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 f269606c1a [python] Remove dynamic bucket base snapshot commit guard
(#10075)
f269606c1a is described below
commit f269606c1aa3c2679bc09c9aadc92536e778bfc4
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 22 11:47:36 2026 +0800
[python] Remove dynamic bucket base snapshot commit guard (#10075)
---
.../pypaimon/tests/write/dynamic_bucket_test.py | 67 ++++++++--------------
paimon-python/pypaimon/write/commit_message.py | 1 -
paimon-python/pypaimon/write/file_store_commit.py | 39 -------------
paimon-python/pypaimon/write/table_write.py | 9 ---
4 files changed, 23 insertions(+), 93 deletions(-)
diff --git a/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
b/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
index be175f1e39..7eb7f2049e 100644
--- a/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
+++ b/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
@@ -503,7 +503,7 @@ class DynamicBucketTest(unittest.TestCase):
for message in messages2 for entry in message.index_adds
]
with self.assertRaisesRegex(
- RuntimeError, 'HASH index assignment conflict'
+ RuntimeError, 'HASH index conflict detected'
):
commit2.commit(messages2)
@@ -544,7 +544,7 @@ class DynamicBucketTest(unittest.TestCase):
commit1.commit(messages1)
with self.assertRaisesRegex(
- RuntimeError, 'HASH index assignment conflict'
+ RuntimeError, 'HASH index conflict detected'
):
commit2.commit(messages2)
@@ -556,7 +556,7 @@ class DynamicBucketTest(unittest.TestCase):
commit1.close()
commit2.close()
- def test_concurrent_disjoint_bucket_replacements_conflict(self):
+ def test_concurrent_disjoint_bucket_replacements_succeed(self):
with tempfile.TemporaryDirectory() as root:
table = self._create_table(
root,
@@ -583,41 +583,28 @@ class DynamicBucketTest(unittest.TestCase):
return_value=1,
):
writer2, commit2, messages2 = self._prepare_indexed_write(
- table, [3]
+ table, [4]
)
self.assertNotEqual(messages1[0].bucket, messages2[0].bucket)
commit1.commit(messages1)
- stale_paths = [
- file.file_path
- for message in messages2
- for file in message.new_files
- ] + [
- entry.index_file.external_path
- or table.path_factory().global_index_path_factory().to_path(
- entry.index_file.file_name
- )
- for message in messages2
- for entry in message.index_adds
- ]
+ commit2.commit(messages2)
- with self.assertRaisesRegex(
- RuntimeError, 'assigned from snapshot.*latest snapshot'
- ):
- commit2.commit(messages2)
-
- self.assertTrue(all(
- table.file_io.exists(path) for path in stale_paths
- ))
+ self.assertEqual(2, len(self._hash_indexes(table)))
+ self.assertEqual(
+ {'id': [1, 2, 3, 4],
+ 'value': ['v-1', 'v-2', 'v-3', 'v-4']},
+ self._read_arrow(table).sort_by('id').to_pydict(),
+ )
writer1.close()
writer2.close()
commit1.close()
commit2.close()
- def test_data_only_upsert_conflicts_after_overwrite_remaps_key(self):
+ def test_data_only_upsert_succeeds_after_concurrent_index_change(self):
with tempfile.TemporaryDirectory() as root:
table = self._create_table(
- root, 'data_only_overwrite_conflict', target_row_num=1
+ root, 'data_only_concurrent_append', target_row_num=1
)
self._commit_arrow(table, [1, 2], ['one', 'two'])
@@ -633,25 +620,17 @@ class DynamicBucketTest(unittest.TestCase):
for message in stale_messages
))
- overwrite_builder = table.new_batch_write_builder().overwrite({})
- overwrite_writer = overwrite_builder.new_write()
- overwrite_writer.write_arrow(
- pa.table({'id': [2], 'value': ['overwrite']})
- )
- overwrite_messages = overwrite_writer.prepare_commit()
- overwrite_commit = overwrite_builder.new_commit()
- overwrite_commit.commit(overwrite_messages)
- overwrite_writer.close()
- overwrite_commit.close()
-
- with self.assertRaisesRegex(
- RuntimeError, 'HASH index assignment conflict'
- ):
- stale_commit.commit(stale_messages)
+ concurrent_messages = self._commit_arrow(table, [3], ['three'])
+ self.assertTrue(any(
+ message.index_adds or message.index_deletes
+ for message in concurrent_messages
+ ))
+ stale_commit.commit(stale_messages)
self.assertEqual(
- {'id': [2], 'value': ['overwrite']},
- self._read_arrow(table).to_pydict(),
+ {'id': [1, 2, 3],
+ 'value': ['one', 'stale-upsert', 'three']},
+ self._read_arrow(table).sort_by('id').to_pydict(),
)
stale_writer.close()
stale_commit.close()
@@ -696,7 +675,7 @@ class DynamicBucketTest(unittest.TestCase):
'_commit_retry_wait',
):
with self.assertRaisesRegex(
- RuntimeError, 'HASH index assignment conflict'
+ RuntimeError, 'HASH index conflict detected'
):
commit.commit(messages)
diff --git a/paimon-python/pypaimon/write/commit_message.py
b/paimon-python/pypaimon/write/commit_message.py
index c170d0d216..e3c6e15287 100644
--- a/paimon-python/pypaimon/write/commit_message.py
+++ b/paimon-python/pypaimon/write/commit_message.py
@@ -34,7 +34,6 @@ class CommitMessage:
index_adds: List['IndexManifestEntry'] = field(default_factory=list)
index_deletes: List['IndexManifestEntry'] = field(default_factory=list)
changelog_files: List[DataFileMeta] = field(default_factory=list)
- hash_index_base_snapshot: Optional[int] = None
total_buckets: Optional[int] = None
def is_empty(self):
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index d56fe5acf2..a06c6e9d63 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -294,10 +294,6 @@ class FileStoreCommit:
for msg in commit_messages:
index_deletes.extend(msg.index_deletes)
index_adds.extend(msg.index_adds)
- hash_index_base_snapshot = self._hash_index_base_snapshot(
- commit_messages
- )
-
if not index_deletes:
from pypaimon.write.global_index_update_checker import (
apply_global_index_update_action,
@@ -351,7 +347,6 @@ class FileStoreCommit:
allow_rollback=allow_rollback,
index_deletes=index_deletes,
index_adds=index_adds,
- hash_index_base_snapshot=hash_index_base_snapshot,
snapshot_properties=snapshot_properties,
allow_empty_commit=not ignore_empty_commit)
@@ -386,10 +381,6 @@ class FileStoreCommit:
index_deletes = [
entry for message in commit_messages for entry in
message.index_deletes
]
- hash_index_base_snapshot = self._hash_index_base_snapshot(
- commit_messages
- )
-
if not skip_overwrite:
index_deletes = self._overwrite_hash_index_deletes(
partition_filter, index_deletes
@@ -404,24 +395,9 @@ class FileStoreCommit:
allow_rollback=False,
index_deletes=index_deletes,
index_adds=index_adds,
- hash_index_base_snapshot=hash_index_base_snapshot,
snapshot_properties=snapshot_properties,
)
- @staticmethod
- def _hash_index_base_snapshot(
- commit_messages: List[CommitMessage],
- ) -> Optional[int]:
- # Include data-only dynamic-bucket upserts. Their existing mappings
- # are stable across append commits, but a concurrent overwrite may
- # rebuild the HASH index and move a key to another bucket.
- base_snapshots = [
- getattr(message, "hash_index_base_snapshot", None)
- for message in commit_messages
- if getattr(message, "hash_index_base_snapshot", None) is not None
- ]
- return min(base_snapshots) if base_snapshots else None
-
def _overwrite_hash_index_deletes(self, partition_filter, deletes):
"""Delete HASH indexes for every partition replaced by overwrite."""
from pypaimon.index.dynamic_bucket import HASH_INDEX
@@ -511,7 +487,6 @@ class FileStoreCommit:
def _try_commit(self, commit_kind, commit_identifier, commit_entries_plan,
detect_conflicts=False, allow_rollback=False,
index_deletes=None,
index_adds=None, changelog_entries=None,
- hash_index_base_snapshot=None,
snapshot_properties: Optional[Dict[str, str]] = None,
allow_empty_commit=False):
@@ -553,7 +528,6 @@ class FileStoreCommit:
allow_rollback=allow_rollback,
index_deletes=index_deletes,
index_adds=index_adds,
- hash_index_base_snapshot=hash_index_base_snapshot,
commit_result_may_be_uncertain=commit_result_may_be_uncertain,
snapshot_properties=snapshot_properties,
)
@@ -633,7 +607,6 @@ class FileStoreCommit:
allow_rollback: bool = False,
index_deletes=None,
index_adds=None,
- hash_index_base_snapshot=None,
commit_result_may_be_uncertain: bool = False,
snapshot_properties: Optional[Dict[str, str]] = None
) -> CommitResult:
@@ -646,18 +619,6 @@ class FileStoreCommit:
notify_callbacks=True):
return SuccessResult()
- latest_snapshot_id = latest_snapshot.id if latest_snapshot else 0
- if (
- hash_index_base_snapshot is not None
- and latest_snapshot_id != hash_index_base_snapshot
- ):
- raise RuntimeError(
- "HASH index assignment conflict detected: assigned from "
- "snapshot {}, but the latest snapshot is {}.".format(
- hash_index_base_snapshot, latest_snapshot_id
- )
- )
-
unique_id = uuid.uuid4()
base_manifest_list = f"manifest-list-{unique_id}-0"
delta_manifest_list = f"manifest-list-{unique_id}-1"
diff --git a/paimon-python/pypaimon/write/table_write.py
b/paimon-python/pypaimon/write/table_write.py
index d9952726e0..37ee3fe4c8 100644
--- a/paimon-python/pypaimon/write/table_write.py
+++ b/paimon-python/pypaimon/write/table_write.py
@@ -319,9 +319,6 @@ class TableWrite:
return commit_messages
index_changes = prepare_indexes()
- base_snapshot_id = getattr(
- self.row_key_extractor, "base_snapshot_id", None
- )
messages_by_bucket = {
(tuple(message.partition), message.bucket): message
for message in commit_messages
@@ -338,12 +335,6 @@ class TableWrite:
messages_by_bucket[(partition, bucket)] = message
message.index_adds.extend(changes.additions)
message.index_deletes.extend(changes.deletions)
- if base_snapshot_id is not None:
- # Data-only upserts must participate too. A concurrent overwrite
- # can rebuild the HASH index and move an existing key, making a
- # stale data file unsafe even when this writer added no mapping.
- for message in commit_messages:
- message.hash_index_base_snapshot = base_snapshot_id
return commit_messages
def _release_prepared_indexes(self) -> None:
