support FFI query result streams that do not pre-collect

Is your feature request related to a problem or challenge? Please describe what you are trying to do.
I'm trying to pass the result of a query into my rust code. Some of the queries I'm doing produce a lot of data, and I would like to process this in a streaming way, without first loading the entire query result into memory (where it might not even fit).

Dataframe has a function __arrow_c_stream__(), which can be used to cross the FFI boundary and get dataframe results into a native component. Unfortunately, this calls .collect() internally. This means I can't actually stream over the results while keeping the memory footprint low. I need to be able to load my entire dataset in memory, and the rest of my processing logic has to wait for this to complete before it can start.

Describe the solution you'd like
I would like __arrow_c_stream__() or a similar function to produce a RecordBatchReader or even a RecordBatchStream (which also appears to be FFI-wrapped), which streams the query result without first collecting into memory.

Describe alternatives you've considered
The alternative is accepting that using results from python in rust will always require a collect on the python side first. Given that the infrastructure seems to be in place to pass around readers and streams, this seems silly.

The Arrow C Data Interface and the PyCapsule Interface (of which __arrow_c_stream__ is a part), are synchronous-only. You could in theory create a RecordBatchIterator in Rust that under the hood blocks using tokio in each next() call, but that will of course be blocking.

Alternatively, you can use RecordBatchStream. That holds a DataFusion RecordBatchStream, and so it shouldn't pre-collect. However then it's impossible to transfer via FFI to another library because Arrow doesn't define an async FFI protocol. I added async iteration support to RecordBatchStream, which may help you.

All this means is that you have to do the iteration via Python. So you can do sync or async iteration over the RecordBatchStream, and then for each produced RecordBatch you can use FFI to convert that to whatever library you're using.

For what it's worth, you can do this to some degree with duckdb:

I realize that FFI source is a specific concept for DF/python, but in the broader sense this is a legitimate request, not having to materialize the full source data when performing an aggregation for example.

FWIW I implemented this in kylebarron/arro3#313, where the RecordBatchStream can be synchronously exported to an ArrowArrayStream

having the same problem, datafusion keep consuming more memory till it crashes

df = ctx.sql("""
              select
             * EXCLUDE("I","XX","SETTLEMENTDATE"),
              CAST(to_timestamp_seconds("SETTLEMENTDATE",'%Y/%m/%d %H:%M:%S') AS TIMESTAMP(6)) as SETTLEMENTDATE,
              cast(EXTRACT(YEAR FROM CAST(to_timestamp_seconds("SETTLEMENTDATE",'%Y/%m/%d %H:%M:%S') AS TIMESTAMP(6))) as integer) as year
              from arrow_dataset
              where "I" ='D' and "UNIT"='DUNIT' and "VERSION" ='3'
              """)
 write_deltalake(f"{table_root}/T{total_files}/datafusion", df,mode="overwrite",partition_by=['year'])

#1222 would close this.

This can be closed?

@Dandandan

Yes! Thanks @Dandandan for point it out.

Navigation Menu

Repository navigation

support FFI query result streams that do not pre-collect #1011

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

support FFI query result streams that do not pre-collect #1011

Description

Activity

kylebarron commented on Feb 5, 2025

aditanase commented on Apr 25, 2025

kylebarron commented on Apr 25, 2025

djouallah commented on Jul 15, 2025

kylebarron commented on Sep 3, 2025

Dandandan commented on Jan 14, 2026

timsaucer commented on Jan 14, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions