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 6b29174a82 [python] Enable native BLOB collection writes and packed
video reads (#10399)
6b29174a82 is described below
commit 6b29174a82a7017be60922589a96d0a4e5ebbd18
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 6 16:57:18 2026 +0800
[python] Enable native BLOB collection writes and packed video reads
(#10399)
---
paimon-python/README.md | 15 +-
paimon-python/pypaimon/daft/daft_datasink.py | 6 +
paimon-python/pypaimon/index/dynamic_bucket.py | 5 +-
.../pypaimon/ray/data_evolution_merge_join.py | 8 +-
.../pypaimon/read/reader/field_indices.py | 3 +
paimon-python/pypaimon/read/split_read.py | 5 +-
paimon-python/pypaimon/read/table_read.py | 4 +-
paimon-python/pypaimon/tests/blob_table_test.py | 4 +-
.../pypaimon/tests/daft/daft_sink_test.py | 40 +++
.../pypaimon/tests/data_evolution_formats_test.py | 32 +-
.../pypaimon/tests/deletion_vector_path_test.py | 5 +-
.../pypaimon/tests/file_store_commit_test.py | 99 +++---
.../pypaimon/tests/multimodal_lerobot_test.py | 15 +-
.../tests/native_blob_collection_write_test.py | 354 +++++++++++++++++++++
.../pypaimon/tests/native_blob_write_test.py | 14 +-
.../tests/native_cross_partition_write_test.py | 4 +-
.../pypaimon/tests/native_dynamic_write_test.py | 4 +-
.../pypaimon/tests/native_nested_update_test.py | 4 +-
.../tests/native_postpone_fixed_write_test.py | 7 +-
.../pypaimon/tests/native_postpone_write_test.py | 2 +-
paimon-python/pypaimon/tests/native_read_test.py | 7 +-
paimon-python/pypaimon/tests/native_update_test.py | 5 +-
.../pypaimon/tests/native_video_read_test.py | 194 +++++++++++
.../pypaimon/tests/py36/ao_simple_test.py | 6 +-
.../tests/ray_data_evolution_merge_into_test.py | 8 +-
.../tests/ray_row_id_conflict_rewriter_test.py | 6 +-
paimon-python/pypaimon/tests/ray_sink_test.py | 6 +-
paimon-python/pypaimon/tests/table_commit_test.py | 65 ++++
paimon-python/pypaimon/tests/table_update_test.py | 112 ++++---
.../pypaimon/tests/table_upsert_by_key_test.py | 69 ++++
.../tests/write/changelog_producer_test.py | 6 +-
.../pypaimon/tests/write/dynamic_bucket_test.py | 49 ++-
.../pypaimon/tests/write/table_write_test.py | 12 +-
.../pypaimon/tests/write/write_buffer_test.py | 1 +
.../write/commit/row_id_conflict_rewriter.py | 13 +-
paimon-python/pypaimon/write/commit_message.py | 1 -
paimon-python/pypaimon/write/file_store_commit.py | 89 +++---
paimon-python/pypaimon/write/file_store_write.py | 16 +-
paimon-python/pypaimon/write/native_update.py | 5 +-
paimon-python/pypaimon/write/native_write.py | 29 +-
paimon-python/pypaimon/write/ray_datasink.py | 4 +
paimon-python/pypaimon/write/table_commit.py | 10 +-
paimon-python/pypaimon/write/table_update.py | 8 +-
.../pypaimon/write/table_update_by_row_id.py | 3 +
paimon-python/pypaimon/write/table_write.py | 15 +-
.../pypaimon/write/writer/composite_data_writer.py | 11 +
paimon-python/pypaimon/write/writer/data_writer.py | 4 +-
47 files changed, 1107 insertions(+), 277 deletions(-)
diff --git a/paimon-python/README.md b/paimon-python/README.md
index 42d38e0561..dedf06db7d 100644
--- a/paimon-python/README.md
+++ b/paimon-python/README.md
@@ -248,13 +248,14 @@ BLOB table selects the Python writer before any native
data is written. Switchin
to row writes after native Arrow writes is rejected. Use
`write.native.enabled=false` when
mixing Arrow batches and Python `Blob` objects in one writer.
-An explicit native writer `abort()` also deletes prepared files that have not
-been passed to a PyPaimon committer. Calling `close()` instead releases those
-files to the caller without deleting them; use `commit.abort(messages)` to
-discard them after closing the writer. Once a commit attempt starts, writer
-abort preserves its files even if the attempt raises, because a snapshot may
-already reference them. Stream writers retain cleanup ownership only for
-messages that have not been submitted to a committer.
+A successful `prepare_commit()` transfers file ownership to the caller.
+Writer `abort()` and `close()` preserve those prepared files and may clean only
+outputs still owned by the writer. Explicit `commit.abort(messages)` deletes
+newly written data, changelog, sidecars and added indexes, matching Java's
+Paimon Table committer. Call it only when those messages are known to be
+uncommitted and will never be submitted. Never call it after a commit with an
+unknown outcome: publication may have succeeded before its response failed.
+Internal failure paths preserve prepared files instead of calling commit abort.
Both native options are disabled by default.
diff --git a/paimon-python/pypaimon/daft/daft_datasink.py
b/paimon-python/pypaimon/daft/daft_datasink.py
index 1486832a26..1376eebe2b 100644
--- a/paimon-python/pypaimon/daft/daft_datasink.py
+++ b/paimon-python/pypaimon/daft/daft_datasink.py
@@ -256,6 +256,7 @@ class PaimonDataSink(DataSink[list[Any]]):
def write(self, micropartitions: Iterator[MicroPartition]) ->
Iterator[WriteResult[list[Any]]]:
table_write = self._write_builder.new_write()
+ commit_messages = []
total_rows = 0
total_bytes = 0
@@ -278,6 +279,8 @@ class PaimonDataSink(DataSink[list[Any]]):
table_write.abort()
except Exception:
logger.warning("Failed to abort Daft Paimon table write.",
exc_info=True)
+ # Never delete files from CommitMessage on failure.
+ # A commit can succeed even when its response raises an exception.
raise
yield WriteResult(
@@ -717,6 +720,7 @@ def make_group_write_udf(
ignore_existing=mode == "overwrite",
base_snapshot_id=group_base_snapshot_id,
)
+ commit_messages = []
try:
if precomputed_bucket:
offset = 0
@@ -749,6 +753,8 @@ def make_group_write_udf(
"Failed to abort Daft Paimon group write.",
exc_info=True,
)
+ # Never delete files from CommitMessage on failure.
+ # A commit can succeed even when its response raises an exception.
raise
return Series.from_arrow(
diff --git a/paimon-python/pypaimon/index/dynamic_bucket.py
b/paimon-python/pypaimon/index/dynamic_bucket.py
index f89862875c..a2838a9494 100644
--- a/paimon-python/pypaimon/index/dynamic_bucket.py
+++ b/paimon-python/pypaimon/index/dynamic_bucket.py
@@ -444,6 +444,7 @@ class DynamicBucketIndexMaintainer:
def prepare_commit(self) -> Dict[Tuple[Tuple, int],
DynamicBucketIndexChanges]:
changes: Dict[Tuple[Tuple, int], DynamicBucketIndexChanges] = {}
+ prepared_states = {}
for key, (hashes, old_entry, modified) in self._states.items():
if not modified:
continue
@@ -460,7 +461,9 @@ class DynamicBucketIndexMaintainer:
)
] if old_entry is not None else [],
)
- self._states[key] = (hashes, new_entry, False)
+ prepared_states[key] = (hashes, new_entry, False)
+ # A failed bucket must not make a retry forget earlier additions.
+ self._states.update(prepared_states)
return changes
def release_prepared(self) -> None:
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
index 01d4f4228d..3a261259a7 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
@@ -345,7 +345,6 @@ def _self_merge_aliases(batch: pa.Table, row_id_name: str)
-> pa.Table:
def _apply_self_merge_update_group(context, file_group, collect_row_ids):
"""Read, transform, and stage one complete first-row-id file group."""
from pypaimon.read.table_read import TableRead
- from pypaimon.write.file_store_commit import _abort_commit_messages
from pypaimon.write.row_id_file_index import RowIdFileIndex
table_read = TableRead(
@@ -394,7 +393,8 @@ def _apply_self_merge_update_group(context, file_group,
collect_row_ids):
try:
messages = updater.update_columns(updates, context.update_cols)
except Exception:
- _abort_commit_messages(context.table, updater.commit_messages)
+ # Never delete files from CommitMessage on failure.
+ # A commit can succeed even when its response raises an exception.
raise
return messages, updates.num_rows, row_ids
@@ -466,8 +466,8 @@ def distributed_self_merge_update_apply(
submit_next()
if first_error is not None:
- from pypaimon.write.file_store_commit import _abort_commit_messages
- _abort_commit_messages(plan.table, messages)
+ # Never delete files from CommitMessage on failure.
+ # A commit can succeed even when its response raises an exception.
raise first_error
return messages, num_updated, row_ids
diff --git a/paimon-python/pypaimon/read/reader/field_indices.py
b/paimon-python/pypaimon/read/reader/field_indices.py
index 78180e707b..b5a6690a6b 100644
--- a/paimon-python/pypaimon/read/reader/field_indices.py
+++ b/paimon-python/pypaimon/read/reader/field_indices.py
@@ -41,6 +41,9 @@ def descriptor_field_names_for_table(table) -> Set[str]:
from pypaimon.common.options.core_options import CoreOptions
names = set(CoreOptions.blob_descriptor_fields(table.options))
+ # Video readers return lazy frame references even in payload mode. Keep
+ # their frame/index locators through row filters and result projection.
+ names |= CoreOptions.video_frame_fields(table.options)
if CoreOptions.blob_as_descriptor(table.options):
names |= CoreOptions.blob_view_fields(table.options)
return names
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index d518ca5abe..7107692b3d 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -108,9 +108,10 @@ def deferred_blob_field_names(table, read_fields:
List[DataField],
or CoreOptions.blob_as_descriptor(table.options)):
return set()
- inline_fields = (
+ non_payload_fields = (
CoreOptions.blob_descriptor_fields(table.options)
| CoreOptions.blob_view_fields(table.options)
+ | CoreOptions.video_frame_fields(table.options)
)
predicate_fields = (
predicate_field_names(predicate) if predicate is not None else set()
@@ -118,7 +119,7 @@ def deferred_blob_field_names(table, read_fields:
List[DataField],
return {
read_fields[index].name
for index in blob_field_indices(read_fields)
- if read_fields[index].name not in inline_fields
+ if read_fields[index].name not in non_payload_fields
and read_fields[index].name not in predicate_fields
}
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index dadbb893e1..e94df3f0af 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -60,7 +60,7 @@ _NATIVE_READ_FILE_FORMATS = frozenset({
})
_NATIVE_READ_FILE_SUFFIXES = tuple(
'.%s' % file_format for file_format in _NATIVE_READ_FILE_FORMATS)
-_NATIVE_BLOB_FILE_SUFFIX = '.blob'
+_NATIVE_BLOB_FILE_SUFFIXES = ('.blob', '.video')
class _ClosableArrowBatchReader:
@@ -810,7 +810,7 @@ class TableRead:
file_name = data_file.file_name.lower()
if ('.vector.' not in file_name
and not file_name.endswith(_NATIVE_READ_FILE_SUFFIXES)
- and not file_name.endswith(_NATIVE_BLOB_FILE_SUFFIX)):
+ and not file_name.endswith(_NATIVE_BLOB_FILE_SUFFIXES)):
return False
return True
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py
b/paimon-python/pypaimon/tests/blob_table_test.py
index 41b86fd57b..6ba1a7d0af 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -1748,7 +1748,7 @@ class DedicatedFormatWriterTest(unittest.TestCase):
'payloads': pa.array([[('bad', None)]], type=map_blob_type),
}, schema=pa_schema)
invalid_writer = table.new_batch_write_builder().new_write()
- with self.assertRaisesRegex(ValueError, "does not allow null values"):
+ with self.assertRaisesRegex(ValueError, "does not allow null
values|non-nullable"):
invalid_writer.write_arrow(invalid_data)
invalid_writer.abort()
@@ -1821,7 +1821,7 @@ class DedicatedFormatWriterTest(unittest.TestCase):
'payloads': pa.array([[b'bad', None]], type=array_blob_type),
}, schema=pa_schema)
invalid_writer = table.new_batch_write_builder().new_write()
- with self.assertRaisesRegex(ValueError, "does not allow null
elements"):
+ with self.assertRaisesRegex(ValueError, "does not allow null
elements|cannot contain nulls"):
invalid_writer.write_arrow(invalid_data)
invalid_writer.abort()
diff --git a/paimon-python/pypaimon/tests/daft/daft_sink_test.py
b/paimon-python/pypaimon/tests/daft/daft_sink_test.py
index f6a2b95ce0..bf14da8a28 100644
--- a/paimon-python/pypaimon/tests/daft/daft_sink_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_sink_test.py
@@ -403,6 +403,46 @@ def test_write_paimon_pk_table(pk_table):
assert result.column("id").to_pylist() == [1, 2, 3]
[email protected]_write
[email protected]('grouped', [False, True])
+def
test_dedicated_close_failure_preserves_prepared_files(local_paimon_catalog,
grouped):
+ from unittest.mock import patch
+ from pypaimon.daft.daft_datasink import make_group_write_udf
+ from pypaimon.write.writer.dedicated_format_writer import
DedicatedFormatWriter
+
+ catalog, _ = local_paimon_catalog
+ arrow_schema = pa.schema([('id', pa.int32()), ('payload',
pa.large_binary())])
+ catalog.create_table('test_db.close_failure',
pypaimon.Schema.from_pyarrow_schema(
+ arrow_schema, options={'row-tracking.enabled': 'true',
'data-evolution.enabled': 'true',
+ 'write.native.enabled': 'false'}), False)
+ table = catalog.get_table('test_db.close_failure')
+ data = pa.table({'id': [1, 2, 3], 'payload': [b'a', b'b', b'c']},
schema=arrow_schema)
+ original_close = DedicatedFormatWriter._close_current_writers
+ calls = []
+
+ def fail_after_prepare(writer):
+ calls.append(writer)
+ if len(calls) == 1:
+ return original_close(writer)
+ raise RuntimeError('Close failed after prepare')
+
+ with patch.object(DedicatedFormatWriter, '_close_current_writers',
fail_after_prepare):
+ with pytest.raises(RuntimeError, match='Close failed after prepare'):
+ if grouped:
+ # Exercise the worker body synchronously so the injected close
+ # error reaches the same writer which prepared the files.
+ with patch('daft.func.batch', side_effect=lambda **kwargs:
lambda function: function):
+ write_group = make_group_write_udf(table, 'append', None)
+ write_group(*(daft.Series.from_arrow(column.combine_chunks())
for column in data.columns))
+ else:
+
list(PaimonDataSink(table).write(iter([MicroPartition.from_arrow(data)])))
+ assert len(calls) == 2
+ assert {os.path.splitext(name)[1]
+ for _, _, names in os.walk(table.table_path) for name in names
+ if name.endswith(('.parquet', '.blob'))} == {'.parquet', '.blob'}
+ assert table.snapshot_manager().get_latest_snapshot() is None
+
+
# ---------------------------------------------------------------------------
# Schema conversion tests
# ---------------------------------------------------------------------------
diff --git a/paimon-python/pypaimon/tests/data_evolution_formats_test.py
b/paimon-python/pypaimon/tests/data_evolution_formats_test.py
index d3fb26ceec..f5844de37c 100644
--- a/paimon-python/pypaimon/tests/data_evolution_formats_test.py
+++ b/paimon-python/pypaimon/tests/data_evolution_formats_test.py
@@ -540,7 +540,8 @@ class DataEvolutionFormatsTest(unittest.TestCase):
self.assertEqual(actual.column('id').to_pylist(), [1, 2, 3])
self.assertEqual(actual.column('payload').to_pylist(), blobs)
- def test_blob_abort_deletes_uncommitted_files(self):
+ @pytest.mark.python_write
+ def test_blob_abort_preserves_prepared_files_until_committer_aborts(self):
pa_schema = pa.schema([
('id', pa.int32()),
('payload', pa.large_binary()),
@@ -569,6 +570,13 @@ class DataEvolutionFormatsTest(unittest.TestCase):
writer.abort()
+ for file_meta in all_files:
+ self.assertTrue(table.file_io.exists(self._file_path(file_meta)))
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ commit.abort(commit_messages)
+ finally:
+ commit.close()
for file_meta in all_files:
self.assertFalse(
table.file_io.exists(self._file_path(file_meta)),
@@ -902,7 +910,8 @@ class DataEvolutionFormatsTest(unittest.TestCase):
# Vector (vortex) file format for embedding columns
# ------------------------------------------------------------------
- def test_vector_abort_deletes_uncommitted_files(self):
+ @pytest.mark.python_write
+ def
test_vector_abort_preserves_prepared_files_until_committer_aborts(self):
pa_schema = pa.schema([
('id', pa.int64()),
('embed', pa.list_(pa.float32(), 3)),
@@ -935,13 +944,23 @@ class DataEvolutionFormatsTest(unittest.TestCase):
writer.abort()
+ # prepare_commit hands these files to the committer, as Java's
+ # drainIncrement does. The writer no longer owns their cleanup.
+ for file_meta in all_files:
+ self.assertTrue(table.file_io.exists(self._file_path(file_meta)))
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ commit.abort(commit_messages)
+ finally:
+ commit.close()
for file_meta in all_files:
self.assertFalse(
table.file_io.exists(self._file_path(file_meta)),
f"Expected abort to delete {file_meta.file_name}",
)
- def test_vector_close_failure_after_prepare_raises(self):
+ @pytest.mark.python_write
+ def test_vector_close_failure_preserves_prepared_files(self):
from unittest.mock import patch
pa_schema = pa.schema([
@@ -977,6 +996,13 @@ class DataEvolutionFormatsTest(unittest.TestCase):
with self.assertRaisesRegex(RuntimeError, "Close error"):
writer.close()
+ for file_meta in all_files:
+ self.assertTrue(table.file_io.exists(self._file_path(file_meta)))
+ commit = table.new_batch_write_builder().new_commit()
+ try:
+ commit.abort(commit_messages)
+ finally:
+ commit.close()
for file_meta in all_files:
self.assertFalse(
table.file_io.exists(self._file_path(file_meta)),
diff --git a/paimon-python/pypaimon/tests/deletion_vector_path_test.py
b/paimon-python/pypaimon/tests/deletion_vector_path_test.py
index 9ed9b6179d..d1f53116e9 100644
--- a/paimon-python/pypaimon/tests/deletion_vector_path_test.py
+++ b/paimon-python/pypaimon/tests/deletion_vector_path_test.py
@@ -34,7 +34,6 @@ from pypaimon import CatalogFactory, Schema
from pypaimon.manifest.index_manifest_file import IndexManifestFile
from pypaimon.read.native_plan import native_plan, native_version_at_least
from pypaimon.schema.data_types import AtomicType
-from pypaimon.write.file_store_commit import _abort_commit_messages
from pypaimon.write.commit_message import CommitMessage
from pypaimon.write.table_delete import TableDeleteByRowId
@@ -205,7 +204,7 @@ def
test_floating_bucket_dv_does_not_search_python_directory(tmp_path, file_uri,
@pytest.mark.parametrize('layout', ['bucket', 'bucket-external',
'global-external'])
-def test_abort_removes_uncommitted_dv_from_its_actual_directory(tmp_path,
layout):
+def
test_explicit_abort_removes_uncommitted_dv_from_its_actual_directory(tmp_path,
layout):
table = _table(tmp_path, layout, 'a/b')
_delete(table, [0])
builder = table.new_batch_write_builder()
@@ -213,7 +212,7 @@ def
test_abort_removes_uncommitted_dv_from_its_actual_directory(tmp_path, layout
uncommitted = [entry for message in messages for entry in
message.index_adds]
assert uncommitted
files_before = set(tmp_path.rglob('index-*'))
- _abort_commit_messages(table, messages)
+ builder.new_commit().abort(messages)
files_after = set(tmp_path.rglob('index-*'))
removed = {path.name for path in files_before - files_after}
assert removed == {entry.index_file.file_name for entry in uncommitted}
diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py
b/paimon-python/pypaimon/tests/file_store_commit_test.py
index 674b590bed..39c0649c8c 100644
--- a/paimon-python/pypaimon/tests/file_store_commit_test.py
+++ b/paimon-python/pypaimon/tests/file_store_commit_test.py
@@ -40,7 +40,6 @@ from pypaimon.write.file_store_commit import (
FileStoreCommit,
RollbackRetryResult,
RewriteResult,
- _abort_commit_messages,
_reject_compact_increment,
_row_id_check_from_messages,
)
@@ -79,67 +78,53 @@ class TestRowIdCheckFromMessages(unittest.TestCase):
None, [CommitMessage((), 0, [], check_from_snapshot=-1)], 1)
-class TestAbortCommitMessages(unittest.TestCase):
+class TestExplicitCommitAbort(unittest.TestCase):
- @staticmethod
- def _file_meta(**kwargs):
- return DataFileMeta(
- file_name='data.parquet', file_size=1, row_count=1,
- min_key=None, max_key=None, key_stats=None, value_stats=None,
- min_sequence_number=0, max_sequence_number=0, schema_id=0,
- level=0, extra_files=kwargs.pop('extra_files', []), **kwargs)
-
- def test_reconstructs_local_path_after_wire_decode(self):
- table = Mock()
- table.path_factory.return_value.bucket_path.return_value =
'/table/p=1/bucket-0'
- file = self._file_meta()
- message = CommitMessage((1,), 0, [file])
- _abort_commit_messages(table, [message])
- table.file_io.delete_quietly.assert_called_once_with(
- '/table/p=1/bucket-0/data.parquet')
-
- def test_reconstructs_aligned_sidecars_after_wire_decode(self):
- table = Mock()
- table.path_factory.return_value.bucket_path.return_value =
'/table/p=1/bucket-0'
- file = self._file_meta(extra_files=['data.parquet.index'])
- _abort_commit_messages(table, [CommitMessage((1,), 0, [file])])
- self.assertEqual(table.file_io.delete_quietly.call_args_list, [
- unittest.mock.call('/table/p=1/bucket-0/data.parquet'),
- unittest.mock.call('/table/p=1/bucket-0/data.parquet.index'),
- ])
-
- def test_deletes_literal_external_paths_and_preserves_metadata(self):
+ def
test_abort_deletes_new_data_sidecars_changelog_compact_and_indexes_only(self):
with TemporaryDirectory() as directory:
parent = Path(directory) / 'pt=a%2Fb%25%3F%23'
parent.mkdir()
- paths = [parent / 'data.parquet', parent / 'data.parquet.index']
+ paths = [parent / name for name in (
+ 'data.parquet', 'data.parquet.index', 'changelog.parquet',
+ 'compact-before.parquet', 'compact-after.parquet',
+ 'compact-changelog.parquet', 'index-file', 'old-index')]
for path in paths:
- path.touch()
- external_path = 'file:' + paths[0].as_posix()
- file = self._file_meta(external_path=external_path,
extra_files=[paths[1].name])
- table = Mock(file_io=LocalFileIO())
- _abort_commit_messages(table, [CommitMessage(('a/b%?#',), 0,
[file])])
- self.assertFalse(any(path.exists() for path in paths))
- self.assertEqual(file.external_path, external_path)
- self.assertEqual(file.extra_files, [paths[1].name])
- table.path_factory.assert_not_called()
-
- def test_index_path_failure_does_not_escape_abort(self):
- table = Mock()
- table.path_factory.side_effect = RuntimeError("path lookup failed")
- index_file = Mock(file_name="index-file", external_path=None)
- message = Mock(
- new_files=[],
- changelog_files=[],
- index_adds=[Mock(index_file=index_file)],
- compact_after=[],
- compact_changelog_files=[],
- compact_index_adds=[],
- )
-
- with self.assertLogs(
- 'pypaimon.write.file_store_commit', level='WARNING'):
- _abort_commit_messages(table, [message])
+ path.write_bytes(b'keep')
+
+ def meta(path, **kwargs):
+ return DataFileMeta(
+ file_name=path.name, file_size=4, row_count=1,
+ min_key=None, max_key=None, key_stats=None,
value_stats=None,
+ min_sequence_number=0, max_sequence_number=0, schema_id=0,
+ level=0, extra_files=kwargs.pop('extra_files', []),
+ external_path='file:' + path.as_posix(), **kwargs)
+
+ index_entry = Mock(index_file=Mock(index_type='BTREE',
external_path=str(paths[-2])))
+ old_index = Mock(index_file=Mock(index_type='BTREE',
external_path=str(paths[-1])))
+ message = CommitMessage(
+ ('a/b%?#',), 0, [meta(paths[0], extra_files=[paths[1].name])],
+ deleted_files=[meta(paths[3])],
changelog_files=[meta(paths[2])],
+ compact_before=[meta(paths[3])],
compact_after=[meta(paths[4])],
+ compact_changelog_files=[meta(paths[5])],
index_adds=[index_entry],
+ index_deletes=[old_index], compact_index_adds=[index_entry],
+ compact_index_deletes=[old_index])
+ commit = FileStoreCommit.__new__(FileStoreCommit)
+ commit.table = Mock(file_io=LocalFileIO())
+ commit.abort([message])
+ commit.abort([message])
+ for i, path in enumerate(paths):
+ self.assertEqual(path.exists(), i in (3, len(paths) - 1),
str(path))
+
+ def test_abort_continues_after_storage_cleanup_errors(self):
+ commit = FileStoreCommit.__new__(FileStoreCommit)
+ commit.table = Mock()
+ commit.table.file_io.delete_quietly.side_effect = [OSError('cleanup
failed'), None]
+ first = Mock()
+ first.collect_files.return_value = ['/first']
+ second = Mock()
+ second.collect_files.return_value = ['/second']
+ commit.abort([CommitMessage((), 0, [first, second])])
+ self.assertEqual(commit.table.file_io.delete_quietly.call_count, 2)
class TestFileStoreCommitRowTracking(unittest.TestCase):
diff --git a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
index 748378f7a8..73e39a74b9 100644
--- a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
@@ -2859,7 +2859,6 @@ class LeRobotValidationTest(unittest.TestCase):
target_frame_values = {
55: 168, 56: 216, 61: 56, 62: 88, 63: 120,
}
- expected_frame_values = [168, 216, 56, 88, 120]
info_dir = temp_dir / "meta"
info_dir.mkdir()
@@ -2900,6 +2899,14 @@ class LeRobotValidationTest(unittest.TestCase):
container.mux(packet)
for packet in stream.encode():
container.mux(packet)
+ # MPEG4 is lossy: compare imported frames with the source video's
+ # decoded pixels rather than its uncompressed input images.
+ with av.open(str(video_path)) as container:
+ expected_frame_values = [
+ float(frame.to_ndarray(format="rgb24").mean())
+ for index, frame in enumerate(container.decode(video=0))
+ if index in target_frame_values
+ ]
camera_b_path = (
temp_dir / "videos/camera_b/chunk-000/file-000.mp4")
camera_b_path.parent.mkdir(parents=True)
@@ -2999,7 +3006,7 @@ class LeRobotValidationTest(unittest.TestCase):
np.testing.assert_allclose(
[row["decoded"][0] for row in decoded_rows],
expected_frame_values,
- atol=5,
+ atol=1e-6,
)
np.testing.assert_allclose(
[row["decoded"][1] for row in decoded_rows],
@@ -3066,8 +3073,8 @@ class LeRobotValidationTest(unittest.TestCase):
float(middle["camera"][0].mean()) * 255,
float(middle["camera"][1].mean()) * 255,
],
- [56, 88],
- atol=5,
+ expected_frame_values[2:4],
+ atol=1e-4,
)
self.assertEqual(
[False, False], middle["camera_is_pad"].tolist())
diff --git a/paimon-python/pypaimon/tests/native_blob_collection_write_test.py
b/paimon-python/pypaimon/tests/native_blob_collection_write_test.py
new file mode 100644
index 0000000000..16525c5493
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_blob_collection_write_test.py
@@ -0,0 +1,354 @@
+# 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.
+
+"""Blob Arrow writes must retain row identity across physical file rolls."""
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.schema.data_types import AtomicType, ArrayType, DataField,
MapType, PyarrowFieldParser
+from pypaimon.table.row.blob import BlobDescriptor
+from pypaimon.write.native_write import NativeTableWrite
+from pypaimon.tests.native_blob_write_test import _read, _physical_files
+
+pytestmark = pytest.mark.native_plan
+
+
+def _table(tmp_path, native, external=False, optimize=True, catalog=None,
partitioned=False, not_null=False):
+ if catalog is None:
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path /
'warehouse')})
+ catalog.create_database('db', True)
+ blob = AtomicType('BLOB', nullable=not not_null)
+ fields = [DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'items', ArrayType(True, blob)),
+ DataField(2, 'attrs', MapType(True, AtomicType('STRING'), blob))]
+ options = {'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true',
+ 'write.native.enabled': str(native).lower(),
'target-file-row-num': '2',
+ 'blob.target-file-size': '32 B',
+ 'data-evolution.write-cols-optimization.enabled':
str(optimize).lower()}
+ if external:
+ options['data-file.external-paths'] = (tmp_path / 'external').as_uri()
+ schema = Schema(fields=fields, options=options, partition_keys=['id'] if
partitioned else [])
+ catalog.create_table('db.t', schema, False)
+ return catalog.get_table('db.t')
+
+
+def _data(table, tmp_path, descriptor, start):
+ payload = b'payload' * 10
+ if descriptor:
+ source = tmp_path / 'payload'
+ source.write_bytes(b'prefix' + payload)
+ value = BlobDescriptor(str(source), 6, -1).serialize()
+ else:
+ value = payload
+ rows = [{'id': -1, 'items': [b'ignored'], 'attrs': [('ignored',
b'ignored')]},
+ {'id': start, 'items': [value, None, b''], 'attrs': [('first',
value), ('', b''), ('null', None)]},
+ {'id': start + 1, 'items': None, 'attrs': None},
+ {'id': start + 2, 'items': [], 'attrs': []}]
+ data = pa.Table.from_pylist(rows,
schema=PyarrowFieldParser.from_paimon_schema(table.fields)).slice(1)
+ expected = [{'id': start, 'items': [payload, None, b''],
+ 'attrs': [('first', payload), ('', b''), ('null', None)]},
rows[2], rows[3]]
+ return data, expected
+
+
[email protected]('native_write', [False, True])
[email protected]('external', [False, True])
[email protected]('descriptor', [False, True])
[email protected]('stream', [False, True])
+def test_collection_blob_write_cross_reads_and_rolls(tmp_path, native_write,
external, descriptor, stream):
+ table = _table(tmp_path, native_write, external)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ expected = []
+ try:
+ assert isinstance(writer, NativeTableWrite) == native_write
+ for identifier in range(1, 3 if stream else 2):
+ data, rows = _data(table, tmp_path, descriptor, identifier * 3)
+ writer.write_arrow(data)
+ expected.extend(rows)
+ messages = writer.prepare_commit(identifier) if stream else
writer.prepare_commit()
+ files = [file for message in messages for file in
message.new_files]
+ for name in ('items', 'attrs'):
+ dedicated = [file for file in files if file.write_cols ==
[name]]
+ assert dedicated and all(file.file_name.endswith('.blob') for
file in dedicated)
+ assert sum(file.row_count for file in dedicated) == 3
+ commit.commit(messages, identifier) if stream else
commit.commit(messages)
+ for planner in (False, True):
+ for reader in (False, True):
+ assert _read(table, planner, reader) == expected
+ if native_write:
+ assert writer._python_writer is None
+ finally:
+ writer.close()
+ commit.close()
+
+
[email protected]('native_commit', [False, True])
[email protected]('optimize', [False, True])
+def test_collection_blob_descriptors_are_payload_ranges(tmp_path,
native_rest_catalog, native_commit, optimize):
+ table = _table(tmp_path, True, optimize=optimize,
catalog=native_rest_catalog).copy(
+ {'commit.native.enabled': str(native_commit).lower()})
+ data, expected = _data(table, tmp_path, False, 1)
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ assert isinstance(writer, NativeTableWrite)
+ writer.write_arrow(data)
+ messages = writer.prepare_commit()
+ normal = [file for message in messages for file in message.new_files
if file.file_name.endswith('.parquet')]
+ assert all(file.write_cols == (None if optimize else ['id']) for file
in normal)
+ commit.commit(messages)
+ assert (commit._native_commit is not None) == native_commit
+ for reader in (False, True):
+ rows = _read(table.copy({'blob-as-descriptor': 'true'}), True,
reader)
+ assert rows[1]['items'] is None and rows[2]['items'] == []
+ assert rows[1]['attrs'] is None and rows[2]['attrs'] == []
+ for descriptors, values in ((rows[0]['items'],
expected[0]['items']),
+ ([item for _, item in
rows[0]['attrs']],
+ [item for _, item in
expected[0]['attrs']])):
+ for encoded, payload in zip(descriptors, values):
+ if payload is None:
+ assert encoded is None
+ else:
+ descriptor = BlobDescriptor.deserialize(encoded)
+ assert descriptor.uri.endswith('.blob') and
descriptor.length == len(payload)
+ from urllib.parse import urlparse
+ with open(urlparse(descriptor.uri).path, 'rb') as
source:
+ source.seek(descriptor.offset)
+ assert source.read(descriptor.length) == payload
+ finally:
+ writer.close()
+ commit.close()
+
+
[email protected]('action', ['close', 'abort', 'failed-write',
'short-read'])
+def test_collection_blob_failed_writes_and_abort_clean_all_groups(tmp_path,
action):
+ table = _table(tmp_path, True)
+ writer = table.new_batch_write_builder().new_write()
+ try:
+ assert isinstance(writer, NativeTableWrite)
+ data, _ = _data(table, tmp_path, False, 1)
+ writer.write_arrow(data)
+ assert {path.suffix for path in _physical_files(tmp_path)} ==
{'.parquet', '.blob'}
+ if action in ('failed-write', 'short-read'):
+ if action == 'short-read':
+ source = tmp_path / 'truncated'
+ source.write_bytes(b'x' * (8 * 1024 * 1024))
+ missing = BlobDescriptor(str(source), 0, 8 * 1024 * 1024 +
1).serialize()
+ else:
+ missing = BlobDescriptor(str(tmp_path / 'missing'), 0,
10).serialize()
+ data = pa.table({'id': [5], 'items': [[b'already-copied',
missing]],
+ 'attrs': [[('x', b'ok')]]}, schema=data.schema)
+ with pytest.raises(Exception):
+ writer.write_arrow(data)
+ else:
+ getattr(writer, action)()
+ assert not _physical_files(tmp_path)
+ assert table.snapshot_manager().get_latest_snapshot() is None
+ finally:
+ writer.close()
+
+
[email protected]('native_write', [False, True])
[email protected]('action', ['abort', 'failed-write'])
+def test_stream_collection_failure_keeps_already_committed_groups(tmp_path,
native_write, action):
+ table = _table(tmp_path, native_write)
+ builder = table.new_stream_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ data, expected = _data(table, tmp_path, False, 1)
+ writer.write_arrow(data)
+ messages = writer.prepare_commit(1)
+ committed = {file.file_name for message in messages for file in
message.new_files}
+ commit.commit(messages, 1)
+ assert writer.prepare_commit(2) == []
+ if action == 'failed-write':
+ missing = BlobDescriptor(str(tmp_path / 'missing'), 0,
1).serialize()
+ with pytest.raises(Exception):
+ writer.write_arrow(pa.table({'id': [10], 'items': [[b'ok',
missing]], 'attrs': [[]]},
+ schema=data.schema))
+ else:
+ writer.write_arrow(data)
+ writer.abort()
+ assert committed <= {path.name for path in _physical_files(tmp_path)}
+ for native_read in (False, True):
+ assert _read(table, True, native_read) == expected
+ finally:
+ writer.close()
+ commit.close()
+
+
[email protected]('native_write', [False, True])
[email protected]('native_commit', [False, True])
[email protected]('stream', [False, True])
[email protected]('published', [False, True])
+def test_collection_commit_exception_preserves_files(
+ tmp_path, native_rest_catalog, native_write, native_commit, stream,
published):
+ from unittest.mock import Mock, patch
+
+ table = _table(tmp_path, native_write, external=True,
catalog=native_rest_catalog).copy(
+ {'commit.native.enabled': str(native_commit).lower()})
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ expected = []
+ try:
+ assert isinstance(writer, NativeTableWrite) == native_write
+ if stream:
+ data, rows = _data(table, tmp_path, True, 1)
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit(1), 1)
+ expected.extend(rows)
+
+ data, rows = _data(table, tmp_path, True, 4)
+ writer.write_arrow(data)
+ messages = writer.prepare_commit(2) if stream else
writer.prepare_commit()
+ files = _physical_files(tmp_path)
+ assert {path.suffix for path in files} >= {'.parquet', '.blob'}
+ if native_commit:
+ prepared = commit._prepare_native_commit(messages)
+ assert prepared is not None
+ native, native_messages = prepared
+ proxy = Mock(wraps=native)
+ publish = native.commit
+ else:
+ publish = commit.file_store_commit.commit
+
+ def fail_commit(*args, **kwargs):
+ if published:
+ publish(*args, **kwargs)
+ raise OSError('Commit outcome is unknown')
+
+ if native_commit:
+ proxy.commit.side_effect = fail_commit
+ failing_commit = patch.object(
+ commit, '_prepare_native_commit', return_value=(proxy,
native_messages))
+ else:
+ failing_commit = patch.object(commit.file_store_commit, 'commit',
side_effect=fail_commit)
+ with failing_commit, \
+ patch.object(commit.file_store_commit, 'abort',
side_effect=AssertionError('Unsafe abort')):
+ with pytest.raises(OSError, match='Commit outcome is unknown'):
+ commit.commit(messages, 2) if stream else
commit.commit(messages)
+ if native_commit:
+ proxy.commit.assert_called_once()
+ proxy.abort.assert_not_called()
+
+ # An unknown commit outcome must never trigger explicit commit abort.
+ writer.abort()
+ writer.close()
+ assert _physical_files(tmp_path) == files
+ assert (tmp_path / 'payload').exists()
+ if published:
+ expected.extend(rows)
+ latest = table.snapshot_manager().get_latest_snapshot()
+ assert (latest.id if latest is not None else 0) == int(stream) +
int(published)
+ for native_read in (False, True):
+ read_builder = table.copy({'read.native.enabled':
str(native_read).lower()}).new_read_builder()
+ actual =
read_builder.new_read().to_arrow(read_builder.new_scan().plan().splits())
+ assert actual.sort_by('id').to_pylist() == expected
+ finally:
+ writer.close()
+ commit.close()
+
+
[email protected]('projection', [
+ ['id', "attrs['first']", "attrs['null']", "attrs['missing']"],
+ {'payload': "attrs['first']", 'array': 'items'},
+ ['attrs', "attrs['first']", 'id'],
+])
+def test_native_collection_projection_keeps_selected_blob_keys(tmp_path,
projection):
+ from unittest.mock import patch
+
+ table = _table(tmp_path, True)
+ data, _ = _data(table, tmp_path, False, 1)
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ for native in (False, True):
+ read_table = table.copy({'read.native.enabled': str(native).lower()})
+ read_builder =
read_table.new_read_builder().with_projection(projection)
+ read = read_builder.new_read()
+ splits = read_builder.new_scan().plan().splits()
+ if native:
+ with patch.object(read, '_create_split_read',
side_effect=AssertionError('Python fallback')):
+ result = read.to_arrow(splits)
+ else:
+ expected = read.to_arrow(splits)
+ continue
+ assert result.schema == expected.schema
+ assert result.to_pylist() == expected.to_pylist()
+
+
+def
test_collection_partial_partition_prepare_keeps_cleanup_ownership(tmp_path):
+ from unittest.mock import patch
+ from pypaimon.write.writer.composite_data_writer import CompositeDataWriter
+
+ table = _table(tmp_path, False, partitioned=True)
+ writer = table.new_batch_write_builder().new_write()
+ try:
+ data, _ = _data(table, tmp_path, False, 1)
+ writer.write_arrow(data)
+ partitions = list(writer.file_store_write.data_writers.values())
+ assert len(partitions) == 3 and all(isinstance(partition,
CompositeDataWriter) for partition in partitions)
+ with patch.object(partitions[1], 'prepare_commit',
side_effect=RuntimeError('Second partition failed')):
+ with pytest.raises(RuntimeError, match='Second partition failed'):
+ writer.prepare_commit()
+ assert partitions[0].committed_files
+ assert _physical_files(tmp_path)
+ writer.abort()
+ assert not _physical_files(tmp_path)
+ assert table.snapshot_manager().get_latest_snapshot() is None
+ finally:
+ writer.close()
+
+
[email protected]('native_write', [False, True])
[email protected]('sliced', [False, True])
[email protected]('aliases', [False, True])
+def test_not_null_collection_children_ignore_masked_and_unreferenced_values(
+ tmp_path, native_write, sliced, aliases):
+ table = _table(tmp_path, native_write, not_null=True)
+ offsets = pa.array([0, 1, 2, 3, 4], type=pa.int32())
+ mask = pa.array([True, False, True, False])
+ values = pa.array([None, b'first', None, b'last'], type=pa.large_binary())
+ items = pa.ListArray.from_arrays(
+ offsets, values, mask=mask,
+ type=pa.list_(pa.field('item' if aliases else 'element',
pa.large_binary(), nullable=False)))
+ attrs = pa.MapArray.from_arrays(
+ offsets, pa.array(['hidden', 'first', 'hidden', 'last']), values,
mask=mask,
+ type=pa.map_(pa.field('source_key' if aliases else 'key', pa.string(),
nullable=False),
+ pa.field('source_value' if aliases else 'value',
pa.large_binary(), nullable=False)))
+ data = pa.table({'id': pa.array(range(4), type=pa.int32()), 'items':
items, 'attrs': attrs})
+ if sliced:
+ data = data.slice(1, 1)
+ data.validate(full=True)
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ assert isinstance(writer, NativeTableWrite) == native_write
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit())
+ for native_read in (False, True):
+ assert _read(table, True, native_read) == data.to_pylist()
+ finally:
+ writer.close()
+ commit.close()
diff --git a/paimon-python/pypaimon/tests/native_blob_write_test.py
b/paimon-python/pypaimon/tests/native_blob_write_test.py
index d78167e33a..9bdfd24fe1 100644
--- a/paimon-python/pypaimon/tests/native_blob_write_test.py
+++ b/paimon-python/pypaimon/tests/native_blob_write_test.py
@@ -150,7 +150,7 @@ def
test_native_blob_cleanup_includes_completed_groups(tmp_path, external, actio
@pytest.mark.parametrize('external', [False, True])
@pytest.mark.parametrize('stream', [False, True])
-def test_native_blob_abort_removes_prepared_and_outstanding_files(tmp_path,
external, stream):
+def
test_native_blob_abort_preserves_prepared_and_removes_outstanding_files(tmp_path,
external, stream):
options = {'data-file.path-directory': 'data/nested'}
if external:
options['data-file.external-paths'] = (tmp_path / 'external').as_uri()
@@ -163,11 +163,13 @@ def
test_native_blob_abort_removes_prepared_and_outstanding_files(tmp_path, exte
writer.write_arrow(_data(identifier * 2))
messages = writer.prepare_commit(identifier) if stream else
writer.prepare_commit()
assert messages
- assert {path.suffix for path in _physical_files(tmp_path)} ==
{'.parquet', '.blob', '.index'}
+ prepared = _physical_files(tmp_path)
+ assert {path.suffix for path in prepared} == {'.parquet', '.blob',
'.index'}
writer.write_arrow(_data(6))
+ assert _physical_files(tmp_path) > prepared
writer.abort()
writer.abort()
- assert not _physical_files(tmp_path)
+ assert _physical_files(tmp_path) == prepared
assert table.snapshot_manager().get_latest_snapshot() is None
finally:
writer.close()
@@ -198,7 +200,7 @@ def
test_native_blob_close_releases_prepared_files_to_committer(tmp_path, extern
@pytest.mark.parametrize('external', [False, True])
@pytest.mark.parametrize('commit_native_option', [False, True])
-def test_native_blob_stream_abort_preserves_submitted_files(tmp_path,
external, commit_native_option):
+def test_native_blob_stream_abort_preserves_all_prepared_files(tmp_path,
external, commit_native_option):
options = {'commit.native.enabled': str(commit_native_option).lower()}
if external:
options['data-file.external-paths'] = (tmp_path / 'external').as_uri()
@@ -211,10 +213,12 @@ def
test_native_blob_stream_abort_preserves_submitted_files(tmp_path, external,
submitted = _physical_files(tmp_path)
writer.write_arrow(_data(2))
assert writer.prepare_commit(2)
+ prepared = _physical_files(tmp_path)
+ assert prepared > submitted
writer.write_arrow(_data(4))
assert _physical_files(tmp_path) > submitted
writer.abort()
- assert _physical_files(tmp_path) == submitted
+ assert _physical_files(tmp_path) == prepared
for planner in (False, True):
for reader in (False, True):
assert _read(table, planner, reader) == _data().to_pylist()
diff --git a/paimon-python/pypaimon/tests/native_cross_partition_write_test.py
b/paimon-python/pypaimon/tests/native_cross_partition_write_test.py
index 04b6ae1059..edd9d963c7 100644
--- a/paimon-python/pypaimon/tests/native_cross_partition_write_test.py
+++ b/paimon-python/pypaimon/tests/native_cross_partition_write_test.py
@@ -154,7 +154,7 @@ def test_cross_partition_rowkind_and_row_input(tmp_path):
@pytest.mark.parametrize('committed', [False, True])
-def test_cross_partition_abort_preserves_committed_files(tmp_path, committed):
+def
test_cross_partition_abort_preserves_prepared_and_committed_files(tmp_path,
committed):
table = _table(tmp_path)
builder = table.new_batch_write_builder()
writer, commit = builder.new_write(), builder.new_commit()
@@ -167,7 +167,7 @@ def
test_cross_partition_abort_preserves_committed_files(tmp_path, committed):
if committed:
commit.commit(messages)
writer.abort()
- assert all(table.file_io.exists(p) == committed for p in paths)
+ assert all(table.file_io.exists(p) for p in paths)
finally:
writer.close()
commit.close()
diff --git a/paimon-python/pypaimon/tests/native_dynamic_write_test.py
b/paimon-python/pypaimon/tests/native_dynamic_write_test.py
index 11ec98ddbf..ef84c1b73b 100644
--- a/paimon-python/pypaimon/tests/native_dynamic_write_test.py
+++ b/paimon-python/pypaimon/tests/native_dynamic_write_test.py
@@ -172,8 +172,8 @@ def test_dynamic_abort_index_ownership(tmp_path,
bucket_local, commit_first):
if commit_first:
builder.new_commit().commit(messages)
writer.abort()
- assert table.file_io.exists(path) == commit_first
- assert all(table.file_io.exists(f.file_path) == commit_first
+ assert table.file_io.exists(path)
+ assert all(table.file_io.exists(f.file_path)
for m in messages for f in m.new_files)
diff --git a/paimon-python/pypaimon/tests/native_nested_update_test.py
b/paimon-python/pypaimon/tests/native_nested_update_test.py
index db691e94f6..042c08acac 100644
--- a/paimon-python/pypaimon/tests/native_nested_update_test.py
+++ b/paimon-python/pypaimon/tests/native_nested_update_test.py
@@ -257,7 +257,7 @@ def test_view_pairs_update_map(tmp_path, native, layout):
@pytest.mark.parametrize('native', [False, True])
@pytest.mark.parametrize('data_type,values', CASES)
@pytest.mark.parametrize('overlap', [False, True])
-def test_nested_grouped_updates_abort_overlap(tmp_path, native, data_type,
values, overlap):
+def test_nested_grouped_updates_preserve_files_on_overlap(tmp_path, native,
data_type, values, overlap):
table, _ = _table(tmp_path, data_type, native)
builder = table.new_batch_write_builder()
update = builder.new_update().with_update_type(['value'])
@@ -267,7 +267,7 @@ def test_nested_grouped_updates_abort_overlap(tmp_path,
native, data_type, value
if overlap:
with pytest.raises(ValueError):
update.update_by_arrow_batches_with_row_id(iter([data, data]))
- assert set(tmp_path.rglob('*.parquet')) == before
+ assert set(tmp_path.rglob('*.parquet')) > before
else:
messages =
update.update_by_arrow_batches_with_row_id(iter([data.slice(0, 0), data]))
_commit(builder, messages, False)
diff --git a/paimon-python/pypaimon/tests/native_postpone_fixed_write_test.py
b/paimon-python/pypaimon/tests/native_postpone_fixed_write_test.py
index d9dafe6148..7ac8d52b8e 100644
--- a/paimon-python/pypaimon/tests/native_postpone_fixed_write_test.py
+++ b/paimon-python/pypaimon/tests/native_postpone_fixed_write_test.py
@@ -185,7 +185,7 @@ def
test_provided_plan_does_not_fill_missing_partitions_from_default(tmp_path):
@pytest.mark.parametrize('prepared', [False, True])
-def test_abort_and_close_keep_commit_message_ownership(tmp_path, prepared):
+def test_abort_and_close_preserve_prepared_files(tmp_path, prepared):
table = _table(tmp_path)
builder = table.new_postpone_fixed_bucket_write_builder()
writer = builder.new_write()
@@ -197,7 +197,10 @@ def
test_abort_and_close_keep_commit_message_ownership(tmp_path, prepared):
assert paths and all(table.file_io.exists(path) for path in paths)
writer.abort()
writer.close()
- assert all(not table.file_io.exists(path) for path in paths)
+ if prepared:
+ assert all(table.file_io.exists(path) for path in paths)
+ else:
+ assert not list(tmp_path.rglob('*.parquet'))
assert table.snapshot_manager().get_latest_snapshot() is None
diff --git a/paimon-python/pypaimon/tests/native_postpone_write_test.py
b/paimon-python/pypaimon/tests/native_postpone_write_test.py
index fcd6d2eaaf..94f70e69fc 100644
--- a/paimon-python/pypaimon/tests/native_postpone_write_test.py
+++ b/paimon-python/pypaimon/tests/native_postpone_write_test.py
@@ -134,7 +134,7 @@ def test_postpone_native_overwrite_and_abort(tmp_path,
native_rest_catalog, nati
messages = writer.prepare_commit()
paths = [f.file_path for m in messages for f in m.new_files]
writer.abort()
- assert paths and all(not table.file_io.exists(path) for path in paths)
+ assert paths and all(table.file_io.exists(path) for path in paths)
finally:
writer.close()
diff --git a/paimon-python/pypaimon/tests/native_read_test.py
b/paimon-python/pypaimon/tests/native_read_test.py
index bc28794225..c076fa2666 100644
--- a/paimon-python/pypaimon/tests/native_read_test.py
+++ b/paimon-python/pypaimon/tests/native_read_test.py
@@ -899,7 +899,7 @@ def test_native_avro_read_uses_native_path():
def test_native_read_falls_back_for_unsupported_dedicated_file():
read = _table_read()
schema = pa.schema([('id', pa.int32())])
- split = _Split('camera.video')
+ split = _Split('camera.unsupported')
split._native_split = object()
with patch('pypaimon.read.native_plan.native_read') as native:
@@ -908,10 +908,11 @@ def
test_native_read_falls_back_for_unsupported_dedicated_file():
native.assert_not_called()
-def test_native_read_supports_blob_file_and_forwards_parallelism():
[email protected]('file_name', ['picture.blob', 'camera.video'])
+def
test_native_read_supports_blob_and_video_files_and_forwards_parallelism(file_name):
read = _table_read()
schema = pa.schema([('id', pa.int32())])
- split = _Split('picture.blob')
+ split = _Split(file_name)
split._native_split = object()
with patch(
diff --git a/paimon-python/pypaimon/tests/native_update_test.py
b/paimon-python/pypaimon/tests/native_update_test.py
index 1c01054aef..cf4e138bbf 100644
--- a/paimon-python/pypaimon/tests/native_update_test.py
+++ b/paimon-python/pypaimon/tests/native_update_test.py
@@ -373,8 +373,9 @@ def
test_native_batch_update_preserves_input_table_boundaries(tmp_path):
.update_by_arrow_batches_with_row_id(interleaved_tables()))
assert staged and all(message.check_from_snapshot == read_snapshot_id
for message in staged)
- from pypaimon.write.file_store_commit import _abort_commit_messages
- _abort_commit_messages(table, staged)
+ # Keep staged files: the outcome of a concurrent commit can be unknown.
+ assert all(table.file_io.exists(file.physical_path())
+ for message in staged for file in message.new_files)
def test_native_predicate_update_invokes_callable_by_file_group(tmp_path):
diff --git a/paimon-python/pypaimon/tests/native_video_read_test.py
b/paimon-python/pypaimon/tests/native_video_read_test.py
new file mode 100644
index 0000000000..1ab9de78cc
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_video_read_test.py
@@ -0,0 +1,194 @@
+# 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 unittest.mock import patch
+from urllib.parse import urlparse
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.schema.data_types import AtomicType, DataField
+from pypaimon.table.row.blob import Blob, VideoFrameDescriptor
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.table.row.video_keyframe_index import VideoKeyframeIndex
+
+pytestmark = pytest.mark.native_plan
+
+
+def _table(tmp_path, multiple, external=False, catalog=None):
+ if catalog is None:
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path /
'warehouse')})
+ catalog.create_database('db', True)
+ names = ['left', 'right'] if multiple else ['left']
+ fields = [DataField(0, 'id', AtomicType('INT'))]
+ fields.extend(DataField(index + 1, name, AtomicType('BLOB')) for index,
name in enumerate(names))
+ options = {'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true',
+ 'video-frame-field': ','.join(names), 'write.native.enabled':
'false'}
+ if external:
+ options['data-file.external-paths'] = (tmp_path / 'external').as_uri()
+ catalog.create_table('db.t', Schema(fields=fields, options=options), False)
+ table = catalog.get_table('db.t')
+ mapping = VideoKeyframeIndex([(0, 1)], [(0, 0, 0)]).serialize()
+ sources = []
+ for ordinal in range(4 if multiple else 2):
+ source = tmp_path / ('source-%d.mp4' % ordinal)
+ source.write_bytes(bytes([ordinal + 1]) * 10 + mapping)
+ sources.append(source.as_uri())
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ for row, (ordinal, frame) in enumerate([(0, 2), (0, 3), (None, None),
(1, 7), (1, 8), (0, 10)]):
+ values = [row]
+ for index in range(len(names)):
+ if ordinal is None:
+ values.append(None)
+ else:
+ uri = sources[ordinal + index * 2]
+ descriptor = VideoFrameDescriptor(uri, 0, 10, frame, 10,
len(mapping))
+
values.append(Blob.from_descriptor(table.file_io.uri_reader_factory.create(uri),
descriptor))
+ writer.write_row(GenericRow(values, table.fields))
+ commit.commit(writer.prepare_commit())
+ finally:
+ writer.close()
+ commit.close()
+ return table
+
+
+def _read(table, native, projection=None, filtered=False, limit=None):
+ table = table.copy({'read.native.enabled': str(native).lower(),
+ 'scan.native-plan.enabled': str(native).lower()})
+ builder = table.new_read_builder()
+ if projection is not None:
+ builder.with_projection(projection)
+ if filtered:
+
builder.with_filter(table.new_read_builder().new_predicate_builder().greater_than('id',
0))
+ if limit is not None:
+ builder.with_limit(limit)
+ read = builder.new_read()
+ splits = builder.new_scan().plan().splits()
+ if native:
+ with patch.object(read, '_create_split_read',
side_effect=AssertionError('Python video read fallback')):
+ return read.to_arrow(splits),
read.to_arrow_batch_reader(splits).read_all()
+ return read.to_arrow(splits), read.to_arrow_batch_reader(splits).read_all()
+
+
[email protected]('multiple', [False, True])
[email protected]('external', [False, True])
[email protected]('descriptor_mode', [False, True])
+def test_native_video_reads_match_java_lazy_frame_and_keyframe_descriptors(
+ tmp_path, multiple, external, descriptor_mode):
+ table = _table(tmp_path, multiple, external).copy({'blob-as-descriptor':
str(descriptor_mode).lower()})
+ expected = sorted(_read(table, False)[0].to_pylist(), key=lambda row:
row['id'])
+ for result in _read(table, True):
+ rows = sorted(result.to_pylist(), key=lambda row: row['id'])
+ assert rows == expected and len(rows) == 6
+ for name in ('left', 'right') if multiple else ('left',):
+ assert rows[2][name] is None
+ frames = [VideoFrameDescriptor.deserialize(row[name]) for row in
rows if row[name] is not None]
+ assert [frame.frame_index for frame in frames] == [2, 3, 7, 8, 10]
+ assert frames[0].payload_descriptor ==
frames[1].payload_descriptor == frames[4].payload_descriptor
+ assert frames[2].payload_descriptor == frames[3].payload_descriptor
+ for frame in frames:
+ index = frame.keyframe_index_descriptor
+ assert index is not None
+ with open(urlparse(index.uri).path, 'rb') as source:
+ source.seek(index.offset)
+ assert
VideoKeyframeIndex.deserialize(source.read(index.length)).keyframes == ((0, 0,
0),)
+
+
[email protected]('projection', [[], ['id'], {'frame': 'left'},
['left', 'left', 'id']])
[email protected]('limit', [None, 3])
+def
test_native_video_projection_filter_and_limit_preserve_logical_rows(tmp_path,
projection, limit):
+ table = _table(tmp_path, True)
+ expected = _read(table, False, projection, True, limit)[0]
+ for result in _read(table, True, projection, True, limit):
+ assert result.schema == expected.schema
+ assert result.to_pylist() == expected.to_pylist()
+ assert result.num_rows == expected.num_rows
+
+
+def test_native_update_of_normal_columns_keeps_packed_video_frames(tmp_path,
native_rest_catalog):
+ table = _table(tmp_path, False, catalog=native_rest_catalog).copy(
+ {'write.native.enabled': 'true', 'commit.native.enabled': 'true'})
+ before = _read(table, True)[0].to_pylist()
+ builder = table.new_batch_write_builder()
+ update = builder.new_update().with_update_type(['id'])
+ with
patch('pypaimon.write.table_update.BatchTableUpdate._update_by_arrow_batches_with_row_id',
+ side_effect=AssertionError('Python update fallback')):
+ messages = update.update_by_arrow_batches_with_row_id(iter([
+ pa.table({'_ROW_ID': [0], 'id': [20]},
schema=pa.schema([('_ROW_ID', pa.int64()), ('id', pa.int32())]))]))
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ assert commit._native_commit is not None
+ finally:
+ commit.close()
+ after = _read(table, True)[0].to_pylist()
+ assert sorted(row['id'] for row in after) == [1, 2, 3, 4, 5, 20]
+ assert [row['left'] for row in after] == [row['left'] for row in before]
+
+
+def
test_native_video_placeholder_falls_back_but_null_stops_older_frames(tmp_path):
+ from dataclasses import replace
+ from pypaimon.common.delta_varint_compressor import DeltaVarintCompressor
+ from pypaimon.read.split import DataSplit
+ import struct
+
+ table = _table(tmp_path, False).copy({'read.native.enabled': 'true'})
+ builder = table.new_read_builder()
+ split = builder.new_scan().plan().splits()[0]
+ read = builder.new_read()
+ before = read.to_arrow([split]).to_pylist()
+ original = next(file for file in split.files if
file.file_name.endswith('.video'))
+ indexes = [[], [], [1, 1, 1, 3], [-2, -1, -2, -2], [0, 0, 0, 0]]
+ compressed = [DeltaVarintCompressor.compress(values) for values in indexes]
+ data = b''.join(compressed) + struct.pack('<IIIIIIB', *(len(index) for
index in compressed), 0x4F454449, 1)
+ newer = tmp_path / 'newest.video'
+ newer.write_bytes(data)
+ added = replace(original, file_name=newer.name, file_path=str(newer),
external_path=str(newer),
+ file_size=len(data),
min_sequence_number=original.max_sequence_number + 1,
+ max_sequence_number=original.max_sequence_number + 1)
+ combined = DataSplit(split.files + [added], split.partition, split.bucket,
+ snapshot_id=split.snapshot_id,
bucket_path=split.bucket_path,
+ total_buckets=split.total_buckets)
+ expected = [dict(row) for row in before]
+ expected[1]['left'] = None
+ with patch.object(read, '_create_split_read',
side_effect=AssertionError('Python fallback')):
+ for result in (read.to_arrow([combined]),
read.to_arrow_batch_reader([combined]).read_all()):
+ assert result.to_pylist() == expected
+
+
+def test_native_video_deletion_vectors_filter_both_arrow_readers(tmp_path,
native_rest_catalog):
+ from pypaimon.write.table_update import TableDeleteByRowId
+
+ table = _table(tmp_path, True, catalog=native_rest_catalog).copy(
+ {'deletion-vectors.enabled': 'true', 'write.native.enabled': 'true',
'commit.native.enabled': 'true'})
+ builder = table.new_batch_write_builder()
+ with patch.object(TableDeleteByRowId, 'delete',
+ side_effect=AssertionError('Python delete fallback')):
+ messages = builder.new_update().delete_by_row_id([0, 2])
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages)
+ assert commit._native_commit is not None
+ finally:
+ commit.close()
+ for result in _read(table, True):
+ rows = sorted(result.to_pylist(), key=lambda row: row['id'])
+ assert [row['id'] for row in rows] == [1, 3, 4, 5]
+ assert all(VideoFrameDescriptor.deserialize(row[name]).frame_index in
(3, 7, 8, 10)
+ for row in rows for name in ('left', 'right'))
diff --git a/paimon-python/pypaimon/tests/py36/ao_simple_test.py
b/paimon-python/pypaimon/tests/py36/ao_simple_test.py
index 662d0b720e..db81161ebb 100644
--- a/paimon-python/pypaimon/tests/py36/ao_simple_test.py
+++ b/paimon-python/pypaimon/tests/py36/ao_simple_test.py
@@ -440,7 +440,7 @@ class AOSimpleTest(RESTBaseTest):
}
pa_table = pa.Table.from_pydict(data1, schema=self.pa_schema)
table_write.write_arrow(pa_table)
- table_write.prepare_commit(0)
+ cm = table_write.prepare_commit(0)
# write 2
data2 = {
'user_id': [5, 6, 7, 8],
@@ -450,7 +450,7 @@ class AOSimpleTest(RESTBaseTest):
}
pa_table = pa.Table.from_pydict(data2, schema=self.pa_schema)
table_write.write_arrow(pa_table)
- table_write.prepare_commit(1)
+ cm.extend(table_write.prepare_commit(1))
# write 3
data3 = {
'user_id': [9, 10],
@@ -460,7 +460,7 @@ class AOSimpleTest(RESTBaseTest):
}
pa_table = pa.Table.from_pydict(data3, schema=self.pa_schema)
table_write.write_arrow(pa_table)
- cm = table_write.prepare_commit(2)
+ cm.extend(table_write.prepare_commit(2))
# commit
table_commit.commit(cm, 2)
table_write.close()
diff --git a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
index cea154d5e4..84fdbd0d89 100644
--- a/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
+++ b/paimon-python/pypaimon/tests/ray_data_evolution_merge_into_test.py
@@ -3232,7 +3232,6 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
def test_self_merge_filters_file_group_in_batches(self):
from pypaimon.ray import data_evolution_merge_into as merge_module
from pypaimon.ray import data_evolution_merge_join as join_module
- from pypaimon.write.file_store_commit import _abort_commit_messages
options = dict(self.de_options)
options['read.batch-size'] = '2'
@@ -3316,9 +3315,10 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
self.assertEqual(count, 3)
self.assertEqual(row_ids, [1, 3, 5])
finally:
- _abort_commit_messages(table, messages)
+ # Do not delete staged messages: another commit attempt may use
them.
+ pass
- def test_self_merge_update_aborts_other_groups_after_failure(self):
+ def test_self_merge_update_preserves_other_groups_after_failure(self):
from pypaimon.ray import data_evolution_merge_into as merge_module
from pypaimon.ray.data_evolution_merge_join import (
distributed_self_merge_update_apply,
@@ -3371,7 +3371,7 @@ class RayDataEvolutionMergeIntoTest(unittest.TestCase):
after = set()
for root, _, files in os.walk(self.warehouse):
after.update(os.path.join(root, name) for name in files)
- self.assertEqual(before, after)
+ self.assertLess(before, after)
@unittest.skipIf(_SKIP_CONDITION, _SKIP_REASON)
def test_self_merge_callable_assignment(self):
diff --git a/paimon-python/pypaimon/tests/ray_row_id_conflict_rewriter_test.py
b/paimon-python/pypaimon/tests/ray_row_id_conflict_rewriter_test.py
index 2273a9c97a..acc012f5a4 100644
--- a/paimon-python/pypaimon/tests/ray_row_id_conflict_rewriter_test.py
+++ b/paimon-python/pypaimon/tests/ray_row_id_conflict_rewriter_test.py
@@ -223,7 +223,7 @@ class RayRowIdConflictRewriterTest(unittest.TestCase):
), patch(
'pypaimon.ray.row_id_conflict_rewriter._retry_wait',
), patch(
- 'pypaimon.write.file_store_commit._abort_commit_messages',
+ 'pypaimon.write.file_store_commit.FileStoreCommit.abort',
) as abort_messages, self.assertRaises(RuntimeError) as context:
commit_self_merge_with_compaction_retry(
table,
@@ -260,7 +260,7 @@ class RayRowIdConflictRewriterTest(unittest.TestCase):
), patch(
'pypaimon.ray.row_id_conflict_rewriter._retry_wait',
), patch(
- 'pypaimon.write.file_store_commit._abort_commit_messages',
+ 'pypaimon.write.file_store_commit.FileStoreCommit.abort',
) as abort_messages, self.assertRaises(
RowIdExistenceConflict,
) as context:
@@ -295,7 +295,7 @@ class RayRowIdConflictRewriterTest(unittest.TestCase):
), patch(
'pypaimon.ray.row_id_conflict_rewriter._retry_wait',
), patch(
- 'pypaimon.write.file_store_commit._abort_commit_messages',
+ 'pypaimon.write.file_store_commit.FileStoreCommit.abort',
) as abort_messages, self.assertRaises(RuntimeError) as context:
commit_self_merge_with_compaction_retry(
table,
diff --git a/paimon-python/pypaimon/tests/ray_sink_test.py
b/paimon-python/pypaimon/tests/ray_sink_test.py
index 6153b37014..3870ab336f 100644
--- a/paimon-python/pypaimon/tests/ray_sink_test.py
+++ b/paimon-python/pypaimon/tests/ray_sink_test.py
@@ -363,7 +363,7 @@ class RaySinkTest(unittest.TestCase):
self.assertEqual({2}, {message.total_buckets for message in messages})
@pytest.mark.python_write
- def
test_write_does_not_return_prepared_messages_when_dedicated_close_aborts(self):
+ def test_write_failure_preserves_prepared_dedicated_files(self):
from pypaimon.write.writer.dedicated_format_writer import
DedicatedFormatWriter
pa_schema = pa.schema([
@@ -400,7 +400,9 @@ class RaySinkTest(unittest.TestCase):
datasink.write([data_table], ctx)
self.assertEqual(close_current_calls['count'], 2)
- self.assertEqual([], self._data_files_under(table))
+ self.assertEqual({'.parquet', '.blob'},
+ {os.path.splitext(path)[1] for path in
self._data_files_under(table)})
+ self.assertIsNone(table.snapshot_manager().get_latest_snapshot())
def test_on_write_complete(self):
from ray.data.datasource.datasink import WriteResult
diff --git a/paimon-python/pypaimon/tests/table_commit_test.py
b/paimon-python/pypaimon/tests/table_commit_test.py
index 82cd49291c..d606a94331 100644
--- a/paimon-python/pypaimon/tests/table_commit_test.py
+++ b/paimon-python/pypaimon/tests/table_commit_test.py
@@ -28,6 +28,71 @@ from pypaimon.write.table_commit import BatchTableCommit,
StreamTableCommit
class TestTableCommit(unittest.TestCase):
+ @parameterized.expand([
+ (mode, stream, published)
+ for mode in ('append', 'fixed', 'dynamic')
+ for stream in (False, True)
+ for published in (False, True)
+ ])
+ def test_writer_abort_preserves_files_after_commit_exception(self, mode,
stream, published):
+ import pyarrow as pa
+ from unittest.mock import patch
+ from pypaimon import CatalogFactory, Schema
+
+ with TemporaryDirectory(prefix='paimon-uncertain-commit-') as
warehouse:
+ catalog = CatalogFactory.create({'warehouse': warehouse})
+ catalog.create_database('default', True)
+ options = {'write.native.enabled': 'false',
'commit.native.enabled': 'false',
+ 'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false'}
+ primary_keys = []
+ if mode != 'append':
+ primary_keys = ['id']
+ options.update({'bucket': '1' if mode == 'fixed' else '-1',
+ 'changelog-producer': 'input'})
+ data = pa.table({'id': pa.array([1, 2], pa.int64()), 'value':
['a', 'b']})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(
+ data.schema, options=options, primary_keys=primary_keys),
False)
+ table = catalog.get_table('default.t')
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.write_arrow(data)
+ messages = writer.prepare_commit(1) if stream else
writer.prepare_commit()
+ paths = [file.physical_path() for message in messages
+ for file in message.new_files +
message.changelog_files]
+ if mode != 'append':
+ self.assertTrue(any(message.changelog_files for message in
messages))
+ if mode == 'dynamic':
+ self.assertTrue(any(message.index_adds for message in
messages))
+ paths.extend(table.path_factory().bucket_index_path(
+ tuple(entry.partition.values), entry.bucket,
entry.index_file, table.file_io)
+ for message in messages for entry in
message.index_adds)
+ self.assertTrue(paths)
+ publish = commit.file_store_commit.commit
+
+ def fail_commit(*args, **kwargs):
+ if published:
+ publish(*args, **kwargs)
+ raise OSError('Commit outcome is unknown')
+
+ with patch.object(commit.file_store_commit, 'commit',
side_effect=fail_commit):
+ with self.assertRaisesRegex(OSError, 'Commit outcome is
unknown'):
+ commit.commit(messages, 1) if stream else
commit.commit(messages)
+ # An unknown commit outcome must never trigger explicit commit
abort.
+ writer.abort()
+ writer.close()
+ for path in paths:
+ self.assertTrue(table.file_io.exists(path), 'Deleted
prepared file: ' + path)
+ snapshot = table.snapshot_manager().get_latest_snapshot()
+ self.assertEqual(snapshot.id if snapshot else 0,
int(published))
+ if published:
+ read_builder = table.new_read_builder()
+ actual =
read_builder.new_read().to_arrow(read_builder.new_scan().plan().splits())
+ self.assertEqual(actual.sort_by('id').to_pylist(),
data.to_pylist())
+ finally:
+ writer.close()
+ commit.close()
+
def test_empty_append_snapshot_is_opt_in_and_can_be_tagged(self):
import pyarrow as pa
import pypaimon.multimodal as pmm
diff --git a/paimon-python/pypaimon/tests/table_update_test.py
b/paimon-python/pypaimon/tests/table_update_test.py
index 262b9726c7..6af386546b 100644
--- a/paimon-python/pypaimon/tests/table_update_test.py
+++ b/paimon-python/pypaimon/tests/table_update_test.py
@@ -370,7 +370,7 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
)
@pytest.mark.python_write
- def test_predicate_update_aborts_groups_after_later_failure(self):
+ def test_predicate_update_preserves_groups_after_later_failure(self):
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
table = self._create_seeded_table()
@@ -401,9 +401,10 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
)
self.assertEqual(2, calls)
- self.assertEqual(before_files, self._list_table_files(table))
+ self.assertLess(before_files, self._list_table_files(table))
- def test_callable_update_aborts_groups_after_later_failure(self):
+ @pytest.mark.python_write
+ def test_callable_update_preserves_groups_after_later_failure(self):
table = self._create_seeded_table()
before_files = self._list_table_files(table)
calls = 0
@@ -424,7 +425,7 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
)
self.assertEqual(2, calls)
- self.assertEqual(before_files, self._list_table_files(table))
+ self.assertLess(before_files, self._list_table_files(table))
def test_array_assignment_spans_file_groups(self):
table = self._create_seeded_table()
@@ -1617,6 +1618,60 @@ class _TableUpdateTestBase(DataEvolutionTestBase):
[[('a', None)], [('k2', 'dict_val')], [('b', '2')]],
result4['meta'].to_pylist())
+ @pytest.mark.python_write
+ def
test_update_by_row_id_preserves_files_after_prepare_commit_failure(self):
+ table_schema = pa.schema([
+ ('id', pa.int32()),
+ ('age', pa.int32()),
+ ('picture', pa.large_binary()),
+ ])
+ table = self._create_table(pa_schema=table_schema)
+ self._write_arrow(table, pa.Table.from_pydict({
+ 'id': [1, 2],
+ 'age': [10, 20],
+ 'picture': [b'blob-1', b'blob-2'],
+ }, schema=table_schema))
+
+ rb = table.new_read_builder().with_projection(['id', '_ROW_ID'])
+ row_ids = rb.new_read().to_arrow(
+ rb.new_scan().plan().splits()).sort_by('id')['_ROW_ID']
+ before_files = self._list_table_files(table)
+
+ prepared_files = []
+
+ def fail_after_prepare_commit(
+ new_files, first_row_id, column_names, blob_columns):
+ prepared_files.extend(new_files)
+ raise RuntimeError("forced failure after prepare_commit")
+
+ for columns in (['age'], ['picture'], ['age', 'picture']):
+ with self.subTest(columns=columns):
+ wb = self._make_write_builder(table)
+ tu = wb.new_update().with_update_type(columns)
+ with mock.patch.object(
+ TableUpdateByRowId,
+ '_assign_update_file_metadata',
+ new=staticmethod(fail_after_prepare_commit)):
+ with self.assertRaisesRegex(
+ RuntimeError, "forced failure after
prepare_commit"):
+ self._apply_update(tu, pa.Table.from_pydict({
+ '_ROW_ID': [row_ids[0].as_py()],
+ 'age': [99],
+ 'picture': [b'updated-blob'],
+ }, schema=pa.schema([
+ ('_ROW_ID', pa.int64()),
+ ('age', pa.int32()),
+ ('picture', pa.large_binary()),
+ ])).select(['_ROW_ID'] + columns),
self._next_commit_id())
+
+ self.assertLess(before_files, self._list_table_files(table))
+ self.assertTrue(prepared_files)
+ self.assertTrue(all(table.file_io.exists(file.physical_path())
+ for file in prepared_files))
+ self.assertEqual([10, 20],
self._read_all(table)['age'].to_pylist())
+ self.assertEqual(
+ [b'blob-1', b'blob-2'],
self._read_all(table)['picture'].to_pylist())
+
# ======================================================================
# Mode-specific mixins (add the ``update_by_arrow_with_row_id`` primitive)
@@ -1668,7 +1723,8 @@ 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):
+ @pytest.mark.python_write
+ def test_update_batches_reject_same_file_and_preserve_staged_files(self):
table = self._create_seeded_table()
before_files = self._list_table_files(table)
update = (
@@ -1684,7 +1740,7 @@ class TableUpdateBatchTest(_BatchModeMixin,
_TableUpdateTestBase, unittest.TestC
pa.Table.from_pydict({'_ROW_ID': [1], 'age': [31]}),
]))
- self.assertEqual(before_files, self._list_table_files(table))
+ self.assertLess(before_files, self._list_table_files(table))
self.assertEqual(
[25, 30, 35, 40, 45],
self._read_all(table)['age'].to_pylist(),
@@ -1807,50 +1863,6 @@ class TableUpdateBatchTest(_BatchModeMixin,
_TableUpdateTestBase, unittest.TestC
result = self._read_all(table).sort_by('id')
self.assertEqual([100, 30, 35, 40, 45], result['age'].to_pylist())
- def test_update_by_row_id_aborts_files_after_prepare_commit_failure(self):
- from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
-
- table_schema = pa.schema([
- ('id', pa.int32()),
- ('age', pa.int32()),
- ('picture', pa.large_binary()),
- ])
- table = self._create_table(pa_schema=table_schema)
- self._write_arrow(table, pa.Table.from_pydict({
- 'id': [1, 2],
- 'age': [10, 20],
- 'picture': [b'blob-1', b'blob-2'],
- }, schema=table_schema))
-
- rb = table.new_read_builder().with_projection(['id', '_ROW_ID'])
- row_ids = rb.new_read().to_arrow(
- rb.new_scan().plan().splits()).sort_by('id')['_ROW_ID']
- before_files = self._list_table_files(table)
-
- def fail_after_prepare_commit(
- new_files, first_row_id, column_names, blob_columns):
- raise RuntimeError("forced failure after prepare_commit")
-
- wb = self._make_write_builder(table)
- tu = wb.new_update().with_update_type(['age', 'picture'])
- with mock.patch.object(
- TableUpdateByRowId,
- '_assign_update_file_metadata',
- new=staticmethod(fail_after_prepare_commit)):
- with self.assertRaisesRegex(
- RuntimeError, "forced failure after prepare_commit"):
- self._apply_update(tu, pa.Table.from_pydict({
- '_ROW_ID': [row_ids[0].as_py()],
- 'age': [99],
- 'picture': [b'updated-blob'],
- }, schema=pa.schema([
- ('_ROW_ID', pa.int64()),
- ('age', pa.int32()),
- ('picture', pa.large_binary()),
- ])), self._next_commit_id())
-
- self.assertEqual(before_files, self._list_table_files(table))
-
class TableUpdateStreamTest(_StreamModeMixin, _TableUpdateTestBase,
unittest.TestCase):
"""All shared update tests under stream (``StreamWriteBuilder``) semantics,
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 8833b1fc88..63ebe839e1 100644
--- a/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
+++ b/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
@@ -646,6 +646,75 @@ class _TableUpsertByKeyTestBase(DataEvolutionTestBase):
self.assertEqual(('Dave', 40, 'Houston'), rows[4])
self.assertTrue(all(os.path.exists(path) for path in stale_paths))
+ @pytest.mark.python_write
+ @pytest.mark.python_commit
+ def
test_failed_commit_rewrite_preserves_prepared_groups_and_can_retry(self):
+ from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
+
+ for fail_after_prepare in (False, True):
+ with self.subTest(fail_after_prepare=fail_after_prepare):
+ table = self._create_table()
+ original_data = pa.Table.from_pydict({
+ 'id': [1, 2, 3, 4],
+ 'name': ['Alice', 'Bob', 'Carol', 'Dave'],
+ 'age': [25, 30, 35, 40],
+ 'city': ['NYC', 'LA', 'Chicago', 'Houston'],
+ }, schema=self.pa_schema)
+ self._write_arrow(table, original_data.slice(0, 2))
+ self._write_arrow(table, original_data.slice(2, 2))
+ wb = self._make_write_builder(table)
+ cid = self._next_commit_id()
+ messages = []
+ for columns, data in [
+ (['age'], {'id': [2], 'name': ['ignored'],
+ 'age': [31], 'city': ['ignored']}),
+ (['city'], {'id': [3], 'name': ['ignored'],
+ 'age': [0], 'city': ['Chicago2']}),
+ ]:
+ update = wb.new_update().with_update_type(columns)
+ messages.extend(self._apply_upsert(
+ update, pa.Table.from_pydict(data,
schema=self.pa_schema), ['id'], cid))
+ staged_paths = [file.file_path for message in messages for
file in message.new_files]
+ self._compact_all_data_files(table)
+ snapshot_id = table.snapshot_manager().get_latest_snapshot().id
+ prepared_paths = []
+ calls = []
+ original_update = TableUpdateByRowId.update_columns
+
+ def fail_second_group(updater, data, columns):
+ calls.append(columns)
+ if len(calls) == 2 and not fail_after_prepare:
+ raise RuntimeError('second rewrite group failed')
+ result = original_update(updater, data, columns)
+ prepared_paths.extend(file.file_path for message in result
for file in message.new_files)
+ if len(calls) == 2:
+ raise RuntimeError('second rewrite group failed')
+ return result
+
+ commit = wb.new_commit()
+ try:
+ with mock.patch.object(TableUpdateByRowId,
'update_columns', fail_second_group):
+ with self.assertRaisesRegex(RuntimeError, 'second
rewrite group failed'):
+ self._apply_commit(commit, messages, cid)
+ # Preserve messages for retry; do not explicitly discard
them after a failed commit.
+ finally:
+ commit.close()
+ self.assertEqual(2, len(calls))
+ self.assertEqual(2 if fail_after_prepare else 1,
len(prepared_paths))
+ self.assertTrue(all(os.path.exists(path) for path in
staged_paths + prepared_paths))
+ self.assertEqual(snapshot_id,
table.snapshot_manager().get_latest_snapshot().id)
+
self.assertTrue(original_data.equals(self._read_all(table).sort_by('id')))
+
+ commit = wb.new_commit()
+ try:
+ self._apply_commit(commit, messages, cid)
+ finally:
+ commit.close()
+ rows = {row['id']: row for row in
self._read_all(table).to_pylist()}
+ self.assertEqual(31, rows[2]['age'])
+ self.assertEqual('Chicago2', rows[3]['city'])
+ self.assertTrue(all(os.path.exists(path) for path in
staged_paths + prepared_paths))
+
def test_commit_rewrite_uses_checked_base_entries(self):
table = self._create_table()
self._write_arrow(table, pa.Table.from_pydict({
diff --git a/paimon-python/pypaimon/tests/write/changelog_producer_test.py
b/paimon-python/pypaimon/tests/write/changelog_producer_test.py
index 0ac4610ee8..f59b32d803 100644
--- a/paimon-python/pypaimon/tests/write/changelog_producer_test.py
+++ b/paimon-python/pypaimon/tests/write/changelog_producer_test.py
@@ -310,7 +310,7 @@ class ChangelogProducerTest(unittest.TestCase):
self.assertIsNotNone(snapshot.changelog_manifest_list)
self.assertEqual(snapshot.changelog_record_count, 2)
- def test_abort_cleans_up_changelog_files(self):
+ def test_explicit_abort_deletes_prepared_data_and_changelog_files(self):
table = self._create_table(
'test_input_abort',
options={'changelog-producer': 'input', 'bucket': '1'}
@@ -331,8 +331,8 @@ class ChangelogProducerTest(unittest.TestCase):
data_files_after = glob.glob(os.path.join(bucket_dir, 'data-*'))
changelog_files_after = glob.glob(os.path.join(bucket_dir,
'changelog-*'))
- self.assertEqual(len(data_files_after), 0, "Data files should be
cleaned up after abort")
- self.assertEqual(len(changelog_files_after), 0, "Changelog files
should be cleaned up after abort")
+ self.assertFalse(data_files_after)
+ self.assertFalse(changelog_files_after)
table_write.close()
table_commit.close()
diff --git a/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
b/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
index 1ce9541e8c..218532bbbf 100644
--- a/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
+++ b/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
@@ -402,7 +402,7 @@ class DynamicBucketTest(unittest.TestCase):
):
table.new_batch_write_builder().new_write()
- def test_batch_writer_abort_after_prepare_deletes_hash_index(self):
+ def test_batch_writer_abort_after_prepare_preserves_hash_index(self):
with tempfile.TemporaryDirectory() as root:
table = self._create_table(root, 'abort_prepared')
writer = table.new_batch_write_builder().new_write()
@@ -418,7 +418,52 @@ class DynamicBucketTest(unittest.TestCase):
writer.abort()
- self.assertFalse(table.file_io.exists(index_path))
+ self.assertTrue(table.file_io.exists(index_path))
+
+ @pytest.mark.python_write
+ def test_failed_hash_index_prepare_retries_the_complete_increment(self):
+ for failed_call in (1, 2):
+ with self.subTest(failed_call=failed_call),
tempfile.TemporaryDirectory() as root:
+ table = self._create_table(root, 'retry_prepare',
target_row_num=1).copy({
+ 'write.native.enabled': 'false', 'commit.native.enabled':
'false',
+ 'read.native.enabled': 'false',
'scan.native-plan.enabled': 'false',
+ 'changelog-producer': 'input'})
+ builder = table.new_stream_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ data = pa.table({'id': [1, 2, 3], 'value': ['v-1', 'v-2',
'v-3']})
+ try:
+ writer.write_arrow(data)
+ maintainer = writer.row_key_extractor._index_maintainer
+ bucket_count = len(maintainer._states)
+ self.assertGreaterEqual(bucket_count, 2)
+ write_index = maintainer._write_index
+ calls = 0
+
+ def fail_index(*args):
+ nonlocal calls
+ calls += 1
+ if calls == failed_call:
+ raise OSError('HASH index prepare failed')
+ return write_index(*args)
+
+ with patch.object(maintainer, '_write_index',
side_effect=fail_index):
+ with self.assertRaisesRegex(OSError, 'HASH index
prepare failed'):
+ writer.prepare_commit(1)
+
self.assertIsNone(table.snapshot_manager().get_latest_snapshot())
+ messages = writer.prepare_commit(1)
+ self.assertEqual(3, sum(file.row_count for message in
messages
+ for file in message.new_files))
+ self.assertEqual(3, sum(file.row_count for message in
messages
+ for file in
message.changelog_files))
+ self.assertEqual(bucket_count, sum(len(message.index_adds)
for message in messages))
+ self.assertEqual([], writer.prepare_commit(2))
+ commit.commit(messages, 1)
+ writer.abort()
+ self.assertEqual(data.to_pydict(),
self._read_arrow(table).sort_by('id').to_pydict())
+ self.assertEqual(bucket_count,
len(self._hash_indexes(table)))
+ finally:
+ writer.close()
+ commit.close()
@pytest.mark.python_write
def test_stream_writer_releases_prepared_hash_index_ownership(self):
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index 4d751ce59e..5b48d2ef3b 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -614,7 +614,7 @@ class TableWriteTest(unittest.TestCase):
}
pa_table = pa.Table.from_pydict(data1, schema=self.pa_schema)
table_write.write_arrow(pa_table)
- table_write.prepare_commit(0)
+ cm = table_write.prepare_commit(0)
# write 2
data2 = {
'user_id': [5, 6, 7, 8],
@@ -624,7 +624,7 @@ class TableWriteTest(unittest.TestCase):
}
pa_table = pa.Table.from_pydict(data2, schema=self.pa_schema)
table_write.write_arrow(pa_table)
- table_write.prepare_commit(1)
+ cm.extend(table_write.prepare_commit(1))
# write 3
data3 = {
'user_id': [9, 10],
@@ -634,7 +634,7 @@ class TableWriteTest(unittest.TestCase):
}
pa_table = pa.Table.from_pydict(data3, schema=self.pa_schema)
table_write.write_arrow(pa_table)
- cm = table_write.prepare_commit(2)
+ cm.extend(table_write.prepare_commit(2))
# commit
table_commit.commit(cm, 2)
table_write.close()
@@ -740,7 +740,7 @@ class TableWriteTest(unittest.TestCase):
}
pa_table = pa.Table.from_pydict(data1, schema=self.pk_pa_schema)
table_write.write_arrow(pa_table)
- table_write.prepare_commit(0)
+ cm = table_write.prepare_commit(0)
# write 2
data2 = {
'user_id': [5, 6, 7, 8],
@@ -750,7 +750,7 @@ class TableWriteTest(unittest.TestCase):
}
pa_table = pa.Table.from_pydict(data2, schema=self.pk_pa_schema)
table_write.write_arrow(pa_table)
- table_write.prepare_commit(1)
+ cm.extend(table_write.prepare_commit(1))
# write 3
data3 = {
'user_id': [9, 10],
@@ -760,7 +760,7 @@ class TableWriteTest(unittest.TestCase):
}
pa_table = pa.Table.from_pydict(data3, schema=self.pk_pa_schema)
table_write.write_arrow(pa_table)
- cm = table_write.prepare_commit(2)
+ cm.extend(table_write.prepare_commit(2))
# commit
table_commit.commit(cm, 2)
table_write.close()
diff --git a/paimon-python/pypaimon/tests/write/write_buffer_test.py
b/paimon-python/pypaimon/tests/write/write_buffer_test.py
index 942798b75a..566f163bdc 100644
--- a/paimon-python/pypaimon/tests/write/write_buffer_test.py
+++ b/paimon-python/pypaimon/tests/write/write_buffer_test.py
@@ -513,6 +513,7 @@ class _StubSidecarWriter:
def __init__(self, row_count: int, file_name: str, fail_times: int = 0,
delete_on_abort: bool = True):
self.committed_files = []
+ self.committed_changelog_files = []
self.pending_row_count = 0
self.prepare_commit_calls = 0
self.aborted = False
diff --git a/paimon-python/pypaimon/write/commit/row_id_conflict_rewriter.py
b/paimon-python/pypaimon/write/commit/row_id_conflict_rewriter.py
index 3fdaf916c6..fa0efbb8d2 100644
--- a/paimon-python/pypaimon/write/commit/row_id_conflict_rewriter.py
+++ b/paimon-python/pypaimon/write/commit/row_id_conflict_rewriter.py
@@ -136,10 +136,12 @@ class RowIdConflictRewriter:
updater.update_columns(update_data, column_names)
)
except Exception:
- self._abort(updater.commit_messages)
+ # Never delete files from CommitMessage on failure.
+ # A commit can succeed even when its response raises an
exception.
raise
except Exception:
- self._abort(new_messages)
+ # Preserve prepared outputs from earlier groups as well.
+ # Never infer deletion ownership from CommitMessage.
raise
rewritten_entries = [
@@ -271,13 +273,6 @@ class RowIdConflictRewriter:
))
return entries
- def _abort(self, messages):
- for message in messages:
- for file in message.new_files:
- path = file.physical_path()
- if path:
- self.table.file_io.delete_quietly(path)
-
@staticmethod
def _range_key(entry):
return (
diff --git a/paimon-python/pypaimon/write/commit_message.py
b/paimon-python/pypaimon/write/commit_message.py
index 0a25ba5cf4..1325260cd2 100644
--- a/paimon-python/pypaimon/write/commit_message.py
+++ b/paimon-python/pypaimon/write/commit_message.py
@@ -41,7 +41,6 @@ class CommitMessage:
compact_changelog_files: List[DataFileMeta] = field(default_factory=list)
compact_index_adds: List['IndexManifestEntry'] =
field(default_factory=list)
compact_index_deletes: List['IndexManifestEntry'] =
field(default_factory=list)
- _native_write_pending: bool = field(default=False, init=False, repr=False,
compare=False)
def is_empty(self):
return (
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 5259be2461..03e718e707 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -85,48 +85,6 @@ def _reject_compact_increment(messages: List[CommitMessage]):
'Committing a compact increment requires a separate COMPACT
snapshot.')
-def _abort_commit_messages(table, commit_messages: List[CommitMessage]):
- """Delete files created by messages known to be uncommitted."""
- for message in commit_messages:
- for file in (list(message.new_files) + list(message.changelog_files)
- + list(message.compact_after)
- + list(message.compact_changelog_files)):
- path = None
- try:
- bucket_path = None if file.physical_path() else
table.path_factory().bucket_path(
- tuple(message.partition), message.bucket)
- for path in file.collect_files(bucket_path):
- table.file_io.delete_quietly(path)
- except Exception as error:
- logger.warning(
- "Failed to clean up file %s during abort: %s",
- path,
- error,
- )
- for entry in message.index_adds + message.compact_index_adds:
- file_name = None
- try:
- index_file = entry.index_file
- file_name = index_file.file_name
- if index_file.index_type in ('DELETION_VECTORS', 'HASH'):
- path = table.path_factory().bucket_index_path(
- tuple(entry.partition.values), entry.bucket,
index_file, table.file_io)
- else:
- path = (
- index_file.external_path
- or table.path_factory()
- .global_index_path_factory()
- .to_path(file_name)
- )
- table.file_io.delete_quietly(path)
- except Exception as error:
- logger.warning(
- "Failed to clean up index file %s during abort: %s",
- file_name,
- error,
- )
-
-
class CommitResult:
"""Base class for commit results."""
@@ -1062,8 +1020,51 @@ class FileStoreCommit:
self.table.file_io.delete_quietly(f"{manifest_path}/{index_manifest}")
def abort(self, commit_messages: List[CommitMessage]):
- """Abort commit and delete files. Uses external_path if available to
ensure proper scheme handling."""
- _abort_commit_messages(self.table, commit_messages)
+ """Delete files for an explicitly abandoned, known-uncommitted write.
+
+ Mirrors Java FileStoreCommitImpl.abort: delete new data, changelog,
+ compaction outputs and added indexes, preserving deleted inputs.
+ Never call after a commit whose outcome is unknown. Internal error
+ paths must preserve prepared files instead of calling this method.
+ """
+ for message in commit_messages:
+ for file in (list(message.new_files) +
list(message.changelog_files)
+ + list(message.compact_after)
+ + list(message.compact_changelog_files)):
+ path = None
+ try:
+ bucket_path = None if file.physical_path() else
self.table.path_factory().bucket_path(
+ tuple(message.partition), message.bucket)
+ for path in file.collect_files(bucket_path):
+ self.table.file_io.delete_quietly(path)
+ except Exception as error:
+ logger.warning(
+ "Failed to clean up file %s during abort: %s",
+ path,
+ error,
+ )
+ for entry in message.index_adds + message.compact_index_adds:
+ file_name = None
+ try:
+ index_file = entry.index_file
+ file_name = index_file.file_name
+ if index_file.index_type in ('DELETION_VECTORS', 'HASH'):
+ path = self.table.path_factory().bucket_index_path(
+ tuple(entry.partition.values), entry.bucket,
index_file, self.table.file_io)
+ else:
+ path = (
+ index_file.external_path
+ or self.table.path_factory()
+ .global_index_path_factory()
+ .to_path(file_name)
+ )
+ self.table.file_io.delete_quietly(path)
+ except Exception as error:
+ logger.warning(
+ "Failed to clean up index file %s during abort: %s",
+ file_name,
+ error,
+ )
def close(self):
"""Close the FileStoreCommit and release resources."""
diff --git a/paimon-python/pypaimon/write/file_store_write.py
b/paimon-python/pypaimon/write/file_store_write.py
index 02ddd0898d..599e301eeb 100644
--- a/paimon-python/pypaimon/write/file_store_write.py
+++ b/paimon-python/pypaimon/write/file_store_write.py
@@ -322,6 +322,12 @@ class FileStoreWrite:
return any(isinstance(f.type, VectorType) for f in
self.table.table_schema.fields)
def prepare_commit(self, commit_identifier) -> List[CommitMessage]:
+ messages = self._prepare_commit_messages(commit_identifier)
+ self._release_prepared_files()
+ return messages
+
+ def _prepare_commit_messages(self, commit_identifier) ->
List[CommitMessage]:
+ """Collect files while their parent is still preparing the complete
increment."""
self.commit_identifier = commit_identifier
commit_messages = []
for (partition, bucket), writer in self.data_writers.items():
@@ -336,15 +342,13 @@ class FileStoreWrite:
total_buckets=self._runtime_total_buckets.get(partition),
)
commit_messages.append(commit_message)
- self._release_prepared_postpone_files()
return commit_messages
- def _release_prepared_postpone_files(self):
+ def _release_prepared_files(self):
# Hand off only after every partition prepared successfully. Until
# then, close/abort must still clean up files from earlier partitions.
for writer in self.data_writers.values():
- if isinstance(writer, PostponeDataWriter):
- writer._release_prepared_files()
+ writer._release_prepared_files()
def close(self):
"""Close all data writers and clean up resources."""
@@ -409,8 +413,8 @@ class PostponeFixedBucketFileStoreWrite(FileStoreWrite):
return {}
return super()._load_seq_number_stats(partition)
- def prepare_commit(self, commit_identifier):
- messages = super().prepare_commit(commit_identifier)
+ def _prepare_commit_messages(self, commit_identifier):
+ messages = super()._prepare_commit_messages(commit_identifier)
for message in messages:
message.check_from_snapshot = self._check_from_snapshot
return messages
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index 1aee337e65..f291369bc4 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -42,7 +42,6 @@ def _native_row_id_table(table):
or table.options.file_format() != 'parquet'
or table.options.data_evolution_row_sidecar_enabled(False)
or table.options.with_vector_format()
- or table.options.video_frame_fields()
or table.options.changelog_producer() != ChangelogProducer.NONE
or any(table.options.options.contains_key(key) for key in
SCAN_KEYS)
or not native_write_available()):
@@ -139,7 +138,9 @@ def _native_update_columns_supported(table, columns):
def create_native_upsert(table, commit_user, data, keys, columns):
"""Prepare one core upsert from full Arrow rows or named row values."""
- if not _native_update_columns_supported(table, columns):
+ if table.options.video_frame_fields() or not
_native_update_columns_supported(table, columns):
+ # An upsert can append unmatched rows; packed video writing is still
+ # provided by the Python writer.
return None
native_table = _native_row_id_table(table)
if native_table is None:
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index a69cd3e1ed..81943baefd 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -22,9 +22,8 @@ import pyarrow as pa
from pypaimon.common.options.core_options import MergeEngine
from pypaimon.schema.arrow_schema import arrow_schemas_compatible,
normalize_arrow_strings
-from pypaimon.schema.data_types import PyarrowFieldParser, is_blob_file_field,
is_blob_type
+from pypaimon.schema.data_types import PyarrowFieldParser, is_blob_file_field
from pypaimon.table.bucket_mode import BucketMode
-from pypaimon.write.file_store_commit import _abort_commit_messages
from pypaimon.write.native_commit import (
create_native_write_table, from_native_commit_messages,
)
@@ -72,11 +71,7 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False,
or table.options.file_format() != 'parquet'
# Rust cannot encode these partition keys yet.
or not _native_partition_types_supported(schema,
table.partition_keys)
- # Append dedicated files currently support top-level scalar Blob
fields.
- or table.options.video_frame_fields()
- or (not table.is_primary_key_table
- and any(is_blob_file_field(field) and not
is_blob_type(field.type)
- for field in table.table_schema.fields))):
+ or table.options.video_frame_fields()):
return None
native_table = create_native_write_table(table)
if native_table is None:
@@ -115,7 +110,6 @@ class NativeTableWrite:
self._native_writer = native_writer
self._python_writer = None
self._written = False
- self._prepared_messages = []
self._schema =
PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
def _switch_to_python(self):
@@ -196,13 +190,7 @@ class NativeTableWrite:
if commit_identifier is not None:
raise TypeError('BatchTableWrite.prepare_commit accepts no
identifier')
messages = self._native_writer.prepare_commit()
- messages = from_native_commit_messages(self.table, messages)
- self._prepared_messages = [message for message in
self._prepared_messages
- if message._native_write_pending]
- for message in messages:
- message._native_write_pending = True
- self._prepared_messages.extend(messages)
- return messages
+ return from_native_commit_messages(self.table, messages)
def close(self):
if self._python_writer is not None:
@@ -210,18 +198,15 @@ class NativeTableWrite:
elif self._native_writer is not None:
self._native_writer.close()
self._native_writer = None
- self._prepared_messages.clear()
def abort(self):
if self._python_writer is not None:
self._python_writer.abort()
else:
- messages = [message for message in self._prepared_messages
- if message._native_write_pending]
- try:
- self.close()
- finally:
- _abort_commit_messages(self.table, messages)
+ # Closing Rust cleans only files still owned by its writer.
+ # Never delete prepared CommitMessage files: publication can
+ # succeed even when its response raises an exception.
+ self.close()
class NativePostponeFixedBucketTableWrite(NativeTableWrite):
diff --git a/paimon-python/pypaimon/write/ray_datasink.py
b/paimon-python/pypaimon/write/ray_datasink.py
index 796f69947e..edfdcb90ba 100644
--- a/paimon-python/pypaimon/write/ray_datasink.py
+++ b/paimon-python/pypaimon/write/ray_datasink.py
@@ -192,6 +192,10 @@ class PaimonDatasink(_DatasinkBase):
f"Error aborting worker-side table_write:
{abort_error}",
exc_info=abort_error
)
+ # prepare_commit has handed these files off, but a failed close
+ # prevents this worker from returning them to the coordinator.
+ # Never delete files from CommitMessage on failure.
+ # A commit can succeed even when its response raises an exception.
raise
@staticmethod
diff --git a/paimon-python/pypaimon/write/table_commit.py
b/paimon-python/pypaimon/write/table_commit.py
index 486dcbcff7..894d616fba 100644
--- a/paimon-python/pypaimon/write/table_commit.py
+++ b/paimon-python/pypaimon/write/table_commit.py
@@ -66,9 +66,7 @@ class TableCommit:
commit_messages: List[CommitMessage],
commit_identifier: int = BATCH_COMMIT_IDENTIFIER,
snapshot_properties: Optional[Dict[str, str]] = None):
- """Release native-writer cleanup ownership before publication can
start."""
- for message in commit_messages:
- message._native_write_pending = False
+ """Publish prepared messages while preserving files on any failure."""
non_empty_messages = [msg for msg in commit_messages if not
msg.is_empty()]
commit_kwargs = {
"commit_messages": non_empty_messages,
@@ -139,6 +137,12 @@ class TableCommit:
return None
def abort(self, commit_messages: List[CommitMessage]):
+ """Delete explicitly abandoned, known-uncommitted files.
+
+ Never call after a commit whose outcome is unknown. Publication may
+ have succeeded before its response failed. Internal failure paths
+ must preserve prepared files instead of calling this method.
+ """
prepared = self._prepare_native_commit(commit_messages)
if prepared is not None:
native, messages = prepared
diff --git a/paimon-python/pypaimon/write/table_update.py
b/paimon-python/pypaimon/write/table_update.py
index a2177c2efe..c3f8159fe5 100644
--- a/paimon-python/pypaimon/write/table_update.py
+++ b/paimon-python/pypaimon/write/table_update.py
@@ -47,7 +47,6 @@ from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.snapshot.time_travel_util import SCAN_KEYS, TimeTravelUtil
from pypaimon.table.special_fields import SpecialFields
from pypaimon.write.commit_message import CommitMessage
-from pypaimon.write.file_store_commit import _abort_commit_messages
from pypaimon.write.row_id_file_index import RowIdFileIndex
from pypaimon.write.table_delete import TableDeleteByRowId
from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
@@ -216,8 +215,8 @@ class TableUpdate:
updater.update_columns(table, cols)
return [] if updater is None else updater.commit_messages
except Exception:
- if updater is not None:
- _abort_commit_messages(self.table, updater.commit_messages)
+ # Never delete files from CommitMessage on failure.
+ # A commit can succeed even when its response raises an exception.
raise
def _upsert_by_arrow_with_key(
@@ -356,7 +355,8 @@ class TableUpdate:
update_table, list(assignments.keys())
)
except Exception:
- _abort_commit_messages(self.table, updater.commit_messages)
+ # Never delete files from CommitMessage on failure.
+ # A commit can succeed even when its response raises an exception.
raise
return updater.commit_messages
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 32de0a2a98..103e4400de 100644
--- a/paimon-python/pypaimon/write/table_update_by_row_id.py
+++ b/paimon-python/pypaimon/write/table_update_by_row_id.py
@@ -948,6 +948,7 @@ class TableUpdateByRowId:
for value in values:
blob_writer.write_blob(value, arrow_type)
new_files.extend(blob_writer.prepare_commit())
+ blob_writer._release_prepared_files()
if new_files:
self._assign_update_file_metadata(
@@ -970,6 +971,8 @@ class TableUpdateByRowId:
else:
if file_store_write is not None:
file_store_write.abort()
+ # Never delete files from CommitMessage on failure.
+ # A commit can succeed even when its response raises an
exception.
for blob_writer in blob_writers:
blob_writer.abort()
diff --git a/paimon-python/pypaimon/write/table_write.py
b/paimon-python/pypaimon/write/table_write.py
index 21fdce5598..e982248dbd 100644
--- a/paimon-python/pypaimon/write/table_write.py
+++ b/paimon-python/pypaimon/write/table_write.py
@@ -316,12 +316,9 @@ class TableWrite:
abort()
def _prepare_commit(self, commit_identifier) -> List[CommitMessage]:
- commit_messages =
self.file_store_write.prepare_commit(commit_identifier)
+ commit_messages =
self.file_store_write._prepare_commit_messages(commit_identifier)
prepare_indexes = getattr(self.row_key_extractor, "prepare_commit",
None)
- if prepare_indexes is None:
- return commit_messages
-
- index_changes = prepare_indexes()
+ index_changes = prepare_indexes() if prepare_indexes is not None else
{}
messages_by_bucket = {
(tuple(message.partition), message.bucket): message
for message in commit_messages
@@ -338,6 +335,10 @@ class TableWrite:
messages_by_bucket[(partition, bucket)] = message
message.index_adds.extend(changes.additions)
message.index_deletes.extend(changes.deletions)
+ # Hand off data, changelog and HASH indexes together. Until every
+ # preparation succeeds, retain the metadata for retry and abort.
+ self.file_store_write._release_prepared_files()
+ self._release_prepared_indexes()
return commit_messages
def _release_prepared_indexes(self) -> None:
@@ -403,6 +404,4 @@ class StreamTableWrite(TableWrite):
super().__init__(table, commit_user, None)
def prepare_commit(self, commit_identifier) -> List[CommitMessage]:
- messages = self._prepare_commit(commit_identifier)
- self._release_prepared_indexes()
- return messages
+ return self._prepare_commit(commit_identifier)
diff --git a/paimon-python/pypaimon/write/writer/composite_data_writer.py
b/paimon-python/pypaimon/write/writer/composite_data_writer.py
index 9cf8abf29c..a9d5c66e6f 100644
--- a/paimon-python/pypaimon/write/writer/composite_data_writer.py
+++ b/paimon-python/pypaimon/write/writer/composite_data_writer.py
@@ -51,6 +51,17 @@ class CompositeDataWriter(DataWriter):
self._close_current_writers()
return self.committed_files.copy()
+ def _release_prepared_files(self) -> List[DataFileMeta]:
+ """Hand a successfully prepared group to its committer, as Java does.
+
+ Until every partition prepares, retain files for retry/abort. Once the
+ parent accepts the messages, later checkpoints must neither resend
+ those files nor delete them when another write fails.
+ """
+ owned_files = super()._release_prepared_files()
+ self._committed_files_to_delete_on_abort.clear()
+ return owned_files
+
def close(self):
if self.closed:
return
diff --git a/paimon-python/pypaimon/write/writer/data_writer.py
b/paimon-python/pypaimon/write/writer/data_writer.py
index a00287caf7..e8f178b974 100644
--- a/paimon-python/pypaimon/write/writer/data_writer.py
+++ b/paimon-python/pypaimon/write/writer/data_writer.py
@@ -157,8 +157,10 @@ class DataWriter(ABC):
Call only after all sidecars are prepared and validated. Until then,
this
writer retains its metadata for retry and its responsibility for
cleanup.
"""
- owned_files = self.committed_files.copy() if
self.delete_file_upon_abort() else []
+ owned_files = (self.committed_files + self.committed_changelog_files
+ if self.delete_file_upon_abort() else [])
self.committed_files.clear()
+ self.committed_changelog_files.clear()
return owned_files
def delete_file_upon_abort(self) -> bool:
