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 71148d053f [python] Support vector index builds on deletion-vector
tables (#10157)
71148d053f is described below
commit 71148d053f23d3853f3734f805ea43828250c367
Author: chaoyang <[email protected]>
AuthorDate: Thu Sep 24 17:43:44 2026 +0800
[python] Support vector index builds on deletion-vector tables (#10157)
---
docs/docs/multimodal-table/global-index/vector.mdx | 29 ++++
paimon-python/pypaimon/globalindex/build_plan.py | 3 +
.../pypaimon/globalindex/create_global_index.py | 18 +-
.../pypaimon/tests/global_index_build_test.py | 4 +-
.../pypaimon/tests/ray_vector_search_test.py | 4 +-
.../pypaimon/tests/vector_filter_exactness_test.py | 2 +-
.../pypaimon/tests/vector_index_dv_build_test.py | 184 +++++++++++++++++++++
7 files changed, 231 insertions(+), 13 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index/vector.mdx
b/docs/docs/multimodal-table/global-index/vector.mdx
index c191389b22..0f52f5fee2 100644
--- a/docs/docs/multimodal-table/global-index/vector.mdx
+++ b/docs/docs/multimodal-table/global-index/vector.mdx
@@ -134,6 +134,35 @@ For `ARRAY<FLOAT>` columns, specify the dimension with
`<index-type>.dimension`.
For `VECTOR<FLOAT, n>` columns, Paimon uses the dimension from the column type.
Every query vector must have that same dimension.
+### Python Tables with Deletion Vectors
+
+Python can build the native vector indexes listed above on data-evolution
tables
+with deletion vectors enabled, including tables created with the multimodal
API's
+default options:
+
+```python
+import pyarrow as pa
+import pypaimon.multimodal as pm
+
+connection = pm.connect(options={"warehouse": "/tmp/paimon-warehouse"})
+table = connection.create_table(
+ "vectors",
+ schema=pa.schema([("id", pa.int64()), ("embedding", pa.list_(pa.float32(),
2))]),
+ options={"file.format": "parquet"},
+)
+table.add([{"id": 1, "embedding": [0.0, 1.0]}, {"id": 2, "embedding": [1.0,
1.0]}])
+table.create_index("embedding", "ivf-flat", options={
+ "ivf-flat.nlist": "1", "ivf-flat.distance.metric": "l2",
+})
+table.delete("id = 1")
+print(table.search([0.0, 1.0]).select(["id"]).limit(1).to_list()) # [{"id":
2}]
+```
+
+Index builds preserve physical row IDs, including rows already marked as
deleted.
+Search applies the deletion vectors from its query snapshot before selecting
+top-K results, so logical deletes do not require rebuilding the index. Logical
+deletion does not physically remove vectors from existing index files.
+
## Vector Search
Python indexed vector searches use the `global-index.thread-num` table option,
diff --git a/paimon-python/pypaimon/globalindex/build_plan.py
b/paimon-python/pypaimon/globalindex/build_plan.py
index 74893d16b2..f4e33dd237 100644
--- a/paimon-python/pypaimon/globalindex/build_plan.py
+++ b/paimon-python/pypaimon/globalindex/build_plan.py
@@ -268,6 +268,9 @@ def append_shard_split(
if not task_ranges:
return
+ # Index the physical row-ID range, including DV-deleted rows, as in Java.
+ # Query-side live-row filtering applies DVs for the pinned query snapshot.
+ # Passing deletion files here would change the source coverage seen by
writers.
data_split = DataSplit(
files=list(files),
partition=partition,
diff --git a/paimon-python/pypaimon/globalindex/create_global_index.py
b/paimon-python/pypaimon/globalindex/create_global_index.py
index 09b8a568f5..5926101148 100644
--- a/paimon-python/pypaimon/globalindex/create_global_index.py
+++ b/paimon-python/pypaimon/globalindex/create_global_index.py
@@ -28,6 +28,7 @@ import pyarrow.compute as pc
from pypaimon.common.options.core_options import CoreOptions
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_writer import (
BTREE_IDENTIFIER,
BTreeIndexWriter,
@@ -160,7 +161,7 @@ class GlobalIndexBuilder:
def build(self) -> List[CommitMessage]:
read_builder = self._table.new_read_builder()
- partition_filter = self._resolve_partition_filter(read_builder)
+ partition_filter = self._resolve_partition_filter()
if partition_filter is not None:
read_builder = read_builder.with_partition_filter(partition_filter)
@@ -486,15 +487,16 @@ class GlobalIndexBuilder:
"(bucket = -1), but table '%s' has bucket = %s."
% (self._table.identifier, bucket)
)
- if self._core_options.deletion_vectors_enabled():
+ if (self._core_options.deletion_vectors_enabled()
+ and not (self._index_type in VINDEX_IDENTIFIERS
+ and self._table.options.data_evolution_enabled())):
raise ValueError(
- "Generic global index does not support tables with deletion "
- "vectors enabled. Table '%s' has "
- "'deletion-vectors.enabled' = true."
- % self._table.identifier
+ "Global index build with deletion vectors requires a native "
+ "vector index on a data-evolution table. Table '%s', index
type '%s'."
+ % (self._table.identifier, self._index_type)
)
- def _resolve_partition_filter(self, read_builder) -> Optional[Predicate]:
+ def _resolve_partition_filter(self) -> Optional[Predicate]:
if self._partition_filter is not None:
return self._partition_filter
if self._partitions is None:
@@ -504,7 +506,7 @@ class GlobalIndexBuilder:
if isinstance(partitions, dict):
partitions = [partitions]
- predicate_builder = read_builder.new_predicate_builder()
+ predicate_builder = PredicateBuilder(self._table.partition_keys_fields)
partition_predicates = []
for partition in partitions:
sub_predicates = []
diff --git a/paimon-python/pypaimon/tests/global_index_build_test.py
b/paimon-python/pypaimon/tests/global_index_build_test.py
index d7f117ae93..a430b19e9c 100644
--- a/paimon-python/pypaimon/tests/global_index_build_test.py
+++ b/paimon-python/pypaimon/tests/global_index_build_test.py
@@ -1048,11 +1048,13 @@ class GlobalIndexBuildTest(
dv_options['deletion-vectors.enabled'] = 'true'
dv_table = self._create_table(pa_schema=schema, options=dv_options)
with self.assertRaisesRegex(ValueError, 'deletion vectors'):
- dv_table.create_global_index(
+ dv_table.copy({'data-evolution.enabled':
'false'}).create_global_index(
'embedding',
index_type='ivf-flat',
options={'ivf-flat.dimension': '2'},
)
+ with self.assertRaisesRegex(ValueError, 'deletion vectors'):
+ dv_table.create_global_index('embedding',
index_type=FULL_TEXT_IDENTIFIER)
def test_vindex_native_options_follow_java_mapping(self):
data_type = ArrayType(True, AtomicType('FLOAT'))
diff --git a/paimon-python/pypaimon/tests/ray_vector_search_test.py
b/paimon-python/pypaimon/tests/ray_vector_search_test.py
index 986e766706..ca1b361521 100644
--- a/paimon-python/pypaimon/tests/ray_vector_search_test.py
+++ b/paimon-python/pypaimon/tests/ray_vector_search_test.py
@@ -66,9 +66,7 @@ def add_rows(table, vectors, start=0):
def build_index(table, metric="l2"):
pytest.importorskip("paimon_vindex")
- # Build before any deletes. The index builder currently requires DVs off;
- # subsequent reads/updates use the table's original DV-enabled options.
- table.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index(
+ table.raw_table.create_global_index(
"embedding", "ivf-flat", options={
"global-index.row-count-per-shard": "3",
"ivf-flat.nlist": "1", "ivf-flat.distance.metric": metric})
diff --git a/paimon-python/pypaimon/tests/vector_filter_exactness_test.py
b/paimon-python/pypaimon/tests/vector_filter_exactness_test.py
index b4fa8be79c..4a1d77861e 100644
--- a/paimon-python/pypaimon/tests/vector_filter_exactness_test.py
+++ b/paimon-python/pypaimon/tests/vector_filter_exactness_test.py
@@ -43,7 +43,7 @@ def table(tmp_path):
"read.batch-size": "1"})
table.add(pa.table({"id": [0, 1, 2], "name": ["alpha", "beta zeta",
"gamma"],
"embedding": [[0., 1.], [1., 1.], [2., 1.]]},
schema=schema))
- table.raw_table.copy({"deletion-vectors.enabled":
"false"}).create_global_index(
+ table.raw_table.create_global_index(
"embedding", "ivf-flat", options={"ivf-flat.nlist": "1",
"ivf-flat.distance.metric": "l2"})
return table
diff --git a/paimon-python/pypaimon/tests/vector_index_dv_build_test.py
b/paimon-python/pypaimon/tests/vector_index_dv_build_test.py
new file mode 100644
index 0000000000..3619aff607
--- /dev/null
+++ b/paimon-python/pypaimon/tests/vector_index_dv_build_test.py
@@ -0,0 +1,184 @@
+# 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 sys
+import types
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+import pypaimon.multimodal as pm
+from pypaimon.globalindex.create_global_index import GlobalIndexBuilder
+from pypaimon.globalindex.vindex.vindex_vector_global_index_reader import
VINDEX_IDENTIFIERS
+from pypaimon.index.index_file_handler import IndexFileHandler
+from pypaimon.tests.global_index_build_test import _FakeVectorIndexTrainer,
_FakeVectorIndexWriter
+
+
+SCHEMA = pa.schema([('id', pa.int64()), ('embedding', pa.list_(pa.float32(),
2)), ('category', pa.string())])
+
+
+def append(table, ids, null_vector=False, category='yes'):
+ # Older Parquet writers cannot encode NULL fixed-size lists.
+ schema = SCHEMA.set(1, pa.field('embedding', pa.list_(pa.float32()))) if
null_vector else SCHEMA
+ table.add(pa.table({'id': ids, 'embedding': [None if null_vector and i ==
3 else [float(i), 1.] for i in ids],
+ 'category': [category] * len(ids)}, schema=schema))
+
+
+def create_table(tmp_path, null_vector=False):
+ schema = SCHEMA.set(1, pa.field('embedding', pa.list_(pa.float32()))) if
null_vector else SCHEMA
+ table = pm.connect(options={'warehouse': str(tmp_path)}).create_table(
+ 'vectors', schema=schema, options={
+ 'file.format': 'parquet', 'vector.file.format': 'parquet',
'read.batch-size': '2',
+ 'global-index.row-count-per-shard': '4',
'global-index.build.parallelism': '1'})
+ append(table, list(range(7)), null_vector)
+ return table
+
+
[email protected]
+def table(tmp_path):
+ return create_table(tmp_path)
+
+
+def entries(table):
+ return sorted((e.index_file for e in
IndexFileHandler(table.raw_table).scan(
+ table.raw_table.snapshot_manager().get_latest_snapshot()) if
e.index_file.index_type in VINDEX_IDENTIFIERS),
+ key=lambda f: f.global_index_meta.row_range_start)
+
+
+def build(table, metric='l2'):
+ pytest.importorskip('paimon_vindex')
+ return table.create_index('embedding', 'ivf-flat', options={
+ 'ivf-flat.dimension': '2', 'ivf-flat.nlist': '1',
'ivf-flat.distance.metric': metric})
+
+
+def search(table, batch=False, refine=False, **kwargs):
+ options = {'ivf.nprobe': '1'}
+ if refine:
+ options['refine_factor'] = '2'
+ query = (table.search_vectors([[1., 1.], [1., 1.]], column='embedding',
options=options, **kwargs) if batch
+ else table.search([1., 1.], column='embedding', options=options,
**kwargs))
+ return query.select(['id']).limit(3).to_list()
+
+
[email protected]('index_type', VINDEX_IDENTIFIERS)
[email protected]('all_deleted', [False, True])
+def test_build_preserves_physical_ids_after_deletion(tmp_path, index_type,
all_deleted):
+ table = create_table(tmp_path, null_vector=True)
+ # Overlapping column files must not duplicate IDs or re-add deleted rows.
+ table.update('id >= 0', {'category': 'changed'})
+ table.delete('id >= 0' if all_deleted else 'id = 0 OR id = 1 OR id = 6')
+ fake = types.SimpleNamespace(VectorIndexTrainer=_FakeVectorIndexTrainer,
VectorIndexWriter=_FakeVectorIndexWriter)
+ with patch.dict(sys.modules, {'paimon_vindex': fake}),
patch.object(_FakeVectorIndexWriter, 'instances', []):
+ assert table.create_index('embedding', index_type, options={index_type
+ '.dimension': '2'}) == 2
+ writers = _FakeVectorIndexWriter.instances
+ # NULL row 3 is counted in coverage but omitted from the native
vectors.
+ assert [w.added_ids for w in writers] == [[0, 1, 2], [0, 1, 2]]
+ assert [w.added_vectors for w in writers] == [
+ [[0., 1.], [1., 1.], [2., 1.]], [[4., 1.], [5., 1.], [6., 1.]]]
+ assert [(f.global_index_meta.row_range_start,
f.global_index_meta.row_range_end, f.row_count)
+ for f in entries(table)] == [(0, 3, 4), (4, 6, 3)]
+ assert table.create_index('embedding', index_type) == 0
+
+
[email protected]('metric', ['l2', 'cosine', 'inner_product'])
[email protected]('refine', [False, True])
+def test_native_search_excludes_deletions_before_and_after_build(table,
metric, refine):
+ table.delete('id = 1 OR id = 6')
+ assert build(table, metric) == 2
+ table.delete('id = 0')
+ table.create_index('category', 'bitmap')
+ expected = [{'id': i} for i in ((3, 4, 5) if metric == 'inner_product'
else (2, 3, 4))]
+ for batch in (False, True):
+ assert search(table, batch, refine, pre_filter="category = 'yes'") == (
+ [expected, expected] if batch else expected)
+ table.delete('id >= 0')
+ assert search(table) == []
+ assert search(table, batch=True) == [[], []]
+ assert build(table, metric) == 0
+
+
+def test_native_build_and_query_keep_snapshot_semantics(table):
+ table.delete('id = 1 OR id = 6')
+ assert build(table) == 2
+ saved = table.raw_table.snapshot_manager().get_latest_snapshot().id
+ builder =
(table.raw_table.new_vector_search_builder().with_vector_column('embedding')
+ .with_query_vector([0.,
1.]).with_limit(1).with_option('ivf.nprobe', '1'))
+ plan = builder.new_vector_search_scan().scan()
+ table.delete('id = 0')
+ assert list(builder.new_vector_search_read().read_plan(plan).results()) ==
[0]
+ assert list(builder.execute_local().results()) == [2]
+ assert search(table, snapshot_id=saved) == [{'id': i} for i in (0, 2, 3)]
+
+
+def test_native_incremental_build_with_concurrent_delete(table):
+ assert build(table) == 2
+ append(table, [7, 8])
+ expected = [{'id': i} for i in (6, 7, 8)]
+ assert table.search([7., 1.],
column='embedding').select(['id']).limit(3).to_list() == expected
+ builder = GlobalIndexBuilder(table.raw_table, 'embedding', 'ivf-flat',
options={
+ 'ivf-flat.dimension': '2', 'ivf-flat.nlist': '1',
'ivf-flat.distance.metric': 'l2'})
+ messages = builder.build()
+ # Deleting a newly indexed row between build and commit must remain
visible.
+ table.delete('id = 7')
+ commit = table.raw_table.new_batch_write_builder().new_commit()
+ try:
+ commit.commit(messages)
+ finally:
+ commit.close()
+ assert [(f.global_index_meta.row_range_start,
f.global_index_meta.row_range_end)
+ for f in entries(table)] == [(0, 3), (4, 6), (7, 7), (8, 8)]
+ expected = [{'id': i} for i in (5, 6, 8)]
+ assert table.search([7., 1.],
column='embedding').select(['id']).limit(3).to_list() == expected
+ assert build(table) == 0
+
+
+def test_native_partition_scoped_build_after_deletion(tmp_path):
+ pytest.importorskip('paimon_vindex')
+ table = pm.connect(options={'warehouse': str(tmp_path)}).create_table(
+ 'vectors', schema=SCHEMA, partitioned=['category'],
options={'file.format': 'parquet'})
+ append(table, [0, 1, 2], category='other')
+ append(table, [3, 4, 5])
+ table.delete("category = 'yes' AND id = 4")
+ assert table.raw_table.create_global_index('embedding', 'ivf-flat',
partitions={'category': 'yes'},
+ options={'ivf-flat.nlist': '1',
'ivf-flat.distance.metric': 'l2'}) == 1
+ assert [(f.global_index_meta.row_range_start,
f.global_index_meta.row_range_end)
+ for f in entries(table)] == [(3, 5)]
+ table.raw_table = table.raw_table.copy({'vector-index.search-mode':
'fast'})
+ assert search(table, pre_filter="category = 'yes'") == [{'id': 3}, {'id':
5}]
+ assert search(table, pre_filter="category = 'other'") == []
+
+
+def test_native_index_commit_rejects_replaced_row_ids(table):
+ pytest.importorskip('paimon_vindex')
+ messages = GlobalIndexBuilder(table.raw_table, 'embedding', 'ivf-flat',
options={
+ 'ivf-flat.nlist': '1'}).build()
+ write_builder = table.raw_table.new_batch_write_builder().overwrite({})
+ writer, commit = write_builder.new_write(), write_builder.new_commit()
+ try:
+ writer.write_arrow(pa.table({'id': [99], 'embedding': [[99., 1.]],
'category': ['yes']}, schema=SCHEMA))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ commit = table.raw_table.new_batch_write_builder().new_commit()
+ try:
+ with pytest.raises(RuntimeError, match='Global index row ID existence
conflict'):
+ commit.commit(messages)
+ finally:
+ commit.close()
+ assert entries(table) == []
