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 158982cd1e [python][ray] Fix update_by_row_id failure on empty
datasets (#9217)
158982cd1e is described below
commit 158982cd1eebfbc60a3395f2a7b628ad66a88905
Author: XiaoHongbo <[email protected]>
AuthorDate: Sat Aug 15 20:18:00 2026 +0800
[python][ray] Fix update_by_row_id failure on empty datasets (#9217)
---
.github/workflows/paimon-python-checks.yml | 9 ++++++---
.../pypaimon/ray/data_evolution_merge_join.py | 17 ++++++++++++++++-
.../tests/ray_data_evolution_merge_into_test.py | 18 ++++++++++++++++++
.../pypaimon/tests/ray_update_by_row_id_test.py | 20 ++++++++++++++++++++
4 files changed, 60 insertions(+), 4 deletions(-)
diff --git a/.github/workflows/paimon-python-checks.yml
b/.github/workflows/paimon-python-checks.yml
index 9cc772dd99..6e56db488d 100755
--- a/.github/workflows/paimon-python-checks.yml
+++ b/.github/workflows/paimon-python-checks.yml
@@ -300,13 +300,16 @@ jobs:
echo "=========================================="
echo "Testing Ray version compatibility"
echo "=========================================="
- for ray_version in 2.44.0 2.48.0 2.53.0; do
+ for ray_version in 2.44.0 2.48.0 2.50.1 2.53.0; do
echo "Testing Ray version: $ray_version"
python -m pip install --no-cache-dir -q ray==$ray_version
python -c "import ray; print(f'Ray version: {ray.__version__}')"
python -c "from packaging.version import parse; import ray; assert
parse(ray.__version__) == parse('$ray_version'), f'Expected Ray $ray_version,
got {ray.__version__}'"
- python -m pytest pypaimon/tests/ray_data_test.py::RayDataTest \
- pypaimon/tests/ray_range_join_test.py::RayRangeJoinTest -v
--tb=short || {
+ tests="pypaimon/tests/ray_data_test.py::RayDataTest
pypaimon/tests/ray_range_join_test.py::RayRangeJoinTest"
+ if [ "$ray_version" = "2.50.1" ]; then
+ tests="$tests
pypaimon/tests/ray_update_by_row_id_test.py::RayUpdateByRowIdTest::test_empty_dataset_after_transform_is_noop"
+ fi
+ python -m pytest $tests -v --tb=short || {
echo "Tests failed for Ray $ray_version"; python -m pip
uninstall -y ray; exit 1;
}
python -m pip uninstall -y ray
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
index 359af8e6ef..4a0dd74b35 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
@@ -503,6 +503,7 @@ def distributed_update_apply(
precomputed_info_ref = ray.put(files_info)
frid_col = "_FIRST_ROW_ID"
+ sentinel_row_id = -1
captured_sorted = sorted_first_row_ids
captured_sorted_arr = np.asarray(captured_sorted, dtype=np.int64)
valid_ranges = planner.valid_row_id_ranges
@@ -544,11 +545,25 @@ def distributed_update_apply(
map_kwargs = _map_kwargs(ray_remote_args)
with_frid = update_ds.map_batches(_assign_frid, **map_kwargs)
+ from pypaimon.schema.data_types import PyarrowFieldParser
+ target_pa =
PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
+ update_schema = build_update_schema(target_pa, cols, row_id_name)
+ # Ray drops the schema of an empty shuffle input. A negative row id keeps
+ # the group-by schema in a separate group that is never written.
+ sentinel = pa.Table.from_arrays(
+ [pa.array([sentinel_row_id], type=pa.int64())]
+ + [pa.nulls(1, type=update_schema.field(col).type) for col in cols],
+ schema=update_schema,
+ ).append_column(
+ frid_col, pa.array([sentinel_row_id], type=pa.int64())
+ )
+ with_frid = with_frid.union(ray.data.from_arrow(sentinel))
+
captured_table = table
captured_cols = cols
def _apply_group(group: pa.Table) -> pa.Table:
- if group.num_rows == 0:
+ if group.column(frid_col)[0].as_py() == sentinel_row_id:
return pa.Table.from_pydict({
"msgs_blob": pa.array([], type=pa.binary()),
"n_updated": pa.array([], type=pa.int64()),
diff --git a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
index cddb11ac37..2e1e8b8e44 100644
--- a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
@@ -849,6 +849,24 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
self.assertEqual(self._snapshot_id(target), before)
+ def test_matched_update_with_no_matches_is_noop(self):
+ target = self._create_table()
+ self._write(target, self._source(ids=(1,)))
+ before = self._snapshot_id(target)
+
+ result = merge_into(
+ target=target,
+ source=self._source(ids=(2,)),
+ catalog_options=self.catalog_options,
+ on=['id'],
+ when_matched=[WhenMatched.update('*')],
+ num_partitions=_TEST_NUM_PARTITIONS,
+ )
+
+ self.assertEqual(result['num_matched'], 0)
+ self.assertEqual(self._snapshot_id(target), before)
+ self.assertEqual(self._read_sorted(target)['id'], [1])
+
def test_matched_on_partitioned_table(self):
pt_schema = pa.schema([
('pt', pa.string()),
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 613fa50d03..801c35218a 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
@@ -130,6 +130,26 @@ class RayUpdateByRowIdTest(unittest.TestCase):
self.assertEqual(got[21], 999)
self.assertTrue(all(v == 0 for k, v in got.items() if k != 21))
+ def test_empty_dataset_after_transform_is_noop(self):
+ target = self._create()
+ self._write(target, pa.Table.from_pydict(
+ {"id": [1], "name": ["a"], "age": [1]}, schema=self.pa_schema))
+ rid = self._rowid_by_id(target)
+ source = ray.data.from_arrow(pa.table({
+ "_ROW_ID": pa.array([rid[1]], pa.int64()),
+ "age": pa.array([9], pa.int32()),
+ })).map_batches(
+ lambda batch: batch.slice(0, 0),
+ batch_format="pyarrow",
+ )
+
+ self.assertEqual(
+ update_by_row_id(
+ target, source, self.catalog_options, update_cols=["age"]),
+ {"num_updated": 0},
+ )
+ self.assertEqual(self._read(target).column("age").to_pylist(), [1])
+
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.
