[SPARK-60586][CONNECT] Export Arrow tables with arrow-rs PyArrow conversion

.github/workflows/coverage.yml

@@ -226,7 +226,9 @@ jobs:
 
       - name: Measure Python coverage with offline test suite
         run: |
-          cd python && PYTHONPATH=$(pwd) python -m coverage run --rcfile=.coveragerc -m pytest tests/test_dropin_offline.py
+          cd python
+          PYTHONPATH=$(pwd) python -m coverage run --rcfile=.coveragerc -m pytest \
+            tests/test_dropin_offline.py tests/test_arrow_handoff.py
           PYTHONPATH=$(pwd) python -m coverage json --rcfile=.coveragerc -o coverage-python.json
 
       - name: Parse Python coverage and create badge JSON

.github/workflows/python-arrow.yml

@@ -0,0 +1,59 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+name: Python Arrow export
+
+on:
+  push:
+    branches: [master, "branch-*"]
+  pull_request:
+    branches: [master, "branch-*"]
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  converter:
+    name: Python Arrow export
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - name: Install Rust
+        run: |
+          rustup toolchain install stable --profile minimal --component clippy
+          rustup default stable
+      - name: Install PyArrow
+        run: python -m pip install "pyarrow==26.0.0"
+      - name: Test Arrow table export
+        # Test separately from pyspark-rs: extension-module feature unification
+        # would disable libpython linking for the embedded Python tests.
+        run: |
+          cargo test -p spark-python-arrow --locked --features pyo3/abi3-py39
+      - name: Check converter Clippy
+        run: |
+          cargo clippy -p spark-python-arrow \
+            --locked --all-targets --features pyo3/abi3-py39 -- -D warnings

Cargo.toml

