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 diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py b/paimon-python/pypaimon/ray/data_evolution_merge_join.py index 359af8e6efa2..4a0dd74b3528 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 _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()), 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()), 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.