feat(c/sedona-extension): Add FFI table provider, exec plan, and synchronous stream exchange

[paleolimbot]

This PR adds an FFI wrapper around a TableProvider and enough infrastructure to replace an actual use case that we can test, which is importing a DataFrame from a separate SedonaDB context.

import pandas as pd
import sedona.db

sd_producer = sedona.db.connect()
sd_consumer = sedona.db.connect()

path = "submodules/geoarrow-data/ns-water/files/ns-water_water-point.parquet"

producer_sql = 'SELECT "OBJECTID", "FEAT_CODE" FROM water_point WHERE "OBJECTID" BETWEEN 10 AND 100 ORDER BY "OBJECTID"'
consumer_sql = 'SELECT "FEAT_CODE", COUNT(*) as cnt FROM df_producer GROUP BY "FEAT_CODE" HAVING COUNT(*) > 1 ORDER BY cnt DESC'

sd_producer.read_parquet(path).to_view("water_point")
df_producer = sd_producer.sql(producer_sql)

# Run the query on the producer (no FFI)
df_producer.to_view("df_producer")
result_no_ffi = sd_producer.sql(consumer_sql).to_pandas()

# Run the query on the consumer (separated via FFI)
sd_consumer.create_data_frame(df_producer).to_view("df_producer")
df_over_ffi = sd_consumer.sql(consumer_sql)
result_over_ffi =df_over_ffi.to_pandas()

# Same results!
pd.testing.assert_frame_equal(result_no_ffi, result_over_ffi)

# ...except our ImportedSedonaCExec wraps the inital plan. Projection pushed down!
explain_pd = df_over_ffi.explain().to_pandas()
print(explain_pd[explain_pd["plan_type"] == "physical_plan"]["plan"].iloc[0])
# SortPreservingMergeExec: [cnt@1 DESC]
#   SortExec: expr=[cnt@1 DESC], preserve_partitioning=[true]
#     ProjectionExec: expr=[FEAT_CODE@0 as FEAT_CODE, count(Int64(1))@1 as cnt]
#       FilterExec: count(Int64(1))@1 > 1
#         AggregateExec: mode=FinalPartitioned, gby=[FEAT_CODE@0 as FEAT_CODE], aggr=[count(Int64(1))]
#           RepartitionExec: partitioning=Hash([FEAT_CODE@0], 12), input_partitions=12
#             AggregateExec: mode=Partial, gby=[FEAT_CODE@0 as FEAT_CODE], aggr=[count(Int64(1))]
#               RepartitionExec: partitioning=RoundRobinBatch(12), input_partitions=1
#                 CooperativeExec
#                   ImportedSedonaCExec: ProjectionExec: expr=[FEAT_CODE@1 as FEAT_CODE]

[Copilot]

Pull request overview

This PR introduces a new Sedona C-extension FFI layer for exporting/importing TableProvider and ExecutionPlan objects across process/runtime boundaries (Python/R/ADBC), enabling cross-SedonaDB-context DataFrame round-tripping while preserving DataFusion optimization opportunities (e.g., projection pushdown).

Changes:

• Add a new sedona-extension crate implementing ABI-stable C interfaces for table providers, execution plans, expressions, and Arrow stream utilities.
• Replace the previous Rust/Python stream-reader utilities with a shared StreamingRecordBatchReader implementation (with optional cancellation / empty-batch skipping).
• Update Python/R/ADBC integrations to export/import providers via __sedonadb_table_provider__ and add Python FFI round-trip tests.

Reviewed changes

