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 09d859879e [python] Preserve postpone events and enable native
postpone merge engines (#10368)
09d859879e is described below
commit 09d859879ee5ff4765a9bd1ae7fe6b16c1188a05
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 4 16:04:55 2026 +0800
[python] Preserve postpone events and enable native postpone merge engines
(#10368)
---
paimon-python/pypaimon/read/split_read.py | 10 +-
paimon-python/pypaimon/tests/arrow_schema_test.py | 36 ++++-
.../tests/native_managed_blob_write_test.py | 165 +++++++++++++++++++++
.../pypaimon/tests/native_postpone_write_test.py | 50 ++++++-
.../tests/write/postpone_event_write_test.py | 152 +++++++++++++++++++
.../pypaimon/tests/write/table_write_test.py | 3 +
paimon-python/pypaimon/write/file_store_write.py | 57 +++----
paimon-python/pypaimon/write/native_write.py | 16 +-
paimon-python/pypaimon/write/row_kind.py | 66 +++++++++
paimon-python/pypaimon/write/table_write.py | 11 +-
.../pypaimon/write/writer/key_value_data_writer.py | 4 +-
.../pypaimon/write/writer/postpone_data_writer.py | 109 ++++++++++++++
12 files changed, 630 insertions(+), 49 deletions(-)
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index 88883f61a2..d518ca5abe 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -99,6 +99,10 @@ def deferred_blob_field_names(table, read_fields:
List[DataField],
predicate: Optional[Predicate],
limit: Optional[int],
has_post_filter: bool = False) -> set:
+ # Only append tables have dedicated Blob files to defer. Primary-key
+ # managed packs are resolved inside Rust's read pipeline.
+ if table.is_primary_key_table:
+ return set()
# An auth filter also selects rows; defer past it too, like a
predicate/limit.
if ((predicate is None and limit is None and not has_post_filter)
or CoreOptions.blob_as_descriptor(table.options)):
@@ -1188,10 +1192,8 @@ class MergeFileSplitRead(SplitRead):
``AGGREGATE`` is special-cased here because building the per-
field aggregators needs the full ``DataField`` objects, the
full primary-key list and the parsed ``CoreOptions`` -- which
- sit outside the dispatch's raw-options contract. The writer-
- side merge buffer falls back to dedupe for aggregation anyway
- (see :meth:`FileStoreWrite._build_pk_merge_function`), so the
- two sides only need to share the simple engines.
+ sit outside the dispatch's raw-options contract. FileStoreWrite uses
+ the same aggregator builder for writer-side folding and validation.
"""
engine = self.table.options.merge_engine()
if engine == MergeEngine.AGGREGATE:
diff --git a/paimon-python/pypaimon/tests/arrow_schema_test.py
b/paimon-python/pypaimon/tests/arrow_schema_test.py
index 7a555f8a27..5abf03e75a 100644
--- a/paimon-python/pypaimon/tests/arrow_schema_test.py
+++ b/paimon-python/pypaimon/tests/arrow_schema_test.py
@@ -19,7 +19,7 @@ import glob
import os
import tempfile
import unittest
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
import pyarrow as pa
import pyarrow.parquet as pq
@@ -121,6 +121,40 @@ class ArrowSchemaTest(unittest.TestCase):
actual = result.to_pydict()
self.assertEqual(sorted(zip(actual['id'], actual['text'])), [(1,
'中文'), (2, None)])
+ @parameterized.expand([('table',), ('batch',)])
+ def test_schema_only_writer_normalizes_input(self, entry):
+ writer = object.__new__(TableWrite)
+ writer.file_store_write = Mock(write_cols=None)
+ writer.table_pyarrow_schema = pa.schema([('text', pa.string())])
+ source = pa.Table.from_pydict({'text': ['中文', None]},
schema=pa.schema([('text', pa.large_string())]))
+ if entry == 'batch':
+ source = source.to_batches()[0]
+
+ result = writer._prepare_arrow_data(source)
+
+ self.assertIsInstance(result, type(source))
+ self.assertEqual(result.schema, writer.table_pyarrow_schema)
+ self.assertEqual(result.to_pydict(), source.to_pydict())
+ writer.file_store_write.write.assert_not_called()
+
+ @parameterized.expand([('table',), ('batch',)])
+ def test_normalization_failure_fails_before_routing(self, entry):
+ writer = object.__new__(TableWrite)
+ writer.file_store_write = Mock(write_cols=None)
+ writer.row_key_extractor = Mock()
+ writer.table_pyarrow_schema = pa.schema([('text', pa.string())])
+ source = pa.Table.from_pydict({'text': ['中文']},
schema=pa.schema([('text', pa.large_string())]))
+ error = ValueError('Cannot convert large_string input to string')
+ with patch('pypaimon.write.table_write.normalize_arrow_strings',
side_effect=error):
+ with self.assertRaises(ValueError) as raised:
+ if entry == 'batch':
+ writer.write_arrow_batch(source.to_batches()[0])
+ else:
+ writer.write_arrow(source)
+ self.assertIs(raised.exception, error)
+
writer.row_key_extractor.extract_partition_bucket_groups.assert_not_called()
+ writer.file_store_write.write.assert_not_called()
+
@unittest.skipUnless(int(pa.__version__.split('.')[0]) == 6, 'Arrow 6
lacks struct cast kernels')
def test_unsupported_nested_cast_fails_before_routing(self):
writer = object.__new__(TableWrite)
diff --git a/paimon-python/pypaimon/tests/native_managed_blob_write_test.py
b/paimon-python/pypaimon/tests/native_managed_blob_write_test.py
new file mode 100644
index 0000000000..997182607e
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_managed_blob_write_test.py
@@ -0,0 +1,165 @@
+# 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.
+
+"""Exercise managed primary-key Blobs exclusively through Rust reads and
writes."""
+
+from unittest.mock import patch
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from pypaimon import CatalogFactory
+from pypaimon.schema.data_types import PyarrowFieldParser
+from pypaimon.table.row.blob import Blob, BlobDescriptor
+from pypaimon.write.native_write import NativeTableWrite
+
+pytestmark = pytest.mark.native_plan
+
+
+def _table(tmp_path, mode='fixed', extra=None):
+ # Table creation and schema validation belong to Rust. Python's catalog
+ # only loads the resulting Java-compatible schema for the native bridge.
+ from pypaimon_rust.datafusion import SQLContext
+ options = {'bucket': '1' if mode == 'fixed' else '-2', 'file.format':
'parquet',
+ 'write.native.enabled': 'true', 'read.native.enabled': 'true',
+ 'scan.native-plan.enabled': 'true', 'rowkind.field': 'op',
+ 'blob.target-file-size': '1 B', 'postpone.default-bucket-num':
'2'}
+ options.update(extra or {})
+ context = SQLContext()
+ context.register_catalog('paimon', {'warehouse': str(tmp_path)})
+ context.sql('CREATE SCHEMA paimon.managed')
+ properties = ', '.join("'%s' = '%s'" % (key, value) for key, value in
options.items())
+ context.sql('CREATE TABLE paimon.managed.t ('
+ 'id INT, payload BLOB, items ARRAY<BLOB>, mapping MAP(STRING,
BLOB), op STRING, '
+ 'PRIMARY KEY (id)) WITH (' + properties + ')')
+ return CatalogFactory.create({'warehouse':
str(tmp_path)}).get_table('managed.t')
+
+
+def _builder(table, mode):
+ return (table.new_postpone_fixed_bucket_write_builder() if mode ==
'postpone-fixed'
+ else table.new_batch_write_builder())
+
+
+def _input(table, values):
+ return pa.Table.from_pylist(values,
schema=PyarrowFieldParser.from_paimon_schema(table.fields))
+
+
+def _read(table, predicate=None, projection=None, limit=None,
descriptors=False):
+ table = table.copy({'blob-as-descriptor': str(descriptors).lower()})
+ builder = table.new_read_builder()
+ if predicate:
+ builder.with_filter(predicate)
+ if projection:
+ builder.with_projection(projection)
+ if limit is not None:
+ builder.with_limit(limit)
+ splits = builder.new_scan().plan().splits()
+ read = builder.new_read()
+ guard = patch.object(read, '_create_split_read',
side_effect=AssertionError('Python read fallback'))
+ with guard:
+ return read.to_arrow(splits, blob_parallelism=2).to_pylist()
+
+
[email protected]('mode', ['fixed', 'postpone-fixed'])
+def
test_managed_blob_roundtrip_nested_nulls_duplicates_and_payload_filters(tmp_path,
mode):
+ table = _table(tmp_path, mode)
+ # Two commits force merging old descriptors with new descriptors.
+ batches = [
+ [dict(id=1, payload=b'old', items=[b'a', None], mapping=[('a', b'x'),
('a', b'y')], op='+I'),
+ dict(id=2, payload=None, items=None, mapping=None, op='+I'),
+ dict(id=3, payload=b'other', items=[], mapping=[], op='+I')],
+ [dict(id=1, payload=b'new', items=[None, b'', b'new'], mapping=[('b',
None), ('b', b'')], op='+U')],
+ ]
+ for rows in batches:
+ builder = _builder(table, mode)
+ writer, commit = builder.new_write(), builder.new_commit()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow(_input(table, rows))
+ messages = writer.prepare_commit()
+ assert all(any(name.endswith('.blobref') for name in f.extra_files)
+ for m in messages for f in m.new_files)
+ commit.commit(messages)
+ finally:
+ writer.close()
+ commit.close()
+ expected = [batches[1][0], *batches[0][1:]]
+ assert sorted(_read(table), key=lambda row: row['id']) == expected
+ predicate =
table.new_read_builder().new_predicate_builder().equal('payload', b'new')
+ assert _read(table, predicate, ['items', 'id'], limit=1) ==
[dict(items=[None, b'', b'new'], id=1)]
+ assert _read(table, predicate, ['payload', 'id'], limit=1) ==
[dict(payload=b'new', id=1)]
+ # Descriptors are exposed only when requested, including nested leaves.
+ descriptors = sorted(_read(table, descriptors=True), key=lambda row:
row['id'])
+ assert BlobDescriptor.is_blob_descriptor(descriptors[0]['payload'])
+ assert BlobDescriptor.is_blob_descriptor(descriptors[0]['items'][1])
+ assert descriptors[0]['items'][0] is None
+ assert descriptors[1]['payload'] is None
+
+
[email protected]('mode', ['fixed', 'postpone-fixed', 'pending'])
+def test_managed_blob_arrow_inputs_copy_referenced_payloads(tmp_path, mode):
+ table = _table(tmp_path, mode)
+ source = tmp_path / 'source.bin'
+ source.write_bytes(b'reference')
+ reference = Blob.from_file(table.file_io, str(source), 0, 9)
+ builder = _builder(table, mode)
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow(_input(table, [dict(
+ id=1, payload=b'row',
items=[reference.to_descriptor().serialize(), None],
+ mapping=[('a', b'')], op='+I')]))
+ writer.write_arrow(_input(table, [dict(id=2, payload=b'arrow',
items=[], mapping=[], op='+I')]))
+ messages = writer.prepare_commit()
+ assert writer._python_writer is None
+ rows = []
+ for message in messages:
+ for file in message.new_files:
+ with table.file_io.new_input_stream(file.file_path) as stream:
+ rows.extend(pq.ParquetFile(stream).read().to_pylist())
+ row = next(row for row in rows if row['id'] == 1)
+ assert Blob.from_descriptor_bytes(row['items'][0],
table.file_io).to_data() == b'reference'
+ assert BlobDescriptor.is_blob_descriptor(row['payload'])
+ assert source.read_bytes() == b'reference'
+ finally:
+ writer.close()
+
+
+def test_pending_blob_retracts_drop_payloads_but_preserve_events(tmp_path):
+ table = _table(tmp_path, 'pending', {'target-file-row-num': '2'})
+ builder = table.new_stream_write_builder()
+ writer = builder.new_write()
+ try:
+ for checkpoint in (1, 2):
+ rows = [dict(id=3, payload=b'a', items=[b'b'], mapping=[('c',
b'd')], op='+I'),
+ dict(id=1, payload=b'ignored', items=[b'ignored'],
mapping=[('c', b'ignored')], op='-D'),
+ dict(id=3, payload=b'e', items=[], mapping=[], op='+U')]
+ writer.write_arrow(_input(table, rows))
+ messages = writer.prepare_commit(checkpoint)
+ files = [f for m in messages for f in m.new_files]
+ assert [f.row_count for f in files] == [3]
+ assert files[0].delete_row_count == 1
+ with table.file_io.new_input_stream(files[0].file_path) as stream:
+ physical = pq.ParquetFile(stream).read().to_pylist()
+ assert [row['id'] for row in physical] == [3, 1, 3]
+ assert physical[1]['payload'] is None
+ assert physical[1]['items'] is None
+ assert physical[1]['mapping'] is None
+ assert writer.prepare_commit(checkpoint + 100) == []
+ finally:
+ writer.close()
diff --git a/paimon-python/pypaimon/tests/native_postpone_write_test.py
b/paimon-python/pypaimon/tests/native_postpone_write_test.py
index 2c77b224ac..fcd6d2eaaf 100644
--- a/paimon-python/pypaimon/tests/native_postpone_write_test.py
+++ b/paimon-python/pypaimon/tests/native_postpone_write_test.py
@@ -139,12 +139,54 @@ def test_postpone_native_overwrite_and_abort(tmp_path,
native_rest_catalog, nati
writer.close()
[email protected]('engine', ['first-row', 'partial-update',
'aggregation'])
-def test_postpone_engine_retract_validation_stays_on_python(tmp_path, engine):
- table = _table(tmp_path, {'merge-engine': engine})
[email protected]('engine,extra,accepted', [
+ ('deduplicate', {}, True),
+ ('first-row', {}, False),
+ ('partial-update', {}, False),
+ ('partial-update', {'partial-update.remove-record-on-delete': 'true'},
True),
+ ('aggregation', {'fields.v.aggregate-function': 'sum'}, True),
+ ('aggregation', {'fields.v.aggregate-function': 'min'}, False),
+ ('aggregation', {'fields.v.aggregate-function': 'min',
'fields.v.ignore-retract': 'true'}, True),
+])
[email protected]('kind', ['-D', '-U'])
+def test_postpone_validates_retract_without_merging_events(tmp_path, engine,
extra, accepted, kind):
+ table = _table(tmp_path, {'merge-engine': engine, **extra})
writer = table.new_batch_write_builder().new_write()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow_batch(_batch([(3, 'a', 10, '+I'), (1, 'a', 20,
'+I'), (3, 'a', 30, '+I')]))
+ retract = [(3, 'a', 5, kind)]
+ if accepted:
+ writer.write_arrow_batch(_batch(retract))
+ files = [f for m in writer.prepare_commit() for f in m.new_files]
+ assert _file_rows(table, files) == [
+ (3, 'a', 10, '+I'), (1, 'a', 20, '+I'), (3, 'a', 30, '+I'),
*retract]
+ else:
+ with pytest.raises(Exception,
match='(?i)retract|DELETE|UPDATE_BEFORE'):
+ writer.write_arrow_batch(_batch(retract))
+ with pytest.raises(Exception, match='(?i)fail'):
+ writer.prepare_commit()
+ assert not list(tmp_path.rglob('*.parquet'))
+ finally:
+ writer.close()
+
+
+def
test_postpone_first_retract_state_survives_checkpoints_and_is_partition_local(tmp_path):
+ table = _table(tmp_path, {'merge-engine': 'aggregation',
'fields.v.aggregate-function': 'min',
+ 'aggregation.remove-record-on-delete': 'true'})
+ writer = table.new_stream_write_builder().new_write()
try:
- assert not isinstance(writer, NativeTableWrite)
+ writer.write_arrow_batch(_batch([(1, 'a', 1, '-D')]))
+ first = writer.prepare_commit(1)
+ # The valid first DELETE is remembered; Java does not validate another
+ # retract in this writer, even if the next kind is UPDATE_BEFORE.
+ writer.write_arrow_batch(_batch([(2, 'a', 2, '-U')]))
+ second = writer.prepare_commit(2)
+ assert _file_rows(table, [f for m in second for f in m.new_files]) ==
[(2, 'a', 2, '-U')]
+ assert writer.prepare_commit(3) == []
+ with pytest.raises(Exception, match='(?i)retract'):
+ writer.write_arrow_batch(_batch([(1, 'b', 1, '-U')]))
+ assert all(table.file_io.exists(f.file_path) for m in first + second
for f in m.new_files)
finally:
writer.close()
diff --git a/paimon-python/pypaimon/tests/write/postpone_event_write_test.py
b/paimon-python/pypaimon/tests/write/postpone_event_write_test.py
new file mode 100644
index 0000000000..16285e2a11
--- /dev/null
+++ b/paimon-python/pypaimon/tests/write/postpone_event_write_test.py
@@ -0,0 +1,152 @@
+# 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.
+
+"""Pending files carry source events, not merged primary-key rows."""
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+
+
+def make_table(tmp_path, options=None, partition_keys=None):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ settings = {'bucket': '-2', 'file.format': 'parquet',
'write.native.enabled': 'false'}
+ settings.update(options or {})
+ schema = pa.schema([('id', pa.int32()), ('v', pa.int32()), ('op',
pa.string())])
+ catalog.create_table('default.events', Schema.from_pyarrow_schema(
+ schema, primary_keys=['id'] + (partition_keys or []),
+ partition_keys=partition_keys or [], options=settings), False)
+ return catalog.get_table('default.events'), schema
+
+
+def stored_rows(table, messages):
+ rows = []
+ for message in messages:
+ for file in message.new_files:
+ with table.file_io.new_input_stream(file.file_path) as source:
+ rows.extend(pq.ParquetFile(source).read().to_pylist())
+ return rows
+
+
[email protected]('engine', ['deduplicate', 'first-row',
'partial-update', 'aggregation'])
+def test_postpone_preserves_unsorted_duplicate_rows(tmp_path, engine):
+ table, schema = make_table(tmp_path, {'merge-engine': engine})
+ writer = table.new_batch_write_builder().new_write()
+ try:
+ writer.write_arrow(pa.table({'id': [3, 1, 3], 'v': [10, 20, 30],
+ 'op': ['+I'] * 3}, schema=schema))
+ messages = writer.prepare_commit()
+ assert [(row['id'], row['v']) for row in stored_rows(table, messages)]
== [(3, 10), (1, 20), (3, 30)]
+ finally:
+ writer.close()
+
+
[email protected]('row_input', [False, True])
[email protected]('options,expected', [
+ ({}, [0, 1, 2, 3]),
+ ({'ignore-delete': 'true'}, [0, 2]),
+ ({'ignore-update-before': 'true'}, [0, 2, 3]),
+])
+def test_python_postpone_row_kinds_are_filtered_before_writing(tmp_path,
row_input, options, expected):
+ from pypaimon.table.row.generic_row import GenericRow
+ table, schema = make_table(tmp_path, {'rowkind.field': 'op', **options})
+ writer = table.new_batch_write_builder().new_write()
+ rows = [{'id': i, 'v': i, 'op': kind} for i, kind in enumerate(['+I',
'-U', '+U', '-D'])]
+ try:
+ if row_input:
+ for row in rows:
+ writer.write_row(GenericRow(list(row.values()), table.fields))
+ else:
+ writer.write_arrow(pa.Table.from_pylist(rows, schema=schema))
+ messages = writer.prepare_commit()
+ physical = stored_rows(table, messages)
+ assert [row['id'] for row in physical] == expected
+ assert [row['_VALUE_KIND'] for row in physical] == expected
+ assert sum(f.delete_row_count for m in messages for f in m.new_files)
== sum(i in (1, 3) for i in expected)
+ finally:
+ writer.close()
+
+
[email protected]('engine', ['first-row', 'partial-update',
'aggregation'])
+def
test_python_postpone_rejects_unsupported_retract_and_aborts_rolled_files(tmp_path,
engine):
+ table, schema = make_table(tmp_path, {
+ 'merge-engine': engine, 'rowkind.field': 'op', 'target-file-row-num':
'1'})
+ writer = table.new_stream_write_builder().new_write()
+ try:
+ writer.write_arrow(pa.table({'id': [1], 'v': [1], 'op': ['+I']},
schema=schema))
+ assert list(tmp_path.rglob('*.parquet'))
+ with pytest.raises((ValueError, NotImplementedError),
match='(?i)retract|DELETE|non-INSERT'):
+ writer.write_arrow(pa.table({'id': [1], 'v': [1], 'op': ['-D']},
schema=schema))
+ assert not list(tmp_path.rglob('*.parquet'))
+ with pytest.raises(RuntimeError, match='failed'):
+ writer.prepare_commit(1)
+ finally:
+ writer.close()
+
+
+def test_python_postpone_checkpoints_do_not_repeat_prepared_files(tmp_path):
+ table, schema = make_table(tmp_path, {'target-file-row-num': '2',
'rowkind.field': 'op'})
+ writer = table.new_stream_write_builder().new_write()
+ try:
+ for checkpoint in range(3):
+ writer.write_arrow(pa.table({'id': [3, 1, 3], 'v': [10, 20, 30],
+ 'op': ['+I', '-U', '-D']},
schema=schema))
+ messages = writer.prepare_commit(checkpoint)
+ rows = stored_rows(table, messages)
+ assert [row['id'] for row in rows] == [3, 1, 3]
+ assert [row['_SEQUENCE_NUMBER'] for row in rows] ==
list(range(checkpoint * 3, checkpoint * 3 + 3))
+ assert [file.row_count for message in messages for file in
message.new_files] == [3]
+ assert writer.prepare_commit(checkpoint + 100) == []
+ paths = [f.file_path for m in messages for f in m.new_files]
+ writer.abort()
+ assert all(table.file_io.exists(path) for path in paths)
+ finally:
+ writer.close()
+
+
+def test_python_aggregation_writer_uses_field_aggregators(tmp_path):
+ table, schema = make_table(tmp_path, {
+ 'bucket': '1', 'merge-engine': 'aggregation',
'fields.v.aggregate-function': 'sum'})
+ writer = table.new_batch_write_builder().new_write()
+ try:
+ writer.write_arrow(pa.table({'id': [3, 1, 3], 'v': [10, 20, 30], 'op':
['+I'] * 3}, schema=schema))
+ rows = stored_rows(table, writer.prepare_commit())
+ assert [(row['id'], row['v']) for row in rows] == [(1, 20), (3, 40)]
+ finally:
+ writer.close()
+
+
+def
test_python_postpone_prepare_failure_keeps_earlier_partition_owned(tmp_path,
monkeypatch):
+ table, schema = make_table(tmp_path, partition_keys=['v'])
+ writer = table.new_stream_write_builder().new_write()
+ writer.write_arrow(pa.table({'id': [1, 2], 'v': [10, 20], 'op': ['+I',
'+I']}, schema=schema))
+ writers = list(writer.file_store_write.data_writers.values())
+ assert len(writers) == 2
+
+ def fail_prepare():
+ raise OSError('injected prepare failure')
+
+ monkeypatch.setattr(writers[1], 'prepare_commit', fail_prepare)
+ try:
+ with pytest.raises(OSError, match='injected'):
+ writer.prepare_commit(1)
+ assert list(tmp_path.rglob('*.parquet'))
+ finally:
+ writer.close()
+ assert not list(tmp_path.rglob('*.parquet'))
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py
b/paimon-python/pypaimon/tests/write/table_write_test.py
index 8006269bea..4d751ce59e 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -227,6 +227,7 @@ class TableWriteTest(unittest.TestCase):
@staticmethod
def _mock_table_write(partitions, buckets):
table_write = object.__new__(TableWrite)
+ table_write.table = Mock(is_primary_key_table=False)
table_write._validate_pyarrow_schema = Mock()
table_write.row_key_extractor = Mock()
table_write.file_store_write = Mock()
@@ -301,6 +302,7 @@ class TableWriteTest(unittest.TestCase):
'payload': [b'a', b'b', b'c', b'd'],
})
table_write = object.__new__(TableWrite)
+ table_write.table = Mock(is_primary_key_table=False)
table_write._validate_pyarrow_schema = Mock()
table_write.file_store_write = Mock()
table_write.row_key_extractor = Mock()
@@ -325,6 +327,7 @@ class TableWriteTest(unittest.TestCase):
'payload': [b'a', b'b', b'c', b'd'],
})
table_write = object.__new__(TableWrite)
+ table_write.table = Mock(is_primary_key_table=False)
table_write._validate_pyarrow_schema = Mock()
table_write.file_store_write = Mock()
table_write.row_key_extractor = Mock()
diff --git a/paimon-python/pypaimon/write/file_store_write.py
b/paimon-python/pypaimon/write/file_store_write.py
index 16f0884516..02ddd0898d 100644
--- a/paimon-python/pypaimon/write/file_store_write.py
+++ b/paimon-python/pypaimon/write/file_store_write.py
@@ -33,6 +33,7 @@ from pypaimon.write.writer.dedicated_format_writer import
DedicatedFormatWriter
from pypaimon.write.writer.data_vector_writer import DataVectorWriter
from pypaimon.write.writer.data_writer import DataWriter
from pypaimon.write.writer.key_value_data_writer import KeyValueDataWriter
+from pypaimon.write.writer.postpone_data_writer import PostponeDataWriter
from pypaimon.table.bucket_mode import BucketMode
@@ -162,19 +163,25 @@ class FileStoreWrite:
f"target-file-row-num should be at most {max_value}")
if row_limit != max_value:
row_rolling_supported = (
- self.table.options.data_evolution_enabled()
- and not self.table.is_primary_key_table)
+ (self.table.options.data_evolution_enabled()
+ and not self.table.is_primary_key_table)
+ or bucket == BucketMode.POSTPONE_BUCKET.value)
if not row_rolling_supported:
raise NotImplementedError(
"target-file-row-num is set on this table but pypaimon
supports row-count "
- "based file rolling only for data-evolution append tables
(no primary key); "
+ "based file rolling only for data-evolution append and
postpone tables; "
"unset it or write with Java/Flink/Spark.")
def max_seq_number():
return self._seq_number_stats(partition).get(bucket, 1)
- # Check if table has blob columns
+ # Dedicated Blob files are an append-table layout. PK tables require
+ # managed packs and references attached to their key-value Parquet
files.
if self._has_blob_columns():
+ if self.table.is_primary_key_table:
+ raise NotImplementedError(
+ 'Primary-key Blob writes require the native writer; '
+ 'enable write.native.enabled and write Arrow batches')
return DedicatedFormatWriter(
table=self.table,
partition=partition,
@@ -196,11 +203,13 @@ class FileStoreWrite:
write_cols=self.write_cols,
)
elif self.table.is_primary_key_table:
- return KeyValueDataWriter(
+ writer_type = (PostponeDataWriter if bucket ==
BucketMode.POSTPONE_BUCKET.value
+ else KeyValueDataWriter)
+ return writer_type(
table=self.table,
partition=partition,
bucket=bucket,
- max_seq_number=max_seq_number(),
+ max_seq_number=(0 if bucket ==
BucketMode.POSTPONE_BUCKET.value else max_seq_number()),
options=options,
merge_function=self._build_pk_merge_function(),
changelog_producer=self.changelog_producer)
@@ -256,8 +265,6 @@ class FileStoreWrite:
from pypaimon.common.merge_engine_dispatch import (
build_merge_function, partial_update_unsupported_options)
from pypaimon.common.options.core_options import MergeEngine
- from pypaimon.read.reader.deduplicate_merge_function import \
- DeduplicateMergeFunction
engine = self.options.merge_engine()
raw_options = self.options.options.to_map()
@@ -286,25 +293,15 @@ class FileStoreWrite:
f.name for f in self.table.table_schema.fields],
)
- # Catch the dispatch's "wholly unsupported engine" raise only
- # for the engines we know are out of scope today; any other
- # NotImplementedError is a bug we want to surface, not swallow.
if engine == MergeEngine.AGGREGATE:
- # Surface the silent semantic mismatch in logs: the file
- # will be PK-unique (better than the pre-PR multi-row
- # corruption), but any reader that honours the declared
- # engine will see wrong values. Users sharing tables
- # across writers especially need to see this.
- logger.warning(
- "merge-engine '%s' is not implemented on the pypaimon "
- "write path; falling back to deduplicate so the flushed "
- "file stays PK-unique. The file contents reflect "
- "deduplicate semantics (latest writer wins), not %s "
- "semantics. Any reader that interprets the file under "
- "the declared engine will return incorrect results. "
- "Avoid the pypaimon writer for tables on this engine.",
- engine.value, engine.value)
- return DeduplicateMergeFunction()
+ from pypaimon.read.reader.aggregation_merge_function import (
+ AggregateMergeFunction, build_field_aggregators)
+ fields = self.table.table_schema.fields
+ return AggregateMergeFunction(
+ key_arity=len(self.table.trimmed_primary_keys),
+ value_arity=len(fields),
+ field_aggregators=build_field_aggregators(
+ fields, self.table.primary_keys, self.options))
all_value_fields = self.table.table_schema.fields
return build_merge_function(
@@ -339,8 +336,16 @@ 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):
+ # 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()
+
def close(self):
"""Close all data writers and clean up resources."""
for writer in self.data_writers.values():
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index d0ca273479..64d0fd774f 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -76,14 +76,7 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False,
BucketMode.BUCKET_UNAWARE,
BucketMode.CROSS_PARTITION,
BucketMode.POSTPONE_MODE)
- # Postpone bypasses the KV merger. Native retract validation for
- # other engines and managed BLOB externalization are still missing.
- or (table.bucket_mode() == BucketMode.POSTPONE_MODE and not
fixed_bucket
- and (table.options.merge_engine() != MergeEngine.DEDUPLICATE
- or any(is_blob_file_field(field) for field in
table.table_schema.fields)))
- or (fixed_bucket and (table.options.deletion_vectors_enabled()
- or any(is_blob_file_field(field)
- for field in
table.table_schema.fields)))
+ or (fixed_bucket and table.options.deletion_vectors_enabled())
or (table.options.deletion_vectors_enabled()
and table.options.merge_engine() in
(MergeEngine.PARTIAL_UPDATE,
MergeEngine.AGGREGATE))
@@ -93,10 +86,11 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False,
or not _native_map_layouts_supported(table, schema)
# Rust cannot encode these partition keys yet.
or not _native_partition_types_supported(schema,
table.partition_keys)
- # Native dedicated files currently support top-level scalar Blob
fields.
+ # Append dedicated files currently support top-level scalar Blob
fields.
or table.options.video_frame_fields()
- or any(is_blob_file_field(field) and not is_blob_type(field.type)
- for field in table.table_schema.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))):
return None
native_table = create_native_write_table(table)
if native_table is None:
diff --git a/paimon-python/pypaimon/write/row_kind.py
b/paimon-python/pypaimon/write/row_kind.py
new file mode 100644
index 0000000000..9e501e6565
--- /dev/null
+++ b/paimon-python/pypaimon/write/row_kind.py
@@ -0,0 +1,66 @@
+# 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.
+
+"""Configured row kinds and pre-routing filters, matching Java
RowKindFilter."""
+
+import pyarrow as pa
+
+from pypaimon.table.row.row_kind import RowKind
+
+
+def row_kinds(options, data):
+ name = options.options.to_map().get('rowkind.field')
+ if name is None:
+ return [0] * data.num_rows
+ if name not in data.schema.names:
+ raise ValueError('Cannot find rowkind field %s in table schema' % name)
+ column = data.column(name)
+ if not pa.types.is_string(column.type):
+ raise ValueError('rowkind.field column must be a string')
+ return [RowKind.from_string(value).value for value in column.to_pylist()]
+
+
+def filter_write_batch(table, data):
+ if not table.is_primary_key_table:
+ return data
+ options = table.options
+ raw = options.options.to_map()
+ if 'rowkind.field' not in raw:
+ return data
+ kinds = row_kinds(options, data)
+ ignore_delete = options.ignore_delete()
+ ignore_before = str(raw.get('ignore-update-before', 'false')).lower() ==
'true'
+ if not ignore_delete and not ignore_before:
+ return data
+ keep = [not _is_filtered(kind, ignore_delete, ignore_before) for kind in
kinds]
+ return data.filter(pa.array(keep, type=pa.bool_()))
+
+
+def skip_write_row(table, values):
+ if not table.is_primary_key_table:
+ return False
+ options = table.options
+ raw = options.options.to_map()
+ name = raw.get('rowkind.field')
+ if name is None:
+ return False
+ kind = RowKind.from_string(values[name]).value
+ return _is_filtered(kind, options.ignore_delete(),
+ str(raw.get('ignore-update-before', 'false')).lower()
== 'true')
+
+
+def _is_filtered(kind, ignore_delete, ignore_before):
+ return (ignore_delete and kind in (1, 3)) or (ignore_before and kind == 1)
diff --git a/paimon-python/pypaimon/write/table_write.py
b/paimon-python/pypaimon/write/table_write.py
index 87867b54b5..21fdce5598 100644
--- a/paimon-python/pypaimon/write/table_write.py
+++ b/paimon-python/pypaimon/write/table_write.py
@@ -197,6 +197,9 @@ class TableWrite:
)
require_columns(values_by_name, column_names, "write_row")
require_columns(values_by_name, self.table.partition_keys, "write_row")
+ from pypaimon.write.row_kind import skip_write_row
+ if skip_write_row(self.table, values_by_name):
+ return
partition, bucket = (
self.row_key_extractor.extract_partition_bucket_row(values_by_name)
)
@@ -344,7 +347,13 @@ class TableWrite:
def _prepare_arrow_data(self, data):
self._validate_pyarrow_schema(data.schema)
- return normalize_arrow_strings(data)
+ data = normalize_arrow_strings(data)
+ # Schema-only writers can normalize input without table options.
+ table = getattr(self, 'table', None)
+ if table is None:
+ return data
+ from pypaimon.write.row_kind import filter_write_batch
+ return filter_write_batch(table, data)
def _validate_pyarrow_schema(self, data_schema: pa.Schema):
if self._is_compatible_pyarrow_schema(data_schema,
self.table_pyarrow_schema):
diff --git a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
index 6992852bbc..9e9998cd3d 100644
--- a/paimon-python/pypaimon/write/writer/key_value_data_writer.py
+++ b/paimon-python/pypaimon/write/writer/key_value_data_writer.py
@@ -269,8 +269,8 @@ class KeyValueDataWriter(DataWriter):
new_arrays.append(sequence_column)
new_fields.append(pa.field('_SEQUENCE_NUMBER', pa.int64(),
nullable=False))
- # TODO: support real row kind here
- value_kind_column = pa.array([0] * num_rows, type=pa.int8())
+ from pypaimon.write.row_kind import row_kinds
+ value_kind_column = pa.array(row_kinds(self.options, data),
type=pa.int8())
new_arrays.append(value_kind_column)
new_fields.append(pa.field('_VALUE_KIND', pa.int8(), nullable=False))
diff --git a/paimon-python/pypaimon/write/writer/postpone_data_writer.py
b/paimon-python/pypaimon/write/writer/postpone_data_writer.py
new file mode 100644
index 0000000000..7f53531a62
--- /dev/null
+++ b/paimon-python/pypaimon/write/writer/postpone_data_writer.py
@@ -0,0 +1,109 @@
+# 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.
+
+"""Write pending primary-key events in arrival order, as Java does."""
+
+import pyarrow as pa
+
+from pypaimon.common.options.core_options import ChangelogProducer
+from pypaimon.table.row.key_value import KeyValue
+from pypaimon.write.writer.key_value_data_writer import KeyValueDataWriter
+
+
+class PostponeDataWriter(KeyValueDataWriter):
+ def __init__(self, **kwargs):
+ # Pending files never restore bucket sequences or produce a separate
+ # input changelog. Their own events become input to bucket assignment.
+ kwargs['max_seq_number'] = -1
+ kwargs['changelog_producer'] = ChangelogProducer.NONE
+ super().__init__(**kwargs)
+ self._retract_validated = False
+ self._failed = False
+ self._writing_stats = None
+
+ def write(self, data):
+ self._ensure_active()
+ try:
+ super().write(data)
+ except Exception:
+ self._failed = True
+ raise
+
+ def _ensure_active(self):
+ if self._failed:
+ raise RuntimeError('Postpone writer failed; create a new writer')
+
+ def _process_data(self, data):
+ enhanced = self._add_system_fields(data)
+ self._validate_retract(enhanced,
enhanced.column('_VALUE_KIND').to_pylist())
+ return pa.Table.from_batches([enhanced])
+
+ def _validate_retract(self, data, kinds):
+ if self._retract_validated:
+ return
+ for index, kind in enumerate(kinds):
+ if kind not in (1, 3):
+ continue
+ row = tuple(column[index].as_py() for column in data.columns)
+ kv = KeyValue(len(self.trimmed_primary_keys),
len(self.table.fields)).replace(row)
+ self._merge_function.reset()
+ self._merge_function.add(kv)
+ self._merge_function.get_result()
+ self._retract_validated = True
+ break
+
+ def _check_and_roll_if_needed(self):
+ if self._buffer.num_rows and (
+ self._buffer.num_rows >= self.target_file_row_num
+ or self._buffer.nbytes >= self.target_file_size):
+ self._flush_all()
+
+ def _flush_all(self):
+ data = self._buffer.materialize()
+ if data is not None and data.num_rows:
+ self._write_data_to_file(data)
+ self._buffer.reset()
+
+ def _write_data_to_file(self, data):
+ sequence = data.column('_SEQUENCE_NUMBER')
+ self._writing_stats = (
+ sequence[0].as_py(), sequence[-1].as_py(),
+ sum(kind in (1, 3) for kind in
data.column('_VALUE_KIND').to_pylist()))
+ try:
+ super()._write_data_to_file(data)
+ finally:
+ self._writing_stats = None
+
+ def _create_data_file_meta(self, **kwargs):
+ minimum, maximum, deletes = self._writing_stats
+ kwargs['min_sequence_number'] = minimum
+ kwargs['max_sequence_number'] = maximum
+ meta = super()._create_data_file_meta(**kwargs)
+ meta.delete_row_count = deletes
+ return meta
+
+ def prepare_commit(self):
+ self._ensure_active()
+ try:
+ return super().prepare_commit()
+ except Exception:
+ self._failed = True
+ self.abort()
+ raise
+
+ def close(self):
+ # Prepared messages own their files; discard only unprepared events.
+ self.abort()
