Stream filesystem conversions into output

docs/api.md

@@ -127,6 +127,15 @@ empty. All batches must match the encoding schema.
 
 Raises `TypeError` for non-Arrow inputs and `ValueError` for missing or mismatched schemas.
 
+### `encode_batches_to(batches, output, *, schema=None)`
+
+Consumes an iterable of Arrow record batches and writes encoded data to a binary file-like
+`output`. Returns `None` and leaves `output` open. `schema` is required when `batches` is
+empty. All batches must match the encoding schema.
+
+Raises `TypeError` for non-Arrow batches or a non-schema `schema`, `ValueError` for missing
+or mismatched schemas, and `OSError` when the output writer fails.
+
 ### `iter_batches(data, *, schema=None, batch_size=1024, row_limit=None, byte_limit=None)`
 
 Returns `DecodedBatchStream`.
@@ -177,6 +186,9 @@ writer accepts batches incrementally but may buffer its output until `finish`.
 
 `Codec::encode_stream` uses the same writer session for all four codecs.
 
+Python `Codec.encode_batches_to` uses an owned writer session backed by the supplied binary
+file object.
+
 ### `Codec::start_writer(schema, output)`
 
 Borrows a `Write + Send` sink and returns a writer session with the same lifetime. This form

docs/explanation.md

@@ -58,8 +58,9 @@ chunks can reside at different offsets. Arrow IPC, CSV, and JSONL consume reader
 sequentially, but use the same seekable reader contract so registry consumers have one input
 boundary.
 
-`DataFileSystem` opens its inner source through fsspec and decodes from that file object. Its
-converted output remains a buffering adapter: schema application collects decoded batches,
-encoding returns complete output bytes, and `_open()` copies those bytes into a seekable
-spooled file. Reader-backed input removes the initial encoded allocation without making the
-complete chained conversion end-to-end streaming.
+`DataFileSystem` opens its inner source through fsspec, applies its schema plan batch by
+batch, and writes encoded output directly into a seekable spooled file. This bounds
+intermediate transport memory without changing the filesystem contract: `_open()` still
+completes conversion before returning the spool, and `info()` may perform a complete
+conversion to determine output size. Individual format writers may also buffer internally;
+in particular, Parquet can defer encoded output until `finish`.

docs/how-to-integrate.md

