This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 679aa01908 [python] Fix flaky
test_concurrent_updates_overlapping_rows_last_writer_wins (#8733)
679aa01908 is described below
commit 679aa0190871c5ed6008ccad466fd85469ae6801
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Jul 20 11:29:50 2026 +0800
[python] Fix flaky
test_concurrent_updates_overlapping_rows_last_writer_wins (#8733)
---
paimon-python/pypaimon/tests/table_update_test.py | 15 ++++++++-------
1 file changed, 8 insertions(+), 7 deletions(-)
diff --git a/paimon-python/pypaimon/tests/table_update_test.py
b/paimon-python/pypaimon/tests/table_update_test.py
index adb08dd995..793871e11a 100644
--- a/paimon-python/pypaimon/tests/table_update_test.py
+++ b/paimon-python/pypaimon/tests/table_update_test.py
@@ -1093,9 +1093,12 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
)
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 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
{'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):
