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 dd73b5c8de [core] Use a single BTree scan for bounded range predicates
(#9844)
dd73b5c8de is described below
commit dd73b5c8de722974380ac44dd5fc7464d903f066
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) {
