feat(ffi): Rust bindings for sirius::ffi::Fragment and Context

rust/crates/sirius-sys/src/lib.rs

@@ -5,9 +5,10 @@
 //! wrappers live in the [`sirius`](https://docs.rs/sirius) crate.
 //!
 //! The bridge binds Sirius's **public C++ surface** (`src/include/sirius_ffi.hpp`):
-//! an RAII [`Context`] held via [`cxx::UniquePtr`]. Constructing it brings up an
-//! initialized engine; dropping the `UniquePtr` tears it down. The header is
-//! lightweight, so the bridge compiles without any of Sirius's internal headers
+//! an RAII [`Context`] held via [`cxx::UniquePtr`], plus the [`Fragment`] it
+//! creates — one plan fragment of a multi-fragment query. Constructing a context
+//! brings up an initialized engine; dropping the `UniquePtr` tears it down. The
+//! header is lightweight, so the bridge compiles without any of Sirius's internal headers
 //! (cudf/rmm/duckdb). It is the seed of the public API `libsirius` will expose;
 //! the bindings link whichever Sirius artifact provides these symbols (the DuckDB
 //! extension today, a dedicated `libsirius` later — see `build.rs`).
@@ -55,7 +56,117 @@ mod ffi {
             plan: &CxxString,
             out_stream_addr: usize,
         ) -> Result<()>;
+
+        /// One plan fragment of a multi-fragment query. Either declares output
+        /// streams (an intermediate fragment, whose results park as native GPU
+        /// batches that outlive its own query) or none (a result fragment, which
+        /// produces Arrow).
+        ///
+        /// Usage order: `declare_*` → `build` → `relay_from` every sender → `run`
+        /// → drain via `relay_from` or `result_to_arrow`. Exactly one fragment may
+        /// sit between its own `build` and `run`; the engine serializes queries.
+        type Fragment;
+
+        /// Create a [`Fragment`] on `context`, which must outlive it.
+        fn make_fragment(context: Pin<&mut Context>) -> Result<UniquePtr<Fragment>>;
+
+        /// Name of the view a plan reads to consume input stream `stream_id` — the
+        /// single definition of the convention, so a front end emitting the read
+        /// and the engine creating the view cannot drift apart.
+        fn stream_view_name(stream_id: u64) -> UniquePtr<CxxString>;
+
+        /// Declare one column of an input stream, in plan order. `ty` is a DuckDB
+        /// type name; a stream has no file to probe, so the schema is given, never
+        /// inferred.
+        fn declare_input_column(
+            self: Pin<&mut Fragment>,
+            stream_id: u64,
+            name: &CxxString,
+            ty: &CxxString,
+        ) -> Result<()>;
+
+        /// Declare a sender that must close this input stream before it ends.
+        fn declare_input_sender(
+            self: Pin<&mut Fragment>,
+            stream_id: u64,
+            sender_id: u32,
+        ) -> Result<()>;
+
+        /// Declare an output stream. A fragment with none is a result fragment.
+        fn declare_output(self: Pin<&mut Fragment>, stream_id: u64) -> Result<()>;
+
+        /// Replicate every batch to all declared outputs. Mutually exclusive with
+        /// [`declare_output_hash_key`](Fragment::declare_output_hash_key), and must
+        /// precede `build`; both are the errors raised here. Whether enough
+        /// destinations exist is not known until `build`.
+        fn declare_output_broadcast(self: Pin<&mut Fragment>) -> Result<()>;
+
+        /// Hash-partition across the declared outputs on `column_index` (call once
+        /// per key column, in key order). Same exclusivity and ordering rules as
+        /// broadcast.
+        fn declare_output_hash_key(self: Pin<&mut Fragment>, column_index: u32) -> Result<()>;
+
+        /// Plan `substrait_plan` against the declared streams and open the
+        /// fragment's query lifecycle. Where the declaration-time rules cannot be
+        /// checked, they are checked here:
+        ///
+        /// * a routing mode declared on a fragment with fewer than two outputs is
+        ///   rejected — every row would reach destination 0 regardless, so a
+        ///   silently-dropped spec would look like it worked;
+        /// * a plan that reads the same declared input stream id more than once is
+        ///   rejected — the second read would overwrite the first's registration
+        ///   and strand its pipeline waiting for a push that never comes.
+        fn build(self: Pin<&mut Fragment>, substrait_plan: &CxxString) -> Result<()>;
+
+        /// Move every batch parked on `source`'s output stream into this fragment's
+        /// input stream as native handles — no Arrow, no file, no copy — then close
+        /// `sender_id`. Returns the number of batches moved.
+        ///
+        /// `source` must have finished [`run`](Fragment::run): before that, an empty
+        /// stream and a finished one are indistinguishable, and the input would be
+        /// closed after zero batches. `input_stream_id` must be declared on this
+        /// fragment, and `source` must have output streams (a result fragment is
+        /// rejected).
+        fn relay_from(
+            self: Pin<&mut Fragment>,
+            source: Pin<&mut Fragment>,
+            source_stream_id: u64,
+            input_stream_id: u64,
+            sender_id: u32,
+        ) -> Result<usize>;
+
+        /// Close `sender_id` on input stream `stream_id`. The end-of-stream mirror
+        /// for senders that are not local fragments — `relay_from` closes its own.
+        /// Idempotent per sender; the stream ends once every expected sender has
+        /// closed.
+        fn close_input(self: Pin<&mut Fragment>, stream_id: u64, sender_id: u32) -> Result<()>;
+
+        /// Execute the fragment and close its query lifecycle. Blocks until its
+        /// pipelines finish.
+        fn run(self: Pin<&mut Fragment>) -> Result<()>;
+
+        /// Write a result fragment's rows into the caller-owned `ArrowArrayStream`
+        /// at `out_stream_addr`.
+        ///
+        /// # Safety
+        /// `out_stream_addr` must be the address of a valid, writable
+        /// `ArrowArrayStream` that outlives this call. The safe
+        /// [`sirius`](https://docs.rs/sirius) wrapper upholds this.
+        unsafe fn result_to_arrow(self: Pin<&mut Fragment>, out_stream_addr: usize) -> Result<()>;
+
+        /// Batches currently parked on an output stream — the evidence that a
+        /// fragment boundary carried native batches rather than nothing.
+        fn output_batch_count(self: &Fragment, stream_id: u64) -> Result<usize>;
+
+        /// DuckDB type names of a built fragment's output (sink) columns — the
+        /// types every batch leaving the fragment actually carries, exactly
+        /// what `relay_from`'s schema guard compares against a receiver's
+        /// declared input columns. Fallible: unbuilt and result fragments both
+        /// surface as `Err`.
+        fn output_types(self: &Fragment) -> Result<UniquePtr<CxxVector<CxxString>>>;
     }
 }
 
-pub use ffi::{Context, make_context, make_context_from_config};
+pub use ffi::{
+    Context, Fragment, make_context, make_context_from_config, make_fragment, stream_view_name,
+};