Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion .github/workflows/coverage.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
59 changes: 59 additions & 0 deletions .github/workflows/python-arrow.yml
Original file line number Diff line number Diff line change
@@ -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

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This installs latest stable and then runs clippy with -D warnings (L59), so the job will start failing whenever a new stable release adds lints, even if this crate hasn't changed. Could it pin a toolchain, or match the workspace lint job's setup?

rustup default stable
- name: Install PyArrow
run: python -m pip install "pyarrow==26.0.0"
Comment on lines +49 to +50

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pyproject.toml declares pyarrow>=10.0.0, and toArrow() now goes through PyArrow's private Schema._import_from_c / RecordBatch._import_from_c instead of the public IPC reader. CI pins only 26.0.0 here, and the other workflows install the latest PyArrow, so the PyArrow 10 run in the PR description isn't repeated in CI. Could this job add a leg at the declared floor? If that isn't practical, the README could state which PyArrow versions are tested.

- 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
10 changes: 10 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 3 additions & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ members = [
"crates/spark-connect-macros",
"crates/spark-connect-build",
"crates/pyspark-rs",
"crates/spark-python-arrow",
"examples",
]
exclude = [
Expand All @@ -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,
Expand Down
1 change: 1 addition & 0 deletions crates/pyspark-rs/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
18 changes: 7 additions & 11 deletions crates/pyspark-rs/src/dataframe.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
33 changes: 33 additions & 0 deletions crates/spark-python-arrow/Cargo.toml
Original file line number Diff line number Diff line change
@@ -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"

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Now that the crate is no longer built outside the workspace, edition, license and arrow could use .workspace = true like every other crate. On rust-version: apache-spark-connect already depends on arrow 58.3, which declares Rust 1.85, so the workspace's 1.82 is already wrong for the core crate too, not just this one (README L76). A separate change that raises the workspace value to 1.85 would let this crate inherit it and make that README paragraph unnecessary.

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"] }
# The consumer chooses the Python ABI floor and embedding/extension-module linking.
pyo3 = { version = "0.28" }

[lints.rust]
unsafe_op_in_unsafe_fn = "deny"
103 changes: 103 additions & 0 deletions crates/spark-python-arrow/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
<!--
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 C Data Interface path preserves batch boundaries, schema and field metadata,
and zero-column row counts. It shares 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.

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 outbound schema helpers also
retain sorted-map and dictionary-order flags when present.

Table export raises `ValueError` if batches have inconsistent schemas, including
field metadata or dictionary ordering. 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_NET_OFFLINE=true CARGO_REGISTRIES_CRATES_IO_PROTOCOL=git \
cargo test -p spark-python-arrow --locked --features pyo3/abi3-py39
Comment on lines +88 to +89

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This command differs from what CI runs (cargo test -p spark-python-arrow --locked --features pyo3/abi3-py39 in python-arrow.yml). CARGO_NET_OFFLINE=true makes it fail unless every dependency is already in the local Cargo cache, and CARGO_REGISTRIES_CRATES_IO_PROTOCOL=git switches away from Cargo's default sparse index. Could the README show the CI command, with the offline variables as an optional note for offline environments?

```

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 importer failures. Scalar regressions
live with the client tests. These are not substitutes for Spark's complete
Connect parity suite.
Loading
Loading