[core] Support projected binary index manifest scans

paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java

@@ -35,17 +35,23 @@
 /** Schema for global index. */
 public class GlobalIndexMeta {
 
+    public static final String ROW_RANGE_START = "_ROW_RANGE_START";
+    public static final String ROW_RANGE_END = "_ROW_RANGE_END";
+    public static final String INDEX_FIELD_ID = "_INDEX_FIELD_ID";
+    public static final String EXTRA_FIELD_IDS = "_EXTRA_FIELD_IDS";
+    public static final String INDEX_META = "_INDEX_META";
+    public static final String SOURCE_META = "_SOURCE_META";
+
     public static final RowType SCHEMA =
             new RowType(
                     true,
                     Arrays.asList(
-                            new DataField(0, "_ROW_RANGE_START", new BigIntType(false)),
-                            new DataField(1, "_ROW_RANGE_END", new BigIntType(false)),
-                            new DataField(2, "_INDEX_FIELD_ID", new IntType(false)),
-                            new DataField(
-                                    3, "_EXTRA_FIELD_IDS", DataTypes.ARRAY(new IntType(false))),
-                            new DataField(4, "_INDEX_META", DataTypes.BYTES()),
-                            new DataField(5, "_SOURCE_META", DataTypes.BYTES())));
+                            new DataField(0, ROW_RANGE_START, new BigIntType(false)),
+                            new DataField(1, ROW_RANGE_END, new BigIntType(false)),
+                            new DataField(2, INDEX_FIELD_ID, new IntType(false)),
+                            new DataField(3, EXTRA_FIELD_IDS, DataTypes.ARRAY(new IntType(false))),
+                            new DataField(4, INDEX_META, DataTypes.BYTES()),
+                            new DataField(5, SOURCE_META, DataTypes.BYTES())));
 
     private final long rowRangeStart;
     private final long rowRangeEnd;

paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java

@@ -28,10 +28,12 @@
 import org.apache.paimon.index.pkfulltext.PkFullTextIndexFile;
 import org.apache.paimon.index.pksorted.PkSortedIndexFile;
 import org.apache.paimon.index.pkvector.PkVectorAnnSegmentFile;
+import org.apache.paimon.manifest.BinaryIndexManifestEntry;
 import org.apache.paimon.manifest.IndexManifestEntry;
 import org.apache.paimon.manifest.IndexManifestEntrySerializer;
 import org.apache.paimon.manifest.IndexManifestFile;
 import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.utils.CloseableIterator;
 import org.apache.paimon.utils.Filter;
 import org.apache.paimon.utils.IndexFilePathFactories;
 import org.apache.paimon.utils.Pair;
@@ -113,6 +115,19 @@ public List<IndexManifestEntry> scan(String indexType) {
         return scan(snapshotManager.latestSnapshot(), indexType);
     }
 