Copilot reviewed 23 out of 24 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
rust/sedona/src/reader.rs	Removes the old SedonaStreamReader implementation.
rust/sedona/src/lib.rs	Stops exporting the removed reader module.
rust/sedona/src/dataframe.rs	Adds a Sedona DataFrame extension trait + write options (currently duplicates context.rs).
rust/sedona-adbc/src/statement.rs	Switches ADBC streaming export to StreamingRecordBatchReader.
rust/sedona-adbc/Cargo.toml	Adds sedona-extension dependency.
r/sedonadb/src/rust/src/dataframe.rs	Switches R streaming export to StreamingRecordBatchReader.
r/sedonadb/src/rust/Cargo.toml	Adds sedona-extension dependency.
python/sedonadb/tests/test_dataframe.py	Updates comment to new __sedonadb_table_provider__ interface name.
python/sedonadb/tests/test_dataframe_ffi.py	Adds end-to-end Python tests validating FFI DataFrame round-trips.
python/sedonadb/src/runtime.rs	Removes now-unused Rust-side “wait for future” helper.
python/sedonadb/src/reader.rs	Replaces Python stream reader with new_py_streaming_reader using cancellation-aware StreamingRecordBatchReader.
python/sedonadb/src/import_from.rs	Switches provider import to new Sedona FFI table provider capsule type.
python/sedonadb/src/dataframe.rs	Exports __sedonadb_table_provider__ via sedona-extension’s exported provider implementation.
python/sedonadb/python/sedonadb/dataframe.py	Renames Python-side provider hook to __sedonadb_table_provider__ and updates docs/comments.
python/sedonadb/Cargo.toml	Adds sedona-extension dependency.
Cargo.lock	Records new crate dependencies for sedona-extension usage.
c/sedona-extension/src/utils.rs	Adds Arrow stream/property utilities + StreamingRecordBatchReader implementation and tests.
c/sedona-extension/src/table_provider.rs	Adds exported/imported TableProvider FFI wrappers and tests.
c/sedona-extension/src/execution_plan.rs	Adds exported/imported ExecutionPlan FFI wrappers and tests.
c/sedona-extension/src/expr.rs	Adds exported/imported expression FFI wrappers and tests.
c/sedona-extension/src/sedona_extension.h	Extends the C header with new ABI structs and callbacks.
c/sedona-extension/src/lib.rs	Exposes new execution_plan, expr, table_provider, and utils modules.
c/sedona-extension/src/extension.rs	Adds Rust FFI struct definitions for new ABI types (error/expr/plan/provider).
c/sedona-extension/Cargo.toml	Adds dependencies needed for new FFI/provider/plan implementations.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 22 out of 23 changed files in this pull request and generated 2 comments.

[paleolimbot]

This is the test that I wanted to make sure I could write before marking this ready for review: we export one of our own DataFrames as a "table", then write a query against it. We currently push projections through the boundary but not filters, but in either case the result should be correct.

[paleolimbot]

This is the C FFI over which a scan is handed across a boundary. This will be stable once we release the next version but we might need to change it to support everything we need to support before then. I designed this part to have as few callbacks as possible to ensure we can update things in a backward-compatible way without breaking the ABI.

[paleolimbot]

Currently this uses blocking ArrowArrayStream/RecordBatchReaders on both sides of the boundary; however, those are implemented in such a way that they can be handed a cancellation token on both ends (e.g., so in Python when you hit Control-C, both the producer and the consumer have a mechanism to know that a cancel was requested)..

