[python][ray] Fix update_by_row_id failure on empty datasets

.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

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 _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 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()),

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()),

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.