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 d631383adc [Python] Enable Native Data Evolution ROW sidecars (#10404)
d631383adc is described below
commit d631383adcd5b97a13dd3c2ee57837ac460b00d4
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 6 21:28:56 2026 +0800
[Python] Enable Native Data Evolution ROW sidecars (#10404)
---
.../read/reader/blob_descriptor_convert_reader.py | 33 +-
.../pypaimon/read/reader/concat_batch_reader.py | 13 +-
.../read/reader/nested_leaf_batch_reader.py | 7 +-
paimon-python/pypaimon/read/split_read.py | 16 +-
paimon-python/pypaimon/read/table_read.py | 4 -
.../pypaimon/tests/concat_batch_reader_test.py | 51 +++
.../pypaimon/tests/multimodal_lerobot_test.py | 4 +-
.../pypaimon/tests/native_row_sidecar_test.py | 369 +++++++++++++++++++++
paimon-python/pypaimon/tests/native_update_test.py | 7 +-
paimon-python/pypaimon/utils/arrow_utils.py | 28 ++
paimon-python/pypaimon/write/native_commit.py | 2 +
paimon-python/pypaimon/write/native_update.py | 1 -
paimon-python/pypaimon/write/native_write.py | 3 -
13 files changed, 519 insertions(+), 19 deletions(-)
diff --git
a/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
b/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
index 1d671f470a..c9c99ba801 100644
--- a/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
+++ b/paimon-python/pypaimon/read/reader/blob_descriptor_convert_reader.py
@@ -23,6 +23,28 @@ from pyarrow import RecordBatch
from pypaimon.common.options.core_options import CoreOptions
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
from pypaimon.table.row.blob import Blob, BlobViewStruct
+from pypaimon.utils.arrow_utils import as_blob_data, clear_blob_data,
is_blob_data
+
+
+class BlobDataBatchReader(RecordBatchReader):
+ """Keep ROW payload provenance after file schema and row-id
normalization."""
+
+ def __init__(self, inner, field_names):
+ self._inner = inner
+ self._field_names = field_names
+ self._adopt_metadata(inner)
+
+ def read_arrow_batch(self):
+ batch = self._inner.read_arrow_batch()
+ if batch is None:
+ return None
+ fields = [as_blob_data(field) if field.name in self._field_names else
field
+ for field in batch.schema]
+ return RecordBatch.from_arrays(
+ batch.columns, schema=pyarrow.schema(fields,
metadata=batch.schema.metadata))
+
+ def close(self):
+ self._inner.close()
class BlobInlineConvertReader(RecordBatchReader):
@@ -93,7 +115,7 @@ class BlobInlineConvertReader(RecordBatchReader):
and not self._blob_as_descriptor):
batch, view_blobs = self._resolve_view_fields(batch,
self._blob_view_lookup)
# Resolve BlobDescriptor -> real bytes (if blob-as-descriptor=false)
- return self._resolve_descriptor_fields(batch, view_blobs)
+ return clear_blob_data(self._resolve_descriptor_fields(batch,
view_blobs))
# ------------------------------------------------------------------
# Stage 1: BlobView prescan (lightweight, only reads view columns)
@@ -114,7 +136,8 @@ class BlobInlineConvertReader(RecordBatchReader):
if batch is None:
break
for field_name in self._view_fields:
- if field_name not in batch.schema.names:
+ if (field_name not in batch.schema.names
+ or is_blob_data(batch.schema.field(field_name))):
continue
for value in batch.column(field_name).to_pylist():
value = self._normalize_blob_to_bytes(value)
@@ -143,7 +166,8 @@ class BlobInlineConvertReader(RecordBatchReader):
"""Replace BlobViewStruct bytes in view fields with descriptor
bytes."""
view_blobs = {}
for field_name in self._view_fields:
- if field_name not in batch.schema.names:
+ if (field_name not in batch.schema.names
+ or is_blob_data(batch.schema.field(field_name))):
continue
values = [self._normalize_blob_to_bytes(v) for v in
batch.column(field_name).to_pylist()]
converted_values = []
@@ -182,7 +206,8 @@ class BlobInlineConvertReader(RecordBatchReader):
return batch
for field_name in self._descriptor_fields:
- if field_name not in batch.schema.names:
+ if (field_name not in batch.schema.names
+ or is_blob_data(batch.schema.field(field_name))):
continue
values = [self._normalize_blob_to_bytes(v) for v in
batch.column(field_name).to_pylist()]
blobs = [
diff --git a/paimon-python/pypaimon/read/reader/concat_batch_reader.py
b/paimon-python/pypaimon/read/reader/concat_batch_reader.py
index da0cee3add..30aa021004 100644
--- a/paimon-python/pypaimon/read/reader/concat_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/concat_batch_reader.py
@@ -23,7 +23,7 @@ import pyarrow as pa
from pyarrow import RecordBatch
from pypaimon.manifest.schema.data_file_meta import DataFileMeta
-from pypaimon.utils.arrow_utils import zero_column_batch
+from pypaimon.utils.arrow_utils import as_blob_data, is_blob_data,
zero_column_batch
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
from pypaimon.schema.data_types import DataField, PyarrowFieldParser
from pypaimon.table.row.blob import Blob
@@ -170,6 +170,11 @@ class MergeAllBatchReader(RecordBatchReader):
if batch.num_rows == 0:
continue
+ # Do not combine reference and materialized-payload batches: one
+ # schema marker cannot describe two different BLOB representations.
+ if batches and not batch.schema.equals(batches[0].schema,
check_metadata=True):
+ self._remainder = batch
+ break
take = min(batch.num_rows, self._batch_size - num_rows)
piece = batch.slice(0, take)
piece_usage = _batch_offset_usage(piece)
@@ -270,11 +275,14 @@ class DataEvolutionMergeReader(RecordBatchReader):
return None
columns = []
+ fields = list(self.schema)
for i in range(len(self.row_offsets)):
batch_index = self.row_offsets[i]
field_index = self.field_offsets[i]
if batch_index >= 0 and batches[batch_index] is not None:
columns.append(batches[batch_index].column(field_index).slice(0, min_rows))
+ if
is_blob_data(batches[batch_index].schema.field(field_index)):
+ fields[i] = as_blob_data(fields[i])
else:
columns.append(pa.nulls(min_rows,
type=self.schema.field(i).type))
@@ -282,7 +290,8 @@ class DataEvolutionMergeReader(RecordBatchReader):
if batches[i] is not None and batches[i].num_rows > min_rows:
self._buffers[i] = batches[i].slice(min_rows,
batches[i].num_rows - min_rows)
- return pa.RecordBatch.from_arrays(columns, schema=self.schema)
+ return pa.RecordBatch.from_arrays(
+ columns, schema=pa.schema(fields, metadata=self.schema.metadata))
def close(self) -> None:
try:
diff --git a/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
b/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
index 874fd447c8..cb4c4f81c4 100644
--- a/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
@@ -29,6 +29,7 @@ from pypaimon.read.reader.field_indices import (
blob_field_indices, descriptor_field_indices, vector_field_indices)
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
from pypaimon.schema.data_types import DataField, PyarrowFieldParser
+from pypaimon.utils.arrow_utils import as_blob_data, is_blob_data
def _struct_field(column, name):
@@ -72,6 +73,7 @@ class NestedLeafBatchReader(RecordBatchReader):
if batch is None:
return None
arrays = []
+ fields = list(self._schema)
for i, path in enumerate(self._paths):
column = batch.column(path[0])
for name in path[1:]:
@@ -84,7 +86,10 @@ class NestedLeafBatchReader(RecordBatchReader):
if column.type != target_type:
column = cast_array_for_schema_evolution(column, target_type)
arrays.append(column)
- return pa.RecordBatch.from_arrays(arrays, schema=self._schema)
+ if len(path) == 1 and is_blob_data(batch.schema.field(path[0])):
+ fields[i] = as_blob_data(fields[i])
+ return pa.RecordBatch.from_arrays(
+ arrays, schema=pa.schema(fields, metadata=self._schema.metadata))
def close(self) -> None:
self._inner.close()
diff --git a/paimon-python/pypaimon/read/split_read.py
b/paimon-python/pypaimon/read/split_read.py
index 7107692b3d..80aeb562f1 100644
--- a/paimon-python/pypaimon/read/split_read.py
+++ b/paimon-python/pypaimon/read/split_read.py
@@ -60,7 +60,8 @@ from pypaimon.read.reader.field_indices import (
descriptor_field_names_for_table, vector_field_indices)
from pypaimon.read.reader.filter_record_reader import FilterRecordReader
from pypaimon.read.reader.format_avro_reader import FormatAvroReader
-from pypaimon.read.reader.blob_descriptor_convert_reader import
BlobInlineConvertReader
+from pypaimon.read.reader.blob_descriptor_convert_reader import (
+ BlobDataBatchReader, BlobInlineConvertReader)
from pypaimon.read.reader.blob_view_read_support import (
needs_blob_inline_convert, wrap_record_reader_with_blob_inline_convert)
from pypaimon.read.reader.filter_record_batch_reader import
FilterRecordBatchReader
@@ -312,8 +313,14 @@ class SplitRead(ABC):
if len(effective_row_ranges) == 0:
return EmptyRecordBatchReader()
+ row_sidecar_selected = False
row_sidecar_file = self._row_sidecar_file_name(file)
+ descriptor_projection = (
+ CoreOptions.blob_as_descriptor(self.table.options)
+ and any(self.read_fields[index].name in read_fields
+ for index in blob_field_indices(self.read_fields)))
if (physical_row_ranges is None
+ and not descriptor_projection
and row_sidecar_file is not None
and self._should_read_row_sidecar(
file,
@@ -323,6 +330,7 @@ class SplitRead(ABC):
self.table.options.data_evolution_row_sidecar_max_selection_ratio())):
file_path = self._aligned_extra_file_path(file, row_sidecar_file)
file_format = ROW_SIDECAR_FORMAT
+ row_sidecar_selected = True
# Prepare file-local native row selection. Existing native formats
# consume row indices; Parquet keeps compact ranges to avoid expanding
@@ -549,6 +557,12 @@ class SplitRead(ABC):
file_data_fields=file_read_fields,
target_data_fields=target_fields)
+ if row_sidecar_selected:
+ inline_fields =
(CoreOptions.blob_descriptor_fields(self.table.options)
+ |
CoreOptions.blob_view_fields(self.table.options))
+ if inline_fields:
+ reader = BlobDataBatchReader(reader, inline_fields)
+
# For non-Vortex formats, wrap with RowIdFilterRecordBatchReader
if (effective_row_ranges is not None
and row_indices is None
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index e94df3f0af..2f52c1fb9c 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -803,10 +803,6 @@ class TableRead:
@staticmethod
def _native_split_files_supported(split):
for data_file in split.files:
- # Rust does not select ROW sidecars for point reads. The primary
- # file need not be available when the sidecar covers this read.
- if any(name.lower().endswith('.row') for name in
data_file.extra_files or []):
- return False
file_name = data_file.file_name.lower()
if ('.vector.' not in file_name
and not file_name.endswith(_NATIVE_READ_FILE_SUFFIXES)
diff --git a/paimon-python/pypaimon/tests/concat_batch_reader_test.py
b/paimon-python/pypaimon/tests/concat_batch_reader_test.py
index 9e16de2df7..9bcfd5e929 100644
--- a/paimon-python/pypaimon/tests/concat_batch_reader_test.py
+++ b/paimon-python/pypaimon/tests/concat_batch_reader_test.py
@@ -15,7 +15,10 @@
# specific language governing permissions and limitations
# under the License.
+import tempfile
import unittest
+from pathlib import Path
+from types import SimpleNamespace
from unittest.mock import patch
import pyarrow as pa
@@ -41,6 +44,54 @@ class _BatchReader(RecordBatchReader):
class ConcatBatchReaderTest(unittest.TestCase):
+ def test_merge_all_resolves_blob_refs_and_keeps_row_payloads(self):
+ from pypaimon.common.options import Options
+ from pypaimon.common.options.core_options import CoreOptions
+ from pypaimon.filesystem.local_file_io import LocalFileIO
+ from pypaimon.read.reader.blob_descriptor_convert_reader import (
+ BlobDataBatchReader, BlobInlineConvertReader)
+ from pypaimon.table.row.blob import BlobDescriptor
+
+ with tempfile.TemporaryDirectory() as root:
+ # Both representations can have the same descriptor-shaped bytes.
+ payload = BlobDescriptor(str(Path(root) / 'must-not-read'), 0,
1).serialize()
+ source = Path(root) / 'source'
+ source.write_bytes(payload)
+ reference = BlobDescriptor(str(source), 0,
len(payload)).serialize()
+ table = SimpleNamespace(
+ options=CoreOptions(Options({'blob-descriptor-field':
'payload'})),
+ file_io=LocalFileIO.create(),
+ catalog_environment=SimpleNamespace(catalog_loader=None))
+ schema = pa.schema([pa.field(
+ 'payload', pa.large_binary(), metadata={b'description':
b'keep'})])
+ for reverse in (False, True):
+ with self.subTest(reverse=reverse):
+ refs = _BatchReader([pa.RecordBatch.from_arrays(
+ [pa.array([reference], type=pa.large_binary())],
schema=schema)])
+ data =
BlobDataBatchReader(_BatchReader([pa.RecordBatch.from_arrays(
+ [pa.array([payload, None], type=pa.large_binary())],
schema=schema)]), {'payload'})
+ sources = [data, refs] if reverse else [refs, data]
+ merged = MergeAllBatchReader(
+ [lambda reader=reader: reader for reader in sources],
batch_size=3)
+ reader = BlobInlineConvertReader(merged, table)
+ values = []
+ sizes = []
+ try:
+ while True:
+ batch = reader.read_arrow_batch()
+ if batch is None:
+ break
+ values.extend(batch.column(0).to_pylist())
+ sizes.append(batch.num_rows)
+ metadata = batch.schema.field(0).metadata or {}
+ self.assertNotIn(b'paimon.row-sidecar.blob-data',
metadata)
+ if batch.num_rows == 2:
+ self.assertEqual(metadata, {b'description':
b'keep'})
+ finally:
+ reader.close()
+ self.assertEqual(values, [payload, None, payload] if
reverse else [payload, payload, None])
+ self.assertEqual(sizes, [2, 1] if reverse else [1, 2])
+
def test_merge_all_zero_column_batches_keep_row_counts(self):
batches = [pa.record_batch([pa.array(range(size))],
names=['id']).select([])
for size in (1, 2, 2)]
diff --git a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
index 73e39a74b9..1f5da1f922 100644
--- a/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_lerobot_test.py
@@ -1595,8 +1595,8 @@ class LeRobotValidationTest(unittest.TestCase):
return_value=loaded), patch(
"pypaimon.multimodal.lerobot.dataset._target_schema",
return_value=pa.schema([])), patch(
- "pypaimon.multimodal.lerobot.dataset.sys.version_info",
- (3, 10)):
+ "pypaimon.multimodal.lerobot.dataset.sys",
+ SimpleNamespace(version_info=(3, 10), modules=sys.modules)):
for invalid in (0, 1, None, "true"):
with self.subTest(return_uint8=invalid):
with self.assertRaisesRegex(
diff --git a/paimon-python/pypaimon/tests/native_row_sidecar_test.py
b/paimon-python/pypaimon/tests/native_row_sidecar_test.py
new file mode 100644
index 0000000000..4241aad7e2
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_row_sidecar_test.py
@@ -0,0 +1,369 @@
+# 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.
+
+import json
+import os
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon import CatalogFactory, Schema
+from pypaimon.data.generic_variant import GenericVariant
+from pypaimon.schema.data_types import PyarrowFieldParser
+from pypaimon.write.native_update import NativeTableUpdateByRowId
+from pypaimon.write.table_update_by_row_id import TableUpdateByRowId
+from pypaimon.write.native_write import NativeTableWrite,
native_write_available
+
+pytestmark = [pytest.mark.native_plan, pytest.mark.skipif(
+ not native_write_available(), reason='pypaimon-rust runtime required')]
+
+
[email protected]('stream', [False, True])
+def test_row_sidecar_options_keep_batch_and_stream_writes_native(tmp_path,
stream):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ schema = pa.schema([('id', pa.int32()), ('value', pa.string())])
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options={
+ 'file.format': 'parquet', 'write.native.enabled': 'true',
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true',
+ 'data-evolution.row-sidecar.enabled': 'true'}), False)
+ table = catalog.get_table('default.t')
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow(pa.Table.from_pydict({'id': [1, 2], 'value': ['a',
'b']}, schema=schema))
+ messages = writer.prepare_commit(7) if stream else
writer.prepare_commit()
+ for message in messages:
+ for file in message.new_files:
+ assert file.extra_files == [file.file_name + '.row']
+ assert all(table.file_io.exists(path) for path in
file.collect_files())
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, 7) if stream else commit.commit(messages)
+ finally:
+ commit.close()
+ finally:
+ writer.close()
+
+
+def _table(tmp_path, schema=None, options=None):
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.create_database('default', True)
+ if schema is None:
+ schema = pa.schema([('id', pa.int32()), ('value', pa.string())])
+ settings = {'file.format': 'parquet', 'write.native.enabled': 'true',
+ 'read.native.enabled': 'true', 'row-tracking.enabled': 'true',
+ 'data-evolution.enabled': 'true', 'deletion-vectors.enabled':
'true',
+ 'data-evolution.row-sidecar.enabled': 'true'}
+ settings.update(options or {})
+ catalog.create_table('default.t', Schema.from_pyarrow_schema(schema,
options=settings), False)
+ return catalog.get_table('default.t')
+
+
+def _commit(builder, messages, stream=False, identifier=1):
+ commit = builder.new_commit()
+ try:
+ commit.commit(messages, identifier) if stream else
commit.commit(messages)
+ finally:
+ commit.close()
+
+
+def _append(table, start=0, count=100):
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ try:
+ schema = PyarrowFieldParser.from_paimon_schema(table.fields)
+ writer.write_arrow(pa.table({'id': list(range(start, start + count)),
+ 'value': ['v%d' % i for i in range(start,
start + count)]},
+ schema=schema))
+ messages = writer.prepare_commit()
+ _commit(builder, messages)
+ return messages
+ finally:
+ writer.close()
+
+
+def _read(table, projection, row_id, native=True):
+ builder = table.new_read_builder().with_projection(projection)
+ predicate = (table.new_read_builder().with_projection(['_ROW_ID'])
+ .new_predicate_builder().equal('_ROW_ID', row_id))
+ builder.with_filter(predicate)
+ read = builder.new_read()
+ splits = builder.new_scan().plan().splits()
+ if native:
+ # Fail if a successful assertion actually used Python's ROW decoder.
+ with patch.object(read, '_create_split_read',
side_effect=AssertionError('Python fallback')):
+ return read.to_arrow(splits)
+ return read.to_arrow(splits)
+
+
+def _remove_primaries(messages):
+ for message in messages:
+ for file in message.new_files:
+ os.remove(file.external_path or file.file_path)
+
+
[email protected]('native_write', [False, True])
[email protected]('native_read', [False, True])
+def test_row_sidecars_interoperate_between_python_and_rust(tmp_path,
native_write, native_read):
+ table = _table(tmp_path, options={'write.native.enabled':
str(native_write).lower()})
+ _append(table, -20, 20)
+ messages = _append(table)
+ _remove_primaries(messages)
+ read_table = table.copy({'read.native.enabled': str(native_read).lower(),
+ 'data-evolution.row-sidecar.enabled': 'false'})
+ assert _read(read_table, ['id', 'value', '_ROW_ID'], 25,
native_read).to_pydict() == {
+ 'id': [5], 'value': ['v5'], '_ROW_ID': [25]}
+
+
[email protected]('stream', [False, True])
[email protected]('incremental', [False, True])
+def test_sidecars_keep_updates_native_and_apply_deletion_vectors(tmp_path,
stream, incremental):
+ table = _table(tmp_path)
+ initial = _append(table)
+ builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ update = builder.new_update()
+ data = pa.table({'_ROW_ID': [5, 7], 'value': ['changed-5', 'changed-7']})
+ if incremental:
+ updater = update.new_update_by_row_id(1) if stream else
update.new_update_by_row_id()
+ assert isinstance(updater, NativeTableUpdateByRowId)
+ messages = updater.update_columns(data, ['value'])
+ else:
+ with patch.object(TableUpdateByRowId, 'update_columns',
+ side_effect=AssertionError('Python update')):
+ messages = (update.update_by_arrow_with_row_id(data, 1) if stream
+ else update.update_by_arrow_with_row_id(data))
+ for message in messages:
+ for file in message.new_files:
+ assert file.extra_files == [file.file_name + '.row']
+ _commit(builder, messages, stream)
+ delete_builder = table.new_stream_write_builder() if stream else
table.new_batch_write_builder()
+ delete = delete_builder.new_update()
+ deletes = delete.delete_by_row_id([6], 2) if stream else
delete.delete_by_row_id([6])
+ _commit(delete_builder, deletes, stream, 2)
+ _remove_primaries(initial + messages)
+ assert _read(table, ['id', 'value', '_ROW_ID'], 5).to_pydict() == {
+ 'id': [5], 'value': ['changed-5'], '_ROW_ID': [5]}
+ assert _read(table, ['id', 'value', '_ROW_ID'], 6).num_rows == 0
+ assert _read(table, ['id', 'value', '_ROW_ID'], 7).to_pydict() == {
+ 'id': [7], 'value': ['changed-7'], '_ROW_ID': [7]}
+
+
[email protected]('layout', ['plain', 'map', 'variant'])
+def test_sidecars_store_logical_nested_map_and_variant_values(tmp_path,
layout):
+ schema = pa.schema([
+ ('id', pa.int32()),
+ ('profile', pa.struct([('score', pa.int32()), ('unused',
pa.string())])),
+ ('attrs', pa.map_(pa.string(), pa.int64())),
+ *([('payload', GenericVariant.to_arrow_array([]).type)] if layout !=
'map' else []),
+ ])
+ options = {}
+ if layout == 'map':
+ options.update({'fields.attrs.map.storage-layout': 'shared-shredding',
+ 'fields.attrs.map.shared-shredding.max-columns': '1'})
+ elif layout == 'variant':
+ options['variant.shreddingSchema'] = json.dumps({'type': 'ROW',
'fields': [{
+ 'name': 'payload', 'type': {'type': 'ROW', 'fields': [
+ {'name': 'score', 'type': 'INT'}, {'name': 'unused', 'type':
'STRING'}]}}]})
+ table = _table(tmp_path, schema, options)
+ schema = PyarrowFieldParser.from_paimon_schema(table.fields)
+ values = {
+ 'id': list(range(100)),
+ 'profile': [{'score': i, 'unused': 'ignore'} if i != 6 else None for i
in range(100)],
+ 'attrs': [[('wanted', i), ('other', -i)] if i != 6 else None for i in
range(100)],
+ }
+ if layout != 'map':
+ values['payload'] = GenericVariant.to_arrow_array([
+ GenericVariant.from_python({'score': i, 'unused': 'ignore'}) if i
!= 6 else None
+ for i in range(100)])
+ data = pa.table(values, schema=schema)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow(data)
+ messages = writer.prepare_commit()
+ _commit(builder, messages)
+ finally:
+ writer.close()
+ _remove_primaries(messages)
+ projection = {'score': 'profile.score', 'wanted': "attrs['wanted']",
+ 'missing': "attrs['missing']",
+ 'row_id': '_ROW_ID'}
+ if layout != 'map':
+ projection['variant'] = "variant_get(payload, '$.score', 'float')"
+ for row_id in [5, 6]:
+ value = 5 if row_id == 5 else None
+ expected = {'score': [value], 'wanted': [value], 'missing': [None],
'row_id': [row_id]}
+ if layout != 'map':
+ expected['variant'] = [value]
+ assert _read(table, projection, row_id).to_pydict() == expected
+
+
[email protected]('enabled', [None, 'false', 'true'])
+def test_missing_sidecar_can_fall_back_to_native_parquet(tmp_path, enabled):
+ options = {} if enabled is None else {'scan.ignore-lost-files': enabled}
+ table = _table(tmp_path, options=options)
+ messages = _append(table)
+ for message in messages:
+ for file in message.new_files:
+ os.remove(next(path for path in file.collect_files() if
path.endswith('.row')))
+ if enabled == 'true':
+ assert _read(table, ['id', 'value'], 5).to_pydict() == {'id': [5],
'value': ['v5']}
+ else:
+ with pytest.raises(ValueError, match=r'parquet\.row'):
+ _read(table, ['id', 'value'], 5)
+
+
[email protected]('enabled', ['false', 'off', 'true', 'on'])
+def test_sidecar_generation_uses_parsed_boolean_options(tmp_path, enabled):
+ table = _table(tmp_path, options={'data-evolution.row-sidecar.enabled':
enabled})
+ messages = _append(table, count=2)
+ for message in messages:
+ for file in message.new_files:
+ assert bool(file.extra_files) == (enabled in ('true', 'on'))
+
+
[email protected]('blob_type', [
+ pa.large_binary(), pa.list_(pa.large_binary()), pa.map_(pa.string(),
pa.large_binary())])
+def test_dedicated_blob_writers_keep_java_sidecar_exclusion(tmp_path,
blob_type):
+ schema = pa.schema([('id', pa.int32()), ('payload', blob_type)])
+ table = _table(tmp_path, schema)
+ if pa.types.is_list(blob_type):
+ values = [[b'a', None], None]
+ elif pa.types.is_map(blob_type):
+ values = [[('key', b'a')], None]
+ else:
+ values = [b'a', None]
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow(pa.table({'id': [1, 2], 'payload': values},
schema=schema))
+ messages = writer.prepare_commit()
+ files = [file for message in messages for file in message.new_files]
+ assert any(file.file_name.endswith('.blob') for file in files)
+ assert any(file.file_name.endswith('.parquet') for file in files)
+ assert all(not file.extra_files for file in files)
+ _commit(builder, messages)
+ finally:
+ writer.close()
+
+
+def
test_explicit_abort_owns_sidecars_but_close_after_prepare_preserves_them(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ writer.write_arrow(pa.table({'id': [1], 'value': ['a']},
+
schema=PyarrowFieldParser.from_paimon_schema(table.fields)))
+ messages = writer.prepare_commit()
+ paths = {path for message in messages for file in message.new_files for
path in file.collect_files()}
+ assert len(paths) == 2
+ writer.close()
+ assert all(table.file_io.exists(path) for path in paths)
+ commit = builder.new_commit()
+ try:
+ # These messages were never submitted and are now permanently
abandoned.
+ commit.abort(messages)
+ finally:
+ commit.close()
+ assert all(not table.file_io.exists(path) for path in paths)
+
+
[email protected]('external', [False, True])
+def test_old_sidecar_schema_evolves_by_field_id(tmp_path, external):
+ from pypaimon.schema.data_types import AtomicType
+ from pypaimon.schema.schema_change import SchemaChange
+
+ options = {'data-file.external-paths': str(tmp_path / 'external data')} if
external else {}
+ table = _table(tmp_path, options=options)
+ messages = _append(table)
+ _remove_primaries(messages)
+ catalog = CatalogFactory.create({'warehouse': str(tmp_path)})
+ catalog.alter_table('default.t', [
+ SchemaChange.rename_column('value', 'renamed_value'),
+ SchemaChange.update_column_type('id', AtomicType('BIGINT')),
+ SchemaChange.add_column('added', AtomicType('INT')),
+ ], False)
+ table = catalog.get_table('default.t')
+ actual = _read(table, ['renamed_value', 'added', 'id', '_ROW_ID'], 5)
+ assert actual.to_pydict() == {
+ 'renamed_value': ['v5'], 'added': [None], 'id': [5], '_ROW_ID': [5]}
+ assert actual.schema.field('id').type == pa.int64()
+
+
+def test_bitmap_and_limit_keep_primary_without_requested_row_ranges(tmp_path):
+ table = _table(tmp_path, options={'file-index.bitmap.columns': 'id'})
+ messages = _append(table)
+ for message in messages:
+ for file in message.new_files:
+ os.remove(next(path for path in file.collect_files() if
path.endswith('.row')))
+ for mode in ['bitmap', 'limit']:
+ builder = table.new_read_builder()
+ if mode == 'bitmap':
+ builder.with_filter(builder.new_predicate_builder().equal('id', 5))
+ else:
+ builder.with_limit(1)
+ read = builder.new_read()
+ with patch.object(read, '_create_split_read',
side_effect=AssertionError('Python fallback')):
+ actual = read.to_arrow(builder.new_scan().plan().splits())
+ expected_id = 5 if mode == 'bitmap' else 0
+ assert actual.to_pydict() == {'id': [expected_id], 'value': ['v%d' %
expected_id]}
+
+
[email protected]('option', ['blob-descriptor-field',
'blob.stored-descriptor-fields'])
[email protected]('native_read', [False, True])
+def test_sidecars_materialize_inline_blob_references(tmp_path, option,
native_read):
+ from pypaimon.table.row.blob import BlobDescriptor
+
+ schema = pa.schema([('id', pa.int32()), ('payload', pa.large_binary()),
('value', pa.string())])
+ table = _table(tmp_path, schema, {
+ option: 'payload', 'read.native.enabled': str(native_read).lower()})
+ # Actual bytes may themselves look exactly like a reference. ROW decoding
+ # must return them intact, without resolving that inner descriptor.
+ payload = BlobDescriptor(str(tmp_path / 'must-not-read'), 0, 1).serialize()
+ source = tmp_path / 'blob-source'
+ source.write_bytes(b'prefix' + payload + b'suffix')
+ reference = BlobDescriptor(str(source), 6, len(payload)).serialize()
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ assert isinstance(writer, NativeTableWrite)
+ try:
+ writer.write_arrow(pa.table({
+ 'id': list(range(100)), 'value': ['v%d' % i for i in range(100)],
+ 'payload': [reference if i != 6 else None for i in range(100)],
+ }, schema=schema))
+ messages = writer.prepare_commit()
+ _commit(builder, messages)
+ finally:
+ writer.close()
+ descriptor_table = table.copy({'blob-as-descriptor': 'true'})
+ assert _read(descriptor_table, ['payload'], 5,
native=native_read).to_pydict() == {'payload': [reference]}
+ read = descriptor_table.new_read_builder().with_projection(['payload'])
+ assert
read.new_read().to_arrow(read.new_scan().plan().splits())['payload'][5].as_py()
== reference
+ update_builder = table.copy({'write.native.enabled':
'true'}).new_batch_write_builder()
+ with patch.object(TableUpdateByRowId, 'update_columns',
side_effect=AssertionError('Python update')):
+ updates = update_builder.new_update().update_by_arrow_with_row_id(
+ pa.table({'_ROW_ID': [5], 'value': ['changed-5']}))
+ _commit(update_builder, updates)
+ source.unlink()
+ _remove_primaries(messages + updates)
+ assert _read(table, ['id', 'payload', 'value', '_ROW_ID'], 5,
native=native_read).to_pydict() == {
+ 'id': [5], 'payload': [payload], 'value': ['changed-5'], '_ROW_ID':
[5]}
+ assert _read(table, ['payload'], 6, native=native_read).to_pydict() ==
{'payload': [None]}
diff --git a/paimon-python/pypaimon/tests/native_update_test.py
b/paimon-python/pypaimon/tests/native_update_test.py
index cf4e138bbf..eb5870a07c 100644
--- a/paimon-python/pypaimon/tests/native_update_test.py
+++ b/paimon-python/pypaimon/tests/native_update_test.py
@@ -355,7 +355,12 @@ def
test_native_batch_update_preserves_input_table_boundaries(tmp_path):
with pytest.raises(RuntimeError, match='input generator failed'):
table.new_batch_write_builder().new_update().update_by_arrow_batches_with_row_id(
failed_input())
- assert set(tmp_path.rglob('*.parquet')) == before
+ # Native update already returned the first table's prepared messages to
+ # the binding. A later iterator failure must not delete handed-off files.
+ assert before < set(tmp_path.rglob('*.parquet'))
+ read = table.new_read_builder()
+ assert
read.new_read().to_arrow(read.new_scan().plan().splits()).sort_by('id').select(
+ ['id', 'age']).to_pydict() == {'id': [2, 3, 4, 10], 'age': [22, 33,
40, 11]}
read_snapshot_id = table.snapshot_manager().get_latest_snapshot().id
diff --git a/paimon-python/pypaimon/utils/arrow_utils.py
b/paimon-python/pypaimon/utils/arrow_utils.py
index a3dea97db4..77e1064813 100644
--- a/paimon-python/pypaimon/utils/arrow_utils.py
+++ b/paimon-python/pypaimon/utils/arrow_utils.py
@@ -23,3 +23,31 @@ def zero_column_batch(num_rows: int, metadata=None) ->
pa.RecordBatch:
empty_struct = pa.Array.from_buffers(
pa.struct([]), num_rows, [None], children=[])
return
pa.RecordBatch.from_struct_array(empty_struct).replace_schema_metadata(metadata)
+
+
+# Internal provenance: Arrow represents BlobData and BlobRef as the same bytes
+# type. Payloads may themselves parse as references, so inspecting their bytes
+# cannot distinguish them. Remove this marker before exposing output batches.
+_ROW_BLOB_DATA = b'paimon.row-sidecar.blob-data'
+
+
+def is_blob_data(field):
+ return _ROW_BLOB_DATA in (field.metadata or {})
+
+
+def as_blob_data(field):
+ metadata = dict(field.metadata or {})
+ metadata[_ROW_BLOB_DATA] = b'true'
+ return field.with_metadata(metadata)
+
+
+def clear_blob_data(batch):
+ if not any(is_blob_data(field) for field in batch.schema):
+ return batch
+ fields = []
+ for field in batch.schema:
+ metadata = dict(field.metadata or {})
+ metadata.pop(_ROW_BLOB_DATA, None)
+ fields.append(field.with_metadata(metadata or None))
+ return pa.RecordBatch.from_arrays(
+ batch.columns, schema=pa.schema(fields,
metadata=batch.schema.metadata))
diff --git a/paimon-python/pypaimon/write/native_commit.py
b/paimon-python/pypaimon/write/native_commit.py
index 6a12343bbd..acbb1d2081 100644
--- a/paimon-python/pypaimon/write/native_commit.py
+++ b/paimon-python/pypaimon/write/native_commit.py
@@ -147,6 +147,8 @@ def create_native_write_table(table):
table.options.snapshot_ignore_empty_commit())
options['row-tracking.partition-group-on-commit'] =
_option_value_to_string(
table.options.row_tracking_partition_group_on_commit())
+ options['data-evolution.row-sidecar.enabled'] = _option_value_to_string(
+ table.options.data_evolution_row_sidecar_enabled())
schema_json = JSON.to_json(table.table_schema.copy(new_options=options))
if environment.supports_version_management:
if not _rest_catalog_supported(table):
diff --git a/paimon-python/pypaimon/write/native_update.py
b/paimon-python/pypaimon/write/native_update.py
index f291369bc4..ec077b7b30 100644
--- a/paimon-python/pypaimon/write/native_update.py
+++ b/paimon-python/pypaimon/write/native_update.py
@@ -40,7 +40,6 @@ def _native_row_id_table(table):
or not table.options.row_tracking_enabled()
or table.is_primary_key_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.changelog_producer() != ChangelogProducer.NONE
or any(table.options.options.contains_key(key) for key in
SCAN_KEYS)
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index 81943baefd..be04e4f9bd 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -56,9 +56,6 @@ def create_native_write(table, commit_user,
static_partition=None, stream=False,
pa.types.is_floating(schema.field(name).type) for name in
sequence_fields)):
return None
if (not native_write_available()
- # Rust does not produce the optional random-access .row sidecars.
- or (table.options.data_evolution_enabled()
- and table.options.data_evolution_row_sidecar_enabled())
or table.bucket_mode() not in (BucketMode.HASH_FIXED,
BucketMode.HASH_DYNAMIC,
BucketMode.BUCKET_UNAWARE,
BucketMode.CROSS_PARTITION,