The Async version is very verbose and we should do it someday, but not today (prototyped in #914 ).

[paleolimbot]

This is the FFI for the table itself.

[paleolimbot]

This is where we take an async SendableRecordBatchStream and export it as a RecordBatchReader that blocks on every call to next().

[paleolimbot]

This is the other direction: A SendableRecordBatchStream that polls a blocking RecordBatchReader from FFI. This would also benefit from the periodic cancel checker evaluation but we didn't specifically wire that up yet and so I didn't implement it (in the other direction this had already been implemented for the Python reader and there's a mechanism to implement it for the ADBC reader, which is why I included it here).

[paleolimbot]

This is the swap that was the target endpoint of this PR

[paleolimbot]

I replaced this reader with the reader in sedona-extension

[paleolimbot]

I also replaced this reader with the reader in sedona-extension

[paleolimbot]

This is a follow-up. This is how filters will get passed to get_property and scan...get_property can be queried for a spatial representation of the filter (e.g., bbox for a particular column), or a completely serialized filter in various forms (substrait, datafusion proto, maybe SQL but at the cost of some precision)

[james-willis]

I havent had time to review this myself yet, but here is a review from fable/ultracode. Do with it what you will knowing its just from the AI.

Correctness bugs

1. c/sedona-extension/src/streaming.rs:312 — The imported stream blocks a tokio worker thread in a synchronous recv() while the producer-side StreamWorker needs the same runtime via Handle::block_on, deadlocking when producer and consumer share a runtime. (Any plan that
   spawns tasks — e.g. contains RepartitionExec — exported and executed on the same runtime hangs forever once the runtime's workers are all blocked in recv(); deterministic on a current-thread runtime. The test comment at streaming.rs:352 about avoiding
   tokio::time::sleep because it "can deadlock with block_on" shows the hazard was already encountered.)
2. Runtime-liveness regression, three sites — python/sedonadb/src/dataframe.rs:453, rust/sedona-adbc/src/statement.rs:102, and r/sedonadb/src/rust/src/dataframe.rs:118 (R site PLAUSIBLE, other two CONFIRMED). The new StreamingRecordBatchReader holds only a
   tokio::runtime::Handle where the deleted readers held Arc, so the runtime can shut down while an exported stream is still being read. (caps = df.execute_stream().arrow_c_stream(); del df; del con drops the last Arc; the next capsule read calls
   block_on on a shut-down runtime and the consumer gets a spurious "Worker thread terminated" error. Same for ADBC releasing the statement before draining the reader, or R garbage-collecting the context mid-stream.)
3. python/sedonadb/src/dataframe.rs:676 — sedonadb_table_provider exports the view with a bare SessionContext::new(), and the producer side uses that empty session for both scan() planning and task_ctx() execution — the comment claiming "the consuming side will use
   its own session" is wrong; the consumer's session is discarded at table_provider.rs:345. (sd.read_parquet('s3://…') imported into another context fails at execute time with "No suitable object store found" because the bare session lacks the producer's object-store
   registrations, options, and Sedona planner extensions.)
4. c/sedona-extension/src/execution_plan.rs:238 — Every FFI error path writes *err = SedonaCError::new(...), which first runs Drop on the previous contents of *err, but the header never requires callers to zero-initialize SedonaCError. (A C consumer passing an
   uninitialized stack SedonaCError triggers a read of a garbage release function pointer and a call through it on the first error — jump to a random address. Applies to every *err assignment in execution_plan.rs and table_provider.rs.)
5. get_schema has no error channel — c/sedona-extension/src/execution_plan.rs:171–182 and c/sedona-extension/src/table_provider.rs:141 (three angles converged here). The callback is declared void, and on FFI_ArrowSchema::try_from failure both implementations silently
   return without writing *out — unlike every sibling callback, which takes a SedonaCError and returns int. (A C consumer following the Arrow convention of passing an uninitialized ArrowSchema then calls a garbage release() — UB; the Rust importer parses an empty schema
   and surfaces a misleading "null format" error with the real cause swallowed. This is an ABI-shape issue worth fixing before the header is published.)
6. c/sedona-extension/src/sedona_extension.h:228 — SedonaCExpr.get_property_schema is missing the struct ArrowSchema* out parameter that the equivalent callbacks on SedonaCExecutionPlan (line 283) and SedonaCTableProvider (line 341) have, so the callback cannot return a
   schema at all. (The doc on SedonaCExpr.get_property says the type "may be retrieved with the get_property_schema callback," but the signature has no output slot. Fixing the function-pointer type after the header ships breaks every compiled consumer, so it should be
   corrected now.)
7. python/sedonadb/src/import_from.rs:49 — Standard DataFusion FFI interop is dropped: import now only recognizes sedonadb_table_provider; nothing handles datafusion_table_provider anymore. (Objects from datafusion-python or delta-rs previously imported
   zero-copy via FFI_TableProvider; now they silently fall through to single-scan arrow_c_stream materialization, or fail with "Can't create table provider" if they have no stream dunder.)
8. c/sedona-extension/src/streaming.rs:149 — Cancellation no longer stops query execution: the deleted wait_for_future_from_rust dropped the polled future on Ctrl+C, but the new worker thread is stuck uninterruptibly in runtime.block_on(stream.next()). (Ctrl+C during a
   slow batch returns control to Python, but the worker keeps burning CPU and the stream's memory/file handles can't be dropped until the batch completes — for a never-yielding stream, they leak for the life of the process.)
9. c/sedona-extension/src/extension.rs:188 — sedona_c_noop_release doesn't set self->release = NULL, violating the header's own contract ("Implementations of this callback must set self->release to NULL") for the UNKNOWN_SEDONA_C_ERROR fallback. (A consumer that uses
   release == NULL to detect a released error sees a still-"valid" error and may double-release or keep reading it; while (err.release) err.release(&err); spins forever.)
10. c/sedona-extension/src/streaming.rs:261 — Iterator::next detects cancellation by downcasting the error fetch_next_batch itself constructed (ExternalError → io::Error → ErrorKind::Interrupted). (If the inner stream yields its own interrupted-io::Error — e.g. an
    interrupted object-store read — the reader misclassifies it as user cancellation, sets cancelled = true, and permanently returns None, silently truncating remaining batches. Simpler: have fetch_next_batch return an explicit Cancelled variant or set the flag itself.)

Efficiency

11. c/sedona-extension/src/table_provider.rs:86 — ExportedTableProvider::scan spawns a brand-new OS thread per scan call and immediately join()s it, parking the calling thread (typically a consumer tokio executor thread) for the entire producer-side planning. (For
    object-store-backed tables, planning does network I/O that can take seconds; several concurrent queries against imported tables stall the consumer runtime. Cheaper: one long-lived planning thread, or a oneshot channel awaited asynchronously.)
12. c/sedona-extension/src/streaming.rs:139 — The zero-capacity rendezvous protocol (worker only fetches after a request arrives) fully serializes producer fetch and consumer processing, with two context switches per batch and one dedicated OS thread per partition
    stream. (Per-batch wall time is T_produce + T_consume instead of max of the two — roughly doubling a 1000-batch export through Python when the costs are comparable; a 32-partition imported plan also spawns 32 extra threads per execution. Cheaper: eager fetch into a
    bounded channel of capacity 1–2 for one-batch pipelining.)
13. c/sedona-extension/src/utils.rs:158 — Every property fetch makes a second FFI roundtrip to get_property_schema even though the exporter always answers Utf8. (EXPLAIN ANALYZE or repeated metrics() polling doubles FFI traffic and does a CString/Field/FFI_ArrowSchema
    allocate-and-parse cycle per read. Cheaper: fetch the property schema once and cache it.)

Altitude

14. r/sedonadb/src/rust/src/dataframe.rs:134 — The R bindings still export/import via the version-locked datafusion-ffi FFI_TableProvider, so the new version-agnostic C mechanism is wired into Python only and two parallel table-provider FFI mechanisms must now be
    maintained. (Every future exchange change — pushdown, cancellation, error surfaces — must be made twice, and an R package and Python package built against different DataFusion versions still can't exchange providers, defeating the PR's ABI-stability goal for one of the
    bindings.)

Reuse / simplification

15. c/sedona-extension/src/utils.rs:234 and :311 — Four near-verbatim copies of the property-fetch protocol (get_plan_property, get_plan_string_property, get_table_provider_string_property, get_table_provider_property_data_type): identical CString setup, unsafe call,
    errno handling, and schema lookup, differing only in struct type and the final decode step. (~180 lines of duplicated unsafe FFI code; any fix to errno mapping or error freeing must land in four places, and a missed copy gives divergent behavior between plan and
    provider property reads — partially masked by the silent default-to-TableType::Base fallback.)
16. c/sedona-extension/src/streaming.rs:243 — The cancellation error (io::Error with ErrorKind::Interrupted) is constructed at two sites with no shared constructor, and next() re-detects it by downcast — an invisible coupling. (Editing one copy to a different error type
    without updating the downcast means the cancelled flag is never set, and the FFI stream consumer re-blocks on the still-running query instead of terminating — a hang that only manifests on the Ctrl+C path. Closely related to finding 10.)
17. c/sedona-extension/src/table_provider.rs:475 — The six test_roundtrip_* tests each repeat the same 10-line export/import/register preamble with only the SQL string differing, while a suitable helper (setup_imported_provider_with, line 731) already exists in the same
    module. (Any signature change to ExportedTableProvider::new or ImportedTableProvider::try_new requires editing six copies; a parameterized (sql, expected) helper cuts ~60 lines.)
18. c/sedona-extension/src/execution_plan.rs:318 — Stale copy-pasted name ImportedTableProviderExec survives in the Debug impl (lines 318, 322), the try_new doc comment (line 328), and the with_new_children error message (line 483) of a type actually named
    ImportedSedonaCExec. (Someone debugging greps for a type that doesn't exist. Use self.name() or fix all four.)
19. c/sedona-extension/Cargo.toml:44 — sedona-geometry is added as a dependency but nothing in the crate references it. (Every build of the extension crate — and Python/R/ADBC downstream — compiles and links it for nothing. Delete the line.)

[Copilot]

Pull request overview

Copilot reviewed 29 out of 30 changed files in this pull request and generated 1 comment.

[james-willis]

these are advanced tactics heretofore unknown to me