diff --git a/docs/docs/multimodal-table/global-index/btree.mdx b/docs/docs/multimodal-table/global-index/btree.mdx index 5a31c6b005e1..7da34552eb9c 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 3c4d80c4417c..71731aa07b93 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 @@ private CompletableFuture> visitParallel( return visitSelectedFiles(selector.get(), visitor); } - private CompletableFuture> visitSelectedFiles( + protected CompletableFuture> visitSelectedFiles( Optional> selectedOpt, Function> 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 75ae2a762bd7..84b403b46e18 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 String identifier() { public List selectFiles( List indexFields, Predicate predicate, List files) { if (indexFields.size() > 1) { - Optional> 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 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 174df0704214..26584d4260c7 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 Optional visitEqual(Object literal) { return createResult(() -> pointQuery(literal)); } + public Optional 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 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 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 02b71aec8909..3acccc79a6bc 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 fields, Predicate predicate) { - List 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(List fields, Predicate predicate) { + List conjuncts = PredicateBuilder.splitAnd(predicate); + List covered = new ArrayList<>(); + List prefix = new ArrayList<>(); + List range = Collections.emptyList(); + boolean empty = false; + for (DataField field : fields) { + List 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 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> match(List fields, Predicate predicate) { - List conjuncts = PredicateBuilder.splitAnd(predicate); - List 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[] comparators; + + @SuppressWarnings("unchecked") + private Bound(List 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 fields; + private final List predicates; + private final Bound lower; + private final Bound upper; + private final int boundColumns; + private final int equalColumns; + private final boolean empty; + + private Plan( + List fields, + List 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 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 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 selectFiles(List 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 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 b14cf376c5b9..6ec85d813e0b 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.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 comparator; private final long totalRowCount; + private final List indexFiles; + private final long fallbackScanMaxSize; @Nullable private final Pair fullRangeBounds; public LazyFilteredBTreeReader( @@ -77,6 +77,8 @@ public LazyFilteredBTreeReader( 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 @@ private Pair fullRangeBounds(List files) { 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 CompletableFuture> visitComposite(Predicate p if (indexFields.size() < 2) { return CompletableFuture.completedFuture(Optional.empty()); } - Optional> matched = - CompositeBTreePredicate.match(indexFields, predicate); - if (!matched.isPresent()) { + Optional 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 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 b477f8547d14..861f892f818e 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 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> { @@ -64,6 +65,11 @@ public void remove() { } 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 compareToTarget) { int left = 0; int right = reader.recordCount() - 1; polledPosition = -1; @@ -72,7 +78,7 @@ public boolean seekTo(MemorySlice targetKey) { 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 8346464f3b4f..ea0575457e4d 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 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 SstFileIterator { */ 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 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 1c05959f13a0..b6425835412e 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 @@ public PositionOutputStream newOutputStream(String name) 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 6172018d29b7..4a63cdd80a0f 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.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 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 fields = @@ -57,42 +55,36 @@ void testMatchesInIndexOrderRegardlessOfPredicateOrder() { Predicate predicate = PredicateBuilder.and( builder.equal(1, -3), builder.equal(0, BinaryString.fromString("a"))); - List 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 @@ void testRequiresCompleteConjunctiveEqualitiesAndMatchingTypes() { 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 @@ void testContradictionsAndSqlNullEqualityPruneAllFiles() { 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 files = Collections.singletonList(file); assertThat(factory.selectFiles(fields, builder.equal(1, 100), files)).isSameAs(files); @@ -154,7 +146,7 @@ void testRetainsFilesWhenCompleteKeyOrMetadataIsUnavailable() { builder.equal(0, BinaryString.fromString("a")), builder.greaterThan(1, 100)), files)) - .isSameAs(files); + .isEmpty(); List 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 @@ void testRetainsFilesWhenCompleteKeyOrMetadataIsUnavailable() { .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 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 000000000000..58a1602851be --- /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 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 files = write(options); + List 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 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 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 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 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 f40b46439903..47c7e6573d89 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 void testUnalignedIterator() throws IOException { 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 a1cdc71ebe38..4fdae682e5fb 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 @@ private Plan planWithIndexQuery() { 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 a6ef515ceb29..e294cc09d118 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 Optional scanWithCoverage(Predicate pred .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 GlobalIndexResult unindexedRows(Predicate predicate) { 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 18737d69f98e..f962282552b3 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.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.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 @@ static GlobalIndexQuery create( RowType rowType, Predicate predicate, List files, - IndexPathFactory pathFactory) { + IndexPathFactory pathFactory, + Options options) { Map> groupsByField = DataEvolutionGlobalIndexScanner.groupIndexFiles( files.stream() @@ -124,90 +125,54 @@ static GlobalIndexQuery create( } 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> groups) { + Predicate predicate, + RowType rowType, + Map> groups, + Options options) { if (predicate instanceof LeafPredicate) { LeafPredicate leaf = (LeafPredicate) predicate; Optional 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 children = new ArrayList<>(); List 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 matched = null; - Predicate conjunction = PredicateBuilder.and(predicates); - for (List fieldGroups : groups.values()) { - for (IndexGroup group : fieldGroups) { - if (!group.isCompositeBTree() - || (selected != null - && group.extraFields.size() - <= selected.extraFields.size())) { - continue; - } - Optional> 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 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 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 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 keyColumns = + selected.group.indexFields().stream() + .map(DataField::name) + .collect(Collectors.toSet()); predicates.removeIf( - child -> { - if (!(child instanceof LeafPredicate)) { - return false; - } - Optional 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 @@ private static GlobalIndexQuery createForPredicate( } } if (query == null) { - query = createForPredicate(child, rowType, groups); + query = createForPredicate(child, rowType, groups, options); } if (query == null) { if (union) { @@ -253,6 +218,101 @@ private static GlobalIndexQuery createForPredicate( : new GlobalIndexQuery(null, union, children, Collections.emptyList()); } + @Nullable + private static CompositeCandidate selectComposite( + Predicate predicate, Map> groups, Options options) { + CompositeCandidate selected = null; + Set> seen = new HashSet<>(); + long budget = options.get(BTreeIndexOptions.BTREE_INDEX_FALLBACK_SCAN_MAX_SIZE).getBytes(); + for (List fieldGroups : groups.values()) { + for (IndexGroup group : fieldGroups) { + if (!group.isCompositeBTree() || !seen.add(group.indexFields())) { + continue; + } + Optional planned = + CompositeBTreePredicate.plan(group.indexFields(), predicate); + if (!planned.isPresent()) { + continue; + } + CompositeBTreePredicate.Plan plan = planned.get(); + List 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 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 scalarCoverage = + Range.sortAndMergeOverlap( + fieldGroups.stream() + .filter(other -> other.extraFields.isEmpty()) + .map(other -> other.range) + .collect(Collectors.toList()), + true); + List 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 groups; + + private CompositeCandidate( + IndexGroup group, CompositeBTreePredicate.Plan plan, List 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 79727b7b3e13..c315a73b2002 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 @@ void testPrunesSortedFilesBeforeSplitSerialization(String indexType) throws Exce 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 @@ void testBoundedRangeUsesSingleScan(boolean lowerInclusive, boolean upperInclusi invocation -> new Path(invocation.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 @@ void testBoundedRangeUsesSingleScan(boolean lowerInclusive, boolean upperInclusi assertThat( GlobalIndexQuery.create( - rowType, PredicateBuilder.or(lower, upper), files, paths)) + rowType, + PredicateBuilder.or(lower, upper), + files, + paths, + new Options())) .isNotNull(); } @@ -249,7 +255,8 @@ public boolean isExternalPath() { 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 ranges = Collections.singletonList(new Range(100, 102)); assertThat(plan).isNotNull(); assertThat(plan.evaluate(fileIO, new Options(), ranges).results().toRangeList()) @@ -608,7 +615,8 @@ void testPlansAvailableIndexesWithoutTypeChecks() throws Exception { 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 @@ void testPlansAvailableIndexesWithoutTypeChecks() throws Exception { 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 @@ void testScalarPlansUseOnlySingleFieldDefinitions(String indexType) { new Path(invocation.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 @@ void testScalarPlansUseOnlySingleFieldDefinitions(String indexType) { 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.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 @@ public boolean isExternalPath() { 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 9976fe8479d4..9e6361a82bf1 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 @@ /** 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 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 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 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 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 @@ void testScalarReadersDoNotOpenMultiFieldIndexFiles(String indexType) throws Exc 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 @@ void testPartialSingleCoverageDoesNotBorrowCompositeCoverage() throws Exception 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 0ae2afbe7a94..d409a8955e7b 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 void testCompositeBTreeIndex() throws Exception { 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 334b68471c39..ed820634dac9 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")