From b34c28a3f633f5d2ef2a2aaad6c5a3130340014d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=B5=B4=E9=95=BF=E9=A3=8E?= Date: Sun, 18 Oct 2026 22:56:11 -0700 Subject: [PATCH 1/4] [core] Support scalar queries on grouped ES indexes --- .../globalindex/GlobalIndexerFactory.java | 11 + .../GlobalIndexerFactoryUtils.java | 6 + .../DataEvolutionGlobalIndexCoverage.java | 12 +- .../DataEvolutionGlobalIndexScanner.java | 25 +- .../paimon/globalindex/GlobalIndexQuery.java | 10 +- .../AbstractDataEvolutionVectorRead.java | 12 +- .../source/DataEvolutionFullTextRead.java | 15 + .../globalindex/GlobalIndexQueryTest.java | 65 +++ .../source/FullTextSearchBuilderTest.java | 40 ++ .../VectorSearchRowFilterExactnessTest.java | 43 ++ paimon-eslib/pom.xml | 48 ++ .../index/ESIndexGlobalIndexerFactory.java | 5 + .../eslib/index/ESIndexScalarTableTest.java | 438 ++++++++++++++++++ 13 files changed, 711 insertions(+), 19 deletions(-) create mode 100644 paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexScalarTableTest.java 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-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..e0be3bc3d40e 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 @@ -330,8 +330,9 @@ 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)); } } @@ -358,6 +359,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); 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..12d2460dce40 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 @@ -253,6 +253,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 +307,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); } 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/test/java/org/apache/paimon/globalindex/GlobalIndexQueryTest.java b/paimon-core/src/test/java/org/apache/paimon/globalindex/GlobalIndexQueryTest.java index 844317fc57f0..89a9632bbcca 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 @@ -76,6 +76,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 +86,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 = 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..3269c2e1c557 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,7 +24,9 @@ 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.GlobalIndexResult; import org.apache.paimon.globalindex.GlobalIndexSingleColumnWriter; import org.apache.paimon.globalindex.btree.BTreeGlobalIndexerFactory; @@ -52,12 +54,19 @@ 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 static org.assertj.core.api.Assertions.assertThat; +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; /** * Vector search with a row filter must rank an exact match set: candidate-only scalar index answers @@ -66,6 +75,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(); 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/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/ESIndexScalarTableTest.java b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexScalarTableTest.java new file mode 100644 index 000000000000..b2210e962f65 --- /dev/null +++ b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexScalarTableTest.java @@ -0,0 +1,438 @@ +/* + * 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.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.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.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.ReadBuilder; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.DataField; +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 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; + +/** 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)); + } + + @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)); + } + + 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; + } + + 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<>(); + 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(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; + } + + 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); + } + } +} From 8e12fe3cb65d6c310260ee8ef6b38ad59192210f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=B5=B4=E9=95=BF=E9=A3=8E?= Date: Mon, 26 Oct 2026 06:26:44 -0700 Subject: [PATCH 2/4] [core] Track scalar predicate completeness before search ranking Propagate explicit predicate completeness through evaluation, ranges, reader wrappers and planning so partial same-field filters cannot bypass refinement. Return unsupported for unsupported ESLib scalar operations, distinguish zero hits, and preserve persisted vector metrics when falling back to data without changing backend conflict policies. Add scalar, coverage, BTree, vector and full-text regressions. Keep ESLib 1.0.7 and the existing vector-primary design. --- .../ContainsRefinementEvaluator.java | 19 +- .../globalindex/GlobalIndexEvaluator.java | 71 ++- .../GlobalIndexPredicateResult.java | 40 ++ .../paimon/globalindex/GlobalIndexReader.java | 18 +- .../globalindex/OffsetGlobalIndexReader.java | 13 + .../globalindex/UnionGlobalIndexReader.java | 42 ++ .../btree/LazyFilteredBTreeReader.java | 10 + .../globalindex/GlobalIndexEvaluatorTest.java | 225 +++++++++- .../LazyFilteredBTreeIndexReaderTest.java | 1 + .../paimon/globalindex/GlobalIndexQuery.java | 83 +++- .../AbstractDataEvolutionVectorRead.java | 63 ++- .../table/source/FilteredRowIdReader.java | 4 + .../globalindex/GlobalIndexQueryTest.java | 119 ++++- .../BTreeScalarFilterCompletenessTest.java | 286 ++++++++++++ .../VectorSearchRowFilterExactnessTest.java | 70 +++ .../eslib/index/ESIndexGlobalIndexReader.java | 38 +- .../index/ESIndexGlobalIndexE2ETest.java | 160 +++++-- .../eslib/index/ESIndexScalarTableTest.java | 407 +++++++++++++++++- 18 files changed, 1556 insertions(+), 113 deletions(-) create mode 100644 paimon-common/src/main/java/org/apache/paimon/globalindex/GlobalIndexPredicateResult.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/source/BTreeScalarFilterCompletenessTest.java 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/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/GlobalIndexQuery.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/GlobalIndexQuery.java index e0be3bc3d40e..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); } } @@ -339,7 +351,8 @@ private static GlobalIndexQuery createForIndexedField( 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. */ @@ -429,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) @@ -499,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( @@ -507,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); @@ -518,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; @@ -531,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; @@ -547,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; @@ -555,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); @@ -568,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), @@ -677,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); @@ -687,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 12d2460dce40..b400363513f4 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; @@ -73,6 +74,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 +102,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])); @@ -152,9 +157,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 +167,7 @@ protected void prepareSplits( List rawSplits) { splitSearchSplits(splits, indexSplits, rawSplits); scalarPreFilter = null; + fallbackVectorMetric = null; if (filter == null || indexSplits.isEmpty()) { return; } @@ -178,9 +184,57 @@ 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. + IndexFileMeta firstFile = split.vectorIndexFiles().get(0); + GlobalIndexer indexer = + indexers.computeIfAbsent(firstFile.indexType(), k -> new HashMap<>()) + .computeIfAbsent( + checkNotNull(firstFile.globalIndexMeta()).getIndexedFieldIds(), + k -> createGlobalIndexer(Collections.singletonList(split))); + 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) { @@ -761,6 +815,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/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 89a9632bbcca..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; @@ -183,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 @@ -260,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); @@ -285,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/VectorSearchRowFilterExactnessTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchRowFilterExactnessTest.java index 3269c2e1c557..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 @@ -27,8 +27,11 @@ 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; @@ -60,9 +63,11 @@ 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; @@ -524,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/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/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 index b2210e962f65..9575e1a2e65c 100644 --- 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 @@ -30,6 +30,7 @@ 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; @@ -46,6 +47,7 @@ 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; @@ -56,6 +58,7 @@ import org.apache.paimon.table.source.ReadBuilder; import org.apache.paimon.table.source.Split; 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; @@ -64,6 +67,7 @@ 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; @@ -78,6 +82,8 @@ 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 { @@ -292,6 +298,290 @@ void testScalarPrimaryAndFullTextPreFilter() throws Exception { .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 = @@ -331,6 +621,110 @@ private FileStoreTable createTable() throws Exception { 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()); @@ -343,6 +737,11 @@ private void build(FileStoreTable table, int from, int to, String... names) thro .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)); @@ -393,7 +792,13 @@ private List read(FileStoreTable table, Predicate predicate) throws Exc List result = new ArrayList<>(); try (RecordReader reader = builder.newRead().executeFilter().createReader(splits)) { - reader.forEachRemaining(row -> result.add(row.getInt(0))); + reader.forEachRemaining( + row -> + result.add( + table.rowType().getTypeAt(0).getTypeRoot() + == DataTypeRoot.BIGINT + ? (int) row.getLong(0) + : row.getInt(0))); } Collections.sort(result); return result; From 59da228582a16c1bebd0ca794b3bcacf8226e5db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=B5=B4=E9=95=BF=E9=A3=8E?= Date: Wed, 28 Oct 2026 20:30:52 -0700 Subject: [PATCH 3/4] [core] Read vector splits with matching index definitions Split range-based vector tasks by index type and ordered field IDs, then reuse one indexer per definition for both single and batch search. Resolve compatible persisted metrics after all shards register metadata, preserving backend shard-conflict policies and scoring for reranking and raw tails. Add real-archive regressions for mixed definitions, field order, overlapping and identical ranges, raw tails, and incompatible metrics. --- .../AbstractDataEvolutionVectorRead.java | 91 +++++++++++- .../source/DataEvolutionBatchVectorRead.java | 15 +- .../table/source/DataEvolutionVectorRead.java | 18 ++- .../eslib/index/ESIndexScalarTableTest.java | 136 ++++++++++++++++++ 4 files changed, 241 insertions(+), 19 deletions(-) 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 b400363513f4..0c595bd9aea9 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 @@ -65,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; @@ -133,6 +134,58 @@ 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) { + Map, GlobalIndexer>> definitions = new HashMap<>(); + GlobalIndexer first = createGlobalIndexer(splits); + 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( @@ -200,12 +253,7 @@ private String fallbackVectorMetric(List splits) { } // 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. - IndexFileMeta firstFile = split.vectorIndexFiles().get(0); - GlobalIndexer indexer = - indexers.computeIfAbsent(firstFile.indexType(), k -> new HashMap<>()) - .computeIfAbsent( - checkNotNull(firstFile.globalIndexMeta()).getIndexedFieldIds(), - k -> createGlobalIndexer(Collections.singletonList(split))); + GlobalIndexer indexer = indexerForDefinition(split, indexers); try (GlobalIndexReader ignored = indexer.createReader( meta -> table.fileIO().newInputStream(meta.filePath()), @@ -772,13 +820,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) { 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/DataEvolutionVectorRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorRead.java index fbac5ef7b786..89ea43a6b91b 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 @@ -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; @@ -73,17 +74,19 @@ protected GlobalIndexResult readSplits(List splits) return GlobalIndexResult.createEmpty(); } - GlobalIndexer globalIndexer = - indexSplits.isEmpty() ? null : createGlobalIndexer(indexSplits); + List globalIndexers = + indexSplits.isEmpty() ? Collections.emptyList() : createGlobalIndexers(indexSplits); ScoredGlobalIndexResult result = indexSplits.isEmpty() ? ScoredGlobalIndexResult.createEmpty() - : readIndexed(indexSplits, globalIndexer); + : readIndexed(indexSplits, globalIndexers); + GlobalIndexer globalIndexer = + globalIndexers.isEmpty() ? null : metricIndexer(globalIndexers); return withRawSearch(result, rawSplits, globalIndexer, rawPreFilter(rawSplits), vector); } protected ScoredGlobalIndexResult readIndexed( - List splits, GlobalIndexer globalIndexer) { + List splits, List globalIndexers) { 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-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexScalarTableTest.java b/paimon-eslib/src/test/java/org/apache/paimon/eslib/index/ESIndexScalarTableTest.java index 9575e1a2e65c..cbed716581da 100644 --- 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 @@ -55,8 +55,10 @@ 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.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; @@ -240,6 +242,140 @@ void testVectorPrimaryDesignIsUnchanged() throws Exception { .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 + @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(); From d31e015a7e0e591b7f0981fae14f36929fe6e545 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=B5=B4=E9=95=BF=E9=A3=8E?= Date: Thu, 29 Oct 2026 03:10:52 -0700 Subject: [PATCH 4/4] [core] Preserve vector reader entry point compatibility --- .../AbstractDataEvolutionVectorRead.java | 6 +- .../table/source/DataEvolutionVectorRead.java | 14 ++--- .../eslib/index/ESIndexScalarTableTest.java | 59 +++++++++++++++++++ 3 files changed, 71 insertions(+), 8 deletions(-) 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 0c595bd9aea9..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 @@ -136,8 +136,12 @@ protected GlobalIndexer createGlobalIndexer(List splits) /** 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<>(); - GlobalIndexer first = createGlobalIndexer(splits); IndexFileMeta firstFile = firstVectorIndexFile(splits); definitions .computeIfAbsent(firstFile.indexType(), k -> new HashMap<>()) 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 89ea43a6b91b..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 @@ -32,7 +32,6 @@ import javax.annotation.Nullable; import java.util.ArrayList; -import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; @@ -74,19 +73,20 @@ protected GlobalIndexResult readSplits(List splits) return GlobalIndexResult.createEmpty(); } - List globalIndexers = - indexSplits.isEmpty() ? Collections.emptyList() : createGlobalIndexers(indexSplits); + GlobalIndexer globalIndexer = + indexSplits.isEmpty() ? null : createGlobalIndexer(indexSplits); ScoredGlobalIndexResult result = indexSplits.isEmpty() ? ScoredGlobalIndexResult.createEmpty() - : readIndexed(indexSplits, globalIndexers); - GlobalIndexer globalIndexer = - globalIndexers.isEmpty() ? null : metricIndexer(globalIndexers); + : readIndexed(indexSplits, globalIndexer); return withRawSearch(result, rawSplits, globalIndexer, rawPreFilter(rawSplits), vector); } protected ScoredGlobalIndexResult readIndexed( - List splits, List globalIndexers) { + 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); 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 index cbed716581da..76fd21e30cab 100644 --- 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 @@ -36,6 +36,8 @@ 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; @@ -56,6 +58,8 @@ 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; @@ -307,6 +311,44 @@ void testVectorDefinitionsPreserveMetricForRawAndRefine(boolean rawTail, boolean 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) @@ -963,6 +1005,23 @@ 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<>();