@@ -9,7 +9,7 @@ Map backend-native values to a `pyarrow.Schema`. Describe both sides of the conv
 before reading data:
 
 ```python
-from fsspec_data import DataFormat, InterchangeRequest, SchemaPolicy
+from fsspec_data import DEFAULT_REGISTRY, DataFormat, InterchangeRequest, SchemaPolicy
 
 request = InterchangeRequest(
     provided_format=DataFormat.PARQUET,
@@ -51,6 +51,18 @@ encoded_arrow = plan.convert(encoded, batch_size=1_024)
 `convert` buffers its encoded result. Use `iter_batches` for incremental scans, previews,
 and database reads.
 
+Write an encoded stream directly to a binary file when the consumer does not need one byte
+buffer:
+
+```python
+with open("result.arrow", "wb") as output:
+    DEFAULT_REGISTRY.get(DataFormat.ARROW).encode_batches_to(
+        plan.iter_batches(encoded),
+        output,
+        schema=plan.requested_schema,
+    )
+```
+
 ## Integrate from Rust
 
 Compose registered codecs with the plan's stream adapter:

fsspec_data/filesystem.py

@@ -70,11 +70,14 @@ def _open(
             raise ValueError("fsspec-data is a read-only filesystem")
 
         path = self._strip_protocol(path)
-        output = self._convert(path)
         file = tempfile.SpooledTemporaryFile(max_size=self.spool_max_size, mode="w+b")
-        file.write(output)
-        file.seek(0)
-        self._sizes[path] = len(output)
+        try:
+            self._convert(path, file)
+            self._sizes[path] = file.tell()
+            file.seek(0)
+        except Exception:
+            file.close()
+            raise
         return file
 
     def info(self, path: str, **kwargs: Any) -> dict[str, Any]:
@@ -94,31 +97,31 @@ def ls(self, path: str, detail: bool = True, **kwargs: Any):
         entry = self.info(path, **kwargs)
         return [entry] if detail else [entry["name"]]
 
-    def _convert(self, path: str) -> bytes:
+    def _convert(self, path: str, output) -> None:
         provided_format = self.provided_format or _format_from_path(self.fo)
         requested_format = self.requested_format or _format_from_path(path)
         with self.fs.open(self.fo, "rb") as source:
-            decoded = DEFAULT_REGISTRY.get(provided_format).decode_batches(
+            decoded = DEFAULT_REGISTRY.get(provided_format).iter_batches(
                 source,
                 schema=self.provided_schema,
                 batch_size=self.batch_size,
                 row_limit=self.row_limit,
                 byte_limit=self.byte_limit,
             )
-        if self.provided_schema is not None and not decoded.schema.equals(self.provided_schema, check_metadata=False):
-            raise ValueError("provided schema does not match the source schema")
-
-        provided_schema = self.provided_schema or decoded.schema
-        requested_schema = self.requested_schema or provided_schema
-        plan = InterchangeRequest(
-            provided_format,
-            requested_format,
-            provided_schema,
-            requested_schema,
-            self.schema_policy,
-        ).plan()
-        batches = tuple(plan.apply_batch(batch) for batch in decoded.batches)
-        return DEFAULT_REGISTRY.get(requested_format).encode_batches(batches, schema=requested_schema)
+            if self.provided_schema is not None and not decoded.schema.equals(self.provided_schema, check_metadata=False):
+                raise ValueError("provided schema does not match the source schema")
+
+            provided_schema = self.provided_schema or decoded.schema
+            requested_schema = self.requested_schema or provided_schema
+            plan = InterchangeRequest(
+                provided_format,
+                requested_format,
+                provided_schema,
+                requested_schema,
+                self.schema_policy,
+            ).plan()
+            batches = (plan.apply_batch(batch) for batch in decoded)
+            DEFAULT_REGISTRY.get(requested_format).encode_batches_to(batches, output, schema=requested_schema)
 
 
 _SUFFIX_FORMATS = {

fsspec_data/interchange.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+from collections.abc import Iterable
 from dataclasses import dataclass, replace
 from enum import Enum
 from importlib import import_module
+from itertools import chain
 from typing import Any, BinaryIO
 
 import pyarrow as pa
@@ -94,6 +96,31 @@ def encode_batches(
                 raise TypeError("batches must contain pyarrow.RecordBatch objects")
         return _rust.encode_batches(self.format.value, schema, batches)
 
+    def encode_batches_to(
+        self,
+        batches: Iterable[pa.RecordBatch],
+        output: BinaryIO,
+        *,
+        schema: pa.Schema | None = None,
+    ) -> None:
+        batches = iter(batches)
+        if schema is None:
+            try:
+                first = next(batches)
+            except StopIteration as error:
+                raise ValueError("schema is required when encoding no record batches") from error
+            if not isinstance(first, pa.RecordBatch):
+                raise TypeError("batches must contain pyarrow.RecordBatch objects")
+            schema = first.schema
+            batches = chain((first,), batches)
+        schema = _ensure_pyarrow(schema)
+        writer = _rust.start_codec_writer(self.format.value, schema, output)
+        for batch in batches:
+            if not isinstance(batch, pa.RecordBatch):
+                raise TypeError("batches must contain pyarrow.RecordBatch objects")
+            writer.write_batch(batch)
+        writer.finish()
+
     def decode_batches(
         self,
         data: bytes | bytearray | memoryview | BinaryIO,

fsspec_data/tests/test_all.py

@@ -1,4 +1,5 @@
 import io
+from typing import Any
 
 import pyarrow as pa
 import pyarrow.parquet as pq
@@ -282,6 +283,43 @@ def test_text_codec_round_trip_preserves_schema_and_rows(format, prefix):
     assert pa.Table.from_batches(decoded.batches).to_pylist() == pa.Table.from_batches(batches).to_pylist()
 
 
+@pytest.mark.parametrize("format", list(DataFormat))
+def test_codec_encodes_batches_to_binary_file(format):
+    provided, _ = schemas()
+    batches = (
+        pa.RecordBatch.from_arrays([pa.array([1, 2]), pa.array(["ada", None])], schema=provided),
+        pa.RecordBatch.from_arrays([pa.array([3]), pa.array(["margaret"])], schema=provided),
+    )
+    codec = DEFAULT_REGISTRY.get(format)
+    output = io.BytesIO()
+
+    codec.encode_batches_to(iter(batches), output)
+
+    decode_schema = provided if format in {DataFormat.CSV, DataFormat.JSONL} else None
+    decoded = codec.decode_batches(output.getvalue(), schema=decode_schema)
+    assert pa.Table.from_batches(decoded.batches).to_pylist() == pa.Table.from_batches(batches).to_pylist()
+
+
+def test_codec_file_writer_requires_schema_for_empty_stream():
+    codec = DEFAULT_REGISTRY.get(DataFormat.ARROW)
+
+    with pytest.raises(ValueError, match="schema is required"):
+        codec.encode_batches_to((), io.BytesIO())
+
+
+def test_codec_file_writer_rejects_non_batches():
+    provided, _ = schemas()
+    codec = DEFAULT_REGISTRY.get(DataFormat.ARROW)
+    invalid: Any = object()
+
+    with pytest.raises(TypeError, match="pyarrow.RecordBatch"):
+        codec.encode_batches_to(iter([invalid]), io.BytesIO())
+
+    batch = pa.record_batch([[1], ["ada"]], schema=provided)
+    with pytest.raises(TypeError, match="pyarrow.RecordBatch"):
+        codec.encode_batches_to(iter([batch, invalid]), io.BytesIO(), schema=provided)
+
+
 @pytest.mark.parametrize("format", [DataFormat.CSV, DataFormat.JSONL])
 def test_text_codec_decode_requires_schema(format):
     codec = DEFAULT_REGISTRY.get(format)

fsspec_data/tests/test_filesystem.py

@@ -5,7 +5,7 @@
 import pyarrow.parquet as pq
 import pytest
 
-from fsspec_data import DataFileSystem
+from fsspec_data import Codec, DataFileSystem, DecodedBatchStream
 
 SCHEMA = pa.schema([pa.field("id", pa.int64(), nullable=False), pa.field("name", pa.string())])
 SCHEMA_OPTIONS = {
@@ -62,6 +62,17 @@ def read(self, size=-1):
     assert bytes_read < len(encoded)
 
 
+def test_chained_filesystem_streams_batches_into_spooled_output(memory_fs, monkeypatch):
+    monkeypatch.setattr(DecodedBatchStream, "collect", lambda self: pytest.fail("decoded batches should not be collected"))
+    monkeypatch.setattr(Codec, "encode_batches", lambda self, *args, **kwargs: pytest.fail("encoded bytes should not be collected"))
+    fs = DataFileSystem(fo="orders.csv", fs=memory_fs, provided_schema=SCHEMA_OPTIONS)
+
+    assert pq.read_table(fs.open("orders.parquet")).to_pylist() == [
+        {"id": 1, "name": "ada"},
+        {"id": 2, "name": "grace"},
+    ]
+
+
 def test_fsspec_url_builds_chained_filesystem(memory_fs):
     fsspec.register_implementation("fsspec-data", DataFileSystem, clobber=True)
 

rust/python/lib.rs

@@ -1,8 +1,8 @@
-use std::io::{self, Read, Seek, SeekFrom};
+use std::io::{self, Read, Seek, SeekFrom, Write};
 
 use ::fsspec_data::{
-    plan_schema as build_plan, CancellationToken, CodecReader, DataFormat, DecodedStream,
-    InterchangeError, SchemaPolicy, StreamOptions, DEFAULT_REGISTRY,
+    plan_schema as build_plan, CancellationToken, CodecReader, CodecWriter, DataFormat,
+    DecodedStream, InterchangeError, SchemaPolicy, StreamOptions, DEFAULT_REGISTRY,
 };
 use arrow::array::RecordBatch;
 use arrow::datatypes::Schema;
@@ -59,11 +59,60 @@ fn python_io_error(error: PyErr) -> io::Error {
     io::Error::other(error.to_string())
 }
 
+struct PythonWriter {
+    sink: Py<PyAny>,
+}
+
+impl Write for PythonWriter {
+    fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
+        Python::with_gil(|py| {
+            self.sink
+                .bind(py)
+                .call_method1("write", (PyBytes::new(py, buffer),))
+                .and_then(|written| written.extract::<usize>())
+                .map_err(python_io_error)
+        })
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Python::with_gil(|py| {
+            self.sink
+                .bind(py)
+                .call_method0("flush")
+                .map(|_| ())
+                .map_err(python_io_error)
+        })
+    }
+}
+
 #[pyclass(unsendable)]
 struct NativeBatchStream {
     stream: DecodedStream,
 }
 
+#[pyclass(unsendable)]
+struct NativeCodecWriter {
+    writer: Option<Box<dyn CodecWriter>>,
+}
+
+#[pymethods]
+impl NativeCodecWriter {
+    fn write_batch(&mut self, batch: PyArrowType<RecordBatch>) -> PyResult<()> {
+        self.writer
+            .as_mut()
+            .ok_or_else(|| pyo3::exceptions::PyValueError::new_err("codec writer is finished"))?
+            .write_batch(&batch.0)
+            .map_err(to_python_error)
+    }
+
+    fn finish(&mut self) -> PyResult<()> {
+        if let Some(writer) = self.writer.take() {
+            writer.finish().map_err(to_python_error)?;
+        }
+        Ok(())
+    }
+}
+
 #[pymethods]
 impl NativeBatchStream {
     fn __iter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> {
@@ -113,6 +162,31 @@ fn encode_batches<'py>(
     Ok(PyBytes::new(py, &encoded))
 }
 
+#[pyfunction]
+fn start_codec_writer(
+    py: Python<'_>,
+    format: &str,
+    schema: PyArrowType<Schema>,
+    sink: Py<PyAny>,
+) -> PyResult<Py<NativeCodecWriter>> {
+    let format = DataFormat::parse(format).map_err(to_python_error)?;
+    let writer = DEFAULT_REGISTRY
+        .get(format)
+        .and_then(|codec| {
+            codec.start_owned_writer(
+                std::sync::Arc::new(schema.0),
+                Box::new(PythonWriter { sink }),
+            )
+        })
+        .map_err(to_python_error)?;
+    Py::new(
+        py,
+        NativeCodecWriter {
+            writer: Some(writer),
+        },
+    )
+}
+
 #[pyfunction]
 #[pyo3(signature = (format, data, schema=None))]
 fn decode_batches(
@@ -248,5 +322,6 @@ fn fsspec_data(m: &Bound<PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(decode_stream, m)?)?;
     m.add_function(wrap_pyfunction!(encode_batches, m)?)?;
     m.add_function(wrap_pyfunction!(plan_schema, m)?)?;
+    m.add_function(wrap_pyfunction!(start_codec_writer, m)?)?;
     Ok(())
 }