This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 71e3c92ff4 [core] Support composite BTree IN and NULL queries (#10347)
71e3c92ff4 is described below
commit 71e3c92ff4c184fa87bcef1a6af37507b2d7cad8
Author: Jingsong Lee <[email protected]>
AuthorDate: Sat Oct 3 17:55:05 2026 +0800
[core] Support composite BTree IN and NULL queries (#10347)
---
docs/docs/multimodal-table/global-index/btree.mdx | 26 +-
.../paimon/globalindex/btree/BTreeIndexReader.java | 42 +--
.../globalindex/btree/CompositeBTreePredicate.java | 327 ++++++++++++++-------
.../globalindex/btree/LazyFilteredBTreeReader.java | 4 +-
.../globalindex/btree/CompositeBTreeIndexTest.java | 10 +-
.../btree/CompositeBTreePredicateTest.java | 87 +++++-
.../globalindex/btree/CompositeBTreeRangeTest.java | 83 ++++++
.../paimon/globalindex/GlobalIndexQuery.java | 23 +-
.../paimon/globalindex/GlobalIndexQueryTest.java | 34 +++
.../sorted/SortedGlobalIndexTestUtils.java | 33 +++
.../paimon/table/CompositeBTreeTableTest.java | 90 +++++-
.../paimon/flink/SortedGlobalIndexITCase.java | 114 +++++--
.../spark/util/SparkExpressionConverter.scala | 2 +-
.../CompositeBTreeIndexProcedureTest.scala | 111 +++++--
14 files changed, 785 insertions(+), 201 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index/btree.mdx
b/docs/docs/multimodal-table/global-index/btree.mdx
index 7da34552eb..c34c611c10 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 26584d4260..2ab03fb49e 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 class BTreeIndexReader implements Closeable {
if (plan.isEmpty()) {
return new RoaringNavigableMap64();
}
- if (plan.isPointLookup()) {
- return pointQuery(plan.pointKey());
- }
RoaringNavigableMap64 result = new RoaringNavigableMap64();
- SstFileReader.SstFileIterator iterator =
reader.createIterator();
- iterator.seekTo(
- key ->
- plan.lower()
- .compareKey(
- (InternalRow)
keySerializer.deserialize(key)));
- BlockIterator batch;
- while ((batch = iterator.readBatch()) != null) {
- while (batch.hasNext()) {
- Map.Entry<MemorySlice, MemorySlice> entry =
batch.next();
- InternalRow key =
- (InternalRow)
keySerializer.deserialize(entry.getKey());
- if (plan.upper().compareKey(key) > 0) {
- return result;
+ 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<MemorySlice, MemorySlice> 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 3acccc79a6..f39bde5c1f 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.KeySerializer;
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.Collections;
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> plan(List<DataField> fields, Predicate
predicate) {
List<Predicate> conjuncts = PredicateBuilder.splitAnd(predicate);
List<Predicate> covered = new ArrayList<>();
- List<Object> prefix = new ArrayList<>();
+ List<Object[]> prefixes = new ArrayList<>();
+ prefixes.add(new Object[0]);
List<LeafPredicate> range = Collections.emptyList();
- boolean empty = false;
+ int equalColumns = 0;
for (DataField field : fields) {
List<LeafPredicate> column = new ArrayList<>();
- LeafPredicate equality = null;
+ List<Predicate> 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<List<Object>> 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<Object[]> 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<Interval> 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<List<Object>> pointValues(DataField field,
List<LeafPredicate> 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<Object> candidates =
+ selected.function() instanceof IsNull
+ ? Collections.singletonList(null)
+ : selected.literals();
+ Comparator<Object> comparator =
KeySerializer.create(field.type()).createComparator();
+ TreeSet<Object> 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<Object> 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 @@ public final class CompositeBTreePredicate {
|| 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 @@ public final class CompositeBTreePredicate {
}
}
- /** 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<DataField> fields;
private final List<Predicate> predicates;
- private final Bound lower;
- private final Bound upper;
+ private final List<Interval> intervals;
private final int boundColumns;
private final int equalColumns;
- private final boolean empty;
private Plan(
List<DataField> fields,
List<Predicate> predicates,
- Bound lower,
- Bound upper,
+ List<Interval> 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<Predicate> predicates() {
return predicates;
}
- public Bound lower() {
- return lower;
- }
-
- public Bound upper() {
- return upper;
+ public List<Interval> intervals() {
+ return intervals;
}
public int boundColumns() {
@@ -257,17 +377,13 @@ public final class CompositeBTreePredicate {
}
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 final class CompositeBTreePredicate {
}
public List<GlobalIndexIOMeta> selectFiles(List<GlobalIndexIOMeta>
files) {
- if (empty) {
+ if (isEmpty()) {
return Collections.emptyList();
}
// Preserve conservative pruning when metadata is unavailable.
@@ -299,18 +415,19 @@ public final class CompositeBTreePredicate {
List<GlobalIndexIOMeta> result = new ArrayList<>();
for (GlobalIndexIOMeta file : files) {
SortedIndexFileMeta meta =
SortedIndexFileMeta.deserialize(file.metadata());
- if (meta.firstKey() == null
- || meta.lastKey() == null
- || (lower.compareKey(
- (InternalRow)
- serializer.deserialize(
-
MemorySlice.wrap(meta.lastKey())))
- > 0
- && upper.compareKey(
- (InternalRow)
- serializer.deserialize(
-
MemorySlice.wrap(meta.firstKey())))
- < 0)) {
+ 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 6ec85d813e..fcbdf92595 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 class LazyFilteredBTreeReader extends
SortedFileGlobalIndexReader<BTreeIn
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<GlobalIndexIOMeta> 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 b642583541..df3975a4f2 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 @@ class CompositeBTreeIndexTest {
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 4a63cdd80a..4bea80c1fc 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.apache.paimon.types.RowType;
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 @@ class CompositeBTreePredicateTest {
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 @@ class CompositeBTreePredicateTest {
builder.equal(1, number), builder.equal(0,
BinaryString.fromString(category)));
}
+ @Test
+ void testDiscreteDomainsIntersectDeduplicateAndBoundExpansion() {
+ List<Object> categories = new ArrayList<>();
+ List<Object> 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<DataField> 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<GlobalIndexIOMeta> 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 58a1602851..ce8e564446 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.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.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 @@ class CompositeBTreeRangeTest {
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 @@ class CompositeBTreeRangeTest {
}
}
+ @ParameterizedTest
+ @ValueSource(ints = {1, 2})
+ void testInAndNullIntervalsAcrossBlocks(int version) throws Exception {
+ Options options = options(version);
+ List<GlobalIndexIOMeta> files = write(options);
+ List<Predicate> 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 @@ class CompositeBTreeRangeTest {
}
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 f962282552..af6a1b03dc 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 @@ class GlobalIndexQuery {
}
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<GlobalIndexQuery> children = new ArrayList<>();
List<Predicate> predicates =
GlobalIndexEvaluator.normalizedChildren(compound);
if (!union) {
@@ -159,17 +165,16 @@ class GlobalIndexQuery {
children.add(selected.query());
predicates.removeAll(selected.plan.predicates());
// Later key conditions remain data filters; do not expand
scalar postings for them.
- Set<String> keyColumns =
+ Set<Integer> 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<Integer> residualFields =
collectFieldIds(rowType, child);
+ return !residualFields.isEmpty()
+ && keyFields.containsAll(residualFields);
+ });
}
}
@@ -257,7 +262,7 @@ class GlobalIndexQuery {
if (plan.boundColumns() == 1) {
List<Range> 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 c315a73b20..844317fc57 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.predicate.PredicateBuilder;
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 @@ import static org.mockito.Mockito.when;
/** 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<IndexFileMeta> 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 58a98ade99..ff6e57f9c6 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 @@ package org.apache.paimon.globalindex.sorted;
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.sink.CommitMessage;
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<Path, byte[]> 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<Path, byte[]> file : files.entrySet()) {
+ try (PositionOutputStream out =
+ table.fileIO().newOutputStream(file.getKey(), false)) {
+ out.write(file.getValue());
+ }
+ }
+ };
+ }
+
public static List<CommitMessage> 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 9e6361a82b..9c8d68fa8f 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 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Composite lookups must avoid single-column postings and retain uncovered
data. */
class CompositeBTreeTableTest extends DataEvolutionTestBase {
+ @Test
+ void 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<Object> 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<Predicate> 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<String> expected = read(withoutIndex, query);
+ for (boolean inReader : Arrays.asList(false, true)) {
+ for (String mode : Arrays.asList("fast", "full", "detail")) {
+ List<String> 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<String> 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 @@ class CompositeBTreeTableTest extends
DataEvolutionTestBase {
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<String> 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 d409a8955e..a73f9b8dbe 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.BinaryString;
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.sink.CommitMessageImpl;
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.table.api.config.TableConfigOptions;
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.Collectors;
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 class SortedGlobalIndexITCase extends
CatalogITCaseBase {
"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<IndexManifestEntry> compositeEntries =
@@ -135,35 +139,58 @@ public class SortedGlobalIndexITCase extends
CatalogITCaseBase {
});
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 class SortedGlobalIndexITCase extends
CatalogITCaseBase {
.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<String> 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 f0a5453d38..ecdc5128e9 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 ed820634da..6351fb55d1 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(
