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 289e9265e1 [core] Support composite BTree prefix and range queries
(#10346)
289e9265e1 is described below
commit 289e9265e11cb2174e6a08985676db9f9b3749af
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 3 16:35:33 2026 +0800
[core] Support composite BTree prefix and range queries (#10346)
---
docs/docs/multimodal-table/global-index/btree.mdx | 36 ++-
.../globalindex/SortedFileGlobalIndexReader.java | 2 +-
.../btree/BTreeGlobalIndexerFactory.java | 28 +-
.../paimon/globalindex/btree/BTreeIndexReader.java | 33 +++
.../globalindex/btree/CompositeBTreePredicate.java | 310 ++++++++++++++++++---
.../globalindex/btree/LazyFilteredBTreeReader.java | 27 +-
.../java/org/apache/paimon/sst/BlockIterator.java | 8 +-
.../java/org/apache/paimon/sst/SstFileReader.java | 9 +-
.../globalindex/btree/CompositeBTreeIndexTest.java | 7 +-
.../btree/CompositeBTreePredicateTest.java | 90 +++---
.../globalindex/btree/CompositeBTreeRangeTest.java | 236 ++++++++++++++++
.../org/apache/paimon/sst/BlockIteratorTest.java | 21 ++
.../paimon/globalindex/DataEvolutionBatchScan.java | 3 +-
.../DataEvolutionGlobalIndexScanner.java | 6 +-
.../paimon/globalindex/GlobalIndexQuery.java | 188 ++++++++-----
.../paimon/globalindex/GlobalIndexQueryTest.java | 89 ++++--
.../paimon/table/CompositeBTreeTableTest.java | 165 ++++++++++-
.../paimon/flink/SortedGlobalIndexITCase.java | 11 +
.../CompositeBTreeIndexProcedureTest.scala | 9 +
19 files changed, 1067 insertions(+), 211 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index/btree.mdx
b/docs/docs/multimodal-table/global-index/btree.mdx
index 5a31c6b005..7da34552eb 100644
--- a/docs/docs/multimodal-table/global-index/btree.mdx
+++ b/docs/docs/multimodal-table/global-index/btree.mdx
@@ -165,18 +165,40 @@ The index stores typed tuples and one row-ID posting list
for each distinct tupl
Equality conditions on every key column use a composite point lookup,
regardless
of their order in the SQL predicate. This avoids expanding each column's
posting
list before intersecting them. Additional predicates remain row filters, and OR
-branches can each select a composite lookup. When several definitions match,
-selection prefers the one with more key columns.
+branches can each select a composite lookup.
-Eligible composite point queries honor `global-index.query-in-reader.enabled`.
+Composite indexes also support an equality prefix followed by a range on the
next
+key column. For an index on `(category, item_number)`, the following
conditions can
+use the index:
+
+| Condition | Index scan |
+| --- | --- |
+| `category = 'category-a'` | All tuples with this category prefix |
+| `category = 'category-a' AND item_number > 7` | A range within the category
prefix |
+| `category = 'category-a' AND item_number BETWEEN 7 AND 9` | A range with
inclusive endpoints within the prefix |
+| `category > 'category-a' AND item_number = 7` | A category range; the
item-number condition remains a data filter |
+
+The range may use `>`, `>=`, `<`, `<=` or `BETWEEN`. Conditions after the first
+range column remain data filters. A condition only on `item_number` cannot use
+this composite index's left prefix.
+
+When several composite definitions match, selection prefers more bounded key
+columns, then more equality columns, then fewer total key columns. If only the
+leading column is bounded, a dedicated single-column index is preferred when
+its coverage includes the composite definition's coverage.
+
+Eligible composite queries honor `global-index.query-in-reader.enabled`.
With reader evaluation enabled, supported index conditions narrow the candidate
rows and remaining conditions are evaluated against the data.
Composite indexes can coexist with single-column indexes on the same columns.
-Composite lookup currently requires equality on all key columns; conditions on
-only some key columns, ranges and IN predicates use available single-column
-indexes or ordinary scans. Vector and full-text search pre-filters use
-single-column indexes or their data fallback.
+Full-key equalities use point probes. Prefix and range scans use
+`btree-index.fallback-scan-max-size`, applied to the selected files in each
index
+group. If any group exceeds the budget, the whole composite definition is
+declined and planning uses other indexes or an ordinary scan. Point probes are
+unaffected by this budget. Dedicated composite `IN` and `IS NULL` optimization
is
+not supported yet. Vector and full-text search pre-filters use single-column
+indexes or their data fallback.
In `full` and `detail` modes, rows outside the selected composite index's
coverage
remain eligible for data filtering. Calling the procedure again with the same
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 3c4d80c441..71731aa07b 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
@@ -327,7 +327,7 @@ public abstract class SortedFileGlobalIndexReader<R extends
Closeable>
return visitSelectedFiles(selector.get(), visitor);
}
- private CompletableFuture<Optional<GlobalIndexResult>> visitSelectedFiles(
+ protected CompletableFuture<Optional<GlobalIndexResult>>
visitSelectedFiles(
Optional<List<GlobalIndexIOMeta>> selectedOpt,
Function<R, Optional<GlobalIndexResult>> visitor) {
if (!selectedOpt.isPresent()) {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexerFactory.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexerFactory.java
index 75ae2a762b..84b403b46e 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexerFactory.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/BTreeGlobalIndexerFactory.java
@@ -18,21 +18,15 @@
package org.apache.paimon.globalindex.btree;
-import org.apache.paimon.data.GenericRow;
-import org.apache.paimon.globalindex.CompositeKeySerializer;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexer;
import org.apache.paimon.globalindex.GlobalIndexerFactory;
import org.apache.paimon.globalindex.KeySerializer;
import org.apache.paimon.globalindex.SortedFileMetaSelector;
import org.apache.paimon.options.Options;
-import org.apache.paimon.predicate.FieldRef;
-import org.apache.paimon.predicate.LeafPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.types.DataField;
-import org.apache.paimon.types.RowType;
-import java.util.Collections;
import java.util.List;
import java.util.Optional;
@@ -50,25 +44,9 @@ public class BTreeGlobalIndexerFactory implements
GlobalIndexerFactory {
public List<GlobalIndexIOMeta> selectFiles(
List<DataField> indexFields, Predicate predicate,
List<GlobalIndexIOMeta> files) {
if (indexFields.size() > 1) {
- Optional<List<LeafPredicate>> matched =
- CompositeBTreePredicate.match(indexFields, predicate);
- if (!matched.isPresent()) {
- return files;
- }
- if (CompositeBTreePredicate.isContradictory(indexFields,
predicate)) {
- return Collections.emptyList();
- }
- if (files.stream().anyMatch(file -> file.metadata() == null)) {
- return files;
- }
- RowType keyType = new RowType(indexFields);
- KeySerializer serializer = new CompositeKeySerializer(keyType);
- Object[] values = matched.get().stream().map(leaf ->
leaf.literals().get(0)).toArray();
- return new SortedFileMetaSelector(files, serializer)
- .visitEqual(
- new FieldRef(0, indexFields.get(0).name(),
keyType),
- GenericRow.of(values))
- .orElse(files);
+ Optional<CompositeBTreePredicate.Plan> plan =
+ CompositeBTreePredicate.plan(indexFields, predicate);
+ return plan.map(value -> value.selectFiles(files)).orElse(files);
}
return SortedFileMetaSelector.selectFiles(
predicate, files,
KeySerializer.create(indexFields.get(0).type()));
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 174df07042..26584d4260 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
@@ -18,6 +18,7 @@
package org.apache.paimon.globalindex.btree;
+import org.apache.paimon.data.InternalRow;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
@@ -339,6 +340,38 @@ public class BTreeIndexReader implements Closeable {
return createResult(() -> pointQuery(literal));
}
+ public Optional<GlobalIndexResult>
visitComposite(CompositeBTreePredicate.Plan plan) {
+ return createResult(
+ () -> {
+ if (plan.isEmpty()) {
+ return new RoaringNavigableMap64();
+ }
+ if (plan.isPointLookup()) {
+ return pointQuery(plan.pointKey());
+ }
+ RoaringNavigableMap64 result = new RoaringNavigableMap64();
+ SstFileReader.SstFileIterator iterator =
reader.createIterator();
+ iterator.seekTo(
+ key ->
+ plan.lower()
+ .compareKey(
+ (InternalRow)
keySerializer.deserialize(key)));
+ BlockIterator batch;
+ while ((batch = iterator.readBatch()) != null) {
+ while (batch.hasNext()) {
+ Map.Entry<MemorySlice, MemorySlice> entry =
batch.next();
+ InternalRow key =
+ (InternalRow)
keySerializer.deserialize(entry.getKey());
+ if (plan.upper().compareKey(key) > 0) {
+ return result;
+ }
+ addRowIdsTo(entry.getValue(), result);
+ }
+ }
+ return result;
+ });
+ }
+
public Optional<GlobalIndexResult> visitGreaterThan(Object literal) {
return createResult(() -> rangeQuery(literal, maxKey, false, true));
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/CompositeBTreePredicate.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/CompositeBTreePredicate.java
index 02b71aec89..3acccc79a6 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/CompositeBTreePredicate.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/btree/CompositeBTreePredicate.java
@@ -18,71 +18,303 @@
package org.apache.paimon.globalindex.btree;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.globalindex.CompositeKeySerializer;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.KeySerializer;
+import org.apache.paimon.globalindex.SortedIndexFileMeta;
+import org.apache.paimon.memory.MemorySlice;
+import org.apache.paimon.predicate.Between;
import org.apache.paimon.predicate.Equal;
+import org.apache.paimon.predicate.GreaterOrEqual;
+import org.apache.paimon.predicate.GreaterThan;
import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.LessOrEqual;
+import org.apache.paimon.predicate.LessThan;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.RowType;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Optional;
-/** Matches a full conjunction of equalities in physical composite-key order.
*/
+/** An equality prefix followed by an optional range in physical composite-key
order. */
public final class CompositeBTreePredicate {
private CompositeBTreePredicate() {}
- /** SQL equalities with NULL, or distinct values for the same key field,
cannot match. */
- public static boolean isContradictory(List<DataField> fields, Predicate
predicate) {
- List<LeafPredicate> matched = match(fields, predicate).get();
- for (int i = 0; i < fields.size(); i++) {
- Object value = matched.get(i).literals().get(0);
- if (value == null) {
- return true;
+ public static Optional<Plan> plan(List<DataField> fields, Predicate
predicate) {
+ List<Predicate> conjuncts = PredicateBuilder.splitAnd(predicate);
+ List<Predicate> covered = new ArrayList<>();
+ List<Object> prefix = new ArrayList<>();
+ List<LeafPredicate> range = Collections.emptyList();
+ boolean empty = false;
+ for (DataField field : fields) {
+ List<LeafPredicate> column = new ArrayList<>();
+ LeafPredicate equality = null;
+ for (Predicate child : conjuncts) {
+ if (!(child instanceof LeafPredicate)) {
+ continue;
+ }
+ LeafPredicate leaf = (LeafPredicate) child;
+ if (!leaf.fieldRefOptional().isPresent()
+ ||
!field.name().equals(leaf.fieldRefOptional().get().name())
+ || !field.type().equalsIgnoreNullable(leaf.type())
+ || (!(leaf.function() instanceof Equal) &&
!isRange(leaf))) {
+ continue;
+ }
+ column.add(leaf);
+ if (leaf.function() instanceof Equal) {
+ equality = leaf;
+ }
+ }
+ if (equality == null) {
+ range = column;
+ covered.addAll(range);
+ break;
}
- Comparator<Object> comparator =
-
KeySerializer.create(fields.get(i).type()).createComparator();
- for (Predicate child : PredicateBuilder.splitAnd(predicate)) {
- if (child instanceof LeafPredicate) {
- LeafPredicate leaf = (LeafPredicate) child;
- if (leaf.function() instanceof Equal
- &&
leaf.fieldRefOptional().equals(matched.get(i).fieldRefOptional())) {
- Object other = leaf.literals().get(0);
- if (other == null || comparator.compare(value, other)
!= 0) {
- return true;
- }
+ Object value = equality.literals().get(0);
+ for (LeafPredicate leaf : column) {
+ // Validate all bounds before using a full-key point lookup or
a prefix.
+ empty |=
+ value == null || !leaf.function().test(leaf.type(),
value, leaf.literals());
+ }
+ prefix.add(value);
+ covered.addAll(column);
+ }
+ int equalColumns = prefix.size();
+ int boundColumns = equalColumns + (range.isEmpty() ? 0 : 1);
+ if (boundColumns == 0) {
+ return Optional.empty();
+ }
+ Object[] values = prefix.toArray();
+ Bound lower = new Bound(fields, values, false);
+ Bound upper = new Bound(fields, values, true);
+ if (!range.isEmpty()) {
+ // Comparisons cannot match a NULL component, even with an
unbounded lower endpoint.
+ lower = new Bound(fields, append(values, null), true);
+ for (LeafPredicate leaf : range) {
+ if (leaf.literals().contains(null)) {
+ empty = true;
+ continue;
+ }
+ if (leaf.function() instanceof GreaterThan
+ || leaf.function() instanceof GreaterOrEqual
+ || leaf.function() instanceof Between) {
+ Bound next =
+ new Bound(
+ fields,
+ append(values, leaf.literals().get(0)),
+ leaf.function() instanceof GreaterThan);
+ if (next.compareTo(lower) > 0) {
+ lower = next;
+ }
+ }
+ if (leaf.function() instanceof LessThan
+ || leaf.function() instanceof LessOrEqual
+ || leaf.function() instanceof Between) {
+ Object value = leaf.literals().get(leaf.function()
instanceof Between ? 1 : 0);
+ Bound next =
+ new Bound(
+ fields,
+ append(values, value),
+ !(leaf.function() instanceof LessThan));
+ if (next.compareTo(upper) < 0) {
+ upper = next;
}
}
}
}
- return false;
+ return Optional.of(
+ new Plan(
+ fields,
+ covered,
+ lower,
+ upper,
+ boundColumns,
+ equalColumns,
+ empty || lower.compareTo(upper) >= 0));
}
- public static Optional<List<LeafPredicate>> match(List<DataField> fields,
Predicate predicate) {
- List<Predicate> conjuncts = PredicateBuilder.splitAnd(predicate);
- List<LeafPredicate> matched = new ArrayList<>();
- for (DataField field : fields) {
- LeafPredicate equality = null;
- for (Predicate conjunct : conjuncts) {
- if (conjunct instanceof LeafPredicate) {
- LeafPredicate leaf = (LeafPredicate) conjunct;
- if (leaf.function() instanceof Equal
- && leaf.fieldRefOptional().isPresent()
- &&
field.name().equals(leaf.fieldRefOptional().get().name())
- && field.type().equalsIgnoreNullable(leaf.type()))
{
- equality = leaf;
- break;
- }
+ private static boolean isRange(LeafPredicate leaf) {
+ return leaf.function() instanceof GreaterThan
+ || leaf.function() instanceof GreaterOrEqual
+ || leaf.function() instanceof LessThan
+ || leaf.function() instanceof LessOrEqual
+ || leaf.function() instanceof Between;
+ }
+
+ private static Object[] append(Object[] prefix, Object value) {
+ Object[] result = Arrays.copyOf(prefix, prefix.length + 1);
+ result[prefix.length] = value;
+ return result;
+ }
+
+ /** A virtual boundary immediately before or after every key with the
specified prefix. */
+ public static final class Bound {
+ private final Object[] values;
+ private final boolean after;
+ private final InternalRow.FieldGetter[] getters;
+ private final Comparator<Object>[] comparators;
+
+ @SuppressWarnings("unchecked")
+ private Bound(List<DataField> fields, Object[] values, boolean after) {
+ this.values = values;
+ this.after = after;
+ getters = new InternalRow.FieldGetter[values.length];
+ comparators = new Comparator[values.length];
+ for (int i = 0; i < values.length; i++) {
+ getters[i] =
InternalRow.createFieldGetter(fields.get(i).type().copy(true), i);
+ comparators[i] =
KeySerializer.create(fields.get(i).type()).createComparator();
+ }
+ }
+
+ /** Compare a persisted key to this virtual endpoint. */
+ public int compareKey(InternalRow key) {
+ for (int i = 0; i < values.length; i++) {
+ int comparison = compare(i, getters[i].getFieldOrNull(key),
values[i]);
+ if (comparison != 0) {
+ return comparison;
}
}
- if (equality == null) {
- return Optional.empty();
+ return after ? -1 : 1;
+ }
+
+ private int compare(int position, Object left, Object right) {
+ return left == null
+ ? (right == null ? 0 : -1)
+ : right == null ? 1 : comparators[position].compare(left,
right);
+ }
+
+ private int compareTo(Bound other) {
+ int count = Math.min(values.length, other.values.length);
+ for (int i = 0; i < count; i++) {
+ int comparison = compare(i, values[i], other.values[i]);
+ if (comparison != 0) {
+ return comparison;
+ }
+ }
+ if (values.length != other.values.length) {
+ return values.length < other.values.length
+ ? (after ? 1 : -1)
+ : (other.after ? -1 : 1);
+ }
+ return Boolean.compare(after, other.after);
+ }
+ }
+
+ /** One tuple interval, with its covered query predicates and point-lookup
classification. */
+ public static final class Plan {
+ private final List<DataField> fields;
+ private final List<Predicate> predicates;
+ private final Bound lower;
+ private final Bound upper;
+ private final int boundColumns;
+ private final int equalColumns;
+ private final boolean empty;
+
+ private Plan(
+ List<DataField> fields,
+ List<Predicate> predicates,
+ Bound lower,
+ Bound upper,
+ int boundColumns,
+ int equalColumns,
+ boolean empty) {
+ this.fields = fields;
+ this.predicates = predicates;
+ this.lower = lower;
+ this.upper = upper;
+ this.boundColumns = boundColumns;
+ this.equalColumns = equalColumns;
+ this.empty = empty;
+ }
+
+ public List<Predicate> predicates() {
+ return predicates;
+ }
+
+ public Bound lower() {
+ return lower;
+ }
+
+ public Bound upper() {
+ return upper;
+ }
+
+ public int boundColumns() {
+ return boundColumns;
+ }
+
+ public int equalColumns() {
+ return equalColumns;
+ }
+
+ public boolean isEmpty() {
+ return empty;
+ }
+
+ public boolean isPointLookup() {
+ return equalColumns == fields.size();
+ }
+
+ public GenericRow pointKey() {
+ return GenericRow.of(lower.values);
+ }
+
+ /**
+ * Point probes are unrestricted; other scans use the existing
selected-file byte budget.
+ */
+ public boolean canScan(List<GlobalIndexIOMeta> selected, long budget) {
+ if (isPointLookup() || selected.isEmpty()) {
+ return true;
+ }
+ if (budget <= 0) {
+ return false;
+ }
+ for (GlobalIndexIOMeta file : selected) {
+ if (file.fileSize() > budget) {
+ return false;
+ }
+ budget -= file.fileSize();
+ }
+ return true;
+ }
+
+ public List<GlobalIndexIOMeta> selectFiles(List<GlobalIndexIOMeta>
files) {
+ if (empty) {
+ return Collections.emptyList();
+ }
+ // Preserve conservative pruning when metadata is unavailable.
+ if (files.stream().anyMatch(file -> file.metadata() == null)) {
+ return files;
+ }
+ CompositeKeySerializer serializer = new CompositeKeySerializer(new
RowType(fields));
+ List<GlobalIndexIOMeta> result = new ArrayList<>();
+ for (GlobalIndexIOMeta file : files) {
+ SortedIndexFileMeta meta =
SortedIndexFileMeta.deserialize(file.metadata());
+ if (meta.firstKey() == null
+ || meta.lastKey() == null
+ || (lower.compareKey(
+ (InternalRow)
+ serializer.deserialize(
+
MemorySlice.wrap(meta.lastKey())))
+ > 0
+ && upper.compareKey(
+ (InternalRow)
+ serializer.deserialize(
+
MemorySlice.wrap(meta.firstKey())))
+ < 0)) {
+ result.add(file);
+ }
}
- matched.add(equality);
+ return result;
}
- return Optional.of(matched);
}
}
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 b14cf376c5..6ec85d813e 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
@@ -18,7 +18,6 @@
package org.apache.paimon.globalindex.btree;
-import org.apache.paimon.data.GenericRow;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.GlobalIndexResult;
import org.apache.paimon.globalindex.KeySerializer;
@@ -28,7 +27,6 @@ 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.LeafPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.TopN;
import org.apache.paimon.types.DataField;
@@ -60,6 +58,8 @@ public class LazyFilteredBTreeReader extends
SortedFileGlobalIndexReader<BTreeIn
@Nullable private final RoaringNavigableMap64 rowIdFilter;
private final Comparator<Object> comparator;
private final long totalRowCount;
+ private final List<GlobalIndexIOMeta> indexFiles;
+ private final long fallbackScanMaxSize;
@Nullable private final Pair<Object, Object> fullRangeBounds;
public LazyFilteredBTreeReader(
@@ -77,6 +77,8 @@ public class LazyFilteredBTreeReader extends
SortedFileGlobalIndexReader<BTreeIn
this.fileReader = fileReader;
this.keySerializer = keySerializer;
this.indexFields = indexFields;
+ this.indexFiles = files;
+ this.fallbackScanMaxSize = fallbackScanMaxSize;
this.rowIdFilter =
rowRanges == null ? null :
GlobalIndexResult.fromRanges(rowRanges).results();
this.comparator = keySerializer.createComparator();
@@ -99,6 +101,9 @@ public class LazyFilteredBTreeReader extends
SortedFileGlobalIndexReader<BTreeIn
return null;
}
remaining -= file.rowCount();
+ if (file.metadata() == null) {
+ return null;
+ }
SortedIndexFileMeta meta =
SortedIndexFileMeta.deserialize(file.metadata());
if (meta.hasNulls() || meta.firstKey() == null || meta.lastKey()
== null) {
return null;
@@ -120,16 +125,20 @@ public class LazyFilteredBTreeReader extends
SortedFileGlobalIndexReader<BTreeIn
if (indexFields.size() < 2) {
return CompletableFuture.completedFuture(Optional.empty());
}
- Optional<List<LeafPredicate>> matched =
- CompositeBTreePredicate.match(indexFields, predicate);
- if (!matched.isPresent()) {
+ Optional<CompositeBTreePredicate.Plan> planned =
+ CompositeBTreePredicate.plan(indexFields, predicate);
+ if (!planned.isPresent()) {
return CompletableFuture.completedFuture(Optional.empty());
}
- if (CompositeBTreePredicate.isContradictory(indexFields, predicate)) {
- return
CompletableFuture.completedFuture(Optional.of(GlobalIndexResult.createEmpty()));
+ CompositeBTreePredicate.Plan plan = planned.get();
+ if (plan.isPointLookup() && !plan.isEmpty()) {
+ return visitEqual((FieldRef) null, plan.pointKey());
+ }
+ List<GlobalIndexIOMeta> selected = plan.selectFiles(indexFiles);
+ if (!plan.canScan(selected, fallbackScanMaxSize)) {
+ return CompletableFuture.completedFuture(Optional.empty());
}
- Object[] values = matched.get().stream().map(leaf ->
leaf.literals().get(0)).toArray();
- return visitEqual((FieldRef) null, GenericRow.of(values));
+ return visitSelectedFiles(Optional.of(selected), reader ->
reader.visitComposite(plan));
}
@Override
diff --git
a/paimon-common/src/main/java/org/apache/paimon/sst/BlockIterator.java
b/paimon-common/src/main/java/org/apache/paimon/sst/BlockIterator.java
index b477f8547d..861f892f81 100644
--- a/paimon-common/src/main/java/org/apache/paimon/sst/BlockIterator.java
+++ b/paimon-common/src/main/java/org/apache/paimon/sst/BlockIterator.java
@@ -24,6 +24,7 @@ import org.apache.paimon.memory.MemorySliceInput;
import java.util.Iterator;
import java.util.Map;
import java.util.NoSuchElementException;
+import java.util.function.ToIntFunction;
/** An {@link Iterator} for a block. */
public class BlockIterator implements Iterator<Map.Entry<MemorySlice,
MemorySlice>> {
@@ -64,6 +65,11 @@ public class BlockIterator implements
Iterator<Map.Entry<MemorySlice, MemorySlic
}
public boolean seekTo(MemorySlice targetKey) {
+ return seekTo(key -> reader.comparator().compare(key, targetKey));
+ }
+
+ /** Seek using a monotone comparison to a key or virtual prefix boundary.
*/
+ public boolean seekTo(ToIntFunction<MemorySlice> compareToTarget) {
int left = 0;
int right = reader.recordCount() - 1;
polledPosition = -1;
@@ -72,7 +78,7 @@ public class BlockIterator implements
Iterator<Map.Entry<MemorySlice, MemorySlic
while (left <= right) {
int mid = left + (right - left) / 2;
- int compare = reader.comparator().compare(readKey(mid), targetKey);
+ int compare = compareToTarget.applyAsInt(readKey(mid));
if (compare == 0) {
polledPosition = mid;
diff --git
a/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java
b/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java
index 8346464f3b..ea0575457e 100644
--- a/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java
+++ b/paimon-common/src/main/java/org/apache/paimon/sst/SstFileReader.java
@@ -32,6 +32,7 @@ import javax.annotation.Nullable;
import java.io.Closeable;
import java.io.IOException;
import java.util.Comparator;
+import java.util.function.ToIntFunction;
import static org.apache.paimon.sst.SstFileUtils.crc32c;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -279,15 +280,19 @@ public class SstFileReader implements Closeable {
*/
public void seekTo(byte[] key) {
MemorySlice keySlice = MemorySlice.wrap(key);
+ seekTo(candidate -> comparator.compare(candidate, keySlice));
+ }
- indexIterator.seekTo(keySlice);
+ /** Seek to a virtual boundary without encoding it as a persisted key.
*/
+ public void seekTo(ToIntFunction<MemorySlice> compareToTarget) {
+ indexIterator.seekTo(compareToTarget);
if (indexIterator.hasNext()) {
seekedDataBlock = getNextBlock(indexIterator);
// The index block entry key is the last key of the
corresponding data block.
// If there is some index entry key >= targetKey, the related
data block must
// also contain some key >= target key, which means
seekedDataBlock.hasNext()
// must be true
- seekedDataBlock.seekTo(keySlice);
+ seekedDataBlock.seekTo(compareToTarget);
Preconditions.checkState(seekedDataBlock.hasNext());
} else {
seekedDataBlock = null;
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreeIndexTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreeIndexTest.java
index 1c05959f13..b642583541 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreeIndexTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreeIndexTest.java
@@ -194,11 +194,8 @@ class CompositeBTreeIndexTest {
BinaryString.fromString("tag"));
for (Predicate unsupported :
Arrays.asList(
- builder.equal(0,
BinaryString.fromString("category-a")),
- PredicateBuilder.and(
- builder.equal(0,
BinaryString.fromString("category-a")),
- builder.greaterThan(1, 7),
- builder.equal(2,
BinaryString.fromString("tag"))),
+ builder.equal(1, 7),
+ builder.isNull(0),
PredicateBuilder.and(
builder.in(
0,
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreePredicateTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreePredicateTest.java
index 6172018d29..4a63cdd80a 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreePredicateTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreePredicateTest.java
@@ -25,7 +25,6 @@ import org.apache.paimon.globalindex.CompositeKeySerializer;
import org.apache.paimon.globalindex.GlobalIndexIOMeta;
import org.apache.paimon.globalindex.KeySerializer;
import org.apache.paimon.globalindex.SortedIndexFileMeta;
-import org.apache.paimon.predicate.LeafPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.types.DataField;
@@ -37,11 +36,10 @@ import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
-/** Complete tuple matching and conservative manifest metadata pruning. */
+/** Typed interval planning and conservative manifest metadata pruning. */
class CompositeBTreePredicateTest {
private final List<DataField> fields =
@@ -57,42 +55,36 @@ class CompositeBTreePredicateTest {
Predicate predicate =
PredicateBuilder.and(
builder.equal(1, -3), builder.equal(0,
BinaryString.fromString("a")));
- List<LeafPredicate> matched = CompositeBTreePredicate.match(fields,
predicate).get();
+ assertThat(CompositeBTreePredicate.plan(fields,
predicate).get().pointKey())
+ .isEqualTo(GenericRow.of(BinaryString.fromString("a"), -3));
+ assertThat(CompositeBTreePredicate.plan(fields,
predicate).get().isEmpty()).isFalse();
assertThat(
- matched.stream()
- .map(leaf -> leaf.literals().get(0))
- .collect(Collectors.toList()))
- .containsExactly(BinaryString.fromString("a"), -3);
- assertThat(CompositeBTreePredicate.isContradictory(fields,
predicate)).isFalse();
- assertThat(
- CompositeBTreePredicate.match(
+ CompositeBTreePredicate.plan(
Arrays.asList(fields.get(1),
fields.get(0)), predicate)
- .get().stream()
- .map(leaf -> leaf.literals().get(0))
- .collect(Collectors.toList()))
- .containsExactly(-3, BinaryString.fromString("a"));
+ .get()
+ .pointKey())
+ .isEqualTo(GenericRow.of(-3, BinaryString.fromString("a")));
}
@Test
- void testRequiresCompleteConjunctiveEqualitiesAndMatchingTypes() {
- assertThat(
- CompositeBTreePredicate.match(
- fields, builder.equal(0,
BinaryString.fromString("a"))))
- .isEmpty();
+ void testRequiresLeadingConstraintAndMatchingTypes() {
+ assertThat(CompositeBTreePredicate.plan(fields, builder.equal(1,
7))).isEmpty();
assertThat(
- CompositeBTreePredicate.match(
+ CompositeBTreePredicate.plan(
fields,
PredicateBuilder.or(
builder.equal(0,
BinaryString.fromString("a")),
builder.equal(1, 7))))
.isEmpty();
assertThat(
- CompositeBTreePredicate.match(
- fields,
- PredicateBuilder.and(
- builder.equal(0,
BinaryString.fromString("a")),
- builder.greaterThan(1, 7))))
- .isEmpty();
+ CompositeBTreePredicate.plan(
+ fields,
+ PredicateBuilder.and(
+ builder.equal(0,
BinaryString.fromString("a")),
+ builder.greaterThan(1, 7)))
+ .get()
+ .boundColumns())
+ .isEqualTo(2);
PredicateBuilder incompatible =
new PredicateBuilder(
new RowType(
@@ -100,7 +92,7 @@ class CompositeBTreePredicateTest {
new DataField(10, "category",
DataTypes.INT()),
fields.get(1))));
assertThat(
- CompositeBTreePredicate.match(
+ CompositeBTreePredicate.plan(
fields,
PredicateBuilder.and(
incompatible.equal(0, 1),
incompatible.equal(1, 7))))
@@ -134,16 +126,16 @@ class CompositeBTreePredicateTest {
PredicateBuilder.and(
builder.equal(0, BinaryString.fromString("a")),
builder.equal(1, null)))) {
- assertThat(CompositeBTreePredicate.isContradictory(fields,
predicate)).isTrue();
+ assertThat(CompositeBTreePredicate.plan(fields,
predicate).get().isEmpty()).isTrue();
assertThat(factory.selectFiles(fields, predicate,
files)).isEmpty();
}
Predicate redundant = PredicateBuilder.and(equal("a", 7),
builder.equal(1, 7));
- assertThat(CompositeBTreePredicate.isContradictory(fields,
redundant)).isFalse();
+ assertThat(CompositeBTreePredicate.plan(fields,
redundant).get().isEmpty()).isFalse();
assertThat(factory.selectFiles(fields, redundant,
files)).containsExactlyElementsOf(files);
}
@Test
- void testRetainsFilesWhenCompleteKeyOrMetadataIsUnavailable() {
+ void testRetainsFilesWhenLeadingConstraintOrMetadataIsUnavailable() {
GlobalIndexIOMeta file = file("first", "a", -3, "a", 7);
List<GlobalIndexIOMeta> files = Collections.singletonList(file);
assertThat(factory.selectFiles(fields, builder.equal(1, 100),
files)).isSameAs(files);
@@ -154,7 +146,7 @@ class CompositeBTreePredicateTest {
builder.equal(0,
BinaryString.fromString("a")),
builder.greaterThan(1, 100)),
files))
- .isSameAs(files);
+ .isEmpty();
List<GlobalIndexIOMeta> unknown =
Arrays.asList(file, new GlobalIndexIOMeta(new Path("unknown"),
1, null));
assertThat(factory.selectFiles(fields, equal("z", 7),
unknown)).isSameAs(unknown);
@@ -166,6 +158,40 @@ class CompositeBTreePredicateTest {
.isEmpty();
}
+ @Test
+ void testPrefixAndRangePruneTypedEndpoints() {
+ GlobalIndexIOMeta first = file("first", "a", -3, "a", 7);
+ GlobalIndexIOMeta second = file("second", "a", 10, "b", 2);
+ GlobalIndexIOMeta third = file("third", "b", 3, "c", 4);
+ List<GlobalIndexIOMeta> files = Arrays.asList(first, second, third);
+ Predicate prefix = builder.equal(0, BinaryString.fromString("a"));
+ assertThat(factory.selectFiles(fields, prefix,
files)).containsExactly(first, second);
+ assertThat(
+ factory.selectFiles(
+ fields,
+ PredicateBuilder.and(prefix,
builder.greaterThan(1, 7)),
+ files))
+ .containsExactly(second);
+ assertThat(
+ factory.selectFiles(
+ fields,
+ PredicateBuilder.and(prefix,
builder.greaterOrEqual(1, 7)),
+ files))
+ .containsExactly(first, second);
+ assertThat(
+ factory.selectFiles(
+ fields,
+ PredicateBuilder.and(prefix,
builder.lessThan(1, -3)),
+ files))
+ .isEmpty();
+ assertThat(
+ factory.selectFiles(
+ fields,
+ PredicateBuilder.and(prefix,
builder.lessOrEqual(1, -3)),
+ files))
+ .containsExactly(first);
+ }
+
private Predicate equal(String category, int number) {
return PredicateBuilder.and(
builder.equal(1, number), builder.equal(0,
BinaryString.fromString(category)));
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreeRangeTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreeRangeTest.java
new file mode 100644
index 0000000000..58a1602851
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/CompositeBTreeRangeTest.java
@@ -0,0 +1,236 @@
+/*
+ * 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.globalindex.btree;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.globalindex.CompositeKeySerializer;
+import org.apache.paimon.globalindex.GlobalIndexIOMeta;
+import org.apache.paimon.globalindex.GlobalIndexReader;
+import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter;
+import org.apache.paimon.globalindex.GlobalIndexer;
+import org.apache.paimon.globalindex.ResultEntry;
+import org.apache.paimon.globalindex.io.GlobalIndexFileWriter;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Range;
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+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.ValueSource;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+
+import static
org.apache.paimon.shade.guava30.com.google.common.util.concurrent.MoreExecutors.newDirectExecutorService;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Prefix and range seeks must include entire boundary suffixes across SST
blocks. */
+class CompositeBTreeRangeTest {
+
+ @TempDir java.nio.file.Path tempPath;
+
+ private final RowType type =
+ RowType.of(
+ new DataField(10, "category", DataTypes.INT()),
+ new DataField(20, "item_number", DataTypes.INT()),
+ new DataField(30, "suffix", DataTypes.INT()));
+ private final PredicateBuilder builder = new PredicateBuilder(type);
+ private final List<GenericRow> rows = new ArrayList<>();
+ private final LocalFileIO io = LocalFileIO.create();
+
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2})
+ void testPrefixAndRangeSeeksAcrossBlocks(int version) throws Exception {
+ Options options = options(version);
+ List<GlobalIndexIOMeta> files = write(options);
+ List<Predicate> queries =
+ Arrays.asList(
+ builder.equal(0, 1),
+ PredicateBuilder.and(builder.equal(0, 1),
builder.equal(1, 7)),
+ PredicateBuilder.and(builder.equal(0, 1),
builder.greaterThan(1, 7)),
+ PredicateBuilder.and(builder.equal(0, 1),
builder.greaterOrEqual(1, 7)),
+ PredicateBuilder.and(builder.equal(0, 1),
builder.lessThan(1, 7)),
+ PredicateBuilder.and(builder.equal(0, 1),
builder.lessOrEqual(1, 7)),
+ PredicateBuilder.and(builder.equal(0, 1),
builder.between(1, -1, 7)),
+ PredicateBuilder.and(
+ builder.equal(0, 1),
+ builder.greaterThan(1, -1),
+ builder.lessThan(1, 7)),
+ PredicateBuilder.and(
+ builder.equal(0, 1),
+ builder.greaterOrEqual(1, -1),
+ builder.lessOrEqual(1, 7)),
+ builder.greaterThan(0, 0),
+ builder.lessOrEqual(0, 0),
+ builder.between(0, -1, 1),
+ PredicateBuilder.and(
+ builder.equal(0, 1), builder.greaterThan(1,
Integer.MAX_VALUE)),
+ PredicateBuilder.and(
+ builder.equal(0, 1), builder.lessThan(1,
Integer.MIN_VALUE)),
+ PredicateBuilder.and(
+ builder.equal(0, 1),
+ builder.greaterThan(1, 7),
+ builder.lessOrEqual(1, 7)),
+ PredicateBuilder.and(
+ builder.equal(0, 1), builder.equal(1, 7),
builder.lessThan(1, 7)),
+ PredicateBuilder.and(builder.equal(0, null),
builder.greaterThan(1, 7)),
+ PredicateBuilder.and(builder.equal(0, 1),
builder.lessThan(1, null)));
+ ExecutorService executor = newDirectExecutorService();
+ try {
+ for (List<Range> ranges :
+ Arrays.asList(null, Collections.singletonList(new
Range(20, 150)))) {
+ try (GlobalIndexReader reader =
+ GlobalIndexer.create("btree", type.getFields(),
options)
+ .createReader(
+ file ->
io.newInputStream(file.filePath()),
+ files,
+ rows.size(),
+ ranges,
+ executor)) {
+ for (Predicate query : queries) {
+ RoaringNavigableMap64 expected = new
RoaringNavigableMap64();
+ for (int i = 0; i < rows.size(); i++) {
+ if ((ranges == null || (i >= 20 && i <= 150))
+ && query.test(rows.get(i))) {
+ expected.add(i);
+ }
+ }
+
assertThat(reader.visitComposite(query).join()).isPresent();
+ assertThat(
+ reader.visitComposite(query)
+ .join()
+ .get()
+ .results()
+ .toRangeList())
+ .as("%s, local ranges %s", query, ranges)
+
.containsExactlyElementsOf(expected.toRangeList());
+ }
+ }
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testBudgetFallbackDoesNotDisablePointLookupsOrProvenEmptyRanges()
throws Exception {
+ Options options = options(2);
+ List<GlobalIndexIOMeta> files = write(options);
+ options.set(BTreeIndexOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE,
MemorySize.ofBytes(0));
+ ExecutorService executor = newDirectExecutorService();
+ try (GlobalIndexReader reader =
+ GlobalIndexer.create("btree", type.getFields(), options)
+ .createReader(
+ file -> io.newInputStream(file.filePath()),
+ files,
+ rows.size(),
+ null,
+ executor)) {
+ assertThat(reader.visitComposite(builder.equal(0,
1)).join()).isEmpty();
+ Predicate point =
+ PredicateBuilder.and(
+ builder.equal(0, 1), builder.equal(1, 7),
builder.equal(2, 0));
+
assertThat(reader.visitComposite(point).join().get().results().isEmpty()).isFalse();
+ assertThat(
+ reader.visitComposite(builder.greaterThan(0,
Integer.MAX_VALUE))
+ .join()
+ .get()
+ .results()
+ .isEmpty())
+ .isTrue();
+ for (Predicate unsupported :
+ Arrays.asList(
+ builder.equal(1, 7),
+ builder.isNull(0),
+ builder.in(0, Arrays.asList(-1, 0, 1)))) {
+
assertThat(reader.visitComposite(unsupported).join()).isEmpty();
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ private Options options(int version) {
+ Options options = new Options();
+ options.set(BTreeIndexOptions.BTREE_INDEX_FILE_VERSION, version);
+ options.set(BTreeIndexOptions.BTREE_INDEX_BLOCK_SIZE,
MemorySize.ofBytes(128));
+ options.set(BTreeIndexOptions.BTREE_INDEX_COMPRESSION, "lz4");
+ options.set(BTreeIndexOptions.BTREE_INDEX_BLOOM_FILTER_ENABLED, true);
+ return options;
+ }
+
+ private List<GlobalIndexIOMeta> write(Options options) throws Exception {
+ for (Integer category :
+ Arrays.asList(null, Integer.MIN_VALUE, -1, 0, 1,
Integer.MAX_VALUE)) {
+ for (Integer number :
+ Arrays.asList(null, Integer.MIN_VALUE, -1, 0, 7, 8,
Integer.MAX_VALUE)) {
+ for (int suffix = 0; suffix < 5; suffix++) {
+ rows.add(GenericRow.of(category, number, suffix));
+ }
+ }
+ }
+ rows.add(GenericRow.of(1, 7, 0));
+ rows.add(GenericRow.of(1, 7, null));
+ CompositeKeySerializer serializer = new CompositeKeySerializer(type);
+ rows.sort((left, right) -> serializer.createComparator().compare(left,
right));
+ Path directory = new Path(tempPath.toUri());
+ GlobalIndexFileWriter files =
+ new GlobalIndexFileWriter() {
+ @Override
+ public String newFileName(String prefix) {
+ return prefix + UUID.randomUUID();
+ }
+
+ @Override
+ public PositionOutputStream newOutputStream(String name)
throws IOException {
+ return io.newOutputStream(new Path(directory, name),
false);
+ }
+ };
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
+ GlobalIndexer.create("btree", type.getFields(),
options)
+ .createWriter(files);
+ for (int i = 0; i < rows.size(); i++) {
+ writer.write(rows.get(i), i);
+ }
+ List<GlobalIndexIOMeta> result = new ArrayList<>();
+ for (ResultEntry entry : writer.finish()) {
+ Path path = new Path(directory, entry.fileName());
+ result.add(
+ new GlobalIndexIOMeta(
+ path, io.getFileSize(path), entry.rowCount(),
entry.meta()));
+ }
+ return result;
+ }
+}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/sst/BlockIteratorTest.java
b/paimon-common/src/test/java/org/apache/paimon/sst/BlockIteratorTest.java
index f40b464399..47c7e6573d 100644
--- a/paimon-common/src/test/java/org/apache/paimon/sst/BlockIteratorTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/sst/BlockIteratorTest.java
@@ -45,6 +45,27 @@ public class BlockIteratorTest {
innerTest(false);
}
+ @Test
+ void testSeekToVirtualPrefixBoundaries() throws IOException {
+ BlockIterator iterator = BlockReader.create(writeBlock(true),
COMPARATOR).iterator();
+ iterator.seekTo(
+ key -> {
+ int comparison = Integer.compare(key.readInt(0) / 10, 3);
+ return comparison == 0 ? 1 : comparison;
+ });
+ Assertions.assertEquals(30, iterator.next().getKey().readInt(0));
+ iterator.seekTo(
+ key -> {
+ int comparison = Integer.compare(key.readInt(0) / 10, 3);
+ return comparison == 0 ? -1 : comparison;
+ });
+ Assertions.assertEquals(40, iterator.next().getKey().readInt(0));
+ iterator.seekTo(key -> -1);
+ Assertions.assertFalse(iterator.hasNext());
+ iterator.seekTo(key -> 1);
+ Assertions.assertEquals(0, iterator.next().getKey().readInt(0));
+ }
+
public void innerTest(boolean aligned) throws IOException {
MemorySlice data = writeBlock(aligned);
BlockReader reader = BlockReader.create(data, COMPARATOR);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
index a1cdc71ebe..4fdae682e5 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java
@@ -388,7 +388,8 @@ public class DataEvolutionBatchScan implements
DataTableScan {
table.rowType(),
indexFilter,
indexFiles,
- table.store().pathFactory().globalIndexFileFactory());
+ table.store().pathFactory().globalIndexFileFactory(),
+ table.coreOptions().toConfiguration());
if (indexQuery == null) {
return dataPlan;
}
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 a6ef515ceb..e294cc09d1 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
@@ -408,7 +408,8 @@ public class DataEvolutionGlobalIndexScanner implements
Closeable {
.noneMatch(
DataEvolutionGlobalIndexScanner::isMultiFieldIndex)
? null
- : GlobalIndexQuery.create(rowType, predicate,
indexFiles, indexPathFactory);
+ : GlobalIndexQuery.create(
+ rowType, predicate, indexFiles,
indexPathFactory, options);
if (query != null && query.hasCompositeQuery()) {
try {
return query.evaluateWithCoverage(
@@ -443,7 +444,8 @@ public class DataEvolutionGlobalIndexScanner implements
Closeable {
GlobalIndexQuery query =
predicate == null
? null
- : GlobalIndexQuery.create(rowType, predicate,
indexFiles, indexPathFactory);
+ : GlobalIndexQuery.create(
+ rowType, predicate, indexFiles,
indexPathFactory, options);
if (query != null && query.hasCompositeQuery() &&
query.hasScalarQuery()) {
// Scalar shards can decline a predicate at runtime, so the legacy
API must use
// evaluated coverage as well. Internal callers retain the
evaluation directly.
diff --git
a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexQuery.java
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexQuery.java
index 18737d69f9..f962282552 100644
---
a/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexQuery.java
+++
b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexQuery.java
@@ -22,6 +22,7 @@ import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import
org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner.IndexMetaFileGroup;
import org.apache.paimon.globalindex.GlobalIndexEvaluator.Evaluation;
+import org.apache.paimon.globalindex.btree.BTreeIndexOptions;
import org.apache.paimon.globalindex.btree.CompositeBTreePredicate;
import org.apache.paimon.index.IndexFileMeta;
import org.apache.paimon.index.IndexPathFactory;
@@ -30,7 +31,6 @@ import org.apache.paimon.io.DataOutputView;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.And;
import org.apache.paimon.predicate.CompoundPredicate;
-import org.apache.paimon.predicate.Equal;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.GreaterOrEqual;
import org.apache.paimon.predicate.LeafPredicate;
@@ -106,7 +106,8 @@ class GlobalIndexQuery {
RowType rowType,
Predicate predicate,
List<IndexFileMeta> files,
- IndexPathFactory pathFactory) {
+ IndexPathFactory pathFactory,
+ Options options) {
Map<Integer, List<IndexMetaFileGroup>> groupsByField =
DataEvolutionGlobalIndexScanner.groupIndexFiles(
files.stream()
@@ -124,90 +125,54 @@ class GlobalIndexQuery {
}
groups.put(fieldId, indexGroups);
});
- return createForPredicate(predicate, rowType, groups);
+ return createForPredicate(predicate, rowType, groups, options);
}
@Nullable
private static GlobalIndexQuery createForPredicate(
- Predicate predicate, RowType rowType, Map<Integer,
List<IndexGroup>> groups) {
+ Predicate predicate,
+ RowType rowType,
+ Map<Integer, List<IndexGroup>> groups,
+ Options options) {
if (predicate instanceof LeafPredicate) {
LeafPredicate leaf = (LeafPredicate) predicate;
Optional<FieldRef> field = leaf.fieldRefOptional();
if (!field.isPresent()) {
return null;
}
- return createForIndexedField(leaf, field.get(), rowType, groups);
+ CompositeCandidate composite = selectComposite(leaf, groups,
options);
+ return composite == null
+ ? createForIndexedField(leaf, field.get(), rowType, groups)
+ : composite.query();
}
CompoundPredicate compound = (CompoundPredicate) predicate;
boolean union = compound.function() instanceof Or;
List<GlobalIndexQuery> children = new ArrayList<>();
List<Predicate> predicates =
GlobalIndexEvaluator.normalizedChildren(compound);
if (!union) {
- // Prefer the longest full composite key before any single-column
posting is read.
while (!predicates.isEmpty()) {
- IndexGroup selected = null;
- List<LeafPredicate> matched = null;
- Predicate conjunction = PredicateBuilder.and(predicates);
- for (List<IndexGroup> fieldGroups : groups.values()) {
- for (IndexGroup group : fieldGroups) {
- if (!group.isCompositeBTree()
- || (selected != null
- && group.extraFields.size()
- <=
selected.extraFields.size())) {
- continue;
- }
- Optional<List<LeafPredicate>> match =
-
CompositeBTreePredicate.match(group.indexFields(), conjunction);
- if (match.isPresent()) {
- selected = group;
- matched = match.get();
- }
- }
- }
+ CompositeCandidate selected =
+ selectComposite(PredicateBuilder.and(predicates),
groups, options);
if (selected == null) {
break;
}
- List<Predicate> covered = new ArrayList<>();
- for (Predicate child : predicates) {
- if (child instanceof LeafPredicate) {
- LeafPredicate leaf = (LeafPredicate) child;
- if (leaf.function() instanceof Equal
- && matched.stream()
- .anyMatch(
- equality ->
-
equality.fieldRefOptional()
-
.equals(leaf.fieldRefOptional()))) {
- covered.add(child);
- }
- }
- }
- Predicate composite = PredicateBuilder.and(covered);
- List<IndexGroup> selectedGroups = new ArrayList<>();
- for (IndexGroup group : groups.get(selected.field.id())) {
- if (group.type.equals(selected.type)
- &&
group.indexFields().equals(selected.indexFields())) {
- selectedGroups.add(group.selectFiles(composite));
- }
- }
- children.add(
- new GlobalIndexQuery(
- composite, false, Collections.emptyList(),
selectedGroups));
- predicates.removeAll(covered);
- // The tuple equality already narrows these columns to one
value. Leave their
- // remaining conditions as data filters instead of reading
scalar postings.
- List<LeafPredicate> keyEqualities = matched;
+ children.add(selected.query());
+ predicates.removeAll(selected.plan.predicates());
+ // Later key conditions remain data filters; do not expand
scalar postings for them.
+ Set<String> keyColumns =
+ selected.group.indexFields().stream()
+ .map(DataField::name)
+ .collect(Collectors.toSet());
predicates.removeIf(
- child -> {
- if (!(child instanceof LeafPredicate)) {
- return false;
- }
- Optional<FieldRef> field = ((LeafPredicate)
child).fieldRefOptional();
- return keyEqualities.stream()
- .anyMatch(
- equality ->
equality.fieldRefOptional().equals(field));
- });
+ child ->
+ child instanceof LeafPredicate
+ && ((LeafPredicate) child)
+ .fieldRefOptional()
+ .filter(field ->
keyColumns.contains(field.name()))
+ .isPresent());
}
}
+
for (int i = 0; i < predicates.size(); i++) {
Predicate child = predicates.get(i);
GlobalIndexQuery query = null;
@@ -238,7 +203,7 @@ class GlobalIndexQuery {
}
}
if (query == null) {
- query = createForPredicate(child, rowType, groups);
+ query = createForPredicate(child, rowType, groups, options);
}
if (query == null) {
if (union) {
@@ -253,6 +218,101 @@ class GlobalIndexQuery {
: new GlobalIndexQuery(null, union, children,
Collections.emptyList());
}
+ @Nullable
+ private static CompositeCandidate selectComposite(
+ Predicate predicate, Map<Integer, List<IndexGroup>> groups,
Options options) {
+ CompositeCandidate selected = null;
+ Set<List<DataField>> seen = new HashSet<>();
+ long budget =
options.get(BTreeIndexOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE).getBytes();
+ for (List<IndexGroup> fieldGroups : groups.values()) {
+ for (IndexGroup group : fieldGroups) {
+ if (!group.isCompositeBTree() ||
!seen.add(group.indexFields())) {
+ continue;
+ }
+ Optional<CompositeBTreePredicate.Plan> planned =
+ CompositeBTreePredicate.plan(group.indexFields(),
predicate);
+ if (!planned.isPresent()) {
+ continue;
+ }
+ CompositeBTreePredicate.Plan plan = planned.get();
+ List<IndexGroup> definition =
+ groups.get(group.field.id()).stream()
+ .filter(
+ other ->
+ other.type.equals(group.type)
+ && other.indexFields()
+
.equals(group.indexFields()))
+ .collect(Collectors.toList());
+ Predicate covered = PredicateBuilder.and(plan.predicates());
+ List<IndexGroup> selectedGroups =
+ definition.stream()
+ .map(part -> part.selectFiles(covered))
+ .collect(Collectors.toList());
+ // Decline the entire definition when any shard exceeds the
same reader budget.
+ // Selecting only some shards would change FAST coverage
between eager and reader
+ // queries.
+ if (selectedGroups.stream().anyMatch(part ->
!plan.canScan(part.files, budget))) {
+ continue;
+ }
+ if (plan.boundColumns() == 1) {
+ List<Range> scalarCoverage =
+ Range.sortAndMergeOverlap(
+ fieldGroups.stream()
+ .filter(other ->
other.extraFields.isEmpty())
+ .map(other -> other.range)
+ .collect(Collectors.toList()),
+ true);
+ List<Range> compositeCoverage =
+ Range.sortAndMergeOverlap(
+ definition.stream()
+ .map(other -> other.range)
+ .collect(Collectors.toList()),
+ true);
+ // A dedicated leading-field lookup avoids expanding the
tuple suffix.
+ if (Range.and(compositeCoverage,
scalarCoverage).equals(compositeCoverage)) {
+ continue;
+ }
+ }
+ CompositeCandidate candidate = new CompositeCandidate(group,
plan, selectedGroups);
+ if (selected == null || candidate.preferredTo(selected)) {
+ selected = candidate;
+ }
+ }
+ }
+ return selected;
+ }
+
+ private static class CompositeCandidate {
+ private final IndexGroup group;
+ private final CompositeBTreePredicate.Plan plan;
+ private final List<IndexGroup> groups;
+
+ private CompositeCandidate(
+ IndexGroup group, CompositeBTreePredicate.Plan plan,
List<IndexGroup> groups) {
+ this.group = group;
+ this.plan = plan;
+ this.groups = groups;
+ }
+
+ private boolean preferredTo(CompositeCandidate other) {
+ if (plan.boundColumns() != other.plan.boundColumns()) {
+ return plan.boundColumns() > other.plan.boundColumns();
+ }
+ if (plan.equalColumns() != other.plan.equalColumns()) {
+ return plan.equalColumns() > other.plan.equalColumns();
+ }
+ return group.indexFields().size() <
other.group.indexFields().size();
+ }
+
+ private GlobalIndexQuery query() {
+ return new GlobalIndexQuery(
+ PredicateBuilder.and(plan.predicates()),
+ false,
+ Collections.emptyList(),
+ groups);
+ }
+ }
+
@Nullable
private static GlobalIndexQuery createForIndexedField(
Predicate predicate,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexQueryTest.java
b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexQueryTest.java
index 79727b7b3e..c315a73b20 100644
---
a/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexQueryTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexQueryTest.java
@@ -124,7 +124,8 @@ class GlobalIndexQueryTest {
builder.equal(0, 25),
PredicateBuilder.and(
builder.greaterThan(0, 15),
builder.lessThan(0, 35)))) {
- GlobalIndexQuery query = GlobalIndexQuery.create(rowType,
predicate, files, paths);
+ GlobalIndexQuery query =
+ GlobalIndexQuery.create(rowType, predicate, files, paths,
new Options());
assertThat(query).isNotNull();
DataOutputSerializer out = new DataOutputSerializer(256);
query.forRanges(Collections.singletonList(new Range(0,
99))).serialize(out);
@@ -152,7 +153,8 @@ class GlobalIndexQueryTest {
invocation ->
new
Path(invocation.<IndexFileMeta>getArgument(0).fileName()));
Options options = new Options();
- GlobalIndexQuery plan = GlobalIndexQuery.create(rowType, range, files,
paths);
+ GlobalIndexQuery plan =
+ GlobalIndexQuery.create(rowType, range, files, paths, new
Options());
assertThat(plan).isNotNull();
IndexQuerySplit split =
new IndexQuerySplit(dataSplit(), plan, options.toMap(),
Collections.emptyList());
@@ -189,7 +191,11 @@ class GlobalIndexQueryTest {
assertThat(
GlobalIndexQuery.create(
- rowType, PredicateBuilder.or(lower, upper),
files, paths))
+ rowType,
+ PredicateBuilder.or(lower, upper),
+ files,
+ paths,
+ new Options()))
.isNotNull();
}
@@ -249,7 +255,8 @@ class GlobalIndexQueryTest {
Arrays.asList(b.isNotNull(0), b.notEqual(0, 1), b.notIn(0,
Arrays.asList(1, 3)));
for (int i = 0; i < predicates.size(); i++) {
GlobalIndexQuery plan =
- GlobalIndexQuery.create(rowType, predicates.get(i), files,
paths);
+ GlobalIndexQuery.create(
+ rowType, predicates.get(i), files, paths, new
Options());
List<Range> ranges = Collections.singletonList(new Range(100,
102));
assertThat(plan).isNotNull();
assertThat(plan.evaluate(fileIO, new Options(),
ranges).results().toRangeList())
@@ -608,7 +615,8 @@ class GlobalIndexQueryTest {
rowType,
PredicateBuilder.and(supportedLeaf, unsupportedLeaf),
files,
- paths);
+ paths,
+ new Options());
assertThat(andPlan).isNotNull();
assertThat(andPlan.contributingFieldIds(rowType)).containsExactlyInAnyOrder(0,
1);
IndexQuerySplit split =
@@ -620,20 +628,22 @@ class GlobalIndexQueryTest {
rowType,
PredicateBuilder.or(supportedLeaf,
unsupportedLeaf),
files,
- paths))
+ paths,
+ new Options()))
.isNotNull();
assertThat(
GlobalIndexQuery.create(
rowType,
supportedLeaf,
Arrays.asList(supported, indexFile("fm", "fm",
0, 99, 0, null)),
- paths))
+ paths,
+ new Options()))
.isNotNull();
}
@ParameterizedTest
@ValueSource(strings = {"btree", "custom-index"})
- void testScalarPlansUseOnlySingleFieldDefinitions(String indexType) {
+ void testCompositePrefixEligibilityIsSeparateFromScalarDefinitions(String
indexType) {
RowType rowType = RowType.of(DataTypes.INT(), DataTypes.INT());
PredicateBuilder builder = new PredicateBuilder(rowType);
IndexFileMeta multi = indexFile(indexType, "multi", 0, 99, 0, new
int[] {1});
@@ -644,10 +654,19 @@ class GlobalIndexQueryTest {
new
Path(invocation.<IndexFileMeta>getArgument(0).fileName()));
for (int field : new int[] {0, 1}) {
Predicate predicate = builder.equal(field, 7);
- assertThat(
- GlobalIndexQuery.create(
- rowType, predicate,
Collections.singletonList(multi), paths))
- .isNull();
+ GlobalIndexQuery prefix =
+ GlobalIndexQuery.create(
+ rowType,
+ predicate,
+ Collections.singletonList(multi),
+ paths,
+ new Options());
+ if (field == 0 && "btree".equals(indexType)) {
+ assertThat(prefix).isNotNull();
+ assertThat(prefix.hasCompositeQuery()).isTrue();
+ } else {
+ assertThat(prefix).isNull();
+ }
Range dedicatedRange = new Range(20 + field * 20, 39 + field * 20);
IndexFileMeta dedicated =
indexFile(
@@ -659,13 +678,50 @@ class GlobalIndexQueryTest {
null);
GlobalIndexQuery query =
GlobalIndexQuery.create(
- rowType, predicate, Arrays.asList(multi,
dedicated), paths);
+ rowType,
+ predicate,
+ Arrays.asList(multi, dedicated),
+ paths,
+ new Options());
assertThat(query).isNotNull();
- assertThat(query.coveredRanges()).containsExactly(dedicatedRange);
- assertThat(query.hasCompositeQuery()).isFalse();
+ if (field == 0 && "btree".equals(indexType)) {
+ // Partial scalar coverage must not replace a fully covered
composite prefix.
+ assertThat(query.coveredRanges()).containsExactly(new Range(0,
99));
+ assertThat(query.hasCompositeQuery()).isTrue();
+ } else {
+
assertThat(query.coveredRanges()).containsExactly(dedicatedRange);
+ assertThat(query.hasCompositeQuery()).isFalse();
+ }
}
}
+ @Test
+ void testDedicatedLeadingIndexPreferredOnlyWithCompleteCoverage() {
+ RowType rowType = RowType.of(DataTypes.INT(), DataTypes.INT());
+ Predicate predicate = new PredicateBuilder(rowType).equal(0, 7);
+ IndexPathFactory paths = mock(IndexPathFactory.class);
+ when(paths.toPath(any(IndexFileMeta.class)))
+ .thenAnswer(
+ invocation ->
+ new
Path(invocation.<IndexFileMeta>getArgument(0).fileName()));
+ IndexFileMeta composite = indexFile("btree", "composite", 0, 99, 0,
new int[] {1});
+ IndexFileMeta first = indexFile("btree", "first", 0, 49, 0, null);
+ IndexFileMeta second = indexFile("bitmap", "second", 50, 99, 0, null);
+ GlobalIndexQuery partial =
+ GlobalIndexQuery.create(
+ rowType, predicate, Arrays.asList(composite, first),
paths, new Options());
+ assertThat(partial.hasCompositeQuery()).isTrue();
+ GlobalIndexQuery complete =
+ GlobalIndexQuery.create(
+ rowType,
+ predicate,
+ Arrays.asList(composite, first, second),
+ paths,
+ new Options());
+ assertThat(complete.hasCompositeQuery()).isFalse();
+ assertThat(complete.coveredRanges()).containsExactly(new Range(0, 99));
+ }
+
private IndexFileMeta indexFile(
String type, String name, long start, long end, int fieldId, int[]
extraFields) {
return new IndexFileMeta(
@@ -720,6 +776,7 @@ class GlobalIndexQueryTest {
return false;
}
};
- return GlobalIndexQuery.create(rowType, predicate,
Collections.singletonList(file), paths);
+ return GlobalIndexQuery.create(
+ rowType, predicate, Collections.singletonList(file), paths,
new Options());
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/CompositeBTreeTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/CompositeBTreeTableTest.java
index 9976fe8479..9e6361a82b 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/CompositeBTreeTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/CompositeBTreeTableTest.java
@@ -67,6 +67,159 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Composite lookups must avoid single-column postings and retain uncovered
data. */
class CompositeBTreeTableTest extends DataEvolutionTestBase {
+ @Test
+ void testPrefixAndRangeQueriesAcrossReaderModes() throws Exception {
+ createTableDefault();
+ FileStoreTable table =
+
table().copy(Collections.singletonMap("sorted-index.records-per-file", "13"));
+ append(table, 0, 60);
+ BatchWriteBuilder writes = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writes.newWrite();
+ BatchTableCommit commit = writes.newCommit()) {
+ write.write(
+ GenericRow.of(
+ null,
+ BinaryString.fromString("category-a"),
+ BinaryString.fromString("null-number")));
+ write.write(GenericRow.of(7, null,
BinaryString.fromString("null-category")));
+ commit.commit(write.prepareCommit());
+ }
+ build(table, "f1", "f0", "f2");
+ FileStoreTable withoutIndex =
+ table.copy(
+
Collections.singletonMap(CoreOptions.GLOBAL_INDEX_ENABLED.key(), "false"));
+ PredicateBuilder b = new PredicateBuilder(table.rowType());
+ Predicate prefix = b.equal(1, BinaryString.fromString("category-a"));
+ List<Predicate> queries =
+ Arrays.asList(
+ prefix,
+ PredicateBuilder.and(prefix, b.equal(0, 7)),
+ PredicateBuilder.and(prefix, b.greaterThan(0, 7)),
+ PredicateBuilder.and(prefix, b.greaterOrEqual(0, 7)),
+ PredicateBuilder.and(prefix, b.lessThan(0, 7)),
+ PredicateBuilder.and(prefix, b.lessOrEqual(0, 7)),
+ PredicateBuilder.and(prefix, b.between(0, 6, 8)),
+ PredicateBuilder.and(prefix, b.greaterThan(0, 6),
b.lessThan(0, 8)),
+ PredicateBuilder.and(prefix, b.greaterThan(0, 7),
b.lessOrEqual(0, 7)),
+ PredicateBuilder.and(
+ prefix,
+ b.greaterThan(0, 7),
+ b.startsWith(2,
BinaryString.fromString("p2"))),
+ PredicateBuilder.and(
+ b.greaterThan(1,
BinaryString.fromString("category-a")),
+ b.equal(0, 7)),
+ b.equal(0, 7),
+ PredicateBuilder.or(
+ PredicateBuilder.and(prefix, b.greaterThan(0,
7)),
+ PredicateBuilder.and(
+ b.equal(1,
BinaryString.fromString("category-b")),
+ b.lessOrEqual(0, 1))));
+ for (Predicate query : queries) {
+ List<String> expected = read(withoutIndex, query);
+ for (boolean inReader : Arrays.asList(false, true)) {
+ for (String mode : Arrays.asList("fast", "full", "detail")) {
+ assertThat(read(configured(table, mode, inReader), query))
+ .as("%s, reader=%s, mode=%s", query, inReader,
mode)
+ .containsExactlyInAnyOrderElementsOf(expected);
+ }
+ }
+ }
+ append(table, 60, 80);
+ Predicate query = PredicateBuilder.and(prefix, b.greaterThan(0, 7));
+ for (boolean inReader : Arrays.asList(false, true)) {
+ assertThat(read(configured(table, "fast", inReader),
query)).hasSize(6);
+ for (String mode : Arrays.asList("full", "detail")) {
+ assertThat(read(configured(table, mode, inReader), query))
+
.containsExactlyInAnyOrderElementsOf(read(withoutIndex, query));
+ }
+ }
+ }
+
+ @Test
+ void testRangesAvoidSingleColumnPostingsAndKeepResidualConditions() throws
Exception {
+ createTableDefault();
+ FileStoreTable table = table();
+ append(table, 0, 100);
+ build(table, "f1");
+ build(table, "f0");
+ build(table, "f1", "f0");
+ deleteIndexFiles(table, 1);
+ PredicateBuilder b = new PredicateBuilder(table.rowType());
+ Predicate range =
+ PredicateBuilder.and(
+ b.equal(1, BinaryString.fromString("category-a")),
b.between(0, 6, 8));
+ Predicate residual = PredicateBuilder.and(range, b.notEqual(0, 7));
+ Predicate union =
+ PredicateBuilder.or(
+ residual,
+ PredicateBuilder.and(
+ b.equal(1,
BinaryString.fromString("category-b")),
+ b.greaterThan(0, 8)));
+ FileStoreTable withoutIndex =
+ table.copy(
+
Collections.singletonMap(CoreOptions.GLOBAL_INDEX_ENABLED.key(), "false"));
+ for (boolean inReader : Arrays.asList(false, true)) {
+ for (Predicate query : Arrays.asList(range, residual, union)) {
+ assertThat(read(configured(table, "fast", inReader), query))
+
.containsExactlyInAnyOrderElementsOf(read(withoutIndex, query));
+ }
+ }
+ }
+
+ @Test
+ void testOneOverBudgetShardFallsBackForTheWholeDefinition() throws
Exception {
+ createTableDefault();
+ FileStoreTable table = table();
+ append(table, 0, 20);
+ build(table, "f1", "f0", "f2");
+ append(table, 20, 2200);
+ build(table, "f1", "f0", "f2");
+ List<IndexFileMeta> files =
+ table.store().newIndexFileHandler().scanEntries().stream()
+ .map(IndexManifestEntry::indexFile)
+ .collect(Collectors.toList());
+ long small =
+ files.stream()
+ .filter(file -> file.globalIndexMeta().rowRange().to
== 19)
+ .mapToLong(IndexFileMeta::fileSize)
+ .sum();
+ long large =
+ files.stream()
+ .filter(file -> file.globalIndexMeta().rowRange().from
== 20)
+ .mapToLong(IndexFileMeta::fileSize)
+ .sum();
+ assertThat(small).isPositive();
+ assertThat(large).isGreaterThan(small);
+ FileStoreTable budgeted =
+ table.copy(
+ Collections.singletonMap(
+ "btree-index.fallback-scan-max-size", small +
" b"));
+ FileStoreTable withoutIndex =
+ table.copy(
+
Collections.singletonMap(CoreOptions.GLOBAL_INDEX_ENABLED.key(), "false"));
+ PredicateBuilder b = new PredicateBuilder(table.rowType());
+ Predicate range =
+ PredicateBuilder.and(
+ b.equal(1, BinaryString.fromString("category-a")),
b.between(0, 7, 8));
+ Predicate conjunction =
+ PredicateBuilder.and(range, b.startsWith(2,
BinaryString.fromString("p")));
+ Predicate union =
+ PredicateBuilder.or(
+ range,
+ PredicateBuilder.and(
+ b.equal(1,
BinaryString.fromString("category-b")),
+ b.greaterThan(0, 8)));
+ for (Predicate query : Arrays.asList(range, conjunction, union)) {
+ List<String> expected = read(withoutIndex, query);
+ for (boolean inReader : Arrays.asList(false, true)) {
+ for (String mode : Arrays.asList("fast", "full", "detail")) {
+ assertThat(read(configured(budgeted, mode, inReader),
query))
+ .containsExactlyInAnyOrderElementsOf(expected);
+ }
+ }
+ }
+ }
+
@Test
void testCoexistsAndAvoidsSingleColumnPostings() throws Exception {
createTableDefault();
@@ -395,7 +548,7 @@ class CompositeBTreeTableTest extends DataEvolutionTestBase
{
DataEvolutionGlobalIndexScanner.create(table, files).get()) {
for (Predicate predicate :
Arrays.asList(
- predicates.equal(0, 7),
+ predicates.notEqual(0, 7),
predicates.equal(1,
BinaryString.fromString("category-a")))) {
assertThat(scanner.scan(predicate)).isEmpty();
assertThat(scanner.unindexedRows(predicate).results().toRangeList())
@@ -619,17 +772,15 @@ class CompositeBTreeTableTest extends
DataEvolutionTestBase {
createTableDefault();
FileStoreTable table = table();
append(table, 0, 20);
- build(table, "f1");
+ build(table, "f0");
append(table, 20, 40);
build(table, "f1", "f0");
- Predicate scalar =
- new PredicateBuilder(table.rowType())
- .equal(1, BinaryString.fromString("category-a"));
+ Predicate scalar = new PredicateBuilder(table.rowType()).equal(0, 7);
for (boolean inReader : Arrays.asList(false, true)) {
- assertThat(read(configured(table, "fast", inReader),
scalar)).hasSize(10);
+ assertThat(read(configured(table, "fast", inReader),
scalar)).hasSize(2);
for (String mode : Arrays.asList("full", "detail")) {
assertThat(read(configured(table, mode, inReader), scalar))
- .hasSize(20)
+ .hasSize(4)
.contains("p27");
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
index 0ae2afbe7a..d409a8955e 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/SortedGlobalIndexITCase.java
@@ -152,6 +152,17 @@ public class SortedGlobalIndexITCase extends
CatalogITCaseBase {
sql(
"SELECT id FROM T_COMPOSITE WHERE category
= 'absent' AND item_number = 7"))
.isEmpty();
+ assertThat(sql("SELECT id FROM T_COMPOSITE WHERE item_number = 7"))
+ .containsExactlyInAnyOrder(
+ Row.of(7), Row.of(17), Row.of(27), Row.of(37),
Row.of(100));
+ assertThat(
+ sql(
+ "SELECT id FROM T_COMPOSITE WHERE
item_number = 7 AND category > 'category-a'"))
+ .containsExactlyInAnyOrder(Row.of(17), Row.of(37));
+ assertThat(
+ sql(
+ "SELECT id FROM T_COMPOSITE WHERE
item_number = 7 AND category BETWEEN 'category-a' AND 'category-b'"))
+ .containsExactlyInAnyOrder(Row.of(7), Row.of(17),
Row.of(27), Row.of(37));
}
insertCompositeRows(40, 60);
buildBTreeIndexForTable("T_COMPOSITE", "item_number,category");
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompositeBTreeIndexProcedureTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompositeBTreeIndexProcedureTest.scala
index 334b68471c..ed820634da 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompositeBTreeIndexProcedureTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompositeBTreeIndexProcedureTest.scala
@@ -99,6 +99,15 @@ class CompositeBTreeIndexProcedureTest extends
PaimonSparkTestBase {
checkAnswer(
sql("SELECT id FROM T WHERE category = 'absent' AND item_number =
7"),
Seq.empty)
+ checkAnswer(
+ sql("SELECT id FROM T WHERE category = 'category-a'"),
+ ((0 until 10) ++ (20 until 30) :+ 101).map(Row(_)))
+ checkAnswer(
+ sql("SELECT id FROM T WHERE category = 'category-a' AND item_number
> 7"),
+ Seq(Row(8), Row(9), Row(28), Row(29)))
+ checkAnswer(
+ sql("SELECT id FROM T WHERE category = 'category-a' AND item_number
BETWEEN 6 AND 8"),
+ Seq(Row(6), Row(7), Row(8), Row(26), Row(27), Row(28)))
}
insert(40, 60)
build("category,item_number")
