Add internalSearch flag for native get-by-id (no Substrait)

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

@@ -0,0 +1,150 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.analytics.spi;
+
+import org.opensearch.common.annotation.ExperimentalApi;
+import org.opensearch.common.lucene.uid.Versions;
+import org.opensearch.common.xcontent.XContentFactory;
+import org.opensearch.core.common.bytes.BytesReference;
+import org.opensearch.core.index.Index;
+import org.opensearch.core.xcontent.XContentBuilder;
+import org.opensearch.index.engine.exec.DocumentMetadataResolver;
+import org.opensearch.index.engine.exec.IndexReaderProvider;
+import org.opensearch.index.engine.exec.Segment;
+import org.opensearch.index.engine.exec.WriterFileSet;
+import org.opensearch.index.get.DocumentLookupResult;
+import org.opensearch.index.seqno.SequenceNumbers;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Core orchestrator for document lookup. Coordinates row-location resolution,
+ * backend-specific execution, and result assembly.
+ *
+ * @opensearch.experimental
+ */
+@ExperimentalApi
+public class DocumentLookupService {
+
+    private final DocumentMetadataResolver documentResolver;
+    private final DocumentRowReader executor;
+
+    public DocumentLookupService(DocumentMetadataResolver documentResolver, DocumentRowReader executor) {
+        this.documentResolver = documentResolver;
+        this.executor = executor;
+    }
+
+    public DocumentLookupResult getById(String id, IndexReaderProvider.Reader reader, Index index) throws IOException {
+        Map<String, Object> row = fetchRow(id, reader);
+        return row == null ? DocumentLookupResult.notFound(id) : buildResultFromRow(id, row);
+    }
+
+    /**
+     * Resolves only the version metadata ({@code _version}/{@code _seq_no}/{@code _primary_term}) for an id,
+     * skipping {@code _source} reconstruction.
+     */
+    public DocumentLookupResult getVersionMetadata(String id, IndexReaderProvider.Reader reader, Index index) throws IOException {
+        Map<String, Object> row = fetchRow(id, reader);
+        if (row == null) {
+            return DocumentLookupResult.notFound(id);
+        }
+        long seqNo = extractLong(row, "_seq_no", SequenceNumbers.UNASSIGNED_SEQ_NO);
+        long primaryTerm = extractLong(row, "_primary_term", SequenceNumbers.UNASSIGNED_PRIMARY_TERM);
+        long version = extractLong(row, "_version", Versions.NOT_FOUND);
+        return new DocumentLookupResult(id, version, true, null, seqNo, primaryTerm, Map.of(), Map.of());
+    }
+
+    /** Locates an id via the resolver and fetches its raw row. Returns null only when the id is not found. */
+    private Map<String, Object> fetchRow(String id, IndexReaderProvider.Reader reader) throws IOException {
+        DocumentMetadataResolver.DocumentMetadata metadata = documentResolver.resolveMetadata(reader, id);
+        if (metadata == null) {
+            return null;
+        }
+
+        WriterFileSet fileSet = reader.catalogSnapshot().findFileSet(executor.formatName(), metadata.writerGeneration());
+        if (fileSet == null) {
+            throw new IllegalStateException(
+                "Resolver located id ["
+                    + id
+                    + "] at writer generation ["
+                    + metadata.writerGeneration()
+                    + "] but no matching file set was found"
+            );
+        }
+
+        Map<String, Object> row = executor.executeSingleRow(metadata.rowId(), fileSet);
+        if (row == null) {
+            throw new IllegalStateException(
+                "Resolver located id ["
+                    + id
+                    + "] at writer generation ["
+                    + metadata.writerGeneration()
+                    + "] rowId ["
+                    + metadata.rowId()
+                    + "] but backend returned no row"
+            );
+        }
+        return row;
+    }
+
+    public List<DocumentLookupResult> getDocsAboveSeqNo(long fromSeqNoExclusive, IndexReaderProvider.Reader reader, Index index)
+        throws IOException {
+        List<WriterFileSet> fileSets = new ArrayList<>();
+        for (Segment segment : reader.catalogSnapshot().getSegments()) {
+            WriterFileSet fileSet = segment.dfGroupedSearchableFiles().get(executor.formatName());
+            if (fileSet != null && !fileSet.files().isEmpty()) {
+                fileSets.add(fileSet);
+            }
+        }
+        List<DocumentLookupResult> results = new ArrayList<>();
+        for (Map<String, Object> row : executor.executeRowsAboveSeqNo(fileSets, fromSeqNoExclusive)) {
+            Object idVal = row.get("_id");
+            if (idVal != null) {
+                results.add(buildResultFromRow(idVal.toString(), row));
+            }
+        }
+        return results;
+    }
+
+    /** Builds a DocumentLookupResult from a raw row, filtering reserved/internal fields. */
+    private static DocumentLookupResult buildResultFromRow(String id, Map<String, Object> row) throws IOException {
+        long seqNo = extractLong(row, "_seq_no", SequenceNumbers.UNASSIGNED_SEQ_NO);
+        long primaryTerm = extractLong(row, "_primary_term", SequenceNumbers.UNASSIGNED_PRIMARY_TERM);
+        long version = extractLong(row, "_version", Versions.NOT_FOUND);
+
+        Map<String, Object> filtered = new LinkedHashMap<>();
+        for (Map.Entry<String, Object> e : row.entrySet()) {
+            if (ReservedFields.RESERVED.contains(e.getKey()) || e.getKey().startsWith("_")) continue;
+            filtered.put(e.getKey(), e.getValue());
+        }
+
+        BytesReference source;
+        try (XContentBuilder xcb = XContentFactory.jsonBuilder()) {
+            xcb.map(filtered);
+            source = BytesReference.bytes(xcb);
+        }
+
+        return new DocumentLookupResult(id, version, true, source, seqNo, primaryTerm, Map.of(), Map.of());
+    }
+
+    public static long extractLong(Map<String, Object> row, String key, long fallback) {
+        Object v = row.get(key);
+        if (v == null) return fallback;
+        if (v instanceof Number) return ((Number) v).longValue();
+        try {
+            return Long.parseLong(v.toString());
+        } catch (NumberFormatException e) {
+            return fallback;
+        }
+    }
+}

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

