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 ae83ef3182 [python] Reject overlapping row-id update batches (#9484)
ae83ef3182 is described below
commit ae83ef318293ff1234f2158e7c10bd5e752a1d42
Author: Yann Byron <[email protected]>
AuthorDate: Mon Aug 31 17:27:10 2026 +0800
[python] Reject overlapping row-id update batches (#9484)
---
paimon-python/pypaimon/tests/table_update_test.py | 75 +++++++++++++++++++++-
paimon-python/pypaimon/write/table_update.py | 7 +-
.../pypaimon/write/table_update_by_row_id.py | 18 ++++++
3 files changed, 98 insertions(+), 2 deletions(-)
diff --git a/paimon-python/pypaimon/tests/table_update_test.py
b/paimon-python/pypaimon/tests/table_update_test.py
index 9b183be85d..130dd53686 100644
--- a/paimon-python/pypaimon/tests/table_update_test.py
+++ b/paimon-python/pypaimon/tests/table_update_test.py
@@ -32,6 +32,7 @@ from pypaimon.tests.data_evolution_test_helpers import (
StreamModeMixin,
)
from pypaimon.write.table_update import BatchTableUpdate
+from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
# ======================================================================
@@ -71,6 +72,30 @@ def test_batch_row_id_update_batches_reuse_file_index():
]
+def test_batch_row_id_update_detects_overlap_before_write():
+ updater = TableUpdateByRowId.__new__(TableUpdateByRowId)
+ updater.table = mock.Mock(field_names=['age'])
+ updater.commit_messages = []
+ updater._updated_first_row_ids_by_column = {}
+
+ def route(data):
+ return data.append_column(
+ TableUpdateByRowId.FIRST_ROW_ID_COLUMN,
+ pa.array([0], type=pa.int64()),
+ )
+
+ with mock.patch.object(
+ updater, '_calculate_first_row_id', side_effect=route):
+ with mock.patch.object(updater, '_write_by_first_row_id') as write:
+ updater.update_columns(
+ pa.table({'_ROW_ID': [0], 'age': [26]}), ['age'])
+ with pytest.raises(ValueError, match='overlapping first_row_ids'):
+ updater.update_columns(
+ pa.table({'_ROW_ID': [1], 'age': [31]}), ['age'])
+
+ assert write.call_count == 1
+
+
class _TableUpdateTestBase(DataEvolutionTestBase):
"""Shared tests for ``TableUpdate.update_by_arrow_with_row_id``.
@@ -1606,6 +1631,50 @@ class _StreamModeMixin(StreamModeMixin):
class TableUpdateBatchTest(_BatchModeMixin, _TableUpdateTestBase,
unittest.TestCase):
"""All shared update tests under batch (``BatchWriteBuilder``)
semantics."""
+ def test_update_batches_reject_same_file_and_abort_staged_files(self):
+ table = self._create_seeded_table()
+ before_files = self._list_table_files(table)
+ update = (
+ self._make_write_builder(table)
+ .new_update()
+ .with_update_type(['age'])
+ )
+
+ with self.assertRaisesRegex(
+ ValueError, "overlapping first_row_ids.*0"):
+ update.update_by_arrow_batches_with_row_id(iter([
+ pa.Table.from_pydict({'_ROW_ID': [0], 'age': [26]}),
+ pa.Table.from_pydict({'_ROW_ID': [1], 'age': [31]}),
+ ]))
+
+ self.assertEqual(before_files, self._list_table_files(table))
+ self.assertEqual(
+ [25, 30, 35, 40, 45],
+ self._read_all(table)['age'].to_pylist(),
+ )
+
+ def test_update_batches_allow_same_file_for_different_columns(self):
+ table = self._create_seeded_table()
+ builder = self._make_write_builder(table)
+ messages = (
+ builder.new_update()
+ .update_by_arrow_batches_with_row_id(iter([
+ pa.Table.from_pydict({'_ROW_ID': [0], 'age': [26]}),
+ pa.Table.from_pydict({'_ROW_ID': [1], 'city': ['Seattle']}),
+ ]))
+ )
+
+ commit = builder.new_commit()
+ commit.commit(messages)
+ commit.close()
+
+ rows = self._read_all(table)
+ self.assertEqual([26, 30, 35, 40, 45], rows['age'].to_pylist())
+ self.assertEqual(
+ ['NYC', 'Seattle', 'Chicago', 'Houston', 'Phoenix'],
+ rows['city'].to_pylist(),
+ )
+
def test_callable_output_preserves_large_offset_chunks(self):
from pypaimon.write.table_update import TableUpdate
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
@@ -1629,13 +1698,17 @@ class TableUpdateBatchTest(_BatchModeMixin,
_TableUpdateTestBase, unittest.TestC
updater = TableUpdateByRowId.__new__(TableUpdateByRowId)
updater.table = mock.Mock(field_names=['payload'])
updater.commit_messages = []
+ updater._updated_first_row_ids_by_column = {}
updates = pa.Table.from_arrays(
[pa.array([1, 0], type=pa.int64()), result],
names=['_ROW_ID', 'payload'],
)
with mock.patch.object(
updater, '_calculate_first_row_id',
- side_effect=lambda data: data) as calculate:
+ side_effect=lambda data: data.append_column(
+ TableUpdateByRowId.FIRST_ROW_ID_COLUMN,
+ pa.array([0, 0], type=pa.int64()),
+ )) as calculate:
with mock.patch.object(updater, '_write_by_first_row_id'):
updater.update_columns(updates, ['payload'])
diff --git a/paimon-python/pypaimon/write/table_update.py
b/paimon-python/pypaimon/write/table_update.py
index a9404b32fe..c358644460 100644
--- a/paimon-python/pypaimon/write/table_update.py
+++ b/paimon-python/pypaimon/write/table_update.py
@@ -647,7 +647,12 @@ class BatchTableUpdate(TableUpdate):
def update_by_arrow_batches_with_row_id(
self, tables: Iterable[pa.Table]
) -> List[CommitMessage]:
- """Apply row-id updates from batches using one target-file index."""
+ """Apply row-id updates from batches using one target-file index.
+
+ For each updated column, batches must target disjoint ``first_row_id``
+ file groups. Conflicting overlap is rejected and all files staged by
+ earlier batches are aborted.
+ """
return self._update_by_arrow_batches_with_row_id(
tables, BATCH_COMMIT_IDENTIFIER)
diff --git a/paimon-python/pypaimon/write/table_update_by_row_id.py
b/paimon-python/pypaimon/write/table_update_by_row_id.py
index 85ccfb4c6e..da9b1a8cdc 100644
--- a/paimon-python/pypaimon/write/table_update_by_row_id.py
+++ b/paimon-python/pypaimon/write/table_update_by_row_id.py
@@ -99,6 +99,7 @@ class TableUpdateByRowId:
self.valid_row_id_ranges = info.valid_row_id_ranges
self.commit_messages: List[CommitMessage] = []
+ self._updated_first_row_ids_by_column: Dict[str, Set[int]] = {}
def _snapshot_files_info(self) -> _FilesInfo:
"""Return the already loaded snapshot file index for broadcast."""
@@ -219,7 +220,24 @@ class TableUpdateByRowId:
raise ValueError(f"Column {col_name} not found in table
schema")
data_with_first_row_id = self._calculate_first_row_id(data)
+ first_row_ids = set(
+ data_with_first_row_id[self.FIRST_ROW_ID_COLUMN].to_pylist()
+ )
+ overlapping = {}
+ for col_name in column_names:
+ ids = first_row_ids.intersection(
+ self._updated_first_row_ids_by_column.get(col_name, set()))
+ if ids:
+ overlapping[col_name] = sorted(ids)
+ if overlapping:
+ raise ValueError(
+ "Input batches contain overlapping first_row_ids by column: "
+ f"{overlapping}"
+ )
self._write_by_first_row_id(data_with_first_row_id, column_names)
+ for col_name in column_names:
+ self._updated_first_row_ids_by_column.setdefault(
+ col_name, set()).update(first_row_ids)
return self.commit_messages
