This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new f75c52d2b7 [core] Support projected binary index manifest scans (#9095)
f75c52d2b7 is described below
commit f75c52d2b74c479c3f7ba0f24a1233832ca4da4d
Author: YeJunHao <[email protected]>
AuthorDate: Fri Aug 7 19:21:26 2026 +0800
[core] Support projected binary index manifest scans (#9095)
---
.../org/apache/paimon/index/GlobalIndexMeta.java | 20 +-
.../org/apache/paimon/index/IndexFileHandler.java | 15 ++
.../paimon/manifest/BinaryIndexManifestEntry.java | 282 +++++++++++++++++++++
.../apache/paimon/manifest/IndexManifestEntry.java | 34 ++-
.../manifest/IndexManifestEntrySerializer.java | 2 +-
.../apache/paimon/manifest/IndexManifestFile.java | 65 ++++-
.../apache/paimon/index/IndexFileHandlerTest.java | 14 +
.../manifest/BinaryIndexManifestEntryTest.java | 152 +++++++++++
8 files changed, 565 insertions(+), 19 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
b/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
index 354c2261cf..026db07867 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/GlobalIndexMeta.java
@@ -35,17 +35,23 @@ import java.util.Objects;
/** 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;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
index 8fa6554886..6621cb56cd 100644
--- a/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
+++ b/paimon-core/src/main/java/org/apache/paimon/index/IndexFileHandler.java
@@ -28,10 +28,12 @@ import org.apache.paimon.fs.Path;
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 class IndexFileHandler {
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();
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryIndexManifestEntry.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/BinaryIndexManifestEntry.java
new file mode 100644
index 0000000000..e951316f09
--- /dev/null
+++
b/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);
+ }
+ }
+}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java
index 10ac3a86a2..f69716d455 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntry.java
@@ -45,23 +45,37 @@ import static
org.apache.paimon.utils.SerializationUtils.newStringType;
@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;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java
index 496ee70835..c37bb77a00 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestEntrySerializer.java
@@ -46,7 +46,7 @@ public class IndexManifestEntrySerializer extends
ObjectSerializer<IndexManifest
private static final int FORMAT_IDENTIFIER = 1;
public IndexManifestEntrySerializer() {
-
super(ManifestSchemaUtils.withFormatIdentifier(IndexManifestEntry.SCHEMA));
+ super(IndexManifestEntry.MANIFEST_ROW_TYPE);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java
b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java
index b4e7240aea..8a312f59cf 100644
---
a/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java
+++
b/paimon-core/src/main/java/org/apache/paimon/manifest/IndexManifestFile.java
@@ -18,6 +18,7 @@
package org.apache.paimon.manifest;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FormatReaderFactory;
import org.apache.paimon.format.FormatWriterFactory;
@@ -25,21 +26,30 @@ import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.table.BucketMode;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.FileStorePathFactory;
+import org.apache.paimon.utils.FileUtils;
import org.apache.paimon.utils.ObjectsFile;
import org.apache.paimon.utils.PathFactory;
import org.apache.paimon.utils.SegmentsCache;
import javax.annotation.Nullable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
/** Index manifest file. */
public class IndexManifestFile extends ObjectsFile<IndexManifestEntry> {
+ private final FileFormat fileFormat;
+ private final RowType manifestType;
+
private IndexManifestFile(
FileIO fileIO,
+ FileFormat fileFormat,
RowType schema,
FormatReaderFactory readerFactory,
FormatWriterFactory writerFactory,
@@ -55,12 +65,64 @@ public class IndexManifestFile extends
ObjectsFile<IndexManifestEntry> {
compression,
pathFactory,
cache);
+ this.fileFormat = fileFormat;
+ this.manifestType = schema;
}
public Path indexManifestFilePath(String fileName) {
return pathFactory.toPath(fileName);
}
+ /**
+ * Scans projected index manifest entries without materializing {@link
IndexManifestEntry}s.
+ *
+ * <p>The returned iterator reuses the same mutable {@link
BinaryIndexManifestEntry} for all
+ * records. An entry is only valid until the next call to {@link
CloseableIterator#hasNext()},
+ * {@link CloseableIterator#next()}, or {@link CloseableIterator#close()},
and must not be
+ * retained. The caller must close the iterator.
+ *
+ * <p>This method intentionally bypasses the manifest cache because cached
entries are
+ * materialized with the complete index manifest schema.
+ */
+ public CloseableIterator<BinaryIndexManifestEntry> scan(
+ String fileName, BinaryIndexManifestEntry.Projection projection) {
+ BinaryIndexManifestEntry entry = projection.createEntry();
+ try {
+ CloseableIterator<InternalRow> rows =
+ FileUtils.createFormatReader(
+ fileIO,
+ fileFormat.createReaderFactory(
+ manifestType,
+ projection.projectedType(),
+ Collections.emptyList()),
+ pathFactory.toPath(fileName),
+ null)
+ .toCloseableIterator();
+ return new CloseableIterator<BinaryIndexManifestEntry>() {
+ @Override
+ public boolean hasNext() {
+ entry.clear();
+ return rows.hasNext();
+ }
+
+ @Override
+ public BinaryIndexManifestEntry next() {
+ entry.clear();
+ InternalRow row = rows.next();
+ return row == null ? null : entry.replace(row);
+ }
+
+ @Override
+ public void close() throws Exception {
+ entry.clear();
+ rows.close();
+ }
+ };
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to read index manifest " +
fileName, e);
+ }
+ }
+
/** Write new index files to index manifest. */
@Nullable
public String writeIndexFiles(
@@ -97,9 +159,10 @@ public class IndexManifestFile extends
ObjectsFile<IndexManifestEntry> {
}
public IndexManifestFile create() {
- RowType schema =
ManifestSchemaUtils.withFormatIdentifier(IndexManifestEntry.SCHEMA);
+ RowType schema = IndexManifestEntry.MANIFEST_ROW_TYPE;
return new IndexManifestFile(
fileIO,
+ fileFormat,
schema,
fileFormat.createReaderFactory(schema, schema, new
ArrayList<>()),
fileFormat.createWriterFactory(schema),
diff --git
a/paimon-core/src/test/java/org/apache/paimon/index/IndexFileHandlerTest.java
b/paimon-core/src/test/java/org/apache/paimon/index/IndexFileHandlerTest.java
index 70d5881e34..dbb759d794 100644
---
a/paimon-core/src/test/java/org/apache/paimon/index/IndexFileHandlerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/index/IndexFileHandlerTest.java
@@ -27,11 +27,13 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.CompactIncrement;
import org.apache.paimon.io.DataIncrement;
+import org.apache.paimon.manifest.BinaryIndexManifestEntry;
import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.IndexFilePathFactories;
import org.apache.paimon.utils.Pair;
@@ -149,6 +151,18 @@ public class IndexFileHandlerTest {
Snapshot snapshot = store.snapshotManager().latestSnapshot();
IndexFileHandler indexFileHandler = store.newIndexFileHandler();
+ int binaryEntryCount = 0;
+ try (CloseableIterator<BinaryIndexManifestEntry> entries =
+ indexFileHandler.scan(snapshot,
BinaryIndexManifestEntry.GLOBAL_INDEX_PROJECTION)) {
+ while (entries.hasNext()) {
+ BinaryIndexManifestEntry entry = entries.next();
+ assertThat(entry.isAdd()).isTrue();
+ assertThat(entry.indexType()).isNotNull();
+ binaryEntryCount++;
+ }
+ }
+ assertThat(binaryEntryCount).isEqualTo(3);
+
assertThat(
indexFileHandler.scanBuckets(
snapshot, DELETION_VECTORS_INDEX,
Collections.emptySet()))
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryIndexManifestEntryTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryIndexManifestEntryTest.java
new file mode 100644
index 0000000000..02d952ed5e
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/BinaryIndexManifestEntryTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.index.GlobalIndexMeta;
+import org.apache.paimon.index.IndexFileMeta;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.TableTestBase;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.CloseableIterator;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link BinaryIndexManifestEntry}. */
+class BinaryIndexManifestEntryTest extends TableTestBase {
+
+ @Test
+ void testProjectedScanAndReusableEntry() throws Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ IndexManifestFile indexManifestFile =
table.store().indexManifestFileFactory().create();
+
+ BinaryRow firstPartition = partition(1);
+ BinaryRow secondPartition = partition(2);
+ IndexManifestEntry add =
+ entry(
+ FileKind.ADD,
+ firstPartition,
+ 3,
+ "btree",
+ new GlobalIndexMeta(10, 19, 1, new int[] {2}, null));
+ IndexManifestEntry delete =
+ entry(FileKind.DELETE, secondPartition, 4, "deletion-vector",
null);
+ String fileName =
indexManifestFile.writeWithoutRolling(Arrays.asList(add, delete));
+
+ try (CloseableIterator<BinaryIndexManifestEntry> entries =
+ indexManifestFile.scan(
+ fileName,
BinaryIndexManifestEntry.GLOBAL_INDEX_PROJECTION)) {
+ assertThat(entries.hasNext()).isTrue();
+ BinaryIndexManifestEntry first = entries.next();
+ assertThat(first.isAdd()).isTrue();
+ assertThat(first.isDelete()).isFalse();
+
assertThat(deserializeBinaryRow(first.partitionBytes())).isEqualTo(firstPartition);
+ assertThat(first.bucket()).isEqualTo(3);
+ assertThat(first.indexType().toString()).isEqualTo("btree");
+ assertThat(first.hasGlobalIndexMeta()).isTrue();
+ assertThat(first.rowRangeStart()).isEqualTo(10);
+ assertThat(first.rowRangeEnd()).isEqualTo(19);
+ assertThat(first.indexFieldId()).isEqualTo(1);
+ assertThat(first.hasExtraFields()).isTrue();
+
+ assertThat(entries.hasNext()).isTrue();
+ assertThatThrownBy(first::bucket)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("not backed by a row");
+
+ BinaryIndexManifestEntry second = entries.next();
+ assertThat(second).isSameAs(first);
+ assertThat(second.isAdd()).isFalse();
+ assertThat(second.isDelete()).isTrue();
+
assertThat(deserializeBinaryRow(second.partitionBytes())).isEqualTo(secondPartition);
+ assertThat(second.bucket()).isEqualTo(4);
+
assertThat(second.indexType().toString()).isEqualTo("deletion-vector");
+ assertThat(second.hasGlobalIndexMeta()).isFalse();
+ assertThatThrownBy(second::rowRangeStart)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("not present");
+
+ assertThat(entries.hasNext()).isFalse();
+ assertThatThrownBy(second::bucket)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("not backed by a row");
+ }
+ }
+
+ @Test
+ void testCustomProjectionAndOrdering() {
+ RowType manifestType = IndexManifestEntry.MANIFEST_ROW_TYPE;
+ BinaryIndexManifestEntry entry =
+ BinaryIndexManifestEntry.Projection.create(
+ new RowType(
+ false,
+ Arrays.asList(
+ manifestType.getField(
+
IndexManifestEntry.INDEX_TYPE),
+
manifestType.getField(IndexManifestEntry.BUCKET),
+
manifestType.getField(IndexManifestEntry.KIND))))
+ .createEntry()
+ .replace(
+ GenericRow.of(
+ BinaryString.fromString("btree"),
+ 3,
+ FileKind.ADD.toByteValue()));
+
+ assertThat(entry.indexType().toString()).isEqualTo("btree");
+ assertThat(entry.bucket()).isEqualTo(3);
+ assertThat(entry.isAdd()).isTrue();
+ assertThatThrownBy(entry::partitionBytes)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining(IndexManifestEntry.PARTITION);
+ assertThatThrownBy(entry::hasGlobalIndexMeta)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining(IndexManifestEntry.GLOBAL_INDEX);
+ }
+
+ private static IndexManifestEntry entry(
+ FileKind kind,
+ BinaryRow partition,
+ int bucket,
+ String indexType,
+ GlobalIndexMeta globalIndexMeta) {
+ return new IndexManifestEntry(
+ kind,
+ partition,
+ bucket,
+ new IndexFileMeta(indexType, "index-file", 100, 10,
globalIndexMeta, null));
+ }
+
+ private static BinaryRow partition(int value) {
+ BinaryRow row = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(row);
+ writer.writeInt(0, value);
+ writer.complete();
+ return row;
+ }
+}
