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 dd2bc70cca [python] Preserve retained snapshots in Ray row ID reads
(#10042)
dd2bc70cca is described below
commit dd2bc70cca87f6f808b815040e07154ada282a06
Author: chaoyang <[email protected]>
AuthorDate: Mon Sep 21 15:04:58 2026 +0800
[python] Preserve retained snapshots in Ray row ID reads (#10042)
---
docs/docs/pypaimon/ray-row-ids.md | 5 ++
.../pypaimon/ray/data_evolution_merge_join.py | 17 ++---
paimon-python/pypaimon/ray/read_by_row_id.py | 16 ++---
.../pypaimon/tests/ray_read_by_row_id_test.py | 76 +++++++++++++++++++++-
4 files changed, 92 insertions(+), 22 deletions(-)
diff --git a/docs/docs/pypaimon/ray-row-ids.md
b/docs/docs/pypaimon/ray-row-ids.md
index 4f65b68f01..ec8eed7269 100644
--- a/docs/docs/pypaimon/ray-row-ids.md
+++ b/docs/docs/pypaimon/ray-row-ids.md
@@ -143,6 +143,11 @@ ds = read_by_row_id(
ranges raise an error; matching numeric IDs from another table cannot be
distinguished. Persist the source table and snapshot or tag with row-ID work
lists, and select that version when reading them.
+- A tag selected through `scan.tag-name` or `scan.version` remains readable
after
+ its main snapshot metadata expires, while the tag retains the snapshot's
files.
+ 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`.
- For a non-empty target, the `row_ids` source is consumed lazily by the
downstream
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
index 5cb42fbf49..835c9a981b 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
@@ -904,22 +904,21 @@ def distributed_read_by_row_id(
*,
num_partitions: Optional[int],
ray_remote_args: Optional[Dict[str, Any]] = None,
- base_snapshot_id: Optional[int] = None,
estimated_size_bytes: Optional[int] = None,
estimated_num_rows: Optional[int] = None,
data_context=None,
):
"""Read ``projection`` for the ``_ROW_ID``s in ``row_ids_ds``, routing
each to its
owning file and reading only the matched rows via ``IndexedSplit`` slicing
(blob
- resolved). Returns a ``ray.data.Dataset`` of ``(*projection, _ROW_ID)``,
or ``None``
- if the target is empty. Read-side mirror of ``distributed_update_apply``.
+ resolved). ``table`` carries the resolved read snapshot. Returns a
+ ``ray.data.Dataset`` of ``(*projection, _ROW_ID)``, or ``None`` if the
target
+ is empty. Read-side mirror of ``distributed_update_apply``.
"""
import numpy as np
import uuid
import ray
- from pypaimon.common.options.core_options import CoreOptions
from pypaimon.globalindex.indexed_split import IndexedSplit
from pypaimon.read.split import DataSplit
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
@@ -935,13 +934,9 @@ def distributed_read_by_row_id(
# Typed empty block so all output blocks share one schema.
empty_out = _read_output_schema(table, read_cols).empty_table()
- # Read-only planner (only scans the manifest); pinned to the base snapshot
for stable routing.
- scan_table = (
- table.copy({CoreOptions.SCAN_SNAPSHOT_ID.key(): str(base_snapshot_id)})
- if base_snapshot_id is not None else table
- )
+ # The caller pinned the resolved snapshot, including any retained tag
metadata.
planner = TableUpdateByRowId(
- scan_table,
+ table,
"_read_by_row_id_planner_" + uuid.uuid4().hex[:8],
BATCH_COMMIT_IDENTIFIER,
)
@@ -995,7 +990,7 @@ def distributed_read_by_row_id(
frid_col, pa.array(sorted_arr[idx], type=pa.int64())
)
- captured_table = scan_table # read at the same pinned snapshot the
planner routed on
+ captured_table = table # read at the same pinned snapshot the planner
routed on
captured_read_cols = read_cols
captured_empty = empty_out
diff --git a/paimon-python/pypaimon/ray/read_by_row_id.py
b/paimon-python/pypaimon/ray/read_by_row_id.py
index d9358f3723..234b6edbe1 100644
--- a/paimon-python/pypaimon/ray/read_by_row_id.py
+++ b/paimon-python/pypaimon/ray/read_by_row_id.py
@@ -169,6 +169,10 @@ def read_by_row_id(
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.common.options.options import Options
base_schema = table.schema_manager.get_schema(base.schema_id)
+ if table.table_schema.id != base_schema.id:
+ raise ValueError(
+ "The time-travel schema changed while resolving the read
snapshot; "
+ "retry 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.")
@@ -179,20 +183,14 @@ def read_by_row_id(
raise ValueError(
f"target '{target}' has no rows; every _ROW_ID in the source
is foreign.")
return _empty_result(table, read_cols)
- # base captures the resolved snapshot; reduce any time-travel key to a
plain snapshot-id
- # so the planner's own snapshot-id pin does not read as a second,
conflicting one.
- from pypaimon.common.options.core_options import CoreOptions
- present = [k for k in SCAN_KEYS if table.options.options.contains_key(k)]
- if present:
- overrides = {k: None for k in present}
- overrides[CoreOptions.SCAN_SNAPSHOT_ID.key()] = str(base.id)
- table = table.copy(overrides)
+ # Carry the resolved metadata into planning and workers: a tag can retain
its
+ # snapshot after the main snapshot file expires, or move before lazy
execution.
+ table = table._copy_with_snapshot(base)
try:
result = distributed_read_by_row_id(
rid_ds, table, projection,
num_partitions=num_partitions,
ray_remote_args=ray_remote_args,
- base_snapshot_id=base.id,
estimated_size_bytes=estimated_size_bytes,
estimated_num_rows=estimated_num_rows,
data_context=data_context,
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 d14392634b..be214002b3 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
@@ -255,6 +255,78 @@ class RayReadByRowIdTest(unittest.TestCase):
read_by_row_id(target, src, self.catalog_options,
projection=["age"],
dynamic_options={"scan.snapshot-id": "1",
"scan.tag-name": "x"})
+ def test_retained_tag_after_snapshot_expiry(self):
+ from pypaimon.schema.schema_change import SchemaChange
+
+ target = self._create()
+ self._write(target, pa.Table.from_pydict(
+ {"id": [1, 2], "name": ["a", "b"], "age": [10, 20]},
schema=self.pa_schema))
+ table = self.catalog.get_table(target)
+ table.create_tag("training", 1)
+ rid = self._rowid_by_id(target)
+ self._write(target, pa.Table.from_pydict(
+ {"id": [3], "name": ["c"], "age": [30]}, schema=self.pa_schema))
+ self.catalog.alter_table(target, [SchemaChange.rename_column("age",
"years")])
+ table.file_io.delete(table.snapshot_manager().get_snapshot_path(1))
+
+ for selector in ("scan.tag-name", "scan.version"):
+ with self.subTest(selector=selector):
+ ds = read_by_row_id(
+ target, pa.table({"_ROW_ID": [rid[2], rid[1], rid[1]]}),
+ self.catalog_options, projection=["id", "age"],
+ dynamic_options={selector: "training",
"scan.native-plan.enabled": "true"})
+ self.assertEqual(self._rows_by_id(ds), {
+ 1: {"id": 1, "age": 10, "_ROW_ID": rid[1]},
+ 2: {"id": 2, "age": 20, "_ROW_ID": rid[2]},
+ })
+
+ def test_rejects_tag_schema_change_during_snapshot_resolution(self):
+ import importlib
+ from pypaimon.schema.schema_change import SchemaChange
+
+ module = importlib.import_module("pypaimon.ray.read_by_row_id")
+ target = self._create()
+ self._write(target, pa.Table.from_pydict(
+ {"id": [1], "name": ["a"], "age": [10]}, schema=self.pa_schema))
+ table = self.catalog.get_table(target)
+ table.create_tag("training", 1)
+ self.catalog.alter_table(target, [SchemaChange.rename_column("age",
"years")])
+ schema = pa.schema([("id", pa.int32()), ("name", pa.string()),
("years", pa.int32())])
+ self._write(target, pa.table({"id": [2], "name": ["b"], "years":
[20]}, schema=schema))
+ resolve = module._read_snapshot
+
+ def resolve_after_tag_moves(read_table):
+ table.replace_tag("training")
+ return resolve(read_table)
+
+ with mock.patch.object(module, "_read_snapshot",
resolve_after_tag_moves):
+ with self.assertRaisesRegex(ValueError, "schema changed.*retry"):
+ read_by_row_id(
+ target, pa.table({"_ROW_ID": [0]}), self.catalog_options,
+ projection=["age"], dynamic_options={"scan.tag-name":
"training"})
+
+ def test_lazy_tag_read_keeps_resolved_snapshot(self):
+ target = self._create()
+ self._write(target, pa.Table.from_pydict(
+ {"id": [1], "name": ["a"], "age": [10]}, schema=self.pa_schema))
+ table = self.catalog.get_table(target)
+ table.create_tag("training", 1)
+ rid = self._rowid_by_id(target)[1]
+ ds = read_by_row_id(
+ target, pa.table({"_ROW_ID": [rid]}), self.catalog_options,
+ projection=["id", "age"], dynamic_options={"scan.tag-name":
"training"})
+
+ from pypaimon.multimodal.table import MultimodalTable
+ MultimodalTable(self.catalog, target, table).update("id = 1", {"age":
99})
+ table.replace_tag("training")
+ table.file_io.delete(table.snapshot_manager().get_snapshot_path(1))
+
+ self.assertEqual(ds.take_all(), [{"id": 1, "age": 10, "_ROW_ID": rid}])
+ latest = read_by_row_id(
+ target, pa.table({"_ROW_ID": [rid]}), self.catalog_options,
+ projection=["age"], dynamic_options={"scan.tag-name": "training"})
+ self.assertEqual(latest.take_all(), [{"age": 99, "_ROW_ID": rid}])
+
def test_row_tracking_cannot_be_enabled_after_data(self):
# row-tracking.enabled / data-evolution.enabled are immutable once the
# table has snapshots (Java parity), so a "snapshot predates
@@ -283,10 +355,10 @@ class RayReadByRowIdTest(unittest.TestCase):
captured = {}
def fake_read(rid_ds, table, projection, *, num_partitions,
- ray_remote_args=None, base_snapshot_id=None,
+ ray_remote_args=None,
estimated_size_bytes=None, estimated_num_rows=None,
data_context=None):
- captured["base_snapshot_id"] = base_snapshot_id
+ captured["base_snapshot_id"] = table._read_snapshot.id
captured["num_partitions"] = num_partitions
captured["estimated_size_bytes"] = estimated_size_bytes
captured["estimated_num_rows"] = estimated_num_rows
