[python] Add opt-in Rust native data reads by JingsongLi · Pull Request #9954 · apache/paimon

JingsongLi · 2026-09-18T05:50:05Z

What changed

add the read.native.enabled opt-in table option
preserve opaque Rust splits from native planning and send them to the Rust reader
use Rust-produced PyArrow batches in to_arrow and to_arrow_batch_reader
retain exact PyPaimon limit and schema behavior
use effective split parallelism to run independent Rust readers concurrently
propagate blob_parallelism to Rust and cap aggregate split-reader × BLOB concurrency at 64
fall back to the Python reader for unsupported controls, transformed/query-auth splits, or native setup failures
cover Data Evolution, including native .blob reads, and treat row-id result ordering as unspecified
require the current unpublished paimon-rust API and validate that native batches already match the requested Arrow field types

Rust binding

Streaming batches are provided by apache/paimon-rust#870. Configurable native BLOB concurrency is provided by apache/paimon-rust#871. Native BLOB values use Arrow LargeBinary end to end in apache/paimon-rust#874, which should merge before this PR.

Tests

native read/plan unit tests: 64 passed, 53 subtests passed
native Data Evolution BLOB integration tests: 2 passed
Python Data Evolution tests excluding Vortex: 99 passed, 10 deselected, 21 subtests passed
paimon-rust BLOB core tests: 129 passed, excluding 2 REST token tests blocked by the sandbox
paimon-rust Data Evolution tests: 75 passed
paimon-rust DataFusion BLOB tests: 29 passed
paimon-rust Python read tests: 81 passed
cargo check -p pypaimon_rust

leaves12138

Reviewed at 4a2f625a706e5c6bed0d640b993f35e55778208c, using the native binding built from paimon-rust ee8a90323a6aa5203ed44cc842c5fb6c33f10a3a.

I reproduced three regressions, detailed in the inline comments: BLOB Arrow-schema compatibility, restored legacy partition paths, and deferred BLOB resolution under LIMIT. I would address these before merging.

The existing native-read and native-plan integration tests passed (36 tests). Twelve additional Python/native comparison checks produced 7 passes and 5 failures; all five failures were on the native path and correspond to these three issues.

leaves12138 · 2026-09-18T09:45:45Z

+            if remaining is not None and batch.num_rows > remaining:
+                batch = batch.slice(0, remaining)
+            batch = self._project_batch_to_output(batch)
+            yield self._try_to_pad_batch_by_schema(batch, schema)


Could we normalize native batches to the requested PyPaimon Arrow schema, including field types, before returning them on both the serial and parallel paths?

For a BLOB column, PyPaimon expects large_binary, but the native reader returns binary. _try_to_pad_batch_by_schema() returns the batch unchanged when the column names match, so it does not handle this difference.

I reproduced this with a two-row Data Evolution table containing an INT column and a BLOB column: the Python path returns large_binary, while native to_arrow() returns binary. More importantly, to_arrow_batch_reader(...).read_all() fails with ArrowInvalid: Schema at index 0 was different, because the reader declares large_binary but emits binary batches. Comparing only to_pydict() in the current tests misses this regression.

Please add schema-equality assertions and a read_all() regression test, in addition to checking the row values.

Fixed in be669d5. Native batches are now normalized against the full requested Arrow schema: matching names are no longer sufficient, mismatched column types are cast (including binary -> large_binary), and the batch is rebuilt with the target schema. Added serial RecordBatchReader.read_all() and parallel native-read regressions, and the live Data Evolution BLOB test now asserts exact schema equality for both to_arrow() and read_all().

Follow-up: fixed the root cause in apache/paimon-rust#874. BLOB now maps to Arrow LargeBinary across scalar/nested format reads, descriptor resolution, Data Evolution, and DataFusion, while BINARY/VARBINARY remain Binary. This PR no longer casts legacy Binary BLOB batches or probes older unpublished Rust bindings; it requires matching native field types and keeps the serial read_all plus parallel schema regressions.

leaves12138 · 2026-09-18T09:45:45Z

