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 0a5678bfbe2a680ea634b79bc5c8747a4bcd1da6 Author: jerry <[email protected]> AuthorDate: Fri Sep 18 11:55:09 2026 +0800 [core] Use a single BTree scan for bounded range predicates (#9844) --- .../paimon/globalindex/GlobalIndexEvaluator.java | 69 ++++++++++++++++- .../paimon/globalindex/GlobalIndexReader.java | 22 ++++++ .../globalindex/OffsetGlobalIndexReader.java | 7 ++ .../globalindex/SortedFileGlobalIndexReader.java | 13 +++- .../paimon/globalindex/UnionGlobalIndexReader.java | 7 ++ .../paimon/globalindex/btree/BTreeIndexReader.java | 5 ++ .../globalindex/btree/LazyFilteredBTreeReader.java | 10 +++ .../globalindex/GlobalIndexEvaluatorTest.java | 48 ++++++++++++ .../btree/LazyFilteredBTreeIndexReaderTest.java | 90 ++++++++++++++++++++++ 9 files changed, 265 insertions(+), 6 deletions(-) diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java index 224bc2805a..48bfb7780a 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java @@ -18,8 +18,11 @@ package org.apache.paimon.globalindex; +import org.apache.paimon.predicate.And; import org.apache.paimon.predicate.CompoundPredicate; import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.predicate.GreaterOrEqual; +import org.apache.paimon.predicate.GreaterThan; import org.apache.paimon.predicate.IsNaN; import org.apache.paimon.predicate.IsNotNull; import org.apache.paimon.predicate.LeafBinaryFunction; @@ -27,6 +30,8 @@ import org.apache.paimon.predicate.LeafFunction; import org.apache.paimon.predicate.LeafNAryFunction; import org.apache.paimon.predicate.LeafPredicate; import org.apache.paimon.predicate.LeafTernaryFunction; +import org.apache.paimon.predicate.LessOrEqual; +import org.apache.paimon.predicate.LessThan; import org.apache.paimon.predicate.Or; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.TopN; @@ -49,6 +54,7 @@ import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutionException; +import java.util.function.Function; import java.util.function.IntFunction; import java.util.stream.Collectors; @@ -142,6 +148,14 @@ public class GlobalIndexEvaluator implements Closeable { return CompletableFuture.completedFuture(Optional.empty()); } FieldRef fieldRef = fieldRefOptional.get(); + return visitFieldAsync( + fieldRef, + reader -> predicate.function().visit(reader, fieldRef, predicate.literals())); + } + + private CompletableFuture<Optional<Evaluation>> visitFieldAsync( + FieldRef fieldRef, + Function<GlobalIndexReader, CompletableFuture<Optional<GlobalIndexResult>>> visitor) { int fieldId = rowType.getField(fieldRef.name()).id(); Collection<GlobalIndexReader> readers = indexReadersCache.computeIfAbsent(fieldId, readersFunction::apply); @@ -149,7 +163,7 @@ public class GlobalIndexEvaluator implements Closeable { List<CompletableFuture<Optional<GlobalIndexResult>>> readerFutures = new ArrayList<>(readers.size()); for (GlobalIndexReader reader : readers) { - readerFutures.add(predicate.function().visit(reader, fieldRef, predicate.literals())); + readerFutures.add(visitor.apply(reader)); } return CompletableFuture.allOf(readerFutures.toArray(new CompletableFuture[0])) @@ -189,8 +203,41 @@ public class GlobalIndexEvaluator implements Closeable { } List<CompletableFuture<Optional<Evaluation>>> childFutures = new ArrayList<>(children.size()); - for (Predicate child : children) { - childFutures.add(visitAsync(child)); + for (int i = 0; i < children.size(); i++) { + Predicate child = children.get(i); + CompletableFuture<Optional<Evaluation>> range = null; + if (predicate.function() instanceof And && isRangeBound(child)) { + LeafPredicate first = (LeafPredicate) child; + // ponytail: pair search is quadratic in filter count; group by field for large + // conjunctions. + for (int j = i + 1; j < children.size(); j++) { + Predicate other = children.get(j); + if (!isRangeBound(other)) { + continue; + } + LeafPredicate second = (LeafPredicate) other; + if (isLowerBound(first) == isLowerBound(second) + || !first.fieldRefOptional().equals(second.fieldRefOptional())) { + continue; + } + LeafPredicate lower = isLowerBound(first) ? first : second; + LeafPredicate upper = isLowerBound(first) ? second : first; + FieldRef field = first.fieldRefOptional().get(); + range = + visitFieldAsync( + field, + reader -> + reader.visitRange( + field, + lower.literals().get(0), + upper.literals().get(0), + lower.function() instanceof GreaterOrEqual, + upper.function() instanceof LessOrEqual)); + children.remove(j); + break; + } + } + childFutures.add(range == null ? visitAsync(child) : range); } return CompletableFuture.allOf(childFutures.toArray(new CompletableFuture[0])) @@ -204,6 +251,22 @@ public class GlobalIndexEvaluator implements Closeable { }); } + private static boolean isRangeBound(Predicate predicate) { + if (!(predicate instanceof LeafPredicate)) { + return false; + } + LeafPredicate leaf = (LeafPredicate) predicate; + return leaf.fieldRefOptional().isPresent() + && (isLowerBound(leaf) + || leaf.function() instanceof LessThan + || leaf.function() instanceof LessOrEqual) + && leaf.literals().get(0) != null; + } + + private static boolean isLowerBound(LeafPredicate leaf) { + return leaf.function() instanceof GreaterThan || leaf.function() instanceof GreaterOrEqual; + } + private Optional<Evaluation> combineResults( List<Optional<Evaluation>> results, CompoundPredicate predicate) { Set<Integer> contributingFieldIds = new HashSet<>(); diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java index c6901c176b..f5d6ab9fad 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java @@ -77,6 +77,28 @@ public interface GlobalIndexReader return CompletableFuture.completedFuture(Optional.empty()); } + /** + * Evaluate a bounded range. Flags specify whether each endpoint is included. Readers without a + * combined scan retain the supported bounds as candidate filters. + */ + default CompletableFuture<Optional<GlobalIndexResult>> visitRange( + FieldRef fieldRef, Object from, Object to, boolean fromInclusive, boolean toInclusive) { + CompletableFuture<Optional<GlobalIndexResult>> lower = + fromInclusive + ? visitGreaterOrEqual(fieldRef, from) + : visitGreaterThan(fieldRef, from); + CompletableFuture<Optional<GlobalIndexResult>> upper = + toInclusive ? visitLessOrEqual(fieldRef, to) : visitLessThan(fieldRef, to); + return lower.thenCombine( + upper, + (left, right) -> { + if (!left.isPresent()) { + return right; + } + return right.isPresent() ? Optional.of(left.get().and(right.get())) : left; + }); + } + @Override default CompletableFuture<Optional<GlobalIndexResult>> visitNotLike( FieldRef fieldRef, Object literal) { diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java index 9c29bb918d..70283659ec 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java @@ -184,6 +184,13 @@ public class OffsetGlobalIndexReader implements ContainsRefiningGlobalIndexReade return wrapped.visitNotIn(fieldRef, literals).thenApply(this::applyOffset); } + @Override + public CompletableFuture<Optional<GlobalIndexResult>> visitRange( + FieldRef fieldRef, Object from, Object to, boolean fromInclusive, boolean toInclusive) { + return wrapped.visitRange(fieldRef, from, to, fromInclusive, toInclusive) + .thenApply(this::applyOffset); + } + @Override public CompletableFuture<Optional<GlobalIndexResult>> visitBetween( FieldRef fieldRef, Object from, Object to) { diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java index 814ec92414..fc78ce190d 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/SortedFileGlobalIndexReader.java @@ -236,12 +236,19 @@ public abstract class SortedFileGlobalIndexReader<R extends Closeable> @Override public CompletableFuture<Optional<GlobalIndexResult>> visitBetween( FieldRef fieldRef, Object from, Object to) { + return visitRange(fieldRef, from, to, reader -> visitBetween(reader, from, to)); + } + + /** Prune files using enclosing inclusive bounds, then apply the exact range in the reader. */ + protected CompletableFuture<Optional<GlobalIndexResult>> visitRange( + FieldRef fieldRef, + Object from, + Object to, + Function<R, Optional<GlobalIndexResult>> visitor) { if (!canFallbackScan(from) || to == null) { return unsupported(); } - return visitFallbackParallel( - () -> fileSelector.visitBetween(fieldRef, from, to), - reader -> visitBetween(reader, from, to)); + return visitFallbackParallel(() -> fileSelector.visitBetween(fieldRef, from, to), visitor); } @Override diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java index f55986c694..dd2cb1eeaa 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java @@ -181,6 +181,13 @@ public class UnionGlobalIndexReader implements ContainsRefiningGlobalIndexReader return unionAsync(reader -> reader.visitNotIn(fieldRef, literals)); } + @Override + public CompletableFuture<Optional<GlobalIndexResult>> visitRange( + FieldRef fieldRef, Object from, Object to, boolean fromInclusive, boolean toInclusive) { + return unionAsync( + reader -> reader.visitRange(fieldRef, from, to, fromInclusive, toInclusive)); + } + @Override public CompletableFuture<Optional<GlobalIndexResult>> visitBetween( FieldRef fieldRef, Object from, Object to) { diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java index d7d6017b5c..b42224106d 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeIndexReader.java @@ -358,6 +358,11 @@ public class BTreeIndexReader implements Closeable { }); } + public Optional<GlobalIndexResult> visitRange( + Object from, Object to, boolean fromInclusive, boolean toInclusive) { + return createResult(() -> rangeQuery(from, to, fromInclusive, toInclusive)); + } + public Optional<GlobalIndexResult> visitBetween(Object from, Object to) { return createResult(() -> rangeQuery(from, to, true, true)); } 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 fdf9b513a4..f47c218350 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 @@ -229,6 +229,16 @@ public class LazyFilteredBTreeReader extends SortedFileGlobalIndexReader<BTreeIn return reader.visitIn(literals); } + @Override + public CompletableFuture<Optional<GlobalIndexResult>> visitRange( + FieldRef fieldRef, Object from, Object to, boolean fromInclusive, boolean toInclusive) { + return visitRange( + fieldRef, + from, + to, + reader -> reader.visitRange(from, to, fromInclusive, toInclusive)); + } + @Override protected Optional<GlobalIndexResult> visitBetween( BTreeIndexReader reader, Object from, Object to) { diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java index 09c593dc86..16d5e61584 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java @@ -87,6 +87,54 @@ class GlobalIndexEvaluatorTest { return new StubGlobalIndexReader(result); } + @Test + void testRangeFallbackAndOrBoundary() { + AtomicInteger rangeCalls = new AtomicInteger(); + GlobalIndexReader reader = + new StubGlobalIndexReader(null) { + @Override + public CompletableFuture<Optional<GlobalIndexResult>> visitRange( + FieldRef field, + Object from, + Object to, + boolean fromInclusive, + boolean toInclusive) { + rangeCalls.incrementAndGet(); + return super.visitRange(field, from, to, fromInclusive, toInclusive); + } + + @Override + public CompletableFuture<Optional<GlobalIndexResult>> visitGreaterOrEqual( + FieldRef field, Object literal) { + return CompletableFuture.completedFuture(Optional.of(resultOf(2, 3))); + } + }; + PredicateBuilder builder = new PredicateBuilder(rowType()); + try (GlobalIndexEvaluator evaluator = + new GlobalIndexEvaluator(rowType(), fieldId -> Collections.singletonList(reader))) { + assertBitmapContainsExactly( + evaluator + .evaluate( + PredicateBuilder.and( + builder.greaterOrEqual(0, 30), + builder.lessThan(0, 120))) + .get() + .results(), + 2L, + 3L); + assertThat(rangeCalls).hasValue(1); + assertThat( + evaluator.evaluate( + PredicateBuilder.or( + builder.greaterOrEqual(0, 30), + builder.lessThan(0, 120)))) + .isEmpty(); + evaluator.evaluate( + PredicateBuilder.and(builder.greaterOrEqual(0, 30), builder.lessThan(1, 120))); + assertThat(rangeCalls).hasValue(1); + } + } + @Test void testSingleFieldSequential() { RowType rowType = rowType(); 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 7cc643f3ec..af7d553979 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 @@ -20,21 +20,28 @@ package org.apache.paimon.globalindex.btree; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.globalindex.GlobalIndexEvaluator; import org.apache.paimon.globalindex.GlobalIndexIOMeta; import org.apache.paimon.globalindex.GlobalIndexReader; import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; import org.apache.paimon.globalindex.OffsetGlobalIndexReader; import org.apache.paimon.globalindex.ResultEntry; +import org.apache.paimon.globalindex.UnionGlobalIndexReader; import org.apache.paimon.globalindex.btree.BTreeIndexReader.KeyRowIds; import org.apache.paimon.globalindex.io.GlobalIndexFileReader; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.And; +import org.apache.paimon.predicate.CompoundPredicate; import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.predicate.TopN; import org.apache.paimon.testutils.junit.parameterized.ParameterizedTestExtension; import org.apache.paimon.types.DataField; import org.apache.paimon.types.IntType; +import org.apache.paimon.types.RowType; import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.SemaphoredDelegatingExecutor; @@ -66,6 +73,11 @@ import static org.apache.paimon.predicate.SortValue.SortDirection.ASCENDING; import static org.apache.paimon.predicate.SortValue.SortDirection.DESCENDING; import static org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; /** Test for {@link LazyFilteredBTreeReader} to read multiple files. */ @ExtendWith(ParameterizedTestExtension.class) @@ -96,6 +108,84 @@ public class LazyFilteredBTreeIndexReaderTest extends AbstractIndexReaderTest { return written; } + @TestTemplate + public void testFusedRangeThroughEvaluatorAndWrappers() throws Exception { + List<Pair<Object, Long>> values = new ArrayList<>(data); + values.add(Pair.of(null, (long) dataNum)); + List<GlobalIndexIOMeta> written = Collections.singletonList(writeData(values)); + RowType rowType = + new RowType(Collections.singletonList(new DataField(1, "testField", dataType))); + PredicateBuilder builder = new PredicateBuilder(rowType); + LazyFilteredBTreeReader indexReader = + spy( + (LazyFilteredBTreeReader) + globalIndexer.createReader( + fileReader, + written, + dataNum + 1, + newDirectExecutorService())); + BTreeIndexReader btreeReader = spy(indexReader.openReader(written.get(0))); + doReturn(btreeReader).when(indexReader).openReader(written.get(0)); + GlobalIndexReader wrapped = + new UnionGlobalIndexReader( + Collections.singletonList( + new OffsetGlobalIndexReader(indexReader, 1000L, 1000L + dataNum))); + try (GlobalIndexEvaluator evaluator = + new GlobalIndexEvaluator(rowType, fieldId -> Collections.singletonList(wrapped))) { + Object from = data.get(dataNum / 4).getKey(); + Object to = data.get(dataNum * 3 / 4).getKey(); + // Include empty and singleton intervals, and both predicate orders. + for (Object[] bounds : + Arrays.asList( + new Object[] {from, to}, + new Object[] {from, from}, + new Object[] {to, from})) { + for (boolean lowerInclusive : Arrays.asList(false, true)) { + for (boolean upperInclusive : Arrays.asList(false, true)) { + Predicate lower = + lowerInclusive + ? builder.greaterOrEqual(0, bounds[0]) + : builder.greaterThan(0, bounds[0]); + Predicate upper = + upperInclusive + ? builder.lessOrEqual(0, bounds[1]) + : builder.lessThan(0, bounds[1]); + for (List<Predicate> order : + Arrays.asList( + Arrays.asList(lower, upper), Arrays.asList(upper, lower))) { + clearInvocations(btreeReader); + GlobalIndexEvaluator.Evaluation result = + evaluator + .evaluateWithContributingFields( + new CompoundPredicate(And.INSTANCE, order)) + .get(); + List<Long> expected = new ArrayList<>(); + for (Pair<Object, Long> value : data) { + int low = comparator.compare(value.getKey(), bounds[0]); + int high = comparator.compare(value.getKey(), bounds[1]); + if ((low > 0 || lowerInclusive && low == 0) + && (high < 0 || upperInclusive && high == 0)) { + expected.add(value.getValue() + 1000L); + } + } + assertResult(result.result(), expected); + assertThat(result.contributingFieldIds()).containsExactly(1); + if (comparator.compare(bounds[0], bounds[1]) <= 0) { + verify(btreeReader) + .visitRange( + bounds[0], + bounds[1], + lowerInclusive, + upperInclusive); + } + verifyNoMoreInteractions(btreeReader); + } + } + } + } + } + } + private int firstStrictlyGreaterKeyIndex() { for (int i = 1; i < dataNum; i++) { if (comparator.compare(data.get(i - 1).getKey(), data.get(i).getKey()) < 0) {
