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 567b57323f [python] Keep Blob row writes and upserts on native core
(#10455)
567b57323f is described below
commit 567b57323fececda9d847811fec9a4e93579429a
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Oct 9 13:23:40 2026 +0800
[python] Keep Blob row writes and upserts on native core (#10455)
---
.../pypaimon/tests/native_blob_row_write_test.py | 447 +++++++++++++++++++++
.../pypaimon/tests/native_blob_update_test.py | 5 +-
.../pypaimon/tests/native_blob_uri_reader_test.py | 96 ++++-
.../pypaimon/tests/native_blob_write_test.py | 15 +-
paimon-python/pypaimon/write/native_blob_rows.py | 72 +++-
paimon-python/pypaimon/write/native_update.py | 46 ++-
paimon-python/pypaimon/write/native_write.py | 41 +-
7 files changed, 679 insertions(+), 43 deletions(-)
diff --git a/paimon-python/pypaimon/tests/native_blob_row_write_test.py
b/paimon-python/pypaimon/tests/native_blob_row_write_test.py
new file mode 100644
index 0000000000..b95153b8e6
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_blob_row_write_test.py
@@ -0,0 +1,447 @@
+# 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.
+
+"""Existing row Blob inputs must retain the core writer and source lifetime."""
+
+import io
+from decimal import Decimal
+from unittest.mock import patch
+
+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 Blob, BlobDescriptor, BlobViewStruct
+from pypaimon.table.row.generic_row import GenericRow
+from pypaimon.write.native_write import NativeTableWrite
+
+pytestmark = pytest.mark.native_plan
+
+
+class StreamBlob(Blob):
+ def __init__(self, payload, error=None):
+ self.payload = payload
+ self.error = error
+ self.streams = []
+ self.opened = 0
+
+ def to_data(self):
+ raise AssertionError('Blob payload was materialized in Python')
+
+ def to_descriptor(self):
+ raise AssertionError('Custom Blob must use its input stream')
+
+ def new_input_stream(self):
+ self.opened += 1
+ if self.error is not None:
+ raise self.error
+
+ class Stream(io.BytesIO):
+ def __init__(self, payload):
+ super().__init__(payload)
+ self.read_sizes = []
+
+ def read(self, size=-1):
+ assert 0 < size <= 3, 'Core must use its configured copy
buffer'
+ self.read_sizes.append(size)
+ return super().read(size)
+
+ stream = Stream(self.payload)
+ self.streams.append(stream)
+ return stream
+
+
+def _type(kind):
+ blob = AtomicType('BLOB')
+ if kind == 'array':
+ return ArrayType(True, blob)
+ if kind == 'map':
+ return MapType(True, AtomicType('STRING'), blob)
+ return blob
+
+
+def _value(kind, payload):
+ if kind == 'array':
+ return [payload, None, b'']
+ if kind == 'map':
+ return [('object', payload), ('null', None), ('empty', b'')]
+ return payload
+
+
+def _table(tmp_path, kind, external=False, extra=None, payload_type=None):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path / 'warehouse')})
+ catalog.create_database('db', True)
+ fields = [DataField(0, 'id', AtomicType('INT')),
+ DataField(1, 'pt', AtomicType('STRING')), DataField(2,
'payload', payload_type or _type(kind))]
+ options = {'row-tracking.enabled': 'true', 'data-evolution.enabled':
'true',
+ 'write.native.enabled': 'true', 'read.native.enabled': 'true',
+ 'scan.native-plan.enabled': 'true', 'target-file-row-num': '2',
+ 'blob.target-file-size': '32 B', 'blob.copy-buffer-size': '3 B'}
+ if external:
+ options['data-file.external-paths'] = (tmp_path / 'external').as_uri()
+ options['data-file.external-paths.strategy'] = 'round-robin'
+ options.update(extra or {})
+ catalog.create_table('db.t', Schema(fields, partition_keys=['pt'],
options=options), False)
+ return catalog.get_table('db.t')
+
+
+def _read(table):
+ builder = table.new_read_builder()
+ read = builder.new_read()
+ with patch.object(read, '_create_split_read',
side_effect=AssertionError('Python read fallback')):
+ return
sorted(read.to_arrow(builder.new_scan().plan().splits()).to_pylist(),
key=lambda row: row['id'])
+
+
+def _commit(builder, messages, stream, identifier=7):
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, identifier) if stream else
commit.commit(messages)
+ finally:
+ commit.close()
+
+
[email protected]('kind', ['scalar', 'array', 'map'])
[email protected]('stream', [False, True])
[email protected]('external', [False, True])
+def test_blob_rows_mix_with_arrow_and_release_sources(tmp_path, kind, stream,
external):
+ table = _table(tmp_path, kind, external)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer = builder.new_write()
+ blob = StreamBlob(b'streamed payload')
+ empty = b'' if kind == 'scalar' else []
+ expected = [dict(id=0, pt='a', payload=_value(kind, b'arrow')),
+ dict(id=1, pt='a', payload=_value(kind, blob.payload)),
+ dict(id=2, pt='b', payload=empty), dict(id=3, pt='b',
payload=None)]
+ try:
+ assert isinstance(writer, NativeTableWrite)
+ with patch.object(writer, '_switch_to_python',
side_effect=AssertionError('Python writer fallback')):
+ writer.write_arrow(pa.Table.from_pylist(
+ expected[:1],
schema=PyarrowFieldParser.from_paimon_schema(table.fields)))
+ writer.write_row(GenericRow([1, 'a', _value(kind, blob)],
table.fields))
+ writer.write_row(GenericRow([2, 'b', empty], table.fields))
+ writer.write_row(GenericRow([3, 'b', None], table.fields))
+ assert blob.opened == 1
+ assert all(source.closed and source.read_sizes for source in
blob.streams)
+ assert writer._blob_rows.readers == {}
+ assert writer._python_writer is None
+ messages = writer.prepare_commit(7) if stream else
writer.prepare_commit()
+ _commit(builder, messages, stream)
+ assert _read(table) == expected
+ finally:
+ writer.close()
+
+
[email protected]('kind', ['scalar', 'array', 'map'])
+def test_stream_blob_rows_keep_routing_after_multiple_prepares(tmp_path, kind):
+ table = _table(tmp_path, kind)
+ builder = table.new_stream_write_builder()
+ writer = builder.new_write()
+ expected = []
+ try:
+ for identifier in [1, 2]:
+ blob = StreamBlob(('payload-%d' % identifier).encode())
+ with patch.object(writer, '_switch_to_python',
side_effect=AssertionError('Python writer fallback')):
+ writer.write_row(GenericRow([identifier, 'a', _value(kind,
blob)], table.fields))
+ assert blob.opened == 1 and all(source.closed for source in
blob.streams)
+ assert writer._blob_rows.readers == {}
+ _commit(builder, writer.prepare_commit(identifier), True,
identifier)
+ expected.append(dict(id=identifier, pt='a', payload=_value(kind,
blob.payload)))
+ assert _read(table) == expected
+ finally:
+ writer.close()
+
+
[email protected]('kind', ['scalar', 'array', 'map'])
+def
test_blob_row_subset_uses_named_fields_without_opening_omitted_blobs(tmp_path,
kind):
+ table = _table(tmp_path, kind)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write().with_write_type(['pt', 'id'])
+ omitted = StreamBlob(b'ignored', AssertionError('omitted Blob opened'))
+ try:
+ with patch.object(writer, '_switch_to_python',
side_effect=AssertionError('Python writer fallback')):
+ writer.write_row(GenericRow([_value(kind, omitted), 1, 'a'],
+ [table.fields[2], table.fields[0],
table.fields[1]]))
+ assert omitted.opened == 0
+ _commit(builder, writer.prepare_commit(), False)
+ assert _read(table) == [dict(id=1, pt='a', payload=None)]
+ finally:
+ writer.close()
+
+
[email protected]('stream', [False, True])
[email protected]('blob_object', [False, True])
+def test_blob_row_map_preserves_java_null_keys(tmp_path, stream, blob_object):
+ from pypaimon.read.reader.format_blob_reader import FormatBlobReader
+ table = _table(tmp_path, 'map', extra={'blob.target-file-size': '1 MB'})
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer = builder.new_write()
+ value = StreamBlob(b'NULL key') if blob_object else b'NULL key'
+ try:
+ with patch.object(writer, '_switch_to_python',
side_effect=AssertionError('Python writer fallback')):
+ writer.write_row(GenericRow([1, 'a', [(None, value), ('normal',
None)]], table.fields))
+ messages = writer.prepare_commit(7) if stream else
writer.prepare_commit()
+ _commit(builder, messages, stream)
+ files = [file for message in messages for file in message.new_files if
file.file_name.endswith('.blob')]
+ assert len(files) == 1
+ reader = FormatBlobReader(table.file_io, files[0].file_path,
['payload'], [table.fields[2]], None, False)
+ try:
+ row = reader.read_values_at([0])[0]
+ assert row[None].to_data() == b'NULL key' and row['normal'] is None
+ finally:
+ reader.close()
+ finally:
+ writer.close()
+
+
+def
test_duplicate_blob_map_keys_are_rejected_before_opening_payloads(tmp_path):
+ table = _table(tmp_path, 'map')
+ writer = table.new_batch_write_builder().new_write()
+ first = StreamBlob(b'first', AssertionError('duplicate map opened'))
+ second = StreamBlob(b'second', AssertionError('duplicate map opened'))
+ try:
+ with patch.object(writer, '_switch_to_python',
side_effect=AssertionError('Python writer fallback')):
+ with pytest.raises(ValueError, match='unique'):
+ writer.write_row(GenericRow([1, 'a', [(None, first), (None,
second)]], table.fields))
+ assert first.opened == second.opened == 0
+ assert writer._blob_rows.readers == {}
+ assert not list(tmp_path.rglob('*.blob'))
+ finally:
+ writer.close()
+
+
[email protected]('kind', ['scalar', 'array', 'map'])
+def
test_blob_row_stream_error_preserves_prepared_files_and_exception(tmp_path,
kind):
+ table = _table(tmp_path, kind)
+ builder = table.new_stream_write_builder()
+ writer = builder.new_write()
+ error = OSError('original Python stream failure')
+ bad = StreamBlob(b'bad', error)
+ try:
+ writer.write_row(GenericRow([0, 'a', _value(kind, b'prepared')],
table.fields))
+ messages = writer.prepare_commit(1)
+ prepared = {path for path in tmp_path.rglob('*') if path.is_file() and
path.suffix in ('.parquet', '.blob')}
+ with patch.object(writer, '_switch_to_python',
side_effect=AssertionError('Python writer fallback')):
+ with pytest.raises(OSError) as caught:
+ writer.write_row(GenericRow([1, 'a', _value(kind, bad)],
table.fields))
+ assert caught.value is error
+ assert writer._blob_rows.readers == {}
+ writer.abort()
+ assert all(path.exists() for path in prepared)
+ _commit(builder, messages, True, 1)
+ assert _read(table) == [dict(id=0, pt='a', payload=_value(kind,
b'prepared'))]
+ finally:
+ writer.close()
+
+
+def
test_blob_object_stream_and_native_file_descriptor_share_one_writer(tmp_path):
+ table = _table(tmp_path, 'scalar')
+ source = tmp_path / 'source'
+ source.write_bytes(b'prefixREFERENCEDsuffix')
+ reference = BlobDescriptor(source.as_uri(), 6, 10).serialize()
+ blob = StreamBlob(b'OBJECT')
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.write_row(GenericRow([0, 'a', blob], table.fields))
+ writer.write_row(GenericRow([1, 'a', reference], table.fields))
+ assert writer._blob_rows.readers == {}
+ _commit(builder, writer.prepare_commit(), False)
+ assert _read(table) == [dict(id=0, pt='a', payload=b'OBJECT'),
+ dict(id=1, pt='a', payload=b'REFERENCED')]
+ finally:
+ writer.close()
+
+
[email protected]('kind', ['scalar', 'array', 'map'])
[email protected]('stream', [False, True])
+def
test_row_upsert_matches_and_deduplicates_before_opening_blob_streams(tmp_path,
kind, stream):
+ from pypaimon.tests.native_blob_update_test import _native_only
+ table = _table(tmp_path, kind)
+ seed_builder = table.new_batch_write_builder()
+ seed = seed_builder.new_write()
+ try:
+ seed.write_arrow(pa.Table.from_pylist([
+ dict(id=0, pt='a', payload=_value(kind, b'old')),
+ dict(id=1, pt='a', payload=_value(kind, b'keep')),
+ ], schema=PyarrowFieldParser.from_paimon_schema(table.fields)))
+ _commit(seed_builder, seed.prepare_commit(), False)
+ finally:
+ seed.close()
+ matched = StreamBlob(b'MATCHED')
+ appended = StreamBlob(b'APPENDED')
+ shadowed = StreamBlob(b'bad', AssertionError('Shadowed upsert source
opened'))
+ source = [GenericRow([0, 'a', _value(kind, shadowed)], table.fields),
+ GenericRow([2, 'b', _value(kind, shadowed)], table.fields),
+ GenericRow([0, 'a', _value(kind, matched)], table.fields),
+ GenericRow([2, 'b', _value(kind, appended)], table.fields)]
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ with _native_only():
+ update = builder.new_update().with_update_type(['payload'])
+ messages = update.upsert_by_key(source, ['id'], 7) if stream else
update.upsert_by_key(source, ['id'])
+ assert shadowed.opened == 0
+ assert matched.opened == appended.opened == 1
+ assert all(source.closed for blob in (matched, appended) for source in
blob.streams)
+ _commit(builder, messages, stream)
+ assert _read(table) == [dict(id=0, pt='a', payload=_value(kind,
b'MATCHED')),
+ dict(id=1, pt='a', payload=_value(kind, b'keep')),
+ dict(id=2, pt='b', payload=_value(kind,
b'APPENDED'))]
+
+
[email protected]('kind', ['scalar', 'array', 'map'])
+def test_row_upsert_source_error_is_not_retried_through_python(tmp_path, kind):
+ from pypaimon.tests.native_blob_update_test import _native_only
+ table = _table(tmp_path, kind)
+ bad = StreamBlob(b'bad', OSError('upsert Blob source failed'))
+ builder = table.new_batch_write_builder()
+ with _native_only(), pytest.raises(OSError) as caught:
+ builder.new_update().upsert_by_key([GenericRow([0, 'a', _value(kind,
bad)], table.fields)], ['id'])
+ assert caught.value is bad.error
+ assert bad.opened == 1
+ assert table.snapshot_manager().get_latest_snapshot() is None
+ assert not list(tmp_path.rglob('*.blob'))
+
+
[email protected]('kind', ['scalar', 'array', 'map'])
+def
test_blob_row_missing_partition_is_rejected_without_opening_streams(tmp_path,
kind):
+ table = _table(tmp_path, kind)
+ blob = StreamBlob(b'not opened')
+ writer = table.new_batch_write_builder().new_write()
+ try:
+ with pytest.raises(ValueError, match='pt'):
+ writer.write_row(GenericRow([0, _value(kind, blob)],
[table.fields[0], table.fields[2]]))
+ assert blob.opened == 0
+ assert not list(tmp_path.rglob('*.blob'))
+ writer.write_row(GenericRow([0, 'a', _value(kind, blob)],
table.fields))
+ assert blob.opened == 1
+ finally:
+ writer.close()
+
+
[email protected]('operation', ['write', 'upsert'])
[email protected]('entry', ['ab', b'ab', None, {'a': b'value'}])
+def test_blob_map_rows_reject_non_pair_entries(tmp_path, operation, entry):
+ table = _table(tmp_path, 'map')
+ blob = StreamBlob(b'not opened')
+ row = GenericRow([0, 'a', [('valid', blob), entry]], table.fields)
+ builder = table.new_batch_write_builder()
+ with pytest.raises(ValueError, match='key/value pairs'):
+ if operation == 'upsert':
+ builder.new_update().upsert_by_key([row], ['id'])
+ else:
+ writer = builder.new_write()
+ try:
+ writer.write_row(row)
+ finally:
+ writer.close()
+ assert blob.opened == 0
+ assert not list(tmp_path.rglob('*.blob'))
+
+
[email protected]('operation', ['write', 'upsert'])
[email protected]('type_name, key, expected', [
+ ('DECIMAL(10,2)', '12.345', '12.35'),
+ ('DECIMAL(10,2)', '-12.345', '-12.35'),
+ ('DECIMAL(38,18)', '12345678901234567890.1234567890123456785',
+ '12345678901234567890.123456789012345679'),
+ ('DECIMAL(10,2)', '0.004999999999999999999999999999999999999999999',
'0.00'),
+])
+def test_blob_map_row_decimal_keys_round_in_core(tmp_path, operation,
type_name, key, expected):
+ from pypaimon.tests.native_blob_update_test import _native_only
+ from pypaimon.read.reader.format_blob_reader import FormatBlobReader
+ table = _table(tmp_path, 'map', payload_type=MapType(True,
AtomicType(type_name), AtomicType('BLOB')))
+ builder = table.new_batch_write_builder()
+ row = GenericRow([0, 'a', [(Decimal(key), StreamBlob(b'value'))]],
table.fields)
+ if operation == 'upsert':
+ seed = builder.new_write()
+ try:
+ seed.write_row(GenericRow([0, 'a', [(Decimal('1'), b'old')]],
table.fields))
+ _commit(builder, seed.prepare_commit(), False)
+ finally:
+ seed.close()
+ with _native_only():
+ if operation == 'upsert':
+ messages =
builder.new_update().with_update_type(['payload']).upsert_by_key([row], ['id'])
+ else:
+ writer = builder.new_write()
+ try:
+ writer.write_row(row)
+ messages = writer.prepare_commit()
+ finally:
+ writer.close()
+ _commit(builder, messages, False)
+ file = next(file for message in messages for file in message.new_files if
file.file_name.endswith('.blob'))
+ reader = FormatBlobReader(table.file_io, file.file_path, ['payload'],
[table.fields[2]], None, False)
+ try:
+ result = reader.read_values_at([0])[0]
+ assert list(result) == [Decimal(expected)]
+ assert result[Decimal(expected)].to_data() == b'value'
+ finally:
+ reader.close()
+
+
[email protected]('option', ['blob-descriptor-field',
'blob.stored-descriptor-fields', 'blob-view-field'])
[email protected]('stream', [False, True])
+def test_inline_blob_row_objects_keep_real_references(tmp_path,
native_rest_catalog, option, stream):
+ from pypaimon.tests.native_blob_update_test import _inline_table, _read as
read_inline
+ table, _, references = _inline_table(tmp_path, option,
catalog=native_rest_catalog)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer = builder.new_write()
+ reference = references[0]
+ value = (Blob.from_view(BlobViewStruct.deserialize(reference)) if option
== 'blob-view-field'
+ else Blob.from_descriptor_bytes(reference, table.file_io))
+ try:
+ with patch.object(writer, '_switch_to_python',
side_effect=AssertionError('Python writer fallback')):
+ with patch.object(value, 'new_input_stream',
side_effect=AssertionError('Inline reference opened')):
+ writer.write_row(GenericRow([4, 'a', 14, value], table.fields))
+ assert writer._blob_rows.readers == {}
+ messages = writer.prepare_commit(7) if stream else
writer.prepare_commit()
+ assert all(not file.file_name.endswith('.blob') for message in
messages for file in message.new_files)
+ _commit(builder, messages, stream)
+ rows = read_inline(table, descriptors=True, resolve_views=False)
+ assert rows[-1] == dict(id=4, pt='a', value=14, payload=reference)
+ finally:
+ writer.close()
+
+
[email protected]('key', ['99999999.995', '-99999999.995'])
+def
test_decimal_blob_map_overflow_is_rejected_before_opening_payload(tmp_path,
key):
+ table = _table(tmp_path, 'map', payload_type=MapType(True,
AtomicType('DECIMAL(10,2)'), AtomicType('BLOB')))
+ blob = StreamBlob(b'not opened')
+ writer = table.new_batch_write_builder().new_write()
+ try:
+ with pytest.raises(ValueError, match='precision'):
+ writer.write_row(GenericRow([0, 'a', [(Decimal(key), blob)]],
table.fields))
+ assert blob.opened == 0
+ assert writer._blob_rows.readers == {}
+ assert not list(tmp_path.rglob('*.blob'))
+ finally:
+ writer.close()
+
+
+def test_decimal_blob_map_duplicate_keys_are_checked_after_rounding(tmp_path):
+ table = _table(tmp_path, 'map', payload_type=MapType(True,
AtomicType('DECIMAL(10,2)'), AtomicType('BLOB')))
+ blob = StreamBlob(b'not opened')
+ writer = table.new_batch_write_builder().new_write()
+ try:
+ with pytest.raises(ValueError, match='unique'):
+ writer.write_row(GenericRow([0, 'a', [(Decimal('12.345'), blob),
(Decimal('12.35'), blob)]], table.fields))
+ assert blob.opened == 0
+ assert writer._blob_rows.readers == {}
+ assert not list(tmp_path.rglob('*.blob'))
+ finally:
+ writer.close()
diff --git a/paimon-python/pypaimon/tests/native_blob_update_test.py
b/paimon-python/pypaimon/tests/native_blob_update_test.py
index 892f7936df..6b263bee47 100644
--- a/paimon-python/pypaimon/tests/native_blob_update_test.py
+++ b/paimon-python/pypaimon/tests/native_blob_update_test.py
@@ -615,7 +615,7 @@ def
test_native_row_blob_map_rejects_duplicate_null_keys_before_record_write(tmp
assert set(tmp_path.rglob('*.blob')) == paths
-def test_raw_blob_row_upsert_keeps_streams_lazy_on_python_path(tmp_path):
+def test_raw_blob_row_upsert_keeps_streams_lazy_in_core(tmp_path):
from pypaimon.tests.blob_table_test import _StreamingOnlyBlob
table, _ = _table(tmp_path)
first, shadowed, surviving = (_StreamingOnlyBlob(value) for value in
(b'first', b'shadow', b'last'))
@@ -623,7 +623,8 @@ def
test_raw_blob_row_upsert_keeps_streams_lazy_on_python_path(tmp_path):
GenericRow([4, 'b', 400, shadowed], table.fields),
GenericRow([4, 'b', 401, surviving], table.fields)]
builder = table.new_batch_write_builder()
- messages =
builder.new_update().with_update_type(['payload']).upsert_by_key(source, ['id'])
+ with _native_only():
+ messages =
builder.new_update().with_update_type(['payload']).upsert_by_key(source, ['id'])
_commit(builder, messages, False)
assert first.opened and surviving.opened and not shadowed.opened
assert [row['payload'] for row in _read(table)] == [b'first', None, b'',
b'last', b'last']
diff --git a/paimon-python/pypaimon/tests/native_blob_uri_reader_test.py
b/paimon-python/pypaimon/tests/native_blob_uri_reader_test.py
index 9d6ece1f34..8233bedf94 100644
--- a/paimon-python/pypaimon/tests/native_blob_uri_reader_test.py
+++ b/paimon-python/pypaimon/tests/native_blob_uri_reader_test.py
@@ -103,6 +103,99 @@ def test_custom_uri_reader_keeps_native_writer(tmp_path):
commit.close()
[email protected]('stream', [False, True])
[email protected]('row_input', [False, True])
+def
test_scoped_application_factory_preserves_native_and_live_row_sources(tmp_path,
stream, row_input):
+ from pypaimon.table.row.generic_row import GenericRow
+ from pypaimon.tests.native_blob_row_write_test import StreamBlob
+ table = _table(tmp_path, {'blob.copy-buffer-size': '3 B'})
+ source = tmp_path / 'source'
+ source.write_bytes(b'prefixFILEIO!suffix')
+ native_reference = BlobDescriptor(source.as_uri(), 6, 7).serialize()
+
+ class ScopedFactory(Factory):
+ def __init__(self):
+ super().__init__()
+ self.selected = []
+ self.created = []
+
+ def _supports_uri(self, uri):
+ self.selected.append(uri)
+ return uri.startswith('custom://')
+
+ def create(self, uri):
+ self.created.append(uri)
+ return super().create(uri)
+
+ factory = ScopedFactory()
+ blob = StreamBlob(b'OBJECT')
+ values = [[0, blob if row_input else b'BYTES', native_reference],
+ [1, BlobDescriptor('custom://source', 6, 7).serialize(), None]]
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ try:
+ writer.with_blob_uri_reader_factory(factory)
+ if row_input:
+ for value in values:
+ writer.write_row(GenericRow(value, table.fields))
+ assert blob.opened == 1 and all(source.closed for source in
blob.streams)
+ else:
+ writer.write_arrow(pa.Table.from_pylist([
+ dict(zip(['id', 'large', 'small'], value)) for value in
values],
+ schema=PyarrowFieldParser.from_paimon_schema(table.fields)))
+ messages = writer.prepare_commit(7) if stream else
writer.prepare_commit()
+ commit.commit(messages, 7) if stream else commit.commit(messages)
+ assert writer._python_writer is None
+ assert sorted(factory.selected) == sorted([source.as_uri(),
'custom://source'])
+ assert factory.created == ['custom://source']
+ assert _read(table, True, True) == [
+ dict(id=0, large=b'OBJECT' if row_input else b'BYTES',
small=b'FILEIO!'),
+ dict(id=1, large=b'PAYLOAD', small=None),
+ ]
+ finally:
+ writer.close()
+ commit.close()
+
+
[email protected]('stream', [False, True])
[email protected]('row_input', [False, True])
+def
test_application_factory_selection_errors_do_not_fall_back_to_native_io(tmp_path,
stream, row_input):
+ from pypaimon.table.row.generic_row import GenericRow
+ table = _table(tmp_path)
+ source = tmp_path / 'source'
+ source.write_bytes(b'native readable')
+ reference = BlobDescriptor(source.as_uri(), 0, -1).serialize()
+ error = OSError('application selector failed')
+ selected = []
+
+ class FailedSelection:
+ def _supports_uri(self, uri):
+ selected.append(uri)
+ raise error
+
+ def create(self, uri):
+ pytest.fail('A selection error must not select any reader')
+
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ writer.with_blob_uri_reader_factory(FailedSelection())
+ with pytest.raises(OSError) as caught:
+ if row_input:
+ writer.write_row(GenericRow([0, reference, None],
table.fields))
+ else:
+ writer.write_arrow(pa.Table.from_pylist([dict(id=0,
large=reference, small=None)],
+
schema=PyarrowFieldParser.from_paimon_schema(table.fields)))
+ assert caught.value is error
+ assert selected == [source.as_uri()]
+ assert writer._python_writer is None
+ assert writer._blob_rows.readers == {}
+ finally:
+ writer.close()
+ assert not list(tmp_path.rglob('*.blob'))
+ assert not list(tmp_path.rglob('*.parquet'))
+
+
@pytest.mark.parametrize('native', [False, True])
@pytest.mark.parametrize('stream', [False, True])
def test_custom_readers_copy_scalar_and_collection_windows(tmp_path, native,
stream):
@@ -195,7 +288,7 @@ def
test_custom_reader_exceptions_are_not_retried_and_streams_close(tmp_path, op
assert not list(tmp_path.rglob('*.parquet'))
-def test_reader_factory_replacement_clear_and_row_fallback(tmp_path):
+def test_reader_factory_replacement_clear_and_native_rows(tmp_path):
from pypaimon.table.row.generic_row import GenericRow
table = _table(tmp_path)
factory = Factory()
@@ -220,6 +313,7 @@ def
test_reader_factory_replacement_clear_and_row_fallback(tmp_path):
writer.with_blob_uri_reader_factory(factory)
writer.write_row(GenericRow([
2, BlobDescriptor('custom://source', 6, 7).serialize(), None],
table.fields))
+ assert writer._python_writer is None
commit.commit(writer.prepare_commit())
assert factory.opened == factory.closed == ['custom://source']
assert _read(table, True, True) == [{'id': 2, 'large': b'PAYLOAD',
'small': None}]
diff --git a/paimon-python/pypaimon/tests/native_blob_write_test.py
b/paimon-python/pypaimon/tests/native_blob_write_test.py
index 9bdfd24fe1..3b738ce237 100644
--- a/paimon-python/pypaimon/tests/native_blob_write_test.py
+++ b/paimon-python/pypaimon/tests/native_blob_write_test.py
@@ -272,7 +272,7 @@ class _StreamingBlob(Blob):
return io.BytesIO(self.data)
-def test_blob_object_selects_python_before_writing(tmp_path):
+def test_blob_object_rows_keep_native_writer(tmp_path):
table = _table(tmp_path)
builder = table.new_batch_write_builder()
writer, commit = builder.new_write(), builder.new_commit()
@@ -282,7 +282,7 @@ def
test_blob_object_selects_python_before_writing(tmp_path):
writer.write_row(GenericRow([0, b'bytes', None], table.fields))
writer.write_row(GenericRow([1, blob, None], table.fields))
writer.write_arrow(_data(2))
- assert writer._python_writer is not None
+ assert writer._python_writer is None
commit.commit(writer.prepare_commit())
assert blob.opened
expected = [{'id': 0, 'large': b'bytes', 'small': None},
@@ -293,18 +293,19 @@ def
test_blob_object_selects_python_before_writing(tmp_path):
commit.close()
-def test_blob_object_cannot_switch_after_native_write(tmp_path):
+def test_blob_object_rows_mix_with_native_arrow_writes(tmp_path):
table = _table(tmp_path)
builder = table.new_batch_write_builder()
writer, commit = builder.new_write(), builder.new_commit()
blob = _StreamingBlob(b'stream')
try:
writer.write_arrow(_data())
- with pytest.raises(RuntimeError, match='after native data was
written'):
- writer.write_row(GenericRow([2, blob, None], table.fields))
- assert not blob.opened
+ writer.write_row(GenericRow([2, blob, None], table.fields))
+ assert blob.opened
+ assert writer._python_writer is None
commit.commit(writer.prepare_commit())
- assert _read(table, True, True) == _data().to_pylist()
+ assert _read(table, True, True) == _data().to_pylist() + [
+ {'id': 2, 'large': b'stream', 'small': None}]
finally:
writer.close()
commit.close()
diff --git a/paimon-python/pypaimon/write/native_blob_rows.py
b/paimon-python/pypaimon/write/native_blob_rows.py
index 463f940789..7505e82272 100644
--- a/paimon-python/pypaimon/write/native_blob_rows.py
+++ b/paimon-python/pypaimon/write/native_blob_rows.py
@@ -18,11 +18,14 @@
"""Transport row Blob streams into the core URI copy path."""
from collections.abc import Mapping
+from decimal import Decimal
from uuid import uuid4
-from pypaimon.common.uri_reader import UriReaderFactory
-from pypaimon.schema.data_types import ArrayType, MapType
+import pyarrow as pa
+
+from pypaimon.schema.data_types import ArrayType, MapType, PyarrowFieldParser,
is_blob_file_field
from pypaimon.table.row.blob import Blob, BlobDescriptor
+from pypaimon.write.row_utils import inline_blob_value, value_for_arrow
class _BlobRowReader:
@@ -42,15 +45,67 @@ class NativeBlobRows:
the core cannot reuse an unrelated Python stream.
"""
- def __init__(self, file_io):
+ def __init__(self, fallback=None):
self.readers = {}
- self.fallback = UriReaderFactory.from_file_io(file_io)
+ self.fallback = fallback
+
+ def _supports_uri(self, uri):
+ # Ordinary descriptors retain the core FileIO/HTTP path. A configured
+ # application factory owns other URIs and its errors must propagate.
+ if uri in self.readers:
+ return True
+ if self.fallback is None:
+ return False
+ if hasattr(self.fallback, '_supports_uri'):
+ return self.fallback._supports_uri(uri)
+ return True
def create(self, uri):
if uri in self.readers:
return self.readers[uri]
return self.fallback.create(uri)
+ def to_batch(self, table, rows, names):
+ """Transport existing row values; core owns physical input
validation."""
+ descriptor_fields = table.options.blob_descriptor_fields()
+ view_fields = table.options.blob_view_fields()
+ fields = []
+ columns = []
+ for name in names:
+ field = table.field_dict[name]
+ arrow_field = PyarrowFieldParser.from_paimon_field(field)
+ inline = name in descriptor_fields or name in view_fields
+ if is_blob_file_field(field) and not inline:
+ if isinstance(field.type, MapType) and not
table.is_primary_key_table:
+ entries = pa.struct([
+ pa.field('key', arrow_field.type.key_type),
+ arrow_field.type.item_field.with_name('value'),
+ ])
+ arrow_field = pa.field(name, pa.list_(entries),
nullable=arrow_field.nullable)
+ values = [self.encode(row[name], field.type) for row in rows]
+ if (isinstance(field.type, MapType) and not
table.is_primary_key_table
+ and
pa.types.is_decimal(arrow_field.type.value_type[0].type)):
+ # Preserve arbitrary source scale. The core rounds and
checks
+ # precision before serializing the declared decimal key.
+ def key_text(key):
+ if key is None:
+ return None
+ if not isinstance(key, Decimal):
+ raise ValueError('MAP<X, BLOB> DECIMAL key must be
a decimal.Decimal.')
+ return format(key, 'f')
+ values = [None if value is None else [(key_text(key),
item) for key, item in value]
+ for value in values]
+ entries = arrow_field.type.value_type
+ entries = pa.struct([entries[0].with_type(pa.string()),
entries[1]])
+ arrow_field = arrow_field.with_type(pa.list_(entries))
+ else:
+ values = [value_for_arrow(inline_blob_value(
+ row[name], name in descriptor_fields, name in
view_fields), field)
+ for row in rows]
+ fields.append(arrow_field)
+ columns.append(pa.array(values, type=arrow_field.type))
+ return pa.RecordBatch.from_arrays(columns, schema=pa.schema(fields))
+
def encode(self, value, data_type):
if hasattr(value, 'as_py'):
value = value.as_py()
@@ -61,8 +116,15 @@ class NativeBlobRows:
raise ValueError('ARRAY<BLOB> field value must be a list or
tuple.')
return [self.encode(item, data_type.element) for item in value]
if isinstance(data_type, MapType):
+ if isinstance(value, (bytes, bytearray, str)) or not
hasattr(value, '__iter__'):
+ raise ValueError('MAP<X, BLOB> field value must be a mapping
or key/value pairs')
entries = value.items() if isinstance(value, Mapping) else value
- return [(key, self.encode(item, data_type.value)) for key, item in
entries]
+ result = []
+ for entry in entries:
+ if not isinstance(entry, (list, tuple)) or len(entry) != 2:
+ raise ValueError('MAP<X, BLOB> entries must be key/value
pairs')
+ result.append((entry[0], self.encode(entry[1],
data_type.value)))
+ return result
if isinstance(value, Blob):
if any(value is placeholder for placeholder in (
Blob.PLACE_HOLDER, Blob.ARRAY_PLACE_HOLDER,
Blob.MAP_PLACE_HOLDER)):
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index 3f5231a1b8..1abf1ba4f9 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -19,7 +19,7 @@
import pyarrow as pa
-from pypaimon.schema.data_types import MapType, PyarrowFieldParser
+from pypaimon.schema.data_types import MapType, PyarrowFieldParser,
is_blob_file_field
from pypaimon.common.options.core_options import ChangelogProducer
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.snapshot.time_travel_util import SCAN_KEYS
@@ -103,7 +103,7 @@ def _native_update_columns_supported(table, columns):
return not names.intersection(table.options.video_frame_fields())
-def _upsert_row_batches(rows, fields, schema):
+def _upsert_row_batches(rows, fields, schema, table=None, blob_rows=None):
"""Encode consecutive row shapes without padding absent fields with NULL.
Keep source order, including duplicates across shapes. Key matching,
@@ -120,10 +120,13 @@ def _upsert_row_batches(rows, fields, schema):
end += 1
selected = [field for field in fields if field.name in names]
batch_schema = pa.schema([schema.field(field.name) for field in
selected])
- batches.append(pa.RecordBatch.from_pydict({
- field.name: [value_for_arrow(row[field.name], field) for row in
rows[start:end]]
- for field in selected
- }, schema=batch_schema))
+ if blob_rows is not None:
+ batches.append(blob_rows.to_batch(table, rows[start:end],
batch_schema.names))
+ else:
+ batches.append(pa.RecordBatch.from_pydict({
+ field.name: [value_for_arrow(row[field.name], field) for row
in rows[start:end]]
+ for field in selected
+ }, schema=batch_schema))
start = end
return batches
@@ -139,16 +142,18 @@ def create_native_upsert(table, commit_user, data, keys,
columns):
return None
fields = table.table_schema.fields
schema = PyarrowFieldParser.from_paimon_schema(fields)
+ blob_rows = None
if any(not _supported_upsert_key_type(schema.field(key).type) for key in
set(keys + table.partition_keys)):
return None
if not isinstance(data, pa.Table):
if not columns or not data:
return None
- # Row upserts can append custom Blob streams. Select the row-aware
- # writer before matching or opening any source, including shadowed
rows.
- if any(_contains_blob_value(value) for row in data for value in
row.values()):
- return None
- data = _upsert_row_batches(data, fields, schema)
+ if any(_contains_blob_value(value) for row in data for value in
row.values()) or any(
+ is_blob_file_field(field) and isinstance(field.type, MapType)
and field.name in row
+ for field in fields for row in data):
+ from pypaimon.write.native_blob_rows import NativeBlobRows
+ blob_rows = NativeBlobRows()
+ data = _upsert_row_batches(data, fields, schema, table, blob_rows)
elif (len(data.column_names) != len(set(data.column_names))
or not set(data.column_names) <= set(schema.names)
or any(data.schema.field(name).type != schema.field(name).type
@@ -158,7 +163,9 @@ def create_native_upsert(table, commit_user, data, keys,
columns):
._with_commit_user(commit_user)
.new_update()
.with_update_type(columns))
- return NativeTableUpsert(table, writer, keys, data)
+ if blob_rows is not None:
+ writer._with_blob_uri_reader_factory(blob_rows)
+ return NativeTableUpsert(table, writer, keys, data, blob_rows)
def create_native_predicate_update(table, commit_user, predicate,
columns=None):
@@ -280,7 +287,7 @@ class NativeTableUpdateByRowId(TableUpdateByRowId):
if not blob_object_columns:
return self.update_columns(data, column_names)
from pypaimon.write.native_blob_rows import NativeBlobRows
- blobs = NativeBlobRows(self.table.file_io)
+ blobs = NativeBlobRows()
for name, values in blob_object_columns.items():
field = self.table.field_dict[name]
arrow_field = PyarrowFieldParser.from_paimon_field(field)
@@ -302,16 +309,21 @@ class NativeTableUpdateByRowId(TableUpdateByRowId):
class NativeTableUpsert:
"""Submit Arrow tables or row-shape batches to the core Rust upsert
writer."""
- def __init__(self, table, writer, keys, data):
+ def __init__(self, table, writer, keys, data, blob_rows=None):
self.table = table
self.writer = writer
self.keys = keys
self.data = data
+ self.blob_rows = blob_rows
def upsert(self):
- return from_native_commit_messages(
- self.table,
- self.writer.upsert_by_arrow_with_key(self.data, self.keys))
+ try:
+ return from_native_commit_messages(
+ self.table,
+ self.writer.upsert_by_arrow_with_key(self.data, self.keys))
+ finally:
+ if self.blob_rows is not None:
+ self.blob_rows.readers.clear()
class NativePredicateTableUpdate:
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index 0a8d0f136f..9a353f60a9 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -27,7 +27,7 @@ from pypaimon.table.bucket_mode import BucketMode
from pypaimon.write.native_commit import (
create_native_write_table, from_native_commit_messages,
)
-from pypaimon.write.row_utils import row_to_named_values,
row_values_to_arrow_table
+from pypaimon.write.row_utils import require_columns, row_to_named_values,
row_values_to_arrow_table
def native_write_available() -> bool:
@@ -112,6 +112,11 @@ class NativeTableWrite:
self._blob_consumer = None
self._blob_uri_reader_factory = None
self._restore_snapshot_id = restore_snapshot_id
+ self._blob_rows = None
+ if any(is_blob_file_field(field) for field in
table.table_schema.fields):
+ from pypaimon.write.native_blob_rows import NativeBlobRows
+ self._blob_rows = NativeBlobRows()
+ self._native_writer.with_blob_uri_reader_factory(self._blob_rows)
def _switch_to_python(self):
if self._python_writer is not None:
@@ -189,7 +194,12 @@ class NativeTableWrite:
return
self._python_writer.with_blob_uri_reader_factory(uri_reader_factory)
if self._written:
raise RuntimeError('with_blob_uri_reader_factory must be called
before any write operation.')
- self._native_writer.with_blob_uri_reader_factory(uri_reader_factory)
+ if self._blob_rows is not None:
+ if uri_reader_factory is not None and not
callable(getattr(uri_reader_factory, 'create', None)):
+ raise TypeError('URI reader factory must provide callable
create')
+ self._blob_rows.fallback = uri_reader_factory
+ else:
+
self._native_writer.with_blob_uri_reader_factory(uri_reader_factory)
self._blob_uri_reader_factory = uri_reader_factory
return self
@@ -239,17 +249,24 @@ class NativeTableWrite:
def write_row(self, row):
if self._python_writer is not None:
return self._python_writer.write_row(row)
- if any(is_blob_file_field(field) and field.name in self._schema.names
- for field in self.table.table_schema.fields):
- # Select the row-aware writer from the first row, including byte
- # values: later rows may provide custom Blob streams or URI
readers.
- return self._switch_to_python().write_row(row)
values = row_to_named_values(row, self.table.table_schema.fields)
names = self._schema.names
- data = self._prepare_native_arrow_data(row_values_to_arrow_table(
- values, self.table.table_schema.fields, names).to_batches()[0])
- from pypaimon.write.row_kind import with_row_kind
- self._write_native_batch(with_row_kind(self.table, data, row))
+ require_columns(values, names, 'write_row')
+ require_columns(values, self.table.partition_keys, 'write_row')
+ from pypaimon.write.row_kind import skip_write_row, with_row_kind
+ if skip_write_row(self.table, values, row.get_row_kind()):
+ return
+ try:
+ data = (self._blob_rows.to_batch(self.table, [values], names)
+ if self._blob_rows is not None else
+ self._prepare_native_arrow_data(row_values_to_arrow_table(
+ values, self.table.table_schema.fields,
names).to_batches()[0]))
+ self._write_native_batch(with_row_kind(self.table, data, row))
+ finally:
+ # Core externalizes Blob values during write, before buffering PK
+ # records. Their live Python objects need not survive the call.
+ if self._blob_rows is not None:
+ self._blob_rows.readers.clear()
def prepare_commit(self, commit_identifier=None):
if self._python_writer is not None:
@@ -272,6 +289,8 @@ class NativeTableWrite:
elif self._native_writer is not None:
self._native_writer.close()
self._native_writer = None
+ if self._blob_rows is not None:
+ self._blob_rows.readers.clear()
def abort(self):
if self._python_writer is not None:
