initial POC

sandbox/libs/analytics-api/src/main/java/org/opensearch/analytics/schema/OpenSearchSchemaBuilder.java

@@ -117,6 +117,12 @@ private static AbstractTable buildTable(Map<String, Object> properties) {
             public RelDataType getRowType(RelDataTypeFactory typeFactory) {
                 RelDataTypeFactory.Builder builder = typeFactory.builder();
                 addLeafFields(builder, typeFactory, properties, "");
+                // Virtual row ID column — always present in parquet files, computed by analytics backend.
+                // Only add if not already in the mapping.
+                if (!properties.containsKey("__row_id__")) {
+                    builder.add("__row_id__", typeFactory.createTypeWithNullability(
+                        typeFactory.createSqlType(SqlTypeName.BIGINT), true));
+                }
                 return builder.build();
             }
         };

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/AnalyticsSearchBackendPlugin.java

@@ -8,6 +8,10 @@
 
 package org.opensearch.analytics.spi;
 
+import org.apache.arrow.memory.BufferAllocator;
+import org.opensearch.analytics.backend.EngineResultStream;
+import org.opensearch.index.engine.exec.IndexReaderProvider;
+
 import java.util.List;
 
 /**
@@ -113,4 +117,23 @@ default FilterDelegationHandle getFilterDelegationHandle(List<DelegatedExpressio
     default void configureFilterDelegation(FilterDelegationHandle handle, BackendExecutionContext backendContext) {
         throw new UnsupportedOperationException("configureFilterDelegation not implemented for [" + name() + "]");
     }
+
+    /**
+     * QTF fetch phase: reads specific rows by global row ID.
+     * Row IDs are passed as a BigIntVector for zero-copy transfer to native.
+     *
+     * @param reader the index reader for the target shard
+     * @param rowIdVector Arrow BigIntVector containing global row IDs
+     * @param columns column names to read
+     * @param allocator Arrow buffer allocator for result import
+     * @return a result stream containing the requested rows
+     */
+    default EngineResultStream fetchByRowIds(
+        IndexReaderProvider.Reader reader,
+        org.apache.arrow.vector.BigIntVector rowIdVector,
+        String[] columns,
+        BufferAllocator allocator
+    ) {
+        throw new UnsupportedOperationException("fetchByRowIds not implemented for [" + name() + "]");
+    }
 }

sandbox/libs/dataformat-native/rust/Cargo.toml

@@ -81,11 +81,11 @@ opensearch-repository-azure = { path = "../../../plugins/native-repository-azure
 opensearch-repository-fs = { path = "../../../plugins/native-repository-fs/src/main/rust" }
 
 [profile.release]
-lto = true
-codegen-units = 1
-incremental = true
 debug = "line-tables-only"
 strip = false
+lto = false
+codegen-units = 4
+incremental = true
 
 [profile.dev]
 opt-level = 1
@@ -94,3 +94,10 @@ codegen-units = 16
 incremental = true
 debug = "full"
 strip = false
+
+
+[profile.release-fast]
+inherits = "release"
+lto = false
+codegen-units = 4
+incremental = true

sandbox/plugins/analytics-backend-datafusion/rust/Cargo.toml

@@ -79,3 +79,7 @@ rand = "=0.8.6"
 [[bench]]
 name = "query_bench"
 harness = false
+
+[[bench]]
+name = "row_id_bench"
+harness = false