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 822c5b29c5 [core] Prune FM index partitions by query row ranges
(#10201)
822c5b29c5 is described below
commit 822c5b29c5713641ff6ee565bbaa7f9594d66bd2
Author: QuakeWang <[email protected]>
AuthorDate: Sat Oct 3 17:06:49 2026 +0800
[core] Prune FM index partitions by query row ranges (#10201)
---
.../globalindex/fmindex/FMGlobalIndexer.java | 16 ++
.../globalindex/fmindex/FMGlobalIndexTest.java | 165 +++++++++++++++++++++
.../apache/paimon/table/IndexQuerySplitTest.java | 100 +++++++++++++
3 files changed, 281 insertions(+)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexer.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexer.java
index 6a8969abc3..45fe62248a 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexer.java
@@ -119,6 +119,7 @@ public class FMGlobalIndexer implements GlobalIndexer {
FMGlobalIndexReader.FileSetRowCountValidator validator =
new FMGlobalIndexReader.FileSetRowCountValidator(files.size(),
totalRowCount);
FMIndexFile.IndexMeta[] indexMetas = new
FMIndexFile.IndexMeta[files.size()];
+ // Validate the complete file set before pruning partitions.
for (int i = 0; i < files.size(); i++) {
byte[] metadata = files.get(i).metadata();
checkArgument(
@@ -134,6 +135,17 @@ public class FMGlobalIndexer implements GlobalIndexer {
FMGlobalIndexReader.ContainerMetadataLoader container =
new
FMGlobalIndexReader.ContainerMetadataLoader(files.get(i), indexMetas[i]);
for (FMIndexFile.PartitionMeta partition :
indexMetas[i].partitions) {
+ if (rowRanges != null
+ && rowRanges.stream()
+ .noneMatch(
+ range ->
+ Range.intersect(
+ range.from,
+ range.to,
+ partition.firstRowId,
+
partition.lastRowId()))) {
+ continue;
+ }
readers.add(
new FMGlobalIndexReader(
fileReader,
@@ -146,6 +158,10 @@ public class FMGlobalIndexer implements GlobalIndexer {
locateCostRatio));
}
}
+ if (readers.isEmpty()) {
+ return FMGlobalIndexReader.empty(
+ executor, readContext, demandPageSize, locateCostRatio);
+ }
return readers.size() == 1 ? readers.get(0) : new
UnionGlobalIndexReader(readers);
}
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java
index 8e7befeb4e..af6a9e1512 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/fmindex/FMGlobalIndexTest.java
@@ -39,10 +39,14 @@ import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.utils.Range;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import java.io.IOException;
import java.io.RandomAccessFile;
@@ -62,6 +66,7 @@ import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Stream;
import static
org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
import static org.assertj.core.api.Assertions.assertThat;
@@ -109,6 +114,166 @@ public class FMGlobalIndexTest {
indexer = new FMGlobalIndexer(dataField, options);
}
+ @ParameterizedTest
+ @MethodSource("partitionRowRanges")
+ public void testPartitionPruning(
+ List<Range> ranges, int expectedPartitions, long[] expectedRows)
throws Exception {
+ List<GlobalIndexIOMeta> files =
+ writeData(
+ Arrays.asList(
+ str("needle-0"),
+ str("other-1"),
+ str("other-2"),
+ str("needle-3"),
+ str("other-4"),
+ str("other-5"),
+ str("needle-6")),
+ 0);
+ assertThat(files).hasSize(1);
+
assertThat(FMIndexFile.readIndexMeta(files.get(0).metadata()).partitions).hasSize(3);
+ AtomicInteger opens = new AtomicInteger();
+ GlobalIndexFileReader countingReader =
+ meta -> {
+ opens.incrementAndGet();
+ return fileIO.newInputStream(meta.filePath());
+ };
+ CountingDirectExecutor executor = new CountingDirectExecutor();
+ try (GlobalIndexReader reader =
+ indexer.createReader(countingReader, files, 7, ranges,
executor)) {
+ assertThat(opens).hasValue(0);
+ assertThat(executor.submittedTasks).hasValue(0);
+ assertRows(reader.visitContains(fieldRef, str("needle")).join(),
expectedRows);
+ if (expectedPartitions == 0) {
+ assertRows(reader.visitIsNull(fieldRef).join());
+ assertRows(reader.visitIsNotNull(fieldRef).join());
+ }
+ }
+ assertThat(executor.submittedTasks).hasValue(expectedPartitions);
+ assertThat(opens).hasValue(expectedPartitions);
+ }
+
+ private static Stream<Arguments> partitionRowRanges() {
+ return Stream.of(
+ Arguments.of(null, 3, new long[] {0, 3, 6}),
+ Arguments.of(Collections.emptyList(), 0, new long[0]),
+ Arguments.of(Collections.singletonList(new Range(0, 1)), 1,
new long[] {0}),
+ Arguments.of(Collections.singletonList(new Range(7, 9)), 0,
new long[0]),
+ Arguments.of(Collections.singletonList(new Range(2, 3)), 2,
new long[] {0, 3}),
+ Arguments.of(Arrays.asList(new Range(2, 2), new Range(0, 0)),
1, new long[] {0}),
+ Arguments.of(Arrays.asList(new Range(0, 1), new Range(1, 2)),
1, new long[] {0}),
+ Arguments.of(Collections.singletonList(new Range(6, 6)), 1,
new long[] {6}),
+ // Boundary partitions retain matching rows outside the
requested range.
+ Arguments.of(Collections.singletonList(new Range(2, 2)), 1,
new long[] {0}),
+ Arguments.of(Collections.singletonList(new Range(3, 3)), 1,
new long[] {3}),
+ Arguments.of(
+ Arrays.asList(new Range(6, 6), new Range(0, 0)), 2,
new long[] {0, 6}));
+ }
+
+ @Test
+ public void testPartitionPruningAcrossFilesPreservesRowIds() throws
Exception {
+ List<GlobalIndexIOMeta> files = new ArrayList<>();
+ files.addAll(writeData(Arrays.asList(str("needle-0"), null,
str("other-2")), 0));
+ files.addAll(writeData(Arrays.asList(str("needle-3"), null,
str("needle-5")), 3));
+ assertThat(files).hasSize(2);
+ AtomicInteger opens = new AtomicInteger();
+ GlobalIndexFileReader countingReader =
+ meta -> {
+
assertThat(meta.filePath()).isEqualTo(files.get(1).filePath());
+ opens.incrementAndGet();
+ return fileIO.newInputStream(meta.filePath());
+ };
+ CountingDirectExecutor executor = new CountingDirectExecutor();
+ try (GlobalIndexReader reader =
+ indexer.createReader(
+ countingReader,
+ files,
+ 6,
+ Collections.singletonList(new Range(4, 4)),
+ executor)) {
+ assertRows(reader.visitContains(fieldRef, str("needle")).join(),
3L, 5L);
+ assertRows(reader.visitIsNull(fieldRef).join(), 4L);
+ }
+ assertThat(executor.submittedTasks).hasValue(2);
+ assertThat(opens).hasValue(2);
+ }
+
+ @Test
+ public void testPrunedFileMetadataIsStillValidated() throws Exception {
+ GlobalIndexIOMeta first = writeData(Arrays.asList(str("first"), null),
0).get(0);
+ GlobalIndexIOMeta second = writeData(Arrays.asList(str("second"),
null), 2).get(0);
+ GlobalIndexIOMeta overlapping =
writeData(Arrays.asList(str("overlap"), null), 1).get(0);
+ byte[] metadata = Arrays.copyOf(second.metadata(),
second.metadata().length);
+ metadata[metadata.length - 1] ^= 1;
+ GlobalIndexIOMeta corrupted =
+ new GlobalIndexIOMeta(second.filePath(), second.fileSize(),
metadata);
+ GlobalIndexFileReader noIO =
+ meta -> {
+ throw new AssertionError("Unexpected I/O");
+ };
+ for (List<Range> ranges :
+ Arrays.asList(
+ Collections.<Range>emptyList(),
+ Collections.singletonList(new Range(0, 0)),
+ Collections.singletonList(new Range(4, 4)))) {
+ assertThatThrownBy(
+ () ->
+ indexer.createReader(
+ noIO,
+ Arrays.asList(first, corrupted),
+ 4,
+ ranges,
+ newDirectExecutorService()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("metadata checksum mismatch");
+ assertThatThrownBy(
+ () ->
+ indexer.createReader(
+ noIO,
+ Arrays.asList(first, overlapping),
+ 4,
+ ranges,
+ newDirectExecutorService()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("do not exactly cover source rows");
+ }
+ }
+
+ @Test
+ public void testMetadataValidationWithEmptyRanges() throws Exception {
+ List<GlobalIndexIOMeta> files =
writeData(Collections.singletonList(str("needle")), 0);
+ GlobalIndexIOMeta file = files.get(0);
+ byte[] metadata = Arrays.copyOf(file.metadata(),
file.metadata().length);
+ metadata[metadata.length - 1] ^= 1;
+ GlobalIndexFileReader noIO =
+ meta -> {
+ throw new AssertionError("Unexpected I/O");
+ };
+ assertThatThrownBy(
+ () ->
+ indexer.createReader(
+ noIO,
+ Collections.singletonList(
+ new GlobalIndexIOMeta(
+ file.filePath(),
+ file.fileSize(),
+ metadata)),
+ 1,
+ Collections.emptyList(),
+ newDirectExecutorService()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("metadata checksum mismatch");
+ assertThatThrownBy(
+ () ->
+ indexer.createReader(
+ noIO,
+ files,
+ 2,
+ Collections.emptyList(),
+ newDirectExecutorService()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("row count mismatch");
+ }
+
@Test
public void testExactContainsForShortUtf8AndRepeatedOccurrences() throws
Exception {
List<GlobalIndexIOMeta> files =
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/IndexQuerySplitTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/IndexQuerySplitTest.java
index a7ca6b9eb2..040519cfe9 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/IndexQuerySplitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/IndexQuerySplitTest.java
@@ -903,6 +903,106 @@ public class IndexQuerySplitTest extends
DataEvolutionTestBase {
return table.copy(options);
}
+ @Test
+ public void testFMPartitionPruningAfterSplitRestore() throws Exception {
+ write(100);
+ appendRows(100, 101);
+ appendRows(101, 104);
+ appendRows(104, 106);
+ FileStoreTable original = smallSplits(getTableDefault());
+ GlobalIndexFileReadWrite io =
+ new GlobalIndexFileReadWrite(
+ original.fileIO(),
original.store().pathFactory().globalIndexFileFactory());
+ Options indexOptions = new Options();
+ indexOptions.set(FMGlobalIndexOptions.PARTITION_ROW_COUNT, 2);
+ indexOptions.set(FMGlobalIndexOptions.SA_SAMPLE_RATE, 1);
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
+ GlobalIndexer.create("fm",
original.rowType().getField("f1"), indexOptions)
+ .createWriter(io);
+ for (int i = 0; i < 6; i++) {
+ writer.write(str("a" + (100 + i)), i);
+ }
+ List<IndexFileMeta> indexes = new ArrayList<>();
+ for (ResultEntry entry : writer.finish()) {
+ indexes.add(
+ new IndexFileMeta(
+ "fm",
+ entry.fileName(),
+ io.fileSize(entry.fileName()),
+ entry.rowCount(),
+ new GlobalIndexMeta(
+ 100,
+ 105,
+ original.rowType().getField("f1").id(),
+ null,
+ entry.meta()),
+ null));
+ }
+ assertThat(indexes).hasSize(1);
+ try (BatchTableCommit commit =
original.newBatchWriteBuilder().newCommit()) {
+ commit.commit(
+ Collections.singletonList(
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ BucketMode.UNAWARE_BUCKET,
+ null,
+ DataIncrement.indexIncrement(indexes),
+ CompactIncrement.emptyIncrement())));
+ }
+ Path indexPath =
+
original.store().pathFactory().globalIndexFileFactory().toPath(indexes.get(0));
+ AtomicInteger opens = new AtomicInteger();
+ FileIO fileIO = spy(original.fileIO());
+ doAnswer(
+ invocation -> {
+ if (indexPath.equals(invocation.getArgument(0))) {
+ opens.incrementAndGet();
+ }
+ return invocation.callRealMethod();
+ })
+ .when(fileIO)
+ .newInputStream(any(Path.class));
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), "fast");
+ options.put(FMGlobalIndexOptions.LOCATE_COST_RATIO.key(), "1");
+ FileStoreTable table =
+ distributedTable(
+ smallSplits(
+ new AppendOnlyFileStoreTable(
+ fileIO, original.location(),
original.schema()))
+ .copy(options));
+ ReadBuilder read =
+ table.newReadBuilder()
+ .withFilter(new
PredicateBuilder(table.rowType()).contains(1, str("a10")));
+ List<Split> splits = read.newScan().plan().splits();
+
assertThat(splits).hasSize(3).allMatch(IndexQuerySplit.class::isInstance);
+ assertThat(opens).hasValue(0);
+ List<Split> restoredSplits = new ArrayList<>();
+ for (Split split : splits) {
+ IndexQuerySplit restored =
+ (IndexQuerySplit)
SplitSerializer.deserialize(SplitSerializer.serialize(split));
+ assertThat(restored).isEqualTo(split);
+ IndexQuerySplit javaRestored =
+ InstantiationUtil.deserializeObject(
+ InstantiationUtil.serializeObject(split),
getClass().getClassLoader());
+ assertThat(javaRestored).isEqualTo(split);
+ long firstRowId =
restored.dataSplit().dataFiles().get(0).nonNullRowIdRange().from;
+ long lastRowId =
restored.dataSplit().dataFiles().get(0).nonNullRowIdRange().to;
+ for (IndexQuerySplit transported : Arrays.asList(restored,
javaRestored)) {
+ opens.set(0);
+ IndexedSplit evaluated = transported.evaluate(fileIO);
+ assertThat(evaluated.rowRanges()).containsExactly(new
Range(firstRowId, lastRowId));
+ assertThat(opens).hasValue(firstRowId == 101 ? 2 : 1);
+ }
+ restoredSplits.add(restored);
+ }
+ assertThat(read(read, restoredSplits)).containsExactly(100, 101, 102,
103, 104, 105);
+ original.fileIO().delete(indexPath, false);
+ assertThatThrownBy(() -> ((IndexQuerySplit)
restoredSplits.get(0)).evaluate(fileIO))
+ .isInstanceOf(IOException.class);
+ }
+
@Test
public void testFMFallbackAndMixedDistributedIndex() throws Exception {
write(100);
