dbtsai opened a new pull request, #108: URL: https://github.com/apache/spark-connect-rust/pull/108
### What changes were proposed in this pull request? **Alternative implementation to #103, for design review under the same [SPARK-60586](https://issues.apache.org/jira/browse/SPARK-60586). These PRs are alternatives, not a stack: please choose one approach rather than merge both.** Replace the extra IPC reserialization in native `DataFrame.toArrow()` with an Arrow C Stream handoff. The same unpublished `spark-python-arrow` export crate and client integration as #103 are retained, but PyArrow imports one `RecordBatchReader` and calls `read_all()` instead of importing each batch through Python. | | #103: C Data | This PR: C Stream | |---|---|---| | Batch handoff | One Python `RecordBatch._import_from_c` call per batch, then `Table.from_batches` | One `RecordBatchReader._import_from_c` call, then `read_all`; batches pass through native callbacks | | Schema validation | Standalone schema import | Same schema-only preflight before the reader import, including for PyArrow 10 | | Ownership | Each array descriptor is consumed separately | Reader owns the stream; yielded array descriptors retain independent buffer ownership | | Additional implementation | Direct batch loop | GIL-independent stream callback adapter; delegates array iteration/release to Arrow and preserves schema ordering flags | Both approaches share compatible Arrow value/offset buffers. Both collect all decoded batches into a Rust `Vec` first: C Stream here is **not network streaming or bounded-memory execution**. PyO3 stays at **0.28.3**. There are no row/scalar, UDF-worker, or Python-to-Rust import changes. This branch starts from upstream `c36c846fcb4f22c9995e5de5c69c45ae76a5ad9a`; #103 is unchanged. Feedback requested: is avoiding the per-batch Python calls worth the additional stream callback/ownership surface, or is #103's direct C Data path preferable? ### Why are the changes needed? Both proposals avoid serializing already-decoded Rust Arrow batches into a new IPC file and copying it into Python bytes just so PyArrow can decode it again. This alternative lets Arrow's native reader loop consume those batches without a Python call or Python list entry for each batch. The Spark Connect wire protocol is unchanged. ### Does this PR introduce _any_ user-facing change? `toArrow()` shares compatible value and offset buffers while preserving values, batch boundaries, metadata, nested ordering flags, and zero-column row counts. `toPandas()` uses the same handoff, but pandas materialization can still copy. As in #103, inconsistent batch schemas (including metadata and dictionary ordering) raise `ValueError` instead of being written under the first schema. An empty batch vector still yields the existing empty-schema result. NUL-containing field names/timezones use the IPC compatibility path over the collected batches, without repeating the RPC. Zero-copy does not cover network IPC decoding, schema descriptors, Python rows, pandas conversion, or the fallback. Sliced validity bitmaps can require realignment. See `crates/spark-python-arrow/README.md` for the contract. ### How was this patch tested? Local macOS arm64 validation, Rust 1.99.1, optimized release (`opt-level=3`, `abi3-py39`) wheels: - **14 converter tests passed**, including buffer identity/lifetime, nested schemas, sorted maps, dictionary ordering, empty shapes, NUL fallback, repeated schema callbacks, and error-string lifetime/forwarding. Strict converter Clippy and workspace formatting passed. - **292 Python tests passed on each of Python 3.12/PyArrow 26 and Python 3.9/PyArrow 10**: 273 existing offline tests plus 19 handoff cases. The handoff tests forbid per-batch Python C Data imports and verify stream/array release exactly once, rejection before/after ownership transfer, partial-read failure, and buffers outliving the session/table. - **386 core Rust unit tests passed**; all integration tests compiled with and without `wasm-udf`. - The freshly rebuilt C Data comparison wheel passed its own **291 Python tests**, which forbid C Stream imports. Native binary hashes and installed wheel identities were checked before measurement. - Public API names across 12 classes are identical between the two variants. This is not a full behavioral/signature parity claim. #### Head-to-head benchmark Used [pyspark-connect-benchmark at `1bbdec9`](https://github.com/dbtsai/pyspark-connect-benchmark/tree/1bbdec9f3665fcd2ec2ee5129bbcdb95ee4b00c4), with C Data and C Stream built from the same upstream base and the same optimized toolchain, PyO3 0.28.3, Arrow 58.3.0, Python 3.12.14, and Python dependencies. This is a fresh comparison between the two handoffs, not a reuse of the earlier IPC baseline numbers. **288/288 samples passed:** 8 workloads x 6 fresh processes x 3 clients (C Data, C Stream, official PySpark) x 2 slot-swapped rounds. All Rust RPC counters matched pairwise, execution positions were balanced, and source/wheel identities stayed unchanged. Rounds are reported separately, without pooling. Every table cell is **C Data / C Stream**; peak RSS is the median of the sampled per-process peaks. | Workload | Forward median ms | Reversed median ms | Forward peak RSS MiB | Reversed peak RSS MiB | |---|---:|---:|---:|---:| | Wide numeric, Arrow | 15.152 / 15.296 | 14.436 / 14.277 | 147.33 / 147.16 | 146.87 / 147.20 | | Wide numeric, pandas | 16.101 / 15.963 | 15.866 / 15.924 | 187.21 / 187.02 | 187.12 / 187.06 | | Strings, 100 batches | 5.677 / 5.207 | 5.029 / 5.310 | 111.55 / 111.13 | 110.84 / 111.28 | | Strings, 10 batches | 3.448 / 3.211 | 3.251 / 3.188 | 111.95 / 112.00 | 111.96 / 111.49 | | First allocation, 4096 x 512 | 9.201 / 9.550 | 9.243 / 9.250 | 127.12 / 127.48 | 127.28 / 127.42 | | Single value | 0.516 / 0.473 | 0.532 / 0.531 | 102.33 / 102.41 | 102.80 / 102.62 | | Numeric, 1049 batches | 30.486 / 29.114 | 32.456 / 30.735 | 112.41 / 112.41 | 112.66 / 112.77 | | 128 MiB, 128 batches | 34.539 / 35.613 | 36.219 / 36.584 | 241.68 / 236.08 | 240.80 / 238.52 | The results do **not** establish a general speedup. C Stream was 4.5% and 5.3% faster for 1049 small batches, but 3.1% and 1.0% slower for the 128 MiB case. Most other differences were small or changed direction. Peak RSS was broadly similar; the 128 MiB case showed 2.3-5.6 MiB lower sampled peaks with C Stream. These are **exploratory shared-host measurements**, not load-isolated results; an unrelated benchmark was observed active shortly before this run. The workload measures client actions receiving canned localhost gRPC responses, not real Spark execution. All cases except first-allocation warm the full result action. RSS samples the whole client process at a 2 ms target interval and can miss short spikes; it is not an allocation count. Buffer-identity and lifetime tests, not timings, establish the zero-copy ownership behavior. #### Remaining draft checks - [ ] Full official parity suite and minimum-compiler build. The converter declares Rust 1.85 because Arrow 58.3 requires it; local builds used 1.99.1. - [ ] Review CI. Workspace Clippy reports approximate-constant errors in unchanged core tests, reproduced in the C Data comparison tree; strict converter Clippy passes. The pre-existing `SparkSession.Hook` API gap is unchanged. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (application version unavailable). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
