[python] Fix flaky test_concurrent_updates_overlapping_rows_last_writer_wins

The test derived the winner from completion_order[-1] (the order threads
returned in), which is not the commit order, so under CI load it diverged
from the persisted data and failed intermittently. Tag each thread's
commits with a distinct commit.user-prefix and read the true winner from
the latest snapshot's commit_user, then assert the persisted rows equal
that thread's whole update and the untouched rows keep their seed values.

Author: XiaoHongbo-Hope

paimon-python/pypaimon/tests/table_update_test.py

@@ -1093,9 +1093,12 @@ def _run_concurrent_updates(self, table, thread_specs, max_retries):
         )
 
         def worker(idx, spec):
+            # Tag each thread's commits so the durable winner can be read back from the
+            # latest snapshot's commit_user (the order threads return in is not the commit order).
+            worker_table = table.copy({'commit.user-prefix': 'w%d' % idx})
             for _ in range(max_retries):
                 try:
-                    self._do_update(table, pa.Table.from_pydict({
+                    self._do_update(worker_table, pa.Table.from_pydict({
                         '_ROW_ID': spec['row_ids'],
                         'age': spec['ages'],
                     }), ['age'])
@@ -1142,13 +1145,11 @@ def test_concurrent_updates_overlapping_rows_last_writer_wins(self):
             {'row_ids': [0, 1, 2], 'ages': [102, 202, 302]},
             {'row_ids': [0, 1, 2], 'ages': [103, 203, 303]},
         ]
-        completion_order = self._run_concurrent_updates(
-            table, specs, max_retries=30
-        )
-        winner = specs[completion_order[-1]]['ages']
+        self._run_concurrent_updates(table, specs, max_retries=30)
         ages = self._read_all(table)['age'].to_pylist()
-        self.assertEqual(winner, ages[:3])
-        # Rows 3 & 4 must remain at seed values
+        # The real winner is the thread whose commit produced the latest snapshot.
+        winner = int(table.snapshot_manager().get_latest_snapshot().commit_user[1:].split('_')[0])
+        self.assertEqual(specs[winner]['ages'], ages[:3])
         self.assertEqual([40, 45], ages[3:])
 
     def test_update_list_and_map_columns(self):