This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 3f84bbdc7c [core][python] Introduce index-specific search mode
defaults (#8844)
3f84bbdc7c is described below
commit 3f84bbdc7caa7319696abfd1516ef7e6a7fa232b
Author: XiaoHongbo <[email protected]>
AuthorDate: Sun Jul 26 15:14:07 2026 +0800
[core][python] Introduce index-specific search mode defaults (#8844)
---
docs/docs/multimodal-table/global-index.mdx | 20 +-
.../multimodal-table/global-index/full-text.mdx | 4 +-
docs/docs/primary-key-table/global-index.mdx | 14 +-
docs/docs/pypaimon/data-evolution.md | 2 +-
docs/generated/core_configuration.html | 22 +-
.../main/java/org/apache/paimon/CoreOptions.java | 44 ++-
.../paimon/globalindex/DataEvolutionBatchScan.java | 2 +-
.../DataEvolutionGlobalIndexCoverage.java | 17 +-
.../DataEvolutionGlobalIndexScanner.java | 7 +-
.../table/source/DataEvolutionFullTextScan.java | 6 +-
.../table/source/DataEvolutionVectorScan.java | 34 ++-
.../table/source/PrimaryKeyFullTextRead.java | 4 +-
.../paimon/table/source/PrimaryKeyVectorRead.java | 2 +-
.../java/org/apache/paimon/CoreOptionsTest.java | 33 +++
.../paimon/table/BtreeGlobalIndexTableTest.java | 14 +-
.../table/source/FullTextSearchBuilderTest.java | 2 +-
.../table/source/PrimaryKeyFullTextReadTest.java | 2 +-
.../table/source/VectorSearchBuilderTest.java | 56 +++-
.../paimon/flink/action/DataEvolutionDelete.java | 3 +-
.../flink/action/DataEvolutionMergeIntoAction.java | 4 +-
.../action/DataEvolutionMergeIntoActionITCase.java | 23 ++
.../action/DeleteActionDataEvolutionITCase.java | 19 ++
paimon-full-text/README.md | 2 +-
.../pypaimon/common/options/core_options.py | 41 ++-
.../data_evolution_global_index_coverage.py | 22 +-
.../data_evolution_global_index_scanner.py | 6 +-
.../pypaimon/read/scanner/file_scanner.py | 4 +-
.../pypaimon/table/source/full_text_scan.py | 5 +-
.../table/source/full_text_search_builder.py | 4 +-
.../table/source/primary_key_full_text_read.py | 4 +-
.../table/source/primary_key_vector_read.py | 6 +-
.../pypaimon/table/source/vector_search_builder.py | 70 ++---
.../pypaimon/table/source/vector_search_read.py | 5 +-
.../pypaimon/table/source/vector_search_scan.py | 31 +-
.../pypaimon/tests/e2e/java_py_read_write_test.py | 6 +-
.../tests/global_index_scalar_search_mode_test.py | 109 +++++++
paimon-python/pypaimon/tests/global_index_test.py | 9 +-
.../tests/primary_key_index_definitions_test.py | 39 ++-
paimon-python/pypaimon/tests/table_update_test.py | 19 +-
.../pypaimon/tests/vector_search_filter_test.py | 322 ++++++++++++++++++---
paimon-python/pypaimon/write/table_update.py | 2 +-
.../MergeIntoPaimonDataEvolutionTable.scala | 5 +-
.../MergeIntoPaimonDataEvolutionTable.scala | 5 +-
.../paimon/spark/sql/RowTrackingTestBase.scala | 2 +
44 files changed, 870 insertions(+), 182 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index.mdx
b/docs/docs/multimodal-table/global-index.mdx
index ace4770ba7..731fb8c078 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:
<TabItem value="sql" label="SQL">
```sql
-ALTER TABLE my_table SET ('global-index.search-mode' = 'full');
+ALTER TABLE my_table SET ('vector-index.search-mode' = 'full');
```
</TabItem>
@@ -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"})
```
</TabItem>
@@ -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:
<Tabs groupId="global-index-enabled">
@@ -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 ae8ea59d61..5886ef2d7f 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)
</Tabs>
-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 3098d71c52..e76845e6c3 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.<column>.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.<column>.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 f5d58d75e4..43f31c63b8 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 d6f44e9f33..bd739d57a4 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -758,6 +758,12 @@ under the License.
<td>Integer</td>
<td>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.</td>
</tr>
+ <tr>
+ <td><h5>full-text-index.search-mode</h5></td>
+ <td style="word-wrap: break-word;">fast</td>
+ <td><p>Enum</p></td>
+ <td>Search mode for full-text index queries.<br /><br />Possible
values:<ul><li>"fast": Only search indexed data.</li><li>"full": Use snapshot
next row id and global index coverage to detect missing row ids, and scan raw
data only when a gap exists.</li><li>"detail": Scan data files to find exact
unindexed rows. This can handle index invalidation caused by updates or
rewrites.</li></ul></td>
+ </tr>
<tr>
<td><h5>global-index.build.max-parallelism</h5></td>
<td style="word-wrap: break-word;">4096</td>
@@ -796,9 +802,9 @@ under the License.
</tr>
<tr>
<td><h5>global-index.search-mode</h5></td>
- <td style="word-wrap: break-word;">fast</td>
+ <td style="word-wrap: break-word;">(none)</td>
<td><p>Enum</p></td>
- <td>Search mode for global index queries. Supported values are
'fast', 'full', and 'detail'.<br /><br />Possible values:<ul><li>"fast": Only
search indexed data.</li><li>"full": Use snapshot next row id and global index
coverage to detect missing row ids, and scan raw data only when a gap
exists.</li><li>"detail": Scan data files to find exact unindexed rows. This
can handle index invalidation caused by updates or rewrites.</li></ul></td>
+ <td>Fallback search mode for global index queries.<br /><br
/>Possible values:<ul><li>"fast": Only search indexed data.</li><li>"full": Use
snapshot next row id and global index coverage to detect missing row ids, and
scan raw data only when a gap exists.</li><li>"detail": Scan data files to find
exact unindexed rows. This can handle index invalidation caused by updates or
rewrites.</li></ul></td>
</tr>
<tr>
<td><h5>global-index.thread-num</h5></td>
@@ -1326,6 +1332,12 @@ For an internal format table in a REST catalog, it also
makes the catalog own th
<td>String</td>
<td>The field that generates the row kind for primary key table,
the row kind determines which data is '+I', '-U', '+U' or '-D'.</td>
</tr>
+ <tr>
+ <td><h5>scalar-index.search-mode</h5></td>
+ <td style="word-wrap: break-word;">full</td>
+ <td><p>Enum</p></td>
+ <td>Search mode for scalar index queries.<br /><br />Possible
values:<ul><li>"fast": Only search indexed data.</li><li>"full": Use snapshot
next row id and global index coverage to detect missing row ids, and scan raw
data only when a gap exists.</li><li>"detail": Scan data files to find exact
unindexed rows. This can handle index invalidation caused by updates or
rewrites.</li></ul></td>
+ </tr>
<tr>
<td><h5>scan.bounded.watermark</h5></td>
<td style="word-wrap: break-word;">(none)</td>
@@ -1753,6 +1765,12 @@ If the data size allocated for the sorting task is
uneven,which may lead to perf
<td>String</td>
<td>Specifies column names that should be stored as vector type.
This is used when you want to treat a ARRAY column as a VECTOR.</td>
</tr>
+ <tr>
+ <td><h5>vector-index.search-mode</h5></td>
+ <td style="word-wrap: break-word;">fast</td>
+ <td><p>Enum</p></td>
+ <td>Search mode for vector index queries.<br /><br />Possible
values:<ul><li>"fast": Only search indexed data.</li><li>"full": Use snapshot
next row id and global index coverage to detect missing row ids, and scan raw
data only when a gap exists.</li><li>"detail": Scan data files to find exact
unindexed rows. This can handle index invalidation caused by updates or
rewrites.</li></ul></td>
+ </tr>
<tr>
<td><h5>vector-search.distribute.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
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 c5989ff9b3..a57810ae45 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2713,11 +2713,27 @@ public class CoreOptions implements Serializable {
public static final ConfigOption<GlobalIndexSearchMode>
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<GlobalIndexSearchMode>
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<GlobalIndexSearchMode>
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<GlobalIndexSearchMode>
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<Integer> GLOBAL_INDEX_THREAD_NUM =
key("global-index.thread-num")
@@ -4324,10 +4340,32 @@ public class CoreOptions implements Serializable {
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<GlobalIndexSearchMode> 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 aacae5ce68..6dbbca8e77 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 @@ public class DataEvolutionBatchScan implements
DataTableScan {
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 55678824e6..122c5f404e 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<Integer, List<Range>> coverageByField;
public DataEvolutionGlobalIndexCoverage(
@@ -58,9 +59,24 @@ public class DataEvolutionGlobalIndexCoverage {
@Nullable Snapshot snapshot,
@Nullable PartitionPredicate partitionFilter,
Collection<IndexFileMeta> indexFiles) {
+ this(
+ table,
+ snapshot,
+ partitionFilter,
+ indexFiles,
+ table.coreOptions().scalarIndexSearchMode());
+ }
+
+ public DataEvolutionGlobalIndexCoverage(
+ FileStoreTable table,
+ @Nullable Snapshot snapshot,
+ @Nullable PartitionPredicate partitionFilter,
+ Collection<IndexFileMeta> 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 class DataEvolutionGlobalIndexCoverage {
}
public List<Range> unindexedRanges(Collection<Integer> 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 5f07e8a17e..215e1b31c5 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 @@ public class DataEvolutionGlobalIndexScanner implements
Closeable {
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<Integer, IndexMetaFileGroup> indexMetas = new HashMap<>();
Map<Integer, List<IndexMetaFileGroup>> 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 1b6ef7adc7..3dc1f36812 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 class DataEvolutionFullTextScan implements
FullTextScan {
if (!allIndexFiles.isEmpty()) {
List<Range> 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 317ffc6d52..e9a26020d5 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 class DataEvolutionVectorScan implements
VectorScan {
splits.add(new IndexVectorSearchSplit(range.from, range.to,
vectorFiles, scalarFiles));
}
+ GlobalIndexSearchMode vectorSearchMode =
table.coreOptions().vectorIndexSearchMode();
List<Range> rawRowRanges =
new DataEvolutionGlobalIndexCoverage(
- table, snapshot, partitionFilter,
vectorIndexFiles)
+ table,
+ snapshot,
+ partitionFilter,
+ vectorIndexFiles,
+ vectorSearchMode)
.unindexedRanges(vectorColumn.id());
if (filter != null) {
+ List<Range> 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 3374141a33..c514301be0 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 class PrimaryKeyFullTextRead implements FullTextRead {
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 class PrimaryKeyFullTextRead implements FullTextRead {
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 46232947f3..6a632ed646 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 @@ public class PrimaryKeyVectorRead implements VectorRead,
Serializable {
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 b3e91a3a4b..96643624a3 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 class CoreOptionsTest {
.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 c0d8ea01aa..ff1d993cf8 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 class BtreeGlobalIndexTableTest extends
DataEvolutionTestBase {
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 class BtreeGlobalIndexTableTest extends
DataEvolutionTestBase {
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 class BtreeGlobalIndexTableTest extends
DataEvolutionTestBase {
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 class BtreeGlobalIndexTableTest extends
DataEvolutionTestBase {
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 class BtreeGlobalIndexTableTest extends
DataEvolutionTestBase {
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 12af712f1d..16c418dc64 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 class FullTextSearchBuilderTest extends
TableTestBase {
(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 c00ecb18a2..a791a605b2 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 @@ class PrimaryKeyFullTextReadTest {
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 2153b44f2b..9f747807c0 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 class VectorSearchBuilderTest extends TableTestBase {
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 class VectorSearchBuilderTest extends
TableTestBase {
@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 class VectorSearchBuilderTest extends
TableTestBase {
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 class VectorSearchBuilderTest extends
TableTestBase {
}
@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 03ea04d144..fc0bd93697 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 @@ class DataEvolutionDelete implements Serializable {
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 b3a57fe732..5c5cce7491 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 class DataEvolutionMergeIntoAction extends
TableActionBase {
// _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 2deff25d13..bcc25cc1bf 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 class DataEvolutionMergeIntoActionITCase extends
ActionITCaseBase {
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 08d3e58b11..f220f84808 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 class DeleteActionDataEvolutionITCase extends
ActionITCaseBase {
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 2ac1ae0e9b..2d17cccfb9 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 20ee9ddb8c..392ea653a0 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 @@ class CoreOptions:
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 07a9213bb8..19aa1c251f 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 @@ class DataEvolutionGlobalIndexCoverage:
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 @@ class DataEvolutionGlobalIndexCoverage:
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 @@ class DataEvolutionGlobalIndexCoverage:
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 @@ class DataEvolutionGlobalIndexCoverage:
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 4ee6b94ced..192c2e99bb 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 @@ class DataEvolutionGlobalIndexScanner:
"""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 e952a9f3a4..c2fcff26a9 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -498,7 +498,9 @@ class FileScanner:
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 5e0038a184..1387b02536 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 @@ class DataEvolutionFullTextScan(FullTextScan):
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 e98a74cef3..1e094e543f 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 @@ class FullTextSearchBuilderImpl(FullTextSearchBuilder):
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 7b447c37a3..2386cdf2d2 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 @@ class PrimaryKeyFullTextRead(DataEvolutionFullTextRead):
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 8850840e39..fd6a65e83e 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 @@ class PrimaryKeyVectorRead(DataEvolutionVectorRead):
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_builder.py
b/paimon-python/pypaimon/table/source/vector_search_builder.py
index 5675f94c10..af823e58c2 100644
--- a/paimon-python/pypaimon/table/source/vector_search_builder.py
+++ b/paimon-python/pypaimon/table/source/vector_search_builder.py
@@ -132,27 +132,20 @@ class AbstractVectorSearchBuilderImpl:
# type: (Predicate) -> VectorSearchBuilder
if predicate is None:
return self
- if self._filter is None:
- self._filter = predicate
- else:
- self._filter = PredicateBuilder.and_predicates([self._filter,
predicate])
- # split out the partition-only conjuncts and store them as
_partition_filter for
- # manifest pruning. Non-partition conjuncts remain in self._filter;
- # the silent drop of non-partition conjuncts *in the extracted copy*
- # is intentional — nothing is lost overall.
- extracted = self._extract_partition_only_conjuncts(predicate)
- if extracted is not None:
- if self._partition_filter is None:
- self._partition_filter = extracted
+ partition_filter, data_filter = self._split_partition_filter(predicate)
+ if partition_filter is not None:
+ self._add_partition_filter(partition_filter)
+ if data_filter is not None:
+ if self._filter is None:
+ self._filter = data_filter
else:
- self._partition_filter = PredicateBuilder.and_predicates(
- [self._partition_filter, extracted])
+ self._filter = PredicateBuilder.and_predicates(
+ [self._filter, data_filter])
return self
def with_partition_filter(self, partition_filter):
# type: (Predicate) -> VectorSearchBuilder
if partition_filter is None:
- self._partition_filter = None
return self
# Strict: every referenced field must be a partition key, otherwise a
# non-partition conjunct would be silently dropped (with_filter has
@@ -169,31 +162,42 @@ class AbstractVectorSearchBuilderImpl:
"Partition filter must reference only partition keys "
"(%s); got non-partition field(s): %s"
% (partition_keys, sorted(extras)))
- self._partition_filter = self._rebuild_leaf_indices_by_name(
- partition_filter,
- {name: idx for idx, name in enumerate(partition_keys)},
- )
+ self._add_partition_filter(
+ self._rebuild_leaf_indices_by_name(
+ partition_filter,
+ {name: idx for idx, name in enumerate(partition_keys)},
+ ))
return self
- def _extract_partition_only_conjuncts(self, predicate):
- """AND-split ``predicate``, keep conjuncts that reference ONLY
- partition keys, and rebuild their leaf indices against the
- partition-only row by field name (so the caller's PredicateBuilder
- convention — full-row or partition-row — doesn't matter).
- """
+ def _split_partition_filter(self, predicate):
+ """Split partition-only and data conjuncts."""
partition_keys = list(self._table.partition_keys or [])
if not partition_keys:
- return None
+ return None, predicate
from pypaimon.read.push_down_utils import _split_and, _get_all_fields
partition_key_set = set(partition_keys)
pk_to_idx = {name: idx for idx, name in enumerate(partition_keys)}
- kept = [p for p in _split_and(predicate)
- if _get_all_fields(p).issubset(partition_key_set)]
- if not kept:
- return None
- rebuilt = [self._rebuild_leaf_indices_by_name(p, pk_to_idx)
- for p in kept]
- return PredicateBuilder.and_predicates(rebuilt)
+ partition_parts = []
+ data_parts = []
+ for part in _split_and(predicate):
+ if _get_all_fields(part).issubset(partition_key_set):
+ partition_parts.append(
+ self._rebuild_leaf_indices_by_name(part, pk_to_idx))
+ else:
+ data_parts.append(part)
+ return (
+ PredicateBuilder.and_predicates(partition_parts),
+ PredicateBuilder.and_predicates(data_parts),
+ )
+
+ def _add_partition_filter(self, partition_filter):
+ if partition_filter is None:
+ return
+ if self._partition_filter is None:
+ self._partition_filter = partition_filter
+ else:
+ self._partition_filter = PredicateBuilder.and_predicates(
+ [self._partition_filter, partition_filter])
@classmethod
def _rebuild_leaf_indices_by_name(cls, predicate, pk_to_idx):
diff --git a/paimon-python/pypaimon/table/source/vector_search_read.py
b/paimon-python/pypaimon/table/source/vector_search_read.py
index eba11094f3..84b0ae7c00 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 @@ class AbstractVectorSearchReadImpl:
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 e9f65a5b61..7e002064db 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 @@ class DataEvolutionVectorScan(VectorSearchScan):
)
)
+ 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 bc48ef3ccc..80a3e7934c 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 @@ class JavaPyReadWriteTest(unittest.TestCase):
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 0000000000..b900c52ca0
--- /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 4d7a1b142b..627da08af0 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 GlobalIndexScalarFallbackTest(unittest.TestCase):
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 GlobalIndexScalarFallbackTest(unittest.TestCase):
[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 bf1e560097..5a599f2699 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 @@ import unittest
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 @@ class PrimaryKeyIndexDefinitionsTest(unittest.TestCase):
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 @@ class PrimaryKeyIndexDefinitionsTest(unittest.TestCase):
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 @@ class PrimaryKeyIndexDefinitionsTest(unittest.TestCase):
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 afbb18b469..40ceaf2489 100644
--- a/paimon-python/pypaimon/tests/table_update_test.py
+++ b/paimon-python/pypaimon/tests/table_update_test.py
@@ -94,13 +94,14 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
}, 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 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
)
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 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
))
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 55fe9bb07b..d677ba2a81 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 @@ import unittest
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 @@ class _StubTable:
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 @@ class FullTextSearchBuilderDslTest(unittest.TestCase):
[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 @@ class FullTextSearchBuilderDslTest(unittest.TestCase):
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 @@ class VectorSearchFilterTest(unittest.TestCase):
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 @@ class VectorSearchFilterTest(unittest.TestCase):
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 @@ class VectorSearchFilterTest(unittest.TestCase):
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)
- def global_index_search_mode(self_inner):
- return
CoreOptions(self_inner.options).global_index_search_mode()
+ 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
+
+ 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 @@ class VectorSearchFilterTest(unittest.TestCase):
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 @@ class VectorSearchFilterTest(unittest.TestCase):
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,
+ )
+
+ 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 global_index_search_mode(self_inner):
- return
CoreOptions(self_inner.options).global_index_search_mode()
+ 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 @@ class VectorSearchFilterTest(unittest.TestCase):
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 @@ class
VectorSearchMultiShardScalarTest(unittest.TestCase):
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 @@ class VectorSearchMultiShardScalarTest(unittest.TestCase):
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])
@@ -1988,6 +2075,8 @@ class
VectorSearchPartitionedFilterTest(unittest.TestCase):
"""A normal full-row predicate ``pt == 2`` must (a) be auto-split
into _partition_filter with indices re-based to the partition-only
row, and (b) drop pt=1 entries during manifest pruning."""
+ _patch_snapshot(
+ self, self.entries, types.SimpleNamespace(next_row_id=10))
pb = PredicateBuilder(self.table.fields)
builder = (VectorSearchBuilderImpl(self.table)
.with_vector_column("embedding")
@@ -1995,6 +2084,7 @@ class
VectorSearchPartitionedFilterTest(unittest.TestCase):
.with_limit(3)
.with_filter(pb.equal("pt", 2)))
+ self.assertIsNone(builder._filter)
# Partition filter's leaf index points into the partition-only row.
self.assertEqual("pt", builder._partition_filter.field)
self.assertEqual(0, builder._partition_filter.index)
@@ -2004,8 +2094,146 @@ class
VectorSearchPartitionedFilterTest(unittest.TestCase):
self.assertEqual(["vec-pt2.index"],
[f.file_name for f in splits[0].vector_index_files])
+ def test_partition_only_filter_is_not_scalar_prefilter(self):
+ from pypaimon.globalindex.vector_search_result import (
+ DictBasedScoredIndexResult,
+ )
+ from pypaimon.table.source.batch_vector_search_builder import (
+ BatchVectorSearchBuilderImpl,
+ )
+
+ self.table.options = CoreOptions(Options({
+ "scalar-index.search-mode": "fast",
+ }))
+ _patch_snapshot(
+ self, self.entries, types.SimpleNamespace(next_row_id=10))
+ searches = []
+
+ def _fake_create(index_type, file_io, index_path,
+ index_io_meta_list, options=None):
+ class _FakeReader:
+ def visit_vector_search(self_inner, search):
+ searches.append(search)
+ scores = {} if (search.include_row_ids is not None and
+ search.include_row_ids.is_empty()) else
{0: 1.0}
+ return _completed_future(
+ DictBasedScoredIndexResult(scores))
+
+ def visit_batch_vector_search(self_inner, search):
+ searches.append(search)
+ scores = {} if (search.include_row_ids is not None and
+ search.include_row_ids.is_empty()) else
{0: 1.0}
+ return _completed_future([
+ DictBasedScoredIndexResult(scores)
+ for _ in range(search.vector_count)
+ ])
+
+ def close(self_inner):
+ pass
+
+ return _FakeReader()
+
+ partition_filter = PredicateBuilder(self.table.fields).equal("pt", 2)
+ with mock.patch(
+
"pypaimon.table.source.vector_search_read._create_vector_reader",
+ side_effect=_fake_create):
+ direct = (
+ VectorSearchBuilderImpl(self.table)
+ .with_vector_column("embedding")
+ .with_query_vector([1.0, 0.0, 0.0, 0.0])
+ .with_limit(3)
+ .with_filter(partition_filter)
+ .execute_local()
+ )
+ batch = (
+ BatchVectorSearchBuilderImpl(self.table)
+ .with_vector_column("embedding")
+ .with_query_vectors([[1.0, 0.0, 0.0, 0.0]])
+ .with_limit(3)
+ .with_filter(partition_filter)
+ .execute_batch_local()
+ )
+
+ self.assertEqual([5], list(direct.results()))
+ self.assertEqual([[5]], [list(result.results()) for result in batch])
+ self.assertTrue(all(search.include_row_ids is None for search in
searches))
+
+ def test_with_filter_keeps_only_data_conjuncts_as_scalar_filter(self):
+ pb = PredicateBuilder(self.table.fields)
+ builder = VectorSearchBuilderImpl(self.table).with_filter(
+ PredicateBuilder.and_predicates([
+ pb.equal("pt", 2),
+ pb.greater_or_equal("id", 5),
+ ]))
+
+ self.assertEqual("pt", builder._partition_filter.field)
+ self.assertEqual("id", builder._filter.field)
+
+ cross_field_or = PredicateBuilder.or_predicates([
+ pb.equal("pt", 2),
+ pb.greater_or_equal("id", 5),
+ ])
+ builder =
VectorSearchBuilderImpl(self.table).with_filter(cross_field_or)
+ self.assertIsNone(builder._partition_filter)
+ self.assertIs(cross_field_or, builder._filter)
+
+ def test_partition_filters_accumulate_in_any_order(self):
+ from pypaimon.table.source.batch_vector_search_builder import (
+ BatchVectorSearchBuilderImpl,
+ )
+
+ _patch_snapshot(
+ self, self.entries, types.SimpleNamespace(next_row_id=10))
+ pb = PredicateBuilder(self.table.fields)
+ auto_filter = pb.equal("pt", 2)
+ explicit_filter = pb.equal("pt", 1)
+
+ for builder_class in (
+ VectorSearchBuilderImpl, BatchVectorSearchBuilderImpl):
+ builders = [
+ builder_class(self.table)
+ .with_vector_column("embedding")
+ .with_filter(auto_filter)
+ .with_partition_filter(explicit_filter),
+ builder_class(self.table)
+ .with_vector_column("embedding")
+ .with_partition_filter(explicit_filter)
+ .with_filter(auto_filter),
+ ]
+ for builder in builders:
+ self.assertEqual(
+ [], builder.new_vector_search_scan().scan().splits())
+
+ def test_none_partition_filter_is_noop(self):
+ from pypaimon.table.source.batch_vector_search_builder import (
+ BatchVectorSearchBuilderImpl,
+ )
+
+ _patch_snapshot(
+ self, self.entries, types.SimpleNamespace(next_row_id=10))
+ partition_filter = PredicateBuilder(self.table.fields).equal("pt", 2)
+
+ for builder_class in (
+ VectorSearchBuilderImpl, BatchVectorSearchBuilderImpl):
+ builders = [
+ builder_class(self.table)
+ .with_vector_column("embedding")
+ .with_partition_filter(partition_filter)
+ .with_partition_filter(None),
+ builder_class(self.table)
+ .with_vector_column("embedding")
+ .with_partition_filter(None)
+ .with_partition_filter(partition_filter),
+ ]
+ for builder in builders:
+ splits = builder.new_vector_search_scan().scan().splits()
+ self.assertEqual(1, len(splits))
+ self.assertEqual(
+ ["vec-pt2.index"],
+ [f.file_name for f in splits[0].vector_index_files])
+
def test_with_partition_filter_rejects_non_partition_field(self):
- """Non-partition conjuncts would be silently dropped by the extractor,
+ """Non-partition conjuncts would be silently dropped by the splitter,
producing wrong results; the API must refuse them up front."""
pb = PredicateBuilder(self.table.fields)
builder = VectorSearchBuilderImpl(self.table)
diff --git a/paimon-python/pypaimon/write/table_update.py
b/paimon-python/pypaimon/write/table_update.py
index a08418591e..8095c16c2a 100644
--- a/paimon-python/pypaimon/write/table_update.py
+++ b/paimon-python/pypaimon/write/table_update.py
@@ -273,7 +273,7 @@ class TableUpdate:
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 5690a65d5f..79e09bc380 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 f287d31959..479515ca46 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 72258ddbec..f88993dba8 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)")
