From 1e784e98e59ec02ec50950e322d6eb9eda1a8233 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 14 Aug 2026 00:46:02 -0700 Subject: [PATCH 1/4] [python][ray] Handle empty row-id update datasets --- .../pypaimon/ray/data_evolution_merge_join.py | 18 +++++++++++++++++ .../tests/ray_update_by_row_id_test.py | 20 +++++++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py b/paimon-python/pypaimon/ray/data_evolution_merge_join.py index 359af8e6efa2..829d14502143 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,10 +545,27 @@ def _assign_frid(batch: pa.Table) -> pa.Table: 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 and is removed before writing. + 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([captured_sorted[0]], 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: + group = group.filter( + pc.not_equal(group.column(row_id_name), sentinel_row_id) + ) if group.num_rows == 0: return pa.Table.from_pydict({ "msgs_blob": pa.array([], type=pa.binary()), 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 613fa50d0319..801c35218af9 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 @@ def test_updates_correct_row_across_files(self): 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. From 6b84024d3da7d2d75fd8ad77450c04eaa0206839 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 14 Aug 2026 01:21:10 -0700 Subject: [PATCH 2/4] [python][ray] Test empty merge update results --- .../ray_data_evolution_merge_into_test.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) 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 cddb11ac37b1..2e1e8b8e4486 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 @@ def test_empty_target_matched_update_is_noop(self): 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()), From 33df27bd2bd10e01a64f056ad6aeaa267498c602 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 14 Aug 2026 01:28:00 -0700 Subject: [PATCH 3/4] [python][ray] Isolate the empty-update sentinel group --- paimon-python/pypaimon/ray/data_evolution_merge_join.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py b/paimon-python/pypaimon/ray/data_evolution_merge_join.py index 829d14502143..4a0dd74b3528 100644 --- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py +++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py @@ -549,13 +549,13 @@ def _assign_frid(batch: pa.Table) -> pa.Table: 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 and is removed before writing. + # 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([captured_sorted[0]], type=pa.int64()) + frid_col, pa.array([sentinel_row_id], type=pa.int64()) ) with_frid = with_frid.union(ray.data.from_arrow(sentinel)) @@ -563,10 +563,7 @@ def _assign_frid(batch: pa.Table) -> pa.Table: captured_cols = cols def _apply_group(group: pa.Table) -> pa.Table: - group = group.filter( - pc.not_equal(group.column(row_id_name), sentinel_row_id) - ) - 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()), From 5fba2c1906446bd3d1c11cf68a10ed8a588189d0 Mon Sep 17 00:00:00 2001 From: "xiaohongbo.xhb" Date: Fri, 14 Aug 2026 01:52:19 -0700 Subject: [PATCH 4/4] [python][ray] Test empty updates on Ray 2.50 --- .github/workflows/paimon-python-checks.yml | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/.github/workflows/paimon-python-checks.yml b/.github/workflows/paimon-python-checks.yml index f28e61fe71d7..2bb90143dec0 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