diff --git a/docs/docs/multimodal-table/global-index/btree.mdx b/docs/docs/multimodal-table/global-index/btree.mdx index 7da34552eb9c..c34c611c1052 100644 --- a/docs/docs/multimodal-table/global-index/btree.mdx +++ b/docs/docs/multimodal-table/global-index/btree.mdx @@ -167,8 +167,8 @@ 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. -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 +Composite indexes also support a prefix restricted by equality, `IN` or `IS NULL`, +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 | @@ -177,13 +177,18 @@ use the index: | `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 | +| `category IN ('category-a', 'category-b') AND item_number = 7` | A point probe for each category | +| `category IN ('category-a', 'category-b') AND item_number > 7` | A separate range within each category prefix | +| `category IS NULL AND item_number > 7` | A range within the NULL category prefix | +| `category = 'category-a' AND item_number IS NULL` | A point probe with a NULL second component | +| `category IS NOT NULL AND item_number = 7` | A non-NULL category range; the item-number condition remains a data filter | -The range may use `>`, `>=`, `<`, `<=` or `BETWEEN`. Conditions after the first +The range may use `>`, `>=`, `<`, `<=`, `BETWEEN` or `IS NOT NULL`. 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 +columns, then more columns restricted by equality, `IN` or `IS NULL`, 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. @@ -192,12 +197,19 @@ 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. -Full-key equalities use point probes. Prefix and range scans use +Restrictions by equality, `IN` or `IS NULL` on every key column use point probes. +Repeated `IN` values are deduplicated, and NULL entries in `IN` lists do not match +NULL keys. `IS NULL` explicitly selects NULL keys; equality to NULL still matches +no rows. The planner expands at most 256 tuple intervals after deduplication and +intersection of conditions on each column. Queries exceeding this limit fall back +to other eligible index paths or an ordinary scan. + +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 +unaffected by this budget. A file shared by several intervals is counted only once. +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 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 26584d4260c7..2ab03fb49e67 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 @@ -346,26 +346,32 @@ public Optional visitComposite(CompositeBTreePredicate.Plan p 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; + for (CompositeBTreePredicate.Interval interval : plan.intervals()) { + if (plan.isPointLookup()) { + result.or(pointQuery(interval.pointKey())); + continue; + } + SstFileReader.SstFileIterator iterator = reader.createIterator(); + iterator.seekTo( + key -> + interval.lower() + .compareKey( + (InternalRow) + keySerializer.deserialize(key))); + BlockIterator batch; + boolean finished = false; + while (!finished && (batch = iterator.readBatch()) != null) { + while (batch.hasNext()) { + Map.Entry entry = batch.next(); + InternalRow key = + (InternalRow) keySerializer.deserialize(entry.getKey()); + if (interval.upper().compareKey(key) > 0) { + finished = true; + break; + } + addRowIdsTo(entry.getValue(), result); } - addRowIdsTo(entry.getValue(), result); } } return result; 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 3acccc79a6bc..f39bde5c1fdf 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 @@ -26,12 +26,17 @@ import org.apache.paimon.globalindex.SortedIndexFileMeta; import org.apache.paimon.memory.MemorySlice; import org.apache.paimon.predicate.Between; +import org.apache.paimon.predicate.CompoundPredicate; import org.apache.paimon.predicate.Equal; import org.apache.paimon.predicate.GreaterOrEqual; import org.apache.paimon.predicate.GreaterThan; +import org.apache.paimon.predicate.In; +import org.apache.paimon.predicate.IsNotNull; +import org.apache.paimon.predicate.IsNull; import org.apache.paimon.predicate.LeafPredicate; import org.apache.paimon.predicate.LessOrEqual; import org.apache.paimon.predicate.LessThan; +import org.apache.paimon.predicate.Or; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.types.DataField; @@ -43,103 +48,204 @@ import java.util.Comparator; import java.util.List; import java.util.Optional; +import java.util.TreeSet; -/** An equality prefix followed by an optional range in physical composite-key order. */ +/** Discrete equality/IN/NULL prefixes followed by an optional range in composite-key order. */ public final class CompositeBTreePredicate { + public static final int MAX_INTERVALS = 256; + private CompositeBTreePredicate() {} public static Optional plan(List fields, Predicate predicate) { List conjuncts = PredicateBuilder.splitAnd(predicate); List covered = new ArrayList<>(); - List prefix = new ArrayList<>(); + List prefixes = new ArrayList<>(); + prefixes.add(new Object[0]); List range = Collections.emptyList(); - boolean empty = false; + int equalColumns = 0; for (DataField field : fields) { List column = new ArrayList<>(); - LeafPredicate equality = null; + List originals = new ArrayList<>(); 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; + LeafPredicate leaf = asLeaf(child); + if (leaf != null + && leaf.fieldRefOptional().isPresent() + && field.name().equals(leaf.fieldRefOptional().get().name()) + && field.type().equalsIgnoreNullable(leaf.type()) + && (isPoint(leaf) || isRange(leaf))) { + column.add(leaf); + originals.add(child); } } - if (equality == null) { + Optional> domain = pointValues(field, column); + covered.addAll(originals); + if (!domain.isPresent()) { range = column; - covered.addAll(range); break; } - 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()); + equalColumns++; + if (domain.get().isEmpty()) { + return Optional.of( + new Plan( + fields, + covered, + Collections.emptyList(), + equalColumns, + equalColumns)); + } + if ((long) prefixes.size() * domain.get().size() > MAX_INTERVALS) { + return Optional.empty(); } - prefix.add(value); - covered.addAll(column); + List expanded = new ArrayList<>(); + for (Object[] prefix : prefixes) { + for (Object value : domain.get()) { + expanded.add(append(prefix, value)); + } + } + prefixes = expanded; } - 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; + List intervals = new ArrayList<>(); + for (Object[] prefix : prefixes) { + Bound lower = new Bound(fields, prefix, false); + Bound upper = new Bound(fields, prefix, true); + boolean empty = false; + if (!range.isEmpty()) { + // Comparisons and IS NOT NULL exclude the entire NULL-component suffix. + lower = new Bound(fields, append(prefix, null), true); + for (LeafPredicate leaf : range) { + if (leaf.function() instanceof IsNotNull) { + continue; } - } - 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; + if (leaf.literals().contains(null)) { + empty = true; + break; + } + if (leaf.function() instanceof GreaterThan + || leaf.function() instanceof GreaterOrEqual + || leaf.function() instanceof Between) { + Bound next = + new Bound( + fields, + append(prefix, 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(prefix, value), + !(leaf.function() instanceof LessThan)); + if (next.compareTo(upper) < 0) { + upper = next; + } } } } + if (!empty && lower.compareTo(upper) < 0) { + intervals.add(new Interval(lower, upper)); + } + } + return Optional.of(new Plan(fields, covered, intervals, boundColumns, equalColumns)); + } + + private static Optional> pointValues(DataField field, List column) { + LeafPredicate selected = null; + for (LeafPredicate leaf : column) { + if (isPoint(leaf) + && (selected == null + || !(leaf.function() instanceof In) + || (selected.function() instanceof In + && leaf.literals().size() < selected.literals().size()))) { + selected = leaf; + } } - return Optional.of( - new Plan( - fields, - covered, - lower, - upper, - boundColumns, - equalColumns, - empty || lower.compareTo(upper) >= 0)); + if (selected == null) { + return Optional.empty(); + } + List candidates = + selected.function() instanceof IsNull + ? Collections.singletonList(null) + : selected.literals(); + Comparator comparator = KeySerializer.create(field.type()).createComparator(); + TreeSet values = + new TreeSet<>( + (left, right) -> + left == null + ? (right == null ? 0 : -1) + : right == null ? 1 : comparator.compare(left, right)); + for (Object value : candidates) { + if (value == null + && (!(selected.function() instanceof IsNull) || !field.type().isNullable())) { + continue; + } + boolean matches = true; + for (LeafPredicate leaf : column) { + if (!leaf.function().test(leaf.type(), value, leaf.literals())) { + matches = false; + break; + } + } + if (matches) { + values.add(value); + if (values.size() > MAX_INTERVALS) { + break; + } + } + } + return Optional.of(new ArrayList<>(values)); + } + + // PredicateBuilder represents small IN lists as ORs of equalities. + private static LeafPredicate asLeaf(Predicate predicate) { + if (predicate instanceof LeafPredicate) { + return (LeafPredicate) predicate; + } + if (!(predicate instanceof CompoundPredicate) + || !(((CompoundPredicate) predicate).function() instanceof Or)) { + return null; + } + LeafPredicate first = null; + List literals = new ArrayList<>(); + for (Predicate child : PredicateBuilder.splitOr(predicate)) { + if (!(child instanceof LeafPredicate)) { + return null; + } + LeafPredicate leaf = (LeafPredicate) child; + if (!(leaf.function() instanceof Equal) + || !leaf.fieldRefOptional().isPresent() + || (first != null + && !first.fieldRefOptional().equals(leaf.fieldRefOptional()))) { + return null; + } + first = leaf; + literals.add(leaf.literals().get(0)); + } + return first == null + ? null + : new LeafPredicate( + In.INSTANCE, + first.type(), + first.fieldRefOptional().get().index(), + first.fieldRefOptional().get().name(), + literals); + } + + private static boolean isPoint(LeafPredicate leaf) { + return leaf.function() instanceof Equal + || leaf.function() instanceof In + || leaf.function() instanceof IsNull; } private static boolean isRange(LeafPredicate leaf) { @@ -147,7 +253,8 @@ private static boolean isRange(LeafPredicate leaf) { || leaf.function() instanceof GreaterOrEqual || leaf.function() instanceof LessThan || leaf.function() instanceof LessOrEqual - || leaf.function() instanceof Between; + || leaf.function() instanceof Between + || leaf.function() instanceof IsNotNull; } private static Object[] append(Object[] prefix, Object value) { @@ -209,43 +316,56 @@ private int compareTo(Bound other) { } } - /** One tuple interval, with its covered query predicates and point-lookup classification. */ + /** One disjoint tuple interval, in physical key order. */ + public static final class Interval { + private final Bound lower; + private final Bound upper; + + private Interval(Bound lower, Bound upper) { + this.lower = lower; + this.upper = upper; + } + + public Bound lower() { + return lower; + } + + public Bound upper() { + return upper; + } + + public GenericRow pointKey() { + return GenericRow.of(lower.values); + } + } + + /** Tuple intervals and their covered predicates, rebuilt from the original query. */ public static final class Plan { private final List fields; private final List predicates; - private final Bound lower; - private final Bound upper; + private final List intervals; private final int boundColumns; private final int equalColumns; - private final boolean empty; private Plan( List fields, List predicates, - Bound lower, - Bound upper, + List intervals, int boundColumns, - int equalColumns, - boolean empty) { + int equalColumns) { this.fields = fields; this.predicates = predicates; - this.lower = lower; - this.upper = upper; + this.intervals = intervals; 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 List intervals() { + return intervals; } public int boundColumns() { @@ -257,17 +377,13 @@ public int equalColumns() { } public boolean isEmpty() { - return empty; + return intervals.isEmpty(); } 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. */ @@ -288,7 +404,7 @@ public boolean canScan(List selected, long budget) { } public List selectFiles(List files) { - if (empty) { + if (isEmpty()) { return Collections.emptyList(); } // Preserve conservative pruning when metadata is unavailable. @@ -299,18 +415,19 @@ public List selectFiles(List files) { 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)) { + if (meta.firstKey() == null || meta.lastKey() == null) { + result.add(file); + continue; + } + InternalRow first = + (InternalRow) serializer.deserialize(MemorySlice.wrap(meta.firstKey())); + InternalRow last = + (InternalRow) serializer.deserialize(MemorySlice.wrap(meta.lastKey())); + if (intervals.stream() + .anyMatch( + interval -> + interval.lower.compareKey(last) > 0 + && interval.upper.compareKey(first) < 0)) { result.add(file); } } 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 6ec85d813e0b..fcbdf9259560 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 @@ -131,8 +131,8 @@ public CompletableFuture> visitComposite(Predicate p return CompletableFuture.completedFuture(Optional.empty()); } CompositeBTreePredicate.Plan plan = planned.get(); - if (plan.isPointLookup() && !plan.isEmpty()) { - return visitEqual((FieldRef) null, plan.pointKey()); + if (plan.isPointLookup() && plan.intervals().size() == 1) { + return visitEqual((FieldRef) null, plan.intervals().get(0).pointKey()); } List selected = plan.selectFiles(indexFiles); if (!plan.canScan(selected, fallbackScanMaxSize)) { 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 b6425835412e..df3975a4f280 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 @@ -195,15 +195,7 @@ public PositionOutputStream newOutputStream(String name) for (Predicate unsupported : Arrays.asList( builder.equal(1, 7), - builder.isNull(0), - PredicateBuilder.and( - builder.in( - 0, - Arrays.asList( - BinaryString.fromString("category-a"), - BinaryString.fromString("category-b"))), - builder.equal(1, 7), - builder.equal(2, BinaryString.fromString("tag"))), + builder.notEqual(0, BinaryString.fromString("category-a")), PredicateBuilder.or(fullKey, builder.equal(1, 8)))) { assertThat(reader.visitComposite(unsupported).join()).isEmpty(); } 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 4a63cdd80a0f..4bea80c1fce7 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 @@ -33,6 +33,7 @@ import org.junit.jupiter.api.Test; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; @@ -55,13 +56,20 @@ void testMatchesInIndexOrderRegardlessOfPredicateOrder() { Predicate predicate = PredicateBuilder.and( builder.equal(1, -3), builder.equal(0, BinaryString.fromString("a"))); - assertThat(CompositeBTreePredicate.plan(fields, predicate).get().pointKey()) + assertThat( + CompositeBTreePredicate.plan(fields, predicate) + .get() + .intervals() + .get(0) + .pointKey()) .isEqualTo(GenericRow.of(BinaryString.fromString("a"), -3)); assertThat(CompositeBTreePredicate.plan(fields, predicate).get().isEmpty()).isFalse(); assertThat( CompositeBTreePredicate.plan( Arrays.asList(fields.get(1), fields.get(0)), predicate) .get() + .intervals() + .get(0) .pointKey()) .isEqualTo(GenericRow.of(-3, BinaryString.fromString("a"))); } @@ -197,6 +205,83 @@ private Predicate equal(String category, int number) { builder.equal(1, number), builder.equal(0, BinaryString.fromString(category))); } + @Test + void testDiscreteDomainsIntersectDeduplicateAndBoundExpansion() { + List categories = new ArrayList<>(); + List numbers = new ArrayList<>(); + for (int i = 0; i < 16; i++) { + categories.add(BinaryString.fromString("category-" + i)); + numbers.add(i); + } + Predicate first = builder.in(0, categories); + Predicate max = PredicateBuilder.and(first, builder.in(1, numbers)); + assertThat(CompositeBTreePredicate.plan(fields, max).get().intervals()).hasSize(256); + numbers.add(null); + numbers.add(7); + assertThat( + CompositeBTreePredicate.plan( + fields, PredicateBuilder.and(first, builder.in(1, numbers))) + .get() + .intervals()) + .hasSize(256); + numbers.add(16); + Predicate overflow = PredicateBuilder.and(first, builder.in(1, numbers)); + assertThat(CompositeBTreePredicate.plan(fields, overflow)).isEmpty(); + assertThat( + CompositeBTreePredicate.plan( + fields, PredicateBuilder.and(overflow, builder.equal(1, 7))) + .get() + .intervals()) + .hasSize(16); + assertThat( + CompositeBTreePredicate.plan( + fields, + PredicateBuilder.and( + overflow, builder.in(1, Arrays.asList(7, 8)))) + .get() + .intervals()) + .hasSize(32); + assertThat( + CompositeBTreePredicate.plan( + fields, + PredicateBuilder.and( + first, + builder.in( + 0, + Collections.singletonList( + BinaryString.fromString( + "absent"))))) + .get() + .isEmpty()) + .isTrue(); + List nonNull = + Arrays.asList( + new DataField(10, "category", DataTypes.STRING().notNull()), fields.get(1)); + assertThat(CompositeBTreePredicate.plan(nonNull, builder.isNull(0)).get().isEmpty()) + .isTrue(); + } + + @Test + void testIntervalFileUnionCountsEachFileOnce() { + GlobalIndexIOMeta first = file("first", "a", -3, "a", 7); + GlobalIndexIOMeta second = file("second", "a", 10, "b", 2); + GlobalIndexIOMeta third = file("third", "b", 3, "c", 4); + Predicate query = + PredicateBuilder.and( + builder.in( + 0, + Arrays.asList( + BinaryString.fromString("b"), + BinaryString.fromString("a"), + BinaryString.fromString("a"))), + builder.between(1, 0, 2)); + CompositeBTreePredicate.Plan plan = CompositeBTreePredicate.plan(fields, query).get(); + List selected = plan.selectFiles(Arrays.asList(first, second, third)); + assertThat(selected).containsExactly(first, second); + assertThat(plan.canScan(selected, 2)).isTrue(); + assertThat(plan.canScan(selected, 1)).isFalse(); + } + private GlobalIndexIOMeta file( String name, String first, int firstNumber, String last, int lastNumber) { byte[] metadata = 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 index 58a1602851be..ce8e5644469e 100644 --- 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 @@ -31,6 +31,8 @@ import org.apache.paimon.globalindex.io.GlobalIndexFileWriter; import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.In; +import org.apache.paimon.predicate.LeafPredicate; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.types.DataField; @@ -162,6 +164,22 @@ void testBudgetFallbackDoesNotDisablePointLookupsOrProvenEmptyRanges() throws Ex PredicateBuilder.and( builder.equal(0, 1), builder.equal(1, 7), builder.equal(2, 0)); assertThat(reader.visitComposite(point).join().get().results().isEmpty()).isFalse(); + Predicate multiPoint = + PredicateBuilder.and( + builder.equal(0, 1), + builder.in(1, Arrays.asList(7, 8)), + builder.in(2, Arrays.asList(0, 1))); + assertThat( + reader.visitComposite(multiPoint) + .join() + .get() + .results() + .getLongCardinality()) + .isEqualTo(5); + Predicate nullPoint = + PredicateBuilder.and(builder.isNull(0), builder.isNull(1), builder.isNull(2)); + assertThat(reader.visitComposite(nullPoint).join().get().results().getLongCardinality()) + .isEqualTo(1); assertThat( reader.visitComposite(builder.greaterThan(0, Integer.MAX_VALUE)) .join() @@ -181,6 +199,70 @@ void testBudgetFallbackDoesNotDisablePointLookupsOrProvenEmptyRanges() throws Ex } } + @ParameterizedTest + @ValueSource(ints = {1, 2}) + void testInAndNullIntervalsAcrossBlocks(int version) throws Exception { + Options options = options(version); + List files = write(options); + List queries = + Arrays.asList( + PredicateBuilder.and( + builder.in(0, Arrays.asList(-1, 1, 1, null)), + builder.between(1, -1, 7)), + PredicateBuilder.and( + builder.equal(0, 1), + in(1, -1, 7, 8, null), + builder.in(2, Arrays.asList(0, 1))), + PredicateBuilder.and(in(0, -1, 1), in(1, -1, 7), in(2, 0, 1)), + PredicateBuilder.and(builder.equal(0, 1), in(1, -1, 7, null), in(1, 7, 8)), + PredicateBuilder.and(builder.equal(0, 1), in(1)), + PredicateBuilder.and(builder.equal(0, 1), in(1, (Object) null)), + PredicateBuilder.and(builder.isNull(0), builder.greaterThan(1, 0)), + PredicateBuilder.and( + builder.isNull(0), builder.isNull(1), builder.isNull(2)), + PredicateBuilder.and( + builder.equal(0, 1), builder.equal(1, 7), builder.isNull(2)), + builder.isNull(0), + builder.isNotNull(0), + PredicateBuilder.and(builder.equal(0, 1), builder.isNotNull(1)), + PredicateBuilder.and(builder.isNull(0), builder.isNotNull(0)), + PredicateBuilder.and(builder.isNull(0), builder.equal(0, 1)), + PredicateBuilder.and(builder.isNull(0), builder.lessThan(0, 7))); + ExecutorService executor = newDirectExecutorService(); + try (GlobalIndexReader reader = + GlobalIndexer.create("btree", type.getFields(), options) + .createReader( + file -> io.newInputStream(file.filePath()), + files, + rows.size(), + null, + executor)) { + for (Predicate query : queries) { + RoaringNavigableMap64 expected = new RoaringNavigableMap64(); + for (int i = 0; i < rows.size(); i++) { + if (query.test(rows.get(i))) { + expected.add(i); + } + } + assertThat(reader.visitComposite(query).join()).as("%s", query).isPresent(); + assertThat(reader.visitComposite(query).join().get().results().toRangeList()) + .as("%s", query) + .containsExactlyElementsOf(expected.toRangeList()); + } + } finally { + executor.shutdownNow(); + } + } + + private Predicate in(int field, Object... values) { + return new LeafPredicate( + In.INSTANCE, + type.getTypeAt(field), + field, + type.getFieldNames().get(field), + Arrays.asList(values)); + } + private Options options(int version) { Options options = new Options(); options.set(BTreeIndexOptions.BTREE_INDEX_FILE_VERSION, version); @@ -202,6 +284,7 @@ private List write(Options options) throws Exception { } rows.add(GenericRow.of(1, 7, 0)); rows.add(GenericRow.of(1, 7, null)); + rows.add(GenericRow.of(null, null, null)); CompositeKeySerializer serializer = new CompositeKeySerializer(type); rows.sort((left, right) -> serializer.createComparator().compare(left, right)); Path directory = new Path(tempPath.toUri()); 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 f962282552b3..af6a1b03dc2c 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 @@ -147,6 +147,12 @@ private static GlobalIndexQuery createForPredicate( } CompoundPredicate compound = (CompoundPredicate) predicate; boolean union = compound.function() instanceof Or; + if (union) { + CompositeCandidate composite = selectComposite(predicate, groups, options); + if (composite != null) { + return composite.query(); + } + } List children = new ArrayList<>(); List predicates = GlobalIndexEvaluator.normalizedChildren(compound); if (!union) { @@ -159,17 +165,16 @@ private static GlobalIndexQuery createForPredicate( children.add(selected.query()); predicates.removeAll(selected.plan.predicates()); // Later key conditions remain data filters; do not expand scalar postings for them. - Set keyColumns = + Set keyFields = selected.group.indexFields().stream() - .map(DataField::name) + .map(DataField::id) .collect(Collectors.toSet()); predicates.removeIf( - child -> - child instanceof LeafPredicate - && ((LeafPredicate) child) - .fieldRefOptional() - .filter(field -> keyColumns.contains(field.name())) - .isPresent()); + child -> { + Set residualFields = collectFieldIds(rowType, child); + return !residualFields.isEmpty() + && keyFields.containsAll(residualFields); + }); } } @@ -257,7 +262,7 @@ private static CompositeCandidate selectComposite( if (plan.boundColumns() == 1) { List scalarCoverage = Range.sortAndMergeOverlap( - fieldGroups.stream() + groups.get(group.field.id()).stream() .filter(other -> other.extraFields.isEmpty()) .map(other -> other.range) .collect(Collectors.toList()), 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 c315a73b2002..844317fc57f0 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 @@ -39,6 +39,7 @@ import org.apache.paimon.stats.SimpleStats; import org.apache.paimon.table.source.DataSplit; import org.apache.paimon.table.source.SplitSerializer; +import org.apache.paimon.types.DataField; import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; @@ -84,6 +85,39 @@ /** Tests predicate type preservation and unsupported predicates in index queries. */ class GlobalIndexQueryTest { + @Test + void testCompositeNonNullRangeLeavesSmallInAsDataFilter() { + RowType rowType = + RowType.of( + new DataField(0, "number", DataTypes.INT()), + new DataField(1, "category", DataTypes.STRING())); + PredicateBuilder b = new PredicateBuilder(rowType); + List files = + Arrays.asList( + new IndexFileMeta( + "btree", + "scalar", + 1, + 100, + new GlobalIndexMeta(0, 99, 0, null, null), + null), + new IndexFileMeta( + "btree", + "composite", + 1, + 100, + new GlobalIndexMeta(0, 99, 1, new int[] {0}, null), + null)); + IndexPathFactory paths = mock(IndexPathFactory.class); + when(paths.toPath(any(IndexFileMeta.class))).thenReturn(new Path("index")); + Predicate predicate = PredicateBuilder.and(b.isNotNull(1), b.in(0, Arrays.asList(7, 8))); + GlobalIndexQuery query = + GlobalIndexQuery.create(rowType, predicate, files, paths, new Options()); + assertThat(query.hasCompositeQuery()).isTrue(); + assertThat(query.hasScalarQuery()).isFalse(); + assertThat(query.contributingFieldIds(rowType)).containsExactly(1); + } + @TempDir java.nio.file.Path tempDir; @ParameterizedTest diff --git a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexTestUtils.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexTestUtils.java index 58a98ade99b5..ff6e57f9c65c 100644 --- a/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexTestUtils.java +++ b/paimon-core/src/test/java/org/apache/paimon/globalindex/sorted/SortedGlobalIndexTestUtils.java @@ -20,9 +20,13 @@ import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.globalindex.CompositeKeySerializer; import org.apache.paimon.globalindex.GlobalIndexKeyExtractor; import org.apache.paimon.globalindex.KeySerializer; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.manifest.IndexManifestEntry; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.SpecialFields; @@ -30,22 +34,51 @@ import org.apache.paimon.table.source.DataSplit; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.CloseableIterator; +import org.apache.paimon.utils.IOUtils; import org.apache.paimon.utils.InternalRowUtils; import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.Range; +import java.io.Closeable; +import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import static org.apache.paimon.globalindex.GlobalIndexBuilderUtils.calcRowRange; +import static org.assertj.core.api.Assertions.assertThat; /** Utilities for building sorted indexes in tests without exposing a test-only production API. */ public final class SortedGlobalIndexTestUtils { private SortedGlobalIndexTestUtils() {} + /** Delete physical index files without changing manifests, and restore them on close. */ + public static Closeable deleteIndexFiles(FileStoreTable table, int fieldCount) + throws IOException { + Map files = new LinkedHashMap<>(); + for (IndexManifestEntry entry : table.store().newIndexFileHandler().scanEntries()) { + IndexFileMeta file = entry.indexFile(); + if (file.globalIndexMeta().getIndexedFieldIds().size() == fieldCount) { + Path path = table.store().pathFactory().globalIndexFileFactory().toPath(file); + files.put(path, IOUtils.readFully(table.fileIO().newInputStream(path), true)); + assertThat(table.fileIO().delete(path, false)).isTrue(); + } + } + assertThat(files).isNotEmpty(); + return () -> { + for (Map.Entry file : files.entrySet()) { + try (PositionOutputStream out = + table.fileIO().newOutputStream(file.getKey(), false)) { + out.write(file.getValue()); + } + } + }; + } + public static List buildIndex( FileStoreTable table, String indexType, 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 9e6361a82bf1..9c8d68fa8f71 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,86 @@ /** Composite lookups must avoid single-column postings and retain uncovered data. */ class CompositeBTreeTableTest extends DataEvolutionTestBase { + @Test + void testInAndNullQueriesAvoidScalarPostingsAcrossReaderModes() 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"))); + write.write(GenericRow.of(null, null, BinaryString.fromString("null-both"))); + commit.commit(write.prepareCommit()); + } + build(table, "f0"); + build(table, "f1", "f0"); + deleteIndexFiles(table, 1); + FileStoreTable withoutIndex = + table.copy( + Collections.singletonMap(CoreOptions.GLOBAL_INDEX_ENABLED.key(), "false")); + PredicateBuilder b = new PredicateBuilder(table.rowType()); + List categories = + new ArrayList<>( + Arrays.asList( + BinaryString.fromString("category-a"), + BinaryString.fromString("category-b"), + null)); + Predicate small = b.in(1, categories); + for (int i = 0; i < 21; i++) { + categories.add(BinaryString.fromString("absent-" + i)); + } + Predicate numbers = b.in(0, Arrays.asList(7, 8, 7, null)); + Predicate joint = PredicateBuilder.and(small, numbers); + List queries = + Arrays.asList( + joint, + PredicateBuilder.and(b.in(1, categories), numbers), + PredicateBuilder.and(small, b.greaterThan(0, 7)), + PredicateBuilder.and(joint, b.greaterThan(0, 7)), + PredicateBuilder.and(b.isNull(1), b.greaterThan(0, 6)), + PredicateBuilder.and(b.isNull(1), b.isNull(0)), + PredicateBuilder.and( + b.equal(1, BinaryString.fromString("category-a")), b.isNull(0)), + PredicateBuilder.and(b.isNotNull(1), numbers), + PredicateBuilder.and(b.isNull(1), b.isNotNull(1)), + PredicateBuilder.and(b.equal(1, null), b.isNull(0)), + PredicateBuilder.or(b.isNull(1), joint)); + for (Predicate query : queries) { + List expected = read(withoutIndex, query); + for (boolean inReader : Arrays.asList(false, true)) { + for (String mode : Arrays.asList("fast", "full", "detail")) { + List actual; + try { + actual = read(configured(table, mode, inReader), query); + } catch (Exception e) { + throw new AssertionError( + query + ", reader=" + inReader + ", mode=" + mode, e); + } + assertThat(actual) + .as("%s, reader=%s, mode=%s", query, inReader, mode) + .containsExactlyInAnyOrderElementsOf(expected); + } + } + } + List indexed = read(withoutIndex, joint); + append(table, 60, 80); + for (boolean inReader : Arrays.asList(false, true)) { + assertThat(read(configured(table, "fast", inReader), joint)) + .containsExactlyInAnyOrderElementsOf(indexed); + for (String mode : Arrays.asList("full", "detail")) { + assertThat(read(configured(table, mode, inReader), joint)) + .containsExactlyInAnyOrderElementsOf(read(withoutIndex, joint)); + } + } + } + @Test void testPrefixAndRangeQueriesAcrossReaderModes() throws Exception { createTableDefault(); @@ -209,7 +289,15 @@ void testOneOverBudgetShardFallsBackForTheWholeDefinition() throws Exception { PredicateBuilder.and( b.equal(1, BinaryString.fromString("category-b")), b.greaterThan(0, 8))); - for (Predicate query : Arrays.asList(range, conjunction, union)) { + Predicate multiRange = + PredicateBuilder.and( + b.in( + 1, + Arrays.asList( + BinaryString.fromString("category-a"), + BinaryString.fromString("category-b"))), + b.between(0, 7, 8)); + for (Predicate query : Arrays.asList(range, conjunction, union, multiRange)) { List expected = read(withoutIndex, query); for (boolean inReader : Arrays.asList(false, true)) { for (String mode : Arrays.asList("fast", "full", "detail")) { 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 d409a8955e7b..a73f9b8dbe4c 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 @@ -23,6 +23,7 @@ import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.globalindex.IndexedSplit; +import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils; import org.apache.paimon.index.DataEvolutionIndexSourceMeta; import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.io.DataFileMeta; @@ -40,6 +41,7 @@ import org.apache.paimon.table.source.ReadBuilder; import org.apache.paimon.table.source.TableScan; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.ExceptionUtils; import org.apache.paimon.utils.InternalRowUtils; import org.apache.paimon.utils.Range; @@ -49,6 +51,8 @@ import org.apache.flink.types.Row; import org.junit.jupiter.api.Test; +import java.io.Closeable; +import java.io.FileNotFoundException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -61,6 +65,7 @@ import java.util.stream.IntStream; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Test case for sorted global indexes. */ public class SortedGlobalIndexITCase extends CatalogITCaseBase { @@ -111,7 +116,6 @@ public void testCompositeBTreeIndex() throws Exception { "INSERT INTO T_COMPOSITE VALUES " + "(100, CAST(NULL AS STRING), 7), (101, 'category-a', CAST(NULL AS INT))"); buildBTreeIndexForTable("T_COMPOSITE", "category"); - buildBTreeIndexForTable("T_COMPOSITE", "item_number"); // Reverse the schema order to verify the index preserves the requested key order. buildBTreeIndexForTable("T_COMPOSITE", "item_number, category"); List compositeEntries = @@ -135,35 +139,58 @@ public void testCompositeBTreeIndex() throws Exception { }); assertThat(compositeEntries.stream().mapToLong(entry -> entry.indexFile().rowCount()).sum()) .isEqualTo(42); - for (boolean inReader : Arrays.asList(false, true)) { - sql( - "ALTER TABLE T_COMPOSITE SET ('global-index.query-in-reader.enabled' = '" - + inReader - + "', 'scalar-index.search-mode' = 'fast')"); - assertThat( - sql( - "SELECT id FROM T_COMPOSITE WHERE item_number = 7 AND category = 'category-a'")) - .containsExactlyInAnyOrder(Row.of(7), Row.of(27)); - assertThat( - sql( - "SELECT id FROM T_COMPOSITE WHERE category = 'category-a' AND item_number = 7")) - .containsExactlyInAnyOrder(Row.of(7), Row.of(27)); - assertThat( - 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)); + // Keep the trailing scalar definition, but make its postings unavailable. + try (Closeable ignored = + SortedGlobalIndexTestUtils.deleteIndexFiles(paimonTable("T_COMPOSITE"), 1)) { + for (boolean inReader : Arrays.asList(false, true)) { + sql( + "ALTER TABLE T_COMPOSITE SET ('global-index.query-in-reader.enabled' = '" + + inReader + + "', 'scalar-index.search-mode' = 'fast')"); + assertCompositeBTreeQuery( + "item_number = 7 AND category = 'category-a'", Row.of(7), Row.of(27)); + assertCompositeBTreeQuery( + "category = 'category-a' AND item_number = 7", Row.of(7), Row.of(27)); + assertCompositeBTreeQuery("category = 'absent' AND item_number = 7"); + assertCompositeBTreeQuery( + "item_number = 7", + Row.of(7), + Row.of(17), + Row.of(27), + Row.of(37), + Row.of(100)); + assertCompositeBTreeQuery( + "item_number = 7 AND category > 'category-a'", Row.of(17), Row.of(37)); + assertCompositeBTreeQuery( + "item_number = 7 AND category BETWEEN 'category-a' AND 'category-b'", + Row.of(7), + Row.of(17), + Row.of(27), + Row.of(37)); + assertCompositeBTreeQuery( + "item_number IN (7, 8) AND category IN ('category-a', 'category-b')", + Row.of(7), + Row.of(8), + Row.of(17), + Row.of(18), + Row.of(27), + Row.of(28), + Row.of(37), + Row.of(38)); + assertCompositeBTreeQuery( + "item_number IN (7, 8) AND category > 'category-a'", + Row.of(17), + Row.of(18), + Row.of(37), + Row.of(38)); + assertCompositeBTreeQuery( + "item_number IS NULL AND category = 'category-a'", Row.of(101)); + assertCompositeBTreeQuery("item_number = 7 AND category IS NULL", Row.of(100)); + assertCompositeBTreeQuery( + "item_number IS NOT NULL AND category IS NULL", Row.of(100)); + } } + buildBTreeIndexForTable("T_COMPOSITE", "item_number"); insertCompositeRows(40, 60); buildBTreeIndexForTable("T_COMPOSITE", "item_number,category"); assertThat( @@ -208,6 +235,35 @@ public void testCompositeBTreeIndex() throws Exception { .containsExactlyInAnyOrder(1, 2); } + private void assertCompositeBTreeQuery(String predicate, Row... expected) throws Exception { + String query = "SELECT id FROM T_COMPOSITE WHERE " + predicate; + assertThat(sql(query)).containsExactlyInAnyOrder(expected); + if (expected.length == 0) { + return; + } + FileStoreTable table = paimonTable("T_COMPOSITE"); + List compositeFiles = + table.store().newIndexFileHandler().scanEntries().stream() + .map(IndexManifestEntry::indexFile) + .filter(file -> file.globalIndexMeta().getIndexedFieldIds().size() == 2) + .map(IndexFileMeta::fileName) + .collect(Collectors.toList()); + try (Closeable ignored = SortedGlobalIndexTestUtils.deleteIndexFiles(table, 2)) { + assertThatThrownBy(() -> sql(query)) + .satisfies( + failure -> { + FileNotFoundException missing = + ExceptionUtils.findThrowable( + failure, FileNotFoundException.class) + .orElseThrow(() -> new AssertionError(failure)); + assertThat( + compositeFiles.stream() + .anyMatch(missing.getMessage()::contains)) + .isTrue(); + }); + } + } + private void insertCompositeRows(int from, int to) { String values = IntStream.range(from, to) diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala index f0a5453d38f9..ecdc5128e9e4 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/SparkExpressionConverter.scala @@ -240,7 +240,7 @@ object SparkExpressionConverter { /** Convert Spark [[Literal]] to Paimon literal. */ def toPaimonLiteral(literal: Literal[_]): Object = { - if (literal == null) { + if (literal == null || literal.value() == null) { return null } 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 ed820634dac9..6351fb55d1d5 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 @@ -18,13 +18,17 @@ package org.apache.paimon.spark.procedure +import org.apache.paimon.globalindex.{IndexedSplit, IndexQuerySplit} +import org.apache.paimon.globalindex.sorted.SortedGlobalIndexTestUtils import org.apache.paimon.options.Options import org.apache.paimon.spark.PaimonSparkTestBase import org.apache.paimon.spark.globalindex.sorted.SortedIndexTopoBuilder import org.apache.paimon.types.DataField +import org.apache.paimon.utils.ExceptionUtils import org.apache.spark.sql.Row +import java.io.FileNotFoundException import java.util.Collections import scala.collection.JavaConverters._ @@ -79,7 +83,6 @@ class CompositeBTreeIndexProcedureTest extends PaimonSparkTestBase { } insert(0, 40) sql("INSERT INTO T VALUES (100, NULL, 7), (101, 'category-a', NULL)") - build("category") build("item_number") build("category,item_number") val indexes = loadTable("T").store().newIndexFileHandler().scanEntries().asScala @@ -90,25 +93,95 @@ class CompositeBTreeIndexProcedureTest extends PaimonSparkTestBase { composite.forall( _.indexFile().globalIndexMeta().getIndexedFieldIds().asScala.toSeq == Seq(1, 2))) assert(composite.map(_.indexFile().rowCount()).sum == 42L) - for (inReader <- Seq(false, true)) { - sql( - s"ALTER TABLE T SET TBLPROPERTIES ('global-index.query-in-reader.enabled' = '$inReader', 'scalar-index.search-mode' = 'fast')") - checkAnswer( - sql("SELECT id FROM T WHERE item_number = 7 AND category = 'category-a'"), - Seq(Row(7), Row(27))) - 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))) + // Keep the trailing scalar definition, but make its postings unavailable. + val scalarFiles = SortedGlobalIndexTestUtils.deleteIndexFiles(loadTable("T"), 1) + try { + for (inReader <- Seq(false, true)) { + sql( + s"ALTER TABLE T SET TBLPROPERTIES ('global-index.query-in-reader.enabled' = '$inReader', 'scalar-index.search-mode' = 'fast')") + def checkCompositeAnswer(query: String, expected: Seq[Row]): Unit = { + val splits = + try { + getPaimonScan(query).inputSplits + } catch { + case failure: Exception => + throw new AssertionError(s"Failed to plan $query, inReader=$inReader", failure) + } + if (expected.nonEmpty) { + assert(splits.nonEmpty) + assert( + splits.forall( + split => + if (inReader) split.isInstanceOf[IndexQuerySplit] + else split.isInstanceOf[IndexedSplit])) + } else if (!inReader) { + assert(splits.isEmpty) + } + try { + checkAnswer(sql(query), expected) + } catch { + case failure: Exception => + throw new AssertionError(s"Failed to read $query, inReader=$inReader", failure) + } + if (expected.nonEmpty) { + val compositeFiles = loadTable("T") + .store() + .newIndexFileHandler() + .scanEntries() + .asScala + .map(_.indexFile()) + .filter(_.globalIndexMeta().getIndexedFieldIds().size() == 2) + .map(_.fileName()) + val removed = SortedGlobalIndexTestUtils.deleteIndexFiles(loadTable("T"), 2) + try { + val failure = intercept[Exception] { + sql(query).collect() + } + val missing = ExceptionUtils.findThrowable(failure, classOf[FileNotFoundException]) + assert(missing.isPresent) + assert(compositeFiles.exists(missing.get().getMessage.contains)) + } finally { + removed.close() + } + } + } + checkCompositeAnswer( + "SELECT id FROM T WHERE item_number = 7 AND category = 'category-a'", + Seq(Row(7), Row(27))) + checkCompositeAnswer( + "SELECT id FROM T WHERE category = 'absent' AND item_number = 7", + Seq.empty) + checkCompositeAnswer( + "SELECT id FROM T WHERE category = 'category-a'", + ((0 until 10) ++ (20 until 30) :+ 101).map(Row(_))) + checkCompositeAnswer( + "SELECT id FROM T WHERE category = 'category-a' AND item_number > 7", + Seq(Row(8), Row(9), Row(28), Row(29))) + checkCompositeAnswer( + "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))) + checkCompositeAnswer( + "SELECT id FROM T WHERE category IN ('category-a', 'category-b') AND item_number IN (7, 8)", + Seq(Row(7), Row(8), Row(17), Row(18), Row(27), Row(28), Row(37), Row(38)) + ) + checkCompositeAnswer( + "SELECT id FROM T WHERE category IN ('category-a', 'category-b') AND item_number > 7", + Seq(Row(8), Row(9), Row(18), Row(19), Row(28), Row(29), Row(38), Row(39)) + ) + checkCompositeAnswer( + "SELECT id FROM T WHERE category IS NULL AND item_number = 7", + Seq(Row(100))) + checkCompositeAnswer( + "SELECT id FROM T WHERE category = 'category-a' AND item_number IS NULL", + Seq(Row(101))) + checkCompositeAnswer( + "SELECT id FROM T WHERE category IN ('category-a', NULL) AND item_number IN (7, 8)", + Seq(Row(7), Row(8), Row(27), Row(28))) + } + } finally { + scalarFiles.close() } + build("category") insert(40, 60) build("category,item_number") checkAnswer(