[JingsongLi]

Purpose

Companion apache/paimon-rust#1187 has merged as e602ec58. Native CI continues to build Rust main; no dependency override or old-extension fallback is needed.

Restrict execution to corresponding Python and Rust implementations of Paimon operations. Shared control logic, transport adapters and third-party framework integrations stay in their owning modules. This also corrects excessive moves introduced in #10699 and #10707.

• Restore Torch datasets, Ray datasources, split providers and Ray datasinks to their original modules. Keep their actual implementations, worker closures, public identities, signatures and override hooks there.
• Restore Ray index task scheduling to pypaimon.ray.vector_index_build and common system-table Arrow output to pypaimon.table.system. Keep the Native system metadata adapter separate.
• Move shared backend selection, request coordination, input/output conversion, explain rendering and resource helpers from the execution root into read, write, globalindex and table.source. Merge shared write input helpers into the existing write.input_adapter module. Restore TableRead's framework conversion methods inline.
• Keep Python file scanning/reading, writing, updates, commits and index algorithms in execution/python, and corresponding Rust adapters in execution/native. Match corresponding capabilities rather than requiring identical module names. The backends remain independent.
• Retain the MERGE improvements: shared condition and transform helpers live in write, so local Python and Native MERGE do not load Ray orchestration. Preserve public Ray helper entry points and signatures.
• Retain the Vindex initialization/close fixes, including concurrent and reentrant callbacks, without replaying Native opening/query failures through Python.
• Retain source cleanup and cancellation fixes. Ray/Torch readers close on completion and errors; cleanup preserves a primary read/conversion failure; Torch producers observe cancellation while waiting on a full queue.
• Document the boundary and add fresh-process import, source-layout and wheel/sdist checks to prevent shared/framework code from moving into execution again.

Existing public API paths, signatures, defaults, class identities and supported hooks are preserved. Stream controls and shard close are additive. No new runtime dependencies. The stream conversion rollback uses the companion Rust scan-state copy API. Native tests used the existing extension without rebuilding or installing it.

Initial layout verification

• Compared 304 public declarations against merged [Python] Decouple public APIs from execution backends #10707 (062b22e0ad): signatures, annotations, decorators and class bases unchanged. Independent review also checked framework serialization and public hooks.
• Final execution selection, backend isolation, Vindex, MERGE, search and writer regressions: 187 passed.
• Torch and Ray sink suite: 87 passed, including real distributed Torch workers. Framework/provider/postpone boundaries: 38 passed. Real Ray Data read/write/BLOB/nested projection cases: 3 passed.
• Real Ray index and local index/boundary regressions: 59 passed.
• System-table/auth/read boundary regressions: 563 passed, 6 skipped. Real Native system-table tests using the installed extension: 66 passed. Restored TableRead framework methods and related adapters: 52 passed.
• Wheel/sdist packaging: 9 passed. Whole-package Flake8, all 1,173 license headers, changed/new Python 3.6 grammar (135 files) and diff checks passed.
• Repeated independent reviews checked module ownership, import cycles, backend dependencies, framework cleanup, public contracts and packaging. Findings were corrected and the affected checks rerun. These focused groups overlap; the full Python/Native suites were not repeated.

Execution API and semantic alignment

Audit the corresponding operations against current Java master (33a8cd34d3) and Rust main (e4ec3a45). The relevant Java/Rust reference APIs are unchanged from the local reference implementations. Companion apache/paimon-rust#1187 adds Core scan-state cloning and the standard Python copy protocol for transactional Native result conversion.

• Give Python and Native commit executors the same five operation signatures. Keep message preparation and FFI conversion before publication. The common coordinator delegates without knowing the binding's batch/stream argument order. Preserve explicit file_store_commit access, overwrite scope, callbacks, empty commits, explicit abort and uncertain-publication file preservation.
• Give both stream scans the same eight control/lifecycle methods: plan, checkpoint, restore, watermark, notify_checkpoint_complete, stream, stream_sync and close. Plan once without polling for a new snapshot; use the iterator for continuous polling. Preserve each physical changelog frame across large backlogs.
• Align initial lookup/full-compaction level selection, empty-frame checkpoint progression and periodic expired/recreated checkpoint diagnosis with Java/Rust. Preserve unacknowledged consumer progress on cancellation, invalidate prefetch on restore/close, and keep failed Native split conversion from advancing the public checkpoint or watermark.
• Fix Python shard updates with deletion vectors: track the actual visible row IDs instead of physical-file capacity. Use the existing row-ID update implementation, pin the snapshot and require a complete read/update cycle. Add close and ensure early reader cancellation releases the source even before its first batch, preserving primary read errors and all prepared files.
• Align early Python BLOB callback/factory validation with Native and preserve a valid existing configuration when replacement fails.
• Document the operation correspondence and the common commit, stream and shard protocols. Internal Python mixins preserve public hooks; Native adapters own Rust handles and remain thin.

