dbtsai opened a new pull request, #117: URL: https://github.com/apache/spark-connect-rust/pull/117
### What changes were proposed in this pull request? **Third alternative for [SPARK-60586](https://issues.apache.org/jira/browse/SPARK-60586), alongside #103 (custom per-batch C Data) and #108 (whole-result C Stream). These are alternatives, not a stack: choose one implementation rather than merge all three.** Implement [Tim's suggestion on #103](https://github.com/apache/spark-connect-rust/pull/103#issuecomment-6458087652): use `arrow::pyarrow::Table::try_new(...).into_pyarrow(py)` for the native client's `toArrow()` handoff. In Arrow 58.3.0 this imports each batch through a native C Stream reader, then calls `pyarrow.Table.from_batches()`. Unlike #108, it does not use one reader for the whole result. - Keep schema-consistency validation, a schema-only PyArrow compatibility preflight, and the NUL-name/timezone IPC fallback without another RPC. - Unlike #103, do not carry over custom C schema descriptor rebuilding and flag manipulation; scope the contract to ordinary Spark-produced results, not generic sorted-map or dictionary-encoded inputs. - Retain the small unpublished export crate and its focused tests for this comparison. The bridge is 137 lines including the license header and inline test, versus 237 in the C Data comparison source. - Keep **PyO3 0.28.3 and Arrow 58.3.0**. Enable Arrow's `pyarrow` feature, adding `arrow-pyarrow 58.3.0`; no PyO3 upgrade or scalar/UDF/import API changes. This is the exact local candidate used in the three-way benchmark below, based on upstream `c36c846fcb4f22c9995e5de5c69c45ae76a5ad9a`. #103 and #108 are unchanged. Feedback requested: is delegating the handoff to arrow-rs preferable to maintaining either custom bridge, given the measured tradeoffs? Whether to inline the remaining wrapper into `pyspark-rs` can be reviewed separately from the handoff choice. ### Why are the changes needed? Avoid serializing already-decoded Arrow batches to another IPC file and copying those bytes into Python just so PyArrow can decode them again. Using arrow-rs's existing conversion reduces custom FFI code and maintenance. The Spark Connect wire protocol does not change. ### Does this PR introduce _any_ user-facing change? `toArrow()` shares compatible decoded Arrow value and offset buffers with PyArrow, preserving values, batch boundaries, metadata, zero-column row counts, and ownership after Rust wrappers are dropped. `toPandas()` uses this handoff but pandas materialization may still copy. As in the other proposals, inconsistent batch schemas, including metadata, now raise `ValueError` instead of being written under the first batch's schema. An empty batch vector still returns an empty-schema table. Field names/timezones containing NUL retain the IPC fallback over already-collected batches. Zero-copy applies to compatible Arrow data buffers, not network IPC decoding, descriptors, Python rows, pandas materialization, or the fallback. Sliced validity bitmaps may require realignment. All batches are collected before handoff: this is **not network streaming or bounded-memory execution**. Sorted-map flags and dictionary-encoded results are outside this implementation's Spark-result contract. See `crates/spark-python-arrow/README.md`. ### How was this patch tested? Local macOS arm64 validation using Rust 1.99.1 and optimized release (`opt-level=3`, `abi3-py39`) builds: - **11 converter tests passed**, including buffer-address identity/lifetime, Spark map layouts, slices, metadata, empty shapes, mismatched schemas, invalid types, and NUL fallback. Strict converter Clippy and workspace formatting passed. - **292 Python tests passed on each environment:** Python 3.12.14/PyArrow 26.0.0 and Python 3.9.6/PyArrow 10.0.0. These include the existing offline suite and 19 public Arrow/pandas handoff cases. - **360/360 fresh benchmark samples passed the harness's result checks.** Independent audit verified those success records, exact RPC counts/payload bytes, execution-order balance, all metric summaries, source/wheel/native hashes, and unchanged identities across both rounds. #### Three-way benchmark Used [dbtsai/pyspark-connect-benchmark at `1bbdec9`](https://github.com/dbtsai/pyspark-connect-benchmark/tree/1bbdec9f3665fcd2ec2ee5129bbcdb95ee4b00c4), driven by a local three-Rust-client scheduler: 10 workloads x 6 fresh processes x 3 implementations x 2 counterbalanced rounds. All three builds used the same upstream base, compiler/release/ABI settings, PyO3 0.28.3, Arrow 58.3.0, Python 3.12.14, PyArrow 26.0.0, and common Python dependencies. Host: Apple M3 Max, 16 logical CPUs, 64 GiB RAM. Representative median wall times in milliseconds. Each cell is **forward / reversed round**; rounds are not pooled and no earlier timings are reused. | `toArrow()` workload | C Data (#103) | C Stream (#108) | arrow-rs Table (this PR) | |---|---:|---:|---:| | 38 MiB numeric, 10 batches | 14.930 / 14.795 | 14.954 / 14.659 | 15.159 / 14.495 | | 128 MiB numeric, 128 batches | 31.721 / 36.810 | 31.044 / 35.972 | 31.038 / 38.291 | | Cold 512-column result | 8.830 / 9.204 | 8.937 / 8.928 | 9.883 / 10.134 | | 1,048,576 rows, 1,049 batches | 27.674 / 27.745 | 27.774 / 27.704 | 29.341 / 29.185 | Median sampled peak RSS in MiB, also **forward / reversed**: | Workload | C Data | C Stream | Table | |---|---:|---:|---:| | 38 MiB numeric | 147.45 / 147.21 | 146.82 / 146.80 | 147.92 / 147.16 | | 128 MiB numeric | 235.39 / 239.58 | 235.30 / 243.85 | 235.97 / 244.48 | | Cold 512 columns | 127.18 / 127.16 | 127.08 / 127.10 | 127.72 / 127.71 | | 1,049 batches | 112.31 / 112.20 | 112.19 / 112.16 | 112.52 / 112.58 | There is **no clear overall winner between C Data and C Stream**. The Table helper is about **10-14% slower for cold wide results** and **5-6% slower for 1,049 batches** in both rounds; the latter also uses about 8-9% more CPU time. Large numeric rankings change between rounds. This proposal offers less custom code, not a demonstrated speed or memory improvement over the other two handoffs. Most per-round peak-RSS medians differ by at most 1.10 MiB, excluding the 128 MiB workload. That workload has an unresolved reversed-round tail: Table peaks at **280.42 MiB**, versus **256.53 MiB for C Stream** and **244.70 MiB for C Data**. Six samples under background load do not establish a regression, but neither do they establish memory equivalence. These are **exploratory shared-desktop measurements** of client receive/conversion against canned localhost RPC responses, not Spark execution or a quiet-host comparison. No competing build/benchmark was detected immediately before launch, but background desktop processes remained active. Most cases warm the full result action; the cold-wide case does not. RSS sampling targets 2 ms (largest observed gap 3.47 ms), can miss short peaks, and is not allocation/leak measurement. Buffer-address and lifetime tests, not benchmark timings, establish buffer sharing. #### Remaining draft checks - [ ] Full official Spark Connect parity suite and minimum-Rust build. The converter declares Rust 1.85 to match Arrow 58.3's requirement; local validation used 1.99.1. - [ ] Review CI and complete repository-wide pre-push validation. Prior comparison builds reproduced workspace Clippy errors in unchanged core tests and the existing `SparkSession.Hook` API gap; this PR does not claim to fix them. Strict converter Clippy passes. ### 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]
