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 3d63e41188 [python] Support deletion vectors in Ray row-id reads and
updates (#10427)
3d63e41188 is described below
commit 3d63e41188f1cbb6521db9375b2b23270250ffa5
Author: chaoyang <[email protected]>
AuthorDate: Thu Oct 8 15:21:32 2026 +0800
[python] Support deletion vectors in Ray row-id reads and updates (#10427)
---
docs/docs/pypaimon/ray-row-ids.md | 16 ++++---
.../pypaimon/ray/data_evolution_merge_join.py | 33 +++++++++++---
paimon-python/pypaimon/ray/read_by_row_id.py | 10 ++---
paimon-python/pypaimon/ray/update_by_row_id.py | 11 +----
.../pypaimon/tests/ray_read_by_row_id_test.py | 31 ++++++++++---
.../pypaimon/tests/ray_update_by_row_id_test.py | 51 ++++++++++++++++++----
6 files changed, 108 insertions(+), 44 deletions(-)
diff --git a/docs/docs/pypaimon/ray-row-ids.md
b/docs/docs/pypaimon/ray-row-ids.md
index ec8eed7269..7a004e1e99 100644
--- a/docs/docs/pypaimon/ray-row-ids.md
+++ b/docs/docs/pypaimon/ray-row-ids.md
@@ -25,7 +25,7 @@ under the License.
# Ray Row IDs and Backfills
-Read selected rows and write derived columns using target-table row IDs. These
APIs require Ray 2.50 or newer and data-evolution tables with row tracking. The
read/update paths on this page reject deletion-vectors-enabled tables; see
[Multimodal Reads](./multimodal-reading) for high-level BLOB processing.
+Read selected rows and write derived columns using target-table row IDs. These
APIs require Ray 2.50 or newer and data-evolution tables with row tracking.
Row-ID reads and updates support deletion vectors; see [Multimodal
Reads](./multimodal-reading) for high-level BLOB processing.
Install a Ray version that supports these APIs in both the driver and workers:
@@ -88,8 +88,10 @@ print(metrics) # {"num_updated": 50}
- Multiple source rows mapping to the same `_ROW_ID` is rejected — deduplicate
first.
- Blob columns cannot be updated through this path.
- Partition columns cannot be updated (in-place rewrite can't move a row
across partitions).
-- Deletion-vectors-enabled tables are not supported yet: a DV-deleted row
still lives
- in its data file, so it can't be told apart from a live row without reading
the target.
+- Updates validate the requested IDs against deletion vectors from the planning
+ snapshot and reject deleted IDs before writing that file group. Physical row
+ positions are preserved during partial-column rewrites. A concurrent delete
+ remains effective even if its row was live when this update was planned.
## Read By Row Id
@@ -148,8 +150,8 @@ ds = read_by_row_id(
Each call captures the resolved snapshot for planning and worker reads;
changing
the tag before a lazy Dataset executes does not switch that Dataset to
another
snapshot.
-- Deletion-vectors-enabled tables are not supported yet, for the same reason as
- `update_by_row_id`.
+- Deleted row IDs are omitted, using the deletion vectors of the resolved
+ snapshot. Reading an earlier retained snapshot can still return those rows.
- For a non-empty target, the `row_ids` source is consumed lazily by the
downstream
action, not read here. A lazy source missing `row_id_col` raises when the
read runs
(a materialized source raises up front).
@@ -262,8 +264,8 @@ process_row_id_ranges(
```
The target must enable `row-tracking.enabled` and
-`data-evolution.enabled`; `embedding` must be nullable and the table must not
-enable deletion vectors. If the source BLOB or embedding model can change,
+`data-evolution.enabled`; `embedding` must be nullable. With deletion vectors,
+a row deleted between reading and update planning causes the update to fail.
If the source BLOB or embedding model can change,
use an additional source/model-version column instead of treating every
non-null embedding as permanently complete. `process_row_id_ranges` does not
retry a failed processor itself—the resumability in this example comes from
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
index 3a261259a7..c745db9e8e 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
@@ -831,6 +831,9 @@ def distributed_update_apply(
)
for_update = group.drop_columns([frid_col])
+ info = ray.get(precomputed_info_ref)
+ _validate_live_row_ids(
+ captured_table, info, int(group[frid_col][0].as_py()),
for_update[row_id_name].to_pylist())
row_ids = (
for_update.column(row_id_name).to_pylist()
if collect_row_ids else []
@@ -908,7 +911,6 @@ def distributed_read_by_row_id(
import ray
from pypaimon.globalindex.indexed_split import IndexedSplit
- from pypaimon.read.split import DataSplit
from pypaimon.table.special_fields import SpecialFields
from pypaimon.utils.range import Range
@@ -983,12 +985,8 @@ def distributed_read_by_row_id(
frid = int(group.column(frid_col)[0].as_py())
info = ray.get(precomputed_info_ref)
owning_split, target_files = info.first_row_id_index[frid]
- origin_split = DataSplit(
- files=target_files,
- partition=owning_split.partition,
- bucket=owning_split.bucket,
- raw_convertible=True,
- )
+ target_names = {file.file_name for file in target_files}
+ origin_split = owning_split.filter_file(lambda file: file.file_name in
target_names)
# Only matched rows (deduped, contiguous ids -> ranges); blob gets
row-index pushdown.
wanted = set(group.column(row_id_name).to_pylist())
indexed = IndexedSplit(origin_split, Range.to_ranges(list(wanted)))
@@ -1275,3 +1273,24 @@ def distributed_write_collect_msgs(
write_kwargs["concurrency"] = concurrency
insert_ds.write_datasink(sink, **write_kwargs)
return sink.collected
+
+
+def _validate_live_row_ids(table, info, first_row_id, row_ids):
+ """Reject deleted updates using only the routed snapshot's DV files.
+
+ The writer must still merge physical rows, including tombstones, to
preserve
+ row-id offsets. Filtering the writer's original data would shift those
offsets.
+ """
+ from pypaimon.deletionvectors.deletion_vector import DeletionVector
+
+ split, files = info.first_row_id_index[first_row_id]
+ if split.data_deletion_files is None:
+ return
+ target_names = {file.file_name for file in files}
+ for file, deletion in zip(split.files, split.data_deletion_files):
+ if deletion is None or file.file_name not in target_names:
+ continue
+ vector = DeletionVector.read(table.file_io, deletion)
+ for row_id in row_ids:
+ if file.row_id_range().contains(row_id) and
vector.is_deleted(row_id - file.first_row_id):
+ raise ValueError("Cannot update deleted _ROW_ID
{}.".format(row_id))
diff --git a/paimon-python/pypaimon/ray/read_by_row_id.py
b/paimon-python/pypaimon/ray/read_by_row_id.py
index 234b6edbe1..4e9011bb26 100644
--- a/paimon-python/pypaimon/ray/read_by_row_id.py
+++ b/paimon-python/pypaimon/ray/read_by_row_id.py
@@ -114,11 +114,7 @@ def read_by_row_id(
if not table.options.row_tracking_enabled():
raise ValueError(
f"read_by_row_id requires 'row-tracking.enabled'='true' on
'{target}'.")
- if table.options.deletion_vectors_enabled():
- # A DV-deleted row still lives in its file, so slicing would surface
it.
- raise ValueError(
- f"read_by_row_id does not support deletion-vectors-enabled tables
yet: "
- f"'{target}'.")
+
if dynamic_options:
# Flipping these would bypass the checks above.
bad = sorted({"data-evolution.enabled", "row-tracking.enabled",
@@ -176,8 +172,8 @@ def read_by_row_id(
if not
CoreOptions(Options(base_schema.options)).row_tracking_enabled():
raise ValueError(
f"the resolved snapshot ({base.id}) predates row-tracking;
read_by_row_id needs it.")
- # No DV (rejected above) -> total_record_count is the live row count; 0 =
empty.
- if base is None or base.total_record_count == 0:
+ if base is None or (base.total_record_count == 0
+ and not table.options.deletion_vectors_enabled()):
# Force an action on the source only in this degenerate branch (like
update_by_row_id).
if rid_ds.limit(1).count() > 0:
raise ValueError(
diff --git a/paimon-python/pypaimon/ray/update_by_row_id.py
b/paimon-python/pypaimon/ray/update_by_row_id.py
index 58679a8a98..a3f7f30a34 100644
--- a/paimon-python/pypaimon/ray/update_by_row_id.py
+++ b/paimon-python/pypaimon/ray/update_by_row_id.py
@@ -85,12 +85,6 @@ def update_by_row_id(
if not table.options.row_tracking_enabled():
raise ValueError(
f"update_by_row_id requires 'row-tracking.enabled'='true' on
'{target}'.")
- if table.options.deletion_vectors_enabled():
- # A DV-deleted row still lives in its data file, so row-id ranges
can't tell it
- # apart without reading the target; refuse rather than update a
deleted row.
- raise ValueError(
- f"update_by_row_id does not support deletion-vectors-enabled
tables yet: "
- f"'{target}'.")
rid = SpecialFields.ROW_ID.name
blob_cols = _blob_col_names(table)
@@ -139,9 +133,8 @@ def update_by_row_id(
update_ds = source_ds.map_batches(_project_cast, batch_format="pyarrow")
base = table.snapshot_manager().get_latest_snapshot()
- # Without deletion vectors (rejected above), total_record_count is the
live row
- # count, so 0 means the target is empty (never written, or emptied by
overwrite).
- if base is None or base.total_record_count == 0:
+ if base is None or (base.total_record_count == 0
+ and not table.options.deletion_vectors_enabled()):
# Every source row id is foreign; don't silently no-op non-empty input.
if update_ds.limit(1).count() > 0:
raise ValueError(
diff --git a/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py
b/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py
index be214002b3..9257c8fd4d 100644
--- a/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py
+++ b/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py
@@ -418,14 +418,33 @@ class RayReadByRowIdTest(unittest.TestCase):
with self.assertRaises(ValueError):
read_by_row_id(target, src, self.catalog_options,
projection=["age"])
- def test_rejects_deletion_vectors_table(self):
+ def test_reads_deletion_vectors_and_pins_snapshot(self):
opts = dict(self.de_options, **{"deletion-vectors.enabled": "true"})
target = self._create(options=opts)
- self._write(target, pa.Table.from_pydict(
- {"id": [1], "name": ["a"], "age": [1]}, schema=self.pa_schema))
- src = pa.table({"_ROW_ID": [0]}, schema=pa.schema([("_ROW_ID",
pa.int64())]))
- with self.assertRaises(ValueError):
- read_by_row_id(target, src, self.catalog_options,
projection=["age"])
+ for ids in ([1, 2, 3], [4, 5, 6]):
+ self._write(target, pa.Table.from_pydict(
+ {"id": ids, "name": ["a"] * 3, "age": ids},
schema=self.pa_schema))
+ table = self.catalog.get_table(target)
+ src = pa.table({"_ROW_ID": [0, 1, 2, 3, 4, 5, 1]})
+ before = read_by_row_id(target, src, self.catalog_options,
projection=["id"])
+ wb = table.new_batch_write_builder()
+ commit = wb.new_commit()
+ try:
+ commit.commit(wb.new_update().delete_by_row_id([0, 2, 4]))
+ after = read_by_row_id(target, src, self.catalog_options,
projection=["id"])
+ self.assertEqual(set(self._rows_by_id(before)), {1, 2, 3, 4, 5, 6})
+ self.assertEqual(set(self._rows_by_id(after)), {2, 4, 6})
+ deleted = read_by_row_id(
+ target, pa.table({"_ROW_ID": [0, 2, 4]}),
self.catalog_options, projection=["id"])
+ self.assertEqual(deleted.take_all(), [])
+ self.assertEqual(deleted.schema().names, ["id", "_ROW_ID"])
+ commit.close()
+ commit = wb.new_commit()
+ commit.commit(wb.new_update().delete_by_row_id([1, 3, 5]))
+ self.assertEqual(read_by_row_id(
+ target, src, self.catalog_options,
projection=["id"]).take_all(), [])
+ finally:
+ commit.close()
def test_rejects_missing_row_id_column(self):
target = self._create()
diff --git a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py
b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py
index a0072770ea..9fb5d3257f 100644
--- a/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py
+++ b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py
@@ -492,17 +492,52 @@ class RayUpdateByRowIdTest(unittest.TestCase):
with self.assertRaises(ValueError):
update_by_row_id(name, src, self.catalog_options,
update_cols=["name"])
- def test_rejects_deletion_vectors_table(self):
- # A DV-deleted row still lives in its file, so update_by_row_id can't
tell it is
- # gone without reading the target; DV tables are refused for now.
+ def test_updates_live_rows_with_deletion_vectors(self):
opts = dict(self.de_options, **{"deletion-vectors.enabled": "true"})
target = self._create(options=opts)
self._write(target, pa.Table.from_pydict(
- {"id": [1], "name": ["a"], "age": [1]}, schema=self.pa_schema))
- src = pa.table({"_ROW_ID": [0], "age": [9]},
- schema=pa.schema([("_ROW_ID", pa.int64()), ("age",
pa.int32())]))
- with self.assertRaises(ValueError):
- update_by_row_id(target, src, self.catalog_options,
update_cols=["age"])
+ {"id": [1, 2, 3, 4], "name": ["a"] * 4, "age": [10, 20, 30, 40]},
+ schema=self.pa_schema))
+ table = self.catalog.get_table(target)
+ wb = table.new_batch_write_builder()
+ commit = wb.new_commit()
+ try:
+ commit.commit(wb.new_update().delete_by_row_id([0, 2]))
+ finally:
+ commit.close()
+ src = pa.table({"_ROW_ID": [1, 3], "age": [99, 88]})
+ self.assertEqual(update_by_row_id(
+ target, src, self.catalog_options, update_cols=["age"]),
{"num_updated": 2})
+ rb = table.new_read_builder().with_projection(["id", "age", "_ROW_ID"])
+ rows =
rb.new_read().to_arrow(rb.new_scan().plan().splits()).to_pylist()
+ self.assertEqual(sorted(rows, key=lambda r: r["id"]), [
+ {"id": 2, "age": 99, "_ROW_ID": 1}, {"id": 4, "age": 88,
"_ROW_ID": 3}])
+ # Deleting after workers have prepared their column files must not
+ # resurrect the row when the update is committed.
+ import importlib
+ module = importlib.import_module("pypaimon.ray.update_by_row_id")
+ original_commit = module._commit_update_messages
+
+ def delete_then_commit(target_table, messages):
+ builder = target_table.new_batch_write_builder()
+ concurrent = builder.new_commit()
+ try:
+ concurrent.commit(builder.new_update().delete_by_row_id([3]))
+ finally:
+ concurrent.close()
+ original_commit(target_table, messages)
+
+ with mock.patch.object(module, "_commit_update_messages",
delete_then_commit):
+ update_by_row_id(target, pa.table({"_ROW_ID": [3], "age": [77]}),
+ self.catalog_options, update_cols=["age"])
+ rb = table.new_read_builder().with_projection(["id", "age", "_ROW_ID"])
+
self.assertEqual(rb.new_read().to_arrow(rb.new_scan().plan().splits()).to_pylist(),
+ [{"id": 2, "age": 99, "_ROW_ID": 1}])
+ snapshot = table.snapshot_manager().get_latest_snapshot().id
+ with self.assertRaisesRegex(ValueError, "deleted _ROW_ID"):
+ update_by_row_id(target, pa.table({"_ROW_ID": [0, 1], "age": [9,
9]}),
+ self.catalog_options, update_cols=["age"])
+ self.assertEqual(table.snapshot_manager().get_latest_snapshot().id,
snapshot)
def test_rejects_blob_column_update(self):
blob_schema = pa.schema([("id", pa.int32()), ("payload",
pa.large_binary())])
