This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit e0233b7ffc6c1348fb94637298b601087da3183f Author: jerry <[email protected]> AuthorDate: Thu Sep 17 14:50:27 2026 +0800 [core] Skip BTree index reads for all-matching predicates (#9834) --- .../paimon/globalindex/GlobalIndexIOMeta.java | 14 ++- .../globalindex/btree/LazyFilteredBTreeReader.java | 111 ++++++++++++++++++ .../globalindex/btree/AbstractIndexReaderTest.java | 1 + .../btree/LazyFilteredBTreeIndexReaderTest.java | 128 ++++++++++++++++++++- .../DataEvolutionGlobalIndexScanner.java | 3 +- .../paimon/table/BtreeGlobalIndexTableTest.java | 75 ++++++++++++ 6 files changed, 328 insertions(+), 4 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexIOMeta.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexIOMeta.java index 40931255a2..cbaf3a35f5 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexIOMeta.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexIOMeta.java @@ -28,11 +28,17 @@ public class GlobalIndexIOMeta { private final Path filePath; private final long fileSize; + private final long rowCount; private final byte[] metadata; public GlobalIndexIOMeta(Path filePath, long fileSize, byte[] metadata) { + this(filePath, fileSize, -1, metadata); + } + + public GlobalIndexIOMeta(Path filePath, long fileSize, long rowCount, byte[] metadata) { this.filePath = filePath; this.fileSize = fileSize; + this.rowCount = rowCount; this.metadata = metadata; } @@ -44,6 +50,11 @@ public class GlobalIndexIOMeta { return fileSize; } + /** Number of indexed rows, or -1 if it is unknown. */ + public long rowCount() { + return rowCount; + } + public byte[] metadata() { return metadata; } @@ -59,12 +70,13 @@ public class GlobalIndexIOMeta { GlobalIndexIOMeta that = (GlobalIndexIOMeta) o; return Objects.equals(filePath, that.filePath) && fileSize == that.fileSize + && rowCount == that.rowCount && Arrays.equals(metadata, that.metadata); } @Override public int hashCode() { - int result = Objects.hash(filePath, fileSize); + int result = Objects.hash(filePath, fileSize, rowCount); result = 31 * result + Arrays.hashCode(metadata); return result; } diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java index 75595f1bd1..fdf9b513a4 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeReader.java @@ -22,17 +22,26 @@ import org.apache.paimon.globalindex.GlobalIndexIOMeta; import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.globalindex.KeySerializer; import org.apache.paimon.globalindex.SortedFileGlobalIndexReader; +import org.apache.paimon.globalindex.SortedIndexFileMeta; import org.apache.paimon.globalindex.io.GlobalIndexFileReader; import org.apache.paimon.io.cache.CacheManager; +import org.apache.paimon.memory.MemorySlice; import org.apache.paimon.predicate.FieldRef; import org.apache.paimon.predicate.TopN; +import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.Range; import org.apache.paimon.utils.RoaringNavigableMap64; +import javax.annotation.Nullable; + import java.io.IOException; +import java.util.Comparator; import java.util.List; import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; +import java.util.function.Predicate; +import java.util.function.Supplier; /** * An Index Reader for BTree which dynamically filters file list by input predicate, then visits @@ -44,6 +53,9 @@ public class LazyFilteredBTreeReader extends SortedFileGlobalIndexReader<BTreeIn private final KeySerializer keySerializer; private final CacheManager cacheManager; private final GlobalIndexFileReader fileReader; + private final Comparator<Object> comparator; + private final long totalRowCount; + @Nullable private final Pair<Object, Object> fullRangeBounds; public LazyFilteredBTreeReader( List<GlobalIndexIOMeta> files, @@ -57,6 +69,105 @@ public class LazyFilteredBTreeReader extends SortedFileGlobalIndexReader<BTreeIn this.cacheManager = cacheManager; this.fileReader = fileReader; this.keySerializer = keySerializer; + this.comparator = keySerializer.createComparator(); + this.totalRowCount = totalRowCount; + this.fullRangeBounds = fullRangeBounds(files); + } + + @Nullable + private Pair<Object, Object> fullRangeBounds(List<GlobalIndexIOMeta> files) { + if (totalRowCount == 0 || files.isEmpty()) { + return null; + } + long remaining = totalRowCount; + Object min = null; + Object max = null; + for (GlobalIndexIOMeta file : files) { + // A file's min/max does not identify its row IDs. Only skip the entire reader + // when its scalar entries cover the complete local row-ID domain without gaps. + if (file.rowCount() < 0 || file.rowCount() > remaining) { + return null; + } + remaining -= file.rowCount(); + SortedIndexFileMeta meta = SortedIndexFileMeta.deserialize(file.metadata()); + if (meta.hasNulls() || meta.firstKey() == null || meta.lastKey() == null) { + return null; + } + Object first = keySerializer.deserialize(MemorySlice.wrap(meta.firstKey())); + Object last = keySerializer.deserialize(MemorySlice.wrap(meta.lastKey())); + if (min == null || comparator.compare(first, min) < 0) { + min = first; + } + if (max == null || comparator.compare(last, max) > 0) { + max = last; + } + } + return remaining == 0 ? Pair.of(min, max) : null; + } + + @Override + public CompletableFuture<Optional<GlobalIndexResult>> visitEqual( + FieldRef fieldRef, Object literal) { + return visitWithAllMatch( + key -> literal != null && comparator.compare(key, literal) == 0, + () -> super.visitEqual(fieldRef, literal)); + } + + @Override + public CompletableFuture<Optional<GlobalIndexResult>> visitLessThan( + FieldRef fieldRef, Object literal) { + return visitWithAllMatch( + key -> literal != null && comparator.compare(key, literal) < 0, + () -> super.visitLessThan(fieldRef, literal)); + } + + @Override + public CompletableFuture<Optional<GlobalIndexResult>> visitLessOrEqual( + FieldRef fieldRef, Object literal) { + return visitWithAllMatch( + key -> literal != null && comparator.compare(key, literal) <= 0, + () -> super.visitLessOrEqual(fieldRef, literal)); + } + + @Override + public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterThan( + FieldRef fieldRef, Object literal) { + return visitWithAllMatch( + key -> literal != null && comparator.compare(key, literal) > 0, + () -> super.visitGreaterThan(fieldRef, literal)); + } + + @Override + public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterOrEqual( + FieldRef fieldRef, Object literal) { + return visitWithAllMatch( + key -> literal != null && comparator.compare(key, literal) >= 0, + () -> super.visitGreaterOrEqual(fieldRef, literal)); + } + + @Override + public CompletableFuture<Optional<GlobalIndexResult>> visitBetween( + FieldRef fieldRef, Object from, Object to) { + return visitWithAllMatch( + key -> + from != null + && to != null + && comparator.compare(key, from) >= 0 + && comparator.compare(key, to) <= 0, + () -> super.visitBetween(fieldRef, from, to)); + } + + // Only use for predicates whose matching keys form one contiguous interval. + private CompletableFuture<Optional<GlobalIndexResult>> visitWithAllMatch( + Predicate<Object> predicate, + Supplier<CompletableFuture<Optional<GlobalIndexResult>>> fallback) { + if (fullRangeBounds != null + && predicate.test(fullRangeBounds.getLeft()) + && predicate.test(fullRangeBounds.getRight())) { + return CompletableFuture.completedFuture( + Optional.of(GlobalIndexResult.fromRange(new Range(0, totalRowCount - 1)))); + } + return fallback.get(); } @Override diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java index fe7ad8b40e..bbb0e2d7e0 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/AbstractIndexReaderTest.java @@ -346,6 +346,7 @@ public abstract class AbstractIndexReaderTest { return new GlobalIndexIOMeta( new Path(new Path(tempPath.toUri()), fileName), fileIO.getFileSize(new Path(new Path(tempPath.toUri()), fileName)), + resultEntry.rowCount(), resultEntry.meta()); } diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java index 3946f82aee..7cc643f3ec 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java @@ -44,6 +44,8 @@ import org.junit.jupiter.api.extension.ExtendWith; import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.Comparator; import java.util.HashSet; import java.util.List; @@ -149,7 +151,7 @@ public class LazyFilteredBTreeIndexReaderTest extends AbstractIndexReaderTest { try (GlobalIndexReader reader = globalIndexer.createReader( fileReader, written, dataNum, newDirectExecutorService())) { - assertThat(reader.visitBetween(ref, min, max).join()).isEmpty(); + assertResult(reader.visitBetween(ref, min, max).join().get(), filter(obj -> true)); GlobalIndexResult result = reader.visitEqual(ref, literal).join().get(); assertResult(result, filter(obj -> comparator.compare(obj, literal) == 0)); @@ -180,13 +182,134 @@ public class LazyFilteredBTreeIndexReaderTest extends AbstractIndexReaderTest { try (GlobalIndexReader reader = globalIndexer.createReader( fileReader, written, dataNum, newDirectExecutorService())) { - assertThat(reader.visitBetween(ref, min, max).join()).isEmpty(); + assertResult(reader.visitBetween(ref, min, max).join().get(), filter(obj -> true)); GlobalIndexResult result = reader.visitGreaterOrEqual(ref, secondFileMin).join().get(); assertResult(result, filter(obj -> comparator.compare(obj, secondFileMin) >= 0)); } } + @TestTemplate + public void testAllMatchAvoidsOpeningBTreeFiles() throws Exception { + Object literal = data.get(dataNum / 2).getKey(); + List<GlobalIndexIOMeta> written = + Arrays.asList( + writeData(Arrays.asList(Pair.of(literal, 0L), Pair.of(literal, 2L))), + writeData(Arrays.asList(Pair.of(literal, 1L), Pair.of(literal, 3L)))); + CountingGlobalIndexFileReader countingReader = new CountingGlobalIndexFileReader(); + FieldRef ref = new FieldRef(1, "testField", dataType); + try (GlobalIndexReader reader = + new OffsetGlobalIndexReader( + globalIndexer.createReader( + countingReader, written, 4, newDirectExecutorService()), + 1000L, + 1003L)) { + assertRows(reader.visitEqual(ref, literal).join().get(), 1000L, 1001L, 1002L, 1003L); + assertRows( + reader.visitGreaterOrEqual(ref, literal).join().get(), + 1000L, + 1001L, + 1002L, + 1003L); + assertRows( + reader.visitLessOrEqual(ref, literal).join().get(), 1000L, 1001L, 1002L, 1003L); + assertRows( + reader.visitBetween(ref, literal, literal).join().get(), + 1000L, + 1001L, + 1002L, + 1003L); + Object lower = data.get(0).getKey(); + Object upper = data.get(dataNum - 1).getKey(); + if (comparator.compare(lower, literal) < 0) { + assertRows( + reader.visitGreaterThan(ref, lower).join().get(), + 1000L, + 1001L, + 1002L, + 1003L); + } + if (comparator.compare(upper, literal) > 0) { + assertRows( + reader.visitLessThan(ref, upper).join().get(), 1000L, 1001L, 1002L, 1003L); + } + assertRows(reader.visitNotEqual(ref, literal).join().get()); + assertRows(reader.visitLessThan(ref, literal).join().get()); + assertRows(reader.visitGreaterThan(ref, literal).join().get()); + assertRows(reader.visitEqual(ref, null).join().get()); + assertThat(countingReader.openedFiles).isEmpty(); + } + } + + @TestTemplate + public void testAllMatchRangeDoesNotConsumeScanBudget() throws Exception { + options.set(BTreeIndexOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE, MemorySize.ofBytes(1)); + globalIndexer = new BTreeGlobalIndexer(new DataField(1, "testField", dataType), options); + List<GlobalIndexIOMeta> written = writeData(); + Object min = data.get(0).getKey(); + Object max = data.get(dataNum - 1).getKey(); + FieldRef ref = new FieldRef(1, "testField", dataType); + CountingGlobalIndexFileReader countingReader = new CountingGlobalIndexFileReader(); + try (GlobalIndexReader reader = + globalIndexer.createReader( + countingReader, written, dataNum, newDirectExecutorService())) { + assertResult(reader.visitBetween(ref, min, max).join().get(), filter(obj -> true)); + assertResult(reader.visitGreaterOrEqual(ref, min).join().get(), filter(obj -> true)); + assertResult(reader.visitLessOrEqual(ref, max).join().get(), filter(obj -> true)); + assertThat(countingReader.openedFiles).isEmpty(); + assertThat(reader.visitLessThan(ref, max).join()).isEmpty(); + } + } + + @TestTemplate + public void testAllMatchFallsBackForNullsAndIncompleteCoverage() throws Exception { + Object literal = data.get(dataNum / 2).getKey(); + FieldRef ref = new FieldRef(1, "testField", dataType); + GlobalIndexIOMeta sparse = + writeData(Arrays.asList(Pair.of(literal, 1L), Pair.of(literal, 3L))); + GlobalIndexIOMeta withNulls = + writeData( + Arrays.asList( + Pair.of(null, 0L), + Pair.of(literal, 1L), + Pair.of(null, 2L), + Pair.of(literal, 3L))); + GlobalIndexIOMeta unknownCount = + new GlobalIndexIOMeta(sparse.filePath(), sparse.fileSize(), sparse.metadata()); + for (GlobalIndexIOMeta meta : Arrays.asList(sparse, withNulls, unknownCount)) { + CountingGlobalIndexFileReader countingReader = new CountingGlobalIndexFileReader(); + try (GlobalIndexReader reader = + globalIndexer.createReader( + countingReader, + Collections.singletonList(meta), + 4, + newDirectExecutorService())) { + assertRows(reader.visitEqual(ref, literal).join().get(), 1L, 3L); + assertThat(countingReader.openedFiles).containsExactly(meta.filePath()); + } + } + } + + @TestTemplate + public void testAllMatchRequiresEveryFileToMatch() throws Exception { + Object min = data.get(0).getKey(); + Object max = data.get(dataNum - 1).getKey(); + if (comparator.compare(min, max) == 0) { + return; + } + List<GlobalIndexIOMeta> written = + Arrays.asList(writeData(singletonData(min, 0L)), writeData(singletonData(max, 1L))); + CountingGlobalIndexFileReader countingReader = new CountingGlobalIndexFileReader(); + FieldRef ref = new FieldRef(1, "testField", dataType); + try (GlobalIndexReader reader = + globalIndexer.createReader( + countingReader, written, 2, newDirectExecutorService())) { + assertRows(reader.visitEqual(ref, min).join().get(), 0L); + assertRows(reader.visitLessThan(ref, max).join().get(), 0L); + assertThat(countingReader.openedFiles).containsExactly(written.get(0).filePath()); + } + } + @TestTemplate public void testOffsetRangeComplementAvoidsOpeningAllBTreeFiles() throws Exception { if (!(dataType instanceof IntType)) { @@ -375,6 +498,7 @@ public class LazyFilteredBTreeIndexReaderTest extends AbstractIndexReaderTest { return new GlobalIndexIOMeta( new Path(new Path(tempPath.toUri()), fileName), fileIO.getFileSize(new Path(new Path(tempPath.toUri()), fileName)), + resultEntry.rowCount(), resultEntry.meta()); } diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java index d391743785..5807e1f1cf 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java @@ -478,7 +478,8 @@ public class DataEvolutionGlobalIndexScanner implements Closeable { GlobalIndexMeta globalIndex = meta.globalIndexMeta(); checkNotNull(globalIndex); Path filePath = indexPathFactory.toPath(meta); - return new GlobalIndexIOMeta(filePath, meta.fileSize(), globalIndex.indexMeta()); + return new GlobalIndexIOMeta( + filePath, meta.fileSize(), meta.rowCount(), globalIndex.indexMeta()); } @Override diff --git a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java index 0349b3290a..76faf508ab 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java @@ -153,6 +153,65 @@ public class BtreeGlobalIndexTableTest extends DataEvolutionTestBase { assertThat(readF1).containsExactly("a200", "a300", "a400", "a56789"); } + @Test + public void testAllMatchSkipsIndexFilesAndPreservesCoverage() throws Exception { + createTableDefault(); + appendDogRows(0, 20); + createIndex("f1"); + createIndex("f0"); + appendDogRows(20, 25); + + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + int skippedFiles = 0; + for (IndexManifestEntry entry : table.store().newIndexFileHandler().scanEntries()) { + IndexFileMeta file = entry.indexFile(); + if ("btree".equals(file.indexType()) + && file.globalIndexMeta().indexFieldId() + == table.rowType().getField("f1").id()) { + // Any attempt to open the all-matching index must fail, including a cold scan. + assertThat( + table.fileIO() + .delete( + table.store() + .pathFactory() + .globalIndexFileFactory() + .toPath(file), + false)) + .isTrue(); + skippedFiles++; + } + } + assertThat(skippedFiles).isPositive(); + + PredicateBuilder builder = new PredicateBuilder(table.rowType()); + Predicate allMatch = builder.equal(1, BinaryString.fromString("dog")); + for (String mode : Arrays.asList("fast", "full", "detail")) { + FileStoreTable configured = tableWithSearchMode(table, mode); + assertThat(readF1(configured, allMatch)) + .containsExactlyElementsOf( + Collections.nCopies(mode.equals("fast") ? 20 : 25, "dog")); + + // An all-matching branch must not disable another column's selective index. + ReadBuilder readBuilder = + configured + .newReadBuilder() + .withFilter(PredicateBuilder.and(allMatch, builder.equal(0, 7))); + TableScan.Plan plan = readBuilder.newScan().plan(); + assertThat(readF1(readBuilder, plan)).containsExactly("dog"); + if (mode.equals("fast")) { + assertThat( + plan.splits().stream() + .map(IndexedSplit.class::cast) + .flatMap(split -> split.rowRanges().stream()) + .collect(Collectors.toList())) + .containsExactly(new Range(7, 7)); + } + assertThat(readF1(configured, PredicateBuilder.or(allMatch, builder.equal(0, 7)))) + .containsExactlyElementsOf( + Collections.nCopies(mode.equals("fast") ? 20 : 25, "dog")); + } + } + @Test public void testFullSearchIgnoresUnindexedAndResidualForCoverage() throws Exception { write(100L); @@ -872,6 +931,22 @@ public class BtreeGlobalIndexTableTest extends DataEvolutionTestBase { Collections.singletonMap(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), searchMode)); } + private void appendDogRows(int fromInclusive, int toExclusive) throws Exception { + FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); + BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + for (int i = fromInclusive; i < toExclusive; i++) { + write.write( + GenericRow.of( + i, + BinaryString.fromString("dog"), + BinaryString.fromString("b" + i))); + } + commit.commit(write.prepareCommit()); + } + } + private void appendRows(int fromInclusive, int toExclusive) throws Exception { FileStoreTable table = (FileStoreTable) catalog.getTable(identifier()); BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
