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 3ab3a2c65f [python] Support raw fallback for global indexes (#8310)
3ab3a2c65f is described below
commit 3ab3a2c65ffa5b46492b6b928a987babba24800e
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Jun 21 22:48:59 2026 +0800
[python] Support raw fallback for global indexes (#8310)
This PR ports the Java raw fallback behavior for global indexes to
Python. It adds `global-index.search-mode` support and lets scalar
global index scans and vector search include unindexed row ranges when
configured with `full` or `detail` search modes.
---
.../test/java/org/apache/paimon/JavaPyE2ETest.java | 71 ++++
paimon-python/dev/run_mixed_tests.sh | 79 +++-
.../pypaimon/common/options/core_options.py | 19 +
.../pypaimon/globalindex/global_index_coverage.py | 145 +++++++
.../pypaimon/globalindex/global_index_scanner.py | 102 ++++-
paimon-python/pypaimon/read/read_builder.py | 8 +-
.../pypaimon/read/scanner/file_scanner.py | 5 +-
paimon-python/pypaimon/read/table_scan.py | 30 +-
.../table/source/batch_vector_search_builder.py | 1 +
.../pypaimon/table/source/vector_search_builder.py | 2 +
.../pypaimon/table/source/vector_search_read.py | 314 ++++++++++++++-
.../pypaimon/table/source/vector_search_scan.py | 123 +++++-
.../pypaimon/table/source/vector_search_split.py | 44 ++-
.../pypaimon/tests/e2e/java_py_read_write_test.py | 62 +++
paimon-python/pypaimon/tests/global_index_test.py | 194 ++++++++++
.../pypaimon/tests/vector_search_filter_test.py | 423 +++++++++++++++++++++
.../test/java/org/apache/paimon/JavaPyE2ETest.java | 117 ++++++
17 files changed, 1685 insertions(+), 54 deletions(-)
diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index 2a7584a0bc..695cc29d76 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -516,6 +516,77 @@ public class JavaPyE2ETest {
testBtreeIndexWriteNull();
}
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testBtreeRawFallbackWrite() throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.STRING(),
DataTypes.STRING()},
+ new String[] {"k", "v"});
+ Options options = new Options();
+ Path tablePath = new Path(warehouse.toString() +
"/default.db/test_btree_raw_fallback");
+ LocalFileIO.create().delete(tablePath, true);
+ options.set(PATH, tablePath.toString());
+ options.set(ROW_TRACKING_ENABLED, true);
+ options.set(DATA_EVOLUTION_ENABLED, true);
+ options.set(GLOBAL_INDEX_ENABLED, true);
+ TableSchema tableSchema =
+ SchemaUtils.forceCommit(
+ new SchemaManager(LocalFileIO.create(), tablePath),
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options.toMap(),
+ ""));
+ AppendOnlyFileStoreTable table =
+ new AppendOnlyFileStoreTable(
+ FileIOFinder.find(tablePath),
+ tablePath,
+ tableSchema,
+ CatalogEnvironment.empty());
+
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(
+ GenericRow.of(BinaryString.fromString("k1"),
BinaryString.fromString("v1")));
+ write.write(
+ GenericRow.of(BinaryString.fromString("k2"),
BinaryString.fromString("v2")));
+ write.write(
+ GenericRow.of(BinaryString.fromString("k3"),
BinaryString.fromString("v3")));
+ commit.commit(write.prepareCommit());
+ }
+
+ SortedGlobalIndexBuilder builder =
+ new SortedGlobalIndexBuilder(table,
"btree").withIndexField("k");
+ try (BatchTableCommit commit = writeBuilder.newCommit()) {
+ commit.commit(
+ builder.build(
+ builder.scan()
+
.map(org.apache.paimon.utils.Pair::getValue)
+ .orElseThrow(
+ () ->
+ new IllegalStateException(
+ "Expected scan
result when building index."))
+ .get(0),
+ IOManager.create(warehouse.toString())));
+ }
+
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(
+ GenericRow.of(BinaryString.fromString("k4"),
BinaryString.fromString("v4")));
+ commit.commit(write.prepareCommit());
+ }
+
+ List<IndexManifestEntry> indexEntries =
+
table.indexManifestFileReader().read(table.latestSnapshot().get().indexManifest);
+ assertThat(indexEntries)
+ .singleElement()
+ .matches(entry -> entry.indexFile().rowCount() == 3);
+ }
+
@Test
@EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
public void testBitmapIndexWrite() throws Exception {
diff --git a/paimon-python/dev/run_mixed_tests.sh
b/paimon-python/dev/run_mixed_tests.sh
index c3d91893df..8da1cb345b 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -219,6 +219,29 @@ run_btree_index_test() {
fi
}
+run_btree_raw_fallback_test() {
+ echo -e "${YELLOW}=== Running BTree Raw Fallback Test (Java Write, Python
Read) ===${NC}"
+
+ cd "$PROJECT_ROOT"
+
+ echo "Running Maven test for JavaPyE2ETest.testBtreeRawFallbackWrite..."
+ if mvn test
-Dtest=org.apache.paimon.JavaPyE2ETest#testBtreeRawFallbackWrite -pl
paimon-core -am -q -DfailIfNoTests=false -Drun.e2e.tests=true; then
+ echo -e "${GREEN}✓ Java test completed successfully${NC}"
+ else
+ echo -e "${RED}✗ Java test failed${NC}"
+ return 1
+ fi
+ cd "$PAIMON_PYTHON_DIR"
+ echo "Running Python test for
JavaPyReadWriteTest.test_read_btree_raw_fallback..."
+ if python -m pytest
java_py_read_write_test.py::JavaPyReadWriteTest::test_read_btree_raw_fallback
-v; then
+ echo -e "${GREEN}✓ Python test completed successfully${NC}"
+ return 0
+ else
+ echo -e "${RED}✗ Python test failed${NC}"
+ return 1
+ fi
+}
+
run_bitmap_index_test() {
echo -e "${YELLOW}=== Step 6b: Running Bitmap Index Test (Java Write,
Python Read) ===${NC}"
@@ -541,6 +564,32 @@ run_vindex_vector_test() {
fi
}
+run_vindex_vector_raw_fallback_test() {
+ echo -e "${YELLOW}=== Running paimon-vindex Vector Raw Fallback Test (Java
Write, Python Read) ===${NC}"
+
+ cd "$PROJECT_ROOT"
+
+ echo "Running Maven test for
JavaPyE2ETest.testVindexVectorRawFallbackWrite..."
+ if mvn test
-Dtest=org.apache.paimon.JavaPyE2ETest#testVindexVectorRawFallbackWrite -pl
paimon-vector -am -q -DfailIfNoTests=false -Drun.e2e.tests=true; then
+ echo -e "${GREEN}✓ Java test completed successfully${NC}"
+ else
+ echo -e "${RED}✗ Java test failed${NC}"
+ return 1
+ fi
+ cd "$PAIMON_PYTHON_DIR"
+ if ! ensure_paimon_vindex; then
+ return 1
+ fi
+ echo "Running Python test for
JavaPyReadWriteTest.test_read_vindex_vector_raw_fallback..."
+ if python -m pytest
java_py_read_write_test.py::JavaPyReadWriteTest::test_read_vindex_vector_raw_fallback
-v; then
+ echo -e "${GREEN}✓ Python test completed successfully${NC}"
+ return 0
+ else
+ echo -e "${RED}✗ Python test failed${NC}"
+ return 1
+ fi
+}
+
run_compact_conflict_test() {
echo -e "${YELLOW}=== Running Compact Conflict Test (Java Write Base,
Python Shard Update + Java Compact) ===${NC}"
@@ -796,6 +845,7 @@ main() {
local java_read_result=0
local pk_dv_result=0
local btree_index_result=0
+ local btree_raw_fallback_result=0
local bitmap_index_result=0
local compressed_global_index_result=0
local compressed_text_result=0
@@ -804,6 +854,7 @@ main() {
local lumina_vector_result=0
local lumina_vector_btree_result=0
local vindex_vector_result=0
+ local vindex_vector_raw_fallback_result=0
local compact_conflict_result=0
local blob_compact_conflict_result=0
local blob_alter_compact_result=0
@@ -873,6 +924,13 @@ main() {
echo ""
+ # Run BTree raw fallback test (Java write indexed + unindexed rows, Python
read)
+ if ! run_btree_raw_fallback_test; then
+ btree_raw_fallback_result=1
+ fi
+
+ echo ""
+
# Run Bitmap index test (Java write, Python read)
if ! run_bitmap_index_test; then
bitmap_index_result=1
@@ -967,9 +1025,16 @@ main() {
if ! run_vindex_vector_test; then
vindex_vector_result=1
fi
+
+ echo ""
+
+ if ! run_vindex_vector_raw_fallback_test; then
+ vindex_vector_raw_fallback_result=1
+ fi
else
echo -e "${YELLOW}⏭ Skipping paimon-vindex Vector Index Test (requires
Python >= 3.9, current: $PYTHON_VERSION)${NC}"
vindex_vector_result=0
+ vindex_vector_raw_fallback_result=0
fi
echo ""
@@ -1073,6 +1138,12 @@ main() {
echo -e "${RED}✗ BTree Index Test (Java Write, Python Read):
FAILED${NC}"
fi
+ if [[ $btree_raw_fallback_result -eq 0 ]]; then
+ echo -e "${GREEN}✓ BTree Raw Fallback Test (Java Write, Python Read):
PASSED${NC}"
+ else
+ echo -e "${RED}✗ BTree Raw Fallback Test (Java Write, Python Read):
FAILED${NC}"
+ fi
+
if [[ $bitmap_index_result -eq 0 ]]; then
echo -e "${GREEN}✓ Bitmap Index Test (Java Write, Python Read):
PASSED${NC}"
else
@@ -1145,6 +1216,12 @@ main() {
echo -e "${RED}✗ paimon-vindex Vector Index Test (Java Write, Python
Read): FAILED${NC}"
fi
+ if [[ $vindex_vector_raw_fallback_result -eq 0 ]]; then
+ echo -e "${GREEN}✓ paimon-vindex Vector Raw Fallback Test (Java Write,
Python Read): PASSED${NC}"
+ else
+ echo -e "${RED}✗ paimon-vindex Vector Raw Fallback Test (Java Write,
Python Read): FAILED${NC}"
+ fi
+
if [[ $compact_conflict_result -eq 0 ]]; then
echo -e "${GREEN}✓ Compact Conflict Test (Java Write+Compact, Python
Read): PASSED${NC}"
else
@@ -1198,7 +1275,7 @@ main() {
# Clean up warehouse directory after all tests
cleanup_warehouse
- if [[ $java_write_result -eq 0 && $python_read_result -eq 0 &&
$python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 &&
$btree_index_result -eq 0 && $bitmap_index_result -eq 0 &&
$compressed_global_index_result -eq 0 && $compressed_text_result -eq 0 &&
$tantivy_fulltext_result -eq 0 && $lumina_vector_result -eq 0 &&
$lumina_vector_btree_result -eq 0 && $vindex_vector_result -eq 0 &&
$compact_conflict_result -eq 0 && $blob_compact_conflict_result -eq 0 && $blob_
[...]
+ if [[ $java_write_result -eq 0 && $python_read_result -eq 0 &&
$python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 &&
$btree_index_result -eq 0 && $btree_raw_fallback_result -eq 0 &&
$bitmap_index_result -eq 0 && $compressed_global_index_result -eq 0 &&
$compressed_text_result -eq 0 && $tantivy_fulltext_result -eq 0 &&
$lumina_vector_result -eq 0 && $lumina_vector_btree_result -eq 0 &&
$vindex_vector_result -eq 0 && $vindex_vector_raw_fallback_result -eq 0 && [...]
echo -e "${GREEN}🎉 All tests passed! Java-Python interoperability
verified.${NC}"
return 0
else
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index 93b7b92b49..2b24bc1be4 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -90,6 +90,12 @@ class GlobalIndexColumnUpdateAction(str, Enum):
DROP_PARTITION_INDEX = "DROP_PARTITION_INDEX"
+class GlobalIndexSearchMode(str, Enum):
+ FAST = "fast"
+ FULL = "full"
+ DETAIL = "detail"
+
+
class CoreOptions:
"""Core options for Paimon tables."""
# File format constants
@@ -618,6 +624,16 @@ class CoreOptions:
.with_description("Whether to enable global index for scan.")
)
+ GLOBAL_INDEX_SEARCH_MODE: ConfigOption[GlobalIndexSearchMode] = (
+ ConfigOptions.key("global-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'."
+ )
+ )
+
GLOBAL_INDEX_THREAD_NUM: ConfigOption[int] = (
ConfigOptions.key("global-index.thread-num")
.int_type()
@@ -1109,6 +1125,9 @@ class CoreOptions:
def global_index_enabled(self, default=None):
return self.options.get(CoreOptions.GLOBAL_INDEX_ENABLED, default)
+ def global_index_search_mode(self):
+ return self.options.get(CoreOptions.GLOBAL_INDEX_SEARCH_MODE)
+
def global_index_thread_num(self) -> Optional[int]:
return self.options.get(CoreOptions.GLOBAL_INDEX_THREAD_NUM)
diff --git a/paimon-python/pypaimon/globalindex/global_index_coverage.py
b/paimon-python/pypaimon/globalindex/global_index_coverage.py
new file mode 100644
index 0000000000..9f94f89495
--- /dev/null
+++ b/paimon-python/pypaimon/globalindex/global_index_coverage.py
@@ -0,0 +1,145 @@
+# 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.
+
+"""Row ranges covered and not covered by global index files."""
+
+from typing import Collection, Dict, List, Optional, Union
+
+from pypaimon.common.options.core_options import CoreOptions,
GlobalIndexSearchMode
+from pypaimon.common.options.options import Options
+from pypaimon.common.predicate import Predicate
+from pypaimon.read.push_down_utils import _get_all_fields
+from pypaimon.schema.data_types import DataField
+from pypaimon.utils.range import Range
+
+
+class GlobalIndexCoverage:
+ """Computes global-index coverage by field id."""
+
+ def __init__(
+ self,
+ table,
+ snapshot,
+ partition_filter,
+ index_files: Collection['IndexFileMeta'],
+ ):
+ self._table = table
+ self._snapshot = snapshot
+ self._partition_filter = partition_filter
+ self._coverage_by_field: Dict[int, List[Range]] = {}
+ for index_file in index_files:
+ meta = index_file.global_index_meta
+ if meta is None:
+ continue
+ row_range = Range(meta.row_range_start, meta.row_range_end)
+ self._add_coverage(meta.index_field_id, row_range)
+ if meta.extra_field_ids is not None:
+ for extra_field_id in meta.extra_field_ids:
+ self._add_coverage(extra_field_id, row_range)
+
+ def unindexed_ranges(
+ self,
+ fields_or_field_id: Union[List[DataField], int],
+ predicate: Optional[Predicate] = None,
+ ) -> List[Range]:
+ if isinstance(fields_or_field_id, int):
+ field_ids = {fields_or_field_id}
+ else:
+ field_by_name = {f.name: f for f in fields_or_field_id}
+ field_ids = set()
+ for name in _get_all_fields(predicate):
+ field = field_by_name.get(name)
+ if field is not None:
+ field_ids.add(field.id)
+ return self._unindexed_ranges(field_ids)
+
+ def _add_coverage(self, field_id: int, row_range: Range) -> None:
+ self._coverage_by_field.setdefault(field_id, []).append(row_range)
+
+ def _indexed_ranges(self, field_ids: Collection[int]) -> List[Range]:
+ ranges = None
+ for field_id in field_ids:
+ field_ranges = self._coverage_by_field.get(field_id)
+ if not field_ranges:
+ return []
+ field_ranges = Range.sort_and_merge_overlap(field_ranges, True)
+ ranges = field_ranges if ranges is None else Range.and_(ranges,
field_ranges)
+ if ranges is None:
+ 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)
+ if search_mode == GlobalIndexSearchMode.FAST:
+ return []
+ next_row_id = getattr(self._snapshot, "next_row_id", None)
+ if self._snapshot is None or next_row_id is None:
+ return []
+ if next_row_id <= 0:
+ return []
+
+ if search_mode == GlobalIndexSearchMode.DETAIL:
+ data_ranges = self._data_ranges_by_data_files()
+ else:
+ data_ranges = [Range(0, next_row_id - 1)]
+
+ indexed_ranges = Range.sort_and_merge_overlap(
+ self._indexed_ranges(field_ids), True)
+ unindexed = []
+ for data_range in Range.sort_and_merge_overlap(data_ranges, True):
+ unindexed.extend(data_range.exclude(indexed_ranges))
+ return Range.sort_and_merge_overlap(unindexed, True)
+
+ def _data_ranges_by_data_files(self) -> List[Range]:
+ if hasattr(self._table, "data_ranges_for_global_index_coverage"):
+ return self._table.data_ranges_for_global_index_coverage(
+ self._snapshot,
+ self._partition_filter,
+ )
+
+ manifest_list_manager = getattr(self._table, "manifest_list_manager",
None)
+ if manifest_list_manager is None:
+ from pypaimon.manifest.manifest_list_manager import
ManifestListManager
+ manifest_list_manager = ManifestListManager(self._table)
+ manifest_files = manifest_list_manager.read_all(self._snapshot)
+ from pypaimon.manifest.manifest_file_manager import ManifestFileManager
+ manager = ManifestFileManager(self._table)
+ entries = manager.read_entries_parallel(
+ manifest_files,
+ self._entry_matches_partition,
+ max_workers=self._table.options.scan_manifest_parallelism(),
+ )
+ data_ranges = []
+ for entry in entries:
+ row_range = entry.file.row_id_range()
+ if row_range is not None:
+ data_ranges.append(row_range)
+ return data_ranges
+
+ def _entry_matches_partition(self, entry) -> bool:
+ if self._partition_filter is None:
+ return True
+ return self._partition_filter.test(entry.partition)
+
+
+def _global_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()
diff --git a/paimon-python/pypaimon/globalindex/global_index_scanner.py
b/paimon-python/pypaimon/globalindex/global_index_scanner.py
index 23af17e405..7f2470d382 100644
--- a/paimon-python/pypaimon/globalindex/global_index_scanner.py
+++ b/paimon-python/pypaimon/globalindex/global_index_scanner.py
@@ -27,6 +27,7 @@ from pypaimon.globalindex.global_index_result import
GlobalIndexResult
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.options.options import Options
from pypaimon.common.predicate import Predicate
+from pypaimon.globalindex.global_index_coverage import GlobalIndexCoverage
from pypaimon.read.push_down_utils import _get_all_fields
from pypaimon.schema.data_types import DataField
from pypaimon.utils.range import Range
@@ -43,11 +44,20 @@ class GlobalIndexScanner:
index_files: Collection['IndexFileMeta'],
thread_num: Optional[int] = None,
options: Optional[CoreOptions] = None,
+ table=None,
+ snapshot=None,
+ partition_filter=None,
):
self._options = options or CoreOptions(Options.from_none())
self._executor = ThreadPoolExecutor(
max_workers=thread_num or 32
)
+ self._fields = fields
+ self._coverage = (
+ GlobalIndexCoverage(table, snapshot, partition_filter, index_files)
+ if table is not None
+ else None
+ )
self._evaluator = self._create_evaluator(
fields, file_io, index_path, index_files
)
@@ -59,17 +69,10 @@ class GlobalIndexScanner:
if global_index_meta is None:
continue
- field_id = global_index_meta.index_field_id
index_type = index_file.index_type
-
- if field_id not in index_metas:
- index_metas[field_id] = {}
- if index_type not in index_metas[field_id]:
- index_metas[field_id][index_type] = {}
-
- range_key = Range(global_index_meta.row_range_start,
global_index_meta.row_range_end)
- if range_key not in index_metas[field_id][index_type]:
- index_metas[field_id][index_type][range_key] = []
+ field_ids = [global_index_meta.index_field_id]
+ if global_index_meta.extra_field_ids is not None:
+ field_ids.extend(global_index_meta.extra_field_ids)
io_meta = GlobalIndexIOMeta(
file_name=index_file.file_name,
@@ -77,7 +80,17 @@ class GlobalIndexScanner:
metadata=global_index_meta.index_meta,
external_path=index_file.external_path,
)
- index_metas[field_id][index_type][range_key].append(io_meta)
+ range_key = Range(
+ global_index_meta.row_range_start,
+ global_index_meta.row_range_end)
+ for field_id in field_ids:
+ if field_id not in index_metas:
+ index_metas[field_id] = {}
+ if index_type not in index_metas[field_id]:
+ index_metas[field_id][index_type] = {}
+ if range_key not in index_metas[field_id][index_type]:
+ index_metas[field_id][index_type][range_key] = []
+ index_metas[field_id][index_type][range_key].append(io_meta)
executor = self._executor
options = self._options
@@ -105,13 +118,17 @@ class GlobalIndexScanner:
if index_files is not None:
if len(index_files) == 0:
return None
+ core_options = _core_options(table)
return GlobalIndexScanner(
fields=table.fields,
file_io=table.file_io,
index_path=table.path_factory().global_index_path_factory().index_path(),
index_files=index_files,
- thread_num=table.options.global_index_thread_num(),
- options=table.options,
+ thread_num=core_options.global_index_thread_num(),
+ options=core_options,
+ table=table,
+ snapshot=_resolve_snapshot(table, snapshot),
+ partition_filter=partition_filter,
)
# Scan index files from snapshot using partition_filter and predicate
@@ -129,29 +146,47 @@ class GlobalIndexScanner:
global_index_meta = entry.index_file.global_index_meta
if global_index_meta is None:
return False
- return global_index_meta.index_field_id in filter_field_ids
+ if global_index_meta.index_field_id in filter_field_ids:
+ return True
+ if global_index_meta.extra_field_ids is not None:
+ return any(
+ field_id in filter_field_ids
+ for field_id in global_index_meta.extra_field_ids
+ )
+ return False
if snapshot is None:
- snapshot = table.snapshot_manager().get_latest_snapshot()
+ snapshot = _resolve_snapshot(table, None)
index_file_handler = IndexFileHandler(table=table)
entries = index_file_handler.scan(snapshot, index_file_filter)
scanned_index_files = [entry.index_file for entry in entries]
if len(scanned_index_files) == 0:
return None
+ core_options = _core_options(table)
return GlobalIndexScanner(
fields=table.fields,
file_io=table.file_io,
index_path=table.path_factory().global_index_path_factory().index_path(),
index_files=scanned_index_files,
- thread_num=table.options.global_index_thread_num(),
- options=table.options,
+ thread_num=core_options.global_index_thread_num(),
+ options=core_options,
+ table=table,
+ snapshot=snapshot,
+ partition_filter=partition_filter,
)
def scan(self, predicate: Optional[Predicate]) ->
Optional[GlobalIndexResult]:
"""Scan the global index with the given predicate."""
return self._evaluator.evaluate(predicate)
+ def unindexed_rows(self, predicate: Optional[Predicate]) ->
GlobalIndexResult:
+ """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))
+
def close(self):
"""Close the scanner and release resources."""
self._evaluator.close()
@@ -164,6 +199,39 @@ class GlobalIndexScanner:
self.close()
+def _resolve_snapshot(table, snapshot):
+ if snapshot is not None:
+ return snapshot
+ snapshot_manager = table.snapshot_manager()
+ if snapshot_manager is None:
+ return None
+ try:
+ from pypaimon.snapshot.time_travel_util import TimeTravelUtil
+ table_options = getattr(table.table_schema, "options", {})
+ scan_keys = getattr(TimeTravelUtil, "SCAN_KEYS", None)
+ if scan_keys is None:
+ from pypaimon.snapshot.time_travel_util import SCAN_KEYS as
scan_keys
+ has_time_travel = any(key in table_options for key in scan_keys)
+ resolved = TimeTravelUtil.try_travel_to_snapshot(
+ Options(table.table_schema.options),
+ table.tag_manager(),
+ snapshot_manager,
+ )
+ if resolved is not None and (
+ has_time_travel or getattr(resolved, "next_row_id", None) is
not None):
+ return resolved
+ except Exception:
+ pass
+ return snapshot_manager.get_latest_snapshot()
+
+
+def _core_options(table):
+ options = getattr(table, "options", None)
+ if options is None:
+ return CoreOptions(Options.from_none())
+ return options
+
+
def _create_readers(file_io, index_path, index_type_metas, field,
executor=None, options=None):
"""Create readers for a specific field, dispatched by index_type.
diff --git a/paimon-python/pypaimon/read/read_builder.py
b/paimon-python/pypaimon/read/read_builder.py
index 5537c97dd6..eea670db53 100644
--- a/paimon-python/pypaimon/read/read_builder.py
+++ b/paimon-python/pypaimon/read/read_builder.py
@@ -44,12 +44,17 @@ class ReadBuilder:
# in ``read_type()`` and downstream consumers.
self._projection: Optional[List[str]] = None
self._nested_paths: Optional[List[List[int]]] = None
+ self._partition_filter: Optional[Predicate] = None
self._limit: Optional[int] = None
def with_filter(self, predicate: Predicate) -> 'ReadBuilder':
self._predicate = predicate
return self
+ def with_partition_filter(self, partition_filter: Predicate) ->
'ReadBuilder':
+ self._partition_filter = partition_filter
+ return self
+
def with_projection(self, projection: List[str]) -> 'ReadBuilder':
"""Project to the given column names.
@@ -77,7 +82,8 @@ class ReadBuilder:
return TableScan(
table=self.table,
predicate=self._predicate,
- limit=self._limit
+ limit=self._limit,
+ partition_predicate=self._partition_filter,
)
def new_read(self) -> TableRead:
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index e5f4e332a1..3facb2f7d5 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -384,7 +384,10 @@ class FileScanner:
if scanner is None:
return None
with scanner:
- return scanner.scan(self.predicate)
+ result = scanner.scan(self.predicate)
+ if result is None:
+ return None
+ return result.or_(scanner.unindexed_rows(self.predicate))
except Exception:
return None
diff --git a/paimon-python/pypaimon/read/table_scan.py
b/paimon-python/pypaimon/read/table_scan.py
index 36568c618f..b2fcaa8604 100755
--- a/paimon-python/pypaimon/read/table_scan.py
+++ b/paimon-python/pypaimon/read/table_scan.py
@@ -33,13 +33,15 @@ class TableScan:
self,
table,
predicate: Optional[Predicate],
- limit: Optional[int]
+ limit: Optional[int],
+ partition_predicate: Optional[Predicate] = None,
):
from pypaimon.table.file_store_table import FileStoreTable
self.table: FileStoreTable = table
self.predicate = predicate
self.limit = limit
+ self.partition_predicate = partition_predicate
self.file_scanner = self._create_file_scanner()
def plan(self) -> Plan:
@@ -79,7 +81,11 @@ class TableScan:
earliest_snapshot = snapshot_manager.try_get_earliest_snapshot()
latest_snapshot = snapshot_manager.get_latest_snapshot()
if earliest_snapshot is None or latest_snapshot is None:
- return FileScanner(self.table, lambda: ([], None))
+ return FileScanner(
+ self.table,
+ lambda: ([], None),
+ partition_predicate=self.partition_predicate,
+ )
start_timestamp = int(ts[0])
end_timestamp = int(ts[1])
if start_timestamp >= end_timestamp:
@@ -87,7 +93,11 @@ class TableScan:
"Ending timestamp %s should be >= starting timestamp %s."
% (end_timestamp, start_timestamp))
if (start_timestamp == end_timestamp or start_timestamp >
latest_snapshot.time_millis
or end_timestamp < earliest_snapshot.time_millis):
- return FileScanner(self.table, lambda: ([], None))
+ return FileScanner(
+ self.table,
+ lambda: ([], None),
+ partition_predicate=self.partition_predicate,
+ )
starting_snapshot =
snapshot_manager.earlier_or_equal_time_mills(start_timestamp)
earliest_snapshot = snapshot_manager.try_get_earliest_snapshot()
@@ -118,7 +128,13 @@ class TableScan:
manifests.extend(manifest_files)
return manifests, end_snapshot
- return FileScanner(self.table, incremental_manifest,
self.predicate, self.limit)
+ return FileScanner(
+ self.table,
+ incremental_manifest,
+ self.predicate,
+ self.limit,
+ partition_predicate=self.partition_predicate,
+ )
if has_time_travel:
def time_travel_manifest_scanner():
@@ -135,7 +151,8 @@ class TableScan:
self.table,
time_travel_manifest_scanner,
self.predicate,
- self.limit
+ self.limit,
+ partition_predicate=self.partition_predicate,
)
def all_manifests():
@@ -146,7 +163,8 @@ class TableScan:
self.table,
all_manifests,
self.predicate,
- self.limit
+ self.limit,
+ partition_predicate=self.partition_predicate,
)
def with_shard(self, idx_of_this_subtask, number_of_para_subtasks) ->
'TableScan':
diff --git a/paimon-python/pypaimon/table/source/batch_vector_search_builder.py
b/paimon-python/pypaimon/table/source/batch_vector_search_builder.py
index f987680a36..199759487b 100644
--- a/paimon-python/pypaimon/table/source/batch_vector_search_builder.py
+++ b/paimon-python/pypaimon/table/source/batch_vector_search_builder.py
@@ -123,5 +123,6 @@ class
BatchVectorSearchBuilderImpl(AbstractVectorSearchBuilderImpl,
self._vector_column,
self._query_vectors,
filter_=self._filter,
+ partition_filter=self._partition_filter,
options=self._options,
)
diff --git a/paimon-python/pypaimon/table/source/vector_search_builder.py
b/paimon-python/pypaimon/table/source/vector_search_builder.py
index f985471929..25b2d2e4c5 100644
--- a/paimon-python/pypaimon/table/source/vector_search_builder.py
+++ b/paimon-python/pypaimon/table/source/vector_search_builder.py
@@ -216,6 +216,7 @@ class AbstractVectorSearchBuilderImpl:
self._vector_column,
filter_=self._filter,
partition_filter=self._partition_filter,
+ options=self._options,
)
@@ -245,5 +246,6 @@ class
VectorSearchBuilderImpl(AbstractVectorSearchBuilderImpl, VectorSearchBuild
self._vector_column,
self._query_vector,
filter_=self._filter,
+ partition_filter=self._partition_filter,
options=self._options,
)
diff --git a/paimon-python/pypaimon/table/source/vector_search_read.py
b/paimon-python/pypaimon/table/source/vector_search_read.py
index 566b093cd6..9656308071 100644
--- a/paimon-python/pypaimon/table/source/vector_search_read.py
+++ b/paimon-python/pypaimon/table/source/vector_search_read.py
@@ -25,6 +25,12 @@ from pypaimon.globalindex.global_index_result import
GlobalIndexResult
from pypaimon.globalindex.offset_global_index_reader import
OffsetGlobalIndexReader
from pypaimon.globalindex.vector_search import VectorSearch
from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult
+from pypaimon.table.source.vector_search_split import (
+ IndexVectorSearchSplit,
+ RawVectorSearchSplit,
+)
+from pypaimon.utils.range import Range
+from pypaimon.utils.roaring_bitmap import RoaringBitmap64
class VectorSearchRead(ABC):
@@ -56,18 +62,27 @@ class BatchVectorSearchRead(ABC):
class AbstractVectorSearchReadImpl:
"""Base implementation for vector search reads."""
- def __init__(self, table, limit, vector_column, filter_=None,
options=None):
+ def __init__(
+ self,
+ table,
+ limit,
+ vector_column,
+ filter_=None,
+ partition_filter=None,
+ options=None,
+ ):
self._table = table
self._limit = limit
self._vector_column = vector_column
self._filter = filter_
+ self._partition_filter = partition_filter
self._options = dict(options or {})
- def _pre_filter(self, splits):
- # type: (list) -> Optional[RoaringBitmap64]
- """Evaluate the scalar filter against scalar global indexes to produce
a row-id bitmap."""
+ def _pre_filters(self, splits):
+ # type: (list) -> List[RoaringBitmap64]
+ """Evaluate scalar indexes and return one include bitmap per index
split."""
if self._filter is None:
- return None
+ return []
# Collect scalar index files across splits, deduplicated by file name.
seen = set()
@@ -79,18 +94,77 @@ class AbstractVectorSearchReadImpl:
seen.add(index_file.file_name)
scalar_files.append(index_file)
+ if not scalar_files:
+ return _empty_bitmaps(len(splits))
+
+ from pypaimon.globalindex.global_index_scanner import
GlobalIndexScanner
+ scanner = GlobalIndexScanner.create(
+ self._table,
+ index_files=scalar_files,
+ partition_filter=self._partition_filter,
+ )
+ if scanner is None:
+ return _empty_bitmaps(len(splits))
+ try:
+ result = scanner.scan(self._filter)
+ if result is None:
+ return _empty_bitmaps(len(splits))
+ matched_rows = result.results()
+ finally:
+ scanner.close()
+
+ include_row_ids = []
+ for split in splits:
+ split_rows = _bitmap_of_range(
+ Range(split.row_range_start, split.row_range_end))
+ include_row_ids.append(RoaringBitmap64.and_(matched_rows,
split_rows))
+ return include_row_ids
+
+ def _pre_filter(self, splits):
+ # Backwards-compatible helper used by older tests/callers.
+ pre_filters = self._pre_filters(splits)
+ if not pre_filters:
+ return None
+ merged = RoaringBitmap64()
+ for bitmap in pre_filters:
+ merged = RoaringBitmap64.or_(merged, bitmap)
+ return merged
+
+ def _raw_pre_filter(self, splits):
+ if self._filter is None:
+ return None
+ raw_rows = _bitmap_of_ranges(_raw_row_ranges(splits))
+ if raw_rows.is_empty():
+ return None
+
+ seen = set()
+ scalar_files = []
+ for split in splits:
+ for index_file in split.scalar_index_files:
+ if index_file.file_name in seen:
+ continue
+ seen.add(index_file.file_name)
+ scalar_files.append(index_file)
if not scalar_files:
return None
from pypaimon.globalindex.global_index_scanner import
GlobalIndexScanner
- scanner = GlobalIndexScanner.create(self._table,
index_files=scalar_files)
+ scanner = GlobalIndexScanner.create(
+ self._table,
+ index_files=scalar_files,
+ partition_filter=self._partition_filter,
+ )
if scanner is None:
return None
try:
result = scanner.scan(self._filter)
if result is None:
return None
- return result.results()
+ include = result.results()
+ include = RoaringBitmap64.or_(
+ include,
+ scanner.unindexed_rows(self._filter).results())
+ return RoaringBitmap64.and_(include, raw_rows)
finally:
scanner.close()
@@ -136,28 +210,90 @@ class AbstractVectorSearchReadImpl:
future.add_done_callback(lambda _: reader.close())
return future
+ def _read_raw_search(self, raw_row_ranges, pre_filter, query_vector,
index_type=None):
+ raw_row_ranges = Range.sort_and_merge_overlap(raw_row_ranges, True)
+ if pre_filter is not None:
+ raw_row_ranges = Range.and_(
+ raw_row_ranges,
+ Range.sort_and_merge_overlap(pre_filter.to_range_list(), True),
+ )
+ if not raw_row_ranges:
+ return DictBasedScoredIndexResult({})
+
+ read_builder = self._table.new_read_builder()
+ if self._partition_filter is not None:
+ read_builder = read_builder.with_partition_filter(
+ self._partition_filter)
+ if self._filter is not None:
+ read_builder = read_builder.with_filter(self._filter)
+ from pypaimon.table.special_fields import SpecialFields
+ projection = [f.name for f in self._table.fields]
+ if SpecialFields.ROW_ID.name not in projection:
+ projection.append(SpecialFields.ROW_ID.name)
+ read_builder = read_builder.with_projection(projection)
+ plan = read_builder.new_scan().with_global_index_result(
+ GlobalIndexResult.from_ranges(raw_row_ranges)).plan()
+ table = read_builder.new_read().to_arrow(plan.splits())
+ if table is None or table.num_rows == 0:
+ return DictBasedScoredIndexResult({})
+
+ row_ids = table.column(SpecialFields.ROW_ID.name).to_pylist()
+ vectors = table.column(self._vector_column.name).to_pylist()
+ metric = _raw_search_metric(
+ self._table, self._vector_column, self._options, index_type)
+ scores = {}
+ for row_id, stored in zip(row_ids, vectors):
+ if stored is None:
+ continue
+ stored_vector = _to_vector_list(stored)
+ if len(stored_vector) != len(query_vector):
+ raise ValueError(
+ "Query vector dimension mismatch: expected %d, got %d"
+ % (len(stored_vector), len(query_vector)))
+ scores[row_id] = _compute_score(query_vector, stored_vector,
metric)
+ return DictBasedScoredIndexResult(scores).top_k(self._limit)
+
class VectorSearchReadImpl(AbstractVectorSearchReadImpl, VectorSearchRead):
"""Implementation for VectorSearchRead."""
def __init__(self, table, limit, vector_column, query_vector, filter_=None,
- options=None):
+ partition_filter=None, options=None):
super().__init__(table, limit, vector_column,
- filter_=filter_, options=options)
+ filter_=filter_,
+ partition_filter=partition_filter,
+ options=options)
self._query_vector = query_vector
def read(self, splits):
# type: (List[VectorSearchSplit]) -> GlobalIndexResult
- if not splits:
+ index_splits, raw_splits = _split_search_splits(splits)
+ if not index_splits and not raw_splits:
return GlobalIndexResult.create_empty()
- pre_filter = self._pre_filter(splits)
+ indexed = (
+ DictBasedScoredIndexResult({})
+ if not index_splits
+ else self._read_indexed(index_splits, self._query_vector)
+ )
+ raw_result = self._read_raw_search(
+ _raw_row_ranges(raw_splits),
+ self._raw_pre_filter(raw_splits),
+ self._query_vector,
+ _raw_search_index_type(raw_splits),
+ )
+ return indexed.or_(raw_result).top_k(self._limit)
+
+ def _read_indexed(self, splits, query_vector):
+ pre_filters = self._pre_filters(splits)
futures = [
self._eval(
split.row_range_start, split.row_range_end,
- split.vector_index_files, self._query_vector, pre_filter
+ split.vector_index_files,
+ query_vector,
+ None if not pre_filters else pre_filters[i]
)
- for split in splits
+ for i, split in enumerate(splits)
]
wait(futures)
@@ -179,25 +315,30 @@ class
BatchVectorSearchReadImpl(AbstractVectorSearchReadImpl,
"""Batch vector search read; result ``i`` corresponds to query vector
``i``."""
def __init__(self, table, limit, vector_column, query_vectors,
- filter_=None, options=None):
+ filter_=None, partition_filter=None, options=None):
super().__init__(table, limit, vector_column,
- filter_=filter_, options=options)
+ filter_=filter_,
+ partition_filter=partition_filter,
+ options=options)
self._query_vectors = list(query_vectors)
def read_batch(self, splits):
# type: (List[VectorSearchSplit]) -> List[GlobalIndexResult]
n = len(self._query_vectors)
- if not splits:
+ index_splits, raw_splits = _split_search_splits(splits)
+ if not index_splits and not raw_splits:
return [GlobalIndexResult.create_empty() for _ in range(n)]
- pre_filter = self._pre_filter(splits)
+ pre_filters = self._pre_filters(index_splits)
futures_by_vector = [
[
self._eval(
split.row_range_start, split.row_range_end,
- split.vector_index_files, vector, pre_filter
+ split.vector_index_files,
+ vector,
+ None if not pre_filters else pre_filters[i]
)
- for split in splits
+ for i, split in enumerate(index_splits)
]
for vector in self._query_vectors
]
@@ -206,6 +347,9 @@ class
BatchVectorSearchReadImpl(AbstractVectorSearchReadImpl,
wait(futures)
results = []
+ raw_pre_filter = self._raw_pre_filter(raw_splits)
+ raw_ranges = _raw_row_ranges(raw_splits)
+ raw_index_type = _raw_search_index_type(raw_splits)
for futures in futures_by_vector:
merged_scores = {}
for future in futures:
@@ -215,7 +359,11 @@ class
BatchVectorSearchReadImpl(AbstractVectorSearchReadImpl,
for row_id in split_result.results():
if row_id not in merged_scores:
merged_scores[row_id] = score_getter(row_id)
-
results.append(DictBasedScoredIndexResult(merged_scores).top_k(self._limit))
+ indexed = DictBasedScoredIndexResult(merged_scores)
+ vector = self._query_vectors[len(results)]
+ raw = self._read_raw_search(
+ raw_ranges, raw_pre_filter, vector, raw_index_type)
+ results.append(indexed.or_(raw).top_k(self._limit))
return results
@@ -238,3 +386,129 @@ def _create_vector_reader(index_type, file_io,
index_path, index_io_meta_list, o
file_io, index_path, index_io_meta_list, options
)
raise ValueError("Unsupported vector index type: '%s'" % index_type)
+
+
+def _split_search_splits(splits):
+ index_splits = []
+ raw_splits = []
+ for split in splits:
+ if isinstance(split, IndexVectorSearchSplit):
+ index_splits.append(split)
+ elif isinstance(split, RawVectorSearchSplit):
+ raw_splits.append(split)
+ return index_splits, raw_splits
+
+
+def _raw_row_ranges(raw_splits):
+ ranges = []
+ for split in raw_splits:
+ ranges.extend(split.row_ranges)
+ return Range.sort_and_merge_overlap(ranges, True)
+
+
+def _raw_search_index_type(raw_splits):
+ for split in raw_splits:
+ if split.index_type is not None:
+ return split.index_type
+ return None
+
+
+def _empty_bitmaps(size):
+ return [RoaringBitmap64() for _ in range(size)]
+
+
+def _bitmap_of_range(row_range):
+ bitmap = RoaringBitmap64()
+ bitmap.add_range(row_range.from_, row_range.to)
+ return bitmap
+
+
+def _bitmap_of_ranges(ranges):
+ bitmap = RoaringBitmap64()
+ for row_range in ranges:
+ bitmap.add_range(row_range.from_, row_range.to)
+ return bitmap
+
+
+def _to_vector_list(value):
+ if hasattr(value, "to_list"):
+ return value.to_list()
+ if hasattr(value, "as_py"):
+ value = value.as_py()
+ return list(value)
+
+
+def _raw_search_metric(table, vector_column, options, index_type=None):
+ candidates = []
+ field_prefix = "fields.%s." % vector_column.name
+ index_prefix = "%s." % index_type if index_type else None
+ for key in [
+ field_prefix + "distance.metric",
+ field_prefix + "metric",
+ *(([
+ index_prefix + "distance.metric",
+ index_prefix + "metric",
+ ]) if index_prefix is not None else []),
+ "test.vector.metric",
+ "lumina.distance.metric",
+ "distance.metric",
+ "metric",
+ ]:
+ if key in options:
+ candidates.append(options[key])
+ table_options = getattr(getattr(table, "options", None), "options", None)
+ table_map = table_options.to_map() if table_options is not None else {}
+ for key in [
+ field_prefix + "distance.metric",
+ field_prefix + "metric",
+ *(([
+ index_prefix + "distance.metric",
+ index_prefix + "metric",
+ ]) if index_prefix is not None else []),
+ "test.vector.metric",
+ "lumina.distance.metric",
+ "distance.metric",
+ "metric",
+ ]:
+ if key in table_map:
+ candidates.append(table_map[key])
+ if candidates:
+ return _normalize_metric(candidates[0])
+
+ inferred = None
+ for key, value in list(options.items()) + list(table_map.items()):
+ if key.endswith(".distance.metric") or key.endswith(".metric"):
+ metric = _normalize_metric(value)
+ if metric in ("l2", "cosine", "inner_product"):
+ if inferred is not None and inferred != metric:
+ return "l2"
+ inferred = metric
+ return inferred or "l2"
+
+
+def _normalize_metric(metric):
+ return str(metric).lower().replace("-", "_")
+
+
+def _compute_score(query, stored, metric):
+ if metric == "l2":
+ sum_sq = 0.0
+ for q, s in zip(query, stored):
+ diff = float(q) - float(s)
+ sum_sq += diff * diff
+ return 1.0 / (1.0 + sum_sq)
+ if metric == "cosine":
+ dot = 0.0
+ norm_a = 0.0
+ norm_b = 0.0
+ for q, s in zip(query, stored):
+ q = float(q)
+ s = float(s)
+ dot += q * s
+ norm_a += q * q
+ norm_b += s * s
+ denominator = (norm_a ** 0.5) * (norm_b ** 0.5)
+ return 0.0 if denominator == 0 else dot / denominator
+ if metric == "inner_product":
+ return sum(float(q) * float(s) for q, s in zip(query, stored))
+ raise ValueError("Unknown vector search metric: %s" % metric)
diff --git a/paimon-python/pypaimon/table/source/vector_search_scan.py
b/paimon-python/pypaimon/table/source/vector_search_scan.py
index 5b8b300d5c..1b6f9bc930 100644
--- a/paimon-python/pypaimon/table/source/vector_search_scan.py
+++ b/paimon-python/pypaimon/table/source/vector_search_scan.py
@@ -20,7 +20,12 @@
from abc import ABC, abstractmethod
from collections import defaultdict
-from pypaimon.table.source.vector_search_split import VectorSearchSplit
+from pypaimon.globalindex.global_index_coverage import GlobalIndexCoverage
+from pypaimon.table.source.vector_search_split import (
+ IndexVectorSearchSplit,
+ RawVectorSearchSplit,
+ VectorSearchSplit,
+)
from pypaimon.utils.range import Range
@@ -48,11 +53,19 @@ class VectorSearchScan(ABC):
class VectorSearchScanImpl(VectorSearchScan):
"""Implementation for VectorSearchScan."""
- def __init__(self, table, vector_column, filter_=None,
partition_filter=None):
+ def __init__(
+ self,
+ table,
+ vector_column,
+ filter_=None,
+ partition_filter=None,
+ options=None,
+ ):
self._table = table
self._vector_column = vector_column
self._filter = filter_
self._partition_filter = partition_filter
+ self._options = dict(options or {})
def scan(self):
# type: () -> VectorSearchScanPlan
@@ -86,6 +99,13 @@ class VectorSearchScanImpl(VectorSearchScan):
partition_filter = self._partition_filter
+ def contains_field(global_index_meta, field_id):
+ if global_index_meta.index_field_id == field_id:
+ return True
+ if global_index_meta.extra_field_ids is not None:
+ return field_id in global_index_meta.extra_field_ids
+ return False
+
def index_file_filter(entry):
if partition_filter is not None:
if not partition_filter.test(entry.partition):
@@ -96,13 +116,17 @@ class VectorSearchScanImpl(VectorSearchScan):
field_id = global_index_meta.index_field_id
if vector_column.id == field_id:
return True
- return field_id in filter_field_ids
+ for filter_field_id in filter_field_ids:
+ if contains_field(global_index_meta, filter_field_id):
+ return True
+ return False
entries = index_file_handler.scan(snapshot, index_file_filter)
all_index_files = [entry.index_file for entry in entries]
# Group vector index files by (rowRangeStart, rowRangeEnd).
vector_by_range = defaultdict(list)
+ vector_index_files = []
for index_file in all_index_files:
meta = index_file.global_index_meta
assert meta is not None
@@ -110,6 +134,12 @@ class VectorSearchScanImpl(VectorSearchScan):
continue
range_key = Range(meta.row_range_start, meta.row_range_end)
vector_by_range[range_key].append(index_file)
+ vector_index_files.append(index_file)
+
+ vector_index_type = _vector_index_type(vector_column, all_index_files)
+ if vector_index_type is None:
+ vector_index_type = _configured_vector_index_type(
+ self._table, vector_column, self._options)
# For each vector range, attach matching scalar index files whose
# row range intersects the vector range.
@@ -125,7 +155,7 @@ class VectorSearchScanImpl(VectorSearchScan):
if range_key.overlaps(scalar_range):
scalar_files.append(index_file)
splits.append(
- VectorSearchSplit(
+ IndexVectorSearchSplit(
range_key.from_,
range_key.to,
vector_files,
@@ -133,4 +163,89 @@ class VectorSearchScanImpl(VectorSearchScan):
)
)
+ raw_row_ranges = GlobalIndexCoverage(
+ self._table,
+ snapshot,
+ partition_filter,
+ vector_index_files,
+ ).unindexed_ranges(vector_column.id)
+ 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:
+ raw_row_ranges = Range.sort_and_merge_overlap(
+ raw_row_ranges
+ + GlobalIndexCoverage(
+ self._table,
+ snapshot,
+ partition_filter,
+ scalar_index_files,
+ ).unindexed_ranges(self._table.fields, self._filter),
+ True,
+ )
+ if raw_row_ranges:
+ splits.append(
+ RawVectorSearchSplit(
+ raw_row_ranges,
+ _scalar_index_files_for_ranges(
+ all_index_files,
+ raw_row_ranges,
+ vector_column.id,
+ ),
+ vector_index_type,
+ )
+ )
+
return VectorSearchScanPlan(splits)
+
+
+def _has_intersection(ranges, row_range):
+ for r in ranges:
+ if r.overlaps(row_range):
+ return True
+ return False
+
+
+def _scalar_index_files_for_ranges(all_index_files, row_ranges,
vector_field_id):
+ scalar_files = []
+ for index_file in all_index_files:
+ meta = index_file.global_index_meta
+ if meta is None or meta.index_field_id == vector_field_id:
+ continue
+ if _has_intersection(row_ranges, Range(meta.row_range_start,
meta.row_range_end)):
+ scalar_files.append(index_file)
+ return scalar_files
+
+
+def _vector_index_type(vector_column, index_files):
+ index_type = None
+ for index_file in index_files:
+ meta = index_file.global_index_meta
+ if meta is None or meta.index_field_id != vector_column.id:
+ continue
+ if index_type is None:
+ index_type = index_file.index_type
+ elif index_type != index_file.index_type:
+ raise ValueError(
+ "Vector column '%s' has multiple index types: %s and %s."
+ % (vector_column.name, index_type, index_file.index_type)
+ )
+ return index_type
+
+
+def _configured_vector_index_type(table, vector_column, options):
+ keys = [
+ "index_type",
+ "index-type",
+ "vector.index-type",
+ "fields.%s.index-type" % vector_column.name,
+ ]
+ table_options = getattr(getattr(table, "options", None), "options", None)
+ table_map = table_options.to_map() if table_options is not None else {}
+ for key in keys:
+ value = options.get(key) or table_map.get(key)
+ if value is not None:
+ return str(value).lower().strip()
+ return None
diff --git a/paimon-python/pypaimon/table/source/vector_search_split.py
b/paimon-python/pypaimon/table/source/vector_search_split.py
index f8279d5c20..a8f2428688 100644
--- a/paimon-python/pypaimon/table/source/vector_search_split.py
+++ b/paimon-python/pypaimon/table/source/vector_search_split.py
@@ -15,19 +15,55 @@
# specific language governing permissions and limitations
# under the License.
-"""Split of vector search."""
+"""Splits of vector search."""
from dataclasses import dataclass, field
-from typing import List
+from typing import List, Optional
from pypaimon.index.index_file_meta import IndexFileMeta
+from pypaimon.utils.range import Range
-@dataclass
class VectorSearchSplit:
- """Split of vector search."""
+ """Base split of vector search.
+
+ The constructor keeps the historical ``VectorSearchSplit(...)`` shape as
+ an alias for an indexed split so existing Python callers/tests continue to
+ work while new code can distinguish split implementations with
+ ``isinstance``.
+ """
+
+ def __new__(
+ cls,
+ row_range_start=None,
+ row_range_end=None,
+ vector_index_files=None,
+ scalar_index_files=None,
+ ):
+ if cls is VectorSearchSplit:
+ return IndexVectorSearchSplit(
+ row_range_start,
+ row_range_end,
+ vector_index_files,
+ scalar_index_files,
+ )
+ return object.__new__(cls)
+
+
+@dataclass
+class IndexVectorSearchSplit(VectorSearchSplit):
+ """Split to read vector index files."""
row_range_start: int
row_range_end: int
vector_index_files: List[IndexFileMeta]
scalar_index_files: List[IndexFileMeta] = field(default_factory=list)
+
+
+@dataclass
+class RawVectorSearchSplit(VectorSearchSplit):
+ """Split to scan raw vectors."""
+
+ row_ranges: List[Range]
+ scalar_index_files: List[IndexFileMeta] = field(default_factory=list)
+ index_type: Optional[str] = None
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 f940f38a69..691bd16dd9 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
@@ -448,6 +448,27 @@ class JavaPyReadWriteTest(unittest.TestCase):
if sys.version_info[:2] >= (3, 7):
self._test_index_manifest_inherited_after_write()
+ def test_read_btree_raw_fallback(self):
+ table = self.catalog.get_table('default.test_btree_raw_fallback')
+ fast_builder = 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.with_filter(
+ read_builder.new_predicate_builder().equal('k', 'k4'))
+ actual = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ expected = pa.Table.from_pydict({
+ 'k': ['k4'],
+ 'v': ['v4'],
+ })
+ self.assertEqual(expected, actual)
+
def _test_read_btree_index_generic(self, table_name: str, k, k_type):
table = self.catalog.get_table('default.' + table_name)
read_builder: ReadBuilder = table.new_read_builder()
@@ -1253,6 +1274,47 @@ class JavaPyReadWriteTest(unittest.TestCase):
print(f"paimon-vindex vector search matched rows: ids={ids}")
self.assertIn(0, ids)
+ def test_read_vindex_vector_raw_fallback(self):
+ """Test raw fallback for a paimon-vindex vector index built by Java."""
+ if sys.version_info < (3, 9):
+ self.skipTest("paimon-vindex requires Python >= 3.9")
+ try:
+ import paimon_vindex # noqa: F401
+ except ImportError:
+ self.skipTest("paimon-vindex is not installed")
+
+ table = self.catalog.get_table(
+ 'default.test_vindex_vector_raw_fallback')
+ fast_result = (table.new_vector_search_builder()
+ .with_vector_column('embedding')
+ .with_query_vector([1.0, 0.0, 0.0, 0.0])
+ .with_limit(1)
+ .execute_local())
+ fast_ids = sorted(list(fast_result.results()))
+ print(
+ "paimon-vindex fast-mode vector search matched rows: "
+ f"ids={fast_ids}")
+ self.assertNotIn(3, fast_ids)
+
+ full_table = table.copy({'global-index.search-mode': 'full'})
+ full_result = (full_table.new_vector_search_builder()
+ .with_vector_column('embedding')
+ .with_query_vector([1.0, 0.0, 0.0, 0.0])
+ .with_limit(1)
+ .execute_local())
+ row_ids = sorted(list(full_result.results()))
+ print(
+ "paimon-vindex full-mode vector search matched rows: "
+ f"ids={row_ids}")
+ self.assertEqual([3], row_ids)
+
+ read_builder = full_table.new_read_builder()
+ scan = read_builder.new_scan().with_global_index_result(full_result)
+ table_read = read_builder.new_read()
+ pa_table = table_read.to_arrow(scan.plan().splits())
+ self.assertEqual(pa_table.num_rows, 1)
+ self.assertEqual([3], pa_table.column('id').to_pylist())
+
def test_read_lumina_vector_with_btree_filter(self):
"""Vector search + btree scalar pre-filter, using a table that Java
populated with both a Lumina vector index on `embedding` and a BTree
diff --git a/paimon-python/pypaimon/tests/global_index_test.py
b/paimon-python/pypaimon/tests/global_index_test.py
index 873e933584..60ca559cda 100644
--- a/paimon-python/pypaimon/tests/global_index_test.py
+++ b/paimon-python/pypaimon/tests/global_index_test.py
@@ -20,8 +20,15 @@ from unittest.mock import patch
import pyarrow as pa
+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.global_index_meta import GlobalIndexMeta
from pypaimon.globalindex.global_index_result import GlobalIndexResult
+from pypaimon.index.index_file_meta import IndexFileMeta
from pypaimon.index.index_file_handler import IndexFileHandler
+from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.snapshot.snapshot_manager import SnapshotManager
from pypaimon.tests.data_evolution_test_helpers import (
BatchModeMixin,
@@ -49,6 +56,193 @@ class GlobalIndexTest(unittest.TestCase):
self.assertEqual(result.results().cardinality(), 10001)
+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()
+
+
+class _CoverageTable:
+ def __init__(self, mode, data_ranges=None):
+ self.options = _CoverageOptions(mode)
+ self.fields = [
+ DataField(0, "id", AtomicType("INT")),
+ DataField(1, "name", AtomicType("STRING")),
+ ]
+ self._data_ranges = data_ranges or []
+
+ def data_ranges_for_global_index_coverage(self, snapshot,
partition_filter):
+ return self._data_ranges
+
+
+class _CoverageSnapshot:
+ def __init__(self, next_row_id):
+ self.next_row_id = next_row_id
+
+
+def _coverage_index_file(field_id, start, end, extra_field_ids=None):
+ return IndexFileMeta(
+ index_type="btree",
+ file_name="idx-%s-%s-%s" % (field_id, start, end),
+ file_size=1,
+ row_count=end - start + 1,
+ global_index_meta=GlobalIndexMeta(
+ row_range_start=start,
+ row_range_end=end,
+ index_field_id=field_id,
+ extra_field_ids=extra_field_ids,
+ index_meta=b"",
+ ),
+ )
+
+
+class GlobalIndexCoverageTest(unittest.TestCase):
+
+ def test_fast_mode_does_not_return_unindexed_ranges(self):
+ from pypaimon.globalindex.global_index_coverage import
GlobalIndexCoverage
+
+ table = _CoverageTable(GlobalIndexSearchMode.FAST)
+ coverage = GlobalIndexCoverage(
+ table,
+ _CoverageSnapshot(10),
+ None,
+ [_coverage_index_file(0, 0, 4)],
+ )
+
+ self.assertEqual([], coverage.unindexed_ranges(0))
+
+ def test_full_mode_uses_snapshot_next_row_id(self):
+ from pypaimon.globalindex.global_index_coverage import
GlobalIndexCoverage
+
+ table = _CoverageTable("full")
+ coverage = GlobalIndexCoverage(
+ table,
+ _CoverageSnapshot(10),
+ None,
+ [_coverage_index_file(0, 0, 4)],
+ )
+
+ self.assertEqual([Range(5, 9)], coverage.unindexed_ranges(0))
+
+ def test_full_mode_intersects_coverage_for_all_predicate_fields(self):
+ from pypaimon.globalindex.global_index_coverage import
GlobalIndexCoverage
+
+ table = _CoverageTable("full")
+ coverage = GlobalIndexCoverage(
+ table,
+ _CoverageSnapshot(10),
+ None,
+ [
+ _coverage_index_file(0, 0, 9),
+ _coverage_index_file(1, 0, 4),
+ ],
+ )
+ predicate = PredicateBuilder.and_predicates(
+ [
+ Predicate(method="equal", index=0, field="id", literals=[1]),
+ Predicate(method="equal", index=1, field="name",
literals=["a"]),
+ ]
+ )
+
+ self.assertEqual(
+ [Range(5, 9)],
+ coverage.unindexed_ranges(table.fields, predicate),
+ )
+
+ def test_extra_fields_count_as_index_coverage(self):
+ from pypaimon.globalindex.global_index_coverage import
GlobalIndexCoverage
+
+ table = _CoverageTable("full")
+ coverage = GlobalIndexCoverage(
+ table,
+ _CoverageSnapshot(10),
+ None,
+ [_coverage_index_file(0, 0, 9, extra_field_ids=[1])],
+ )
+
+ self.assertEqual([], coverage.unindexed_ranges(1))
+
+ def test_detail_mode_uses_table_data_ranges(self):
+ from pypaimon.globalindex.global_index_coverage import
GlobalIndexCoverage
+
+ table = _CoverageTable("detail", data_ranges=[Range(0, 2), Range(7,
9)])
+ coverage = GlobalIndexCoverage(
+ table,
+ _CoverageSnapshot(10),
+ None,
+ [_coverage_index_file(0, 0, 4)],
+ )
+
+ self.assertEqual([Range(7, 9)], coverage.unindexed_ranges(0))
+
+
+class GlobalIndexScalarFallbackTest(unittest.TestCase):
+
+ def
test_eval_global_index_merges_unindexed_rows_when_index_scan_succeeds(self):
+ from pypaimon.read.scanner.file_scanner import FileScanner
+
+ class _Options:
+ def global_index_enabled(self):
+ return True
+
+ class _Table:
+ options = _Options()
+
+ predicate = Predicate(method="equal", index=0, field="id",
literals=[1])
+ scanner = FileScanner.__new__(FileScanner)
+ scanner.predicate = predicate
+ scanner.partition_key_predicate = None
+ scanner.table = _Table()
+
+ index_result = GlobalIndexResult.from_range(Range(1, 1))
+ unindexed = GlobalIndexResult.from_range(Range(5, 6))
+ fake_scanner = unittest.mock.MagicMock()
+ fake_scanner.scan.return_value = index_result
+ fake_scanner.unindexed_rows.return_value = unindexed
+ fake_scanner.__enter__.return_value = fake_scanner
+ fake_scanner.__exit__.return_value = None
+
+ with unittest.mock.patch(
+
"pypaimon.globalindex.global_index_scanner.GlobalIndexScanner.create",
+ return_value=fake_scanner):
+ result = scanner._eval_global_index(snapshot=object())
+
+ self.assertEqual(
+ [Range(1, 1), Range(5, 6)],
+ result.results().to_range_list(),
+ )
+
+ def test_eval_global_index_keeps_none_as_full_scan(self):
+ from pypaimon.read.scanner.file_scanner import FileScanner
+
+ class _Options:
+ def global_index_enabled(self):
+ return True
+
+ class _Table:
+ options = _Options()
+
+ scanner = FileScanner.__new__(FileScanner)
+ scanner.predicate = Predicate(
+ method="equal", index=0, field="id", literals=[1])
+ scanner.partition_key_predicate = None
+ scanner.table = _Table()
+
+ fake_scanner = unittest.mock.MagicMock()
+ fake_scanner.scan.return_value = None
+ fake_scanner.__enter__.return_value = fake_scanner
+ fake_scanner.__exit__.return_value = None
+
+ with unittest.mock.patch(
+
"pypaimon.globalindex.global_index_scanner.GlobalIndexScanner.create",
+ return_value=fake_scanner):
+ result = scanner._eval_global_index(snapshot=object())
+
+ self.assertIsNone(result)
+
+
class PlanSnapshotFetchRegressionTest(
BatchModeMixin, DataEvolutionTestBase, unittest.TestCase):
diff --git a/paimon-python/pypaimon/tests/vector_search_filter_test.py
b/paimon-python/pypaimon/tests/vector_search_filter_test.py
index f2b453346e..fe63890a42 100644
--- a/paimon-python/pypaimon/tests/vector_search_filter_test.py
+++ b/paimon-python/pypaimon/tests/vector_search_filter_test.py
@@ -44,6 +44,7 @@ from pypaimon.schema.data_types import AtomicType, DataField
from pypaimon.table.row.generic_row import GenericRow
from pypaimon.table.source.vector_search_builder import VectorSearchBuilderImpl
from pypaimon.utils.roaring_bitmap import RoaringBitmap64
+from pypaimon.utils.range import Range
# ----------------------------- table stubs ---------------------------------
@@ -122,6 +123,15 @@ def _entry(partition_row, field_id, index_type, file_name,
def _patch_snapshot(testcase, entries):
"""Stub IndexFileHandler.scan + snapshot resolution."""
+ mock.patch.stopall()
+ for attr in ("_scan_patch", "_travel_patch"):
+ patcher = getattr(testcase, attr, None)
+ if patcher is not None:
+ try:
+ patcher.stop()
+ except RuntimeError:
+ pass
+
def _scan(snapshot, entry_filter=None):
if entry_filter is None:
return list(entries)
@@ -1005,6 +1015,174 @@ class VectorSearchFilterTest(unittest.TestCase):
self.assertEqual("oss://bucket/id-btree-0.index",
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.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.assertEqual(2, len(splits))
+ self.assertTrue(any(isinstance(s, IndexVectorSearchSplit)
+ for s in splits))
+ raw = [s for s in splits if isinstance(s, RawVectorSearchSplit)]
+ 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
+ from pypaimon.table.source.vector_search_split import
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)
+
+ # Vector index covers all rows, but there is no scalar id index, so
+ # full mode must produce a raw split for the scalar-filtered path.
+ table = _StubTable(
+ fields=[self.id_field, self.embedding_field],
+ entries=[
+ _entry(None, field_id=1,
+ index_type="lumina-vector-ann",
+ file_name="vec-all.index",
+ row_range_start=0,
+ row_range_end=9)
+ ],
+ )
+ table.options = _Options()
+ table.snapshot_manager = lambda: _Snapshots()
+ self._scan_patch.stop()
+ self._travel_patch.stop()
+ _patch_snapshot(self, table._entries)
+ self._travel_patch.stop()
+ 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 = [s for s in splits if isinstance(s, RawVectorSearchSplit)]
+ 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
+ from pypaimon.table.source.vector_search_split import
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=[])
+ table.options = _Options()
+ table.snapshot_manager = lambda: _Snapshots()
+ self._scan_patch.stop()
+ self._travel_patch.stop()
+ _patch_snapshot(self, [])
+ 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)
+ .with_option("index-type", "ivf-flat")
+ .new_vector_search_scan()
+ .scan()
+ .splits()
+ )
+
+ raw = [s for s in splits if isinstance(s, RawVectorSearchSplit)]
+ self.assertEqual(1, len(raw))
+ self.assertEqual("ivf-flat", raw[0].index_type)
+
+ def test_scan_attaches_scalar_index_when_filter_hits_extra_field(self):
+ id_name_index = _entry(None, field_id=2, index_type="btree",
+ file_name="name-id.index",
+ row_range_start=0,
+ row_range_end=9)
+ id_name_index.index_file.global_index_meta.extra_field_ids = [0]
+ table = _StubTable(fields=[
+ self.id_field,
+ self.embedding_field,
+ _field(2, "name", "STRING"),
+ ], entries=[
+ self.entries[0],
+ id_name_index,
+ ])
+ self._scan_patch.stop()
+ self._travel_patch.stop()
+ _patch_snapshot(self, table._entries)
+ filter_pred = Predicate(method="equal", 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()
+ )
+
+ self.assertEqual(["name-id.index"],
+ [f.file_name for f in splits[0].scalar_index_files])
+
class VectorSearchMultiShardScalarTest(unittest.TestCase):
"""Scalar pre-filter across multiple btree shards of the same field.
@@ -1016,6 +1194,9 @@ class VectorSearchMultiShardScalarTest(unittest.TestCase):
- An empty first shard does NOT short-circuit subsequent shards.
"""
+ def tearDown(self):
+ mock.patch.stopall()
+
def test_hit_only_in_later_shard_returns_global_row_id(self):
from pypaimon.globalindex.global_index_result import GlobalIndexResult
from pypaimon.globalindex.global_index_scanner import (
@@ -1202,6 +1383,95 @@ class
VectorSearchMultiShardScalarTest(unittest.TestCase):
self.assertIsNotNone(result)
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.global_index_scanner import (
+ GlobalIndexScanner,
+ )
+
+ 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)
+
+ id_field = _field(0, "id")
+ emb_field = _field(1, "embedding", "FLOAT")
+ indexed = _entry(None, field_id=0, index_type="btree",
+ 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.snapshot_manager = lambda: _Snapshots()
+
+ scanner = GlobalIndexScanner.create(table, index_files=[indexed])
+ try:
+ result = scanner.unindexed_rows(
+ Predicate(method="equal", index=0, field="id", literals=[7]))
+ finally:
+ scanner.close()
+
+ self.assertEqual([Range(5, 9)], result.results().to_range_list())
+
+ def test_scanner_create_selects_extra_field_indexes(self):
+ from pypaimon.globalindex.global_index_scanner import (
+ GlobalIndexScanner,
+ )
+
+ name_field = _field(0, "name", "STRING")
+ id_field = _field(1, "id")
+ emb_field = _field(2, "embedding", "FLOAT")
+ multi = _entry(None, field_id=0, index_type="btree",
+ file_name="name-id.index",
+ row_range_start=0, row_range_end=9).index_file
+ multi.global_index_meta.extra_field_ids = [1]
+ table = _StubTable(
+ fields=[name_field, id_field, emb_field],
+ entries=[
+ IndexManifestEntry(kind=0, partition=None,
+ bucket=0, index_file=multi)
+ ],
+ )
+ _patch_snapshot(self, table._entries)
+
+ class _StubBTreeReader:
+ def __init__(self_inner, key_serializer, file_io, index_path,
+ io_meta):
+ pass
+
+ def visit_equal(self_inner, literal):
+ return GlobalIndexResult.create_empty()
+
+ def close(self_inner):
+ pass
+
+ with mock.patch(
+
"pypaimon.globalindex.btree.lazy_filtered_btree_reader.BTreeIndexReader",
+ _StubBTreeReader):
+ with mock.patch(
+
"pypaimon.globalindex.sorted_file_global_index_reader.SortedIndexFileMeta.deserialize",
+ return_value=BTreeIndexMeta(first_key=b'',
last_key=b'zzzz', has_nulls=False)):
+ scanner = GlobalIndexScanner.create(
+ table,
+ predicate=Predicate(method="equal", index=1, field="id",
+ literals=[3]),
+ )
+ try:
+ self.assertIsNotNone(scanner)
+ readers = scanner._evaluator._readers_function(id_field)
+ self.assertTrue(readers)
+ finally:
+ if scanner is not None:
+ scanner.close()
+
class VectorSearchPartitionedFilterTest(unittest.TestCase):
"""Partitioned-table paths: with_filter auto-split + partition-filter
@@ -1489,6 +1759,159 @@ class VectorSearchManySplitsTest(unittest.TestCase):
scores = sorted(result.score_getter()(rid) for rid in result.results())
self.assertEqual(scores, [float(i) for i in range(1190, 1200)])
+ def test_read_merges_raw_search_results(self):
+ from pypaimon.globalindex.vector_search_result import (
+ DictBasedScoredIndexResult,
+ )
+ from pypaimon.table.source.vector_search_read import
VectorSearchReadImpl
+ from pypaimon.table.source.vector_search_split import (
+ IndexVectorSearchSplit,
+ RawVectorSearchSplit,
+ )
+
+ embedding_field = _field(1, "embedding", "FLOAT")
+ entry = _entry(None, field_id=1, index_type="lumina-vector-ann",
+ file_name="vec.index",
+ row_range_start=0, row_range_end=4)
+ table = _StubTable(fields=[embedding_field], entries=[entry])
+
+ def _fake_create(index_type, file_io, index_path,
+ index_io_meta_list, options=None):
+ class _FakeReader:
+ def visit_vector_search(self_inner, vs):
+ return _completed_future(
+ DictBasedScoredIndexResult({1: 0.1}))
+
+ def close(self_inner):
+ pass
+
+ return _FakeReader()
+
+ split = IndexVectorSearchSplit(
+ row_range_start=0,
+ row_range_end=4,
+ vector_index_files=[entry.index_file],
+ )
+ raw = RawVectorSearchSplit([Range(5, 9)], [], "lumina-vector-ann")
+
+ with mock.patch(
+
"pypaimon.table.source.vector_search_read._create_vector_reader",
+ side_effect=_fake_create):
+ reader = VectorSearchReadImpl(
+ table, limit=2, vector_column=embedding_field,
+ query_vector=[1.0], filter_=None)
+ with mock.patch.object(
+ reader,
+ "_read_raw_search",
+ return_value=DictBasedScoredIndexResult({8: 0.9})) as
raw_read:
+ result = reader.read([split, raw])
+
+ raw_read.assert_called_once()
+ self.assertEqual([1, 8], sorted(list(result.results())))
+
+ def test_read_uses_empty_index_prefilter_when_scalar_index_missing(self):
+ from pypaimon.table.source.vector_search_read import
VectorSearchReadImpl
+ from pypaimon.table.source.vector_search_split import
IndexVectorSearchSplit
+
+ id_field = _field(0, "id")
+ embedding_field = _field(1, "embedding", "FLOAT")
+ entry = _entry(None, field_id=1, index_type="lumina-vector-ann",
+ file_name="vec.index",
+ row_range_start=0, row_range_end=4)
+ table = _StubTable(fields=[id_field, embedding_field], entries=[entry])
+ filter_pred = Predicate(method="equal", index=0, field="id",
+ literals=[3])
+ split = IndexVectorSearchSplit(
+ row_range_start=0,
+ row_range_end=4,
+ vector_index_files=[entry.index_file],
+ scalar_index_files=[],
+ )
+
+ reader = VectorSearchReadImpl(
+ table, limit=2, vector_column=embedding_field,
+ query_vector=[1.0], filter_=filter_pred)
+
+ pre_filters = reader._pre_filters([split])
+
+ self.assertEqual(1, len(pre_filters))
+ self.assertEqual(0, pre_filters[0].cardinality())
+
+ def test_raw_search_uses_partition_filter_and_index_type_metric(self):
+ import pyarrow as pa
+
+ from pypaimon.table.source.vector_search_read import
VectorSearchReadImpl
+
+ id_field = _field(0, "id")
+ embedding_field = _field(1, "embedding", "FLOAT")
+ table = _StubTable(fields=[id_field, embedding_field], entries=[])
+ partition_filter = Predicate(method="equal", index=0, field="pt",
+ literals=[1])
+ filter_pred = Predicate(method="greaterOrEqual", index=0, field="id",
+ literals=[0])
+ calls = {}
+
+ class _Plan:
+ def splits(self_inner):
+ return ["split"]
+
+ class _Scan:
+ def with_global_index_result(self_inner, result):
+ calls["global_index_ranges"] = result.results().to_range_list()
+ return self_inner
+
+ def plan(self_inner):
+ return _Plan()
+
+ class _Read:
+ def to_arrow(self_inner, splits):
+ calls["splits"] = list(splits)
+ return pa.table({
+ "id": pa.array([5, 6], type=pa.int32()),
+ "embedding": pa.array([[1.0], [0.0]]),
+ "_ROW_ID": pa.array([5, 6], type=pa.int64()),
+ })
+
+ class _Builder:
+ def with_partition_filter(self_inner, predicate):
+ calls["partition_filter"] = predicate
+ return self_inner
+
+ def with_filter(self_inner, predicate):
+ calls["filter"] = predicate
+ return self_inner
+
+ def with_projection(self_inner, projection):
+ calls["projection"] = list(projection)
+ return self_inner
+
+ def new_scan(self_inner):
+ return _Scan()
+
+ def new_read(self_inner):
+ return _Read()
+
+ table.new_read_builder = lambda: _Builder()
+ reader = VectorSearchReadImpl(
+ table,
+ limit=1,
+ vector_column=embedding_field,
+ query_vector=[1.0],
+ filter_=filter_pred,
+ partition_filter=partition_filter,
+ options={"ivf-flat.metric": "inner_product"},
+ )
+
+ result = reader._read_raw_search(
+ [Range(5, 6)], None, [1.0], "ivf-flat")
+
+ self.assertIs(partition_filter, calls["partition_filter"])
+ self.assertIs(filter_pred, calls["filter"])
+ self.assertEqual([Range(5, 6)], calls["global_index_ranges"])
+ self.assertIn("_ROW_ID", calls["projection"])
+ self.assertEqual(["split"], calls["splits"])
+ self.assertEqual([5], sorted(list(result.results())))
+
def tearDown(self):
mock.patch.stopall()
diff --git a/paimon-vector/src/test/java/org/apache/paimon/JavaPyE2ETest.java
b/paimon-vector/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index 033245f9eb..4c8cbfd59c 100644
--- a/paimon-vector/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-vector/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -197,4 +197,121 @@ public class JavaPyE2ETest {
assertThat(indexEntries.get(0).indexFile().indexType())
.isEqualTo(IvfFlatVectorGlobalIndexerFactory.IDENTIFIER);
}
+
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testVindexVectorRawFallbackWrite() throws Exception {
+ String tableName = "test_vindex_vector_raw_fallback";
+ Path tablePath = new Path(warehouse.toString() + "/default.db/" +
tableName);
+ LocalFileIO fileIO = LocalFileIO.create();
+ if (fileIO.exists(tablePath)) {
+ fileIO.delete(tablePath, true);
+ }
+
+ int dimension = 4;
+
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), new ArrayType(new
FloatType())},
+ new String[] {"id", "embedding"});
+
+ Options options = new Options();
+ options.set(PATH, tablePath.toString());
+ options.set(ROW_TRACKING_ENABLED, true);
+ options.set(DATA_EVOLUTION_ENABLED, true);
+ options.set(GLOBAL_INDEX_ENABLED, true);
+ options.setString(
+ IvfFlatVectorGlobalIndexerFactory.IDENTIFIER + ".dimension",
+ String.valueOf(dimension));
+ options.setString(IvfFlatVectorGlobalIndexerFactory.IDENTIFIER +
".metric", "l2");
+ options.setString(IvfFlatVectorGlobalIndexerFactory.IDENTIFIER +
".nlist", "2");
+
+ TableSchema tableSchema =
+ SchemaUtils.forceCommit(
+ new SchemaManager(fileIO, tablePath),
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ options.toMap(),
+ ""));
+
+ AppendOnlyFileStoreTable table =
+ new AppendOnlyFileStoreTable(
+ FileIOFinder.find(tablePath),
+ tablePath,
+ tableSchema,
+ CatalogEnvironment.empty());
+
+ float[][] indexedVectors =
+ new float[][] {
+ new float[] {0.0f, 1.0f, 0.0f, 0.0f},
+ new float[] {0.0f, 0.0f, 1.0f, 0.0f},
+ new float[] {0.0f, 0.0f, 0.0f, 1.0f}
+ };
+
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ for (int i = 0; i < indexedVectors.length; i++) {
+ write.write(GenericRow.of(i, new
GenericArray(indexedVectors[i])));
+ }
+ commit.commit(write.prepareCommit());
+ }
+
+ DataField embeddingField = table.rowType().getField("embedding");
+ Options indexOptions = table.coreOptions().toConfiguration();
+
+ GlobalIndexSingleColumnWriter writer =
+ (GlobalIndexSingleColumnWriter)
+ GlobalIndexBuilderUtils.createIndexWriter(
+ table,
+ IvfFlatVectorGlobalIndexerFactory.IDENTIFIER,
+ embeddingField,
+ indexOptions);
+
+ for (int i = 0; i < indexedVectors.length; i++) {
+ writer.write(indexedVectors[i], i);
+ }
+
+ List<ResultEntry> entries = writer.finish();
+ assertThat(entries).hasSize(1);
+ assertThat(entries.get(0).rowCount()).isEqualTo(indexedVectors.length);
+
+ Range rowRange = new Range(0, indexedVectors.length - 1);
+ List<IndexFileMeta> indexFiles =
+ GlobalIndexBuilderUtils.toIndexFileMetas(
+ table.fileIO(),
+ table.store().pathFactory().globalIndexFileFactory(),
+ table.coreOptions(),
+ rowRange,
+ embeddingField.id(),
+ IvfFlatVectorGlobalIndexerFactory.IDENTIFIER,
+ entries);
+
+ DataIncrement dataIncrement = DataIncrement.indexIncrement(indexFiles);
+ CommitMessage message =
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ null,
+ dataIncrement,
+ CompactIncrement.emptyIncrement());
+ try (BatchTableCommit commit = writeBuilder.newCommit()) {
+ commit.commit(Collections.singletonList(message));
+ }
+
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(GenericRow.of(3, new GenericArray(new float[] {1.0f,
0.0f, 0.0f, 0.0f})));
+ commit.commit(write.prepareCommit());
+ }
+
+ List<org.apache.paimon.manifest.IndexManifestEntry> indexEntries =
+
table.indexManifestFileReader().read(table.latestSnapshot().get().indexManifest());
+ assertThat(indexEntries).hasSize(1);
+
assertThat(indexEntries.get(0).indexFile().rowCount()).isEqualTo(indexedVectors.length);
+ assertThat(indexEntries.get(0).indexFile().indexType())
+ .isEqualTo(IvfFlatVectorGlobalIndexerFactory.IDENTIFIER);
+ }
}