API alignment verification

• Final execution selection, isolation, read, writer, search, MERGE and Vindex boundary regressions: 201 passed.
• Commit protocol and related lifecycle/publication tests: 284 passed, 144 skipped in the Python environment. The new common-protocol tests cover operation signatures, scope, identifiers, properties and uncertain outcomes without replay/deletion. Real Native publication/overwrite/checkpoint/callback/abort tests using the existing extension: 19 passed.
• Stream/auth group: 358 passed, 11 skipped. Before the final Rust-dependent copy bridge, real shared Python/Native stream protocol and manifest/Parquet initial-level regressions had 38 passed, exercising 25 Native plans. The final bridge requires [core] Support copying stream scan state in Python bindings paimon-rust#1187; the local extension is deliberately not rebuilt or installed.
• Python shard alignment regressions: 19 passed, plus the existing 9 shard tests and 8 Python partition-carrier cases. Real Native shard ordering/deletion-vector/invalid-input/close/empty-projection comparisons: 9 passed. Independent verification repeated the 3 early-close/read-error resource cases successfully.
• The 61 existing public declarations touched in this follow-up retain their signatures, decorators and annotations; the public shard close method is additive. Both stream implementations' eight method signatures are checked in a fresh process.
• Whole-package Flake8, all 1,177 license headers, Python 3.6 grammar for 19 changed/new Python files and diff checks passed. Independent reviews revalidated shard source ownership and the final stream state repairs. Focused groups overlap; the full Python/Native suites were not repeated.

Known pre-existing capability difference: Python Data Evolution reads reject nested ROW field projection while Rust supports it. This PR does not add that Python read capability or expand Rust-only features.

Rust companion verification: Core clone behavior test passed; Python binding cargo check and both scoped Rust 1.98 Clippy gates passed. The new copy API owns callback references independently and uses the standard Python copy protocol. No Python version probes or extra metadata reads were added.

The companion Rust embedded-Python GC unit test also passed (1 case): both callback ownership forms, independent cursors and collection of both wrappers' reference cycles, without rebuilding or installing a Python extension.

Review against merged Rust main

Re-audited execution inputs, outputs, lifecycle and failure semantics against Rust main e602ec58 and current Java. Preserve all existing public signatures and keep both backends independent.

• Normalize stream write/update/commit identifiers through Python's integer index protocol into Java long / Rust i64 before consuming input, running callbacks, preparing files or publishing snapshots. Invalid identifiers cannot publish unreadable snapshots or consume an update operation.
• Reject empty or duplicate write-column selections before changing a valid existing configuration or selecting a different backend. Java rejects duplicate RowType fields and Parquet rejects an empty physical schema. Pure previously committed two zero-column rows whose full read returned zero rows.
• Preserve the primary Pure read/conversion error when source close also fails in row, serial batch and parallel worker paths. EOF and explicit cancellation still surface cleanup errors, matching the existing common resource contract.
• Normalize global-index build/drop partition predicates from either table or partition schema positions. Reuse public partition validation and existing name rebinding, preserve the original predicate and reject non-partition conditions without broadening deletion scope.
• Document that corresponding operation protocols do not require identical private helpers or capabilities. Pure DE nested ROW projection remains unsupported, and ordinary Pure Plan.stats() remains optional while Native records lightweight counters. Post-write reconfiguration restrictions and the Native batch adapter's existing optional argument syntax are unchanged.

Final coherent protocol, regression and backend-boundary group: 256 passed, 8 skipped (Native integration cases require an available SDK). Additional writer/lifecycle regressions: 206 passed, 50 skipped; existing index-build/partition-predicate regressions: 68 passed. Groups overlap. Independent reviews revalidated input boundaries, immutable predicates, exact index deletion scope, cleanup errors and backend isolation. No Python extension was installed or rebuilt.

Final static gates passed: whole-package Flake8 using the repository configuration, all 1,182 Python license headers, changed/new Python 3.6 grammar (19 files), diff checks, and 286 existing class/function contracts unchanged versus the preceding HEAD. The separate broad RAT scan found only a pre-existing untracked dev/log runtime log; no source-header failures were found.

[leaves12138]

LGTM