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 f2de0ee193 [python][ray] Add distributed update_by_row_id for
data-evolution tables (#8445)
f2de0ee193 is described below
commit f2de0ee193e7991575f4418db44343f35fef92bb
Author: XiaoHongbo <[email protected]>
AuthorDate: Sat Jul 4 12:45:47 2026 +0800
[python][ray] Add distributed update_by_row_id for data-evolution tables
(#8445)
---
docs/docs/pypaimon/ray-data.md | 43 ++++
paimon-python/pypaimon/ray/__init__.py | 2 +
.../pypaimon/ray/data_evolution_merge_into.py | 2 +-
.../pypaimon/ray/data_evolution_merge_join.py | 10 +-
paimon-python/pypaimon/ray/update_by_row_id.py | 156 +++++++++++
.../pypaimon/tests/ray_update_by_row_id_test.py | 286 +++++++++++++++++++++
6 files changed, 497 insertions(+), 2 deletions(-)
diff --git a/docs/docs/pypaimon/ray-data.md b/docs/docs/pypaimon/ray-data.md
index fb56c54ac6..fc1e6cc9a9 100644
--- a/docs/docs/pypaimon/ray-data.md
+++ b/docs/docs/pypaimon/ray-data.md
@@ -499,3 +499,46 @@ For an end-to-end feature update workflow on Blob tables,
see
- Blob columns can be updated and inserted by `merge_into`. With `update="*"`
or `insert="*"`, the source must include the corresponding blob columns.
If an insert mapping omits a blob column, that column is written as `NULL`.
+
+## Update By Row Id
+
+`update_by_row_id` updates columns of a **data-evolution** table straight from
a
+source that already carries `_ROW_ID` and the new values. Each row is routed
to the
+data file that owns its row id and only those files are rewritten — the target
is
+**never fully read** and there is **no join against it** (unlike
+`merge_into(on=["_ROW_ID"])`, which reads and shuffle-joins the whole target).
It
+pairs with `bucket_join`, which produces the row ids without a shuffle.
Requires
+`ray >= 2.50` and a target with `data-evolution.enabled` and
`row-tracking.enabled`.
+
+```python
+from pypaimon.ray import update_by_row_id
+
+metrics = update_by_row_id(
+ target="database_name.table_name",
+ source=ray_dataset, # ray.data.Dataset / pa.Table / pandas,
carrying _ROW_ID
+ catalog_options={"warehouse": "/path/to/warehouse"},
+ update_cols=["feature"], # non-blob columns to overwrite
+)
+print(metrics) # {"num_updated": 50}
+```
+
+**Parameters:**
+- `source`: a `ray.data.Dataset`, `pyarrow.Table`, or `pandas.DataFrame`
carrying the
+ target `_ROW_ID` and every column in `update_cols`; extra columns are
ignored, and
+ values are cast to the target column types. A table-name source is not
accepted: a
+ table's system `_ROW_ID` is its own and cannot address the target's rows.
+- `update_cols`: the non-blob columns to overwrite. Must be non-empty.
+- `num_partitions`: parallelism for grouping the update rows by target file;
+ defaults to `max(1, cluster_cpus * 2)`.
+- `ray_remote_args`: Ray remote options applied to the update tasks.
+
+**Returns:** `{"num_updated": <rows>}`.
+
+**Notes:**
+- The row ids must exist in the target's current snapshot; a stale or foreign
+ `_ROW_ID` raises rather than silently writing.
+- 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.
diff --git a/paimon-python/pypaimon/ray/__init__.py
b/paimon-python/pypaimon/ray/__init__.py
index ba88fe5c7a..f2daad3a69 100644
--- a/paimon-python/pypaimon/ray/__init__.py
+++ b/paimon-python/pypaimon/ray/__init__.py
@@ -27,12 +27,14 @@ from pypaimon.ray.data_evolution_merge_transform import (
target_col,
lit,
)
+from pypaimon.ray.update_by_row_id import update_by_row_id
__all__ = [
"read_paimon",
"write_paimon",
"bucket_join",
"merge_into",
+ "update_by_row_id",
"WhenMatched",
"WhenNotMatched",
"source_col",
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
index 499ac12a35..ec85b784a4 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_into.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_into.py
@@ -468,7 +468,7 @@ def _require_ray_join() -> None:
if parse(ray.__version__) < parse("2.50.0"):
raise RuntimeError(
- f"merge_into requires ray>=2.50; "
+ f"this Ray operation requires ray>=2.50; "
f"installed ray is {ray.__version__}."
)
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
index 2671daf678..03fc068ea2 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
@@ -441,8 +441,16 @@ def distributed_update_apply(
f"Column '{col}' is not in target table schema."
)
+ # Pin the planner to the caller's base snapshot so row-id routing and the
+ # commit-time conflict check agree even if a concurrent commit lands
(mirrors
+ # the delete path).
+ from pypaimon.common.options.core_options import CoreOptions
+ scan_table = (
+ table.copy({CoreOptions.SCAN_SNAPSHOT_ID.key(): str(base_snapshot_id)})
+ if base_snapshot_id is not None else table
+ )
planner = TableUpdateByRowId(
- table,
+ scan_table,
"_merge_into_planner_" + uuid.uuid4().hex[:8],
BATCH_COMMIT_IDENTIFIER,
)
diff --git a/paimon-python/pypaimon/ray/update_by_row_id.py
b/paimon-python/pypaimon/ray/update_by_row_id.py
new file mode 100644
index 0000000000..a538fb2f4b
--- /dev/null
+++ b/paimon-python/pypaimon/ray/update_by_row_id.py
@@ -0,0 +1,156 @@
+# 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.
+
+"""Distributed row-id update on Ray for data-evolution tables.
+
+Update columns of a data-evolution table straight from a Ray Dataset that
already
+carries ``_ROW_ID`` and the new values -- no full-target read and no big-table
+shuffle join (unlike ``merge_into(on=["_ROW_ID"])``, which reads and joins the
whole
+target). Pairs with ``bucket_join``, which produces the row ids.
+"""
+
+from typing import Any, Dict, List, Optional
+
+import pyarrow as pa
+
+from pypaimon.ray.data_evolution_merge_into import (
+ _normalize_source,
+ _reraise_inner,
+ _require_ray_join,
+ _resolve_num_partitions,
+)
+from pypaimon.ray.data_evolution_merge_join import distributed_update_apply
+from pypaimon.ray.data_evolution_merge_transform import build_update_schema
+
+__all__ = ["update_by_row_id"]
+
+
+def _blob_col_names(table: "FileStoreTable") -> set:
+ return {f.name for f in table.table_schema.fields
+ if getattr(f.type, "type", None) == "BLOB"}
+
+
+def update_by_row_id(
+ target: str,
+ source: Any,
+ catalog_options: Dict[str, str],
+ *,
+ update_cols: List[str],
+ num_partitions: Optional[int] = None,
+ ray_remote_args: Optional[Dict[str, Any]] = None,
+) -> Dict[str, int]:
+ """Update ``update_cols`` of a data-evolution table by ``_ROW_ID``.
+
+ ``source`` (a ``ray.data.Dataset`` / ``pyarrow.Table`` /
``pandas.DataFrame``)
+ must already carry the target ``_ROW_ID`` and the new values. Each row is
+ routed to the data file owning its row id and only those files are
rewritten --
+ the target is never fully read and there is no join against it. Requires
+ ``ray >= 2.50`` and a target with ``data-evolution.enabled`` +
``row-tracking.enabled``.
+
+ Returns ``{"num_updated": <rows>}``.
+ """
+ from pypaimon.catalog.catalog_factory import CatalogFactory
+ from pypaimon.schema.data_types import PyarrowFieldParser
+ from pypaimon.table.special_fields import SpecialFields
+
+ _require_ray_join()
+ if not update_cols:
+ raise ValueError("update_cols must be non-empty.")
+ update_cols = list(dict.fromkeys(update_cols)) # de-dup, keep order
+ num_partitions = _resolve_num_partitions(num_partitions)
+
+ table = CatalogFactory.create(catalog_options).get_table(target)
+ if not table.options.data_evolution_enabled():
+ raise ValueError(
+ f"update_by_row_id requires 'data-evolution.enabled'='true' on
'{target}'.")
+ 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)
+ partition_keys = set(table.partition_keys or [])
+ for col in update_cols:
+ if col not in table.field_names:
+ raise ValueError(f"update column {col!r} is not in target
'{target}'.")
+ if col in blob_cols:
+ # Update writes plain data files; blob deltas are a separate path.
+ raise ValueError(f"update_by_row_id cannot update blob column
{col!r}.")
+ if col in partition_keys:
+ # In-place rewrite can't move a row across partitions.
+ raise ValueError(
+ f"update_by_row_id cannot update partition column {col!r}; "
+ "cross-partition row movement is not supported.")
+
+ if isinstance(source, str):
+ # A table's system _ROW_ID is its own, independent of the target's, so
a
+ # table-name source can't address target rows. Require in-memory data
that
+ # already carries the target row ids (e.g. produced by bucket_join).
+ raise ValueError(
+ "update_by_row_id does not accept a table-name source; pass a
ray.data."
+ f"Dataset / pyarrow.Table / pandas.DataFrame carrying the target
{rid}.")
+ source_ds = _normalize_source(source, catalog_options)
+ src_cols = set(source_ds.schema().names)
+ missing = [c for c in [rid] + update_cols if c not in src_cols]
+ if missing:
+ raise ValueError(
+ f"source is missing columns {missing}; it must carry {rid} and
{update_cols}.")
+
+ # Cast to the on-disk schema (int64 _ROW_ID + target column types) so the
writer
+ # gets exactly the target types regardless of the source's arrow types.
+ target_pa =
PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
+ update_schema = build_update_schema(target_pa, update_cols, rid)
+
+ def _project_cast(batch: pa.Table) -> pa.Table:
+ return batch.select([rid] + update_cols).cast(update_schema)
+
+ 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:
+ # Every source row id is foreign; don't silently no-op non-empty input.
+ if update_ds.limit(1).count() > 0:
+ raise ValueError(
+ f"target '{target}' has no rows; every _ROW_ID in the source
is foreign.")
+ return {"num_updated": 0}
+ try:
+ msgs, num_updated, _ = distributed_update_apply(
+ update_ds, table, update_cols,
+ num_partitions=num_partitions,
+ ray_remote_args=ray_remote_args,
+ base_snapshot_id=base.id,
+ )
+ except Exception as e:
+ _reraise_inner(e)
+ raise # _reraise_inner always raises; keeps msgs/num_updated defined
for linters
+
+ if msgs:
+ wb = table.new_batch_write_builder()
+ tc = wb.new_commit()
+ try:
+ tc.commit(msgs)
+ finally:
+ tc.close()
+ return {"num_updated": num_updated}
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
new file mode 100644
index 0000000000..8b3c71d80a
--- /dev/null
+++ b/paimon-python/pypaimon/tests/ray_update_by_row_id_test.py
@@ -0,0 +1,286 @@
+# 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 os
+import shutil
+import tempfile
+import unittest
+import uuid
+from unittest import mock
+
+import pyarrow as pa
+import pytest
+
+pypaimon = pytest.importorskip("pypaimon")
+ray = pytest.importorskip("ray")
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.ray import update_by_row_id
+
+
+class RayUpdateByRowIdTest(unittest.TestCase):
+ """Distributed row-id update: rewrite only the files owning the given row
ids,
+ without reading or joining the whole target (unlike
merge_into(on=_ROW_ID))."""
+
+ pa_schema = pa.schema([
+ ("id", pa.int32()),
+ ("name", pa.string()),
+ ("age", pa.int32()),
+ ])
+ de_options = {"row-tracking.enabled": "true", "data-evolution.enabled":
"true"}
+
+ @classmethod
+ def setUpClass(cls):
+ cls.tempdir = tempfile.mkdtemp()
+ cls.catalog_options = {"warehouse": os.path.join(cls.tempdir, "wh")}
+ cls.catalog = CatalogFactory.create(cls.catalog_options)
+ cls.catalog.create_database("default", True)
+ if not ray.is_initialized():
+ ray.init(ignore_reinit_error=True, num_cpus=2)
+
+ @classmethod
+ def tearDownClass(cls):
+ try:
+ if ray.is_initialized():
+ ray.shutdown()
+ except Exception:
+ pass
+ shutil.rmtree(cls.tempdir, ignore_errors=True)
+
+ def _create(self, options=None):
+ name = f"default.u_{uuid.uuid4().hex[:8]}"
+ opts = self.de_options if options is None else options
+ self.catalog.create_table(
+ name, Schema.from_pyarrow_schema(self.pa_schema, options=opts),
False)
+ return name
+
+ def _write(self, target, data):
+ t = self.catalog.get_table(target)
+ wb = t.new_batch_write_builder()
+ w = wb.new_write()
+ w.write_arrow(data)
+ wb.new_commit().commit(w.prepare_commit())
+ w.close()
+
+ def _read(self, target, projection=None):
+ t = self.catalog.get_table(target)
+ rb = t.new_read_builder()
+ if projection is not None:
+ rb = rb.with_projection(projection)
+ return rb.new_read().to_arrow(rb.new_scan().plan().splits())
+
+ def _rowid_by_id(self, target):
+ tab = self._read(target, ["_ROW_ID", "id"])
+ return dict(zip(tab.column("id").to_pylist(),
tab.column("_ROW_ID").to_pylist()))
+
+ def test_update_by_row_id_basic(self):
+ target = self._create()
+ self._write(target, pa.Table.from_pydict(
+ {"id": list(range(1, 7)), "name": [f"n{i}" for i in range(1, 7)],
+ "age": [i * 10 for i in range(1, 7)]}, schema=self.pa_schema))
+ rid = self._rowid_by_id(target)
+
+ # update age for ids 2 and 5 only, addressed by their _ROW_ID
+ src = pa.table({"_ROW_ID": [rid[2], rid[5]], "age": [999, 888]},
+ schema=pa.schema([("_ROW_ID", pa.int64()), ("age",
pa.int32())]))
+
+ # Proof of no full-target read: read_paimon is never called (source is
a
+ # Dataset, and the update routes by manifest metadata, not a scan).
+ import pypaimon.ray.ray_paimon as rp
+ with mock.patch.object(rp, "read_paimon",
+ side_effect=AssertionError("target was read!")):
+ stats = update_by_row_id(target, ray.data.from_arrow(src),
+ self.catalog_options, update_cols=["age"])
+ self.assertEqual(stats, {"num_updated": 2})
+
+ back = self._read(target).sort_by("id").to_pydict()
+ self.assertEqual(back["age"], [10, 999, 30, 40, 888, 60])
+ self.assertEqual(back["name"], [f"n{i}" for i in range(1, 7)]) #
untouched
+
+ def test_updates_correct_row_across_files(self):
+ # A _ROW_ID owned by a middle data file must update only that row.
+ target = self._create()
+ for chunk in ([10, 11, 12], [20, 21], [30, 31, 32, 33]):
+ self._write(target, pa.Table.from_pydict(
+ {"id": chunk, "name": ["x"] * len(chunk), "age": [0] *
len(chunk)},
+ schema=self.pa_schema))
+ rid = self._rowid_by_id(target)
+ src = pa.table({"_ROW_ID": [rid[21]], "age": [999]},
+ schema=pa.schema([("_ROW_ID", pa.int64()), ("age",
pa.int32())]))
+ stats = update_by_row_id(target, ray.data.from_arrow(src),
+ self.catalog_options, update_cols=["age"])
+ self.assertEqual(stats, {"num_updated": 1})
+ back = self._read(target).sort_by("id").to_pydict()
+ got = dict(zip(back["id"], back["age"]))
+ self.assertEqual(got[21], 999)
+ self.assertTrue(all(v == 0 for k, v in got.items() if k != 21))
+
+ def test_pins_base_snapshot_for_conflict_detection(self):
+ # The update pins its base snapshot and threads it to
distributed_update_apply,
+ # which uses it for commit-time conflict detection against concurrent
writers.
+ import importlib
+ m = importlib.import_module("pypaimon.ray.update_by_row_id") #
module, not the fn
+ target = self._create()
+ self._write(target, pa.Table.from_pydict(
+ {"id": [1, 2], "name": ["a", "b"], "age": [1, 2]},
schema=self.pa_schema))
+ expected_sid = self.catalog.get_table(
+ target).snapshot_manager().get_latest_snapshot().id
+ rid = self._rowid_by_id(target)
+ src = pa.table({"_ROW_ID": [rid[1]], "age": [9]},
+ schema=pa.schema([("_ROW_ID", pa.int64()), ("age",
pa.int32())]))
+
+ captured = {}
+
+ def fake_apply(update_ds, table, cols, *, num_partitions,
+ ray_remote_args=None, base_snapshot_id=None):
+ captured["base_snapshot_id"] = base_snapshot_id
+ return [], 0, []
+
+ with mock.patch.object(m, "distributed_update_apply", fake_apply):
+ update_by_row_id(target, src, self.catalog_options,
update_cols=["age"])
+ self.assertEqual(captured["base_snapshot_id"], expected_sid)
+
+ def test_accepts_pyarrow_and_pandas_source(self):
+ target = self._create()
+ self._write(target, pa.Table.from_pydict(
+ {"id": [1, 2], "name": ["a", "b"], "age": [1, 2]},
schema=self.pa_schema))
+ rid = self._rowid_by_id(target)
+ # pyarrow.Table source
+ update_by_row_id(
+ target,
+ pa.table({"_ROW_ID": [rid[1]], "age": [77]},
+ schema=pa.schema([("_ROW_ID", pa.int64()), ("age",
pa.int32())])),
+ self.catalog_options, update_cols=["age"])
+ self.assertEqual(self._read(target).sort_by("id").to_pydict()["age"],
[77, 2])
+
+ # pandas.DataFrame source, updating multiple columns at once
+ import pandas as pd
+ update_by_row_id(
+ target,
+ pd.DataFrame({"_ROW_ID": pd.array([rid[2]], dtype="int64"),
+ "name": ["z"], "age": pd.array([88],
dtype="int32")}),
+ self.catalog_options, update_cols=["name", "age"])
+ back = self._read(target).sort_by("id").to_pydict()
+ self.assertEqual(back["age"], [77, 88])
+ self.assertEqual(back["name"], ["a", "z"])
+
+ def test_rejects_table_name_source(self):
+ # A source table's system _ROW_ID is its own, not the target's row
ids, so a
+ # table-name source is rejected rather than silently updating wrong
rows.
+ target = self._create()
+ self._write(target, pa.Table.from_pydict(
+ {"id": [1], "name": ["a"], "age": [1]}, schema=self.pa_schema))
+ with self.assertRaises(ValueError):
+ update_by_row_id(target, "default.some_source",
self.catalog_options,
+ update_cols=["age"])
+
+ def test_rejects_non_data_evolution_table(self):
+ target = self._create(options={}) # plain append table
+ 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"])
+
+ def test_rejects_missing_row_id_column(self):
+ target = self._create()
+ self._write(target, pa.Table.from_pydict(
+ {"id": [1], "name": ["a"], "age": [1]}, schema=self.pa_schema))
+ src = pa.table({"age": [9]}, schema=pa.schema([("age", pa.int32())]))
+ with self.assertRaises(ValueError):
+ update_by_row_id(target, src, self.catalog_options,
update_cols=["age"])
+
+ def test_rejects_partition_column_update(self):
+ name = f"default.u_{uuid.uuid4().hex[:8]}"
+ s = Schema.from_pyarrow_schema(self.pa_schema, partition_keys=["name"],
+ options=self.de_options)
+ self.catalog.create_table(name, s, False)
+ self._write(name, pa.Table.from_pydict(
+ {"id": [1], "name": ["a"], "age": [1]}, schema=self.pa_schema))
+ src = pa.table({"_ROW_ID": [0], "name": ["b"]},
+ schema=pa.schema([("_ROW_ID", pa.int64()), ("name",
pa.string())]))
+ 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.
+ 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"])
+
+ def test_rejects_blob_column_update(self):
+ blob_schema = pa.schema([("id", pa.int32()), ("payload",
pa.large_binary())])
+ name = f"default.u_{uuid.uuid4().hex[:8]}"
+ self.catalog.create_table(
+ name, Schema.from_pyarrow_schema(blob_schema,
options=self.de_options), False)
+ self._write(name, pa.Table.from_pydict(
+ {"id": [1], "payload": pa.array([b"x"], pa.large_binary())},
schema=blob_schema))
+ src = pa.table({"_ROW_ID": [0], "payload": pa.array([b"y"],
pa.large_binary())},
+ schema=pa.schema([("_ROW_ID", pa.int64()), ("payload",
pa.large_binary())]))
+ with self.assertRaises(ValueError):
+ update_by_row_id(name, src, self.catalog_options,
update_cols=["payload"])
+
+ def test_empty_target_foreign_row_id_raises(self):
+ src = pa.table({"_ROW_ID": [0], "age": [9]},
+ schema=pa.schema([("_ROW_ID", pa.int64()), ("age",
pa.int32())]))
+ empty_src = pa.table({"_ROW_ID": pa.array([], pa.int64()),
+ "age": pa.array([], pa.int32())})
+
+ # (a) never written -> no snapshot
+ target = self._create()
+ with self.assertRaises(ValueError):
+ update_by_row_id(target, src, self.catalog_options,
update_cols=["age"])
+ # empty source against an empty target is a no-op, not an error
+ self.assertEqual(
+ update_by_row_id(target, empty_src, self.catalog_options,
update_cols=["age"]),
+ {"num_updated": 0})
+
+ # (b) written then emptied by overwrite -> snapshot exists but 0 live
rows
+ target2 = self._create()
+ self._write(target2, pa.Table.from_pydict(
+ {"id": [1], "name": ["a"], "age": [1]}, schema=self.pa_schema))
+ wb =
self.catalog.get_table(target2).new_batch_write_builder().overwrite()
+ w = wb.new_write()
+ w.write_arrow(pa.Table.from_pydict(
+ {"id": pa.array([], pa.int32()), "name": pa.array([], pa.string()),
+ "age": pa.array([], pa.int32())}, schema=self.pa_schema))
+ wb.new_commit().commit(w.prepare_commit())
+ w.close()
+ with self.assertRaises(ValueError):
+ update_by_row_id(target2, src, self.catalog_options,
update_cols=["age"])
+
+ def test_rejects_unknown_and_empty_update_cols(self):
+ target = self._create()
+ 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=["nope"])
+ with self.assertRaises(ValueError):
+ update_by_row_id(target, src, self.catalog_options, update_cols=[])
+
+
+if __name__ == "__main__":
+ unittest.main()
