[python][ray] Rebase exact-range self-merge updates

Author: XiaoHongbo-Hope

paimon-python/pypaimon/ray/row_id_conflict_rewriter.py

@@ -86,11 +86,11 @@ def commit_self_merge_with_compaction_retry(
         if result is not None:
             current_updates = result.update_messages
             logger.info(
-                "Rewrote %d stale self-merge file(s) against snapshot %d "
-                "before committing to table %s.",
-                result.rewritten_file_count,
+                "Rebased stale self-merge updates against snapshot %d "
+                "before committing to table %s; rewrote %d file(s).",
                 latest_snapshot.id,
                 table.identifier,
+                result.rewritten_file_count,
             )
 
     # Match Spark's PaimonSparkWriter: every outer rebase attempt creates a
@@ -152,11 +152,11 @@ def commit_self_merge_with_compaction_retry(
             raise conflict
 
         logger.info(
-            "Rewrote %d stale self-merge file(s) against snapshot %d "
-            "before retrying commit to table %s.",
-            result.rewritten_file_count,
+            "Rebased stale self-merge updates against snapshot %d "
+            "before retrying commit to table %s; rewrote %d file(s).",
             latest_snapshot.id,
             table.identifier,
+            result.rewritten_file_count,
         )
         _retry_wait(table, retry_count)
         retry_count += 1
@@ -241,8 +241,9 @@ def _rewrite_updates(
             latest_snapshot.next_row_id,
         )
     ]
-    if not candidates:
-        return None
+    # Compaction can replace an anchor while preserving its exact row-id
+    # range. There is nothing to rewrite then, but the snapshot check below
+    # still needs to advance past that compaction.
     if not _ranges_are_still_covered(current_files, candidates):
         return None
 

paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py

@@ -115,16 +115,18 @@ def _snapshot_id(self, target):
         return snap.id if snap is not None else None
 
     def _compact_all_data_files(self, table):
-        """Replace all current data files with one COMPACT output file."""
+        """Replace current data files with one COMPACT file per file group."""
         from pypaimon.table.special_fields import SpecialFields
 
         read_builder = table.new_read_builder().with_projection(
             list(table.field_names) + [SpecialFields.ROW_ID.name]
         )
         plan = read_builder.new_scan().plan_for_write()
-        old_files = [
-            file for split in plan.splits() for file in split.files
-        ]
+        old_files_by_group = {}
+        for split in plan.splits():
+            old_files_by_group.setdefault(
+                (tuple(split.partition.values), split.bucket), [],
+            ).extend(split.files)
         current = read_builder.new_read().to_arrow(plan.splits()).sort_by(
             [(SpecialFields.ROW_ID.name, 'ascending')]
         ).select(list(table.field_names))
@@ -133,12 +135,20 @@ def _compact_all_data_files(self, table):
         writer = write_builder.new_write()
         writer.write_arrow(current)
         messages = writer.prepare_commit()
-        self.assertEqual(len(messages), 1)
-        self.assertEqual(len(messages[0].new_files), 1)
-        messages[0].new_files = [
-            messages[0].new_files[0].assign_first_row_id(0)
-        ]
-        messages[0].deleted_files.extend(old_files)
+        self.assertEqual(
+            {(message.partition, message.bucket) for message in messages},
+            set(old_files_by_group),
+        )
+        for message in messages:
+            old_files = old_files_by_group[
+                (message.partition, message.bucket)
+            ]
+            self.assertEqual(len(message.new_files), 1)
+            first_row_id = min(file.first_row_id for file in old_files)
+            message.new_files = [
+                message.new_files[0].assign_first_row_id(first_row_id)
+            ]
+            message.deleted_files.extend(old_files)
 
         commit = write_builder.new_commit()
         file_store_commit = commit.file_store_commit
@@ -2376,7 +2386,11 @@ def test_self_merge_rebases_staged_updates_after_compaction(self):
         target = 'default.tbl_{}'.format(uuid.uuid4().hex[:8])
         self.catalog.create_table(
             target,
-            Schema.from_pyarrow_schema(schema, options=options),
+            Schema.from_pyarrow_schema(
+                schema,
+                partition_keys=['content_key'],
+                options=options,
+            ),
             False,
         )