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 c834ed39b7 [python] Add conflict check for blob updating (#8244)
c834ed39b7 is described below
commit c834ed39b71da830e962a385e6d170f72a821b60
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Jun 15 22:21:08 2026 +0800
[python] Add conflict check for blob updating (#8244)
A blob update prepared against snapshot N commits successfully even when
a COMPACT between N and now has removed the anchor file. Subsequent
reads can fail, and two such stale commits in a row break the table.
Detect the COMPACT in the row-id conflict check and reject the commit
with a `RuntimeError`.
---
.../test/java/org/apache/paimon/JavaPyE2ETest.java | 74 +++++++++++++-
paimon-python/dev/run_mixed_tests.sh | 40 +++++++-
.../pypaimon/tests/e2e/java_py_read_write_test.py | 40 ++++++++
.../pypaimon/tests/partition_predicate_test.py | 35 +++++++
.../tests/write/conflict_detection_test.py | 106 +++++++++++++++++++++
.../pypaimon/write/commit/commit_scanner.py | 21 ++++
.../pypaimon/write/commit/conflict_detection.py | 57 +++++++++++
.../pypaimon/write/table_update_by_row_id.py | 8 +-
8 files changed, 377 insertions(+), 4 deletions(-)
diff --git a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
index 379331a1c2..543ca8eab6 100644
--- a/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/JavaPyE2ETest.java
@@ -1288,6 +1288,73 @@ public class JavaPyE2ETest {
LOG.info("compact_conflict_test: compact done, 5 files merged into 1
(1000 rows)");
}
+ /**
+ * Step 1 for blob compact conflict test: write a blob table with 2 data
files so compaction
+ * will merge them. Each file has 100 rows of (id, name, blob_data).
+ */
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testBlobCompactConflictWriteBase() throws Exception {
+ Identifier id = identifier("blob_compact_conflict_test");
+ try {
+ catalog.dropTable(id, true);
+ } catch (Exception ignore) {
+ }
+ Schema schema =
+ Schema.newBuilder()
+ .column("f0", DataTypes.INT())
+ .column("f1", DataTypes.STRING())
+ .column("f2", DataTypes.BLOB())
+ .option("target-file-size", "100 MB")
+ .option(ROW_TRACKING_ENABLED.key(), "true")
+ .option(DATA_EVOLUTION_ENABLED.key(), "true")
+ .option("compaction.min.file-num", "2")
+ .option(BUCKET.key(), "-1")
+ .build();
+ catalog.createTable(id, schema, false);
+
+ byte[] blobBytes = new byte[64];
+ java.util.Random rng = new java.util.Random(42);
+
+ FileStoreTable table = (FileStoreTable) catalog.getTable(id);
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite w = builder.newWrite()) {
+ for (int i = 0; i < 100; i++) {
+ rng.nextBytes(blobBytes);
+ w.write(
+ GenericRow.of(
+ i,
+ BinaryString.fromString("name" + i),
+ new
org.apache.paimon.data.BlobData(blobBytes.clone())));
+ }
+ builder.newCommit().commit(w.prepareCommit());
+ }
+
+ table = (FileStoreTable) catalog.getTable(id);
+ builder = table.newBatchWriteBuilder();
+ try (BatchTableWrite w = builder.newWrite()) {
+ for (int i = 100; i < 200; i++) {
+ rng.nextBytes(blobBytes);
+ w.write(
+ GenericRow.of(
+ i,
+ BinaryString.fromString("name" + i),
+ new
org.apache.paimon.data.BlobData(blobBytes.clone())));
+ }
+ builder.newCommit().commit(w.prepareCommit());
+ }
+ LOG.info("blob_compact_conflict_test: 2 base files written (100 rows
each, total 200)");
+ }
+
+ /** Step 3 for blob compact conflict test: run compact. */
+ @Test
+ @EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
+ public void testBlobCompactConflictRunCompact() throws Exception {
+ Identifier id = identifier("blob_compact_conflict_test");
+ doDataEvolutionCompact((FileStoreTable) catalog.getTable(id), true);
+ LOG.info("blob_compact_conflict_test: compact done
(compactBlob=true)");
+ }
+
@Test
@EnabledIfSystemProperty(named = "run.e2e.tests", matches = "true")
public void testDataEvolutionWrite() throws Exception {
@@ -1373,8 +1440,13 @@ public class JavaPyE2ETest {
}
private void doDataEvolutionCompact(FileStoreTable table) throws Exception
{
+ doDataEvolutionCompact(table, false);
+ }
+
+ private void doDataEvolutionCompact(FileStoreTable table, boolean
compactBlob)
+ throws Exception {
DataEvolutionCompactCoordinator coordinator =
- new DataEvolutionCompactCoordinator(table, false, false);
+ new DataEvolutionCompactCoordinator(table, compactBlob, false);
List<CommitMessage> messages = new ArrayList<>();
try {
List<DataEvolutionCompactTask> tasks;
diff --git a/paimon-python/dev/run_mixed_tests.sh
b/paimon-python/dev/run_mixed_tests.sh
index c2b75f476d..2014caa781 100755
--- a/paimon-python/dev/run_mixed_tests.sh
+++ b/paimon-python/dev/run_mixed_tests.sh
@@ -462,6 +462,30 @@ run_compact_conflict_test() {
fi
}
+run_blob_compact_conflict_test() {
+ echo -e "${YELLOW}=== Running Blob Compact Conflict Test (Java Write Base,
Python Blob Update + Java Compact) ===${NC}"
+
+ cd "$PROJECT_ROOT"
+
+ echo "Running Maven test for
JavaPyE2ETest.testBlobCompactConflictWriteBase..."
+ if mvn test
-Dtest=org.apache.paimon.JavaPyE2ETest#testBlobCompactConflictWriteBase -pl
paimon-core -q -Drun.e2e.tests=true; then
+ echo -e "${GREEN}✓ Java write base blob files completed
successfully${NC}"
+ else
+ echo -e "${RED}✗ Java write base blob files failed${NC}"
+ return 1
+ fi
+
+ cd "$PAIMON_PYTHON_DIR"
+ echo "Running Python test for
JavaPyReadWriteTest.test_blob_compact_conflict_update..."
+ if python -m pytest
java_py_read_write_test.py::JavaPyReadWriteTest::test_blob_compact_conflict_update
-v; then
+ echo -e "${GREEN}✓ Python blob compact conflict test completed
successfully${NC}"
+ return 0
+ else
+ echo -e "${RED}✗ Python blob compact conflict test failed${NC}"
+ return 1
+ fi
+}
+
run_data_evolution_test() {
echo -e "${YELLOW}=== Running Data Evolution Test (Java Write, Python
Read) ===${NC}"
@@ -673,6 +697,7 @@ main() {
local lumina_vector_result=0
local lumina_vector_btree_result=0
local compact_conflict_result=0
+ local blob_compact_conflict_result=0
local blob_alter_compact_result=0
local data_evolution_result=0
local data_evolution_py_write_result=0
@@ -823,6 +848,13 @@ main() {
echo ""
+ # Run blob compact conflict test (Java write base + Python blob update +
Java compact)
+ if ! run_blob_compact_conflict_test; then
+ blob_compact_conflict_result=1
+ fi
+
+ echo ""
+
# Run blob alter+compact test (Java write+alter+compact, Python read)
if ! run_blob_alter_compact_test; then
blob_alter_compact_result=1
@@ -968,6 +1000,12 @@ main() {
echo -e "${RED}✗ Compact Conflict Test (Java Write+Compact, Python
Read): FAILED${NC}"
fi
+ if [[ $blob_compact_conflict_result -eq 0 ]]; then
+ echo -e "${GREEN}✓ Blob Compact Conflict Test (Java Write+Compact,
Python Blob Update): PASSED${NC}"
+ else
+ echo -e "${RED}✗ Blob Compact Conflict Test (Java Write+Compact,
Python Blob Update): FAILED${NC}"
+ fi
+
if [[ $blob_alter_compact_result -eq 0 ]]; then
echo -e "${GREEN}✓ Blob Alter+Compact Test (Java Write+Alter+Compact,
Python Read): PASSED${NC}"
else
@@ -1009,7 +1047,7 @@ main() {
# Clean up warehouse directory after all tests
cleanup_warehouse
- if [[ $java_write_result -eq 0 && $python_read_result -eq 0 &&
$python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 &&
$btree_index_result -eq 0 && $compressed_text_result -eq 0 &&
$tantivy_fulltext_result -eq 0 && $lumina_vector_result -eq 0 &&
$lumina_vector_btree_result -eq 0 && $compact_conflict_result -eq 0 &&
$blob_alter_compact_result -eq 0 && $data_evolution_result -eq 0 &&
$data_evolution_py_write_result -eq 0 && $java_variant_write_py_read_result -eq
[...]
+ if [[ $java_write_result -eq 0 && $python_read_result -eq 0 &&
$python_write_result -eq 0 && $java_read_result -eq 0 && $pk_dv_result -eq 0 &&
$btree_index_result -eq 0 && $compressed_text_result -eq 0 &&
$tantivy_fulltext_result -eq 0 && $lumina_vector_result -eq 0 &&
$lumina_vector_btree_result -eq 0 && $compact_conflict_result -eq 0 &&
$blob_compact_conflict_result -eq 0 && $blob_alter_compact_result -eq 0 &&
$data_evolution_result -eq 0 && $data_evolution_py_write_result -eq 0 && [...]
echo -e "${GREEN}🎉 All tests passed! Java-Python interoperability
verified.${NC}"
return 0
else
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 0c84a828bc..4f1df33585 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
@@ -1190,6 +1190,46 @@ class JavaPyReadWriteTest(unittest.TestCase):
tc.close()
print(f"Conflict detected as expected: {ctx.exception}")
+ def test_blob_compact_conflict_update(self):
+ import subprocess
+
+ table = self.catalog.get_table('default.blob_compact_conflict_test')
+ snapshot_before =
table.new_read_builder().new_scan().plan().snapshot_id
+
+ wb = table.new_batch_write_builder()
+ table_update = wb.new_update().with_update_type(['f2'])
+ update_data = pa.Table.from_pydict({
+ '_ROW_ID': pa.array([50], type=pa.int64()),
+ 'f2': pa.array([b'blob50-updated'], type=pa.large_binary()),
+ })
+ stale_commit_msgs =
table_update.update_by_arrow_with_row_id(update_data)
+
+ project_root = os.path.join(self.tempdir, '..', '..', '..', '..')
+ result = subprocess.run(
+ ['mvn', 'test',
+ '-pl', 'paimon-core',
+
'-Dtest=org.apache.paimon.JavaPyE2ETest#testBlobCompactConflictRunCompact',
+ '-Drun.e2e.tests=true',
+ '-Dsurefire.failIfNoSpecifiedTests=false',
+ '-q'],
+ cwd=os.path.abspath(project_root),
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+ universal_newlines=True, timeout=300
+ )
+ self.assertEqual(result.returncode, 0,
+ f"Java compact
failed:\n{result.stdout}\n{result.stderr}")
+
+ table = self.catalog.get_table('default.blob_compact_conflict_test')
+ snapshot_after = table.new_read_builder().new_scan().plan().snapshot_id
+ self.assertGreater(snapshot_after, snapshot_before)
+
+ tc = wb.new_commit()
+ try:
+ with self.assertRaises(RuntimeError):
+ tc.commit(stale_commit_msgs)
+ finally:
+ tc.close()
+
@parameterized.expand(get_file_format_params())
def test_read_data_evolution_table(self, file_format):
"""Read data evolution tables written by Java and verify merged
results."""
diff --git a/paimon-python/pypaimon/tests/partition_predicate_test.py
b/paimon-python/pypaimon/tests/partition_predicate_test.py
index abf5212d0e..dbc50d0585 100644
--- a/paimon-python/pypaimon/tests/partition_predicate_test.py
+++ b/paimon-python/pypaimon/tests/partition_predicate_test.py
@@ -345,3 +345,38 @@ class
TestCommitScannerPartitionPredicate(unittest.TestCase):
self.assertIn('partition_predicate', kwargs)
self.assertIsNotNone(kwargs['partition_predicate'])
self.assertNotIn('predicate', kwargs)
+
+ @patch('pypaimon.write.commit.commit_scanner.ManifestFileManager')
+ def test_raw_entries_preserve_delete_kind(self, mock_mfm_cls):
+ added = ManifestEntry(
+ kind=0, partition=GenericRow(['p1', 'us'], PARTITION_FIELDS),
+ bucket=0, total_buckets=1, file=Mock())
+ deleted = ManifestEntry(
+ kind=1, partition=GenericRow(['p1', 'us'], PARTITION_FIELDS),
+ bucket=0, total_buckets=1, file=Mock())
+ mock_mfm_cls.return_value.read.return_value = [added, deleted]
+
+ scanner = self._scanner()
+ scanner.manifest_list_manager.read_delta.return_value =
[Mock(file_name='m1')]
+ result = scanner.read_incremental_raw_entries_from_changed_partitions(
+ Mock(), [_manifest_entry(['p1', 'us'])])
+
+ self.assertEqual([e.kind for e in result], [0, 1])
+
+ @patch('pypaimon.write.commit.commit_scanner.ManifestFileManager')
+ def test_raw_entries_filter_unmatched_partition(self, mock_mfm_cls):
+ in_part = ManifestEntry(
+ kind=1, partition=GenericRow(['p1', 'us'], PARTITION_FIELDS),
+ bucket=0, total_buckets=1, file=Mock())
+ out_part = ManifestEntry(
+ kind=1, partition=GenericRow(['p2', 'eu'], PARTITION_FIELDS),
+ bucket=0, total_buckets=1, file=Mock())
+ mock_mfm_cls.return_value.read.return_value = [in_part, out_part]
+
+ scanner = self._scanner()
+ scanner.manifest_list_manager.read_delta.return_value =
[Mock(file_name='m1')]
+ result = scanner.read_incremental_raw_entries_from_changed_partitions(
+ Mock(), [_manifest_entry(['p1', 'us'])])
+
+ self.assertEqual(len(result), 1)
+ self.assertEqual(tuple(result[0].partition.values), ('p1', 'us'))
diff --git a/paimon-python/pypaimon/tests/write/conflict_detection_test.py
b/paimon-python/pypaimon/tests/write/conflict_detection_test.py
index 72ac50f79b..b035b38056 100644
--- a/paimon-python/pypaimon/tests/write/conflict_detection_test.py
+++ b/paimon-python/pypaimon/tests/write/conflict_detection_test.py
@@ -205,6 +205,112 @@ class TestCheckRowIdExistence(unittest.TestCase):
detection.check_row_id_existence(base, delta, next_row_id=None))
+class _FakeSnapshot:
+
+ def __init__(self, snapshot_id, commit_kind, next_row_id=None):
+ self.id = snapshot_id
+ self.commit_kind = commit_kind
+ self.next_row_id = next_row_id
+
+
+class _FakeSnapshotManager:
+
+ def __init__(self, snapshots):
+ self._by_id = {s.id: s for s in snapshots}
+
+ def get_snapshot_by_id(self, snapshot_id):
+ return self._by_id.get(snapshot_id)
+
+
+class _FakeCommitScanner:
+
+ def __init__(self, entries_by_snapshot_id,
raw_entries_by_snapshot_id=None):
+ self._by_id = entries_by_snapshot_id
+ self._raw_by_id = raw_entries_by_snapshot_id or {}
+
+ def read_incremental_entries_from_changed_partitions(self, snapshot, _):
+ return self._by_id.get(snapshot.id, [])
+
+ def read_incremental_raw_entries_from_changed_partitions(self, snapshot,
_):
+ return self._raw_by_id.get(snapshot.id, self._by_id.get(snapshot.id,
[]))
+
+
+class _FakeTable:
+
+ def __init__(self, schema_manager):
+ self.schema_manager = schema_manager
+
+
+class TestCheckRowIdFromSnapshot(unittest.TestCase):
+
+ def _make_detection(self, snapshots, raw_entries_by_snapshot_id):
+ detection = ConflictDetection(
+ data_evolution_enabled=True,
+ snapshot_manager=_FakeSnapshotManager(snapshots),
+ manifest_list_manager=None,
+ table=_FakeTable(_FakeSchemaManager([_DEFAULT_SCHEMA])),
+ commit_scanner=_FakeCommitScanner({}, raw_entries_by_snapshot_id),
+ )
+ detection._row_id_check_from_snapshot = 1
+ return detection
+
+ def _blob_delta(self):
+ return [_make_entry("d.blob", first_row_id=0, row_count=51,
+ write_cols=["col_a"])]
+
+ def test_compact_blob_delete_raises_at_first_match(self):
+ check_snap = _FakeSnapshot(1, "APPEND", next_row_id=200)
+ compact1 = _FakeSnapshot(2, "COMPACT", next_row_id=200)
+ compact2 = _FakeSnapshot(3, "COMPACT", next_row_id=200)
+ entries = {
+ 2: [_make_entry("first.blob", kind=1, first_row_id=0,
row_count=200)],
+ 3: [_make_entry("second.blob", kind=1, first_row_id=0,
row_count=200)],
+ }
+ detection = self._make_detection(
+ [check_snap, compact1, compact2], entries)
+ result = detection.check_row_id_from_snapshot(compact2,
self._blob_delta())
+ self.assertIsNotNone(result)
+ self.assertIn("snapshot 2", str(result))
+ self.assertIn("COMPACT", str(result))
+
+ def test_compact_other_file_type_does_not_raise(self):
+ check_snap = _FakeSnapshot(1, "APPEND", next_row_id=200)
+ compact_snap = _FakeSnapshot(2, "COMPACT", next_row_id=200)
+ compact_entries = [
+ _make_entry("old.parquet", kind=1, first_row_id=0, row_count=100),
+ _make_entry("merged.parquet", kind=0, first_row_id=0,
row_count=200),
+ ]
+ detection = self._make_detection(
+ [check_snap, compact_snap], {2: compact_entries})
+ self.assertIsNone(
+ detection.check_row_id_from_snapshot(compact_snap,
self._blob_delta()))
+
+ def test_compact_no_conflict_when_no_matching_delete(self):
+ check_snap = _FakeSnapshot(1, "APPEND", next_row_id=400)
+ compact_snap = _FakeSnapshot(2, "COMPACT", next_row_id=400)
+ col_a_delta = self._blob_delta()
+ col_b_delta = [_make_entry("d.parquet", first_row_id=0, row_count=51,
+ write_cols=["col_b"])]
+ cases = [
+ ("disjoint_range", col_a_delta, [
+ _make_entry("old.blob", kind=1, first_row_id=200,
row_count=200),
+ ]),
+ ("add_only", col_a_delta, [
+ _make_entry("merged.blob", kind=0, first_row_id=0,
row_count=200),
+ ]),
+ ("other_column_shard", col_b_delta, [
+ _make_entry("old.parquet", kind=1, first_row_id=0,
row_count=100,
+ write_cols=["col_a"]),
+ ]),
+ ]
+ for name, delta, compact_entries in cases:
+ with self.subTest(case=name):
+ detection = self._make_detection(
+ [check_snap, compact_snap], {2: compact_entries})
+ self.assertIsNone(
+ detection.check_row_id_from_snapshot(compact_snap, delta))
+
+
class TestRowIdColumnConflictChecker(unittest.TestCase):
def _make_checker(self, delta_files, schema=None):
diff --git a/paimon-python/pypaimon/write/commit/commit_scanner.py
b/paimon-python/pypaimon/write/commit/commit_scanner.py
index 697d95f216..95702f1d02 100644
--- a/paimon-python/pypaimon/write/commit/commit_scanner.py
+++ b/paimon-python/pypaimon/write/commit/commit_scanner.py
@@ -21,6 +21,7 @@ Manifest entries scanner for commit operations.
from typing import Optional, List
from pypaimon.common.predicate_builder import PredicateBuilder
+from pypaimon.manifest.manifest_file_manager import ManifestFileManager
from pypaimon.manifest.manifest_list_manager import ManifestListManager
from pypaimon.manifest.schema.manifest_entry import ManifestEntry
from pypaimon.read.scanner.file_scanner import FileScanner
@@ -95,6 +96,26 @@ class CommitScanner:
self.table, lambda: ([], None),
partition_predicate=partition_filter
).read_manifest_entries(delta_manifests)
+ def read_incremental_raw_entries_from_changed_partitions(self, snapshot:
Snapshot,
+ commit_entries:
List[ManifestEntry]):
+ """Like ``read_incremental_entries_from_changed_partitions`` but
+ preserves DELETE entries (kind=1). The regular method funnels through
+ ``read_entries_parallel`` which discards standalone DELETEs.
+ """
+ delta_manifests = self.manifest_list_manager.read_delta(snapshot)
+ if not delta_manifests:
+ return []
+
+ partition_filter =
self._build_partition_filter_from_entries(commit_entries)
+ mfm = ManifestFileManager(self.table)
+ entries = []
+ for mf in delta_manifests:
+ for entry in mfm.read(mf.file_name):
+ if partition_filter is not None and not
partition_filter.test(entry.partition):
+ continue
+ entries.append(entry)
+ return entries
+
def _build_partition_filter_from_entries(self, entries:
List[ManifestEntry]):
"""Build a partition predicate that matches all partitions present in
the given entries.
diff --git a/paimon-python/pypaimon/write/commit/conflict_detection.py
b/paimon-python/pypaimon/write/commit/conflict_detection.py
index 0d04f464ce..0a7ae8e30b 100644
--- a/paimon-python/pypaimon/write/commit/conflict_detection.py
+++ b/paimon-python/pypaimon/write/commit/conflict_detection.py
@@ -301,13 +301,27 @@ class ConflictDetection:
"{snapshot}.".format(snapshot=self._row_id_check_from_snapshot))
check_next_row_id = check_snapshot.next_row_id
+ # Pair each delta with its anchor file type so a parquet-only
+ # compact does not flag a blob delta whose .blob anchor is intact.
+ delta_signatures = []
+ for f in delta_files:
+ r = f.row_id_range()
+ if r is not None:
+ delta_signatures.append(
+ (DataFileMeta.is_blob_file(f.file_name), r.from_, r.to))
+
for snapshot_id in range(
self._row_id_check_from_snapshot + 1,
latest_snapshot.id + 1):
snapshot = self.snapshot_manager.get_snapshot_by_id(snapshot_id)
if snapshot is None:
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
continue
incremental_entries =
self.commit_scanner.read_incremental_entries_from_changed_partitions(
@@ -325,3 +339,46 @@ class ConflictDetection:
"ineffective.")
return None
+
+ def _compact_conflicts_with_delta(self, snapshot, delta_signatures,
+ column_checker, commit_entries):
+ """Return RuntimeError if a COMPACT snapshot deleted a same-kind
+ anchor file whose row-id range AND write columns overlap any
+ staged delta; otherwise None.
+
+ File-type match guards against `write_cols=None` ambiguity (an
+ initial full-row parquet does not actually contain blob columns);
+ column_checker guards against unrelated column-write shards
+ (compacting an f1-only parquet must not block an f2 update on
+ the same row range).
+ """
+ if not delta_signatures:
+ return None
+ raw_entries =
self.commit_scanner.read_incremental_raw_entries_from_changed_partitions(
+ snapshot, commit_entries)
+ for entry in raw_entries:
+ if entry.kind != 1:
+ continue
+ file_range = entry.file.row_id_range()
+ if file_range is None:
+ continue
+ deleted_is_blob = DataFileMeta.is_blob_file(entry.file.file_name)
+ for delta_is_blob, from_, to in delta_signatures:
+ if delta_is_blob != deleted_is_blob:
+ continue
+ if file_range.from_ > to or from_ > file_range.to:
+ continue
+ if not column_checker.conflicts_with(entry.file):
+ continue
+ return RuntimeError(
+ "Blob/row-id update conflicts with concurrent COMPACT "
+ "(snapshot {sid}): anchor file {name} [{ff}, {ft}] "
+ "was compacted away, overlaps staged delta "
+ "[{df}, {dt}].".format(
+ sid=snapshot.id,
+ name=entry.file.file_name,
+ ff=file_range.from_,
+ ft=file_range.to,
+ df=from_,
+ dt=to))
+ return None
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 d5653cce98..a44bee4e64 100644
--- a/paimon-python/pypaimon/write/table_update_by_row_id.py
+++ b/paimon-python/pypaimon/write/table_update_by_row_id.py
@@ -177,7 +177,11 @@ class TableUpdateByRowId:
if col_name not in self.table.field_names:
raise ValueError(f"Column {col_name} not found in table
schema")
- sorted_data = data.sort_by([(SpecialFields.ROW_ID.name, "ascending")])
+ sort_keys = [(SpecialFields.ROW_ID.name, "ascending")]
+ if hasattr(data, "sort_by"):
+ sorted_data = data.sort_by(sort_keys)
+ else:
+ sorted_data = data.take(pc.sort_indices(data, sort_keys=sort_keys))
data_with_first_row_id = self._calculate_first_row_id(sorted_data)
self._write_by_first_row_id(data_with_first_row_id, column_names)
@@ -311,7 +315,7 @@ class TableUpdateByRowId:
# Build a boolean mask: True at positions that need to be updated
all_indices = pa.array(range(original_data.num_rows), type=pa.int64())
- mask = pc.is_in(all_indices, relative_indices)
+ mask = pc.is_in(all_indices, value_set=relative_indices)
# Build the merged table column by column
merged_columns = {}
