diff --git a/docs/docs/multimodal-table/global-index.mdx b/docs/docs/multimodal-table/global-index.mdx index ace4770ba7e8..731fb8c078ab 100644 --- a/docs/docs/multimodal-table/global-index.mdx +++ b/docs/docs/multimodal-table/global-index.mdx @@ -311,8 +311,8 @@ catalog.alter_table( Global index files cover row-id ranges. If more rows are appended after an index is built, those new rows are not automatically covered by the existing index files. Build the global index again -to create index files for newly uncovered data. By default, queries use fast search and only read -indexed row ranges; rows in uncovered ranges are not returned for that indexed query. +to create index files for newly uncovered data. Scalar queries default to `full` search, while +vector and full-text queries default to `fast` and only search indexed row ranges. To improve freshness for query types that support raw-data search, set: @@ -321,7 +321,7 @@ To improve freshness for query types that support raw-data search, set: ```sql -ALTER TABLE my_table SET ('global-index.search-mode' = 'full'); +ALTER TABLE my_table SET ('vector-index.search-mode' = 'full'); ``` @@ -333,14 +333,14 @@ from pypaimon.schema.schema_change import SchemaChange catalog.alter_table( "db.my_table", - [SchemaChange.set_option("global-index.search-mode", "full")], + [SchemaChange.set_option("vector-index.search-mode", "full")], ) ``` For a read-only override on an existing `Table` instance: ```python -full_table = table.copy({"global-index.search-mode": "full"}) +full_table = table.copy({"vector-index.search-mode": "full"}) ``` @@ -353,6 +353,11 @@ only when such a gap exists. Use `detail` search when data files may have been r after index creation; it scans data file metadata to find the exact unindexed row ranges and can handle index invalidation caused by updates or rewrites. +Use `scalar-index.search-mode`, `vector-index.search-mode`, or +`full-text-index.search-mode` for one index family. The legacy +`global-index.search-mode` has no default and is used as a fallback when the corresponding +family-specific option is not explicitly configured. + To temporarily disable global-index scan acceleration while keeping the index files, set: @@ -389,7 +394,10 @@ These table options affect global index build and read behavior: | Option | Default | Description | |---|---|---| | `global-index.enabled` | `true` | Whether scans can use global indexes. | -| `global-index.search-mode` | `fast` | Search mode for global-index queries. `fast` searches indexed data only. `full` checks snapshot `nextRowId` against global index row-id coverage and scans raw data only if a gap exists. `detail` scans data file metadata to find exact unindexed rows and can handle index invalidation caused by updates or rewrites. | +| `global-index.search-mode` | Not set | Legacy fallback search mode for global-index queries. Family-specific options take precedence. | +| `scalar-index.search-mode` | `full` | Search mode for BTree and Bitmap queries. | +| `vector-index.search-mode` | `fast` | Search mode for vector queries. | +| `full-text-index.search-mode` | `fast` | Search mode for full-text queries. | | `global-index.external-path` | Not set | Root directory for global index files. If not set, files are stored under the table index directory. | | `global-index.column-update-action` | `THROW_ERROR` | Action for updates to indexed columns. `THROW_ERROR` rejects the update, `DROP_PARTITION_INDEX` drops affected partition indexes, and `IGNORE` keeps existing index files unchanged. Use `IGNORE` only when a stale index is acceptable; for example, a vector updated from `NULL` to a value remains invisible until the index is rebuilt. | | `sorted-index.records-per-range` | `10000000` | Expected number of records per sorted global index file for BTree and Bitmap builds. | diff --git a/docs/docs/multimodal-table/global-index/full-text.mdx b/docs/docs/multimodal-table/global-index/full-text.mdx index ae8ea59d61de..5886ef2d7fcc 100644 --- a/docs/docs/multimodal-table/global-index/full-text.mdx +++ b/docs/docs/multimodal-table/global-index/full-text.mdx @@ -282,8 +282,8 @@ print(pa_table) -By default, `global-index.search-mode=fast` searches indexed row ranges only. -When `global-index.search-mode` is `full` or `detail`, Paimon also covers +By default, `full-text-index.search-mode=fast` searches indexed row ranges only. +When `full-text-index.search-mode` is `full` or `detail`, Paimon also covers unindexed row ranges by reading raw rows and building a temporary native full-text index for the searched column. diff --git a/docs/docs/primary-key-table/global-index.mdx b/docs/docs/primary-key-table/global-index.mdx index 3098d71c52e4..e76845e6c390 100644 --- a/docs/docs/primary-key-table/global-index.mdx +++ b/docs/docs/primary-key-table/global-index.mdx @@ -234,7 +234,9 @@ schema validation. | `fields..pk-btree.index.options` | Not set | JSON object containing BTree build options. Unqualified keys are scoped to `btree-index`. | | `pk-bitmap.index.columns` | Not set | Comma-separated columns which own independent Bitmap indexes. | | `fields..pk-bitmap.index.options` | Not set | JSON object containing Bitmap build options. Unqualified keys are scoped to `bitmap-index`. | -| `global-index.search-mode` | `fast` | Search mode for primary-key Vector and Full Text queries. `fast` searches indexed data only, so uncovered files are omitted. For Vector, `full` and `detail` search uncovered files exactly. Primary-key Full Text supports only `fast`. | +| `global-index.search-mode` | Not set | Legacy fallback search mode. Family-specific options take precedence. | +| `vector-index.search-mode` | `fast` | Search mode for primary-key Vector queries. `full` and `detail` search uncovered files exactly. | +| `full-text-index.search-mode` | `fast` | Search mode for primary-key Full Text queries. Only `fast` is currently supported. | For algorithm-specific options, see the corresponding [BTree](../multimodal-table/global-index/btree), @@ -272,10 +274,10 @@ Index construction can execute asynchronously inside the writer. A writer which compaction also waits for active index maintenance; a non-blocking writer can complete maintenance in a later commit. Coverage can therefore be temporarily partial. -Coverage behavior depends on the index family and search mode. `global-index.search-mode` defaults -to `fast`. For primary-key Vector and Full Text queries, `fast` searches only data covered by an -active index group. Uncovered files are not searched, so partial coverage can make results -incomplete. +Coverage behavior depends on the index family and search mode. `vector-index.search-mode` and +`full-text-index.search-mode` default to `fast`. In `fast` mode, primary-key Vector and Full Text +queries search only data covered by an active index group. Uncovered files are not searched, so +partial coverage can make results incomplete. - BTree and Bitmap scans always read uncovered files through the ordinary data path. Partial coverage affects their acceleration, not result completeness. The original scalar predicate and @@ -283,7 +285,7 @@ incomplete. - Vector search in `full` or `detail` mode evaluates files without an active ANN group exactly and merges those results with ANN candidates. In the default `fast` mode, those files are omitted. -Primary-key Full Text currently supports only `global-index.search-mode = fast`. It searches +Primary-key Full Text currently supports only `full-text-index.search-mode = fast`. It searches persistent archives and ignores uncovered files; `full` and `detail` are rejected because a merge-aware logical-row fallback is not implemented. Consequently, newly appended Level-0 rows become full-text searchable only after compaction publishes an eligible data file and archive. diff --git a/docs/docs/pypaimon/data-evolution.md b/docs/docs/pypaimon/data-evolution.md index f5d58d75e4e9..43f31c63b845 100644 --- a/docs/docs/pypaimon/data-evolution.md +++ b/docs/docs/pypaimon/data-evolution.md @@ -99,7 +99,7 @@ You can use `update_by_predicate` for SQL-like `UPDATE ... SET ... WHERE ...` operations. The `Predicate` identifies rows to update, and the assignment map contains literal values for updated columns. When global indexes are available, `update_by_predicate` discovers matching -`_ROW_ID` values with `global-index.search-mode=full` on the configured +`_ROW_ID` values with `scalar-index.search-mode=full` on the configured point-in-time scan snapshot or, if none is configured, the latest snapshot. ```python diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 2841618e7c34..539ceaef90c6 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -752,6 +752,12 @@ Integer For streaming write, full compaction will be constantly triggered after delta commits. For batch write, full compaction will be triggered with each commit as long as this value is greater than 0. + +
full-text-index.search-mode
+ fast +

Enum

+ Search mode for full-text index queries.

Possible values: +
global-index.build.max-parallelism
4096 @@ -790,9 +796,9 @@
global-index.search-mode
- fast + (none)

Enum

- Search mode for global index queries. Supported values are 'fast', 'full', and 'detail'.

Possible values: + Fallback search mode for global index queries.

Possible values:
global-index.thread-num
@@ -1320,6 +1326,12 @@ String The field that generates the row kind for primary key table, the row kind determines which data is '+I', '-U', '+U' or '-D'. + +
scalar-index.search-mode
+ full +

Enum

+ Search mode for scalar index queries.

Possible values: +
scan.bounded.watermark
(none) @@ -1747,6 +1759,12 @@ String Specifies column names that should be stored as vector type. This is used when you want to treat a ARRAY column as a VECTOR. + +
vector-index.search-mode
+ fast +

Enum

+ Search mode for vector index queries.

Possible values: +
vector-search.distribute.enabled
false diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index e0c7fd4004c7..39c0d16e14db 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -2705,11 +2705,27 @@ public String toString() { public static final ConfigOption GLOBAL_INDEX_SEARCH_MODE = key("global-index.search-mode") + .enumType(GlobalIndexSearchMode.class) + .noDefaultValue() + .withDescription("Fallback search mode for global index queries."); + + public static final ConfigOption SCALAR_INDEX_SEARCH_MODE = + key("scalar-index.search-mode") + .enumType(GlobalIndexSearchMode.class) + .defaultValue(GlobalIndexSearchMode.FULL) + .withDescription("Search mode for scalar index queries."); + + public static final ConfigOption VECTOR_INDEX_SEARCH_MODE = + key("vector-index.search-mode") .enumType(GlobalIndexSearchMode.class) .defaultValue(GlobalIndexSearchMode.FAST) - .withDescription( - "Search mode for global index queries. " - + "Supported values are 'fast', 'full', and 'detail'."); + .withDescription("Search mode for vector index queries."); + + public static final ConfigOption FULL_TEXT_INDEX_SEARCH_MODE = + key("full-text-index.search-mode") + .enumType(GlobalIndexSearchMode.class) + .defaultValue(GlobalIndexSearchMode.FAST) + .withDescription("Search mode for full-text index queries."); public static final ConfigOption GLOBAL_INDEX_THREAD_NUM = key("global-index.thread-num") @@ -4312,10 +4328,32 @@ public boolean globalIndexEnabled() { return options.get(GLOBAL_INDEX_ENABLED); } + @Nullable public GlobalIndexSearchMode globalIndexSearchMode() { return options.get(GLOBAL_INDEX_SEARCH_MODE); } + public GlobalIndexSearchMode scalarIndexSearchMode() { + return indexSearchMode(SCALAR_INDEX_SEARCH_MODE); + } + + public GlobalIndexSearchMode vectorIndexSearchMode() { + return indexSearchMode(VECTOR_INDEX_SEARCH_MODE); + } + + public GlobalIndexSearchMode fullTextIndexSearchMode() { + return indexSearchMode(FULL_TEXT_INDEX_SEARCH_MODE); + } + + private GlobalIndexSearchMode indexSearchMode( + ConfigOption familySearchMode) { + if (options.contains(familySearchMode)) { + return options.get(familySearchMode); + } + return options.getOptional(GLOBAL_INDEX_SEARCH_MODE) + .orElseGet(() -> options.get(familySearchMode)); + } + public Integer globalIndexThreadNum() { return options.get(GLOBAL_INDEX_THREAD_NUM); } diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java index aacae5ce681b..6dbbca8e7705 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java @@ -315,7 +315,7 @@ private Optional evalGlobalIndex() { LOG.info( "Scan table '{}' with global index. searchMode='{}', total={} ms, metadata={} ms, lookup={} ms, coverage={} ms.", table.name(), - options.globalIndexSearchMode(), + options.scalarIndexSearchMode(), totalDuration / 1_000_000, metadataDuration / 1_000_000, lookupDuration / 1_000_000, 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 55678824e62c..122c5f404e81 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 @@ -51,6 +51,7 @@ public class DataEvolutionGlobalIndexCoverage { private final FileStoreTable table; @Nullable private final Snapshot snapshot; @Nullable private final PartitionPredicate partitionFilter; + private final GlobalIndexSearchMode searchMode; private final Map> coverageByField; public DataEvolutionGlobalIndexCoverage( @@ -58,9 +59,24 @@ public DataEvolutionGlobalIndexCoverage( @Nullable Snapshot snapshot, @Nullable PartitionPredicate partitionFilter, Collection indexFiles) { + this( + table, + snapshot, + partitionFilter, + indexFiles, + table.coreOptions().scalarIndexSearchMode()); + } + + public DataEvolutionGlobalIndexCoverage( + FileStoreTable table, + @Nullable Snapshot snapshot, + @Nullable PartitionPredicate partitionFilter, + Collection indexFiles, + GlobalIndexSearchMode searchMode) { this.table = table; this.snapshot = snapshot; this.partitionFilter = partitionFilter; + this.searchMode = checkNotNull(searchMode); this.coverageByField = new HashMap<>(); for (IndexFileMeta indexFile : indexFiles) { GlobalIndexMeta meta = checkNotNull(indexFile.globalIndexMeta()); @@ -83,7 +99,6 @@ public List unindexedRanges(int fieldId) { } public List unindexedRanges(Collection fieldIds) { - GlobalIndexSearchMode searchMode = table.coreOptions().globalIndexSearchMode(); if (searchMode == GlobalIndexSearchMode.FAST) { return Collections.emptyList(); } 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 5f07e8a17e5d..215e1b31c56d 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 @@ -93,7 +93,12 @@ private DataEvolutionGlobalIndexScanner( GlobalIndexReadThreadPool.getExecutorService(options.get(GLOBAL_INDEX_THREAD_NUM)); this.indexPathFactory = indexPathFactory; this.coverage = - new DataEvolutionGlobalIndexCoverage(table, snapshot, partitionFilter, indexFiles); + new DataEvolutionGlobalIndexCoverage( + table, + snapshot, + partitionFilter, + indexFiles, + table.coreOptions().scalarIndexSearchMode()); GlobalIndexFileReader indexFileReader = meta -> fileIO.newInputStream(meta.filePath()); Map indexMetas = new HashMap<>(); Map> extraIndexMetas = new HashMap<>(); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java index 1b6ef7adc71d..3dc1f36812f0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionFullTextScan.java @@ -117,7 +117,11 @@ public Plan scan() { if (!allIndexFiles.isEmpty()) { List rawRowRanges = new DataEvolutionGlobalIndexCoverage( - table, snapshot, partitionFilter, allIndexFiles) + table, + snapshot, + partitionFilter, + allIndexFiles, + table.coreOptions().fullTextIndexSearchMode()) .unindexedRanges(textColumnIds); if (!rawRowRanges.isEmpty()) { splits.add(new RawFullTextSearchSplit(rawRowRanges)); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java index 317ffc6d5235..e9a26020d5b1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataEvolutionVectorScan.java @@ -18,6 +18,7 @@ package org.apache.paimon.table.source; +import org.apache.paimon.CoreOptions.GlobalIndexSearchMode; import org.apache.paimon.Snapshot; import org.apache.paimon.globalindex.DataEvolutionGlobalIndexCoverage; import org.apache.paimon.index.GlobalIndexMeta; @@ -144,22 +145,33 @@ public Plan scan() { splits.add(new IndexVectorSearchSplit(range.from, range.to, vectorFiles, scalarFiles)); } + GlobalIndexSearchMode vectorSearchMode = table.coreOptions().vectorIndexSearchMode(); List rawRowRanges = new DataEvolutionGlobalIndexCoverage( - table, snapshot, partitionFilter, vectorIndexFiles) + table, + snapshot, + partitionFilter, + vectorIndexFiles, + vectorSearchMode) .unindexedRanges(vectorColumn.id()); if (filter != null) { + List scalarUnindexedRanges = + new DataEvolutionGlobalIndexCoverage( + table, + snapshot, + partitionFilter, + scalarIndexFiles(allIndexFiles), + table.coreOptions().scalarIndexSearchMode()) + .unindexedRanges(table.rowType(), filter); + if (vectorSearchMode == GlobalIndexSearchMode.FAST) { + scalarUnindexedRanges = + Range.and( + scalarUnindexedRanges, + Range.sortAndMergeOverlap( + new ArrayList<>(vectorByRange.keySet()), true)); + } rawRowRanges = - Range.sortAndMergeOverlap( - addAll( - rawRowRanges, - new DataEvolutionGlobalIndexCoverage( - table, - snapshot, - partitionFilter, - scalarIndexFiles(allIndexFiles)) - .unindexedRanges(table.rowType(), filter)), - true); + Range.sortAndMergeOverlap(addAll(rawRowRanges, scalarUnindexedRanges), true); } if (!rawRowRanges.isEmpty()) { splits.add( diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java index 3374141a33e7..c514301be05d 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyFullTextRead.java @@ -69,7 +69,7 @@ public PrimaryKeyFullTextRead( definition.fieldId() == textField.id(), "Full-text definition does not match field %s.", textField.name()); - checkFastSearchMode(table.coreOptions().globalIndexSearchMode()); + checkFastSearchMode(table.coreOptions().fullTextIndexSearchMode()); ProductionSearch production = new ProductionSearch(table, definition, textField, query, limit); this.limit = limit; @@ -87,7 +87,7 @@ public PrimaryKeyFullTextRead( private static void checkFastSearchMode(GlobalIndexSearchMode searchMode) { if (searchMode != GlobalIndexSearchMode.FAST) { throw new UnsupportedOperationException( - "Primary-key full-text search only supports the FAST global-index search mode; " + "Primary-key full-text search only supports the FAST full-text-index search mode; " + "FULL and DETAIL require merge-aware logical-row fallback."); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java index 46232947f3b1..6a632ed64607 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/PrimaryKeyVectorRead.java @@ -313,7 +313,7 @@ CompletableFuture> searchBatchAsync( annSearcher, searchOptions, metric, - table.coreOptions().globalIndexSearchMode()); + table.coreOptions().vectorIndexSearchMode()); return bucketSearch .searchBatchAsync( state, diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index b3e91a3a4b53..96643624a3a0 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java @@ -107,6 +107,39 @@ public void testIgnoreGlobalIndexColumnUpdateAction() { .isEqualTo(CoreOptions.GlobalIndexColumnUpdateAction.IGNORE); } + @Test + public void testIndexSearchModes() { + Options conf = new Options(); + CoreOptions options = new CoreOptions(conf); + assertThat(options.globalIndexSearchMode()).isNull(); + assertThat(options.scalarIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FULL); + assertThat(options.vectorIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FAST); + assertThat(options.fullTextIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FAST); + + conf.set(CoreOptions.GLOBAL_INDEX_SEARCH_MODE, CoreOptions.GlobalIndexSearchMode.DETAIL); + options = new CoreOptions(conf); + assertThat(options.scalarIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.DETAIL); + assertThat(options.vectorIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.DETAIL); + assertThat(options.fullTextIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.DETAIL); + + conf.set(CoreOptions.SCALAR_INDEX_SEARCH_MODE, CoreOptions.GlobalIndexSearchMode.FAST); + conf.set(CoreOptions.VECTOR_INDEX_SEARCH_MODE, CoreOptions.GlobalIndexSearchMode.FULL); + conf.set(CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE, CoreOptions.GlobalIndexSearchMode.FULL); + options = new CoreOptions(conf); + assertThat(options.scalarIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FAST); + assertThat(options.vectorIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FULL); + assertThat(options.fullTextIndexSearchMode()) + .isEqualTo(CoreOptions.GlobalIndexSearchMode.FULL); + } + @Test public void testBlobSplitByFileSizeDefault() { Options conf = new Options(); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java index c0d8ea01aaa6..ff1d993cf8eb 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java @@ -208,7 +208,7 @@ public void testGlobalIndexDiagnosticLogs() throws Exception { assertThat(logs) .containsPattern( "INFO Scan table '[^']+' with global index\\. " - + "searchMode='fast', total=\\d+ ms, metadata=\\d+ ms, " + + "searchMode='full', total=\\d+ ms, metadata=\\d+ ms, " + "lookup=\\d+ ms, coverage=\\d+ ms\\.") .containsPattern( "INFO Global index lookup table='[^']+', type='btree', " @@ -238,8 +238,9 @@ public void testBTreeGlobalIndexSearchModeControlsUnindexedData() throws Excepti BinaryString.fromString("a100"), BinaryString.fromString("a700"))); - assertThat(readF1(table, predicate)).containsExactly("a100"); + assertThat(readF1(table, predicate)).containsExactly("a100", "a700"); + assertThat(readF1(tableWithSearchMode(table, "fast"), predicate)).containsExactly("a100"); assertThat(readF1(tableWithSearchMode(table, "full"), predicate)) .containsExactly("a100", "a700"); assertThat(readF1(tableWithSearchMode(table, "detail"), predicate)) @@ -251,7 +252,8 @@ public void testBTreeGlobalIndexSearchModeControlsUnindexedData() throws Excepti builder.equal(1, BinaryString.fromString("a700")), builder.equal(2, BinaryString.fromString("b700"))); - assertThat(readF1(table, andWithUnindexedField)).isEmpty(); + assertThat(readF1(table, andWithUnindexedField)).containsExactly("a700"); + assertThat(readF1(tableWithSearchMode(table, "fast"), andWithUnindexedField)).isEmpty(); assertThat(readF1(tableWithSearchMode(table, "full"), andWithUnindexedField)) .containsExactly("a700"); assertThat(readF1(tableWithSearchMode(table, "detail"), andWithUnindexedField)) @@ -358,7 +360,8 @@ public void testBTreeGlobalIndexSearchModeUsesAllPredicateFieldCoverage() throws builder.equal(1, BinaryString.fromString("a700")), builder.equal(2, BinaryString.fromString("b700"))); - assertThat(readF1(table, andPredicate)).isEmpty(); + assertThat(readF1(table, andPredicate)).containsExactly("a700"); + assertThat(readF1(tableWithSearchMode(table, "fast"), andPredicate)).isEmpty(); assertThat(readF1(tableWithSearchMode(table, "full"), andPredicate)) .containsExactly("a700"); assertThat(readF1(tableWithSearchMode(table, "detail"), andPredicate)) @@ -369,7 +372,8 @@ public void testBTreeGlobalIndexSearchModeUsesAllPredicateFieldCoverage() throws builder.equal(1, BinaryString.fromString("a700")), builder.equal(2, BinaryString.fromString("b701"))); - assertThat(readF1(table, orPredicate)).containsExactly("a701"); + assertThat(readF1(table, orPredicate)).containsExactly("a700", "a701"); + assertThat(readF1(tableWithSearchMode(table, "fast"), orPredicate)).containsExactly("a701"); assertThat(readF1(tableWithSearchMode(table, "full"), orPredicate)) .containsExactly("a700", "a701"); assertThat(readF1(tableWithSearchMode(table, "detail"), orPredicate)) 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 12af712f1d36..16c418dc64c5 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 @@ -190,7 +190,7 @@ public void testFullTextSearchNonFastModesScanUnindexedData() throws Exception { (FileStoreTable) table.copy( Collections.singletonMap( - CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), + CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE.key(), searchMode)); GlobalIndexResult result = nonFastModeTable diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java index c00ecb18a27b..a791a605b279 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/PrimaryKeyFullTextReadTest.java @@ -100,7 +100,7 @@ void testRejectsIncompleteSearchModes(GlobalIndexSearchMode mode) { new PrimaryKeyFullTextRead( mode, 10, split -> Collections.emptyList())) .isInstanceOf(UnsupportedOperationException.class) - .hasMessageContaining("only supports the FAST global-index search mode"); + .hasMessageContaining("only supports the FAST full-text-index search mode"); } private static PrimaryKeySearchPosition position( diff --git a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java index 2153b44f2b3a..9f747807c03a 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/source/VectorSearchBuilderTest.java @@ -366,7 +366,7 @@ public void testFullModeRawOnlyUsesConfiguredMetric() throws Exception { catalog.createTable( identifier("full_search_raw_only_cosine_table"), vectorSchemaBuilder(VECTOR_FIELD_NAME) - .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), "full") + .option(CoreOptions.VECTOR_INDEX_SEARCH_MODE.key(), "full") .option("test.vector.metric", "cosine") .build(), false); @@ -1067,12 +1067,10 @@ public void testPreFilterMatchesZeroRows() throws Exception { @Test public void testPartialScalarPreFilterMustNotDropUnindexedScalarRows() throws Exception { catalog.createTable( - identifier("full_search_partial_scalar_unindexed_table"), - vectorSchemaBuilder(VECTOR_FIELD_NAME) - .option(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), "full") - .build(), + identifier("default_scalar_full_partial_index_table"), + vectorSchemaBuilder(VECTOR_FIELD_NAME).build(), false); - FileStoreTable table = getTable(identifier("full_search_partial_scalar_unindexed_table")); + FileStoreTable table = getTable(identifier("default_scalar_full_partial_index_table")); float[][] vectors = new float[10][]; for (int i = 0; i < vectors.length; i++) { @@ -1104,6 +1102,41 @@ public void testPartialScalarPreFilterMustNotDropUnindexedScalarRows() throws Ex assertThat(ids).containsExactly(8); } + @Test + public void testFastVectorModeLimitsScalarFallbackToVectorCoverage() throws Exception { + catalog.createTable( + identifier("fast_vector_partial_coverage_table"), + vectorSchemaBuilder(VECTOR_FIELD_NAME) + .option(CoreOptions.VECTOR_INDEX_SEARCH_MODE.key(), "fast") + .option(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "full") + .build(), + false); + FileStoreTable table = getTable(identifier("fast_vector_partial_coverage_table")); + + float[][] vectors = new float[10][]; + for (int i = 0; i < vectors.length; i++) { + vectors[i] = new float[] {Math.abs(i - 8), 0.0f}; + } + writeVectors(table, vectors); + + buildAndCommitVectorIndex(table, Arrays.copyOf(vectors, 5), new Range(0, 4)); + buildAndCommitBTreeIndex(table, new int[] {2, 3, 4}, new Range(2, 4)); + + Predicate idFilter = new PredicateBuilder(table.rowType()).greaterOrEqual(0, 5); + VectorScan.Plan plan = + table.newVectorSearchBuilder() + .withVector(new float[] {0.0f, 0.0f}) + .withLimit(1) + .withVectorColumn(VECTOR_FIELD_NAME) + .withFilter(idFilter) + .newVectorScan() + .scan(); + + assertThat(rawVectorSearchSplits(plan.splits())).hasSize(1); + assertThat(rawVectorSearchSplits(plan.splits()).get(0).rowRanges()) + .containsExactly(new Range(0, 1)); + } + @Test public void testFullModeFilterWithoutScalarIndexMustNotLetVectorIndexPolluteTopK() throws Exception { @@ -1144,9 +1177,14 @@ public void testFullModeFilterWithoutScalarIndexMustNotLetVectorIndexPolluteTopK } @Test - public void testFastModePartialScalarPreFilterOnlyUsesIndexedRows() throws Exception { - createTableDefault(); - FileStoreTable table = getTableDefault(); + public void testFastScalarModePartialPreFilterOnlyUsesIndexedRows() throws Exception { + catalog.createTable( + identifier("fast_scalar_partial_index_table"), + vectorSchemaBuilder(VECTOR_FIELD_NAME) + .option(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), "fast") + .build(), + false); + FileStoreTable table = getTable(identifier("fast_scalar_partial_index_table")); float[][] vectors = new float[10][]; for (int i = 0; i < vectors.length; i++) { diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java index 03ea04d144e5..fc0bd93697d7 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionDelete.java @@ -168,11 +168,12 @@ TableResult runInternal() { String query = String.format( "SELECT `_ROW_ID` FROM `%s`.`%s`.`%s$row_tracking` " - + "/*+ OPTIONS('scan.snapshot-id'='%d') */ WHERE %s", + + "/*+ OPTIONS('scan.snapshot-id'='%d', '%s'='full') */ WHERE %s", action.catalogName, action.identifier.getDatabaseName(), action.identifier.getObjectName(), baseSnapshotId, + CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), filter); LOG.info("Data-evolution delete source query: {}", query); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java index b3a57fe7329b..5c5cce749109 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoAction.java @@ -304,11 +304,13 @@ public Tuple2, RowType> buildSource() { // _ROW_ID is the first field of joined table. query = String.format( - "SELECT %s, %s FROM %s INNER JOIN %s AS RT ON %s", + "SELECT %s, %s FROM %s INNER JOIN %s " + + "/*+ OPTIONS('%s'='full') */ AS RT ON %s", "`RT`.`_ROW_ID` as `_ROW_ID`", String.join(",", project), escapedSourceName(), escapedRowTrackingTargetName(), + CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), rewriteMergeCondition(mergeCondition)); } diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java index 2deff25d13b9..bcc25cc1bf3a 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DataEvolutionMergeIntoActionITCase.java @@ -621,6 +621,29 @@ public void testUpdateAction() throws Exception { assertFalse(indexFileExists("T")); } + @Test + public void testMergeUnindexedRowWithFastSearchMode() throws Exception { + executeSQL( + "CALL sys.create_global_index(`table` => 'default.T', " + + "index_column => 'name', index_type => 'btree')", + false, + true); + insertInto("T", "(21, 'new', 2.1, '01-22')"); + executeSQL("ALTER TABLE T SET ('scalar-index.search-mode' = 'fast')", false, true); + + builder(warehouse, database, "T") + .withMergeCondition("T.name = 'new' AND S.id = 21") + .withMatchedUpdateSet("T.value = S.`value`") + .withSourceTable("S") + .withSinkParallelism(1) + .build() + .run(); + + testBatchRead( + "SELECT id, name, `value` FROM T WHERE id = 21", + Collections.singletonList(changelogRow("+I", 21, "new", 102.1))); + } + private boolean indexFileExists(String tableName) throws Exception { FileStoreTable table = getFileStoreTable(tableName); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java index 08d3e58b118d..f220f848089b 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/DeleteActionDataEvolutionITCase.java @@ -118,6 +118,25 @@ public void testNoMatchedRowsDoesNotCreateSnapshot() throws Exception { assertThat(deletionVectorCardinality(table)).isZero(); } + @Test + public void testDeleteUnindexedRowWithFastSearchMode() throws Exception { + createDataEvolutionTable(TABLE, false, true); + insertInto(TABLE, "(1, 'old', 'A')"); + executeSQL( + "CALL sys.create_global_index(`table` => 'default.T', " + + "index_column => 'name', index_type => 'btree')", + false, + true); + insertInto(TABLE, "(2, 'new', 'A')"); + executeSQL("ALTER TABLE T SET ('scalar-index.search-mode' = 'fast')", false, true); + + action(TABLE, "name = 'new'", 1).run(); + + testBatchRead( + "SELECT id, name, dt FROM T ORDER BY id", + Collections.singletonList(changelogRow("+I", 1, "old", "A"))); + } + @Test public void testDeleteWithBoundedExternalCandidates() throws Exception { createDataEvolutionTable(TABLE, false, true); diff --git a/paimon-full-text/README.md b/paimon-full-text/README.md index 2ac1ae0e9b34..2d17cccfb95d 100644 --- a/paimon-full-text/README.md +++ b/paimon-full-text/README.md @@ -119,7 +119,7 @@ Level-1-or-higher data level. Data compaction replaces the affected level archiv archive can cover multiple ordered source files; its row IDs concatenate their physical row positions. -Primary-key full-text search currently supports only `global-index.search-mode=fast`. Level-0 and +Primary-key full-text search currently supports only `full-text-index.search-mode=fast`. Level-0 and other uncovered files are not searched; their rows become searchable after compaction publishes an eligible data file and persistent archive. Search applies each source file's deletion vector, preserves native relevance scores, and selects a global Top-K. Only Hybrid search rewrites route diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 20ee9ddb8c25..392ea653a094 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -731,11 +731,29 @@ class CoreOptions: GLOBAL_INDEX_SEARCH_MODE: ConfigOption[GlobalIndexSearchMode] = ( ConfigOptions.key("global-index.search-mode") .enum_type(GlobalIndexSearchMode) + .no_default_value() + .with_description("Fallback search mode for global index queries.") + ) + + SCALAR_INDEX_SEARCH_MODE: ConfigOption[GlobalIndexSearchMode] = ( + ConfigOptions.key("scalar-index.search-mode") + .enum_type(GlobalIndexSearchMode) + .default_value(GlobalIndexSearchMode.FULL) + .with_description("Search mode for scalar index queries.") + ) + + VECTOR_INDEX_SEARCH_MODE: ConfigOption[GlobalIndexSearchMode] = ( + ConfigOptions.key("vector-index.search-mode") + .enum_type(GlobalIndexSearchMode) .default_value(GlobalIndexSearchMode.FAST) - .with_description( - "Search mode for global index queries. " - "Supported values are 'fast', 'full', and 'detail'." - ) + .with_description("Search mode for vector index queries.") + ) + + FULL_TEXT_INDEX_SEARCH_MODE: ConfigOption[GlobalIndexSearchMode] = ( + ConfigOptions.key("full-text-index.search-mode") + .enum_type(GlobalIndexSearchMode) + .default_value(GlobalIndexSearchMode.FAST) + .with_description("Search mode for full-text index queries.") ) GLOBAL_INDEX_EXTERNAL_PATH: ConfigOption[str] = ( @@ -1372,6 +1390,21 @@ def global_index_enabled(self, default=None): def global_index_search_mode(self): return self.options.get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE) + def scalar_index_search_mode(self): + return self._family_index_search_mode(CoreOptions.SCALAR_INDEX_SEARCH_MODE) + + def vector_index_search_mode(self): + return self._family_index_search_mode(CoreOptions.VECTOR_INDEX_SEARCH_MODE) + + def full_text_index_search_mode(self): + return self._family_index_search_mode(CoreOptions.FULL_TEXT_INDEX_SEARCH_MODE) + + def _family_index_search_mode(self, option): + if self.options.contains(option): + return self.options.get(option) + global_mode = self.global_index_search_mode() + return global_mode if global_mode is not None else self.options.get(option) + def global_index_external_path(self, default=None): value = self.options.get(CoreOptions.GLOBAL_INDEX_EXTERNAL_PATH, default) if value is None: diff --git a/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py b/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py index 07a9213bb830..19aa1c251fe2 100644 --- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py +++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_coverage.py @@ -55,6 +55,7 @@ def unindexed_ranges( self, fields_or_field_id: Union[List[DataField], Collection[int], int], predicate: Optional[Predicate] = None, + search_mode: Optional[GlobalIndexSearchMode] = None, ) -> List[Range]: if isinstance(fields_or_field_id, int): field_ids = {fields_or_field_id} @@ -67,7 +68,7 @@ def unindexed_ranges( field = field_by_name.get(name) if field is not None: field_ids.add(field.id) - return self._unindexed_ranges(field_ids) + return self._unindexed_ranges(field_ids, search_mode) def _add_coverage(self, field_id: int, row_range: Range) -> None: self._coverage_by_field.setdefault(field_id, []).append(row_range) @@ -84,8 +85,13 @@ def _indexed_ranges(self, field_ids: Collection[int]) -> List[Range]: return [] return Range.sort_and_merge_overlap(ranges, True) - def _unindexed_ranges(self, field_ids: Collection[int]) -> List[Range]: - search_mode = _global_index_search_mode(self._table) + def _unindexed_ranges( + self, + field_ids: Collection[int], + search_mode: Optional[GlobalIndexSearchMode] = None, + ) -> List[Range]: + if search_mode is None: + search_mode = _scalar_index_search_mode(self._table) if search_mode == GlobalIndexSearchMode.FAST: return [] next_row_id = getattr(self._snapshot, "next_row_id", None) @@ -138,13 +144,13 @@ def _entry_matches_partition(self, entry) -> bool: return self._partition_filter.test(entry.partition) -def _global_index_search_mode(table): +def _scalar_index_search_mode(table): options = getattr(table, "options", None) if options is None: - return GlobalIndexSearchMode.FAST - if hasattr(options, "global_index_search_mode"): - return options.global_index_search_mode() - return CoreOptions(Options.from_none()).global_index_search_mode() + return GlobalIndexSearchMode.FULL + if hasattr(options, "scalar_index_search_mode"): + return options.scalar_index_search_mode() + return CoreOptions(Options.from_none()).scalar_index_search_mode() def _is_field_id_collection(fields_or_field_id): diff --git a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py index 4ee6b94ced30..192c2e99bb8d 100644 --- a/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py +++ b/paimon-python/pypaimon/globalindex/data_evolution_global_index_scanner.py @@ -209,12 +209,14 @@ def scan(self, predicate: Optional[Predicate]) -> Optional[GlobalIndexResult]: """Scan the global index with the given predicate.""" return self._evaluator.evaluate(predicate) - def unindexed_rows(self, predicate: Optional[Predicate]) -> GlobalIndexResult: + def unindexed_rows(self, predicate: Optional[Predicate], + search_mode=None) -> GlobalIndexResult: """Return coarse row ids not covered by global indexes.""" if self._coverage is None: return GlobalIndexResult.create_empty() return GlobalIndexResult.from_ranges( - self._coverage.unindexed_ranges(self._fields, predicate)) + self._coverage.unindexed_ranges( + self._fields, predicate, search_mode=search_mode)) def close(self): """Close the scanner and release resources.""" diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py b/paimon-python/pypaimon/read/scanner/file_scanner.py index e952a9f3a4c8..c2fcff26a9c9 100755 --- a/paimon-python/pypaimon/read/scanner/file_scanner.py +++ b/paimon-python/pypaimon/read/scanner/file_scanner.py @@ -498,7 +498,9 @@ def _eval_global_index(self, snapshot=None): result = scanner.scan(self.predicate) if result is None: return None - return result.or_(scanner.unindexed_rows(self.predicate)) + scalar_mode = self.table.options.scalar_index_search_mode() + return result.or_( + scanner.unindexed_rows(self.predicate, search_mode=scalar_mode)) except Exception: return None diff --git a/paimon-python/pypaimon/table/source/full_text_scan.py b/paimon-python/pypaimon/table/source/full_text_scan.py index 5e0038a1841c..1387b02536bc 100644 --- a/paimon-python/pypaimon/table/source/full_text_scan.py +++ b/paimon-python/pypaimon/table/source/full_text_scan.py @@ -122,7 +122,10 @@ def index_file_filter(entry): snapshot, partition_filter, all_index_files, - ).unindexed_ranges(list(text_column_ids)) + ).unindexed_ranges( + list(text_column_ids), + search_mode=self._table.options.full_text_index_search_mode(), + ) if raw_row_ranges: splits.append(RawFullTextSearchSplit(raw_row_ranges)) diff --git a/paimon-python/pypaimon/table/source/full_text_search_builder.py b/paimon-python/pypaimon/table/source/full_text_search_builder.py index e98a74cef351..1e094e543f62 100644 --- a/paimon-python/pypaimon/table/source/full_text_search_builder.py +++ b/paimon-python/pypaimon/table/source/full_text_search_builder.py @@ -137,11 +137,11 @@ def new_full_text_read(self) -> FullTextRead: from pypaimon.common.options.core_options import GlobalIndexSearchMode mode = CoreOptions( Options(dict(self._table.table_schema.options)) - ).global_index_search_mode() + ).full_text_index_search_mode() if mode != GlobalIndexSearchMode.FAST: raise NotImplementedError( "Primary-key full-text search only supports the FAST " - "global-index search mode; FULL and DETAIL require " + "full-text index search mode; FULL and DETAIL require " "merge-aware logical-row fallback.") from pypaimon.table.source.primary_key_full_text_read import PrimaryKeyFullTextRead return PrimaryKeyFullTextRead( diff --git a/paimon-python/pypaimon/table/source/primary_key_full_text_read.py b/paimon-python/pypaimon/table/source/primary_key_full_text_read.py index 7b447c37a319..2386cdf2d2f2 100644 --- a/paimon-python/pypaimon/table/source/primary_key_full_text_read.py +++ b/paimon-python/pypaimon/table/source/primary_key_full_text_read.py @@ -40,11 +40,11 @@ def __init__(self, table, limit, text_column, query, definition, partition_filter=None): mode = CoreOptions( Options(dict(table.table_schema.options)) - ).global_index_search_mode() + ).full_text_index_search_mode() if mode != GlobalIndexSearchMode.FAST: raise NotImplementedError( "Primary-key full-text search only supports the FAST " - "global-index search mode; FULL and DETAIL require " + "full-text index search mode; FULL and DETAIL require " "merge-aware logical-row fallback.") self._definition = definition super().__init__(table, limit, text_column, query, partition_filter) diff --git a/paimon-python/pypaimon/table/source/primary_key_vector_read.py b/paimon-python/pypaimon/table/source/primary_key_vector_read.py index 8850840e3943..fd6a65e83ec4 100644 --- a/paimon-python/pypaimon/table/source/primary_key_vector_read.py +++ b/paimon-python/pypaimon/table/source/primary_key_vector_read.py @@ -18,6 +18,7 @@ from concurrent.futures import wait from heapq import nsmallest +from pypaimon.common.options.core_options import GlobalIndexSearchMode from pypaimon.index.pk.primary_key_index_source_meta import PrimaryKeyIndexSourceMeta from pypaimon.table.source.primary_key_scored_result import ( PrimaryKeyScoredResult, PrimaryKeySearchPosition, _partition_bytes) @@ -87,7 +88,10 @@ def indexed_candidate_iter(): plan, indexed_candidates, index_type) else: indexed_candidates = _top_k(indexed_candidates, self._limit) - exact_candidates = _top_k(self._raw_candidates(plan), self._limit) + exact_candidates = [] + if (self._table.options.vector_index_search_mode() + != GlobalIndexSearchMode.FAST): + exact_candidates = _top_k(self._raw_candidates(plan), self._limit) candidates = _top_k( (candidate for group in (indexed_candidates, exact_candidates) diff --git a/paimon-python/pypaimon/table/source/vector_search_read.py b/paimon-python/pypaimon/table/source/vector_search_read.py index eba11094f3fd..84b0ae7c0014 100644 --- a/paimon-python/pypaimon/table/source/vector_search_read.py +++ b/paimon-python/pypaimon/table/source/vector_search_read.py @@ -190,7 +190,10 @@ def _raw_pre_filter(self, splits): include = result.results() include = RoaringBitmap64.or_( include, - scanner.unindexed_rows(self._filter).results()) + scanner.unindexed_rows( + self._filter, + search_mode=self._table.options.scalar_index_search_mode(), + ).results()) return RoaringBitmap64.and_(include, raw_rows) finally: scanner.close() diff --git a/paimon-python/pypaimon/table/source/vector_search_scan.py b/paimon-python/pypaimon/table/source/vector_search_scan.py index e9f65a5b6147..7e002064dbd7 100644 --- a/paimon-python/pypaimon/table/source/vector_search_scan.py +++ b/paimon-python/pypaimon/table/source/vector_search_scan.py @@ -20,6 +20,7 @@ from abc import ABC, abstractmethod from collections import defaultdict +from pypaimon.common.options.core_options import GlobalIndexSearchMode from pypaimon.globalindex.data_evolution_global_index_coverage import DataEvolutionGlobalIndexCoverage from pypaimon.table.source.vector_search_split import ( IndexVectorSearchSplit, @@ -163,26 +164,40 @@ def index_file_filter(entry): ) ) + vector_search_mode = self._table.options.vector_index_search_mode() raw_row_ranges = DataEvolutionGlobalIndexCoverage( self._table, snapshot, partition_filter, vector_index_files, - ).unindexed_ranges(vector_column.id) + ).unindexed_ranges( + vector_column.id, + search_mode=vector_search_mode, + ) scalar_index_files = [ f for f in all_index_files if f.global_index_meta is not None and f.global_index_meta.index_field_id != vector_column.id ] if self._filter is not None: + scalar_unindexed_ranges = DataEvolutionGlobalIndexCoverage( + self._table, + snapshot, + partition_filter, + scalar_index_files, + ).unindexed_ranges( + self._table.fields, + self._filter, + search_mode=self._table.options.scalar_index_search_mode(), + ) + if vector_search_mode == GlobalIndexSearchMode.FAST: + scalar_unindexed_ranges = Range.and_( + scalar_unindexed_ranges, + Range.sort_and_merge_overlap( + list(vector_by_range.keys()), True), + ) raw_row_ranges = Range.sort_and_merge_overlap( - raw_row_ranges - + DataEvolutionGlobalIndexCoverage( - self._table, - snapshot, - partition_filter, - scalar_index_files, - ).unindexed_ranges(self._table.fields, self._filter), + raw_row_ranges + scalar_unindexed_ranges, True, ) if raw_row_ranges: diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py index bc48ef3ccc4c..80a3e7934c8d 100644 --- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py +++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py @@ -463,15 +463,15 @@ def test_read_btree_index_table(self): def test_read_btree_raw_fallback(self): table = self.catalog.get_table('default.test_btree_raw_fallback') - fast_builder = table.new_read_builder() + fast_table = table.copy({'scalar-index.search-mode': 'fast'}) + fast_builder = fast_table.new_read_builder() fast_predicate = fast_builder.new_predicate_builder().equal('k', 'k4') fast_builder.with_filter(fast_predicate) fast_result = fast_builder.new_read().to_arrow( fast_builder.new_scan().plan().splits()) self.assertEqual(0, fast_result.num_rows) - full_table = table.copy({'global-index.search-mode': 'full'}) - read_builder = full_table.new_read_builder() + read_builder = table.new_read_builder() read_builder.with_filter( read_builder.new_predicate_builder().equal('k', 'k4')) actual = read_builder.new_read().to_arrow( diff --git a/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py new file mode 100644 index 000000000000..b900c52ca0c1 --- /dev/null +++ b/paimon-python/pypaimon/tests/global_index_scalar_search_mode_test.py @@ -0,0 +1,109 @@ +# 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. + +import unittest +from types import SimpleNamespace + +from pypaimon.common.options.core_options import CoreOptions, GlobalIndexSearchMode +from pypaimon.common.options.options import Options +from pypaimon.globalindex.data_evolution_global_index_coverage import ( + DataEvolutionGlobalIndexCoverage, +) +from pypaimon.globalindex.data_evolution_global_index_scanner import ( + DataEvolutionGlobalIndexScanner, +) + + +def _ranges(result): + inner = result.results() if hasattr(result, "results") else result + for attr in ("to_range_list", "to_ranges", "ranges"): + if hasattr(inner, attr): + value = getattr(inner, attr) + value = value() if callable(value) else value + return [(r.from_, r.to) for r in value] + raise AssertionError("cannot extract ranges from %r" % (result,)) + + +def _coverage(options): + meta = SimpleNamespace( + row_range_start=0, row_range_end=99, index_field_id=1, extra_field_ids=None) + snapshot = SimpleNamespace(next_row_id=200) + table = SimpleNamespace(options=options) + return DataEvolutionGlobalIndexCoverage( + table, snapshot, None, [SimpleNamespace(global_index_meta=meta)]) + + +class ScalarGlobalIndexSearchModeTest(unittest.TestCase): + + def test_default_values(self): + options = CoreOptions(Options.from_none()) + self.assertIsNone(options.global_index_search_mode()) + self.assertEqual( + GlobalIndexSearchMode.FULL, options.scalar_index_search_mode()) + self.assertEqual( + GlobalIndexSearchMode.FAST, options.vector_index_search_mode()) + self.assertEqual( + GlobalIndexSearchMode.FAST, options.full_text_index_search_mode()) + + def test_legacy_global_mode_is_family_fallback(self): + options = CoreOptions(Options({"global-index.search-mode": "detail"})) + for getter in ( + options.scalar_index_search_mode, + options.vector_index_search_mode, + options.full_text_index_search_mode): + self.assertEqual(GlobalIndexSearchMode.DETAIL, getter()) + + def test_family_modes_override_global_mode(self): + options = CoreOptions(Options({ + "global-index.search-mode": "detail", + "scalar-index.search-mode": "fast", + "vector-index.search-mode": "full", + "full-text-index.search-mode": "fast", + })) + self.assertEqual( + GlobalIndexSearchMode.FAST, options.scalar_index_search_mode()) + self.assertEqual( + GlobalIndexSearchMode.FULL, options.vector_index_search_mode()) + self.assertEqual( + GlobalIndexSearchMode.FAST, options.full_text_index_search_mode()) + + def test_coverage_honours_search_mode_override(self): + coverage = _coverage(CoreOptions(Options.from_none())) + self.assertEqual( + [], coverage.unindexed_ranges(1, search_mode=GlobalIndexSearchMode.FAST)) + full = coverage.unindexed_ranges(1, search_mode=GlobalIndexSearchMode.FULL) + self.assertEqual([(100, 199)], [(r.from_, r.to) for r in full]) + self.assertEqual( + [(100, 199)], + [(r.from_, r.to) for r in coverage.unindexed_ranges(1)]) + + def test_scanner_applies_passed_scalar_mode(self): + coverage = _coverage(CoreOptions(Options.from_none())) + scanner = SimpleNamespace(_coverage=coverage, _fields=[1]) + result = DataEvolutionGlobalIndexScanner.unindexed_rows( + scanner, None, search_mode=GlobalIndexSearchMode.FULL) + self.assertEqual([(100, 199)], _ranges(result)) + + def test_scanner_default_is_scalar_mode(self): + coverage = _coverage(CoreOptions(Options.from_none())) + scanner = SimpleNamespace(_coverage=coverage, _fields=[1]) + result = DataEvolutionGlobalIndexScanner.unindexed_rows(scanner, None) + self.assertEqual([(100, 199)], _ranges(result)) + + +if __name__ == "__main__": + unittest.main() diff --git a/paimon-python/pypaimon/tests/global_index_test.py b/paimon-python/pypaimon/tests/global_index_test.py index 4d7a1b142bc2..627da08af0bc 100644 --- a/paimon-python/pypaimon/tests/global_index_test.py +++ b/paimon-python/pypaimon/tests/global_index_test.py @@ -80,8 +80,8 @@ class _CoverageOptions: def __init__(self, mode): self.options = Options({"global-index.search-mode": mode}) - def global_index_search_mode(self): - return CoreOptions(self.options).global_index_search_mode() + def scalar_index_search_mode(self): + return CoreOptions(self.options).scalar_index_search_mode() class _CoverageTable: @@ -223,6 +223,9 @@ class _Options: def global_index_enabled(self): return True + def scalar_index_search_mode(self): + return GlobalIndexSearchMode.FULL + class _Table: options = _Options() @@ -249,6 +252,8 @@ class _Table: [Range(1, 1), Range(5, 6)], result.results().to_range_list(), ) + fake_scanner.unindexed_rows.assert_called_once_with( + predicate, search_mode=GlobalIndexSearchMode.FULL) def test_eval_global_index_keeps_none_as_full_scan(self): from pypaimon.read.scanner.file_scanner import FileScanner diff --git a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py index bf1e56009752..5a599f2699e1 100644 --- a/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py +++ b/paimon-python/pypaimon/tests/primary_key_index_definitions_test.py @@ -18,6 +18,7 @@ from unittest import mock from types import SimpleNamespace +from pypaimon.common.options.core_options import GlobalIndexSearchMode from pypaimon.index.pk.primary_key_index_definitions import PrimaryKeyIndexDefinitions from pypaimon.index.pk.primary_key_index_source_file import PrimaryKeyIndexSourceFile from pypaimon.index.pk.primary_key_index_source_meta import PrimaryKeyIndexSourceMeta @@ -129,6 +130,22 @@ def builder(mode): self.assertIsInstance( builder("fast").new_full_text_read(), PrimaryKeyFullTextRead) + schema = TableSchema( + fields=fields, + options={ + "pk-full-text.index.columns": "content", + "global-index.search-mode": "full", + "full-text-index.search-mode": "fast", + }) + table = SimpleNamespace( + fields=fields, table_schema=schema, partition_keys=[]) + self.assertIsInstance( + FullTextSearchBuilderImpl(table) + .with_query("content", "paimon") + .with_limit(2) + .new_full_text_read(), + PrimaryKeyFullTextRead, + ) for mode in ("full", "detail"): with self.subTest(mode=mode), self.assertRaisesRegex( NotImplementedError, "only supports the FAST"): @@ -194,7 +211,8 @@ def test_empty_pk_vector_ranges_mean_no_candidates(self): def test_pk_vector_refine_factor_overflow_is_rejected_before_search(self): table = SimpleNamespace(options=SimpleNamespace( - primary_key_vector_index_type=lambda column: "ivf-pq")) + primary_key_vector_index_type=lambda column: "ivf-pq", + vector_index_search_mode=lambda: GlobalIndexSearchMode.FAST)) reader = PrimaryKeyVectorRead( table, 2147483647, DataField(1, "embedding", AtomicType("VECTOR(1)")), [0.0], options={"ivf.refine_factor": "2"}) @@ -205,6 +223,25 @@ def test_pk_vector_refine_factor_overflow_is_rejected_before_search(self): with self.assertRaisesRegex(ValueError, "limit overflow"): reader.read_plan(plan) + def test_pk_vector_raw_fallback_uses_vector_search_mode(self): + from pypaimon.table.source.primary_key_vector_scan import ( + PrimaryKeyVectorScanPlan) + + field = DataField(1, "embedding", AtomicType("VECTOR(1)")) + plan = PrimaryKeyVectorScanPlan(1, []) + for mode, expected_calls in ( + (GlobalIndexSearchMode.FAST, 0), + (GlobalIndexSearchMode.FULL, 1), + (GlobalIndexSearchMode.DETAIL, 1)): + table = SimpleNamespace(options=SimpleNamespace( + primary_key_vector_index_type=lambda column: "ivf-pq", + vector_index_search_mode=lambda mode=mode: mode)) + reader = PrimaryKeyVectorRead(table, 1, field, [0.0]) + with self.subTest(mode=mode), mock.patch.object( + reader, "_raw_candidates", return_value=[]) as raw_candidates: + reader.read_plan(plan) + self.assertEqual(expected_calls, raw_candidates.call_count) + def test_full_text_payload_planner_rejects_duplicate_and_stale_levels(self): files = [SimpleNamespace(file_name="a.parquet", row_count=2, level=1)] source_meta = PrimaryKeyIndexSourceMeta( diff --git a/paimon-python/pypaimon/tests/table_update_test.py b/paimon-python/pypaimon/tests/table_update_test.py index afbb18b469db..40ceaf2489f6 100644 --- a/paimon-python/pypaimon/tests/table_update_test.py +++ b/paimon-python/pypaimon/tests/table_update_test.py @@ -94,13 +94,14 @@ def _create_seeded_table(self, partition_keys=None, options=None): }, schema=self.pa_schema)) return table - def _create_global_indexed_table_for_predicate_update(self): + def _create_global_indexed_table_for_predicate_update(self, extra_options=None): options = dict(self.table_options) options.update({ 'global-index.enabled': 'true', 'bucket': '-1', 'file.format': 'parquet', }) + options.update(extra_options or {}) table = self._create_table(options=options) self._write_arrow(table, pa.Table.from_pydict({ 'id': [1, 2], @@ -329,7 +330,9 @@ def test_update_by_predicate_rejects_uncastable_assignment(self): ) def test_update_by_predicate_with_global_index_updates_unindexed_rows(self): - table = self._create_global_indexed_table_for_predicate_update() + table = self._create_global_indexed_table_for_predicate_update({ + 'scalar-index.search-mode': 'fast', + }) pb = table.new_read_builder().new_predicate_builder() self._do_update_by_predicate( @@ -345,6 +348,18 @@ def test_update_by_predicate_with_global_index_updates_unindexed_rows(self): )) self.assertEqual({1: 10, 2: 15, 3: 21, 4: 30}, ages_by_id) + def test_delete_by_predicate_with_global_index_deletes_unindexed_rows(self): + table = self._create_global_indexed_table_for_predicate_update({ + 'deletion-vectors.enabled': 'true', + 'scalar-index.search-mode': 'fast', + }) + + pb = table.new_read_builder().new_predicate_builder() + self._do_delete_by_predicate(table, pb.equal('name', 'new')) + + result = self._read_all(table).sort_by('id') + self.assertEqual([1, 2, 4], result['id'].to_pylist()) + def test_update_by_predicate_with_global_index_falls_back_to_full_scan(self): table = self._create_global_indexed_table_for_predicate_update() diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py b/paimon-python/pypaimon/tests/vector_search_filter_test.py index 55fe9bb07bc4..d3c75613c502 100644 --- a/paimon-python/pypaimon/tests/vector_search_filter_test.py +++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py @@ -30,6 +30,8 @@ from typing import List from unittest import mock +from pypaimon.common.options.core_options import CoreOptions, GlobalIndexSearchMode +from pypaimon.common.options.options import Options from pypaimon.common.predicate import Predicate from pypaimon.common.predicate_builder import PredicateBuilder from pypaimon.globalindex.btree.btree_index_meta import BTreeIndexMeta @@ -95,6 +97,7 @@ def __init__(self, fields, entries, partition_fields=None): self.partition_keys: List[str] = [ f.name for f in self.partition_keys_fields] self.table_schema = _StubSchema() + self.options = CoreOptions(Options.from_none()) self.file_io = object() self._entries = entries @@ -805,26 +808,18 @@ def test_full_text_scan_ignores_other_index_types_on_same_column(self): [f.file_name for f in splits[0].full_text_index_files]) def test_full_text_scan_adds_raw_split_for_uncovered_ranges(self): - from pypaimon.common.options.core_options import CoreOptions - from pypaimon.common.options.options import Options from pypaimon.table.source.full_text_scan import DataEvolutionFullTextScan from pypaimon.table.source.full_text_search_split import ( IndexFullTextSearchSplit, RawFullTextSearchSplit, ) - class _Options: - options = Options({"global-index.search-mode": "full"}) - - def global_index_search_mode(self_inner): - return CoreOptions(self_inner.options).global_index_search_mode() - text_field = _field(1, "content", "STRING") entry = _entry( None, field_id=1, index_type="full-text", file_name="ft.index", row_range_start=0, row_range_end=4) table = _StubTable(fields=[text_field], entries=[entry]) - table.options = _Options() + table.options = CoreOptions(Options({"global-index.search-mode": "full"})) _patch_snapshot(self, [entry], types.SimpleNamespace(next_row_id=10)) splits = DataEvolutionFullTextScan(table, [text_field]).scan().splits() @@ -835,6 +830,26 @@ def global_index_search_mode(self_inner): self.assertEqual(1, len(raw)) self.assertEqual([Range(5, 9)], raw[0].row_ranges) + def test_full_text_mode_overrides_global_mode_for_uncovered_ranges(self): + from pypaimon.table.source.full_text_scan import DataEvolutionFullTextScan + from pypaimon.table.source.full_text_search_split import RawFullTextSearchSplit + + text_field = _field(1, "content", "STRING") + entry = _entry( + None, field_id=1, index_type="full-text", + file_name="ft.index", row_range_start=0, row_range_end=4) + table = _StubTable(fields=[text_field], entries=[entry]) + table.options = CoreOptions(Options({ + "global-index.search-mode": "full", + "full-text-index.search-mode": "fast", + })) + _patch_snapshot(self, [entry], types.SimpleNamespace(next_row_id=10)) + + splits = DataEvolutionFullTextScan(table, [text_field]).scan().splits() + + self.assertFalse(any(isinstance(split, RawFullTextSearchSplit) + for split in splits)) + class VectorSearchFilterTest(unittest.TestCase): """Non-partitioned wiring: scan + read + external_path plumbing.""" @@ -1224,6 +1239,9 @@ def test_refine_factor_query_options_override_table_options(self): class _Options: options = Options({"ivf.refine_factor": "2"}) + def vector_index_search_mode(self_inner): + return CoreOptions(self_inner.options).vector_index_search_mode() + entry = _entry(None, field_id=1, index_type="ivf-pq", file_name="vec.index", row_range_start=0, row_range_end=9) @@ -1317,26 +1335,18 @@ def close(self_inner): captured_io_metas[0][0].external_path) def test_full_mode_scan_adds_raw_split_for_unindexed_vector_rows(self): - from pypaimon.common.options.core_options import CoreOptions - from pypaimon.common.options.options import Options from pypaimon.table.source.vector_search_split import ( IndexVectorSearchSplit, RawVectorSearchSplit, ) - class _Options: - options = Options({"global-index.search-mode": "full"}) - - def global_index_search_mode(self_inner): - return CoreOptions(self_inner.options).global_index_search_mode() - class _Snapshots: def get_latest_snapshot(self_inner): return types.SimpleNamespace(next_row_id=10) table = _StubTable(fields=[self.id_field, self.embedding_field], entries=[self.entries[0]]) - table.options = _Options() + table.options = CoreOptions(Options({"global-index.search-mode": "full"})) table.snapshot_manager = lambda: _Snapshots() self._scan_patch.stop() self._travel_patch.stop() @@ -1360,16 +1370,77 @@ def get_latest_snapshot(self_inner): self.assertEqual(1, len(raw)) self.assertEqual([Range(5, 9)], raw[0].row_ranges) - def test_full_mode_scan_adds_raw_split_for_uncovered_scalar_filter(self): - from pypaimon.common.options.core_options import CoreOptions - from pypaimon.common.options.options import Options + def test_vector_mode_overrides_global_mode_for_unindexed_vector_rows(self): from pypaimon.table.source.vector_search_split import RawVectorSearchSplit - class _Options: - options = Options({"global-index.search-mode": "full"}) + class _Snapshots: + def get_latest_snapshot(self_inner): + return types.SimpleNamespace(next_row_id=10) + + table = _StubTable(fields=[self.id_field, self.embedding_field], + entries=[self.entries[0]]) + table.options = CoreOptions(Options({ + "global-index.search-mode": "full", + "vector-index.search-mode": "fast", + })) + table.snapshot_manager = lambda: _Snapshots() + self._scan_patch.stop() + self._travel_patch.stop() + _patch_snapshot(self, [self.entries[0]]) + self._travel_patch.stop() + + splits = ( + VectorSearchBuilderImpl(table) + .with_vector_column("embedding") + .with_query_vector([1.0, 0.0, 0.0, 0.0]) + .with_limit(3) + .new_vector_search_scan() + .scan() + .splits() + ) + + self.assertFalse(any(isinstance(split, RawVectorSearchSplit) + for split in splits)) + + def test_fast_vector_mode_limits_scalar_fallback_to_vector_coverage(self): + from pypaimon.table.source.vector_search_split import RawVectorSearchSplit - def global_index_search_mode(self_inner): - return CoreOptions(self_inner.options).global_index_search_mode() + entries = [ + _entry(None, field_id=1, index_type="lumina-vector-ann", + file_name="vec.index", row_range_start=0, + row_range_end=4), + _entry(None, field_id=0, index_type="btree", + file_name="id.index", row_range_start=2, + row_range_end=4), + ] + table = _StubTable(fields=[self.id_field, self.embedding_field], + entries=entries) + table.options = CoreOptions(Options({ + "vector-index.search-mode": "fast", + "scalar-index.search-mode": "full", + })) + _patch_snapshot(self, entries, types.SimpleNamespace(next_row_id=10)) + filter_pred = Predicate(method="greaterOrEqual", index=0, field="id", + literals=[5]) + + splits = ( + VectorSearchBuilderImpl(table) + .with_vector_column("embedding") + .with_query_vector([1.0, 0.0, 0.0, 0.0]) + .with_limit(3) + .with_filter(filter_pred) + .new_vector_search_scan() + .scan() + .splits() + ) + + raw = [split for split in splits + if isinstance(split, RawVectorSearchSplit)] + self.assertEqual(1, len(raw)) + self.assertEqual([Range(0, 1)], raw[0].row_ranges) + + def test_full_mode_scan_adds_raw_split_for_uncovered_scalar_filter(self): + from pypaimon.table.source.vector_search_split import RawVectorSearchSplit class _Snapshots: def get_latest_snapshot(self_inner): @@ -1387,7 +1458,10 @@ def get_latest_snapshot(self_inner): row_range_end=9) ], ) - table.options = _Options() + table.options = CoreOptions(Options({ + "scalar-index.search-mode": "full", + "vector-index.search-mode": "fast", + })) table.snapshot_manager = lambda: _Snapshots() self._scan_patch.stop() self._travel_patch.stop() @@ -1411,16 +1485,40 @@ def get_latest_snapshot(self_inner): self.assertEqual(1, len(raw)) self.assertEqual([Range(0, 9)], raw[0].row_ranges) - def test_scan_threads_builder_options_to_raw_split_index_type(self): - from pypaimon.common.options.core_options import CoreOptions - from pypaimon.common.options.options import Options + def test_raw_vector_pre_filter_uses_scalar_search_mode(self): + from pypaimon.table.source.vector_search_read import DataEvolutionVectorRead from pypaimon.table.source.vector_search_split import RawVectorSearchSplit - class _Options: - options = Options({"global-index.search-mode": "full"}) + predicate = Predicate(method="equal", index=0, field="id", literals=[5]) + scalar_file = self.entries[2].index_file + table = _StubTable(fields=[self.id_field, self.embedding_field], entries=[]) + table.options = CoreOptions(Options({ + "global-index.search-mode": "fast", + "scalar-index.search-mode": "detail", + })) + scanner = mock.MagicMock() + scanner.scan.return_value = GlobalIndexResult.create_empty() + scanner.unindexed_rows.return_value = GlobalIndexResult.create_empty() + reader = DataEvolutionVectorRead( + table, + limit=3, + vector_column=self.embedding_field, + query_vector=[1.0, 0.0, 0.0, 0.0], + filter_=predicate, + ) - def global_index_search_mode(self_inner): - return CoreOptions(self_inner.options).global_index_search_mode() + with mock.patch( + "pypaimon.globalindex.data_evolution_global_index_scanner." + "DataEvolutionGlobalIndexScanner.create", + return_value=scanner): + reader._raw_pre_filter([ + RawVectorSearchSplit([Range(0, 9)], [scalar_file])]) + + scanner.unindexed_rows.assert_called_once_with( + predicate, search_mode=GlobalIndexSearchMode.DETAIL) + + def test_scan_threads_builder_options_to_raw_split_index_type(self): + from pypaimon.table.source.vector_search_split import RawVectorSearchSplit class _Snapshots: def get_latest_snapshot(self_inner): @@ -1428,7 +1526,7 @@ def get_latest_snapshot(self_inner): table = _StubTable(fields=[self.id_field, self.embedding_field], entries=[]) - table.options = _Options() + table.options = CoreOptions(Options({"vector-index.search-mode": "full"})) table.snapshot_manager = lambda: _Snapshots() self._scan_patch.stop() self._travel_patch.stop() @@ -1865,21 +1963,10 @@ def close(self_inner): self.assertEqual([3], sorted(list(result.results()))) def test_scanner_reports_unindexed_rows_for_full_mode(self): - from pypaimon.common.options.core_options import CoreOptions - from pypaimon.common.options.options import Options from pypaimon.globalindex.data_evolution_global_index_scanner import ( DataEvolutionGlobalIndexScanner, ) - class _Options: - options = Options({"global-index.search-mode": "full"}) - - def global_index_search_mode(self_inner): - return CoreOptions(self_inner.options).global_index_search_mode() - - def global_index_thread_num(self_inner): - return 32 - class _Snapshots: def get_latest_snapshot(self_inner): return types.SimpleNamespace(next_row_id=10) @@ -1890,7 +1977,7 @@ def get_latest_snapshot(self_inner): file_name="id-0.index", row_range_start=0, row_range_end=4).index_file table = _StubTable(fields=[id_field, emb_field], entries=[]) - table.options = _Options() + table.options = CoreOptions(Options({"scalar-index.search-mode": "full"})) table.snapshot_manager = lambda: _Snapshots() scanner = DataEvolutionGlobalIndexScanner.create(table, index_files=[indexed]) diff --git a/paimon-python/pypaimon/write/table_update.py b/paimon-python/pypaimon/write/table_update.py index a08418591efd..8095c16c2ac4 100644 --- a/paimon-python/pypaimon/write/table_update.py +++ b/paimon-python/pypaimon/write/table_update.py @@ -273,7 +273,7 @@ def _matched_update_scan_table(self): return self.table dynamic_options = { - CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(): + CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(): GlobalIndexSearchMode.FULL.value, CoreOptions.SCAN_MODE.key(): StartupMode.DEFAULT.value, CoreOptions.SCAN_SNAPSHOT_ID.key(): str(snapshot.id), diff --git a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala index 5690a65d5fc7..79e09bc38029 100644 --- a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala +++ b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala @@ -1047,14 +1047,13 @@ object MergeIntoPaimonDataEvolutionTable { val snapshotId = snapshot.id().toString if ( configuredSnapshotId.contains(snapshotId) && - fullSearchMode.equalsIgnoreCase( - table.options().get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key())) + table.coreOptions().scalarIndexSearchMode() == CoreOptions.GlobalIndexSearchMode.FULL ) { (v2Table, relation) } else { val dynamicOptions = new JHashMap[String, String]() timeTravelOptionKeys.foreach(dynamicOptions.put(_, null)) - dynamicOptions.put(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), fullSearchMode) + dynamicOptions.put(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), fullSearchMode) dynamicOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId) val scanTable = SparkTable.of(table.copy(dynamicOptions)) diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala index f287d31959d0..479515ca4669 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala @@ -1046,14 +1046,13 @@ object MergeIntoPaimonDataEvolutionTable { val snapshotId = snapshot.id().toString if ( configuredSnapshotId.contains(snapshotId) && - fullSearchMode.equalsIgnoreCase( - table.options().get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key())) + table.coreOptions().scalarIndexSearchMode() == CoreOptions.GlobalIndexSearchMode.FULL ) { (v2Table, relation) } else { val dynamicOptions = new JHashMap[String, String]() timeTravelOptionKeys.foreach(dynamicOptions.put(_, null)) - dynamicOptions.put(CoreOptions.GLOBAL_INDEX_SEARCH_MODE.key(), fullSearchMode) + dynamicOptions.put(CoreOptions.SCALAR_INDEX_SEARCH_MODE.key(), fullSearchMode) dynamicOptions.put(CoreOptions.SCAN_SNAPSHOT_ID.key(), snapshotId) val scanTable = SparkTable.of(table.copy(dynamicOptions)) diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala index 72258ddbec5d..f88993dba846 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala @@ -1035,6 +1035,7 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase with AdaptiveSpar |CREATE TABLE t (id INT, name STRING, b INT) TBLPROPERTIES ( | 'row-tracking.enabled' = 'true', | 'data-evolution.enabled' = 'true', + | 'scalar-index.search-mode' = 'fast', | 'btree-index.records-per-range' = '1000') |""".stripMargin) sql("INSERT INTO t VALUES (1, 'old', 10)") @@ -1062,6 +1063,7 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase with AdaptiveSpar |CREATE TABLE t (id INT, name STRING, b INT) TBLPROPERTIES ( | 'row-tracking.enabled' = 'true', | 'data-evolution.enabled' = 'true', + | 'scalar-index.search-mode' = 'fast', | 'btree-index.records-per-range' = '1000') |""".stripMargin) sql("INSERT INTO t VALUES (1, 'old', 10)")