diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/ContainsRefinementEvaluator.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/ContainsRefinementEvaluator.java index a5e138a11b1c..410df5070338 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/ContainsRefinementEvaluator.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/ContainsRefinementEvaluator.java @@ -132,7 +132,8 @@ CompletableFuture> evaluate( new GlobalIndexEvaluator.Evaluation( GlobalIndexResult.createEmpty(), emptyResultFields, - null))); + null, + false))); } } List groupedContains = new ArrayList<>(groups.values()); @@ -171,7 +172,8 @@ CompletableFuture> evaluate( GlobalIndexResult .createEmpty(), emptyResultFields, - null))); + null, + false))); } } @@ -201,8 +203,10 @@ CompletableFuture> evaluate( exactFutures .size() + 1); - if (remainingEvaluation - .isPresent()) { + // An absent result for existing + // remaining conditions is not + // the same as no conditions. + if (!remaining.isEmpty()) { results.add( remainingEvaluation); } @@ -269,10 +273,15 @@ private CompletableFuture> visitContai result -> result.map( value -> + // Legacy wrappers can retain only supported + // contains literals. This opaque result does not + // prove completeness; search already treats + // contains as candidate-only. new GlobalIndexEvaluator.Evaluation( value, Collections.singleton(group.fieldId), - null))); + null, + false))); } private CompletableFuture> intersectReaderResults( diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java index d43193aae0a8..fc5b00f95615 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexEvaluator.java @@ -157,11 +157,27 @@ private CompletableFuture> visitLeafAsync(LeafPredicate pre private CompletableFuture> visitFieldAsync( FieldRef fieldRef, Function>> visitor) { + return visitFieldWithCompletenessAsync( + fieldRef, + reader -> + visitor.apply(reader) + .thenApply( + result -> + result.map( + value -> + new GlobalIndexPredicateResult( + value, true)))); + } + + private CompletableFuture> visitFieldWithCompletenessAsync( + FieldRef fieldRef, + Function>> + visitor) { int fieldId = rowType.getField(fieldRef.name()).id(); Collection readers = indexReadersCache.computeIfAbsent(fieldId, readersFunction::apply); - List>> readerFutures = + List>> readerFutures = new ArrayList<>(readers.size()); for (GlobalIndexReader reader : readers) { readerFutures.add(visitor.apply(reader)); @@ -171,26 +187,37 @@ private CompletableFuture> visitFieldAsync( .thenApply( v -> { Optional compoundResult = Optional.empty(); - for (CompletableFuture> f : readerFutures) { - Optional childResult = f.join(); + boolean fullyEvaluated = false; + for (CompletableFuture> f : + readerFutures) { + Optional childResult = f.join(); if (!childResult.isPresent()) { continue; } + // One complete alternative index suffices for this predicate. + // Intersecting safe candidates cannot invalidate that proof. + fullyEvaluated |= childResult.get().fullyEvaluated(); if (compoundResult.isPresent()) { compoundResult = Optional.of( - compoundResult.get().and(childResult.get())); + compoundResult + .get() + .and(childResult.get().result())); } else { - compoundResult = childResult; + compoundResult = Optional.of(childResult.get().result()); } if (compoundResult.get().results().isEmpty()) { break; } } - return compoundResult.map( - result -> + return compoundResult.isPresent() + ? Optional.of( new Evaluation( - result, Collections.singleton(fieldId), null)); + compoundResult.get(), + Collections.singleton(fieldId), + null, + fullyEvaluated)) + : Optional.empty(); }); } @@ -225,10 +252,10 @@ private CompletableFuture> visitCompoundAsync( LeafPredicate upper = isLowerBound(first) ? second : first; FieldRef field = first.fieldRefOptional().get(); range = - visitFieldAsync( + visitFieldWithCompletenessAsync( field, reader -> - reader.visitRange( + reader.visitRangeWithCompleteness( field, lower.literals().get(0), upper.literals().get(0), @@ -271,6 +298,11 @@ static boolean isLowerBound(LeafPredicate leaf) { private Optional combineResults( List> results, CompoundPredicate predicate) { Set contributingFieldIds = new HashSet<>(); + // Field coverage is insufficient: a > 0 AND isNaN(a) still contributes field a even + // when the reader declines isNaN. Track every child, including before an empty shortcut. + boolean fullyEvaluated = + results.stream() + .allMatch(child -> child.isPresent() && child.get().fullyEvaluated()); if (predicate.function() instanceof Or) { GlobalIndexResult compoundResult = GlobalIndexResult.createEmpty(); for (Optional child : results) { @@ -280,7 +312,8 @@ private Optional combineResults( compoundResult = compoundResult.or(child.get().result()); contributingFieldIds.addAll(child.get().contributingFieldIds()); } - return Optional.of(new Evaluation(compoundResult, contributingFieldIds, null)); + return Optional.of( + new Evaluation(compoundResult, contributingFieldIds, null, fullyEvaluated)); } else { Optional compoundResult = Optional.empty(); for (Optional child : results) { @@ -297,7 +330,8 @@ private Optional combineResults( break; } } - return compoundResult.map(result -> new Evaluation(result, contributingFieldIds, null)); + return compoundResult.map( + result -> new Evaluation(result, contributingFieldIds, null, fullyEvaluated)); } } @@ -309,12 +343,15 @@ public static final class Evaluation { private final GlobalIndexResult result; private final Set contributingFieldIds; @Nullable private final List coveredRanges; + private final boolean fullyEvaluated; Evaluation( GlobalIndexResult result, Collection contributingFieldIds, - @Nullable List coveredRanges) { + @Nullable List coveredRanges, + boolean fullyEvaluated) { this.result = result; + this.fullyEvaluated = fullyEvaluated; this.coveredRanges = coveredRanges == null ? null @@ -335,6 +372,14 @@ public List coveredRanges() { public Set contributingFieldIds() { return contributingFieldIds; } + + /** + * Whether all necessary predicates were evaluated within the reported coverage. This does + * not prove full row-range coverage or exactness of candidate-only index operators. + */ + public boolean fullyEvaluated() { + return fullyEvaluated; + } } /** Shared normalization for eager evaluation and metadata-only index planning. */ diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexPredicateResult.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexPredicateResult.java new file mode 100644 index 000000000000..39910c1c6ccd --- /dev/null +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexPredicateResult.java @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.globalindex; + +/** Candidates and whether every condition sent to a reader was evaluated. */ +public final class GlobalIndexPredicateResult { + + private final GlobalIndexResult result; + private final boolean fullyEvaluated; + + public GlobalIndexPredicateResult(GlobalIndexResult result, boolean fullyEvaluated) { + this.result = result; + this.fullyEvaluated = fullyEvaluated; + } + + public GlobalIndexResult result() { + return result; + } + + /** Completeness is necessary, but not sufficient, for an exact match set. */ + public boolean fullyEvaluated() { + return fullyEvaluated; + } +} diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java index 537ca1f0bf00..7d93615d3d48 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexReader.java @@ -89,6 +89,17 @@ default CompletableFuture> visitArrayContainsAll( */ default CompletableFuture> visitRange( FieldRef fieldRef, Object from, Object to, boolean fromInclusive, boolean toInclusive) { + return visitRangeWithCompleteness(fieldRef, from, to, fromInclusive, toInclusive) + .thenApply(result -> result.map(GlobalIndexPredicateResult::result)); + } + + /** + * Evaluate both range bounds without mistaking a supported single bound for the whole range. + * Readers with a combined scan can override this method to retain that optimization and report + * whether both bounds were evaluated. Existing readers default to evaluating the two leaves. + */ + default CompletableFuture> visitRangeWithCompleteness( + FieldRef fieldRef, Object from, Object to, boolean fromInclusive, boolean toInclusive) { CompletableFuture> lower = fromInclusive ? visitGreaterOrEqual(fieldRef, from) @@ -99,9 +110,12 @@ default CompletableFuture> visitRange( upper, (left, right) -> { if (!left.isPresent()) { - return right; + return right.map(result -> new GlobalIndexPredicateResult(result, false)); } - return right.isPresent() ? Optional.of(left.get().and(right.get())) : left; + return Optional.of( + new GlobalIndexPredicateResult( + right.isPresent() ? left.get().and(right.get()) : left.get(), + right.isPresent())); }); } diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java index 2e2d9166e689..e6877bf013c3 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactory.java @@ -34,6 +34,17 @@ default boolean supportsFullTextSearch() { return false; } + /** + * Whether a grouped index definition can serve independent scalar lookups on its fields. + * + *

Grouped fields are queried independently, unlike the ordered columns of a composite key. + * Readers may still decline individual predicates; coverage must include only the shards that + * actually evaluated the predicate. + */ + default boolean supportsGroupedIndexScalarQueries() { + return false; + } + /** * Select index files using metadata only. Implementations must retain every file that may * match; the default keeps all files when no safe pruning is available. diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactoryUtils.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactoryUtils.java index 6e045ec79c6e..247a59bf1933 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactoryUtils.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexerFactoryUtils.java @@ -58,6 +58,12 @@ public static GlobalIndexerFactory load(String type) { return globalIndexerFactory; } + /** Unknown types must not opt grouped definitions into independent scalar lookups. */ + public static boolean supportsGroupedIndexScalarQueries(String type) { + GlobalIndexerFactory factory = factories.get(type); + return factory != null && factory.supportsGroupedIndexScalarQueries(); + } + /** Keep unrecognized index types for the reader to resolve at execution time. */ public static List selectFiles( String type, diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java index 70283659ec0a..6b4ec78ba2b8 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/OffsetGlobalIndexReader.java @@ -191,6 +191,19 @@ public CompletableFuture> visitRange( .thenApply(this::applyOffset); } + @Override + public CompletableFuture> visitRangeWithCompleteness( + FieldRef fieldRef, Object from, Object to, boolean fromInclusive, boolean toInclusive) { + return wrapped.visitRangeWithCompleteness(fieldRef, from, to, fromInclusive, toInclusive) + .thenApply( + result -> + result.map( + value -> + new GlobalIndexPredicateResult( + value.result().offset(offset), + value.fullyEvaluated()))); + } + @Override public CompletableFuture> visitBetween( FieldRef fieldRef, Object from, Object to) { diff --git a/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java b/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java index dd2cb1eeaa90..2ebba5e2871f 100644 --- a/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java +++ b/paimon-common/src/main/java/org/apache/paimon/globalindex/UnionGlobalIndexReader.java @@ -188,6 +188,48 @@ public CompletableFuture> visitRange( reader -> reader.visitRange(fieldRef, from, to, fromInclusive, toInclusive)); } + @Override + public CompletableFuture> visitRangeWithCompleteness( + FieldRef fieldRef, Object from, Object to, boolean fromInclusive, boolean toInclusive) { + long start = durationConsumer == null ? 0L : System.nanoTime(); + List>> futures = new ArrayList<>(); + for (GlobalIndexReader reader : readers) { + futures.add( + reader.visitRangeWithCompleteness( + fieldRef, from, to, fromInclusive, toInclusive)); + } + CompletableFuture> result = + CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) + .thenApply( + ignored -> { + GlobalIndexResult union = null; + boolean fullyEvaluated = true; + for (CompletableFuture> + future : futures) { + Optional current = + future.join(); + if (!current.isPresent()) { + return Optional.empty(); + } + GlobalIndexPredicateResult value = current.get(); + union = + union == null + ? value.result() + : union.or(value.result()); + fullyEvaluated &= value.fullyEvaluated(); + } + return union == null + ? Optional.empty() + : Optional.of( + new GlobalIndexPredicateResult( + union, fullyEvaluated)); + }); + return durationConsumer == null + ? result + : result.whenComplete( + (ignored, throwable) -> durationConsumer.accept(System.nanoTime() - start)); + } + @Override public CompletableFuture> visitBetween( FieldRef fieldRef, Object from, Object to) { 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 fcbdf9259560..b16466554719 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 @@ -19,6 +19,7 @@ package org.apache.paimon.globalindex.btree; import org.apache.paimon.globalindex.GlobalIndexIOMeta; +import org.apache.paimon.globalindex.GlobalIndexPredicateResult; import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.globalindex.KeySerializer; import org.apache.paimon.globalindex.SortedFileGlobalIndexReader; @@ -278,6 +279,15 @@ public CompletableFuture> visitRange( reader -> reader.visitRange(from, to, fromInclusive, toInclusive)); } + @Override + public CompletableFuture> visitRangeWithCompleteness( + FieldRef fieldRef, Object from, Object to, boolean fromInclusive, boolean toInclusive) { + // The combined BTree scan applies both bounds; do not replace it with two postings scans. + return visitRange(fieldRef, from, to, fromInclusive, toInclusive) + .thenApply( + result -> result.map(value -> new GlobalIndexPredicateResult(value, true))); + } + @Override protected Optional visitBetween( BTreeIndexReader reader, Object from, Object to) { diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java index e3d9942ff203..9a6066aaf928 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/GlobalIndexEvaluatorTest.java @@ -94,14 +94,16 @@ void testRangeFallbackAndOrBoundary() { GlobalIndexReader reader = new StubGlobalIndexReader(null) { @Override - public CompletableFuture> visitRange( - FieldRef field, - Object from, - Object to, - boolean fromInclusive, - boolean toInclusive) { + public CompletableFuture> + visitRangeWithCompleteness( + FieldRef field, + Object from, + Object to, + boolean fromInclusive, + boolean toInclusive) { rangeCalls.incrementAndGet(); - return super.visitRange(field, from, to, fromInclusive, toInclusive); + return super.visitRangeWithCompleteness( + field, from, to, fromInclusive, toInclusive); } @Override @@ -154,6 +156,215 @@ void testSingleFieldSequential() { evaluator.close(); } + @Test + void testCompletenessTracksPredicatesNotFields() { + RowType type = RowType.of(new DataField(0, "price", DataTypes.DOUBLE())); + PredicateBuilder b = new PredicateBuilder(type); + GlobalIndexReader reader = + new StubGlobalIndexReader(resultOf(1, 2)) { + @Override + public CompletableFuture> visitGreaterThan( + FieldRef field, Object literal) { + return CompletableFuture.completedFuture(Optional.of(resultOf(1, 2))); + } + }; + try (GlobalIndexEvaluator evaluator = + new GlobalIndexEvaluator(type, field -> Collections.singletonList(reader))) { + Predicate supported = PredicateBuilder.and(b.equal(0, 10.0), b.greaterThan(0, 0.0)); + Predicate partial = PredicateBuilder.and(b.greaterThan(0, 0.0), b.isNaN(0)); + GlobalIndexEvaluator.Evaluation complete = + evaluator.evaluateWithContributingFields(supported).get(); + GlobalIndexEvaluator.Evaluation incomplete = + evaluator.evaluateWithContributingFields(partial).get(); + assertThat(complete.contributingFieldIds()) + .isEqualTo(incomplete.contributingFieldIds()); + assertThat(complete.fullyEvaluated()).isTrue(); + assertThat(incomplete.fullyEvaluated()).isFalse(); + assertThat( + evaluator + .evaluateWithContributingFields( + PredicateBuilder.or(partial, supported)) + .get() + .fullyEvaluated()) + .isFalse(); + assertThat( + evaluator + .evaluateWithContributingFields( + PredicateBuilder.and( + supported, + PredicateBuilder.or( + b.equal(0, 5.0), b.isNaN(0)))) + .get() + .fullyEvaluated()) + .isFalse(); + assertThat( + evaluator.evaluateWithContributingFields( + PredicateBuilder.or(b.equal(0, 5.0), b.isNaN(0)))) + .isEmpty(); + } + } + + @Test + void testRangeCompletenessSurvivesReaderWrappers() throws Exception { + FieldRef field = new FieldRef(0, "a", DataTypes.INT()); + for (boolean upperSupported : Arrays.asList(false, true)) { + GlobalIndexReader delegate = + new StubGlobalIndexReader(null) { + @Override + public CompletableFuture> visitGreaterOrEqual( + FieldRef ref, Object value) { + return CompletableFuture.completedFuture(Optional.of(resultOf(1, 2))); + } + + @Override + public CompletableFuture> visitLessThan( + FieldRef ref, Object value) { + return CompletableFuture.completedFuture( + upperSupported ? Optional.of(resultOf(2)) : Optional.empty()); + } + }; + try (GlobalIndexReader reader = + new UnionGlobalIndexReader( + Collections.singletonList( + new OffsetGlobalIndexReader(delegate, 10, 19)))) { + GlobalIndexPredicateResult result = + reader.visitRangeWithCompleteness(field, 0, 5, true, false).get().get(); + assertThat(result.fullyEvaluated()).isEqualTo(upperSupported); + assertBitmapContainsExactly( + result.result().results(), + upperSupported ? new long[] {12} : new long[] {11, 12}); + try (GlobalIndexEvaluator evaluator = + new GlobalIndexEvaluator( + rowType(), id -> Collections.singletonList(reader))) { + PredicateBuilder b = new PredicateBuilder(rowType()); + assertThat( + evaluator + .evaluateWithContributingFields( + PredicateBuilder.and( + b.greaterOrEqual(0, 0), + b.lessThan(0, 5))) + .get() + .fullyEvaluated()) + .isEqualTo(upperSupported); + } + } + } + } + + @Test + void testAlternativeIndexMayProvideCompleteAnswer() { + Predicate filter = new PredicateBuilder(rowType()).equal(0, 1); + try (GlobalIndexEvaluator evaluator = + new GlobalIndexEvaluator( + rowType(), + id -> + Arrays.asList( + new StubGlobalIndexReader(null), + new StubGlobalIndexReader(resultOf(1))))) { + assertThat(evaluator.evaluateWithContributingFields(filter).get().fullyEvaluated()) + .isTrue(); + } + } + + @Test + void testRangeUnionRequiresCompletenessFromEveryShard() throws Exception { + GlobalIndexReader complete = + new StubGlobalIndexReader(null) { + @Override + public CompletableFuture> + visitRangeWithCompleteness( + FieldRef field, + Object from, + Object to, + boolean lowerInclusive, + boolean upperInclusive) { + return CompletableFuture.completedFuture( + Optional.of(new GlobalIndexPredicateResult(resultOf(1), true))); + } + }; + GlobalIndexReader partial = + new StubGlobalIndexReader(null) { + @Override + public CompletableFuture> + visitRangeWithCompleteness( + FieldRef field, + Object from, + Object to, + boolean lowerInclusive, + boolean upperInclusive) { + return CompletableFuture.completedFuture( + Optional.of(new GlobalIndexPredicateResult(resultOf(2), false))); + } + }; + FieldRef field = new FieldRef(0, "a", DataTypes.INT()); + try (GlobalIndexReader union = + new UnionGlobalIndexReader(Arrays.asList(complete, partial))) { + GlobalIndexPredicateResult result = + union.visitRangeWithCompleteness(field, 0, 5, true, false).get().get(); + assertBitmapContainsExactly(result.result().results(), 1L, 2L); + assertThat(result.fullyEvaluated()).isFalse(); + } + try (GlobalIndexReader union = + new UnionGlobalIndexReader( + Arrays.asList(complete, new StubGlobalIndexReader(null)))) { + assertThat(union.visitRangeWithCompleteness(field, 0, 5, true, false).get()).isEmpty(); + } + } + + @Test + void testEmptyShortcutDoesNotInventCompleteness() { + PredicateBuilder b = new PredicateBuilder(rowType()); + try (GlobalIndexEvaluator evaluator = + new GlobalIndexEvaluator( + rowType(), + id -> + Collections.singletonList( + new StubGlobalIndexReader( + GlobalIndexResult.createEmpty())))) { + GlobalIndexEvaluator.Evaluation result = + evaluator + .evaluateWithContributingFields( + PredicateBuilder.and(b.equal(0, 1), b.lessThan(0, 5))) + .get(); + assertThat(result.result().results()).isEmpty(); + assertThat(result.fullyEvaluated()).isFalse(); + } + } + + @Test + void testContainsDoesNotForgetUnsupportedRemainingCondition() { + RowType type = RowType.of(new DataField(0, "name", DataTypes.STRING())); + PredicateBuilder b = new PredicateBuilder(type); + ContainsRefiningGlobalIndexReader reader = + new StubContainsRefiningGlobalIndexReader() { + @Override + public CompletableFuture> visitContainsCandidates( + FieldRef ref, List literals, GlobalIndexResult candidates) { + return CompletableFuture.completedFuture(Optional.of(resultOf(1))); + } + + @Override + public CompletableFuture> visitContainsConjunction( + FieldRef ref, List literals, GlobalIndexResult candidates) { + return CompletableFuture.completedFuture(Optional.of(resultOf(1))); + } + }; + try (GlobalIndexEvaluator evaluator = + new GlobalIndexEvaluator(type, id -> Collections.singletonList(reader))) { + GlobalIndexEvaluator.Evaluation result = + evaluator + .evaluateWithContributingFields( + PredicateBuilder.and( + b.contains(0, BinaryString.fromString("a")), + b.startsWith(0, BinaryString.fromString("ab")) + .negate() + .get())) + .get(); + assertBitmapContainsExactly(result.result().results(), 1L); + assertThat(result.fullyEvaluated()).isFalse(); + } + } + @Test void testEvaluateAsyncDoesNotWaitForReaderResult() { RowType rowType = rowType(); diff --git a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java index 658d74cafbcb..ee3c4e2111fa 100644 --- a/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java +++ b/paimon-common/src/test/java/org/apache/paimon/globalindex/btree/LazyFilteredBTreeIndexReaderTest.java @@ -172,6 +172,7 @@ public void testFusedRangeThroughEvaluatorAndWrappers() throws Exception { } assertResult(result.result(), expected); assertThat(result.contributingFieldIds()).containsExactly(1); + assertThat(result.fullyEvaluated()).isTrue(); if (comparator.compare(bounds[0], bounds[1]) <= 0) { verify(btreeReader) .visitRange( diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexCoverage.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexCoverage.java index 6ff33b520203..2d759cc8dfcf 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexCoverage.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexCoverage.java @@ -70,13 +70,17 @@ public DataEvolutionGlobalIndexCoverage( if (!DataEvolutionGlobalIndexScanner.isIndexInSchema(table.rowType(), indexFile)) { continue; } - // Single-field coverage belongs to single-field index definitions. Multi-field - // query paths supply their actual coverage explicitly. - if (meta.extraFieldIds() != null && meta.extraFieldIds().length > 0) { + // Composite keys do not imply independent coverage for each of their columns. + if (meta.extraFieldIds() != null + && meta.extraFieldIds().length > 0 + && !GlobalIndexerFactoryUtils.supportsGroupedIndexScalarQueries( + indexFile.indexType())) { continue; } Range range = new Range(meta.rowRangeStart(), meta.rowRangeEnd()); - addCoverage(meta.indexFieldId(), range); + for (int fieldId : meta.getIndexedFieldIds()) { + addCoverage(fieldId, range); + } } } diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java index e294cc09d118..f9183f90fc8d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionGlobalIndexScanner.java @@ -220,7 +220,7 @@ public static Optional create( return create(table, null, null, indexFiles); } - /** Search pre-filters use per-column coverage, so they require single-field definitions. */ + /** Search pre-filters require indexes that can serve independent scalar lookups. */ public static Optional createForScalarFilters( FileStoreTable table, @Nullable Snapshot pinnedSnapshot, @@ -231,7 +231,12 @@ public static Optional createForScalarFilters( pinnedSnapshot, partitionFilter, indexFiles.stream() - .filter(file -> !isMultiFieldIndex(file)) + .filter( + file -> + !isMultiFieldIndex(file) + || GlobalIndexerFactoryUtils + .supportsGroupedIndexScalarQueries( + file.indexType())) .collect(Collectors.toList())); } @@ -410,7 +415,7 @@ public Optional scanWithCoverage(Predicate pred ? null : GlobalIndexQuery.create( rowType, predicate, indexFiles, indexPathFactory, options); - if (query != null && query.hasCompositeQuery()) { + if (query != null && query.hasMultiFieldQuery()) { try { return query.evaluateWithCoverage( fileIO, @@ -419,7 +424,7 @@ public Optional scanWithCoverage(Predicate pred .map(file -> file.globalIndexMeta().rowRange()) .collect(Collectors.toList())); } catch (IOException e) { - throw new RuntimeException("Failed to evaluate composite global index query", e); + throw new RuntimeException("Failed to evaluate multi-field global index query", e); } } return globalIndexEvaluator.evaluateWithContributingFields(predicate); @@ -446,7 +451,7 @@ public GlobalIndexResult unindexedRows(Predicate predicate) { ? null : GlobalIndexQuery.create( rowType, predicate, indexFiles, indexPathFactory, options); - if (query != null && query.hasCompositeQuery() && query.hasScalarQuery()) { + if (query != null && query.hasMultiFieldQuery() && query.hasScalarQuery()) { // Scalar shards can decline a predicate at runtime, so the legacy API must use // evaluated coverage as well. Internal callers retain the evaluation directly. Optional evaluation = scanWithCoverage(predicate); @@ -456,7 +461,7 @@ public GlobalIndexResult unindexedRows(Predicate predicate) { coverage.unindexedRangesFromCoverage(Collections.emptyList(), null)); } List unindexed = - query != null && query.hasCompositeQuery() + query != null && query.hasMultiFieldQuery() ? coverage.unindexedRangesFromCoverage(query.coveredRanges(), null) : coverage.unindexedRanges(rowType, predicate); for (Range range : unindexed) { @@ -496,13 +501,13 @@ private Collection createReaders( GlobalIndexFileReader indexFileReadWrite, IndexMetaFileGroup group, RowType rowType) { List indexFields = group.fieldIds.stream().map(rowType::getField).collect(Collectors.toList()); - if (indexFields.size() > 1) { - return Collections.emptyList(); - } - Set readers = new HashSet<>(); for (Map.Entry>> entry : group.metas.entrySet()) { String indexType = entry.getKey(); + if (indexFields.size() > 1 + && !GlobalIndexerFactoryUtils.supportsGroupedIndexScalarQueries(indexType)) { + continue; + } Map> metas = entry.getValue(); GlobalIndexerFactory globalIndexerFactory = GlobalIndexerFactoryUtils.load(indexType); GlobalIndexer globalIndexer = globalIndexerFactory.create(indexFields, options); 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 af6a1b03dc2c..7fe4898863c5 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 @@ -90,15 +90,20 @@ class GlobalIndexQuery { /** A query's index groups; empty when none overlap this data split. */ private final List groups; + /** Runtime proof relative to the original filter, not just the retained child queries. */ + private final boolean fullyPlanned; + private GlobalIndexQuery( @Nullable Predicate predicate, boolean union, List children, - List groups) { + List groups, + boolean fullyPlanned) { this.predicate = predicate; this.union = union; this.children = children; this.groups = groups; + this.fullyPlanned = fullyPlanned; } @Nullable @@ -154,6 +159,7 @@ private static GlobalIndexQuery createForPredicate( } } List children = new ArrayList<>(); + boolean fullyPlanned = true; List predicates = GlobalIndexEvaluator.normalizedChildren(compound); if (!union) { while (!predicates.isEmpty()) { @@ -169,12 +175,14 @@ private static GlobalIndexQuery createForPredicate( selected.group.indexFields().stream() .map(DataField::id) .collect(Collectors.toSet()); - predicates.removeIf( - child -> { - Set residualFields = collectFieldIds(rowType, child); - return !residualFields.isEmpty() - && keyFields.containsAll(residualFields); - }); + boolean residual = + predicates.removeIf( + child -> { + Set residualFields = collectFieldIds(rowType, child); + return !residualFields.isEmpty() + && keyFields.containsAll(residualFields); + }); + fullyPlanned &= !residual; } } @@ -214,13 +222,16 @@ private static GlobalIndexQuery createForPredicate( if (union) { return null; } + fullyPlanned = false; } else { children.add(query); + fullyPlanned &= query.fullyPlanned; } } return children.isEmpty() ? null - : new GlobalIndexQuery(null, union, children, Collections.emptyList()); + : new GlobalIndexQuery( + null, union, children, Collections.emptyList(), fullyPlanned); } @Nullable @@ -314,7 +325,8 @@ private GlobalIndexQuery query() { PredicateBuilder.and(plan.predicates()), false, Collections.emptyList(), - groups); + groups, + true); } } @@ -330,15 +342,17 @@ private static GlobalIndexQuery createForIndexedField( } List selectedGroups = new ArrayList<>(); for (IndexGroup group : fieldGroups) { - // Scalar lookups use independent single-field index definitions. - if (group.extraFields.isEmpty()) { + // Composite-key indexes require their own planner, not independent field lookups. + if (group.extraFields.isEmpty() + || GlobalIndexerFactoryUtils.supportsGroupedIndexScalarQueries(group.type)) { selectedGroups.add(group.selectFiles(predicate)); } } if (selectedGroups.isEmpty()) { return null; } - return new GlobalIndexQuery(predicate, false, Collections.emptyList(), selectedGroups); + return new GlobalIndexQuery( + predicate, false, Collections.emptyList(), selectedGroups, true); } /** Only prune when metadata proves no match throughout the requested row ranges. */ @@ -358,6 +372,11 @@ boolean hasCompositeQuery() { || children.stream().anyMatch(GlobalIndexQuery::hasCompositeQuery); } + boolean hasMultiFieldQuery() { + return groups.stream().anyMatch(group -> !group.extraFields.isEmpty()) + || children.stream().anyMatch(GlobalIndexQuery::hasMultiFieldQuery); + } + boolean hasScalarQuery() { return groups.stream().anyMatch(group -> !group.isCompositeBTree()) || children.stream().anyMatch(GlobalIndexQuery::hasScalarQuery); @@ -423,7 +442,7 @@ GlobalIndexQuery forRanges(List ranges) { selectedChildren.add(child.forRanges(ranges)); } // Keep children without local groups so readers can retain their uncertain candidates. - return new GlobalIndexQuery(predicate, union, selectedChildren, selected); + return new GlobalIndexQuery(predicate, union, selectedChildren, selected, fullyPlanned); } GlobalIndexResult evaluate(FileIO fileIO, Options options, List ranges) @@ -493,6 +512,7 @@ private Optional evaluateWithExecutor( GlobalIndexResult result = null; Set fields = new HashSet<>(); List coverage = null; + boolean fullyEvaluated = fullyPlanned; for (GlobalIndexQuery child : children) { Optional evaluation = child.evaluateWithExecutor( @@ -501,8 +521,10 @@ private Optional evaluateWithExecutor( if (union) { return Optional.empty(); } + fullyEvaluated = false; continue; } + fullyEvaluated &= evaluation.get().fullyEvaluated(); GlobalIndexResult matches = evaluation.get().result(); result = result == null ? matches : union ? result.or(matches) : result.and(matches); @@ -512,12 +534,13 @@ private Optional evaluateWithExecutor( } return result == null ? Optional.empty() - : Optional.of(new Evaluation(result, fields, coverage)); + : Optional.of(new Evaluation(result, fields, coverage, fullyEvaluated)); } GlobalIndexResult result = GlobalIndexResult.createEmpty(); Set fields = new HashSet<>(); List coverage = new ArrayList<>(); boolean supported = groups.isEmpty(); + boolean fullyEvaluated = fullyPlanned; for (IndexGroup group : groups) { if (group.files.isEmpty()) { supported = true; @@ -525,8 +548,8 @@ private Optional evaluateWithExecutor( coverage.add(group.range); continue; } - Function>> query = - predicateQuery(group); + Function>> + query = predicateQuery(group); GlobalIndexResult splitRows = localSplitRows(ranges, group.range); if (splitRows.results().isEmpty()) { supported = true; @@ -541,7 +564,7 @@ private Optional evaluateWithExecutor( group.range.count(), splitRows.results().toRangeList(), executor)) { - Optional matches = query.apply(reader).get(); + Optional matches = query.apply(reader).get(); if (!matches.isPresent()) { if (allowUnsupported) { continue; @@ -549,10 +572,11 @@ private Optional evaluateWithExecutor( throw new IOException("Index reader does not support predicate: " + predicate); } supported = true; + fullyEvaluated &= matches.get().fullyEvaluated(); fields.addAll(collectFieldIds(new RowType(group.indexFields()), predicate)); coverage.add(group.range); // Clip in index-local coordinates before offset() iterates the retained rows. - result = result.or(splitRows.and(matches.get()).offset(group.range.from)); + result = result.or(splitRows.and(matches.get().result()).offset(group.range.from)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException("Interrupted while evaluating index query split", e); @@ -562,26 +586,44 @@ private Optional evaluateWithExecutor( } return supported ? Optional.of( - new Evaluation(result, fields, Range.sortAndMergeOverlap(coverage, true))) + new Evaluation( + result, + fields, + Range.sortAndMergeOverlap(coverage, true), + fullyEvaluated)) : Optional.empty(); } - private Function>> + private Function>> predicateQuery(IndexGroup group) { if (group.isCompositeBTree()) { - return reader -> reader.visitComposite(predicate); + return reader -> + reader.visitComposite(predicate) + .thenApply( + result -> + result.map( + value -> + new GlobalIndexPredicateResult( + value, true))); } if (predicate instanceof LeafPredicate) { LeafPredicate leaf = (LeafPredicate) predicate; return reader -> - leaf.function().visit(reader, leaf.fieldRefOptional().get(), leaf.literals()); + leaf.function() + .visit(reader, leaf.fieldRefOptional().get(), leaf.literals()) + .thenApply( + result -> + result.map( + value -> + new GlobalIndexPredicateResult( + value, true))); } List bounds = ((CompoundPredicate) predicate).children(); LeafPredicate lower = (LeafPredicate) bounds.get(0); LeafPredicate upper = (LeafPredicate) bounds.get(1); FieldRef field = lower.fieldRefOptional().get(); return reader -> - reader.visitRange( + reader.visitRangeWithCompleteness( field, lower.literals().get(0), upper.literals().get(0), @@ -671,7 +713,8 @@ static GlobalIndexQuery deserialize(DataInputView in) throws IOException { } groups.add(new IndexGroup(indexType, field, extraFields, range, files)); } - return new GlobalIndexQuery(predicate, false, Collections.emptyList(), groups); + // V1 transports candidate queries, not proof that planning retained the full filter. + return new GlobalIndexQuery(predicate, false, Collections.emptyList(), groups, false); } if (type != 1 && type != 2) { throw new IOException("Unknown index query node: " + type); @@ -681,7 +724,7 @@ static GlobalIndexQuery deserialize(DataInputView in) throws IOException { for (int i = 0; i < size; i++) { children.add(deserialize(in)); } - return new GlobalIndexQuery(null, type == 2, children, Collections.emptyList()); + return new GlobalIndexQuery(null, type == 2, children, Collections.emptyList(), false); } static void writeString(DataOutputView out, String value) throws IOException { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java index 5163bb580ef0..ea5f0c4cd062 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataEvolutionVectorRead.java @@ -27,6 +27,7 @@ import org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner; import org.apache.paimon.globalindex.GlobalIndexEvaluator; import org.apache.paimon.globalindex.GlobalIndexIOMeta; +import org.apache.paimon.globalindex.GlobalIndexReadThreadPool; import org.apache.paimon.globalindex.GlobalIndexReader; import org.apache.paimon.globalindex.GlobalIndexer; import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils; @@ -64,6 +65,7 @@ import java.util.Collections; import java.util.Comparator; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; @@ -73,6 +75,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutorService; +import static org.apache.paimon.CoreOptions.GLOBAL_INDEX_THREAD_NUM; import static org.apache.paimon.utils.Preconditions.checkNotNull; /** Base implementation for vector reads. */ @@ -100,6 +103,9 @@ public abstract class AbstractDataEvolutionVectorRead implements Serializable { */ @Nullable private transient RoaringNavigableMap64 scalarPreFilter; + /** Metric of indexed splits moved to the raw path for the current read. */ + @Nullable private transient String fallbackVectorMetric; + private static final Comparator WEAKEST_SCORE_FIRST = Comparator.comparingDouble(a -> Float.intBitsToFloat((int) a[1])) .thenComparing((a, b) -> Long.compare(b[0], a[0])); @@ -128,6 +134,62 @@ protected GlobalIndexer createGlobalIndexer(List splits) return createGlobalIndexer(firstFile.indexType(), firstMeta); } + /** One indexer per definition, aligned with the splits and shared across matching shards. */ + protected List createGlobalIndexers(List splits) { + return createGlobalIndexers(splits, createGlobalIndexer(splits)); + } + + protected List createGlobalIndexers( + List splits, GlobalIndexer first) { + Map, GlobalIndexer>> definitions = new HashMap<>(); + IndexFileMeta firstFile = firstVectorIndexFile(splits); + definitions + .computeIfAbsent(firstFile.indexType(), k -> new HashMap<>()) + .put(checkNotNull(firstFile.globalIndexMeta()).getIndexedFieldIds(), first); + List indexers = new ArrayList<>(splits.size()); + for (IndexVectorSearchSplit split : splits) { + indexers.add( + split.vectorIndexFiles().isEmpty() + ? first + : indexerForDefinition(split, definitions)); + } + return indexers; + } + + private GlobalIndexer indexerForDefinition( + IndexVectorSearchSplit split, + Map, GlobalIndexer>> definitions) { + IndexFileMeta file = split.vectorIndexFiles().get(0); + return definitions + .computeIfAbsent(file.indexType(), k -> new HashMap<>()) + .computeIfAbsent( + checkNotNull(file.globalIndexMeta()).getIndexedFieldIds(), + k -> createGlobalIndexer(Collections.singletonList(split))); + } + + /** Resolve metrics after every shard has registered its metadata with its own indexer. */ + protected GlobalIndexer metricIndexer(List indexers) { + GlobalIndexer first = indexers.get(0); + String metric = null; + for (GlobalIndexer indexer : indexers) { + if (indexer == first) { + continue; + } + if (metric == null) { + metric = rawSearchMetric(first); + } + String current = rawSearchMetric(indexer); + if (!metric.equals(current)) { + throw new IllegalArgumentException( + "Cannot combine vector indexes with different metrics: " + + metric + + " and " + + current); + } + } + return first; + } + protected GlobalIndexer createGlobalIndexer(String indexType) { return GlobalIndexerFactoryUtils.load(indexType) .create( @@ -152,9 +214,9 @@ private GlobalIndexer createGlobalIndexer(String indexType, GlobalIndexMeta meta * silently drops every covered row) nor ignore it (the top-K would be polluted by non-matching * rows). Instead, suppress the index splits and route their ranges through the raw search, * where the exact final-read filter decides; the raw read merges ranges, so overlap with an - * existing raw split is deduplicated. Splits routed this way reuse the vector index type so raw - * scoring keeps its metric. The computed scalar pre-filter is cached for {@link #preFilters} to - * reuse. + * existing raw split is deduplicated. Splits routed this way retain the metric from the vector + * index metadata, which can differ from the current table options. The computed scalar + * pre-filter is cached for {@link #preFilters} to reuse. */ protected void prepareSplits( List splits, @@ -162,6 +224,7 @@ protected void prepareSplits( List rawSplits) { splitSearchSplits(splits, indexSplits, rawSplits); scalarPreFilter = null; + fallbackVectorMetric = null; if (filter == null || indexSplits.isEmpty()) { return; } @@ -178,9 +241,52 @@ protected void prepareSplits( } rawSplits.add( new RawVectorSearchSplit(ranges, scalarIndexFiles, vectorIndexType(indexSplits))); + fallbackVectorMetric = fallbackVectorMetric(indexSplits); indexSplits.clear(); } + private String fallbackVectorMetric(List splits) { + Map, GlobalIndexer>> indexers = new HashMap<>(); + IndexPathFactory indexPathFactory = table.store().pathFactory().globalIndexFileFactory(); + ExecutorService executor = + GlobalIndexReadThreadPool.getExecutorService( + table.coreOptions().toConfiguration().get(GLOBAL_INDEX_THREAD_NUM)); + for (IndexVectorSearchSplit split : splits) { + if (split.vectorIndexFiles().isEmpty()) { + continue; + } + // Share an indexer within each definition so the backend retains its own policy for + // conflicting shard metrics. Different definitions must use their own field lists. + GlobalIndexer indexer = indexerForDefinition(split, indexers); + try (GlobalIndexReader ignored = + indexer.createReader( + meta -> table.fileIO().newInputStream(meta.filePath()), + buildIOMetaList(indexPathFactory, split.vectorIndexFiles()), + split.rowRangeEnd() - split.rowRangeStart() + 1, + null, + executor)) { + // Reader creation registers persisted metadata; no search is necessary. + } catch (IOException e) { + throw new RuntimeException("Failed to initialize vector fallback metric", e); + } + } + String metric = null; + for (Map, GlobalIndexer> definitions : indexers.values()) { + for (GlobalIndexer indexer : definitions.values()) { + String definitionMetric = rawSearchMetric(indexer); + if (metric != null && !metric.equals(definitionMetric)) { + throw new IllegalArgumentException( + "Cannot combine vector indexes with different metrics: " + + metric + + " and " + + definitionMetric); + } + metric = definitionMetric; + } + } + return metric; + } + protected List preFilters(List splits) { List indexedRowRanges = new ArrayList<>(splits.size()); for (IndexVectorSearchSplit split : splits) { @@ -253,6 +359,14 @@ private RoaringNavigableMap64 scalarMatchedRows(List spl if (!evaluation.isPresent()) { return null; } + List evaluatedRanges = evaluation.get().coveredRanges(); + List indexedRanges = splitRows.toRangeList(); + if (evaluatedRanges != null + && !Range.and(evaluatedRanges, indexedRanges).equals(indexedRanges)) { + // Metadata coverage is not proof that every scalar reader accepted the filter. + // Reuse the raw path to decide unsupported shards before vector top-k ranking. + return null; + } RoaringNavigableMap64 matched = evaluation.get().result().results(); if (FilteredRowIdReader.isExact(table.rowType(), filter, evaluation.get())) { return matched; @@ -299,9 +413,7 @@ protected RoaringNavigableMap64 rawPreFilter(List splits) return null; } include.or(result.get().result().results()); - include.or( - scanner.unindexedRowsForContributingFields(result.get().contributingFieldIds()) - .results()); + include.or(scanner.unindexedRowsForEvaluation(result.get()).results()); } catch (IOException e) { throw new RuntimeException(e); } @@ -712,13 +824,42 @@ private static void splitSearchSplits( List rawSplits) { for (VectorSearchSplit split : splits) { if (split instanceof IndexVectorSearchSplit) { - indexSplits.add((IndexVectorSearchSplit) split); + addIndexDefinitionSplits((IndexVectorSearchSplit) split, indexSplits); } else if (split instanceof RawVectorSearchSplit) { rawSplits.add((RawVectorSearchSplit) split); } } } + private static void addIndexDefinitionSplits( + IndexVectorSearchSplit split, List indexSplits) { + Map, List>> definitions = new LinkedHashMap<>(); + for (IndexFileMeta file : split.vectorIndexFiles()) { + definitions + .computeIfAbsent(file.indexType(), k -> new LinkedHashMap<>()) + .computeIfAbsent( + checkNotNull(file.globalIndexMeta()).getIndexedFieldIds(), + k -> new ArrayList<>()) + .add(file); + } + if (definitions.values().stream().mapToInt(Map::size).sum() <= 1) { + indexSplits.add(split); + return; + } + // A range can contain several definitions. Keep each reader's field list homogeneous + // without changing the serialized scan plan or the row-id offset of any shard. + for (Map, List> byFields : definitions.values()) { + for (List files : byFields.values()) { + indexSplits.add( + new IndexVectorSearchSplit( + split.rowRangeStart(), + split.rowRangeEnd(), + files, + split.scalarIndexFiles())); + } + } + } + protected static List rawRowRanges(List rawSplits) { List rawRowRanges = new ArrayList<>(); for (RawVectorSearchSplit split : rawSplits) { @@ -755,6 +896,9 @@ private float[] getVector(InternalRow row, int vectorIndex) { } protected String rawSearchMetric(@Nullable GlobalIndexer globalIndexer) { + if (fallbackVectorMetric != null) { + return fallbackVectorMetric; + } String metric = null; if (globalIndexer != null) { if (!(globalIndexer instanceof VectorGlobalIndexer)) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java index a94f0d6dba9b..aae9f22403fa 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionBatchVectorRead.java @@ -32,6 +32,7 @@ import javax.annotation.Nullable; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; @@ -79,12 +80,14 @@ private List readBatch(List splits) { return empty; } - GlobalIndexer globalIndexer = - indexSplits.isEmpty() ? null : createGlobalIndexer(indexSplits); + List globalIndexers = + indexSplits.isEmpty() ? Collections.emptyList() : createGlobalIndexers(indexSplits); ScoredGlobalIndexResult[] indexedResults = indexSplits.isEmpty() ? emptyScoredResults(n) - : readIndexedBatch(indexSplits, globalIndexer); + : readIndexedBatch(indexSplits, globalIndexers); + GlobalIndexer globalIndexer = + globalIndexers.isEmpty() ? null : metricIndexer(globalIndexers); List results = new ArrayList<>(n); RoaringNavigableMap64 rawPreFilter = rawPreFilter(rawSplits); @@ -97,7 +100,7 @@ private List readBatch(List splits) { } protected ScoredGlobalIndexResult[] readIndexedBatch( - List splits, GlobalIndexer globalIndexer) { + List splits, List globalIndexers) { int n = vectors.length; List preFilters = preFilters(splits); String indexType = vectorIndexType(splits); @@ -114,7 +117,7 @@ protected ScoredGlobalIndexResult[] readIndexedBatch( IndexVectorSearchSplit split = splits.get(i); futures.add( evalBatch( - globalIndexer, + globalIndexers.get(i), indexPathFactory, split.rowRangeStart(), split.rowRangeEnd(), @@ -145,7 +148,7 @@ protected ScoredGlobalIndexResult[] readIndexedBatch( for (int i = 0; i < n; i++) { merged[i] = ScoredGlobalIndexResult.merge(resultsByQuery.get(i)); } - return maybeRerankIndexedBatchResults(merged, indexType, globalIndexer); + return maybeRerankIndexedBatchResults(merged, indexType, metricIndexer(globalIndexers)); } protected ScoredGlobalIndexResult[] maybeRerankIndexedBatchResults( diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java index 1ed32cb51f00..e45108b19a8a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextRead.java @@ -231,6 +231,21 @@ private RoaringNavigableMap64 matchedRows( Optional evaluation = evaluateWithIndexes(scalarIndexFiles, planSnapshot); if (evaluation.isPresent()) { + if (evaluation.get().coveredRanges() != null) { + RoaringNavigableMap64 evaluated = new RoaringNavigableMap64(); + for (Range range : evaluation.get().coveredRanges()) { + evaluated.addRange(range); + } + if (table.coreOptions().scalarIndexSearchMode() != GlobalIndexSearchMode.FAST) { + RoaringNavigableMap64 unsupported = + RoaringNavigableMap64.or( + new RoaringNavigableMap64(), decidedByIndex); + unsupported.andNot(evaluated); + unindexed.or(unsupported); + } + // A shard that declined the predicate must not contribute candidates to top-k. + decidedByIndex.and(evaluated); + } RoaringNavigableMap64 fromIndex = RoaringNavigableMap64.and( evaluation.get().result().results(), decidedByIndex); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java index fbac5ef7b786..b87a502dfe8a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java @@ -84,6 +84,9 @@ protected GlobalIndexResult readSplits(List splits) protected ScoredGlobalIndexResult readIndexed( List splits, GlobalIndexer globalIndexer) { + // Keep the caller's indexer for the first definition so its persisted metric is also + // available to Flink/Spark callers when they subsequently search raw splits. + List globalIndexers = createGlobalIndexers(splits, globalIndexer); List preFilters = preFilters(splits); String indexType = vectorIndexType(splits); int searchLimit = indexedSearchLimit(indexType); @@ -99,7 +102,7 @@ protected ScoredGlobalIndexResult readIndexed( IndexVectorSearchSplit split = splits.get(i); futures.add( eval( - globalIndexer, + globalIndexers.get(i), indexPathFactory, split.rowRangeStart(), split.rowRangeEnd(), @@ -120,6 +123,9 @@ protected ScoredGlobalIndexResult readIndexed( } } return maybeRerankIndexedResult( - ScoredGlobalIndexResult.merge(results), indexType, globalIndexer, vector); + ScoredGlobalIndexResult.merge(results), + indexType, + metricIndexer(globalIndexers), + vector); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/FilteredRowIdReader.java b/paimon-core/src/main/java/org/apache/paimon/table/source/FilteredRowIdReader.java index da333e6ef583..557016ee601b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/FilteredRowIdReader.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/FilteredRowIdReader.java @@ -84,6 +84,10 @@ class FilteredRowIdReader { */ static boolean isExact( RowType rowType, Predicate filter, GlobalIndexEvaluator.Evaluation evaluation) { + // Same-field supported predicates do not prove that an unsupported sibling was evaluated. + if (!evaluation.fullyEvaluated()) { + return false; + } Set filterFieldIds = collectFieldIds(rowType, filter); if (!evaluation.contributingFieldIds().containsAll(filterFieldIds)) { return false; 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 844317fc57f0..c040d214046c 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 @@ -30,10 +30,13 @@ import org.apache.paimon.index.IndexFileMeta; import org.apache.paimon.index.IndexPathFactory; import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.DataInputDeserializer; import org.apache.paimon.io.DataOutputSerializer; import org.apache.paimon.options.Options; import org.apache.paimon.predicate.And; +import org.apache.paimon.predicate.CastTransform; import org.apache.paimon.predicate.CompoundPredicate; +import org.apache.paimon.predicate.FieldRef; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.stats.SimpleStats; @@ -76,6 +79,7 @@ import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.CALLS_REAL_METHODS; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.times; @@ -85,6 +89,70 @@ /** Tests predicate type preservation and unsupported predicates in index queries. */ class GlobalIndexQueryTest { + @ParameterizedTest + @ValueSource(ints = {0, 1}) + void testGroupedIndexScalarCoverageExcludesUnsupportedShards(int field) throws Exception { + RowType rowType = RowType.of(DataTypes.INT(), DataTypes.INT()); + Predicate predicate = new PredicateBuilder(rowType).equal(field, 7); + String type = "independent-fields"; + List files = + Arrays.asList( + indexFile(type, "supported", 10, 19, 0, new int[] {1}), + indexFile(type, "unsupported", 20, 29, 0, new int[] {1}), + indexFile("unknown-multi", "ignored", 30, 39, 0, new int[] {1})); + IndexPathFactory paths = mock(IndexPathFactory.class); + when(paths.toPath(any(IndexFileMeta.class))) + .thenAnswer( + invocation -> + new Path(invocation.getArgument(0).fileName())); + GlobalIndexReader supported = mock(GlobalIndexReader.class); + when(supported.visitEqual(any(), eq(7))) + .thenReturn( + CompletableFuture.completedFuture( + Optional.of(GlobalIndexResult.fromRange(new Range(2, 2))))); + GlobalIndexReader unsupported = mock(GlobalIndexReader.class); + when(unsupported.visitEqual(any(), eq(7))) + .thenReturn(CompletableFuture.completedFuture(Optional.empty())); + GlobalIndexer indexer = mock(GlobalIndexer.class); + when(indexer.createReader(any(), anyList(), eq(10L), anyList(), any())) + .thenAnswer( + invocation -> { + List selected = invocation.getArgument(1); + return selected.get(0).filePath().getName().equals("supported") + ? supported + : unsupported; + }); + GlobalIndexerFactory factory = mock(GlobalIndexerFactory.class); + when(factory.create(eq(rowType.getFields()), any())).thenReturn(indexer); + try (MockedStatic factories = + mockStatic(GlobalIndexerFactoryUtils.class, CALLS_REAL_METHODS)) { + factories + .when(() -> GlobalIndexerFactoryUtils.supportsGroupedIndexScalarQueries(type)) + .thenReturn(true); + factories.when(() -> GlobalIndexerFactoryUtils.load(type)).thenReturn(factory); + GlobalIndexQuery query = + GlobalIndexQuery.create(rowType, predicate, files, paths, new Options()); + assertThat(query).isNotNull(); + assertThat(query.hasMultiFieldQuery()).isTrue(); + assertThat(query.hasCompositeQuery()).isFalse(); + assertThat(query.coveredRanges()).containsExactly(new Range(10, 29)); + List ranges = Collections.singletonList(new Range(0, 39)); + GlobalIndexEvaluator.Evaluation evaluation = + query.evaluateWithCoverage(mock(FileIO.class), new Options(), ranges).get(); + assertThat(evaluation.result().results().toRangeList()) + .containsExactly(new Range(12, 12)); + assertThat(evaluation.coveredRanges()).containsExactly(new Range(10, 19)); + assertThat( + query.evaluateCandidates(mock(FileIO.class), new Options(), ranges) + .get() + .results() + .toRangeList()) + .containsExactly(new Range(0, 9), new Range(12, 12), new Range(20, 39)); + verify(supported, times(2)).close(); + verify(unsupported, times(2)).close(); + } + } + @Test void testCompositeNonNullRangeLeavesSmallInAsDataFilter() { RowType rowType = @@ -118,6 +186,105 @@ void testCompositeNonNullRangeLeavesSmallInAsDataFilter() { assertThat(query.contributingFieldIds(rowType)).containsExactly(1); } + @Test + void testGroupedCompletenessSurvivesPlanningClippingAndV1Restore() throws Exception { + RowType type = + RowType.of( + new DataField(0, "embedding", DataTypes.INT()), + new DataField(1, "price", DataTypes.DOUBLE()), + new DataField(2, "other", DataTypes.INT())); + PredicateBuilder b = new PredicateBuilder(type); + Predicate equal = b.equal(1, 10.0); + List predicates = + Arrays.asList( + equal, + PredicateBuilder.and(equal, b.isNotNull(1)), + PredicateBuilder.and(equal, b.isNaN(1)), + PredicateBuilder.and(equal, b.equal(2, 1)), + PredicateBuilder.or(PredicateBuilder.and(equal, b.isNaN(1)), equal), + // Unlike isNaN, this same-field predicate is dropped before reader + // execution. + PredicateBuilder.and( + equal, + b.equal( + new CastTransform( + new FieldRef(1, "price", DataTypes.DOUBLE()), + DataTypes.INT()), + 10)), + PredicateBuilder.and(b.greaterThan(1, 0.0), b.lessThan(1, 100.0))); + List files = + Collections.singletonList( + indexFile("test-grouped", "index", 0, 9, 0, new int[] {1})); + IndexPathFactory paths = mock(IndexPathFactory.class); + when(paths.toPath(any(IndexFileMeta.class))).thenReturn(new Path("index")); + GlobalIndexReader reader = mock(GlobalIndexReader.class, CALLS_REAL_METHODS); + when(reader.visitEqual(any(), any())) + .thenReturn( + CompletableFuture.completedFuture( + Optional.of(GlobalIndexResult.fromRange(new Range(2, 2))))); + when(reader.visitGreaterThan(any(), any())) + .thenReturn( + CompletableFuture.completedFuture( + Optional.of(GlobalIndexResult.fromRange(new Range(2, 2))))); + when(reader.visitLessThan(any(), any())) + .thenReturn(CompletableFuture.completedFuture(Optional.empty())); + GlobalIndexer indexer = mock(GlobalIndexer.class); + when(indexer.createReader(any(), anyList(), anyLong(), anyList(), any())) + .thenReturn(reader); + GlobalIndexerFactory factory = mock(GlobalIndexerFactory.class); + when(factory.create(anyList(), any())).thenReturn(indexer); + try (MockedStatic factories = + mockStatic(GlobalIndexerFactoryUtils.class, CALLS_REAL_METHODS)) { + factories + .when( + () -> + GlobalIndexerFactoryUtils.supportsGroupedIndexScalarQueries( + "test-grouped")) + .thenReturn(true); + factories + .when(() -> GlobalIndexerFactoryUtils.load("test-grouped")) + .thenReturn(factory); + List ranges = Collections.singletonList(new Range(0, 9)); + for (int i = 0; i < predicates.size(); i++) { + GlobalIndexQuery query = + GlobalIndexQuery.create( + type, predicates.get(i), files, paths, new Options()); + assertThat(query).isNotNull(); + for (GlobalIndexQuery local : Arrays.asList(query, query.forRanges(ranges))) { + GlobalIndexEvaluator.Evaluation result = + local.evaluateWithCoverage(mock(FileIO.class), new Options(), ranges) + .get(); + assertThat(result.fullyEvaluated()) + .as("%s", predicates.get(i)) + .isEqualTo(i < 2); + assertThat(result.result().results().toRangeList()) + .containsExactly(new Range(2, 2)); + } + DataOutputSerializer out = new DataOutputSerializer(256); + query.serialize(out); + GlobalIndexQuery restored = + GlobalIndexQuery.deserialize( + new DataInputDeserializer(out.getCopyOfBuffer())); + assertThat( + restored.evaluateWithCoverage( + mock(FileIO.class), new Options(), ranges) + .get() + .fullyEvaluated()) + .isFalse(); + assertThat( + restored.evaluateCandidates( + mock(FileIO.class), new Options(), ranges) + .get() + .results() + .toRangeList()) + .containsExactly(new Range(2, 2)); + DataOutputSerializer roundTrip = new DataOutputSerializer(256); + restored.serialize(roundTrip); + assertThat(roundTrip.getCopyOfBuffer()).isEqualTo(out.getCopyOfBuffer()); + } + } + } + @TempDir java.nio.file.Path tempDir; @ParameterizedTest @@ -195,14 +362,17 @@ void testBoundedRangeUsesSingleScan(boolean lowerInclusive, boolean upperInclusi assertThat(SplitSerializer.deserialize(SplitSerializer.serialize(split))).isEqualTo(split); GlobalIndexReader reader = mock(GlobalIndexReader.class); - when(reader.visitRange(any(), eq(5), eq(9), eq(lowerInclusive), eq(upperInclusive))) + when(reader.visitRangeWithCompleteness( + any(), eq(5), eq(9), eq(lowerInclusive), eq(upperInclusive))) .thenReturn( CompletableFuture.completedFuture( Optional.of( - GlobalIndexResult.fromRange( - new Range( - lowerInclusive ? 5 : 6, - upperInclusive ? 9 : 8))))); + new GlobalIndexPredicateResult( + GlobalIndexResult.fromRange( + new Range( + lowerInclusive ? 5 : 6, + upperInclusive ? 9 : 8)), + true)))); GlobalIndexer indexer = mock(GlobalIndexer.class); when(indexer.createReader(any(), anyList(), eq(100L), anyList(), any())).thenReturn(reader); GlobalIndexerFactory factory = mock(GlobalIndexerFactory.class); @@ -220,7 +390,9 @@ void testBoundedRangeUsesSingleScan(boolean lowerInclusive, boolean upperInclusi eq(100L), anyList(), any()); - verify(reader).visitRange(any(), eq(5), eq(9), eq(lowerInclusive), eq(upperInclusive)); + verify(reader) + .visitRangeWithCompleteness( + any(), eq(5), eq(9), eq(lowerInclusive), eq(upperInclusive)); } assertThat( diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/BTreeScalarFilterCompletenessTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/BTreeScalarFilterCompletenessTest.java new file mode 100644 index 000000000000..b2412d09fe5c --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/BTreeScalarFilterCompletenessTest.java @@ -0,0 +1,286 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.source; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericArray; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner; +import org.apache.paimon.globalindex.GlobalIndexBuilderUtils; +import org.apache.paimon.globalindex.GlobalIndexEvaluator; +import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexerFactory; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.TableTestBase; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.types.ArrayType; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.Range; + +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertAll; + +/** Scalar BTree candidates must satisfy the complete filter before vector ranking. */ +public class BTreeScalarFilterCompletenessTest extends TableTestBase { + private static final String[] NAMES = {"abacus", "alpha", "zebra"}; + private static final double[] PRICES = {10.0d, 20.0d, 30.0d}; + private static final float[][] VECTORS = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + + @ParameterizedTest + @CsvSource({"full,true", "full,false", "detail,true", "detail,false"}) + public void sameFieldUnsupportedStringMustNotRankNonMatchingRow(String mode, boolean refine) + throws Exception { + Fixture fixture = prepare(mode, refine); + PredicateBuilder b = new PredicateBuilder(fixture.table.rowType()); + // Both leaves use name, but only the first is supported. Field coverage cannot + // distinguish this from the fully supported conjunction tested below. + Predicate filter = + PredicateBuilder.and( + b.startsWith(1, BinaryString.fromString("a")), + b.startsWith(1, BinaryString.fromString("ab")).negate().get()); + Observation actual = observe(fixture, filter); + assertAll( + () -> assertThat(actual.scalar).containsExactly(1L), + () -> assertThat(actual.exact).as("partial predicate must not be exact").isFalse(), + () -> + assertThat(actual.vector) + .as("filter must be checked before top-k") + .isEqualTo( + refine + ? Collections.singletonList(1L) + : Collections.emptyList())); + } + + @ParameterizedTest + @CsvSource({"full,true", "full,false", "detail,true", "detail,false"}) + public void sameFieldUnsupportedIsNaNMustReturnNoRows(String mode, boolean refine) + throws Exception { + Fixture fixture = prepare(mode, refine); + PredicateBuilder b = new PredicateBuilder(fixture.table.rowType()); + Predicate filter = PredicateBuilder.and(b.greaterThan(2, 0.0d), b.isNaN(2)); + Observation actual = observe(fixture, filter); + assertAll( + () -> assertThat(actual.scalar).isEmpty(), + () -> assertThat(actual.exact).as("partial predicate must not be exact").isFalse(), + () -> assertThat(actual.vector).as("finite prices cannot satisfy isNaN").isEmpty()); + } + + @ParameterizedTest + @CsvSource({"full,true", "full,false", "detail,true", "detail,false"}) + public void fullySupportedEqualityControl(String mode, boolean refine) throws Exception { + Fixture fixture = prepare(mode, refine); + Predicate filter = + new PredicateBuilder(fixture.table.rowType()) + .equal(1, BinaryString.fromString("alpha")); + Observation actual = observe(fixture, filter); + assertAll( + () -> assertThat(actual.scalar).containsExactly(1L), + () -> assertThat(actual.exact).isTrue(), + () -> assertThat(actual.vector).containsExactly(1L)); + } + + @ParameterizedTest + @CsvSource({"full,true", "full,false", "detail,true", "detail,false"}) + public void differentFieldUnsupportedPredicateControl(String mode, boolean refine) + throws Exception { + Fixture fixture = prepare(mode, refine); + PredicateBuilder b = new PredicateBuilder(fixture.table.rowType()); + Predicate filter = + PredicateBuilder.and( + b.greaterThan(2, 0.0d), + b.startsWith(1, BinaryString.fromString("ab")).negate().get()); + Observation actual = observe(fixture, filter); + assertAll( + () -> assertThat(actual.scalar).containsExactly(1L, 2L), + () -> assertThat(actual.exact).isFalse(), + () -> + assertThat(actual.vector) + .isEqualTo( + refine + ? Collections.singletonList(1L) + : Collections.emptyList())); + } + + @ParameterizedTest + @CsvSource({"full,true", "full,false", "detail,true", "detail,false"}) + public void fullySupportedSameFieldConjunctionStaysExact(String mode, boolean refine) + throws Exception { + Fixture fixture = prepare(mode, refine); + PredicateBuilder b = new PredicateBuilder(fixture.table.rowType()); + Predicate filter = + PredicateBuilder.and( + b.startsWith(1, BinaryString.fromString("a")), + b.notEqual(1, BinaryString.fromString("abacus"))); + Observation actual = observe(fixture, filter); + assertAll( + () -> assertThat(actual.scalar).containsExactly(1L), + () -> assertThat(actual.exact).isTrue(), + () -> assertThat(actual.vector).containsExactly(1L)); + } + + private Fixture prepare(String mode, boolean refine) throws Exception { + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("name", DataTypes.STRING()) + .column("price", DataTypes.DOUBLE()) + .column("vec", new ArrayType(DataTypes.FLOAT())) + .option(CoreOptions.BUCKET.key(), "-1") + .option(CoreOptions.ROW_TRACKING_ENABLED.key(), "true") + .option(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true") + .option(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), mode) + .option( + CoreOptions.GLOBAL_INDEX_FILTER_REFINE_FROM_DATA.key(), + Boolean.toString(refine)) + .option("test.vector.dimension", "2") + .option("test.vector.metric", "l2") + .build(); + catalog.createTable(identifier(), schema, false); + FileStoreTable table = getTableDefault(); + InternalRow[] rows = new InternalRow[NAMES.length]; + for (int i = 0; i < rows.length; i++) { + rows[i] = + GenericRow.of( + i, + BinaryString.fromString(NAMES[i]), + PRICES[i], + new GenericArray(VECTORS[i])); + } + write(table, rows); + + List scalarFiles = new ArrayList<>(); + scalarFiles.addAll( + buildIndex( + table, + "btree", + "name", + Arrays.stream(NAMES).map(BinaryString::fromString).toArray())); + scalarFiles.addAll( + buildIndex( + table, "btree", "price", new Object[] {PRICES[0], PRICES[1], PRICES[2]})); + List allFiles = new ArrayList<>(scalarFiles); + allFiles.addAll( + buildIndex(table, TestVectorGlobalIndexerFactory.IDENTIFIER, "vec", VECTORS)); + CommitMessage message = + new CommitMessageImpl( + BinaryRow.EMPTY_ROW, + 0, + null, + DataIncrement.indexIncrement(allFiles), + CompactIncrement.emptyIncrement()); + try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { + commit.commit(Collections.singletonList(message)); + } + return new Fixture(table, scalarFiles); + } + + private static List buildIndex( + FileStoreTable table, String type, String column, Object[] sortedValues) + throws Exception { + Options options = table.coreOptions().toConfiguration(); + DataField field = table.rowType().getField(column); + GlobalIndexSingleColumnWriter writer = + (GlobalIndexSingleColumnWriter) + GlobalIndexBuilderUtils.createIndexWriter( + table, type, Collections.singletonList(field), options); + for (int i = 0; i < sortedValues.length; i++) { + writer.write(sortedValues[i], i); + } + return GlobalIndexBuilderUtils.toIndexFileMetas( + table.fileIO(), + table.store().pathFactory().globalIndexFileFactory(), + table.coreOptions(), + new Range(0, sortedValues.length - 1), + field.id(), + type, + writer.finish()); + } + + private static Observation observe(Fixture fixture, Predicate filter) throws Exception { + FileStoreTable table = fixture.table; + GlobalIndexEvaluator.Evaluation evaluation; + try (DataEvolutionGlobalIndexScanner scanner = + DataEvolutionGlobalIndexScanner.createForScalarFilters( + table, null, null, fixture.scalarFiles) + .get()) { + evaluation = scanner.scanWithCoverage(filter).get(); + } + boolean exact = FilteredRowIdReader.isExact(table.rowType(), filter, evaluation); + ReadBuilder readBuilder = table.newReadBuilder().withFilter(filter); + List scalar = new ArrayList<>(); + try (RecordReader reader = + readBuilder.newRead().executeFilter().createReader(readBuilder.newScan().plan())) { + reader.forEachRemaining(row -> scalar.add((long) row.getInt(0))); + } + List vector = new ArrayList<>(); + table.newVectorSearchBuilder() + .withVector(new float[] {1.0f, 0.0f}) + .withVectorColumn("vec") + .withLimit(1) + .withFilter(filter) + .executeLocal() + .results() + .forEach(vector::add); + return new Observation(exact, scalar, vector); + } + + private static final class Fixture { + final FileStoreTable table; + final List scalarFiles; + + Fixture(FileStoreTable table, List scalarFiles) { + this.table = table; + this.scalarFiles = scalarFiles; + } + } + + private static final class Observation { + final boolean exact; + final List scalar; + final List vector; + + Observation(boolean exact, List scalar, List vector) { + this.exact = exact; + this.scalar = scalar; + this.vector = vector; + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java index 7baeed2e102f..f9190580c0f7 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/FullTextSearchBuilderTest.java @@ -25,7 +25,9 @@ import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.serializer.InternalRowSerializer; +import org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner; import org.apache.paimon.globalindex.GlobalIndexBuilderUtils; +import org.apache.paimon.globalindex.GlobalIndexEvaluator; import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; import org.apache.paimon.globalindex.ResultEntry; @@ -62,6 +64,8 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.MockedStatic; import javax.annotation.Nullable; @@ -73,16 +77,52 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Optional; import static org.apache.paimon.table.source.DeletionVectorTestUtils.commitDeletionVectors; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.when; /** Tests for {@link FullTextSearchBuilder} using test-only brute-force full-text index. */ public class FullTextSearchBuilderTest extends TableTestBase { private static final String TEXT_FIELD_NAME = "content"; + @ParameterizedTest + @ValueSource(strings = {"fast", "full", "detail"}) + public void testDeclinedScalarShardUsesActualCoverage(String mode) throws Exception { + FileStoreTable table = createTable("declined_scalar_shard_" + mode, mode); + writeDocuments(table, RANKED_DOCUMENTS); + buildAndCommitIndex(table, RANKED_DOCUMENTS); + buildAndCommitIdBTreeIndex(table, RANKED_DOCUMENTS.length); + Predicate filter = new PredicateBuilder(table.rowType()).equal(0, 5); + GlobalIndexEvaluator.Evaluation evaluation = mock(GlobalIndexEvaluator.Evaluation.class); + when(evaluation.result()).thenReturn(GlobalIndexResult.createEmpty()); + when(evaluation.contributingFieldIds()).thenReturn(Collections.singleton(0)); + when(evaluation.coveredRanges()).thenReturn(Collections.singletonList(new Range(0, 4))); + DataEvolutionGlobalIndexScanner scanner = mock(DataEvolutionGlobalIndexScanner.class); + when(scanner.scanWithCoverage(filter)).thenReturn(Optional.of(evaluation)); + try (MockedStatic scanners = + mockStatic(DataEvolutionGlobalIndexScanner.class, CALLS_REAL_METHODS)) { + scanners.when( + () -> + DataEvolutionGlobalIndexScanner.createForScalarFilters( + any(), any(), any(), any())) + .thenReturn(Optional.of(scanner)); + RoaringNavigableMap64 rows = searchWithFilter(table, filter, 1).results(); + if (mode.equals("fast")) { + assertThat(rows).isEmpty(); + } else { + assertThat(rows).containsExactly(5L); + } + } + } + @Override protected Schema schemaDefault() { return Schema.newBuilder() diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchRowFilterExactnessTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchRowFilterExactnessTest.java index c0bd13300f8b..66e117f93ad6 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchRowFilterExactnessTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchRowFilterExactnessTest.java @@ -24,9 +24,14 @@ import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericRow; +import org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner; import org.apache.paimon.globalindex.GlobalIndexBuilderUtils; +import org.apache.paimon.globalindex.GlobalIndexEvaluator; +import org.apache.paimon.globalindex.GlobalIndexReader; import org.apache.paimon.globalindex.GlobalIndexResult; import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; +import org.apache.paimon.globalindex.GlobalIndexer; +import org.apache.paimon.globalindex.VectorGlobalIndexer; import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory; import org.apache.paimon.globalindex.testvector.TestVectorGlobalIndexerFactory; import org.apache.paimon.index.IndexFileMeta; @@ -52,12 +57,21 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.MockedStatic; import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicInteger; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.CALLS_REAL_METHODS; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.when; /** * Vector search with a row filter must rank an exact match set: candidate-only scalar index answers @@ -66,6 +80,40 @@ */ public class VectorSearchRowFilterExactnessTest extends TableTestBase { + @ParameterizedTest + @ValueSource(strings = {"full", "detail"}) + public void testDeclinedScalarShardFallsBackBeforeVectorRanking(String mode) throws Exception { + FileStoreTable table = + createTable( + "declined_scalar_" + mode, + schemaBuilder(true) + .option(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), mode)); + String[] names = {"alpha", "beta", "gamma"}; + float[][] vectors = {{1.0f, 0.0f}, {0.6f, 0.8f}, {0.0f, 1.0f}}; + write(table, names, vectors); + buildAndCommitVectorIndex(table, vectors); + buildAndCommitNameBTreeIndex(table, names); + Predicate filter = + new PredicateBuilder(table.rowType()).equal(1, BinaryString.fromString("beta")); + GlobalIndexEvaluator.Evaluation evaluation = mock(GlobalIndexEvaluator.Evaluation.class); + when(evaluation.result()).thenReturn(GlobalIndexResult.createEmpty()); + when(evaluation.contributingFieldIds()).thenReturn(Collections.singleton(1)); + when(evaluation.coveredRanges()).thenReturn(Collections.singletonList(new Range(0, 0))); + DataEvolutionGlobalIndexScanner scanner = mock(DataEvolutionGlobalIndexScanner.class); + when(scanner.scanWithCoverage(filter)).thenReturn(Optional.of(evaluation)); + when(scanner.unindexedRowsForEvaluation(evaluation)) + .thenReturn(GlobalIndexResult.fromRange(new Range(1, 2))); + try (MockedStatic scanners = + mockStatic(DataEvolutionGlobalIndexScanner.class, CALLS_REAL_METHODS)) { + scanners.when( + () -> + DataEvolutionGlobalIndexScanner.createForScalarFilters( + any(), any(), any(), any())) + .thenReturn(Optional.of(scanner)); + assertThat(search(table, filter, 1)).containsExactly(1L); + } + } + @Override protected Schema schemaDefault() { return schemaBuilder(true).build(); @@ -481,6 +529,71 @@ public void testFilterOnUnindexedColumn(String scalarMode) throws Exception { } } + @Test + public void testRawFallbackPreservesBackendMetricConflictPolicy() throws Exception { + FileStoreTable table = + createTable( + "vector_fallback_metric_policy", + schemaBuilder(false) + .option(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "full")); + float[][] vectors = {{100.0f, 0.0f}, {0.9f, 0.1f}}; + write(table, new String[] {"keep", "keep"}, vectors); + buildAndCommitVectorIndex(table, vectors, new Range(0, 0)); + buildAndCommitVectorIndex(table, vectors, new Range(1, 1)); + Predicate filter = + new PredicateBuilder(table.rowType()).equal(1, BinaryString.fromString("keep")); + AtomicInteger indexerCount = new AtomicInteger(); + AtomicInteger readerCount = new AtomicInteger(); + DataEvolutionVectorRead read = + new DataEvolutionVectorRead( + table, + null, + filter, + 1, + table.rowType().getField("vec"), + new float[] {1.0f, 0.0f}, + Collections.emptyMap()) { + @Override + protected GlobalIndexer createGlobalIndexer( + List splits) { + indexerCount.incrementAndGet(); + AtomicInteger registeredShards = new AtomicInteger(); + VectorGlobalIndexer indexer = mock(VectorGlobalIndexer.class); + when(indexer.createReader(any(), any(), anyLong(), any(), any())) + .thenAnswer( + invocation -> { + readerCount.incrementAndGet(); + registeredShards.incrementAndGet(); + return mock(GlobalIndexReader.class); + }); + // Model a backend such as Lumina: after conflicting metadata is + // registered, it chooses the configured metric instead of throwing. + when(indexer.metric()) + .thenAnswer( + invocation -> + registeredShards.get() == 1 + ? "l2" + : "inner_product"); + return indexer; + } + }; + VectorScan.Plan plan = + table.newVectorSearchBuilder() + .withVectorColumn("vec") + .withVector(new float[] {1.0f, 0.0f}) + .withLimit(1) + .withFilter(filter) + .newVectorScan() + .scan(); + assertThat(read.read(plan).results()).containsExactly(0L); + assertThat(indexerCount.get()).isEqualTo(1); + assertThat(readerCount.get()).isEqualTo(2); + // A read object may be reused: no previous read's metric or indexer should leak into it. + assertThat(read.read(plan).results()).containsExactly(0L); + assertThat(indexerCount.get()).isEqualTo(2); + assertThat(readerCount.get()).isEqualTo(4); + } + private void write(FileStoreTable table, String[] names, float[][] vectors) throws Exception { BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder(); try (BatchTableWrite write = writeBuilder.newWrite(); diff --git a/paimon-eslib/pom.xml b/paimon-eslib/pom.xml index ec07e73e1723..9740da38adfb 100644 --- a/paimon-eslib/pom.xml +++ b/paimon-eslib/pom.xml @@ -90,6 +90,54 @@ under the License. + + org.apache.paimon + paimon-core + ${project.version} + test + + + org.apache.paimon + paimon-format + ${project.version} + test + + + org.apache.hadoop + hadoop-mapreduce-client-core + ${hadoop.version} + test + + + org.apache.avro + avro + + + com.google.protobuf + protobuf-java + + + ch.qos.reload4j + reload4j + + + org.slf4j + slf4j-reload4j + + + log4j + log4j + + + org.slf4j + slf4j-log4j12 + + + jdk.tools + jdk.tools + + + org.apache.paimon paimon-common diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java index 4d0093abb1fe..6309d9ac39fe 100644 --- a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java +++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexReader.java @@ -1436,23 +1436,6 @@ private Optional executeFilter(String fieldName, IndexFilter } } - private Optional conservativeAllRows(String logicalField) { - if (physicalField(logicalField) == null) { - return Optional.empty(); - } - try { - ensureLoaded(); - RoaringNavigableMap64 bitmap = new RoaringNavigableMap64(); - for (long id : searcher.allRowIds()) { - bitmap.add(id); - } - return Optional.of(GlobalIndexResult.create(bitmap)); - } catch (IOException e) { - throw new RuntimeException( - "Failed to read conservative ES index rows for field: " + logicalField, e); - } - } - private boolean canEvaluateFilter(FieldRef fieldRef, IndexFilter filter) { String physicalField = physicalField(fieldRef.name()); FieldIndexConfig config = logicalFieldConfig(fieldRef.name()); @@ -1489,7 +1472,9 @@ private Optional dispatchFilter(FieldRef fieldRef, IndexFilte String physicalField = physicalField(fieldRef.name()); FieldIndexConfig config = logicalFieldConfig(fieldRef.name()); if (!canEvaluateFilter(fieldRef, filter)) { - return conservativeAllRows(fieldRef.name()); + // Unsupported is not an exact all-rows result. In particular, search pre-filters + // must fall back or refine instead of admitting unfiltered rows into top-k. + return Optional.empty(); } if (filter.filterType() == IndexFilter.FilterType.EXISTS && fieldRef.type() instanceof ArrayType) { @@ -1497,7 +1482,7 @@ private Optional dispatchFilter(FieldRef fieldRef, IndexFilte try { return executeFilter(presenceField, filter); } catch (UnsupportedOperationException | IllegalArgumentException e) { - return conservativeAllRows(fieldRef.name()); + return Optional.empty(); } } if (config.indexType() == FieldIndexConfig.IndexType.FULLTEXT) { @@ -1505,20 +1490,19 @@ private Optional dispatchFilter(FieldRef fieldRef, IndexFilte // the raw column value, so ordinary predicates (=, <>, <, >, IN, LIKE, IS NULL, ...) // cannot run on it directly. Route them to the keyword multi-field sub-field // (content.keyword) when present — it holds the exact value and serves these correctly. - // Without the sub-field, conservatively retain every row so the caller can evaluate - // the predicate. Full-text search (visitFullTextSearch) still targets the analyzed - // primary field. + // Without the sub-field, canEvaluateFilter returns false so the caller can fall back. + // Full-text search (visitFullTextSearch) still targets the analyzed primary field. String subField = indexOptions.keywordSubField(physicalField); try { return executeFilter(subField, filter); } catch (UnsupportedOperationException | IllegalArgumentException e) { - return conservativeAllRows(fieldRef.name()); + return Optional.empty(); } } try { return executeFilter(physicalField, filter); } catch (UnsupportedOperationException | IllegalArgumentException e) { - return conservativeAllRows(fieldRef.name()); + return Optional.empty(); } } @@ -1550,7 +1534,7 @@ private Optional dispatchExistingRowsNotMatching( FieldRef fieldRef, IndexFilter matchingFilter) { if (!canEvaluateFilter(fieldRef, IndexFilter.exists()) || !canEvaluateFilter(fieldRef, matchingFilter)) { - return conservativeAllRows(fieldRef.name()); + return Optional.empty(); } Optional existing = dispatchFilter(fieldRef, IndexFilter.exists()); if (existing.isEmpty()) { @@ -1641,7 +1625,7 @@ public CompletableFuture> visitEqual( () -> { FieldIndexConfig config = logicalFieldConfig(fieldRef.name()); if (config == null) { - return conservativeAllRows(fieldRef.name()); + return Optional.empty(); } FieldIndexConfig.IndexType type = config.indexType(); if (type == FieldIndexConfig.IndexType.KEYWORD @@ -1785,7 +1769,7 @@ public CompletableFuture> visitIn( @Override public CompletableFuture> visitNotIn( FieldRef fieldRef, List literals) { - if (literals == null || literals.contains(null)) { + if (literals == null || literals.stream().anyMatch(Objects::isNull)) { return emptyFilterFuture(); } return async( diff --git a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerFactory.java b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerFactory.java index 98cc1d8deeb5..b0e87d6611f5 100644 --- a/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerFactory.java +++ b/paimon-eslib/src/main/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexerFactory.java @@ -43,6 +43,11 @@ public boolean supportsFullTextSearch() { return true; } + @Override + public boolean supportsGroupedIndexScalarQueries() { + return true; + } + @Override public GlobalIndexer create(List indexFields, Options options) { return new ESIndexGlobalIndexer(indexFields, options); diff --git a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexE2ETest.java b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexE2ETest.java index 464d1ac8a47e..446ae3739c18 100644 --- a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexE2ETest.java +++ b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexGlobalIndexE2ETest.java @@ -44,10 +44,15 @@ import org.apache.paimon.types.DataTypes; import org.apache.paimon.utils.RoaringNavigableMap64; +import org.elasticsearch.eslib.api.ESIndexSearcher; +import org.elasticsearch.eslib.api.model.IndexFilter; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import java.io.IOException; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Proxy; import java.nio.file.Files; import java.util.Arrays; import java.util.HashMap; @@ -67,6 +72,7 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -444,13 +450,11 @@ void writeArchiveThenReadAndSearch(@TempDir java.nio.file.Path tmp) throws IOExc 20, reader.visitLike(titleRef, "doc%").join().get().results().getIntCardinality(), "LIKE 'doc%' on title.keyword matches all 20 'document ...' titles"); - // A range predicate on a string keyword sub-field is unsupported. Return the complete - // shard as a conservative result so a union with differently configured shards cannot - // drop rows; Paimon's residual predicate performs the actual comparison. + // A range predicate on a string keyword sub-field is unsupported. Do not return all rows + // as though they exactly matched: search callers must fall back before selecting top-k. Optional unsupportedTitleRange = reader.visitGreaterThan(titleRef, "abc").join(); - assertTrue(unsupportedTitleRange.isPresent()); - assertEquals(20, unsupportedTitleRange.get().results().getIntCardinality()); + assertTrue(unsupportedTitleRange.isEmpty()); // --- scalar filter: price >= 100 means rows 10..19 (price = i*10) --- FieldRef priceRef = new FieldRef(3, "price", DataTypes.INT()); @@ -594,11 +598,9 @@ void readsIndexAfterIndexedColumnsAreRenamed(@TempDir java.nio.file.Path tmp) .visitLessThan( new FieldRef(3, "cost", DataTypes.BIGINT()), 3_000_000_000L) .join(); - assertTrue(widenedFallback.isPresent()); - assertEquals( - 2, - widenedFallback.get().results().getIntCardinality(), - "a widened field must keep every old INT-indexed row"); + assertTrue( + widenedFallback.isEmpty(), + "an incompatible old INT index must request fallback for the widened field"); } finally { widenedReader.close(); } @@ -678,11 +680,17 @@ void mixedSchemaShardsDoNotDropRowsFromAnOlderScalarType(@TempDir java.nio.file. union.visitLessThan( new FieldRef(0, "value", DataTypes.BIGINT()), 3_000_000_000L) .join(); - assertTrue(candidates.isPresent()); - assertEquals(3, candidates.get().results().getIntCardinality()); - assertTrue(contains(candidates.get().results(), 0L)); - assertTrue(contains(candidates.get().results(), 1L)); - assertTrue(contains(candidates.get().results(), 3L)); + assertTrue( + candidates.isEmpty(), + "one unsupported shard makes the union unsupported, not a partial answer"); + Optional supported = + currentReader + .visitLessThan( + new FieldRef(0, "value", DataTypes.BIGINT()), 3_000_000_000L) + .join(); + assertTrue(supported.isPresent()); + assertEquals(1, supported.get().results().getIntCardinality()); + assertTrue(contains(supported.get().results(), 1L)); } finally { union.close(); } @@ -838,16 +846,33 @@ void nullLiteralsFollowSqlThreeValuedLogicAndKeywordRangeFallsBack( notInWithNull.get().results().isEmpty(), "NOT IN containing NULL is UNKNOWN for every row"); + Optional immutableNotIn = + reader.visitNotIn(kRef, List.of("a")).join(); + assertTrue(immutableNotIn.isPresent()); + assertEquals(2, immutableNotIn.get().results().getIntCardinality()); + assertTrue(contains(immutableNotIn.get().results(), 0L)); + assertTrue(contains(immutableNotIn.get().results(), 3L)); + Optional nullRange = reader.visitGreaterThan(kRef, null).join(); assertTrue(nullRange.isPresent()); assertTrue(nullRange.get().results().isEmpty()); - Optional keywordRange = reader.visitGreaterThan(kRef, "a").join(); - assertTrue(keywordRange.isPresent()); - assertEquals( - 4, - keywordRange.get().results().getIntCardinality(), - "unsupported KEYWORD ordering must conservatively retain the whole shard"); + for (CompletableFuture> keywordRange : + Arrays.asList( + reader.visitGreaterThan(kRef, "a"), + reader.visitGreaterOrEqual(kRef, "a"), + reader.visitLessThan(kRef, "b"), + reader.visitLessOrEqual(kRef, "b"), + reader.visitBetween(kRef, "a", "b"), + reader.visitNotBetween(kRef, "a", "b"))) { + assertTrue( + keywordRange.join().isEmpty(), + "unsupported KEYWORD ordering must not claim an exact all-rows result"); + } + + Optional zeroHits = reader.visitEqual(kRef, "absent").join(); + assertTrue(zeroHits.isPresent(), "supported zero-hit predicates remain supported"); + assertTrue(zeroHits.get().results().isEmpty()); } finally { reader.close(); } @@ -1569,17 +1594,100 @@ void highPrecisionTimestampPredicatesFallBackToRawScan(@TempDir java.nio.file.Pa FieldRef ts = new FieldRef(0, "ts", DataTypes.TIMESTAMP(6)); Optional fallback = reader.visitEqual(ts, Timestamp.fromEpochMillis(1_000L, 2_000)).join(); - assertTrue(fallback.isPresent()); - assertEquals( - 1, - fallback.get().results().getIntCardinality(), - "a lossy millisecond index must conservatively retain every row"); + assertTrue( + fallback.isEmpty(), + "a lossy millisecond index cannot evaluate microsecond comparisons"); + assertTrue( + reader.visitNotEqual(ts, Timestamp.fromEpochMillis(1_000L, 2_000)) + .join() + .isEmpty()); + assertTrue( + reader.visitBetween( + ts, + Timestamp.fromEpochMillis(1_000L, 1_000), + Timestamp.fromEpochMillis(1_000L, 2_000)) + .join() + .isEmpty()); + assertTrue( + reader.visitNotBetween( + ts, + Timestamp.fromEpochMillis(1_000L, 1_000), + Timestamp.fromEpochMillis(1_000L, 2_000)) + .join() + .isEmpty()); assertTrue(reader.visitIsNotNull(ts).join().isPresent()); } finally { reader.close(); } } + @Test + void unsupportedSdkOperationsMustNotBecomeExactMatches(@TempDir java.nio.file.Path tmp) + throws Exception { + List fields = List.of(new DataField(0, "n", DataTypes.INT())); + ESIndexOptions options = new ESIndexOptions(fields, new Options()); + java.nio.file.Path dir = tmp.resolve("sdk-unsupported"); + Files.createDirectories(dir); + ESIndexGlobalIndexWriter writer = + new ESIndexGlobalIndexWriter(new LocalDirWriter(dir), fields, options); + writer.write(1, 0); + writer.write(2, 1); + ResultEntry entry = writer.finish().get(0); + try (ESIndexGlobalIndexReader reader = + new ESIndexGlobalIndexReader( + new LocalFileReader(), List.of(ioMeta(dir, entry)), fields, options)) { + FieldRef ref = new FieldRef(0, "n", DataTypes.INT()); + assertTrue(reader.visitEqual(ref, 1).join().isPresent()); + Field searcherField = ESIndexGlobalIndexReader.class.getDeclaredField("searcher"); + searcherField.setAccessible(true); + ESIndexSearcher actual = (ESIndexSearcher) searcherField.get(reader); + AtomicBoolean failExists = new AtomicBoolean(); + AtomicReference failure = new AtomicReference<>(); + // Keep the real archive loaded; inject only an SDK capability or I/O failure. + searcherField.set( + reader, + Proxy.newProxyInstance( + ESIndexSearcher.class.getClassLoader(), + new Class[] {ESIndexSearcher.class}, + (proxy, method, args) -> { + if (method.getName().equals("filter") + && (((IndexFilter) args[1]).filterType() + == IndexFilter.FilterType.EXISTS) + == failExists.get()) { + throw failure.get(); + } + try { + return method.invoke(actual, args); + } catch (InvocationTargetException e) { + throw e.getCause(); + } + })); + for (Exception unsupported : + Arrays.asList( + new UnsupportedOperationException("unsupported SDK operation"), + new IllegalArgumentException("unsupported SDK type"))) { + failure.set(unsupported); + for (boolean exists : new boolean[] {false, true}) { + failExists.set(exists); + assertTrue(reader.visitNotEqual(ref, 1).join().isEmpty()); + assertTrue(reader.visitNotIn(ref, List.of(1)).join().isEmpty()); + assertTrue(reader.visitNotBetween(ref, 1, 2).join().isEmpty()); + if (exists) { + assertTrue(reader.visitIsNotNull(ref).join().isEmpty()); + } else { + assertTrue(reader.visitEqual(ref, 1).join().isEmpty()); + assertTrue(reader.visitLessThan(ref, 2).join().isEmpty()); + } + } + } + failExists.set(false); + failure.set(new IOException("read failure is not an unsupported operation")); + CompletionException ioFailure = + assertThrows(CompletionException.class, () -> reader.visitEqual(ref, 1).join()); + assertTrue(ioFailure.getCause().getCause() instanceof IOException); + } + } + @Test void standardBooleanDslAndCamelCaseFuzzyParamsAreApplied(@TempDir java.nio.file.Path tmp) throws IOException { diff --git a/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexScalarTableTest.java b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexScalarTableTest.java new file mode 100644 index 000000000000..76fd21e30cab --- /dev/null +++ b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexScalarTableTest.java @@ -0,0 +1,1038 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.eslib.index; + +import org.apache.paimon.CoreOptions.GlobalIndexSearchMode; +import org.apache.paimon.Snapshot; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.data.GenericArray; +import org.apache.paimon.data.GenericRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.SeekableInputStream; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.globalindex.DataEvolutionGlobalIndexCoverage; +import org.apache.paimon.globalindex.DataEvolutionGlobalIndexScanner; +import org.apache.paimon.globalindex.GlobalIndexEvaluator; +import org.apache.paimon.globalindex.GlobalIndexFileReadWrite; +import org.apache.paimon.globalindex.GlobalIndexMultiColumnWriter; +import org.apache.paimon.globalindex.GlobalIndexer; +import org.apache.paimon.globalindex.GlobalIndexerFactoryUtils; +import org.apache.paimon.globalindex.ResultEntry; +import org.apache.paimon.globalindex.ScoredGlobalIndexResult; +import org.apache.paimon.globalindex.VectorGlobalIndexer; +import org.apache.paimon.index.DataEvolutionIndexSourceMeta; +import org.apache.paimon.index.GlobalIndexMeta; +import org.apache.paimon.index.IndexFileMeta; +import org.apache.paimon.io.CompactIncrement; +import org.apache.paimon.io.DataIncrement; +import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.table.sink.BatchTableCommit; +import org.apache.paimon.table.sink.BatchTableWrite; +import org.apache.paimon.table.sink.BatchWriteBuilder; +import org.apache.paimon.table.sink.CommitMessageImpl; +import org.apache.paimon.table.sink.TableCommitImpl; +import org.apache.paimon.table.source.BatchVectorSearchBuilder; +import org.apache.paimon.table.source.DataEvolutionVectorRead; +import org.apache.paimon.table.source.IndexVectorSearchSplit; +import org.apache.paimon.table.source.ReadBuilder; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.table.source.VectorSearchBuilder; +import org.apache.paimon.types.DataField; +import org.apache.paimon.types.DataTypeRoot; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.utils.InstantiationUtil; +import org.apache.paimon.utils.Range; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.jupiter.api.Assertions.assertAll; + +/** Real ES archives must accelerate scalar predicates without dropping uncovered table rows. */ +class ESIndexScalarTableTest { + + @TempDir java.nio.file.Path temporary; + + private final AuditedFileIO fileIO = new AuditedFileIO(); + private final List indexFiles = new ArrayList<>(); + + @ParameterizedTest + @CsvSource({ + "fast,false", + "fast,true", + "full,false", + "full,true", + "detail,false", + "detail,true" + }) + void testScalarPredicatesUseGroupedIndex(String mode, boolean inReader) throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 12, "embedding", "category", "price"); + FileStoreTable indexed = configured(table, mode, inReader); + PredicateBuilder b = new PredicateBuilder(table.rowType()); + List predicates = + Arrays.asList( + b.equal(2, BinaryString.fromString("c1")), + b.greaterOrEqual(3, 3.0), + PredicateBuilder.and(b.greaterOrEqual(3, 3.0), b.lessThan(3, 9.0)), + b.in( + 2, + Arrays.asList( + BinaryString.fromString("c0"), + BinaryString.fromString("c2"))), + b.isNull(2), + b.isNotNull(3), + b.notEqual(2, BinaryString.fromString("c1")), + b.notIn( + 2, + Arrays.asList( + BinaryString.fromString("c0"), + BinaryString.fromString("c2"))), + b.startsWith(2, BinaryString.fromString("c")), + PredicateBuilder.and( + b.equal(2, BinaryString.fromString("c1")), b.greaterThan(3, 2.0)), + PredicateBuilder.or( + b.equal(2, BinaryString.fromString("c1")), b.lessThan(3, 3.0))); + for (Predicate predicate : predicates) { + List expected = read(withoutIndex(table), predicate); + fileIO.indexOpens.set(0); + assertThat(read(indexed, predicate)) + .as("%s", predicate) + .containsExactlyElementsOf(expected); + assertThat(fileIO.indexOpens.get()) + .as("ES index must serve %s", predicate) + .isPositive(); + } + } + + @ParameterizedTest + @CsvSource({"full,false", "full,true", "detail,false", "detail,true"}) + void testPartialAndDifferentDefinitionsRetainUncoveredRows(String mode, boolean inReader) + throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 4, "embedding", "category", "price"); + build(table, 6, 8, "embedding", "price"); + PredicateBuilder b = new PredicateBuilder(table.rowType()); + Predicate category = b.equal(2, BinaryString.fromString("c1")); + Predicate price = b.greaterOrEqual(3, 2.0); + Predicate residual = b.equal(4, BinaryString.fromString("keep")); + for (Predicate predicate : + Arrays.asList( + category, + price, + PredicateBuilder.and(category, price), + PredicateBuilder.or(category, price), + PredicateBuilder.and(price, residual), + PredicateBuilder.or(category, residual), + PredicateBuilder.or(category, b.isNaN(3)))) { + List expected = read(withoutIndex(table), predicate); + assertThat(read(configured(table, mode, inReader), predicate)) + .as("%s", predicate) + .containsExactlyElementsOf(expected); + } + } + + @Test + void testCoverageIncludesOnlyFieldsOfEachEligibleDefinition() throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 4, "embedding", "category", "price"); + build(table, 6, 8, "embedding", "price"); + DataEvolutionGlobalIndexCoverage coverage = + new DataEvolutionGlobalIndexCoverage( + table, + table.snapshotManager().latestSnapshot(), + null, + indexFiles, + GlobalIndexSearchMode.FULL); + assertThat(coverage.unindexedRanges(2)).containsExactly(new Range(4, 11)); + assertThat(coverage.unindexedRanges(3)).containsExactly(new Range(4, 5), new Range(8, 11)); + assertThat(coverage.unindexedRanges(4)).containsExactly(new Range(0, 11)); + assertThat(coverage.unindexedRanges(Arrays.asList(2, 3), null)) + .containsExactly(new Range(4, 11)); + try (DataEvolutionGlobalIndexScanner scanner = + DataEvolutionGlobalIndexScanner.createForScalarFilters( + table, table.snapshotManager().latestSnapshot(), null, indexFiles) + .orElseThrow( + () -> + new AssertionError( + "Grouped ES must be eligible for scalar filters"))) { + assertThat(scanner.scan(new PredicateBuilder(table.rowType()).greaterThan(3, 1.0))) + .isPresent(); + } + } + + @Test + void testFastModeDoesNotIncludeUnindexedTail() throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 8, "embedding", "category", "price"); + Predicate predicate = new PredicateBuilder(table.rowType()).greaterOrEqual(3, 2.0); + List expected = + read(withoutIndex(table), predicate).stream() + .filter(id -> id < 8) + .collect(Collectors.toList()); + for (boolean inReader : Arrays.asList(false, true)) { + assertThat(read(configured(table, "fast", inReader), predicate)) + .containsExactlyElementsOf(expected); + } + } + + @Test + void testVectorPrimaryDesignIsUnchanged() throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 12, "embedding", "category", "price"); + assertThat( + table.newVectorSearchBuilder() + .withVectorColumn("embedding") + .withVector(vector(5)) + .withLimit(1) + .executeLocal() + .results() + .toRangeList()) + .containsExactly(new Range(5, 5)); + assertThat( + table.newVectorSearchBuilder() + .withVectorColumn("embedding") + .withVector(vector(5)) + .withLimit(1) + .withFilter( + new PredicateBuilder(table.rowType()) + .equal(2, BinaryString.fromString("c1"))) + .executeLocal() + .results() + .toRangeList()) + .containsExactly(new Range(4, 4)); + } + + @ParameterizedTest + @CsvSource({ + "disjoint,false", + "disjoint,true", + "same-range,false", + "same-range,true", + "reordered,false", + "reordered,true", + "same-definition,false", + "same-definition,true", + "overlap,false", + "overlap,true", + "raw-tail,false", + "raw-tail,true" + }) + void testVectorQueriesUseMatchingDefinitions(String layout, boolean batch) throws Exception { + FileStoreTable table = createTable(); + if ("same-range".equals(layout)) { + build(table, 0, 12, "embedding", "category", "price"); + build(table, 0, 12, "embedding", "category"); + } else { + int firstEnd = "overlap".equals(layout) ? 8 : 6; + int secondStart = "overlap".equals(layout) ? 4 : 6; + int secondEnd = "raw-tail".equals(layout) ? 8 : 12; + build(table, 0, firstEnd, "embedding", "category", "price"); + if ("same-definition".equals(layout)) { + build(table, secondStart, secondEnd, "embedding", "category", "price"); + } else if ("reordered".equals(layout)) { + build(table, secondStart, secondEnd, "embedding", "price", "category"); + } else { + build(table, secondStart, secondEnd, "embedding", "category"); + } + } + table = configured(table, "full", false); + Predicate filter = + new PredicateBuilder(table.rowType()).equal(2, BinaryString.fromString("c1")); + assertThat(read(withoutIndex(table), filter)).containsExactly(1, 4, 7); + fileIO.indexOpens.set(0); + assertDefinitionVectorQuery(table, 5, null, batch, false, 5); + assertThat(fileIO.indexOpens.get()).isPositive(); + assertDefinitionVectorQuery(table, 5, filter, batch, false, 4); + Predicate unsupported = + new PredicateBuilder(table.rowType()).greaterThan(2, BinaryString.fromString("c1")); + assertDefinitionVectorQuery(table, 5, unsupported, batch, false, 2); + } + + @ParameterizedTest + @CsvSource({"false,false", "false,true", "true,false", "true,true"}) + void testVectorDefinitionsPreserveMetricForRawAndRefine(boolean rawTail, boolean batch) + throws Exception { + FileStoreTable table = createTable(); + FileStoreTable cosine = + table.copy( + Collections.singletonMap( + "global-index.es-index.fields.embedding.metric", "cosine")); + build(cosine, 0, 6, "embedding", "category", "price"); + build(cosine, 6, rawTail ? 8 : 12, "embedding", "category"); + Predicate filter = + new PredicateBuilder(table.rowType()).equal(2, BinaryString.fromString("c2")); + // The table still says L2, but all index archives persist cosine. Row 8 wins in cosine, + // while row 2 wins in L2; row 8 is outside the index ranges when rawTail is true. + assertDefinitionVectorQuery(table, 5, filter, batch, false, 8); + assertDefinitionVectorQuery(table, 5, filter, batch, true, 8); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testSubclassIndexedReadPreservesDefinitionsAndMetric(boolean sameDefinition) + throws Exception { + FileStoreTable table = createTable(); + FileStoreTable cosine = + table.copy( + Collections.singletonMap( + "global-index.es-index.fields.embedding.metric", "cosine")); + build(cosine, 0, 6, "embedding", "category", "price"); + if (sameDefinition) { + build(cosine, 6, 8, "embedding", "category", "price"); + } else { + build(cosine, 6, 8, "embedding", "category"); + } + table = + configured(table, "full", false) + .copy(Collections.singletonMap("vector-index.search-mode", "full")); + Predicate filter = + new PredicateBuilder(table.rowType()).equal(2, BinaryString.fromString("c2")); + SubclassVectorRead read = new SubclassVectorRead(table, filter); + fileIO.indexOpens.set(0); + assertThat( + read.read( + table.newVectorSearchBuilder() + .withVectorColumn("embedding") + .withVector(vector(5)) + .withFilter(filter) + .withLimit(1) + .newVectorScan() + .scan()) + .results() + .toRangeList()) + .containsExactly(new Range(8, 8)); + assertThat(fileIO.indexOpens.get()).isPositive(); + assertThat(((VectorGlobalIndexer) read.suppliedIndexer).metric()).isEqualTo("cosine"); + } + + @ParameterizedTest + @CsvSource({"false,false", "false,true", "true,false", "true,true"}) + void testVectorDefinitionsRejectIncompatibleMetrics(boolean sameDefinition, boolean batch) + throws Exception { + FileStoreTable table = createTable(); + build( + table.copy( + Collections.singletonMap( + "global-index.es-index.fields.embedding.metric", "cosine")), + 0, + 6, + "embedding", + "category", + "price"); + if (sameDefinition) { + build(table, 6, 12, "embedding", "category", "price"); + } else { + build(table, 6, 12, "embedding", "category"); + } + assertThatThrownBy(() -> assertDefinitionVectorQuery(table, 5, null, batch, false, 5)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("different"); + } + + private void assertDefinitionVectorQuery( + FileStoreTable table, + int query, + Predicate filter, + boolean batch, + boolean refine, + int expected) { + table = + configured(table, "full", false) + .copy(Collections.singletonMap("vector-index.search-mode", "full")); + Map queryOptions = + refine + ? Collections.singletonMap("es-index.refine-factor", "2") + : Collections.emptyMap(); + if (batch) { + BatchVectorSearchBuilder builder = + table.newBatchVectorSearchBuilder() + .withVectorColumn("embedding") + .withVectors(new float[][] {vector(query), vector(query)}) + .withLimit(1) + .withOptions(queryOptions); + if (filter != null) { + builder.withFilter(filter); + } + assertThat(builder.executeBatchLocal()) + .hasSize(2) + .allSatisfy( + result -> + assertThat(result.results().toRangeList()) + .containsExactly(new Range(expected, expected))); + } else { + VectorSearchBuilder builder = + table.newVectorSearchBuilder() + .withVectorColumn("embedding") + .withVector(vector(query)) + .withLimit(1) + .withOptions(queryOptions); + if (filter != null) { + builder.withFilter(filter); + } + assertThat(builder.executeLocal().results().toRangeList()) + .containsExactly(new Range(expected, expected)); + } + } + + @Test + void testCompanionVectorIsNotSelectedForAnn() throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 12, "category", "embedding", "price"); + Predicate predicate = + new PredicateBuilder(table.rowType()).equal(2, BinaryString.fromString("c1")); + for (boolean inReader : Arrays.asList(false, true)) { + fileIO.indexOpens.set(0); + assertThat(read(configured(table, "full", inReader), predicate)) + .containsExactly(1, 4, 7); + assertThat(fileIO.indexOpens.get()).isPositive(); + } + FileStoreTable fast = + table.copy(Collections.singletonMap("vector-index.search-mode", "fast")); + fileIO.indexOpens.set(0); + assertThat( + fast.newVectorSearchBuilder() + .withVectorColumn("embedding") + .withVector(vector(5)) + .withLimit(1) + .executeLocal() + .results() + .isEmpty()) + .isTrue(); + assertThat(fileIO.indexOpens.get()).isZero(); + } + + @Test + void testOnlyEsOptsIntoGroupedIndexScalarLookups() { + assertThat(GlobalIndexerFactoryUtils.supportsGroupedIndexScalarQueries("es-index")) + .isTrue(); + assertThat(GlobalIndexerFactoryUtils.supportsGroupedIndexScalarQueries("btree")).isFalse(); + assertThat(GlobalIndexerFactoryUtils.supportsGroupedIndexScalarQueries("bitmap")).isFalse(); + assertThat(GlobalIndexerFactoryUtils.supportsGroupedIndexScalarQueries("unknown")) + .isFalse(); + } + + @Test + void testScalarPrimaryAndFullTextPreFilter() throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 12, "note", "category", "price"); + Predicate predicate = + new PredicateBuilder(table.rowType()).equal(2, BinaryString.fromString("c1")); + for (boolean inReader : Arrays.asList(false, true)) { + assertThat(read(configured(table, "full", inReader), predicate)) + .containsExactlyElementsOf(read(withoutIndex(table), predicate)); + } + assertThat( + table.newFullTextSearchBuilder() + .withQuery("note", "{\"match\":{\"query\":\"keep\"}}") + .withFilter(predicate) + .withLimit(10) + .executeLocal() + .results() + .toRangeList()) + .containsExactly(new Range(4, 4)); + } + + @ParameterizedTest + @CsvSource({"full,false", "full,true", "detail,false", "detail,true"}) + void testUnsupportedScalarPredicatesFallBack(String mode, boolean inReader) throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 4, "embedding", "category", "price"); + build(table, 6, 10, "embedding", "category", "price"); + PredicateBuilder b = new PredicateBuilder(table.rowType()); + Predicate range = b.greaterThan(2, BinaryString.fromString("c1")); + Predicate price = b.greaterThan(3, 0.0); + for (Predicate filter : + Arrays.asList( + range, + b.greaterOrEqual(2, BinaryString.fromString("c1")), + b.lessThan(2, BinaryString.fromString("c1")), + b.lessOrEqual(2, BinaryString.fromString("c1")), + b.between(2, BinaryString.fromString("c0"), BinaryString.fromString("c1")), + b.between(2, BinaryString.fromString("c0"), BinaryString.fromString("c1")) + .negate() + .get(), + PredicateBuilder.and(range, price), + PredicateBuilder.or(range, price))) { + assertThat(read(configured(table, mode, inReader), filter)) + .as("%s", filter) + .containsExactlyElementsOf(read(withoutIndex(table), filter)); + } + } + + @Test + void testUnsupportedFilterDoesNotClaimCoverageButZeroHitsDo() throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 12, "embedding", "category", "price"); + PredicateBuilder b = new PredicateBuilder(table.rowType()); + Predicate range = b.greaterThan(2, BinaryString.fromString("c1")); + try (DataEvolutionGlobalIndexScanner scanner = + DataEvolutionGlobalIndexScanner.createForScalarFilters( + table, table.snapshotManager().latestSnapshot(), null, indexFiles) + .get()) { + assertThat(scanner.scanWithCoverage(range)).isEmpty(); + assertThat(scanner.scanWithCoverage(PredicateBuilder.or(range, b.equal(3, 6.0)))) + .isEmpty(); + + GlobalIndexEvaluator.Evaluation partial = + scanner.scanWithCoverage(PredicateBuilder.and(range, b.greaterThan(3, 0.0))) + .get(); + assertThat(partial.fullyEvaluated()).isFalse(); + assertThat(partial.coveredRanges()).containsExactly(new Range(0, 11)); + assertThat(partial.contributingFieldIds()).containsExactly(3); + + // Supported but unmatched is an exact empty bitmap, not an unsupported operation. + GlobalIndexEvaluator.Evaluation zero = + scanner.scanWithCoverage(b.equal(2, BinaryString.fromString("absent"))).get(); + assertThat(zero.fullyEvaluated()).isTrue(); + assertThat(zero.coveredRanges()).containsExactly(new Range(0, 11)); + assertThat(zero.result().results().isEmpty()).isTrue(); + } + } + + @ParameterizedTest + @CsvSource({ + "fast,true", + "fast,false", + "full,true", + "full,false", + "detail,true", + "detail,false" + }) + void testUnsupportedKeywordRangeCannotBypassSearchFilter(String mode, boolean refine) + throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 12, "embedding", "category", "price"); + build(table, 0, 12, "note", "category", "price"); + table = + configured(table, mode, false) + .copy( + Collections.singletonMap( + "global-index.filter.refine-from-data", + Boolean.toString(refine))); + Predicate filter = + new PredicateBuilder(table.rowType()).greaterThan(2, BinaryString.fromString("c1")); + assertThat(read(withoutIndex(table), filter)).containsExactly(2, 8, 11); + + // Row 5 is the nearest vector but has a null category. Returning all indexed rows as an + // exact scalar result would incorrectly select it, even with data refinement enabled. + // Unsupported filters use the existing fallback, independently of candidate refinement. + assertSearchResults( + table, + filter, + 5, + Collections.singletonList(new Range(2, 2)), + "fast".equals(mode) + ? Collections.emptyList() + : Arrays.asList(new Range(2, 2), new Range(8, 8)), + false); + assertThat( + table.newFullTextSearchBuilder() + .withQuery("note", "{\"match\":{\"query\":\"keep\"}}") + .withFilter(filter) + .withLimit(1) + .executeLocal() + .results() + .toRangeList()) + .containsExactlyElementsOf( + "fast".equals(mode) + ? Collections.emptyList() + : Collections.singletonList(new Range(2, 2))); + } + + @ParameterizedTest + @CsvSource({ + "fast,true", + "fast,false", + "full,true", + "full,false", + "detail,true", + "detail,false" + }) + void testUnsupportedKeywordRangeInSearchConjunctionAndDisjunction(String mode, boolean refine) + throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 12, "embedding", "category", "price"); + build(table, 0, 12, "note", "category", "price"); + table = + configured(table, mode, false) + .copy( + Collections.singletonMap( + "global-index.filter.refine-from-data", + Boolean.toString(refine))); + PredicateBuilder b = new PredicateBuilder(table.rowType()); + Predicate range = b.greaterThan(2, BinaryString.fromString("c1")); + // AND retains supported candidates, but must refine the complete predicate before top-k. + assertSearchResults( + table, + PredicateBuilder.and(range, b.greaterThan(3, 0.0)), + 5, + refine ? Collections.singletonList(new Range(2, 2)) : Collections.emptyList(), + refine ? Collections.singletonList(new Range(2, 2)) : Collections.emptyList(), + true); + // OR cannot use only its supported branch: it must also retain matches of the range. + assertSearchResults( + table, + PredicateBuilder.or(range, b.equal(3, 6.0)), + 5, + Collections.singletonList(new Range(6, 6)), + "fast".equals(mode) + ? Collections.emptyList() + : Arrays.asList(new Range(2, 2), new Range(6, 6), new Range(8, 8)), + false); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testUnsupportedFilterPreservesPersistedVectorMetric(boolean missingBuildOptions) + throws Exception { + FileStoreTable table = createTable(); + build( + table.copy( + Collections.singletonMap( + "global-index.es-index.fields.embedding.metric", "cosine")), + 0, + 12, + "embedding", + "category", + "price"); + if (missingBuildOptions) { + Map remove = new HashMap<>(); + remove.put("global-index.es-index.fields.embedding.metric", null); + remove.put("global-index.es-index.fields.embedding.dimension", null); + table = table.copy(remove); + } + PredicateBuilder b = new PredicateBuilder(table.rowType()); + // These filters are equivalent for this data. Only the equality is supported by ESLib. + // Both paths must use the metric stored with the index, not the table's current L2 option. + for (Predicate filter : + Arrays.asList( + b.equal(2, BinaryString.fromString("c2")), + b.greaterThan(2, BinaryString.fromString("c1")))) { + assertThat( + table.newVectorSearchBuilder() + .withVectorColumn("embedding") + .withVector(vector(5)) + .withLimit(1) + .withFilter(filter) + .executeLocal() + .results() + .toRangeList()) + .as("%s", filter) + .containsExactly(new Range(8, 8)); + assertThat( + table.newBatchVectorSearchBuilder() + .withVectorColumn("embedding") + .withVectors(new float[][] {vector(5), vector(5)}) + .withLimit(1) + .withFilter(filter) + .executeBatchLocal()) + .allSatisfy( + result -> + assertThat(result.results().toRangeList()) + .containsExactly(new Range(8, 8))); + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testUnsupportedFilterValidatesMetricsAcrossDifferentDefinitions(boolean mixedMetrics) + throws Exception { + FileStoreTable table = createTable(); + FileStoreTable cosine = + table.copy( + Collections.singletonMap( + "global-index.es-index.fields.embedding.metric", "cosine")); + build(cosine, 0, 6, "embedding", "category", "price"); + build(mixedMetrics ? table : cosine, 6, 12, "embedding", "category"); + Predicate filter = + new PredicateBuilder(table.rowType()).greaterThan(2, BinaryString.fromString("c1")); + if (mixedMetrics) { + assertThatThrownBy( + () -> + table.newVectorSearchBuilder() + .withVectorColumn("embedding") + .withVector(vector(5)) + .withLimit(1) + .withFilter(filter) + .executeLocal()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("different metrics"); + } else { + assertThat( + table.newVectorSearchBuilder() + .withVectorColumn("embedding") + .withVector(vector(5)) + .withLimit(1) + .withFilter(filter) + .executeLocal() + .results() + .toRangeList()) + .containsExactly(new Range(8, 8)); + } + } + + @ParameterizedTest + @CsvSource({"full,false", "full,true", "detail,false", "detail,true"}) + void testTypeEvolutionDoesNotDropUnsupportedShards(String mode, boolean inReader) + throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 4, "embedding", "id", "price"); + build(table, 0, 4, "note", "id", "price"); + new FileSystemSchemaManager(fileIO, table.location()) + .commitChanges(SchemaChange.updateColumnType("id", DataTypes.BIGINT())); + table = FileStoreTableFactory.create(fileIO, table.location()); + build(table, 6, 10, "embedding", "id", "price"); + build(table, 6, 10, "note", "id", "price"); + table = configured(table, mode, inReader); + Predicate filter = new PredicateBuilder(table.rowType()).lessThan(0, 9L); + assertThat(read(table, filter)) + .containsExactlyElementsOf(read(withoutIndex(table), filter)); + try (DataEvolutionGlobalIndexScanner scanner = + DataEvolutionGlobalIndexScanner.createForScalarFilters( + table, table.snapshotManager().latestSnapshot(), null, indexFiles) + .get()) { + GlobalIndexEvaluator.Evaluation result = scanner.scanWithCoverage(filter).get(); + assertThat(result.coveredRanges()).containsExactly(new Range(6, 9)); + } + assertThat( + table.newVectorSearchBuilder() + .withVectorColumn("embedding") + .withVector(vector(2)) + .withFilter(filter) + .withLimit(1) + .executeLocal() + .results() + .toRangeList()) + .containsExactly(new Range(2, 2)); + assertThat( + table.newFullTextSearchBuilder() + .withQuery("note", "{\"match\":{\"query\":\"keep\"}}") + .withFilter(filter) + .withLimit(10) + .executeLocal() + .results() + .toRangeList()) + .containsExactly( + new Range(0, 0), new Range(2, 2), new Range(6, 6), new Range(8, 8)); + } + + private FileStoreTable createTable() throws Exception { + Path path = new Path(temporary.toUri()); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("embedding", DataTypes.ARRAY(DataTypes.FLOAT())) + .column("category", DataTypes.STRING()) + .column("price", DataTypes.DOUBLE()) + .column("note", DataTypes.STRING()) + .option("bucket", "-1") + .option("write-only", "true") + .option("row-tracking.enabled", "true") + .option("data-evolution.enabled", "true") + .option("global-index.enabled", "true") + .option("global-index.es-index.fields.embedding.dimension", "3") + .option("global-index.es-index.fields.embedding.algorithm", "hnsw") + .option("global-index.es-index.fields.embedding.metric", "l2") + .option("global-index.es-index.fields.category.type", "keyword") + .option("global-index.es-index.fields.note.type", "fulltext") + .build(); + new FileSystemSchemaManager(fileIO, path).createTable(schema); + FileStoreTable table = FileStoreTableFactory.create(fileIO, path); + BatchWriteBuilder builder = table.newBatchWriteBuilder(); + try (BatchTableWrite writer = builder.newWrite(); + BatchTableCommit commit = builder.newCommit()) { + for (int id = 0; id < 12; id++) { + writer.write( + GenericRow.of( + id, + new GenericArray(vector(id)), + category(id), + price(id), + BinaryString.fromString(id % 2 == 0 ? "keep" : "skip"))); + } + commit.commit(writer.prepareCommit()); + } + return table; + } + + @ParameterizedTest + @CsvSource({ + "fast,true", + "fast,false", + "full,true", + "full,false", + "detail,true", + "detail,false" + }) + void testSameFieldPartialFilterCannotBypassSearchRefinement(String mode, boolean refine) + throws Exception { + FileStoreTable table = createTable(); + build(table, 0, 12, "embedding", "category", "price"); + build(table, 0, 12, "note", "category", "price"); + table = + configured(table, mode, false) + .copy( + Collections.singletonMap( + "global-index.filter.refine-from-data", + Boolean.toString(refine))); + PredicateBuilder b = new PredicateBuilder(table.rowType()); + + // A field ID cannot distinguish these two conjuncts. The range is supported; isNaN + // is not, and all stored prices are finite or null. Both searches must return nothing. + Predicate empty = PredicateBuilder.and(b.greaterThan(3, 0.0), b.isNaN(3)); + assertThat(read(withoutIndex(table), empty)).isEmpty(); + assertSearchFilters(table, empty, Collections.emptyList(), Collections.emptyList()); + + // Row 4 is closest but has category c1. Refinement must keep row 3 before top-k, + // not just discard row 4 afterwards. An always-empty fix would fail this positive case. + Predicate partial = + PredicateBuilder.and( + b.startsWith(2, BinaryString.fromString("c")), + b.startsWith(2, BinaryString.fromString("c1")).negate().get()); + assertSearchFilters( + table, + partial, + refine ? Collections.singletonList(new Range(3, 3)) : Collections.emptyList(), + refine + ? Arrays.asList(new Range(2, 2), new Range(6, 6), new Range(8, 8)) + : Collections.emptyList()); + + // Multiple conditions on the same field are not inherently inexact. All-supported + // conjunctions must continue to work even with data refinement disabled. + Predicate complete = + PredicateBuilder.and( + b.startsWith(2, BinaryString.fromString("c")), + b.notEqual(2, BinaryString.fromString("c1"))); + assertSearchFilters( + table, + complete, + Collections.singletonList(new Range(3, 3)), + Arrays.asList(new Range(2, 2), new Range(6, 6), new Range(8, 8))); + } + + private void assertSearchFilters( + FileStoreTable table, + Predicate filter, + List vectorRows, + List fullTextRows) { + assertSearchResults(table, filter, 4, vectorRows, fullTextRows, true); + } + + private void assertSearchResults( + FileStoreTable table, + Predicate filter, + int queryVector, + List vectorRows, + List fullTextRows, + boolean mustReadIndex) { + assertAll( + () -> { + fileIO.indexOpens.set(0); + assertThat( + table.newVectorSearchBuilder() + .withVectorColumn("embedding") + .withVector(vector(queryVector)) + .withLimit(1) + .withFilter(filter) + .executeLocal() + .results() + .toRangeList()) + .containsExactlyElementsOf(vectorRows); + if (mustReadIndex) { + assertThat(fileIO.indexOpens.get()).isPositive(); + } + }, + () -> { + fileIO.indexOpens.set(0); + assertThat( + table.newFullTextSearchBuilder() + .withQuery("note", "{\"match\":{\"query\":\"keep\"}}") + .withFilter(filter) + .withLimit(10) + .executeLocal() + .results() + .toRangeList()) + .containsExactlyElementsOf(fullTextRows); + if (mustReadIndex) { + assertThat(fileIO.indexOpens.get()).isPositive(); + } + }); + } + + private void build(FileStoreTable table, int from, int to, String... names) throws Exception { + List fields = + Arrays.stream(names).map(table.rowType()::getField).collect(Collectors.toList()); + GlobalIndexFileReadWrite io = + new GlobalIndexFileReadWrite( + fileIO, table.store().pathFactory().globalIndexFileFactory()); + GlobalIndexMultiColumnWriter writer = + (GlobalIndexMultiColumnWriter) + GlobalIndexer.create("es-index", fields, Options.fromMap(table.options())) + .createWriter(io); + for (int id = from; id < to; id++) { + Map values = new HashMap<>(); + if (table.rowType().getTypeAt(0).getTypeRoot() == DataTypeRoot.BIGINT) { + values.put("id", (long) id); + } else { + values.put("id", id); + } + values.put("embedding", new GenericArray(vector(id))); + values.put("category", category(id)); + values.put("price", price(id)); + values.put("note", BinaryString.fromString(id % 2 == 0 ? "keep" : "skip")); + writer.write(id - from, GenericRow.of(Arrays.stream(names).map(values::get).toArray())); + } + Snapshot snapshot = table.snapshotManager().latestSnapshot(); + List added = new ArrayList<>(); + for (ResultEntry result : writer.finish()) { + added.add( + new IndexFileMeta( + "es-index", + result.fileName(), + io.fileSize(result.fileName()), + result.rowCount(), + new GlobalIndexMeta( + from, + to - 1, + fields.get(0).id(), + fields.stream().skip(1).mapToInt(DataField::id).toArray(), + result.meta(), + new DataEvolutionIndexSourceMeta(snapshot.id()).serialize()), + null)); + fileIO.indexFiles.add(result.fileName()); + } + try (TableCommitImpl commit = table.newCommit("es-scalar-test")) { + commit.commit( + snapshot.id(), + Collections.singletonList( + new CommitMessageImpl( + BinaryRow.EMPTY_ROW, + 0, + null, + DataIncrement.indexIncrement(added), + CompactIncrement.emptyIncrement()))); + } + indexFiles.addAll(added); + } + + private List read(FileStoreTable table, Predicate predicate) throws Exception { + ReadBuilder builder = table.newReadBuilder().withFilter(predicate); + List splits = new ArrayList<>(); + for (Split split : builder.newScan().plan().splits()) { + splits.add( + InstantiationUtil.deserializeObject( + InstantiationUtil.serializeObject(split), getClass().getClassLoader())); + } + List result = new ArrayList<>(); + try (RecordReader reader = + builder.newRead().executeFilter().createReader(splits)) { + reader.forEachRemaining( + row -> + result.add( + table.rowType().getTypeAt(0).getTypeRoot() + == DataTypeRoot.BIGINT + ? (int) row.getLong(0) + : row.getInt(0))); + } + Collections.sort(result); + return result; + } + + private FileStoreTable configured(FileStoreTable table, String mode, boolean inReader) { + Map options = new HashMap<>(); + options.put("scalar-index.search-mode", mode); + options.put("global-index.query-in-reader.enabled", Boolean.toString(inReader)); + return table.copy(options); + } + + private FileStoreTable withoutIndex(FileStoreTable table) { + return table.copy(Collections.singletonMap("global-index.enabled", "false")); + } + + private static float[] vector(int id) { + return new float[] {id + 0.1f, id * 0.02f, 1.0f}; + } + + private static BinaryString category(int id) { + return id % 5 == 0 ? null : BinaryString.fromString("c" + id % 3); + } + + private static Double price(int id) { + return id % 4 == 0 ? null : (double) id; + } + + /** Uses the protected entry point also called by the Flink and Spark local search paths. */ + private static class SubclassVectorRead extends DataEvolutionVectorRead { + + private GlobalIndexer suppliedIndexer; + + private SubclassVectorRead(FileStoreTable table, Predicate filter) { + super(table, null, filter, 1, table.rowType().getField("embedding"), vector(5), null); + } + + @Override + protected ScoredGlobalIndexResult readIndexed( + List splits, GlobalIndexer globalIndexer) { + suppliedIndexer = globalIndexer; + return super.readIndexed(splits, globalIndexer); + } + } + + private static class AuditedFileIO extends LocalFileIO { + private static final long serialVersionUID = 1L; + private final Set indexFiles = new HashSet<>(); + private final AtomicInteger indexOpens = new AtomicInteger(); + + @Override + public SeekableInputStream newInputStream(Path path) throws IOException { + if (indexFiles.contains(path.getName())) { + indexOpens.incrementAndGet(); + } + return super.newInputStream(path); + } + } +}