@@ -24,6 +24,7 @@ members = [
     "crates/spark-connect-macros",
     "crates/spark-connect-build",
     "crates/pyspark-rs",
+    "crates/spark-python-arrow",
     "examples",
 ]
 exclude = [
@@ -34,6 +35,8 @@ exclude = [
 # Rust-native `examples` crate both link the pure-Rust core directly; both are
 # workspace members but excluded from `default-members` so `cargo test`/`cargo build`
 # stay clean. Build them explicitly (`cargo build -p pyspark-rs` / `-p examples`).
+# `spark-python-arrow` is also explicit: its tests embed Python and require PyArrow.
+# Test it separately to avoid extension-module feature unification disabling libpython linking.
 #
 # The `examples/wasm-udfs` and `examples/wasm-udf-inline` crates are excluded from
 # the workspace entirely because they require the `wasm32-unknown-unknown` target,

crates/pyspark-rs/Cargo.toml

@@ -22,6 +22,7 @@ non_snake_case = "allow"
 spark-connect = { package = "apache-spark-connect", path = "../spark-connect" }
 spark-connect-core = { package = "apache-spark-connect-core", path = "../spark-connect-core" }
 spark-connect-proto = { package = "apache-spark-connect-proto", path = "../spark-connect-proto" }
+spark-python-arrow = { path = "../spark-python-arrow" }
 pyo3.workspace = true
 arrow.workspace = true
 tokio.workspace = true

crates/pyspark-rs/src/dataframe.rs

@@ -1166,19 +1166,15 @@ impl PyDataFrame {
     /// Collect the DataFrame into a `pyarrow.Table`.
     ///
     /// Mirrors `pyspark.sql.DataFrame.toArrow()`: the result is decoded from the
-    /// server's Arrow IPC stream and handed to `pyarrow` (no row-by-row copy).
+    /// server's Arrow IPC stream and handed to `pyarrow` through the shared
+    /// converter, sharing compatible decoded Arrow value buffers. Bitmap alignment
+    /// and schemas requiring the IPC compatibility fallback can still copy.
     #[pyo3(name = "toArrow")]
     fn to_arrow(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
-        use pyo3::types::PyBytes;
-        // `DataFrame::to_arrow` produces Arrow IPC *file* bytes (FileWriter).
-        let bytes = py.detach(|| self.dataframe.to_arrow()).to_pyerr()?;
-        let pa = py.import("pyarrow")?;
-        let py_bytes = PyBytes::new(py, &bytes);
-        let table = pa
-            .getattr("ipc")?
-            .call_method1("open_file", (py_bytes,))?
-            .call_method0("read_all")?;
-        Ok(table.unbind())
+        let batches = py
+            .detach(|| self.dataframe.collect_record_batches())
+            .to_pyerr()?;
+        spark_python_arrow::bridge::record_batches_to_pyarrow(py, batches)
     }
 
     /// Collect the DataFrame into a pandas DataFrame.

crates/spark-python-arrow/Cargo.toml

@@ -0,0 +1,33 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+[package]
+name = "spark-python-arrow"
+version = "0.1.0"
+edition = "2021"
+rust-version = "1.85"
+license = "Apache-2.0"
+publish = false
+description = "Arrow record batch export to PyArrow for Spark clients"
+repository = "https://github.com/apache/spark-connect-rust"
+
+[dependencies]
+arrow = { version = "58", default-features = false, features = ["ffi", "ipc", "pyarrow"] }
+# The consumer chooses the Python ABI floor and embedding/extension-module linking.
+pyo3 = { version = "0.28" }
+
+[lints.rust]
+unsafe_op_in_unsafe_fn = "deny"

crates/spark-python-arrow/README.md

@@ -0,0 +1,121 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Arrow table export to PyArrow
+
+`spark-python-arrow` exports decoded Rust Arrow record batches to PyArrow for
+`pyspark-rs`. It has one public entry point,
+`bridge::record_batches_to_pyarrow`, used by `DataFrame.toArrow()`. It does not
+depend on Spark Connect transport, generated protobufs, or an async runtime.
+
+Python-to-Rust Arrow imports, standalone array/type exports, and worker-side UDF
+conversion are outside this change. Scalar input conversion and row
+materialization remain in `pyspark-rs` until they have another concrete consumer.
+
+## Export contract
+
+The bridge uses `arrow::pyarrow::Table` to export standard Spark result batches.
+It preserves batch boundaries, schema and field metadata, and zero-column row
+counts, sharing compatible Arrow value and offset buffers with PyArrow. The
+caller supplies valid Rust record batches and owns its Python interpreter; the
+library does not initialize Python or retain session state.
+
+`toArrow()` collects all decoded batches into a Rust vector before the handoff.
+In Arrow 58.3.0, the table helper imports each batch through its own native
+C Stream reader and calls `read_next_batch()`, then builds a Python list and
+calls `pyarrow.Table.from_batches()`. It does not export the whole result through
+one reader or stream network results into Python. C Stream transfers buffer
+descriptors and ownership, not serialized IPC bytes.
+
+The contract covers Spark's ordinary map representation (`keys_sorted=false`).
+Sorted-map flags and dictionary-encoded arrays are outside this Spark result
+contract; the bridge does not add compatibility fixes for those generic Arrow
+inputs.
+
+PyArrow consumes the exported descriptors and retains their release callbacks,
+so the data survives after Rust's temporary wrappers have been dropped. If an
+importer raises before consuming a descriptor, Rust releases it. PyArrow and
+Python code in the same interpreter are trusted, including the public aliases
+used by the export path.
+
+Field names or timezone strings containing NUL use an IPC compatibility fallback
+over the already collected batches, without another RPC. Length-prefixed schema
+metadata can contain NUL without this fallback. The bridge preflights
+C-compatible schemas through PyArrow's standalone schema importer
+before the table helper imports any batches. This lets older PyArrow versions
+reject unsupported schemas before entering the C Stream path.
+
+`Table::try_new` validates batch schema consistency, including field metadata,
+before either export path. A mismatch raises `ValueError`. This is stricter than
+the previous IPC export path, which wrote batches under the first batch's schema.
+An empty batch vector still produces an empty table with an empty schema;
+preserving schemas from schema-only IPC streams requires a separate change to
+the decoding path.
+
+Zero-copy describes compatible Arrow **value and offset buffers**, not schema
+descriptors, Python scalar/row objects, pandas materialization, IPC decoding, or
+the IPC fallback. Sliced arrays whose value and null offsets differ can require
+copying the validity bitmap during C Data export.
+
+## Integration
+
+This is an unpublished workspace implementation crate (`publish = false`), not
+a separately released package. `pyspark-rs` uses a path dependency. The current
+package name and `0.1.0` version do not identify a crates.io release; independent
+publication would require Apache package naming and release-flow integration.
+Source-based consumers can pin a repository revision and validate it in their
+own dependency graph. The repository's root `LICENSE` covers this crate.
+
+The library does not enable PyO3's `extension-module` feature or select a minimum Python ABI;
+`pyspark-rs` selects these features for the Python extension.
+
+PyO3 must resolve to one compatible version in each consuming dependency graph.
+The manifest uses the same PyO3 0.28 line as the Python extension, with 0.28.3
+pinned in the workspace lockfile. This change does not upgrade PyO3.
+This crate declares Rust 1.85, matching Arrow 58.3.0's declared minimum.
+The existing workspace default of 1.82 is unchanged; it is not a claim that
+these Python/Arrow packages build on 1.82. The minimum compiler has not been
+independently tested locally.
+
+The Python Arrow export CI job tests this crate with the workspace lockfile.
+These tests run separately from the extension build and are not repeated in coverage.
+
+## Tests
+
+With PyArrow installed in the Python environment used by PyO3:
+
+```sh
+cargo test -p spark-python-arrow --locked --features pyo3/abi3-py39
+```
+
+For an offline build with cached dependencies, prefix the command with
+`CARGO_NET_OFFLINE=true CARGO_REGISTRIES_CRATES_IO_PROTOCOL=git`.
+
+Run this package separately from the extension's test/build command: Cargo
+feature unification with an extension-module consumer can disable libpython
+linking required by the embedded test executable. For a virtual environment,
+set `PYO3_PYTHON` to its interpreter and expose its site-packages to the embedded
+interpreter with `PYTHONPATH` when necessary.
+
+The standalone tests exercise the table export entry point: buffer identity and
+lifetime, schema metadata, slices, chunks, empty shapes, schema mismatch errors,
+and NUL fallback. The Python client's `test_arrow_handoff.py` exercises the public
+`toArrow()` and `toPandas()` paths, including malformed IPC responses. Scalar regressions
+live with the client tests. These are not substitutes for Spark's complete
+Connect parity suite.