+        # conversion. Any Python split transformation creates a fresh object
+        # without this marker and thus safely falls back to the Python reader.
+        for split, rust_split in zip(splits, rust_splits):
+            split._native_split = rust_split


The retained Rust split can become stale immediately after this assignment: _restore_python_partition_paths() updates the Python file paths in place, but does not update or invalidate _native_split.

I reproduced this with a PyPaimon-written table partitioned by a string column whose value is a/b. The Python reader successfully reads the legacy part=a/b/... location, while enabling read.native.enabled makes Rust read part=a%2Fb/... and fail with NotFound. The existing legacy-partition integration test exercises native planning with Python reading, so it does not cover the new native-read path.

Could we synchronize the corrected paths into the Rust split, or remove the native marker whenever path restoration changes the Python split so this case falls back safely?

Fixed in be669d5. Whenever _restore_python_partition_paths() changes a file path, it now invalidates that split’s retained _native_split, so _try_native_batches() falls back before invoking Rust. The integration test now enables both native planning and native reading for the a/b legacy partition case, verifies the rows, and asserts that native_read was not called.

leaves12138 · 2026-09-18T09:45:45Z

+                "Native read failed, falling back to the Python reader: %s", e)
+            return None
+        blob_parallelism_available = native_blob_parallelism_available()
+        if ((blob_parallelism is not None and blob_parallelism > 1)


Support for with_blob_parallelism() does not guarantee that a pruning LIMIT is applied before BLOB payload resolution. Could we keep deferred-BLOB reads with a potentially pruning LIMIT on the Python path until the native reader has that capability, or pass the remaining row budget into Rust before payloads are fetched?

The native iterator resolves a whole batch before _convert_native_batches() slices it to the requested limit. Making split processing serial only prevents cross-split prefetch; it does not prevent payload reads for discarded rows within that batch.

With 32 rows referencing 32 distinct, valid local payload files, LIMIT 1 returns the same one row on both paths, but filesystem open-event tracking shows that Python opens 1 payload file and native opens all 32. With two rows where only the second, discarded row references a missing payload, Python succeeds and native fails while opening that unused payload. This regresses the existing deferred-BLOB LIMIT behavior and can be expensive for image/video workloads.

Please add a native-read regression test that checks payload I/O, not just the final row count.

Fixed in be669d5. A Data Evolution read with deferred BLOB fields now falls back before native invocation whenever its LIMIT may prune rows. Limits proven to cover all split rows remain native-eligible. The new live regression tracks BlobRef.to_data() calls and verifies LIMIT 1 materializes exactly one payload while native_read is never invoked.

leaves12138

Re-reviewed at dec74fef3af4795a12ef91622711a3f26c18a87c using the native binding built from apache/paimon-rust#874 (fa324609dbd7b78a5d836fe17be3fb3be0d7e19e).

Confirmed that the scalar BLOB Arrow-schema/read_all issue and the restored legacy partition-path issue are fixed. Two issues remain, reproduced with the real binding and detailed below: the LIMIT fallback misses descriptor-backed BLOB fields, and TIMESTAMP(0)/TIMESTAMP_LTZ(0) still have incompatible Arrow timestamp units.

leaves12138 · 2026-09-18T12:02:48Z


+    def _deferred_blob_limit_may_prune(self, splits: List[Split]) -> bool:
+        return (self.limit is not None
+                and self._deferred_blob_fields


Could we include descriptor-backed BLOB fields in the native LIMIT fallback instead of relying only on _deferred_blob_fields?

deferred_blob_field_names() explicitly excludes fields configured through blob-descriptor-field, so this guard is false for a Data Evolution table with blob-descriptor-field=payload, even when LIMIT will discard rows. Rust then resolves the whole batch before the Python wrapper applies the output limit.

I re-ran this against the current head and the binding from paimon-rust#874. With 32 rows referencing 32 distinct, valid local payload files and LIMIT 1, filesystem open-event tracking shows that Python opens 1 payload while native opens all 32. This happens with both to_arrow(..., parallelism=1) and to_arrow_batch_reader(...).read_all(). With two rows where only the second, discarded descriptor references a missing payload, Python succeeds while native fails with NotFound.

The added regression covers dedicated .blob fields, but not this descriptor-backed case. Please extend the native-read eligibility check to cover these payload-resolving fields, and add a real-binding regression for blob-descriptor-field that checks discarded-payload I/O, not just the returned row count.

Fixed in b606d8a. Native LIMIT eligibility now also detects configured blob-descriptor-field/blob-view-field payloads and falls back to the Python reader when the limit may prune rows. Added a real-binding regression with a discarded missing descriptor payload and verified only the retained payload is opened.

leaves12138 · 2026-09-18T12:02:48Z

        for field in target_schema:
            if field.name in batch.schema.names:
                col = batch.column(field.name)
+                if col.type != field.type:


Fixing BLOB in paimon-rust#874 resolves that type mismatch, but valid TIMESTAMP(0) and TIMESTAMP_LTZ(0) columns still fail this check.

PyPaimon's PyarrowFieldParser maps precision 0 to timestamp[s] (with UTC for LTZ), while Rust's timestamp_time_unit() maps precisions 0 through 3 to milliseconds. I reproduced this with an ordinary two-row Parquet table, using the actual #874 binding: Python reading succeeds, but native reading raises TypeError: Batch field 'value' has type timestamp[ms], expected timestamp[s]. Both to_arrow() and to_arrow_batch_reader(...).read_all() fail; the LTZ case fails with the corresponding UTC types.

Could we align the timestamp representations at the native/Python boundary, or temporarily fall back for these types rather than only asserting equality? Please cover both TIMESTAMP(0) and TIMESTAMP_LTZ(0) with real native-read schema tests.

Fixed in b606d8a. Native reads now fall back for precision-zero timestamp[s] fields, including TIMESTAMP_LTZ(0) with UTC, because the current Rust binding emits milliseconds for these precisions. The schema check is recursive for nested Arrow types, and real-binding regression tests cover both timestamp variants.

leaves12138

LGTM. Re-reviewed the latest changes and confirmed that the previously reported issues are resolved: BLOB Arrow-schema compatibility, restored legacy partition paths, descriptor-backed BLOB reads under a pruning LIMIT, and precision-zero timestamp compatibility.

The latter two cases now correctly fall back to Python. Verified that LIMIT 1 opens only the retained payload, including when a discarded payload is missing, and that TIMESTAMP(0)/TIMESTAMP_LTZ(0) preserve values and schemas in both materialized and streaming reads. Nested timestamp cases also pass, while compatible timestamp and BLOB reads remain native.

191 focused tests passed locally. Vortex and the full CI suite were not run. No remaining blocking issues found.

JingsongLi added 4 commits September 18, 2026 13:48

[python] Add opt-in Rust native data reads

7f6c108

[python] Validate native reads on data evolution tables

762d2d3

[python] Parallelize native split reads

a5215d7

[python] Support native BLOB read parallelism

4a2f625

leaves12138 reviewed Sep 18, 2026

View reviewed changes

[python] Fix native read compatibility regressions

be669d5

JingsongLi mentioned this pull request Sep 18, 2026

fix(arrow): map BLOB to LargeBinary apache/paimon-rust#874

Merged

[python] Require current native BLOB schema

dec74fe

leaves12138 reviewed Sep 18, 2026

View reviewed changes

[python] Fix native BLOB limit and timestamp fallbacks

b606d8a

leaves12138 approved these changes Sep 18, 2026

View reviewed changes

JingsongLi added 2 commits September 18, 2026 21:05

[python] Allow query auth masking type casts

fbb0636

[python] Fix native read lint formatting

2ebb0d2

JingsongLi merged commit 2650ae1 into apache:master Sep 18, 2026
13 checks passed

Conversation

JingsongLi commented Sep 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changed

Rust binding

Tests

Uh oh!

leaves12138 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leaves12138 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leaves12138 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

JingsongLi commented Sep 18, 2026 •

edited

Loading