@@ -0,0 +1,51 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.analytics.spi;
+
+import org.opensearch.common.annotation.ExperimentalApi;
+import org.opensearch.index.engine.exec.WriterFileSet;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Backend-specific execution contract for reading document rows. Backends implement this
+ * to perform the actual storage read (e.g., DataFusion native parquet scan) from a file set
+ * that the Core layer has already resolved from the catalog snapshot.
+ *
+ * @opensearch.experimental
+ */
+@ExperimentalApi
+public interface DocumentRowReader {
+
+    /**
+     * The storage format this backend reads (e.g. {@code "parquet"}). Lets the Core layer
+     * resolve the backend's candidate {@link WriterFileSet} from the catalog snapshot.
+     */
+    String formatName();
+
+    /**
+     * Fetch a single row at the given offset from the pre-resolved file set.
+     *
+     * @param rowId the row offset to fetch
+     * @param fileSet the file set to read from
+     * @return the row as a field-name → value map, or null if not found
+     */
+    Map<String, Object> executeSingleRow(long rowId, WriterFileSet fileSet) throws IOException;
+
+    /**
+     * Fetch all rows with {@code _seq_no > fromSeqNoExclusive} from the Core-resolved file sets
+     * (one per segment for this backend's format).
+     *
+     * @param fileSets the file sets to scan
+     * @param fromSeqNoExclusive the exclusive lower bound on {@code _seq_no}
+     */
+    List<Map<String, Object>> executeRowsAboveSeqNo(List<WriterFileSet> fileSets, long fromSeqNoExclusive) throws IOException;
+}

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

@@ -0,0 +1,36 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.analytics.spi;
+
+import java.util.Set;
+
+/**
+ * Metadata fields managed by the storage layer that must not appear in
+ * user-visible {@code _source}. Backends use this set to filter internal
+ * columns from document lookup results.
+ *
+ * @opensearch.internal
+ */
+public final class ReservedFields {
+
+    private ReservedFields() {}
+
+    /** Fields reserved by the engine/storage layer — excluded from _source reconstruction. */
+    public static final Set<String> RESERVED = Set.of(
+        "_id",
+        "_seq_no",
+        "_primary_term",
+        "_version",
+        "_doc_count",
+        "_size",
+        "_routing",
+        "_ignored",
+        "__row_id__"
+    );
+}