+    public CloseableIterator<BinaryIndexManifestEntry> scan(
+            BinaryIndexManifestEntry.Projection projection) {
+        return scan(snapshotManager.latestSnapshot(), projection);
+    }
+
+    public CloseableIterator<BinaryIndexManifestEntry> scan(
+            @Nullable Snapshot snapshot, BinaryIndexManifestEntry.Projection projection) {
+        if (snapshot == null || snapshot.indexManifest() == null) {
+            return CloseableIterator.empty();
+        }
+        return indexManifestFile.scan(snapshot.indexManifest(), projection);
+    }
+
     public List<IndexManifestEntry> scan(@Nullable Snapshot snapshot, String indexType) {
         if (snapshot == null) {
             return Collections.emptyList();

paimon-core/src/main/java/org/apache/paimon/manifest/BinaryIndexManifestEntry.java

@@ -0,0 +1,282 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.manifest;
+
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.Arrays;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+import static org.apache.paimon.utils.Preconditions.checkState;
+
+/** Reusable binary view of a projected index manifest entry. */
+public final class BinaryIndexManifestEntry {
+
+    public static final Projection GLOBAL_INDEX_PROJECTION = createGlobalIndexProjection();
+
+    private final Projection projection;
+    private @Nullable InternalRow row;
+
+    private BinaryIndexManifestEntry(Projection projection) {
+        this.projection = projection;
+    }
+
+    private static Projection createGlobalIndexProjection() {
+        RowType manifestType = IndexManifestEntry.MANIFEST_ROW_TYPE;
+        return Projection.create(
+                new RowType(
+                        false,
+                        Arrays.asList(
+                                manifestType.getField(IndexManifestEntry.KIND),
+                                manifestType.getField(IndexManifestEntry.PARTITION),
+                                manifestType.getField(IndexManifestEntry.BUCKET),
+                                manifestType.getField(IndexManifestEntry.INDEX_TYPE),
+                                manifestType
+                                        .getField(IndexManifestEntry.GLOBAL_INDEX)
+                                        .newType(
+                                                GlobalIndexMeta.SCHEMA.project(
+                                                        GlobalIndexMeta.ROW_RANGE_START,
+                                                        GlobalIndexMeta.ROW_RANGE_END,
+                                                        GlobalIndexMeta.INDEX_FIELD_ID,
+                                                        GlobalIndexMeta.EXTRA_FIELD_IDS)))));
+    }
+
+    BinaryIndexManifestEntry replace(InternalRow row) {
+        checkArgument(row != null, "Index manifest row cannot be null.");
+        checkArgument(
+                row.getFieldCount() == projection.projectedType.getFieldCount(),
+                "Index manifest row field count %s does not match projected field count %s.",
+                row.getFieldCount(),
+                projection.projectedType.getFieldCount());
+        this.row = row;
+        return this;
+    }
+
+    void clear() {
+        row = null;
+    }
+
+    public boolean isAdd() {
+        return current().getByte(requiredPosition(projection.kindPosition, IndexManifestEntry.KIND))
+                == FileKind.ADD.toByteValue();
+    }
+
+    public boolean isDelete() {
+        return current().getByte(requiredPosition(projection.kindPosition, IndexManifestEntry.KIND))
+                == FileKind.DELETE.toByteValue();
+    }
+
+    public byte[] partitionBytes() {
+        byte[] partition =
+                current()
+                        .getBinary(
+                                requiredPosition(
+                                        projection.partitionPosition,
+                                        IndexManifestEntry.PARTITION));
+        checkState(partition != null, "Serialized index manifest partition cannot be null.");
+        return partition;
+    }
+
+    public int bucket() {
+        return current()
+                .getInt(requiredPosition(projection.bucketPosition, IndexManifestEntry.BUCKET));
+    }
+
+    public BinaryString indexType() {
+        BinaryString indexType =
+                current()
+                        .getString(
+                                requiredPosition(
+                                        projection.indexTypePosition,
+                                        IndexManifestEntry.INDEX_TYPE));
+        checkState(indexType != null, "Index type cannot be null.");
+        return indexType;
+    }
+
+    public boolean hasGlobalIndexMeta() {
+        return !current()
+                .isNullAt(
+                        requiredPosition(
+                                projection.globalIndexPosition, IndexManifestEntry.GLOBAL_INDEX));
+    }
+
+    public long rowRangeStart() {
+        return globalIndex()
+                .getLong(
+                        requiredPosition(
+                                projection.rowRangeStartPosition, GlobalIndexMeta.ROW_RANGE_START));
+    }
+
+    public long rowRangeEnd() {
+        return globalIndex()
+                .getLong(
+                        requiredPosition(
+                                projection.rowRangeEndPosition, GlobalIndexMeta.ROW_RANGE_END));
+    }
+
+    public int indexFieldId() {
+        return globalIndex()
+                .getInt(
+                        requiredPosition(
+                                projection.indexFieldIdPosition, GlobalIndexMeta.INDEX_FIELD_ID));
+    }
+
+    public boolean hasExtraFields() {
+        int position =
+                requiredPosition(projection.extraFieldIdsPosition, GlobalIndexMeta.EXTRA_FIELD_IDS);
+        InternalRow global = globalIndex();
+        return !global.isNullAt(position) && global.getArray(position).size() > 0;
+    }
+
+    private InternalRow globalIndex() {
+        InternalRow global =
+                current()
+                        .getRow(
+                                requiredPosition(
+                                        projection.globalIndexPosition,
+                                        IndexManifestEntry.GLOBAL_INDEX),
+                                projection.projectedGlobalIndexFieldCount);
+        checkState(global != null, "Global index metadata is not present.");
+        return global;
+    }
+
+    private InternalRow current() {
+        checkState(row != null, "Binary index manifest entry is not backed by a row.");
+        return row;
+    }
+
+    private static int requiredPosition(int position, String fieldName) {
+        if (position < 0) {
+            throw new UnsupportedOperationException(
+                    String.format(
+                            "The selected binary index manifest projection does not contain %s.",
+                            fieldName));
+        }
+        return position;
+    }
+
+    /** Projected index manifest schema together with its bound binary field layout. */
+    public static final class Projection {
+
+        private final RowType projectedType;
+        private final int kindPosition;
+        private final int partitionPosition;
+        private final int bucketPosition;
+        private final int indexTypePosition;
+        private final int globalIndexPosition;
+        private final int projectedGlobalIndexFieldCount;
+        private final int rowRangeStartPosition;
+        private final int rowRangeEndPosition;
+        private final int indexFieldIdPosition;
+        private final int extraFieldIdsPosition;
+
+        private Projection(
+                RowType projectedType,
+                int kindPosition,
+                int partitionPosition,
+                int bucketPosition,
+                int indexTypePosition,
+                int globalIndexPosition,
+                int projectedGlobalIndexFieldCount,
+                int rowRangeStartPosition,
+                int rowRangeEndPosition,
+                int indexFieldIdPosition,
+                int extraFieldIdsPosition) {
+            this.projectedType = projectedType;
+            this.kindPosition = kindPosition;
+            this.partitionPosition = partitionPosition;
+            this.bucketPosition = bucketPosition;
+            this.indexTypePosition = indexTypePosition;
+            this.globalIndexPosition = globalIndexPosition;
+            this.projectedGlobalIndexFieldCount = projectedGlobalIndexFieldCount;
+            this.rowRangeStartPosition = rowRangeStartPosition;
+            this.rowRangeEndPosition = rowRangeEndPosition;
+            this.indexFieldIdPosition = indexFieldIdPosition;
+            this.extraFieldIdsPosition = extraFieldIdsPosition;
+        }
+
+        public static Projection create(RowType projectedType) {
+            checkArgument(projectedType != null, "Projected index manifest type cannot be null.");
+            validateProjection(projectedType);
+
+            int globalIndexPosition = projectedType.getFieldIndex(IndexManifestEntry.GLOBAL_INDEX);
+            int projectedGlobalIndexFieldCount = 0;
+            int rowRangeStartPosition = -1;
+            int rowRangeEndPosition = -1;
+            int indexFieldIdPosition = -1;
+            int extraFieldIdsPosition = -1;
+            if (globalIndexPosition >= 0) {
+                RowType globalIndexType =
+                        (RowType) projectedType.getFields().get(globalIndexPosition).type();
+                projectedGlobalIndexFieldCount = globalIndexType.getFieldCount();
+                rowRangeStartPosition =
+                        globalIndexType.getFieldIndex(GlobalIndexMeta.ROW_RANGE_START);
+                rowRangeEndPosition = globalIndexType.getFieldIndex(GlobalIndexMeta.ROW_RANGE_END);
+                indexFieldIdPosition =
+                        globalIndexType.getFieldIndex(GlobalIndexMeta.INDEX_FIELD_ID);
+                extraFieldIdsPosition =
+                        globalIndexType.getFieldIndex(GlobalIndexMeta.EXTRA_FIELD_IDS);
+            }
+
+            return new Projection(
+                    projectedType,
+                    projectedType.getFieldIndex(IndexManifestEntry.KIND),
+                    projectedType.getFieldIndex(IndexManifestEntry.PARTITION),
+                    projectedType.getFieldIndex(IndexManifestEntry.BUCKET),
+                    projectedType.getFieldIndex(IndexManifestEntry.INDEX_TYPE),
+                    globalIndexPosition,
+                    projectedGlobalIndexFieldCount,
+                    rowRangeStartPosition,
+                    rowRangeEndPosition,
+                    indexFieldIdPosition,
+                    extraFieldIdsPosition);
+        }
+
+        private static void validateProjection(RowType projectedType) {
+            for (DataField projectedField : projectedType.getFields()) {
+                checkArgument(
+                        IndexManifestEntry.MANIFEST_ROW_TYPE.containsField(projectedField.id()),
+                        "Unknown projected index manifest field '%s' (id %s).",
+                        projectedField.name(),
+                        projectedField.id());
+                DataField manifestField =
+                        IndexManifestEntry.MANIFEST_ROW_TYPE.getField(projectedField.id());
+                checkArgument(
+                        projectedField.isPrunedFrom(manifestField),
+                        "Projected index manifest field '%s' does not match %s.",
+                        projectedField.name(),
+                        manifestField);
+            }
+        }
+
+        RowType projectedType() {
+            return projectedType;
+        }
+
+        public BinaryIndexManifestEntry createEntry() {
+            return new BinaryIndexManifestEntry(this);
+        }
+    }
+}

paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java

@@ -45,23 +45,37 @@
 @Public
 public class IndexManifestEntry {
 
+    public static final String KIND = "_KIND";
+    public static final String PARTITION = "_PARTITION";
+    public static final String BUCKET = "_BUCKET";
+    public static final String INDEX_TYPE = "_INDEX_TYPE";
+    public static final String FILE_NAME = "_FILE_NAME";
+    public static final String FILE_SIZE = "_FILE_SIZE";
+    public static final String ROW_COUNT = "_ROW_COUNT";
+    public static final String DELETION_VECTORS_RANGES = "_DELETIONS_VECTORS_RANGES";
+    public static final String EXTERNAL_PATH = "_EXTERNAL_PATH";
+    public static final String GLOBAL_INDEX = "_GLOBAL_INDEX";
+
     public static final RowType SCHEMA =
             new RowType(
                     false,
                     Arrays.asList(
-                            new DataField(0, "_KIND", new TinyIntType(false)),
-                            new DataField(1, "_PARTITION", newBytesType(false)),
-                            new DataField(2, "_BUCKET", new IntType(false)),
-                            new DataField(3, "_INDEX_TYPE", newStringType(false)),
-                            new DataField(4, "_FILE_NAME", newStringType(false)),
-                            new DataField(5, "_FILE_SIZE", new BigIntType(false)),
-                            new DataField(6, "_ROW_COUNT", new BigIntType(false)),
+                            new DataField(0, KIND, new TinyIntType(false)),
+                            new DataField(1, PARTITION, newBytesType(false)),
+                            new DataField(2, BUCKET, new IntType(false)),
+                            new DataField(3, INDEX_TYPE, newStringType(false)),
+                            new DataField(4, FILE_NAME, newStringType(false)),
+                            new DataField(5, FILE_SIZE, new BigIntType(false)),
+                            new DataField(6, ROW_COUNT, new BigIntType(false)),
                             new DataField(
                                     7,
-                                    "_DELETIONS_VECTORS_RANGES",
+                                    DELETION_VECTORS_RANGES,
                                     new ArrayType(true, DeletionVectorMeta.SCHEMA)),
-                            new DataField(8, "_EXTERNAL_PATH", newStringType(true)),
-                            new DataField(9, "_GLOBAL_INDEX", GlobalIndexMeta.SCHEMA)));
+                            new DataField(8, EXTERNAL_PATH, newStringType(true)),
+                            new DataField(9, GLOBAL_INDEX, GlobalIndexMeta.SCHEMA)));
+
+    public static final RowType MANIFEST_ROW_TYPE =
+            ManifestSchemaUtils.withFormatIdentifier(SCHEMA);
 
     private final FileKind kind;
     private final BinaryRow partition;