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 6464ef3011 [python] Recover row-id updates after concurrent compaction
(#8915)
6464ef3011 is described below
commit 6464ef3011170a9a72452f6292d7a234e300e14a
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Jul 30 16:01:31 2026 +0800
[python] Recover row-id updates after concurrent compaction (#8915)
---
docs/docs/pypaimon/data-evolution.md | 27 ++
docs/generated/core_configuration.html | 6 +
.../main/java/org/apache/paimon/CoreOptions.java | 14 +
.../pypaimon/common/options/core_options.py | 19 ++
.../pypaimon/tests/e2e/java_py_read_write_test.py | 17 +-
.../pypaimon/tests/file_store_commit_test.py | 38 ++-
.../pypaimon/tests/table_upsert_by_key_test.py | 261 ++++++++++++++++++
.../pypaimon/write/commit/conflict_detection.py | 40 +--
.../write/commit/row_id_conflict_rewriter.py | 293 +++++++++++++++++++++
paimon-python/pypaimon/write/file_store_commit.py | 102 ++++++-
.../pypaimon/write/table_update_by_row_id.py | 43 ++-
11 files changed, 826 insertions(+), 34 deletions(-)
diff --git a/docs/docs/pypaimon/data-evolution.md
b/docs/docs/pypaimon/data-evolution.md
index 43f31c63b8..279e942357 100644
--- a/docs/docs/pypaimon/data-evolution.md
+++ b/docs/docs/pypaimon/data-evolution.md
@@ -577,6 +577,33 @@ commit.close()
same order for that shard.
- **Parallelism**: run multiple shards by calling
`new_shard_updator(shard_idx, num_shards)` for each shard.
+## Concurrent Compaction Recovery
+
+A partial-column update records the row-ID boundary of each data file it read.
+If compaction merges those files before `commit`, PyPaimon automatically
+rebases regular (non-BLOB and non-VECTOR) staged update files onto the latest
+file boundaries and retries the commit.
+
+The recovery is bounded by the total size of the current data files whose
+row-ID ranges are affected:
+
+```python
+options = {
+ 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true',
+ 'data-evolution.row-id-conflict-rewrite.max-size': '256 MB',
+}
+```
+
+The default is `256 MB`. Set the option to `0 B` to disable automatic
+rewriting. If the affected files exceed the configured size, or if the row IDs
+were removed by an overwrite, the commit keeps the normal
+`Row ID existence conflict` behavior. Logical concurrent updates are still
+checked and are never hidden by compaction recovery.
+
+Recovery is not attempted when deletion vectors are enabled, or when the same
+commit contains existing-row BLOB or VECTOR staged files.
+
## Stream Mode
Data evolution also supports stream mode. The operation semantics are the same
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index 5670cc65c8..78d3e5a168 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -518,6 +518,12 @@ under the License.
<td>Long</td>
<td>Strictly contiguous same-partition logical row-id runs
containing more than this number of rows are excluded from row-id reassignment.
Set to 0 to disable this filtering.</td>
</tr>
+ <tr>
+ <td><h5>data-evolution.row-id-conflict-rewrite.max-size</h5></td>
+ <td style="word-wrap: break-word;">256 mb</td>
+ <td>MemorySize</td>
+ <td>Maximum total size of current data files whose row-id ranges
PyPaimon may automatically rebase staged updates against when a concurrent
compaction changes file boundaries. Set to 0 B to disable.</td>
+ </tr>
<tr>
<td><h5>data-evolution.row-sidecar.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 7449142e8f..a405d1318e 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -2506,6 +2506,16 @@ public class CoreOptions implements Serializable {
+ "more than this number of rows are
excluded from row-id "
+ "reassignment. Set to 0 to disable this
filtering.");
+ public static final ConfigOption<MemorySize>
DATA_EVOLUTION_ROW_ID_CONFLICT_REWRITE_MAX_SIZE =
+ key("data-evolution.row-id-conflict-rewrite.max-size")
+ .memoryType()
+ .defaultValue(MemorySize.ofMebiBytes(256))
+ .withDescription(
+ "Maximum total size of current data files whose
row-id ranges PyPaimon "
+ + "may automatically rebase staged updates
against when a "
+ + "concurrent compaction changes file
boundaries. Set to 0 B "
+ + "to disable.");
+
public static final ConfigOption<Boolean>
DATA_EVOLUTION_ROW_SIDECAR_ENABLED =
key("data-evolution.row-sidecar.enabled")
.booleanType()
@@ -4235,6 +4245,10 @@ public class CoreOptions implements Serializable {
return threshold;
}
+ public long dataEvolutionRowIdConflictRewriteMaxSize() {
+ return
options.get(DATA_EVOLUTION_ROW_ID_CONFLICT_REWRITE_MAX_SIZE).getBytes();
+ }
+
public boolean dataEvolutionRowSidecarEnabled() {
return options.get(DATA_EVOLUTION_ROW_SIDECAR_ENABLED);
}
diff --git a/paimon-python/pypaimon/common/options/core_options.py
b/paimon-python/pypaimon/common/options/core_options.py
index f53c3722b4..8ee6016e64 100644
--- a/paimon-python/pypaimon/common/options/core_options.py
+++ b/paimon-python/pypaimon/common/options/core_options.py
@@ -659,6 +659,18 @@ class CoreOptions:
.with_description("Whether to enable data evolution.")
)
+ DATA_EVOLUTION_ROW_ID_CONFLICT_REWRITE_MAX_SIZE: ConfigOption[MemorySize]
= (
+ ConfigOptions.key("data-evolution.row-id-conflict-rewrite.max-size")
+ .memory_type()
+ .default_value(MemorySize.of_mebi_bytes(256))
+ .with_description(
+ "Maximum total size of current data files whose row-id ranges "
+ "PyPaimon may automatically rebase staged updates against when "
+ "a concurrent compaction changes file boundaries. Set to 0 B "
+ "to disable."
+ )
+ )
+
DATA_EVOLUTION_ROW_SIDECAR_ENABLED: ConfigOption[bool] = (
ConfigOptions.key("data-evolution.row-sidecar.enabled")
.boolean_type()
@@ -1261,6 +1273,13 @@ class CoreOptions:
def data_evolution_enabled(self, default=None):
return self.options.get(CoreOptions.DATA_EVOLUTION_ENABLED, default)
+ def data_evolution_row_id_conflict_rewrite_max_size(self, default=None):
+ value = self.options.get(
+ CoreOptions.DATA_EVOLUTION_ROW_ID_CONFLICT_REWRITE_MAX_SIZE,
+ default,
+ )
+ return value.get_bytes()
+
def data_evolution_row_sidecar_enabled(self, default=None):
return
self.options.get(CoreOptions.DATA_EVOLUTION_ROW_SIDECAR_ENABLED, default)
diff --git a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
index 89b9833f1e..33bfe83e19 100644
--- a/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
+++ b/paimon-python/pypaimon/tests/e2e/java_py_read_write_test.py
@@ -1613,7 +1613,7 @@ class JavaPyReadWriteTest(unittest.TestCase):
1. Java writes 5 base files (testCompactConflictWriteBase)
2. pypaimon ShardTableUpdator scans table, prepares evolution
3. Java runs compact (testCompactConflictRunCompact)
- 4. pypaimon commits stale evolution -> conflict detected, raises
RuntimeError
+ 4. pypaimon rebases the stale evolution files and commits successfully
"""
import subprocess
@@ -1659,13 +1659,18 @@ class JavaPyReadWriteTest(unittest.TestCase):
f"Java compact
failed:\n{result.stdout}\n{result.stderr}")
print("Java compact completed")
- # Step 4: pypaimon commits stale evolution -> conflict detected
+ # Step 4: pypaimon rewrites stale evolution files against the
compacted range
tc = wb.new_commit()
- with self.assertRaises(RuntimeError) as ctx:
- tc.commit(stale_commit_msgs)
- self.assertIn("conflict", str(ctx.exception))
+ tc.commit(stale_commit_msgs)
tc.close()
- print(f"Conflict detected as expected: {ctx.exception}")
+
+ read_builder = table.new_read_builder()
+ result = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits())
+ self.assertEqual(
+ rows_read,
+ sum(value is not None for value in
result.column('f2').to_pylist()),
+ )
def test_blob_compact_conflict_update(self):
import subprocess
diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py
b/paimon-python/pypaimon/tests/file_store_commit_test.py
index c6451f0a9b..88350bfd5f 100644
--- a/paimon-python/pypaimon/tests/file_store_commit_test.py
+++ b/paimon-python/pypaimon/tests/file_store_commit_test.py
@@ -24,8 +24,12 @@ from pypaimon.manifest.schema.data_file_meta import
DataFileMeta
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
from pypaimon.snapshot.snapshot_commit import PartitionStatistics
from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.write.commit.row_id_conflict_rewriter import RowIdRewriteResult
from pypaimon.write.commit_message import CommitMessage
-from pypaimon.write.file_store_commit import FileStoreCommit
+from pypaimon.write.file_store_commit import (
+ FileStoreCommit,
+ RewriteResult,
+)
@patch('pypaimon.write.file_store_commit.ManifestFileManager')
@@ -503,6 +507,38 @@ class TestFileStoreCommit(unittest.TestCase):
result = file_store_commit._write_manifest_files(entries,
"manifest-test")
self.assertIsNotNone(result)
+ def test_row_id_rewrite_respects_commit_retry_limit(
+ self, mock_manifest_list_manager, mock_manifest_file_manager):
+ file_store_commit = self._create_file_store_commit()
+ file_store_commit.commit_max_retries = 1
+ file_store_commit.commit_timeout = 10 ** 9
+ file_store_commit._commit_retry_wait = Mock()
+
+ latest_snapshot = Mock()
+ latest_snapshot.id = 7
+ file_store_commit.snapshot_manager.get_latest_snapshot.return_value = (
+ latest_snapshot
+ )
+
+ commit_entry = Mock()
+ rewrite = RewriteResult(RowIdRewriteResult([commit_entry], 1))
+ file_store_commit._try_commit_once = Mock(side_effect=[
+ rewrite,
+ rewrite,
+ AssertionError("rewrite retry budget was not enforced"),
+ ])
+
+ with self.assertRaises(RuntimeError) as ctx:
+ file_store_commit._try_commit(
+ commit_kind="APPEND",
+ commit_identifier=11,
+ commit_entries_plan=lambda snapshot: [commit_entry],
+ )
+
+ self.assertIn("with 1 retries", str(ctx.exception))
+ self.assertEqual(2, file_store_commit._try_commit_once.call_count)
+ file_store_commit._commit_retry_wait.assert_called_once_with(0)
+
@staticmethod
def _to_entries(commit_messages):
commit_entries = []
diff --git a/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
b/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
index bd05a276ff..feb98ae3c6 100644
--- a/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
+++ b/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
@@ -15,10 +15,13 @@
# specific language governing permissions and limitations
# under the License.
+import os
import unittest
+from unittest import mock
import pyarrow as pa
+from pypaimon.table.special_fields import SpecialFields
from pypaimon.tests.data_evolution_test_helpers import (
BatchModeMixin,
DataEvolutionTestBase,
@@ -87,6 +90,43 @@ class _TableUpsertByKeyTestBase(DataEvolutionTestBase):
tc.close()
return msgs
+ def _compact_all_data_files(self, table):
+ """Replace all current data files with one COMPACT output file."""
+ 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
+ ]
+ current = read_builder.new_read().to_arrow(plan.splits()).sort_by(
+ [(SpecialFields.ROW_ID.name, "ascending")]
+ ).select(list(table.field_names))
+
+ wb = table.new_batch_write_builder()
+ writer = wb.new_write()
+ writer.write_arrow(current)
+ messages = writer.prepare_commit()
+ self.assertEqual(1, len(messages))
+ self.assertEqual(1, len(messages[0].new_files))
+ messages[0].new_files = [
+ messages[0].new_files[0].assign_first_row_id(0)
+ ]
+ messages[0].deleted_files.extend(old_files)
+
+ commit = wb.new_commit()
+ file_store_commit = commit.file_store_commit
+ original_try_commit = file_store_commit._try_commit
+ file_store_commit._try_commit = (
+ lambda commit_kind, *args, **kwargs:
+ original_try_commit("COMPACT", *args, **kwargs)
+ )
+ commit.commit(messages)
+ writer.close()
+ commit.close()
+
# ==================================================================
# Basic upsert tests (non-partitioned)
# ==================================================================
@@ -359,6 +399,227 @@ class _TableUpsertByKeyTestBase(DataEvolutionTestBase):
self.assertEqual('Dave', rows[4])
self.assertEqual('Eve', rows[5])
+ def test_commit_rewrites_stale_update_after_compaction(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'name': ['Alice', 'Bob'],
+ 'age': [25, 30],
+ 'city': ['NYC', 'LA'],
+ }, schema=self.pa_schema))
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [3, 4],
+ 'name': ['Carol', 'Dave'],
+ 'age': [35, 40],
+ 'city': ['Chicago', 'Houston'],
+ }, schema=self.pa_schema))
+
+ wb = self._make_write_builder(table)
+ update = wb.new_update().with_update_type(['age', 'city'])
+ commit_identifier = self._next_commit_id()
+ messages = self._apply_upsert(
+ update,
+ pa.Table.from_pydict({
+ 'id': [2, 3],
+ 'name': ['ignored', 'ignored'],
+ 'age': [31, 36],
+ 'city': ['LA2', 'Chicago2'],
+ }, schema=self.pa_schema),
+ ['id'],
+ commit_identifier,
+ )
+ stale_paths = [
+ file.file_path
+ for message in messages
+ for file in message.new_files
+ ]
+
+ self._compact_all_data_files(table)
+
+ commit = wb.new_commit()
+ self._apply_commit(commit, messages, commit_identifier)
+ commit.close()
+
+ rows = {
+ row['id']: (row['name'], row['age'], row['city'])
+ for row in self._read_all(table).to_pylist()
+ }
+ self.assertEqual(('Bob', 31, 'LA2'), rows[2])
+ self.assertEqual(('Carol', 36, 'Chicago2'), rows[3])
+ self.assertEqual(('Dave', 40, 'Houston'), rows[4])
+ self.assertTrue(all(os.path.exists(path) for path in stale_paths))
+
+ def test_commit_rewrite_uses_checked_base_entries(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'name': ['Alice', 'Bob'],
+ 'age': [25, 30],
+ 'city': ['NYC', 'LA'],
+ }, schema=self.pa_schema))
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [3, 4],
+ 'name': ['Carol', 'Dave'],
+ 'age': [35, 40],
+ 'city': ['Chicago', 'Houston'],
+ }, schema=self.pa_schema))
+
+ wb = self._make_write_builder(table)
+ update = wb.new_update().with_update_type(['age'])
+ commit_identifier = self._next_commit_id()
+ messages = self._apply_upsert(
+ update,
+ pa.Table.from_pydict({
+ 'id': [2],
+ 'name': ['ignored'],
+ 'age': [31],
+ 'city': ['ignored'],
+ }, schema=self.pa_schema),
+ ['id'],
+ commit_identifier,
+ )
+ self._compact_all_data_files(table)
+
+ from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
+
+ original_build = TableUpdateByRowId._files_info_from_entries
+ advanced = [False]
+
+ def build_after_concurrent_compaction(
+ updater_cls, current_table, snapshot_id, entries):
+ if not advanced[0]:
+ advanced[0] = True
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [5],
+ 'name': ['Eve'],
+ 'age': [45],
+ 'city': ['Boston'],
+ }, schema=self.pa_schema))
+ self._compact_all_data_files(table)
+ return original_build(current_table, snapshot_id, entries)
+
+ with mock.patch.object(
+ TableUpdateByRowId,
+ '_load_existing_files_info',
+ side_effect=AssertionError("unexpected snapshot scan"),
+ ), mock.patch.object(
+ TableUpdateByRowId,
+ '_files_info_from_entries',
+ classmethod(build_after_concurrent_compaction)):
+ commit = wb.new_commit()
+ self._apply_commit(commit, messages, commit_identifier)
+ commit.close()
+
+ rows = {
+ row['id']: row['age']
+ for row in self._read_all(table).to_pylist()
+ }
+ self.assertEqual(31, rows[2])
+ self.assertEqual(45, rows[5])
+
+ def test_commit_rewrite_respects_max_size(self):
+ options = dict(self.table_options)
+ options[
+ 'data-evolution.row-id-conflict-rewrite.max-size'
+ ] = '1 B'
+ table = self._create_table(options=options)
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'name': ['Alice', 'Bob'],
+ 'age': [25, 30],
+ 'city': ['NYC', 'LA'],
+ }, schema=self.pa_schema))
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [3, 4],
+ 'name': ['Carol', 'Dave'],
+ 'age': [35, 40],
+ 'city': ['Chicago', 'Houston'],
+ }, schema=self.pa_schema))
+
+ wb = self._make_write_builder(table)
+ update = wb.new_update().with_update_type(['age'])
+ commit_identifier = self._next_commit_id()
+ messages = self._apply_upsert(
+ update,
+ pa.Table.from_pydict({
+ 'id': [2],
+ 'name': ['ignored'],
+ 'age': [31],
+ 'city': ['ignored'],
+ }, schema=self.pa_schema),
+ ['id'],
+ commit_identifier,
+ )
+ self._compact_all_data_files(table)
+
+ commit = wb.new_commit()
+ with self.assertRaises(RuntimeError) as ctx:
+ self._apply_commit(commit, messages, commit_identifier)
+ commit.close()
+ self.assertIn('Row ID existence conflict', str(ctx.exception))
+ self.assertIn(
+ 'data-evolution.row-id-conflict-rewrite.max-size',
+ str(ctx.exception),
+ )
+
+ def test_compaction_rewrite_does_not_hide_logical_update_conflict(self):
+ table = self._create_table()
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'name': ['Alice', 'Bob'],
+ 'age': [25, 30],
+ 'city': ['NYC', 'LA'],
+ }, schema=self.pa_schema))
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [3, 4],
+ 'name': ['Carol', 'Dave'],
+ 'age': [35, 40],
+ 'city': ['Chicago', 'Houston'],
+ }, schema=self.pa_schema))
+
+ wb = self._make_write_builder(table)
+ update = wb.new_update().with_update_type(['age'])
+ commit_identifier = self._next_commit_id()
+ stale_messages = self._apply_upsert(
+ update,
+ pa.Table.from_pydict({
+ 'id': [2],
+ 'name': ['ignored'],
+ 'age': [31],
+ 'city': ['ignored'],
+ }, schema=self.pa_schema),
+ ['id'],
+ commit_identifier,
+ )
+
+ self._upsert(
+ table,
+ pa.Table.from_pydict({
+ 'id': [2],
+ 'name': ['ignored'],
+ 'age': [99],
+ 'city': ['ignored'],
+ }, schema=self.pa_schema),
+ ['id'],
+ update_cols=['age'],
+ )
+ self._compact_all_data_files(table)
+
+ commit = wb.new_commit()
+ with self.assertRaises(RuntimeError):
+ self._apply_commit(
+ commit,
+ stale_messages,
+ commit_identifier,
+ )
+ commit.close()
+
+ rows = {
+ row['id']: row['age']
+ for row in self._read_all(table).to_pylist()
+ }
+ self.assertEqual(99, rows[2])
+
def test_large_table_upsert(self):
"""Upsert that touches a wide selection of rows in a 200-row table."""
table = self._create_table()
diff --git a/paimon-python/pypaimon/write/commit/conflict_detection.py
b/paimon-python/pypaimon/write/commit/conflict_detection.py
index 3534960c1c..3799223959 100644
--- a/paimon-python/pypaimon/write/commit/conflict_detection.py
+++ b/paimon-python/pypaimon/write/commit/conflict_detection.py
@@ -153,6 +153,23 @@ class CommitConflictError(RuntimeError):
"""A deterministic pre-snapshot conflict which is safe to abort."""
+class RowIdExistenceConflict(RuntimeError):
+ """A staged row-id file no longer matches the current base-file layout."""
+
+ def __init__(self, entry):
+ self.entry = entry
+ super().__init__(
+ "Row ID existence conflict: file '{}' references "
+ "firstRowId={}, rowCount={} in bucket {}, "
+ "but no matching file exists in the current snapshot. "
+ "The referenced file may have been rewritten by a "
+ "concurrent compaction or removed by an overwrite.".format(
+ entry.file.file_name,
+ entry.file.first_row_id,
+ entry.file.row_count,
+ entry.bucket))
+
+
class ConflictDetection:
"""Detects conflicts between base and delta files during commit."""
@@ -512,16 +529,7 @@ class ConflictDetection:
key = (entry.partition, entry.bucket,
entry.file.first_row_id, entry.file.row_count)
if key not in existing_index:
- return RuntimeError(
- "Row ID existence conflict: file '{}' references "
- "firstRowId={}, rowCount={} in bucket {}, "
- "but no matching file exists in the current snapshot. "
- "The referenced file may have been rewritten by a "
- "concurrent compaction or removed by an overwrite.".format(
- entry.file.file_name,
- entry.file.first_row_id,
- entry.file.row_count,
- entry.bucket))
+ return RowIdExistenceConflict(entry)
return None
@@ -635,7 +643,8 @@ class ConflictDetection:
count=entry.file.row_count,
)
- def check_row_id_from_snapshot(self, latest_snapshot, commit_entries):
+ def check_row_id_from_snapshot(
+ self, latest_snapshot, commit_entries, check_compaction=True):
if not self.data_evolution_enabled:
return None
if self._row_id_check_from_snapshot is None:
@@ -672,10 +681,11 @@ class ConflictDetection:
continue
if snapshot.commit_kind == "COMPACT":
- err = self._compact_conflicts_with_delta(
- snapshot, delta_signatures, column_checker, commit_entries)
- if err is not None:
- return err
+ if check_compaction:
+ err = self._compact_conflicts_with_delta(
+ snapshot, delta_signatures, column_checker,
commit_entries)
+ if err is not None:
+ return err
continue
incremental_entries =
self.commit_scanner.read_incremental_entries_from_changed_partitions(
diff --git a/paimon-python/pypaimon/write/commit/row_id_conflict_rewriter.py
b/paimon-python/pypaimon/write/commit/row_id_conflict_rewriter.py
new file mode 100644
index 0000000000..7eb623271e
--- /dev/null
+++ b/paimon-python/pypaimon/write/commit/row_id_conflict_rewriter.py
@@ -0,0 +1,293 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from dataclasses import dataclass
+from typing import List, Optional
+
+import pyarrow as pa
+
+from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.manifest_entry import ManifestEntry
+from pypaimon.read.split import DataSplit
+from pypaimon.read.table_read import TableRead
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.special_fields import SpecialFields
+from pypaimon.utils.range import Range
+from pypaimon.write.commit_message import CommitMessage
+from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
+
+
+@dataclass
+class RowIdRewriteResult:
+ commit_entries: List[ManifestEntry]
+ rewritten_file_count: int
+
+
+class RowIdConflictRewriter:
+ """Rebase stale partial-column files onto current row-id file ranges."""
+
+ def __init__(
+ self,
+ table,
+ commit_user: str,
+ commit_identifier: int,
+ max_rewrite_size: int):
+ self.table = table
+ self.commit_user = commit_user
+ self.commit_identifier = commit_identifier
+ self.max_rewrite_size = max_rewrite_size
+
+ def rewrite(
+ self,
+ latest_snapshot,
+ base_entries: List[ManifestEntry],
+ delta_entries: List[ManifestEntry],
+ ) -> Optional[RowIdRewriteResult]:
+ if self.max_rewrite_size <= 0:
+ return None
+ if self.table.options.deletion_vectors_enabled(False):
+ return None
+
+ current_exact_ranges = {
+ self._range_key(entry)
+ for entry in base_entries
+ if self._is_normal_row_id_file(entry.file)
+ }
+ candidates = [
+ entry
+ for entry in delta_entries
+ if self._is_rewrite_candidate(
+ entry, current_exact_ranges, latest_snapshot.next_row_id)
+ ]
+ if not candidates:
+ return None
+
+ candidate_ids = {id(entry) for entry in candidates}
+ if any(
+ self._is_dedicated_file(entry.file)
+ and entry.file.first_row_id is not None
+ and entry.file.first_row_id < latest_snapshot.next_row_id
+ for entry in delta_entries
+ ):
+ return None
+
+ if not self._ranges_are_still_covered(base_entries, candidates):
+ return None
+
+ affected_files = self._affected_current_files(base_entries, candidates)
+ affected_size = sum(file.file_size for file in affected_files)
+ if affected_size > self.max_rewrite_size:
+ raise RuntimeError(
+ "Automatic row-id conflict rewrite requires reading "
+ "{} bytes across {} current data file(s), exceeding "
+ "'data-evolution.row-id-conflict-rewrite.max-size'={} bytes."
+ .format(
+ affected_size,
+ len(affected_files),
+ self.max_rewrite_size,
+ )
+ )
+
+ new_messages = []
+ try:
+ files_info = TableUpdateByRowId._files_info_from_entries(
+ self.table,
+ latest_snapshot.id,
+ base_entries,
+ )
+ groups = self._group_by_write_columns(candidates)
+ for column_names, entries in groups:
+ update_tables = [
+ self._read_staged_update(entry, column_names)
+ for entry in entries
+ ]
+ update_data = (
+ update_tables[0]
+ if len(update_tables) == 1
+ else pa.concat_tables(update_tables)
+ )
+ row_ids = update_data[SpecialFields.ROW_ID.name].to_pylist()
+ if len(row_ids) != len(set(row_ids)):
+ raise RuntimeError(
+ "Automatic row-id conflict rewrite cannot merge "
+ "overlapping staged files for columns {}.".format(
+ column_names)
+ )
+ updater = TableUpdateByRowId(
+ self.table,
+ self.commit_user,
+ self.commit_identifier,
+ _precomputed_files_info=files_info,
+ )
+ try:
+ new_messages.extend(
+ updater.update_columns(update_data, column_names)
+ )
+ except Exception:
+ self._abort(updater.commit_messages)
+ raise
+ except Exception:
+ self._abort(new_messages)
+ raise
+
+ rewritten_entries = [
+ entry for entry in delta_entries if id(entry) not in candidate_ids
+ ]
+ rewritten_entries.extend(self._to_manifest_entries(new_messages))
+ return RowIdRewriteResult(
+ commit_entries=rewritten_entries,
+ rewritten_file_count=len(candidates),
+ )
+
+ def _is_rewrite_candidate(
+ self, entry, current_exact_ranges, next_row_id):
+ file = entry.file
+ return (
+ self._is_normal_row_id_file(file)
+ and file.first_row_id < next_row_id
+ and bool(file.write_cols)
+ and not any(
+ SpecialFields.is_system_field(name)
+ for name in file.write_cols
+ )
+ and self._range_key(entry) not in current_exact_ranges
+ )
+
+ def _ranges_are_still_covered(self, base_entries, candidates):
+ current_ranges = {}
+ for entry in base_entries:
+ if not self._is_normal_row_id_file(entry.file):
+ continue
+ key = (tuple(entry.partition.values), entry.bucket)
+ current_ranges.setdefault(key, []).append(
+ entry.file.row_id_range())
+ current_ranges = {
+ key: Range.sort_and_merge_overlap(ranges, True, True)
+ for key, ranges in current_ranges.items()
+ }
+
+ for entry in candidates:
+ key = (tuple(entry.partition.values), entry.bucket)
+ if entry.file.row_id_range().exclude(
+ current_ranges.get(key, [])):
+ return False
+ return True
+
+ def _affected_current_files(self, base_entries, candidates):
+ affected = {}
+ for base in base_entries:
+ if not self._is_normal_row_id_file(base.file):
+ continue
+ base_key = (tuple(base.partition.values), base.bucket)
+ for candidate in candidates:
+ candidate_key = (
+ tuple(candidate.partition.values), candidate.bucket)
+ if (
+ base_key == candidate_key
+ and base.file.row_id_range().overlaps(
+ candidate.file.row_id_range())
+ ):
+ key = (
+ base_key,
+ base.file.external_path or base.file.file_path
+ or base.file.file_name,
+ )
+ affected[key] = base.file
+ break
+ return list(affected.values())
+
+ @staticmethod
+ def _group_by_write_columns(candidates):
+ groups = {}
+ for entry in candidates:
+ key = tuple(entry.file.write_cols)
+ groups.setdefault(key, []).append(entry)
+ return [
+ (list(columns), entries)
+ for columns, entries in groups.items()
+ ]
+
+ def _read_staged_update(self, entry, column_names):
+ read_fields = [self.table.field_dict[name] for name in column_names]
+ read_fields.append(SpecialFields.ROW_ID)
+ split = DataSplit(
+ files=[entry.file],
+ partition=entry.partition,
+ bucket=entry.bucket,
+ raw_convertible=True,
+ )
+ result = TableRead(
+ self.table,
+ predicate=None,
+ read_type=read_fields,
+ ).to_arrow([split])
+ if result.num_rows != entry.file.row_count:
+ raise RuntimeError(
+ "Automatic row-id conflict rewrite read {} rows from staged "
+ "file '{}', expected {}.".format(
+ result.num_rows,
+ entry.file.file_name,
+ entry.file.row_count,
+ )
+ )
+ return result
+
+ def _to_manifest_entries(self, messages: List[CommitMessage]):
+ entries = []
+ for message in messages:
+ partition = GenericRow(
+ list(message.partition),
+ self.table.partition_keys_fields,
+ )
+ for file in message.new_files:
+ entries.append(ManifestEntry(
+ kind=0,
+ partition=partition,
+ bucket=message.bucket,
+ total_buckets=self.table.total_buckets,
+ file=file,
+ ))
+ return entries
+
+ def _abort(self, messages):
+ for message in messages:
+ for file in message.new_files:
+ path = file.external_path or file.file_path
+ if path:
+ self.table.file_io.delete_quietly(path)
+
+ @staticmethod
+ def _range_key(entry):
+ return (
+ tuple(entry.partition.values),
+ entry.bucket,
+ entry.file.first_row_id,
+ entry.file.row_count,
+ )
+
+ @staticmethod
+ def _is_normal_row_id_file(file):
+ return (
+ file.first_row_id is not None
+ and not RowIdConflictRewriter._is_dedicated_file(file)
+ )
+
+ @staticmethod
+ def _is_dedicated_file(file):
+ return (
+ DataFileMeta.is_blob_file(file.file_name)
+ or DataFileMeta.is_vector_file(file.file_name)
+ )
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 8c7484071a..73f3d22429 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -41,6 +41,11 @@ from pypaimon.write.commit.commit_scanner import
CommitScanner
from pypaimon.write.commit.conflict_detection import (
CommitConflictError,
ConflictDetection,
+ RowIdExistenceConflict,
+)
+from pypaimon.write.commit.row_id_conflict_rewriter import (
+ RowIdConflictRewriter,
+ RowIdRewriteResult,
)
from pypaimon.write.commit.overwrite_changes_provider import
OverwriteChangesProvider
from pypaimon.table.special_fields import SpecialFields
@@ -79,6 +84,15 @@ class RetryResult(CommitResult):
return False
+class RewriteResult(CommitResult):
+
+ def __init__(self, rewrite: RowIdRewriteResult):
+ self.rewrite = rewrite
+
+ def is_success(self) -> bool:
+ return False
+
+
class FileStoreCommit:
"""
Core commit logic for file store operations.
@@ -112,6 +126,9 @@ class FileStoreCommit:
self.commit_timeout = table.options.commit_timeout()
self.commit_min_retry_wait = table.options.commit_min_retry_wait()
self.commit_max_retry_wait = table.options.commit_max_retry_wait()
+ self.row_id_conflict_rewrite_max_size = (
+ table.options.data_evolution_row_id_conflict_rewrite_max_size()
+ )
self.commit_scanner = CommitScanner(table, self.manifest_list_manager)
@@ -360,10 +377,15 @@ class FileStoreCommit:
retry_count = 0
retry_result = None
+ rewritten_commit_entries = None
start_time_ms = int(time.time() * 1000)
while True:
latest_snapshot = self.snapshot_manager.get_latest_snapshot()
- commit_entries = commit_entries_plan(latest_snapshot)
+ commit_entries = (
+ rewritten_commit_entries
+ if rewritten_commit_entries is not None
+ else commit_entries_plan(latest_snapshot)
+ )
# No entries to commit (e.g. drop_partitions with no matching
data): skip commit
# to avoid creating manifest/snapshot with empty partition_stats
(causes read errors).
@@ -384,7 +406,22 @@ class FileStoreCommit:
hash_index_base_snapshot=hash_index_base_snapshot,
)
- if result.is_success():
+ if isinstance(result, RewriteResult):
+ rewritten_commit_entries = result.rewrite.commit_entries
+ self.conflict_detection._row_id_check_from_snapshot = (
+ latest_snapshot.id
+ )
+ # No snapshot commit was attempted for the conflicting files,
+ # so the rewritten attempt is still deterministic.
+ retry_result = None
+ logger.info(
+ "Rewrote %d stale row-id file(s) against snapshot %d "
+ "before retrying commit to table %s.",
+ result.rewrite.rewritten_file_count,
+ latest_snapshot.id,
+ self.table.identifier,
+ )
+ elif result.is_success():
commit_duration_ms = int(time.time() * 1000) - start_time_ms
if commit_kind == "OVERWRITE":
logger.info(
@@ -399,8 +436,8 @@ class FileStoreCommit:
commit_duration_ms,
)
break
-
- retry_result = result
+ else:
+ retry_result = result
elapsed_ms = int(time.time() * 1000) - start_time_ms
if elapsed_ms > self.commit_timeout or retry_count >=
self.commit_max_retries:
@@ -421,7 +458,7 @@ class FileStoreCommit:
f"after {elapsed_ms} millis with {retry_count} retries, "
f"there maybe exist commit conflicts between multiple
jobs."
)
- if retry_result.exception:
+ if retry_result is not None and retry_result.exception:
raise RuntimeError(error_msg) from retry_result.exception
else:
raise RuntimeError(error_msg)
@@ -505,6 +542,18 @@ class FileStoreCommit:
)
if conflict_exception is not None:
+ rewrite_result = self._try_rewrite_row_id_conflict(
+ retry_result,
+ conflict_exception,
+ latest_snapshot,
+ base_data_files,
+ commit_entries,
+ commit_kind,
+ commit_identifier,
+ changelog_entries,
+ )
+ if rewrite_result is not None:
+ return RewriteResult(rewrite_result)
if allow_rollback and self.rollback is not None:
if self.rollback.try_to_rollback(latest_snapshot):
# Rolled back: base/snapshot no longer valid; next
attempt
@@ -658,6 +707,49 @@ class FileStoreCommit:
return SuccessResult()
+ def _try_rewrite_row_id_conflict(
+ self,
+ retry_result,
+ conflict_exception,
+ latest_snapshot,
+ base_data_files,
+ commit_entries,
+ commit_kind,
+ commit_identifier,
+ changelog_entries):
+ if not isinstance(conflict_exception, RowIdExistenceConflict):
+ return None
+ if commit_kind != "APPEND" or changelog_entries:
+ return None
+ if retry_result is not None and retry_result.exception is not None:
+ return None
+
+ non_compaction_conflict = (
+ self.conflict_detection.check_row_id_from_snapshot(
+ latest_snapshot,
+ commit_entries,
+ check_compaction=False,
+ )
+ )
+ if non_compaction_conflict is not None:
+ return None
+
+ try:
+ return RowIdConflictRewriter(
+ self.table,
+ self.commit_user,
+ commit_identifier,
+ self.row_id_conflict_rewrite_max_size,
+ ).rewrite(
+ latest_snapshot,
+ base_data_files,
+ commit_entries,
+ )
+ except RuntimeError as rewrite_error:
+ raise CommitConflictError(
+ "{} {}".format(conflict_exception, rewrite_error)
+ ) from conflict_exception
+
def _write_manifest_files(self, commit_entries, base_name):
return self.manifest_file_manager.rolling_write(
commit_entries, self.manifest_target_size, base_name)
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 8963bf4abc..b0caa6ad03 100644
--- a/paimon-python/pypaimon/write/table_update_by_row_id.py
+++ b/paimon-python/pypaimon/write/table_update_by_row_id.py
@@ -24,6 +24,10 @@ import pyarrow as pa
import pyarrow.compute as pc
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
+from pypaimon.manifest.schema.manifest_entry import ManifestEntry
+from pypaimon.read.scanner.data_evolution_split_generator import (
+ DataEvolutionSplitGenerator,
+)
from pypaimon.read.split import DataSplit
from pypaimon.read.table_read import TableRead
from pypaimon.schema.data_types import (
@@ -97,7 +101,7 @@ class TableUpdateByRowId:
self.commit_messages: List[CommitMessage] = []
def _snapshot_files_info(self) -> _FilesInfo:
- """Internal: return the current snapshot's file index for broadcast."""
+ """Return the already loaded snapshot file index for broadcast."""
return _FilesInfo(
snapshot_id=self.snapshot_id,
first_row_ids=self.first_row_ids,
@@ -115,8 +119,27 @@ class TableUpdateByRowId:
"""
scan = self.table.new_read_builder().new_scan()
plan = scan.plan_for_write()
- splits = plan.splits()
-
+ snapshot_id = plan.snapshot_id if plan.snapshot_id is not None else -1
+ return self._files_info_from_splits(snapshot_id, plan.splits())
+
+ @classmethod
+ def _files_info_from_entries(
+ cls,
+ table,
+ snapshot_id: int,
+ entries: List[ManifestEntry],
+ ) -> _FilesInfo:
+ """Build a file index from an already resolved snapshot entry set."""
+ splits = DataEvolutionSplitGenerator(
+ table,
+ table.options.source_split_target_size(),
+ table.options.source_split_open_file_cost(),
+ ).create_splits(entries)
+ return cls._files_info_from_splits(snapshot_id, splits)
+
+ @classmethod
+ def _files_info_from_splits(
+ cls, snapshot_id: int, splits: List[DataSplit]) -> _FilesInfo:
index: Dict[int, Tuple[DataSplit, List[DataFileMeta]]] = {}
row_id_ranges: List[Range] = []
for split in splits:
@@ -128,14 +151,19 @@ class TableUpdateByRowId:
if not DataFileMeta.is_blob_file(file.file_name)
]
for file in split.files:
- if file.first_row_id is None or
DataFileMeta.is_blob_file(file.file_name):
+ if (
+ file.first_row_id is None
+ or DataFileMeta.is_blob_file(file.file_name)
+ ):
continue
row_id_ranges.append(file.row_id_range())
for file in data_files:
target_files = [
target_file
for target_file in files_with_row_id
- if self._overlaps(file.row_id_range(),
target_file.row_id_range())
+ if cls._overlaps(
+ file.row_id_range(), target_file.row_id_range()
+ )
]
entry = index.get(file.first_row_id)
@@ -143,7 +171,9 @@ class TableUpdateByRowId:
index[file.first_row_id] = (split, target_files)
else:
existing_files = entry[1]
- existing_names = {existing.file_name for existing in
existing_files}
+ existing_names = {
+ existing.file_name for existing in existing_files
+ }
existing_files.extend(
target_file
for target_file in target_files
@@ -155,7 +185,6 @@ class TableUpdateByRowId:
else:
merged = []
- snapshot_id = plan.snapshot_id if plan.snapshot_id is not None else -1
return _FilesInfo(
snapshot_id=snapshot_id,
first_row_ids=sorted(index.keys()),
