feat: zero-copy Arrow C Data Interface at the Python/Rust boundary

[cpsievert]

Summary

• Replaces Arrow IPC serialization with zero-copy Arrow C Data Interface (PyCapsule) transfers for all data crossing the Python↔Rust boundary. All nine call sites funnel through py_to_df()/df_to_py() in src/lib.rs, so there are no signature changes — those two functions are now dispatchers.
• The legacy IPC path is retained behind GGSQL_ARROW_IPC=1 as an escape hatch (e.g., for exotic types that fail to round-trip through the C stream).
• Hand-rolled arrow::ffi_stream was chosen over the arrow-pyarrow crate: arrow-pyarrow 58 requires pyo3 0.28 (this crate is on 0.26), and arrow-pyarrow 57.x pins arrow-array 57 (incompatible with ggsql's arrow-58 DataFrame types). A pyo3 upgrade can happen later on its own merits.
• Minimum pyarrow is now 15 (the public RecordBatchReader.from_stream consumption API landed in 15.0; 14 only has the export side). The arrow crate dependency gains its ffi feature.
• Export direction (df_to_py): the RecordBatch is wrapped in an FFI_ArrowArrayStream, exported as a PyCapsule with a C Data Interface-correct destructor, and consumed via a small #[pyclass] shim exposing __arrow_c_stream__() → pa.RecordBatchReader.from_stream(...).read_all().
• Import direction (py_to_df, the hot path): table.__arrow_c_stream__() → ArrowArrayStreamReader::from_raw → concat_batches, with the same empty-table semantics as the IPC path.

Verification

• Full test suite passes in both boundary modes: uv run pytest tests/ with and without GGSQL_ARROW_IPC=1 (136 passed, 5 skipped in each), including 15 new edge-case/parity tests (dictionary-encoded columns, timestamps with timezones, empty results, chunked multi-batch tables, nullable columns, extension types, IPC-fallback smoke test).
• cargo fmt -- --check and cargo clippy -- -D warnings are clean.
• Leak smoke test: 200 register+read iterations show flat ru_maxrss (Δ ~1 MB).
• Boundary microbenchmark (1M rows × 50 cols int64, 5 iterations of register + SELECT *; script below, deliberately not committed): zero-copy stream 17.5s / 3246 MB peak RSS vs IPC 99.2s / 4247 MB — ~5.7× faster with ~1 GB lower peak RSS. Note DuckDB ArrowVTab ingestion (2048-row chunks) is a constant cost in both modes; the delta isolates the boundary savings.

Benchmark script (run with uv run python bench_arrow_boundary.py)

#!/usr/bin/env python3
"""Microbenchmark for the Python<->Rust Arrow boundary.

Registers a large pyarrow Table and reads it back through DuckDBReader,
exercising py_to_df (register) and df_to_py (execute_sql). Runs the inner
benchmark in two subprocesses -- default (zero-copy C stream) and
GGSQL_ARROW_IPC=1 (legacy IPC) -- and prints a comparison.

Note: DuckDB's ArrowVTab ingestion (chunked at 2048 rows) is a constant cost
in both modes; the delta between modes isolates the boundary savings.

Usage: uv run python scripts/bench_arrow_boundary.py [--rows N] [--cols N] [--iters N]
"""

import argparse
import os
import subprocess
import sys

INNER = r"""
import os
import resource
import sys
import time

import pyarrow as pa

import ggsql

rows, cols, iters = (int(x) for x in sys.argv[1:4])

# ru_maxrss is bytes on macOS, KiB on Linux
RSS_UNIT = 1 if sys.platform == "darwin" else 1024

cols_data = {f"c{i}": pa.array(range(rows), type=pa.int64()) for i in range(cols)}
table = pa.table(cols_data)

reader = ggsql.DuckDBReader("duckdb://memory")

# Warmup (also establishes the high-water mark baseline)
reader.register("t", table)
reader.execute_sql("SELECT * FROM t")

start = time.perf_counter()
for _ in range(iters):
    reader.register("t", table, replace=True)
    out = reader.execute_sql("SELECT * FROM t")
elapsed = time.perf_counter() - start

mode = "ipc" if os.environ.get("GGSQL_ARROW_IPC") else "stream"
peak_mb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss * RSS_UNIT / 1e6
print(f"RESULT mode={mode} elapsed={elapsed:.3f} peak_rss_mb={peak_mb:.0f} rows_out={out.num_rows}")
"""


def run_mode(mode: str, rows: int, cols: int, iters: int) -> str:
    env = dict(os.environ)
    if mode == "ipc":
        env["GGSQL_ARROW_IPC"] = "1"
    else:
        env.pop("GGSQL_ARROW_IPC", None)
    proc = subprocess.run(
        [sys.executable, "-c", INNER, str(rows), str(cols), str(iters)],
        env=env,
        capture_output=True,
        text=True,
        check=True,
    )
    for line in proc.stdout.splitlines():
        if line.startswith("RESULT"):
            return line
    raise RuntimeError(f"no RESULT line; stdout={proc.stdout} stderr={proc.stderr}")


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--rows", type=int, default=1_000_000)
    parser.add_argument("--cols", type=int, default=50)
    parser.add_argument("--iters", type=int, default=5)
    args = parser.parse_args()

    print(f"rows={args.rows} cols={args.cols} iters={args.iters}")
    for mode in ("stream", "ipc"):
        print(run_mode(mode, args.rows, args.cols, args.iters))


if __name__ == "__main__":
    main()

[Copilot]

🟢 Approval recommended

The FFI ownership handling is sound and the new behavior has comprehensive parity and edge-case coverage.

0 open findings

What changed in this PR

Replaces Arrow IPC serialization at the Python/Rust boundary with the Arrow C Data Interface while retaining an IPC fallback.

Changes:

• Adds zero-copy stream conversion and capsule lifecycle handling.
• Raises minimum PyArrow to 15 and enables Arrow FFI.
• Adds parity tests and a boundary benchmark.

File	Description
src/​lib.rs	Implements C stream transfers and IPC fallback.
tests/​test_ggsql.py	Tests both boundary modes and edge cases.
scripts/​bench_arrow_boundary.py	Benchmarks stream versus IPC transfers.
Cargo.toml	Enables Arrow’s FFI feature.
pyproject.toml	Raises the minimum PyArrow version.
uv.lock	Records updated PyArrow constraints.
CHANGELOG.md	Documents the new transfer path and fallback.

🧠 Review effort: Balanced

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.