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 9e22d0f8d0 [python] Enable native read in Python Rust Plan CI (#10067)
9e22d0f8d0 is described below
commit 9e22d0f8d0e7e96edc249605af66aa3c567ada0d
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 22 11:33:06 2026 +0800
[python] Enable native read in Python Rust Plan CI (#10067)
---
.github/workflows/ci-python.yml | 3 +-
paimon-python/conftest.py | 99 ++++++++++++++++------
paimon-python/dev/lint-python.sh | 3 +-
paimon-python/pypaimon/data/variant_shredding.py | 25 ++++--
paimon-python/pypaimon/read/native_plan.py | 10 +--
.../pypaimon/read/reader/data_file_batch_reader.py | 39 +++++++--
.../read/reader/nested_leaf_batch_reader.py | 4 +-
.../pypaimon/tests/batch_vector_raw_scan_test.py | 4 +
paimon-python/pypaimon/tests/blob_test.py | 2 +
.../tests/contiguous_window_dataset_test.py | 2 +
.../pypaimon/tests/data_evolution_formats_test.py | 1 +
.../pypaimon/tests/deferred_blob_resolve_test.py | 2 +
.../pypaimon/tests/deletion_vector_path_test.py | 5 +-
.../pypaimon/tests/multimodal_table_test.py | 3 +
.../pypaimon/tests/multimodal_temporal_test.py | 3 +
.../pypaimon/tests/native_plan_incremental_test.py | 25 +++++-
paimon-python/pypaimon/tests/native_read_test.py | 15 ++++
.../pypaimon/tests/parquet_page_index_test.py | 1 +
.../pypaimon/tests/query_auth_validation_test.py | 2 +
.../pypaimon/tests/reader_parallel_test.py | 4 +
.../pypaimon/tests/rest/rest_base_test.py | 25 ++----
.../pypaimon/tests/rest/rest_read_write_test.py | 28 +++---
.../tests/schema_evolution_nested_read_test.py | 36 ++++++++
.../pypaimon/tests/schema_evolution_read_test.py | 27 +++---
.../pypaimon/tests/test_early_row_range_filter.py | 2 +
paimon-python/pypaimon/tests/variant_test.py | 37 ++++++++
26 files changed, 315 insertions(+), 92 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index 0d9fd9abcf..ea2a988695 100644
--- a/.github/workflows/ci-python.yml
+++ b/.github/workflows/ci-python.yml
@@ -232,10 +232,11 @@ jobs:
assert hasattr(ReadBuilder, 'with_row_ranges'), 'Missing
ReadBuilder.with_row_ranges'
PY
- - name: Run Python tests with Rust planning
+ - name: Run Python tests with Rust planning and reading
shell: bash
env:
PYPAIMON_TEST_NATIVE_PLAN: '1'
+ PYPAIMON_TEST_NATIVE_READ: '1'
run: |
bash paimon-python/dev/lint-python.sh -i pytest
diff --git a/paimon-python/conftest.py b/paimon-python/conftest.py
index 7928854abc..b5090ed771 100644
--- a/paimon-python/conftest.py
+++ b/paimon-python/conftest.py
@@ -20,8 +20,11 @@ import pytest
_NATIVE_PLAN_ENV = "PYPAIMON_TEST_NATIVE_PLAN"
+_NATIVE_READ_ENV = "PYPAIMON_TEST_NATIVE_READ"
_native_plan_count = 0
+_native_read_count = 0
_force_native_for_test = False
+_force_native_read_for_test = False
def pytest_addoption(parser):
@@ -35,60 +38,106 @@ def _native_plan_enabled():
return os.environ.get(_NATIVE_PLAN_ENV) == "1"
+def _native_read_enabled():
+ return os.environ.get(_NATIVE_READ_ENV) == "1"
+
+
def pytest_configure(config):
config.addinivalue_line(
"markers", "python_plan: keep Python planner assertions on the Python
lane")
+ config.addinivalue_line(
+ "markers", "python_read: keep Python reader assertions on the Python
lane")
config.addinivalue_line(
"markers", "native_plan: exercise the real Rust planner in the Rust
main CI job")
- if not _native_plan_enabled():
- return
+ if _native_plan_enabled():
+ from pypaimon.read.table_scan import TableScan
+
+ original_plan = TableScan._try_native_plan
+
+ def tracked_plan(self):
+ global _native_plan_count
+ plan = original_plan(self)
+ if plan is not None and _force_native_for_test:
+ _native_plan_count += 1
+ return plan
+
+ TableScan._try_native_plan = tracked_plan
- from pypaimon.read.table_scan import TableScan
+ if _native_read_enabled():
+ from pypaimon.read.table_read import TableRead
- original = TableScan._try_native_plan
+ original_read = TableRead._try_native_batches
- def tracked(self):
- global _native_plan_count
- plan = original(self)
- if plan is not None and _force_native_for_test:
- _native_plan_count += 1
- return plan
+ def tracked_read(self, splits, *args, **kwargs):
+ global _native_read_count
+ batches = original_read(self, splits, *args, **kwargs)
+ if batches is not None and splits and _force_native_read_for_test:
+ _native_read_count += 1
+ return batches
- TableScan._try_native_plan = tracked
+ TableRead._try_native_batches = tracked_read
+
+
+def pytest_collection_modifyitems(items):
+ if _native_plan_enabled():
+ return
+ skip_native = pytest.mark.skip(reason="native plan tests run in the Rust
Plan job")
+ for item in items:
+ if item.get_closest_marker("native_plan") is not None:
+ item.add_marker(skip_native)
@pytest.fixture(autouse=True)
-def enable_native_plan(request, monkeypatch):
- global _force_native_for_test
- if (not _native_plan_enabled()
- or request.node.get_closest_marker("python_plan") is not None
- or request.path.name in (
- "native_plan_test.py", "native_plan_integration_test.py",
- "native_plan_capabilities_test.py")):
+def enable_native_plan_and_read(request, monkeypatch):
+ global _force_native_for_test, _force_native_read_for_test
+ python_plan = request.node.get_closest_marker("python_plan") is not None
+ python_read = request.node.get_closest_marker("python_read") is not None
+ native_plan_test = request.path.name in (
+ "native_plan_test.py", "native_plan_integration_test.py",
+ "native_plan_capabilities_test.py")
+ force_plan = _native_plan_enabled() and not python_plan and not
native_plan_test
+ force_read = (_native_read_enabled() and not python_plan and not
python_read
+ and not native_plan_test)
+ if not (force_plan or force_read):
yield
return
from pypaimon.common.options.core_options import CoreOptions
- original = CoreOptions.native_plan_enabled
+ if force_plan:
+ original_plan = CoreOptions.native_plan_enabled
- def enabled(self, default=None):
- return original(self, True if default is None else default)
+ def plan_enabled(self, default=None):
+ return original_plan(self, True if default is None else default)
- monkeypatch.setattr(CoreOptions, "native_plan_enabled", enabled)
- _force_native_for_test = True
+ monkeypatch.setattr(CoreOptions, "native_plan_enabled", plan_enabled)
+ if force_read:
+ original_read = CoreOptions.native_read_enabled
+
+ def read_enabled(self, default=None):
+ return original_read(self, True if default is None else default)
+
+ monkeypatch.setattr(CoreOptions, "native_read_enabled", read_enabled)
+ _force_native_for_test = force_plan
+ _force_native_read_for_test = force_read
try:
yield
finally:
_force_native_for_test = False
+ _force_native_read_for_test = False
def pytest_sessionfinish(session, exitstatus):
- if _native_plan_enabled() and exitstatus == 0 and _native_plan_count == 0:
- session.exitstatus = pytest.ExitCode.TESTS_FAILED
+ if exitstatus == 0:
+ if ((_native_plan_enabled() and _native_plan_count == 0)
+ or (_native_read_enabled() and _native_read_count == 0)):
+ session.exitstatus = pytest.ExitCode.TESTS_FAILED
def pytest_terminal_summary(terminalreporter):
if _native_plan_enabled():
terminalreporter.write_line(
"native plans exercised: %d" % _native_plan_count)
+ if _native_read_enabled():
+ terminalreporter.write_line(
+ "native reads exercised: %d" % _native_read_count)
diff --git a/paimon-python/dev/lint-python.sh b/paimon-python/dev/lint-python.sh
index 8ce00549e3..1795379f17 100755
--- a/paimon-python/dev/lint-python.sh
+++ b/paimon-python/dev/lint-python.sh
@@ -247,8 +247,9 @@ function pytest_torch_check() {
}
# Mixed tests check - runs Java-Python interoperability tests
function mixed_check() {
- # Native-plan coverage is asserted only by the main pytest session.
+ # Native plan/read coverage is asserted only by the main pytest session.
unset PYPAIMON_TEST_NATIVE_PLAN
+ unset PYPAIMON_TEST_NATIVE_READ
# Get Python version
PYTHON_VERSION=$(python -c "import sys;
print(f'{sys.version_info.major}.{sys.version_info.minor}')")
diff --git a/paimon-python/pypaimon/data/variant_shredding.py
b/paimon-python/pypaimon/data/variant_shredding.py
index d67af9bf7a..ae95d5f800 100644
--- a/paimon-python/pypaimon/data/variant_shredding.py
+++ b/paimon-python/pypaimon/data/variant_shredding.py
@@ -315,10 +315,14 @@ def _append_scalar(builder, value, arrow_type:
pa.DataType) -> None:
# Object / array binary construction
# ---------------------------------------------------------------------------
-def _build_object_value(fields: List[Tuple[int, bytes]]) -> bytes:
+def _build_object_value(
+ fields: List[Tuple[int, bytes]],
+ key_dict: Optional[Dict[str, int]] = None,
+) -> bytes:
"""Build object variant value bytes from ``(key_id, value_bytes)`` pairs.
- The variant spec requires fields sorted by key_id.
+ Variant object fields are ordered by key name, not metadata key ID. The
+ metadata dictionary may assign IDs in a different order.
"""
if not fields:
# Empty object: header + size=0 + one zero-offset sentinel
@@ -328,7 +332,14 @@ def _build_object_value(fields: List[Tuple[int, bytes]])
-> bytes:
buf.append(0) # offset[0] = 0 (sentinel)
return bytes(buf)
- fields = sorted(fields, key=lambda f: f[0])
+ if key_dict is None:
+ fields = sorted(fields, key=lambda f: f[0])
+ else:
+ id_to_name = {key_id: name for name, key_id in key_dict.items()}
+ fields = sorted(
+ fields,
+ key=lambda f: id_to_name[f[0]].encode('utf-8'),
+ )
size = len(fields)
data = b''.join(vb for _, vb in fields)
data_size = len(data)
@@ -520,7 +531,7 @@ def _rebuild_object(
if overflow_bytes:
fields.extend(_extract_overflow_fields(bytes(overflow_bytes)))
- return _build_object_value(fields)
+ return _build_object_value(fields, key_dict)
def _rebuild_array(
@@ -901,7 +912,8 @@ def _decompose_field_bytes(
else:
typed_value[fname] = {'value': None, 'typed_value': None}
- overflow_bytes = _build_object_value(overflow_pairs) if overflow_pairs
else None
+ overflow_bytes = (_build_object_value(overflow_pairs, key_dict)
+ if overflow_pairs else None)
return {'value': overflow_bytes, 'typed_value': typed_value}
# No shredding sub-schema: treat field bytes as overflow
@@ -959,7 +971,8 @@ def decompose_variant(
else:
typed_value[fname] = {'value': None, 'typed_value': None}
- overflow_bytes = _build_object_value(overflow_pairs) if overflow_pairs
else None
+ overflow_bytes = (_build_object_value(overflow_pairs, key_dict)
+ if overflow_pairs else None)
return {'metadata': metadata, 'value': overflow_bytes, 'typed_value':
typed_value}
diff --git a/paimon-python/pypaimon/read/native_plan.py
b/paimon-python/pypaimon/read/native_plan.py
index adae794d29..0908be7dc3 100644
--- a/paimon-python/pypaimon/read/native_plan.py
+++ b/paimon-python/pypaimon/read/native_plan.py
@@ -415,11 +415,11 @@ def native_plan(
for split in rust_splits
]
if table.options.native_read_enabled():
- # Retain the opaque Rust split next to the Python metadata view. The
- # normal planner/reader contract remains a Python Split list, while
- # native reads can consume the exact Rust split without a second lossy
- # conversion. Any Python split transformation creates a fresh object
- # without this marker and thus safely falls back to the Python reader.
+ # Retain the opaque Rust split next to the Python metadata view so an
+ # unchanged native plan can be read without reserializing each split.
+ # A caller that needs different metadata (such as an endpoint DV)
+ # passes a new Python split, which the native reader converts at read
+ # time from its current fields.
for split, rust_split in zip(splits, rust_splits):
split._native_split = rust_split
_restore_python_partition_paths(table, splits)
diff --git a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
index bd98bccf2a..f739b12961 100644
--- a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+from decimal import Decimal, localcontext, ROUND_HALF_UP
from typing import List, Optional, Tuple
import pyarrow as pa
@@ -42,6 +43,32 @@ def _is_character_string_type(data_type) -> bool:
return t == 'STRING' or t.startswith('VARCHAR') or t.startswith('CHAR')
+def cast_array_for_schema_evolution(array, target_type):
+ """Cast old-file values using Java's DECIMAL rounding and overflow rules.
+
+ PyArrow's unsafe DECIMAL scale reduction truncates and can retain values
+ outside the target precision. Java rounds HALF_UP and returns NULL for
+ values that do not fit after rounding. Other conversions retain the
+ existing unsafe-cast behavior (such as DOUBLE -> INT truncation).
+ """
+ if (pa.types.is_decimal(array.type) and pa.types.is_decimal(target_type)
+ and array.type.scale > target_type.scale):
+ values = []
+ with localcontext() as context:
+ # Rounding a maximal value can carry into one extra digit.
+ context.prec = max(array.type.precision, target_type.precision) + 1
+ quantum = Decimal(1).scaleb(-target_type.scale)
+ for value in array.to_pylist():
+ if value is None:
+ values.append(None)
+ continue
+ rounded = value.quantize(quantum, rounding=ROUND_HALF_UP)
+ values.append(rounded if len(rounded.as_tuple().digits)
+ <= target_type.precision else None)
+ return pa.array(values, type=target_type)
+ return array.cast(target_type, safe=False)
+
+
def _unslice(array):
"""Re-materialize a sliced array so offsets/buffers start at zero.
@@ -306,7 +333,7 @@ class DataFileBatchReader(RecordBatchReader):
# Leaf / non-nested: cast to the target type when it differs.
target_pa_type = PyarrowFieldParser.from_paimon_type(target_type)
if array.type != target_pa_type:
- return array.cast(target_pa_type, safe=False)
+ return cast_array_for_schema_evolution(array, target_pa_type)
return array
def read_arrow_batch(self, start_idx=None, end_idx=None) ->
Optional[RecordBatch]:
@@ -392,11 +419,9 @@ class DataFileBatchReader(RecordBatchReader):
type. Columns whose type already matches are reused as-is, keeping the
common (non-evolution) path zero-copy.
- Casts use ``safe=False`` to match Java ``CastExecutors`` semantics for
- the read-time conversions a user-approved schema evolution implies
- (e.g. DECIMAL scale-down or DOUBLE -> INT truncate rather than raise).
- Evolution legality is the writer's concern (``DataTypeCasts``); the
read
- path only materializes the result.
+ DECIMAL scale reductions use Java's HALF_UP and overflow-to-NULL
+ semantics; other conversions retain the existing unsafe Arrow cast.
+ Evolution legality is the writer's concern (``DataTypeCasts``).
"""
out_arrays = []
out_fields = []
@@ -405,7 +430,7 @@ class DataFileBatchReader(RecordBatchReader):
if target_field is None:
target_field = pa.field(name, array.type)
elif array.type != target_field.type:
- array = array.cast(target_field.type, safe=False)
+ array = cast_array_for_schema_evolution(array,
target_field.type)
out_arrays.append(array)
out_fields.append(target_field)
return pa.RecordBatch.from_arrays(out_arrays,
schema=pa.schema(out_fields))
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 f85c818081..874fd447c8 100644
--- a/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/nested_leaf_batch_reader.py
@@ -23,6 +23,8 @@ from pyarrow import RecordBatch
from pypaimon.data.map_shared_shredding import \
assemble_normal_map_selected_keys
+from pypaimon.read.reader.data_file_batch_reader import \
+ cast_array_for_schema_evolution
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
@@ -80,7 +82,7 @@ class NestedLeafBatchReader(RecordBatchReader):
column = _struct_field(column, name)
target_type = self._schema.field(i).type
if column.type != target_type:
- column = column.cast(target_type, safe=False)
+ column = cast_array_for_schema_evolution(column, target_type)
arrays.append(column)
return pa.RecordBatch.from_arrays(arrays, schema=self._schema)
diff --git a/paimon-python/pypaimon/tests/batch_vector_raw_scan_test.py
b/paimon-python/pypaimon/tests/batch_vector_raw_scan_test.py
index 6bc5b4f818..80ae700f16 100644
--- a/paimon-python/pypaimon/tests/batch_vector_raw_scan_test.py
+++ b/paimon-python/pypaimon/tests/batch_vector_raw_scan_test.py
@@ -20,6 +20,7 @@ import threading
from unittest.mock import patch
import pyarrow as pa
+import pytest
from pypaimon.read.table_read import TableRead
from pypaimon.table.source.vector_search_read import BatchVectorSearchReadImpl
@@ -117,6 +118,7 @@ class BatchVectorRawScanTest(BatchModeMixin,
DataEvolutionTestBase, unittest.Tes
results = reader._read_raw_batch_search([Range(0, 99)], None,
'ivf-flat')
self.assertEqual([{}, {}], [_scores(r) for r in results])
+ @pytest.mark.python_read
def test_scoring_finishes_each_batch_before_reading_the_next(self):
table = self._create_table()
self._write_arrow(table, self._data([[1, 0], [0, 1], [2, 0], [0, 2]]))
@@ -198,6 +200,7 @@ class BatchVectorRawScanTest(BatchModeMixin,
DataEvolutionTestBase, unittest.Tes
self.assertEqual([1.0], list(_scores(current[0]).values()))
self.assertNotEqual(list(old[0].results()), list(current[0].results()))
+ @pytest.mark.python_read
def test_public_batch_search_preserves_split_parallelism(self):
table = self._create_table(partition_keys=['pt'])
for partition in range(4):
@@ -246,6 +249,7 @@ class BatchVectorRawScanTest(BatchModeMixin,
DataEvolutionTestBase, unittest.Tes
self.assertEqual({'active': 0, 'peak':
expected_workers,
'closed': expected_workers}, state)
+ @pytest.mark.python_read
def test_parallel_failure_closes_all_started_readers(self):
table = self._create_table(
partition_keys=['pt'], options=dict(self.table_options,
**{'read.parallelism': '2'}))
diff --git a/paimon-python/pypaimon/tests/blob_test.py
b/paimon-python/pypaimon/tests/blob_test.py
index dee9d21488..7e25b720a3 100644
--- a/paimon-python/pypaimon/tests/blob_test.py
+++ b/paimon-python/pypaimon/tests/blob_test.py
@@ -31,6 +31,7 @@ from pathlib import Path
from unittest.mock import MagicMock, patch
import pyarrow as pa
+import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.common.delta_varint_compressor import DeltaVarintCompressor
@@ -5664,6 +5665,7 @@ class BlobParallelismTest(unittest.TestCase):
for i in range(20):
self.assertEqual(got[i], self.payloads[i])
+ @pytest.mark.python_read
def test_blob_fallback_parallelism_end_to_end(self):
t = self.catalog.get_table('default.bp_test')
diff --git a/paimon-python/pypaimon/tests/contiguous_window_dataset_test.py
b/paimon-python/pypaimon/tests/contiguous_window_dataset_test.py
index c0e5d4d7dd..c498cca3d0 100644
--- a/paimon-python/pypaimon/tests/contiguous_window_dataset_test.py
+++ b/paimon-python/pypaimon/tests/contiguous_window_dataset_test.py
@@ -24,6 +24,7 @@ import unittest
from unittest.mock import patch
import pyarrow as pa
+import pytest
import torch
import pypaimon.multimodal as pmm
@@ -780,6 +781,7 @@ class ContiguousWindowDatasetTest(unittest.TestCase):
with self.assertRaisesRegex(ValueError, r"masks \['episode'\]"):
self._dataset(table)
+ @pytest.mark.python_read
def test_reads_only_the_files_and_row_ranges_a_window_touches(self):
table = self.conn.create_table(
"many_files", schema=self._schema(), options=_TABLE_OPTIONS)
diff --git a/paimon-python/pypaimon/tests/data_evolution_formats_test.py
b/paimon-python/pypaimon/tests/data_evolution_formats_test.py
index 73057665ee..d3fb26ceec 100644
--- a/paimon-python/pypaimon/tests/data_evolution_formats_test.py
+++ b/paimon-python/pypaimon/tests/data_evolution_formats_test.py
@@ -197,6 +197,7 @@ class DataEvolutionFormatsTest(unittest.TestCase):
for file_meta in all_files:
self.assertEqual([], self._row_sidecar_files(file_meta))
+ @pytest.mark.python_read
def test_row_sidecar_serves_sparse_row_id_read(self):
pa_schema = pa.schema([
('id', pa.int32()),
diff --git a/paimon-python/pypaimon/tests/deferred_blob_resolve_test.py
b/paimon-python/pypaimon/tests/deferred_blob_resolve_test.py
index e23ff0e8df..ee013017d6 100644
--- a/paimon-python/pypaimon/tests/deferred_blob_resolve_test.py
+++ b/paimon-python/pypaimon/tests/deferred_blob_resolve_test.py
@@ -24,6 +24,7 @@ from unittest.mock import patch
import pyarrow as pa
import pyarrow.compute as pc
+import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.catalog.table_query_auth import TableQueryAuthResult
@@ -104,6 +105,7 @@ class _PayloadAuthResult(TableQueryAuthResult):
batch.column("payload"), self._expected_payload)
[email protected]_read
class DeferredBlobResolveTest(unittest.TestCase):
@classmethod
diff --git a/paimon-python/pypaimon/tests/deletion_vector_path_test.py
b/paimon-python/pypaimon/tests/deletion_vector_path_test.py
index da61d04dc4..b1d9d4ce72 100644
--- a/paimon-python/pypaimon/tests/deletion_vector_path_test.py
+++ b/paimon-python/pypaimon/tests/deletion_vector_path_test.py
@@ -195,8 +195,11 @@ def
test_missing_explicit_dv_is_not_replaced_by_a_local_copy(tmp_path, planner):
with table.file_io.new_output_stream(directory + '/' + file.file_name)
as stream:
stream.write(data)
table.file_io.delete_quietly(file.external_path)
- with pytest.raises(FileNotFoundError):
+ # The Python filesystem raises FileNotFoundError; the native reader wraps
+ # the same missing explicit path in its storage error.
+ with pytest.raises((FileNotFoundError, ValueError)) as error:
_read(table, planner, 2, [1, 2, 3])
+ assert file.file_name in str(error.value)
@pytest.mark.parametrize('planner', _PLANNERS)
diff --git a/paimon-python/pypaimon/tests/multimodal_table_test.py
b/paimon-python/pypaimon/tests/multimodal_table_test.py
index 3d4de045ee..2e7e680f72 100644
--- a/paimon-python/pypaimon/tests/multimodal_table_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_table_test.py
@@ -26,6 +26,7 @@ from datetime import timedelta
from unittest.mock import patch
import pyarrow as pa
+import pytest
import pypaimon.multimodal as pmm
from pypaimon.multimodal import source_col
from pypaimon.common.predicate_builder import PredicateBuilder
@@ -630,6 +631,7 @@ class MultimodalTableTest(unittest.TestCase):
store.delete_object("images/cat.jpg")
self.assertEqual([], store.list_objects(prefix="images/"))
+ @pytest.mark.python_read
def test_blob_store_list_reads_batches_and_stops_at_limit(self):
from pypaimon.read.table_read import TableRead
@@ -1079,6 +1081,7 @@ class MultimodalTableTest(unittest.TestCase):
self.assertEqual(1, result.num_rows)
self.assertEqual([1], result["id"].to_pylist())
+ @pytest.mark.python_read
def test_scan_to_arrow_batch_reader(self):
users = self.conn.create_table(
"batch_users",
diff --git a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
index 74eb806885..f74b8fe245 100644
--- a/paimon-python/pypaimon/tests/multimodal_temporal_test.py
+++ b/paimon-python/pypaimon/tests/multimodal_temporal_test.py
@@ -26,6 +26,7 @@ from unittest import mock
import numpy as np
import pyarrow as pa
+import pytest
import pypaimon.multimodal as pmm
from pypaimon.multimodal import temporal
from pypaimon.catalog.table_query_auth import TableQueryAuthResult
@@ -741,6 +742,7 @@ class MultimodalTemporalTest(unittest.TestCase):
)
self.assertEqual(expected, row["matches"])
+ @pytest.mark.python_read
def test_window_join_prunes_unaggregated_right_columns(self):
anchors = self._table("window_projection_anchors", {
"episode_id": pa.int32(),
@@ -2402,6 +2404,7 @@ class MultimodalTemporalTest(unittest.TestCase):
self.assertIsNone(aligned.resolved_snapshots["right_1"]["snapshot_id"])
self.assertIsNone(aligned.to_list()[0]["value"])
+ @pytest.mark.python_read
def test_alignment_reuses_decoded_parquet_row_groups_across_batches(self):
anchors = self._table("cached_anchors", {
"episode_id": pa.int32(),
diff --git a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
index 17b4627b1a..54a32fd67a 100644
--- a/paimon-python/pypaimon/tests/native_plan_incremental_test.py
+++ b/paimon-python/pypaimon/tests/native_plan_incremental_test.py
@@ -28,6 +28,7 @@ import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.common.identifier import Identifier
from pypaimon.read.native_plan import native_method_available
+from pypaimon.read.split import DataSplit
from pypaimon.utils.range import Range
@@ -538,8 +539,24 @@ def
test_streaming_reader_honors_explicit_split_deletion_vector(catalog, native,
assert len(plan.splits()) == 1
split = plan.splits()[0]
assert split.is_streaming
- split.data_deletion_files = [DeletionFile(str(path), 0, len(encoded) - 8,
1)]
- # Planners do not attach endpoint DVs, but an explicit split DV is part of
- # the reader contract, including in Java streaming frames.
- result =
table.new_read_builder().new_read().to_arrow(plan.splits()).to_pylist()
+ # The planner cannot attach an endpoint DV. Build the reader input split
+ # with that DV instead of mutating a previously planned (and cached) split.
+ dv_split = DataSplit(
+ files=split.files,
+ partition=split.partition,
+ bucket=split.bucket,
+ raw_convertible=split.raw_convertible,
+ data_deletion_files=[DeletionFile(str(path), 0, len(encoded) - 8, 1)],
+ snapshot_id=split.snapshot_id,
+ is_streaming=split.is_streaming,
+ bucket_path=split.bucket_path,
+ total_buckets=split.total_buckets,
+ )
+ read_table = table.copy({'read.native.enabled': str(native).lower()})
+ with ExitStack() as stack:
+ if native:
+ stack.enter_context(patch(
+ 'pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('explicit DV native read fell
back')))
+ result =
read_table.new_read_builder().new_read().to_arrow([dv_split]).to_pylist()
assert result == [{'k': 1, 'v': '1'}, {'k': 3, 'v': '3'}]
diff --git a/paimon-python/pypaimon/tests/native_read_test.py
b/paimon-python/pypaimon/tests/native_read_test.py
index 6a33c32374..63e49ed50b 100644
--- a/paimon-python/pypaimon/tests/native_read_test.py
+++ b/paimon-python/pypaimon/tests/native_read_test.py
@@ -700,6 +700,21 @@ def
test_native_read_falls_back_for_unsupported_file_format():
native.assert_not_called()
+def test_native_avro_read_uses_native_path():
+ read = _table_read()
+ read.table.options.file_format.return_value = 'avro'
+ split = _Split('data.avro')
+ split._native_split = object()
+
+ with patch('pypaimon.read.native_plan.native_read',
+ return_value=[_id_batch([7])]) as native:
+ batches = list(read._try_native_batches(
+ [split], pa.schema([('id', pa.int32())])))
+
+ assert batches[0].column('id').to_pylist() == [7]
+ native.assert_called_once()
+
+
def test_native_read_falls_back_for_unsupported_dedicated_file():
read = _table_read()
schema = pa.schema([('id', pa.int32())])
diff --git a/paimon-python/pypaimon/tests/parquet_page_index_test.py
b/paimon-python/pypaimon/tests/parquet_page_index_test.py
index 8a33e0b1dc..5af33654d4 100644
--- a/paimon-python/pypaimon/tests/parquet_page_index_test.py
+++ b/paimon-python/pypaimon/tests/parquet_page_index_test.py
@@ -403,6 +403,7 @@ def
test_page_index_switch_bypasses_metadata_processing_when_disabled(fixture, v
@pytest.mark.parametrize('nested', [False, True])
[email protected]_plan
def test_table_option_and_copy_control_page_index_reads(tmp_path, nested):
from pypaimon import CatalogFactory, Schema
diff --git a/paimon-python/pypaimon/tests/query_auth_validation_test.py
b/paimon-python/pypaimon/tests/query_auth_validation_test.py
index d95affffab..25ed0392bb 100644
--- a/paimon-python/pypaimon/tests/query_auth_validation_test.py
+++ b/paimon-python/pypaimon/tests/query_auth_validation_test.py
@@ -23,6 +23,7 @@ import unittest
from types import SimpleNamespace
import pyarrow as pa
+import pytest
from pypaimon.common.options import CoreOptions, Options
from pypaimon.read.query_auth_split import QueryAuthSplit
@@ -833,6 +834,7 @@ class TestSchemaRefreshAsksTheCatalog(unittest.TestCase):
read.to_arrow([QueryAuthSplit(object(), auth)], parallelism=1)
self.assertEqual(table.catalog_environment.get_table_calls, 3)
+ @pytest.mark.python_read
def test_a_read_without_rules_never_asks_the_catalog(self):
fields = [_string_field(0, "id"), _string_field(1, "email")]
table = self._table(fields, fields)
diff --git a/paimon-python/pypaimon/tests/reader_parallel_test.py
b/paimon-python/pypaimon/tests/reader_parallel_test.py
index ea0c3d8f61..b31ba265e4 100644
--- a/paimon-python/pypaimon/tests/reader_parallel_test.py
+++ b/paimon-python/pypaimon/tests/reader_parallel_test.py
@@ -24,6 +24,7 @@ import unittest
from unittest import mock
import pyarrow as pa
+import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.read.table_read import TableRead, _RemainingRows
@@ -237,6 +238,7 @@ class ParallelReaderAppendOnlyTest(unittest.TestCase):
self.assertEqual(serial, auto)
self.assertEqual(auto.num_rows, self.expected_rows)
+ @pytest.mark.python_read
def test_default_none_auto_takes_parallel_path_when_multicore(self):
read = self.table.new_read_builder().new_read()
splits = self._scan_splits(self.table.new_read_builder())
@@ -263,6 +265,7 @@ class ParallelReaderAppendOnlyTest(unittest.TestCase):
splits = self._scan_splits(self.table_opt_4.new_read_builder())
read.to_arrow(splits, parallelism=1)
+ @pytest.mark.python_read
def test_method_arg_overrides_option_to_parallel(self):
# option=1 (forces serial) but caller passes 4: should enable
parallelism.
read = self.table_opt_1.new_read_builder().new_read()
@@ -349,6 +352,7 @@ class ParallelReaderAppendOnlyTest(unittest.TestCase):
df = rb.new_read().to_pandas(splits, parallelism=4)
self.assertEqual(len(df), limit)
+ @pytest.mark.python_read
def test_parallel_reader_error_propagates(self):
rb = self.table.new_read_builder()
splits = self._scan_splits(rb)
diff --git a/paimon-python/pypaimon/tests/rest/rest_base_test.py
b/paimon-python/pypaimon/tests/rest/rest_base_test.py
index 2ae483e6f1..763ac39023 100644
--- a/paimon-python/pypaimon/tests/rest/rest_base_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_base_test.py
@@ -199,22 +199,15 @@ class RESTBaseTest(unittest.TestCase):
self.assertTrue(os.path.exists(self.warehouse +
"/default/test_table/dt=p1"))
self.assertEqual(len(glob.glob(self.warehouse +
"/default/test_table/manifest/*")), 3)
- def _write_test_table(self, table):
+ def _write_test_table(self, table, data=None):
write_builder = table.new_batch_write_builder()
- table_pa_schema = self.pk_pa_schema if table.primary_keys else
self.pa_schema
+ if data is None:
+ data = self.pk_expected if table.primary_keys else self.expected
# first write
table_write = write_builder.new_write()
table_commit = write_builder.new_commit()
- data1 = {
- 'user_id': [1, 2, 3, 4],
- 'item_id': [1001, 1002, 1003, 1004],
- 'behavior': ['a', 'b', 'c', None],
- 'dt': ['p1', 'p1', 'p2', 'p1'],
- 'long-dt': ['2024-10-10', '2024-10-10', '2024-10-10',
'2024-01-01'],
- }
- pa_table = pa.Table.from_pydict(data1, schema=table_pa_schema)
- table_write.write_arrow(pa_table)
+ table_write.write_arrow(data.slice(0, 4))
table_commit.commit(table_write.prepare_commit())
table_write.close()
table_commit.close()
@@ -222,15 +215,7 @@ class RESTBaseTest(unittest.TestCase):
# second write
table_write = write_builder.new_write()
table_commit = write_builder.new_commit()
- data2 = {
- 'user_id': [5, 6, 7, 8],
- 'item_id': [1005, 1006, 1007, 1008],
- 'behavior': ['e', 'f', 'g', 'h'],
- 'dt': ['p2', 'p1', 'p2', 'p2'],
- 'long-dt': ['2024-10-10', '2025-01-23', 'abcdefghijklmnopk',
'2025-08-08'],
- }
- pa_table = pa.Table.from_pydict(data2, schema=table_pa_schema)
- table_write.write_arrow(pa_table)
+ table_write.write_arrow(data.slice(4, 4))
table_commit.commit(table_write.prepare_commit())
table_write.close()
table_commit.close()
diff --git a/paimon-python/pypaimon/tests/rest/rest_read_write_test.py
b/paimon-python/pypaimon/tests/rest/rest_read_write_test.py
index f46ce5444b..4c9a9fd066 100644
--- a/paimon-python/pypaimon/tests/rest/rest_read_write_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_read_write_test.py
@@ -34,6 +34,12 @@ import ray
class RESTTableReadWriteTest(RESTBaseTest):
+ @staticmethod
+ def _avro_field_names(data):
+ return data.rename_columns([
+ 'long_dt' if name == 'long-dt' else name for name in
data.schema.names
+ ])
+
def test_overwrite(self):
simple_pa_schema = pa.schema([
('f0', pa.int32()),
@@ -201,14 +207,15 @@ class RESTTableReadWriteTest(RESTBaseTest):
self.assertEqual(actual, self.expected)
def test_avro_ao_reader(self):
- schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'], options={'file.format': 'avro'})
+ expected = self._avro_field_names(self.expected)
+ schema = Schema.from_pyarrow_schema(expected.schema,
partition_keys=['dt'], options={'file.format': 'avro'})
self.rest_catalog.create_table('default.test_append_only_avro',
schema, False)
table = self.rest_catalog.get_table('default.test_append_only_avro')
- self._write_test_table(table)
+ self._write_test_table(table, expected)
read_builder = table.new_read_builder()
actual = self._read_test_table(read_builder).sort_by('user_id')
- self.assertEqual(actual, self.expected)
+ self.assertEqual(actual, expected)
def test_lance_ao_reader(self):
schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'], options={'file.format': 'lance'})
@@ -286,15 +293,15 @@ class RESTTableReadWriteTest(RESTBaseTest):
self.assertEqual(actual, expected)
def test_avro_ao_reader_with_projection(self):
- schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'], options={'file.format': 'avro'})
+ expected = self._avro_field_names(self.expected)
+ schema = Schema.from_pyarrow_schema(expected.schema,
partition_keys=['dt'], options={'file.format': 'avro'})
self.rest_catalog.create_table('default.test_avro_append_only_projection',
schema, False)
table =
self.rest_catalog.get_table('default.test_avro_append_only_projection')
- self._write_test_table(table)
+ self._write_test_table(table, expected)
read_builder = table.new_read_builder().with_projection(['dt',
'user_id'])
actual = self._read_test_table(read_builder).sort_by('user_id')
- expected = self.expected.select(['dt', 'user_id'])
- self.assertEqual(actual, expected)
+ self.assertEqual(actual, expected.select(['dt', 'user_id']))
def test_ao_reader_with_limit(self):
schema = Schema.from_pyarrow_schema(self.pa_schema,
partition_keys=['dt'])
@@ -344,7 +351,8 @@ class RESTTableReadWriteTest(RESTBaseTest):
self.assertEqual(col_a, col_b)
def test_pk_avro_reader(self):
- schema = Schema.from_pyarrow_schema(self.pa_schema,
+ expected = self._avro_field_names(self.pk_expected)
+ schema = Schema.from_pyarrow_schema(expected.schema,
partition_keys=['dt'],
primary_keys=['user_id', 'dt'],
options={
@@ -353,11 +361,11 @@ class RESTTableReadWriteTest(RESTBaseTest):
})
self.rest_catalog.create_table('default.test_pk_avro', schema, False)
table = self.rest_catalog.get_table('default.test_pk_avro')
- self._write_test_table(table)
+ self._write_test_table(table, expected)
read_builder = table.new_read_builder()
actual = self._read_test_table(read_builder).sort_by('user_id')
- self.assertEqual(actual, self.pk_expected)
+ self.assertEqual(actual, expected)
def test_pk_lance_reader(self):
schema = Schema.from_pyarrow_schema(self.pa_schema,
diff --git a/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
b/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
index 7a9971c449..74bfec80f3 100644
--- a/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
+++ b/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
@@ -28,6 +28,7 @@ Two layers are covered:
is not revived, and a type change is cast at read time.
"""
+import decimal
import os
import shutil
import tempfile
@@ -560,6 +561,41 @@ class SchemaEvolutionNestedContainerTest(_NestedBase):
MapType(True, AtomicType('STRING'), AtomicType('BIGINT')))
self.assertEqual(out.to_pylist(), [[('b', 2)], None])
+ def test_nested_decimal_scale_down_rounds_and_nulls_overflow(self):
+ from pypaimon.read.reader.data_file_batch_reader import \
+ DataFileBatchReader
+ reader = DataFileBatchReader.__new__(DataFileBatchReader)
+ source_type = RowType(True, [
+ DataField(1, 'price', AtomicType('DECIMAL(6, 3)'))])
+ target_type = RowType(True, [
+ DataField(1, 'price', AtomicType('DECIMAL(3, 2)'))])
+ source = pa.array([
+ {'price': decimal.Decimal('9.994')},
+ {'price': decimal.Decimal('9.995')},
+ {'price': decimal.Decimal('-4.565')},
+ None,
+ ], type=pa.struct([pa.field('price', pa.decimal128(6, 3))]))
+
+ actual = reader._align_array_by_id(source, source_type, target_type)
+
+ self.assertEqual(actual.to_pylist(), [
+ {'price': decimal.Decimal('9.99')},
+ {'price': None},
+ {'price': decimal.Decimal('-4.57')},
+ None,
+ ])
+
+ def test_decimal_scale_down_preserves_max_precision_values(self):
+ from pypaimon.read.reader.data_file_batch_reader import \
+ cast_array_for_schema_evolution
+ value = decimal.Decimal('12345678901234567890123456789012345.678')
+ source = pa.array([value], type=pa.decimal128(38, 3))
+
+ actual = cast_array_for_schema_evolution(source, pa.decimal128(38, 2))
+
+ self.assertEqual(actual.to_pylist(), [
+ decimal.Decimal('12345678901234567890123456789012345.68')])
+
def test_map_wrapper_token_validated(self):
# The token consumed when descending through a MAP must be 'value'.
val = pa.struct([('a', pa.int64())])
diff --git a/paimon-python/pypaimon/tests/schema_evolution_read_test.py
b/paimon-python/pypaimon/tests/schema_evolution_read_test.py
index 916058812c..60705d5cad 100644
--- a/paimon-python/pypaimon/tests/schema_evolution_read_test.py
+++ b/paimon-python/pypaimon/tests/schema_evolution_read_test.py
@@ -247,12 +247,11 @@ class SchemaEvolutionReadTest(unittest.TestCase):
# CastExecutors), so 1.2/2.8 read back as 1/2.
("double_to_int", pa.float64(), pa.int32(), 'INT',
[1.2, 2.8], [1, 2], [3, 4]),
- # Lossy DECIMAL scale-down: (10,4) -> (10,2) truncates the extra
- # scale rather than raising.
+ # Java's DECIMAL cast rounds half-up when reducing scale.
("decimal_scale_down",
pa.decimal128(10, 4), pa.decimal128(10, 2), 'DECIMAL(10, 2)',
- [decimal.Decimal('1.2345'), decimal.Decimal('4.5678')],
- [decimal.Decimal('1.23'), decimal.Decimal('4.56')],
+ [decimal.Decimal('1.2345'), decimal.Decimal('-4.5650')],
+ [decimal.Decimal('1.23'), decimal.Decimal('-4.57')],
[decimal.Decimal('7.89'), decimal.Decimal('0.12')]),
]
@@ -307,16 +306,20 @@ class SchemaEvolutionReadTest(unittest.TestCase):
# Reading ONLY old-schema files after a lossy type change (no
# newer-schema file in the splits). The output type must equal the
# current read schema regardless of which files the read spans, and the
- # conversion must truncate to match Java CastExecutors rather than
- # raise. (A previous fix that relied on pyarrow's safe cast crashed
- # here on lossy evolutions.)
+ # DECIMAL scale reduction rounds half-up, while DOUBLE -> INT still
+ # truncates. Neither conversion should raise on these old-file reads.
import decimal
cases = [
("scale_down",
pa.decimal128(10, 4), pa.decimal128(10, 2), 'DECIMAL(10, 2)',
- [decimal.Decimal('1.2345'), decimal.Decimal('4.5678')],
- [decimal.Decimal('1.23'), decimal.Decimal('4.56')]),
+ [decimal.Decimal('4.5678'), decimal.Decimal('-4.5650')],
+ [decimal.Decimal('4.57'), decimal.Decimal('-4.57')]),
+ ("scale_and_precision_down",
+ pa.decimal128(6, 3), pa.decimal128(3, 2), 'DECIMAL(3, 2)',
+ [decimal.Decimal('9.994'), decimal.Decimal('9.995'),
+ decimal.Decimal('-9.995'), decimal.Decimal('999.999')],
+ [decimal.Decimal('9.99'), None, None, None]),
("double_to_int", pa.float64(), pa.int32(), 'INT',
[1.2, 2.8], [1, 2]),
]
@@ -336,7 +339,8 @@ class SchemaEvolutionReadTest(unittest.TestCase):
table_write = write_builder.new_write()
table_commit = write_builder.new_commit()
table_write.write_arrow(pa.Table.from_pydict(
- {'k': [1, 2], 'v': write_vals}, schema=old_schema))
+ {'k': list(range(1, len(write_vals) + 1)), 'v':
write_vals},
+ schema=old_schema))
table_commit.commit(table_write.prepare_commit())
table_write.close()
table_commit.close()
@@ -353,7 +357,8 @@ class SchemaEvolutionReadTest(unittest.TestCase):
actual = read_builder.new_read().to_arrow(
self._scan_table(read_builder))
expected = pa.Table.from_pydict(
- {'k': [1, 2], 'v': read_vals}, schema=new_schema)
+ {'k': list(range(1, len(read_vals) + 1)), 'v': read_vals},
+ schema=new_schema)
self.assertEqual(expected, actual)
def test_schema_evolution_with_scan_filter(self):
diff --git a/paimon-python/pypaimon/tests/test_early_row_range_filter.py
b/paimon-python/pypaimon/tests/test_early_row_range_filter.py
index 2253518a9f..b27184ab55 100644
--- a/paimon-python/pypaimon/tests/test_early_row_range_filter.py
+++ b/paimon-python/pypaimon/tests/test_early_row_range_filter.py
@@ -22,6 +22,7 @@ import unittest
from unittest.mock import patch
import pyarrow as pa
+import pytest
from pypaimon import CatalogFactory, Schema
from pypaimon.common.predicate import Predicate
@@ -77,6 +78,7 @@ class TestManifestReadRowRangePerformance(unittest.TestCase):
def tearDownClass(cls):
shutil.rmtree(cls.tempdir, ignore_errors=True)
+ @pytest.mark.python_plan
def test_scan_constructs_all_entries_without_early_row_range_filter(self):
"""With manifest.merge-min-count=1, all entries are in one manifest.
Querying with _ROW_ID BETWEEN 5 AND 14 should return 2 files, but
diff --git a/paimon-python/pypaimon/tests/variant_test.py
b/paimon-python/pypaimon/tests/variant_test.py
index fd5a6c9fef..6e85718d17 100644
--- a/paimon-python/pypaimon/tests/variant_test.py
+++ b/paimon-python/pypaimon/tests/variant_test.py
@@ -62,6 +62,7 @@ from pypaimon.data.variant_shredding import (
_NULL_VALUE_BYTES,
_build_array_value,
_build_object_value,
+ _extract_overflow_fields,
_encode_scalar_to_value_bytes,
assemble_shredded_column,
build_variant_schema,
@@ -746,6 +747,17 @@ class TestBuildBinary(unittest.TestCase):
gv = GenericVariant(obj_bytes, meta)
self.assertEqual(gv.to_python(), {'age': 30})
+ def test_build_object_orders_keys_not_metadata_ids(self):
+ metadata = _make_metadata('z', 'a')
+ key_dict = parse_metadata_dict(metadata)
+ scalar = _encode_scalar_to_value_bytes(1, pa.int64())
+ value = _build_object_value(
+ [(key_dict['z'], scalar), (key_dict['a'], scalar)], key_dict)
+ self.assertEqual(
+ [key_id for key_id, _ in _extract_overflow_fields(value)],
+ [key_dict['a'], key_dict['z']],
+ )
+
def test_build_array_empty(self):
arr_bytes = _build_array_value([])
gv = GenericVariant(arr_bytes, b'\x01\x00')
@@ -1075,6 +1087,31 @@ class TestShreddingWrite(unittest.TestCase):
overflow_gv = GenericVariant(overflow, result['metadata'])
self.assertIn('name', overflow_gv.to_python())
+ def test_unicode_overflow_and_rebuild_use_utf8_key_order(self):
+ bmp_key = '\uff21'
+ supplementary_key = '\U0001f600'
+ obj_fields = self._obj_fields_for('col', [('keep', 'BIGINT')])
+ original = GenericVariant.from_python({
+ 'keep': 0, bmp_key: 1, supplementary_key: 2})
+ key_dict = parse_metadata_dict(original.metadata())
+
+ shredded = decompose_variant(original, obj_fields)
+ self.assertEqual(
+ [key_id for key_id, _ in
_extract_overflow_fields(shredded['value'])],
+ [key_dict[bmp_key], key_dict[supplementary_key]],
+ )
+
+ schema =
build_variant_schema(shredding_schema_to_arrow_type(obj_fields))
+ rebuilt_value, metadata = rebuild(shredded, schema, key_dict)
+ self.assertEqual(
+ [key_id for key_id, _ in _extract_overflow_fields(rebuilt_value)],
+ [key_dict['keep'], key_dict[bmp_key], key_dict[supplementary_key]],
+ )
+ self.assertEqual(
+ GenericVariant(rebuilt_value, metadata).to_python(),
+ original.to_python(),
+ )
+
def test_decompose_absent_field_is_null(self):
"""A shredded field absent from the variant yields {value: None,
typed_value: None}."""
obj_fields = self._obj_fields_for('col', [('missing_field', 'BIGINT')])
