Skip to content

feat(exec): streaming fragment — plan builder, blocking runner, and Fragment FFI (#839) - #1481

Open
aocsa wants to merge 5 commits into
sirius-db:devfrom
aocsa:stream/15-fragment
Open

feat(exec): streaming fragment — plan builder, blocking runner, and Fragment FFI (#839)#1481
aocsa wants to merge 5 commits into
sirius-db:devfrom
aocsa:stream/15-fragment

Conversation

@aocsa

@aocsa aocsa commented Aug 12, 2026

Copy link
Copy Markdown
Contributor

Part of #839 — the execution half. Last of the three PRs that replaced #1432; #1479 and #1480
have merged, so this now targets dev directly and stands alone.

Not closing #839. The issue asks for a session that "starts it on the existing task
scheduler without blocking". streaming_fragment::run() blocks: it goes through
sirius_engine::execute(), which takes the future from start_query() and immediately waits.
Fragments therefore run store-and-forward, one at a time, which is why the FFI orders
relay_from(...) → run() and notes that only one fragment may sit between its own build()
and run(). The Fragment surface also exposes no push/pull/wait, so a genuinely remote
sender cannot feed a fragment yet. Both are tracked in #1590 rather than claimed here.

Base: #1480 (stream/14-session). Review that first; this PR's diff is the third commit only.

What. Turns a plan plus a set of stream ids into something runnable. streaming_fragment
binds a logical plan against a stream_bind_catalog, rewrites its stream leaves to the
batch_stream handles the session resolves, builds the physical plan, and runs it to completion.

Design: bind before the streams exist. A fragment has to be constructible before its peers
have started — otherwise startup is ordered and a distributed plan deadlocks on its own topology.
stream_bind_catalog and stream_plan_bindings are what make that possible: a plan names its
inputs and outputs by id, and binding resolves those ids through the session at build time rather
than at plan time. The rewrite step is where the id-addressed world of part 2 meets the
pointer-addressed world of the operators.

FFI. sirius::ffi::Fragment exposes one fragment to Rust: relay_from to chain fragments,
result_to_arrow to pull the tail out as Arrow, and the build/run lifecycle. This is the surface
the multi-node work sits on.


Reviewing. Two commits: the original, plus a second holding the review fixes so you can see
what moved since you last looked rather than having it amended away. Suggested reading order:

  1. streaming_fragment.hpp — the lifecycle contract
  2. stream_bind_catalog.hpp / stream_plan_bindings.hpp — how a plan names streams
  3. streaming_fragment.cpp — bind → rewrite → build → run
  4. sirius_ffi.cpp — the Rust-facing surface

Tests. test_streaming_fragment.cpp covers build/run, relay between two fragments, the error
plane, and key-cast normalization. test_stream_bind_catalog.cpp covers the binding layer alone.


Stack.

Part PR Branch Scope State
1 #1479 stream/13-sink STREAMING_SINK + partitioning (#837, #838) merged
2 #1480 stream/14-session exec::stream_session (#839, addressing) merged
3 #1481 (this) stream/15-fragment streaming_fragment + FFI (#839, execution) ← on dev

This tip is tree-identical to #1432's head on every streaming path, so #1433
(sirius::ffi::Fragment for Rust) rebases onto it without change. #1432 is closed as superseded.

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Adds the missing “runnable fragment” layer for streaming execution: a fragment can bind/plan against id-addressed streams via a per-connection bind catalog, rewrite those leaves into STREAMING_SOURCE operators, root the plan in a STREAMING_SINK, and run to completion. It also exposes this lifecycle to Rust via a new sirius::ffi::Fragment API and adds integration/unit tests plus documentation.

Changes:

  • Introduces exec::streaming_fragment, exec::stream_bind_catalog, and sirius_stream_source(id) plan binding glue to bridge bind-time schema resolution with plan-time operator construction.
  • Adds sirius::ffi::Fragment (declare → build → relay/close → run → result) on the embedded DuckDB path.
  • Adds tests for fragment build/run/relay and for bind-catalog semantics; updates docs and build wiring.

Reviewed changes

Copilot reviewed 18 out of 18 changed files in this pull request and generated 4 comments.

Show a summary per file
File Description
test/cpp/utils/pipeline_conversion_test_utils.hpp Adds sql_plan_source helper declaration for tests to produce bound logical plans.
test/cpp/utils/pipeline_conversion_test_utils.cpp Implements sql_plan_source helper used by streaming fragment tests.
test/cpp/exec/test_streaming_fragment.cpp New integration tests for fragment build/run, chaining via streams, and multi-batch draining.
test/cpp/exec/test_stream_bind_catalog.cpp New unit tests for bind catalog behavior and sirius_stream_source binding rules.
src/include/exec/stream_bind_catalog.hpp Defines per-connection stream schema/repo declarations + built-operator backpointer.
src/exec/stream_bind_catalog.cpp Implements validation, lookup, and built-pointer recording for the bind catalog.
src/include/exec/stream_plan_bindings.hpp Declares sirius_stream_source table function + bind data used for planning.
src/exec/stream_plan_bindings.cpp Implements sirius_stream_source binding and idempotent registration.
src/include/exec/streaming_fragment.hpp Defines fragment_spec and streaming_fragment lifecycle/contract.
src/exec/streaming_fragment.cpp Implements bind → plan → sink root → session wiring → blocking run.
src/include/planner/sirius_physical_plan_generator.hpp Adds create_streaming_source_plan hook for stream reads.
src/planner/sirius_plan_get.cpp Routes sirius_stream_source scans to STREAMING_SOURCE built from the bind catalog.
src/sirius_extension.cpp Registers sirius_stream_source function on extension load (transparent path).
src/include/sirius_ffi.hpp Adds Fragment public API for Rust/C++ callers, plus stream_view_name.
src/sirius_ffi.cpp Implements Fragment planning/execution/relay logic and shared Substrait lowering helper.
docs/super-sirius/streaming-sessions.md Documents exec::streaming_fragment and Fragment FFI contracts and invariants.
docs/super-sirius/README.md Adds streaming sessions doc link in Super Sirius index.
CMakeLists.txt Wires new sources and new tests into the build.

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

Comment thread src/exec/streaming_fragment.cpp
Comment thread src/exec/stream_plan_bindings.cpp
Comment thread src/sirius_ffi.cpp
Comment thread src/sirius_ffi.cpp
@aocsa
aocsa deleted the branch sirius-db:dev August 12, 2026 14:01
@aocsa aocsa closed this Aug 12, 2026
@aocsa aocsa reopened this Aug 13, 2026
@aocsa
aocsa requested a review from 9prady9 August 13, 2026 13:18
@dhruv9vats
dhruv9vats self-requested a review August 14, 2026 12:51
@aocsa
aocsa force-pushed the stream/15-fragment branch from d1c0dc6 to cc90c81 Compare August 17, 2026 03:24
@aocsa
aocsa force-pushed the stream/14-session branch from 641b77e to a649cf5 Compare August 17, 2026 03:24
@aocsa

aocsa commented Aug 17, 2026

Copy link
Copy Markdown
Contributor Author

Pushed d61c9c36 — kept as its own commit on top of cc90c816 so the delta since review is visible rather than amended away.

All four Copilot findings are fixed and answered inline. A local two-axis review (repo standards + issue #839) turned up four more, fixed in the same commit:

  • streaming_fragment cleared the whole per-connection bind catalog on teardown and rebuild, discarding a peer fragment's declarations — the header always claimed it cleared only its own. Now erases just its own ids; stream_bind_catalog::erase added, covered by CAT-5b. This is a prerequisite for ever running fragments concurrently.
  • Fragment::run() left outputs neither closed nor failed when execution threw, so a peer parked in wait() blocked forever with no error anywhere — the S2/S3 hazard the design doc names. It now poisons every output before unwinding.
  • build() silently ignored a partition mode on a single-output fragment, so a mistaken declare_output_hash_key() looked like it worked while every row went to destination 0. Rejected now, matching the sink's own rule.
  • Three copies of "resolve the catalog or throw" (bind function, fragment, plan generator) had drifted to three different exception types. Collapsed into one exec::catalog_for().

Scope change worth flagging: this no longer says Closes #839. The issue's central requirement is a session that "starts it on the existing task scheduler without blocking", and run() blocks — sirius_engine::execute() takes the future from start_query() and immediately waits. Combined with no push/pull/wait on the FFI, a genuinely remote sender can't feed a fragment. That's tracked in #1590; this PR is now "Part of #839".

⚠️ The base commit cc90c816 still carries a Closes #839 trailer in its message. If this is squash-merged with the default generated message, it will auto-close #839 anyway — worth trimming that line at merge time (or say the word and I'll amend the message).

@aocsa
aocsa force-pushed the stream/15-fragment branch from d61c9c3 to 8a2277e Compare August 17, 2026 04:47
@aocsa
aocsa force-pushed the stream/14-session branch from a649cf5 to 42ed7f3 Compare August 17, 2026 04:47
@aocsa
aocsa requested review from a team as code owners August 17, 2026 04:47
@aocsa
aocsa requested a balanced review from Copilot August 17, 2026 17:56

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Copilot reviewed 18 out of 18 changed files in this pull request and generated 3 comments.

Suppressed comments (3)

src/sirius_ffi.cpp:519

  • Validate sender_id before draining the source. For an undeclared sender, all batches are currently moved first and only the final close_input() throws, leaving the source consumed and the destination waiting for its actual expected sender set. resolved_inputs already contains the normalized sender set, so this can fail before any data moves.
  auto declared_it = impl_->resolved_inputs.find(input_stream_id);
  if (declared_it == impl_->resolved_inputs.end()) {
    throw sirius::invalid_input_exception("Fragment: relay target input stream " +
                                          std::to_string(input_stream_id) +
                                          " was never declared on this fragment");
  }

src/sirius_ffi.cpp:500

  • Enforce the documented “before this->run()” precondition here. After the destination has run, this method currently pulls and removes the first source batch before the destination push reports that its input is terminal, so a misuse irreversibly loses source data before throwing.

This issue also appears on line 514 of the same file.

  if (!impl_->built) {
    throw sirius::invalid_input_exception("Fragment: build() must run before relay_from()");
  }

src/include/sirius_ffi.hpp:119

  • This public contract says broadcast with one output is a no-op, but Fragment::build() rejects broadcast whenever fewer than two outputs are declared. Align the documentation with the enforced API so callers do not discover the restriction only at build time.
  /// Every output receives the full fragment output (broadcast sink). With a single declared
  /// output this is a no-op. Mutually exclusive with declare_output_hash_key.
  /// @throws after build().

Comment thread src/sirius_extension.cpp
Comment thread src/exec/streaming_fragment.cpp Outdated
Comment thread src/exec/stream_bind_catalog.cpp
@aocsa
aocsa force-pushed the stream/15-fragment branch from 8a2277e to a7bb47e Compare August 18, 2026 00:50
@aocsa
aocsa requested review from a team as code owners August 18, 2026 00:50
@aocsa
aocsa requested review from felipeblazing and removed request for a team August 18, 2026 00:50
@aocsa
aocsa changed the base branch from stream/14-session to dev August 18, 2026 00:50
aocsa added 2 commits August 18, 2026 00:57
…ragment FFI

Turns a plan and a set of stream ids into something runnable. `streaming_fragment`
binds a logical plan against a `stream_bind_catalog`, rewrites its stream leaves
to the `batch_stream` handles the session resolves, builds the physical plan, and
runs it to completion.

`stream_bind_catalog` and `stream_plan_bindings` are the binding layer: they let a
plan name its inputs and outputs by id before those streams exist, which is what
makes a fragment constructible before its peers have started.

The FFI surface (`sirius::ffi::Fragment`) exposes one fragment to Rust:
`relay_from` to chain fragments, `result_to_arrow` to pull the tail, and the
build/run lifecycle.

Closes sirius-db#839
Kept as its own commit so reviewers can see what moved since they last looked.

**Correctness**

- `normalize_key_cast_types` indexed `output_types[key]` unchecked. The sink validates key
  ranges too, but it is constructed *after* this runs, so a negative or out-of-range key was
  undefined behaviour before the sink's own guard could fire.
- `sirius_stream_source` cast a signed INT64 stream id straight to `stream_id_t`. A negative id
  wrapped to a huge unsigned value and surfaced as "no input stream declared with id
  18446744073709551615"; it is now rejected as negative.
- `Fragment::Impl::end_lifecycle()` committed when the transaction was still open. That branch
  is only reachable when setup failed — `build()` clears the flag as soon as its own commit
  succeeds — so it persisted a half-declared fragment. Rolls back instead.
- `Fragment::relay_from()` drained until `pull()` returned nullopt, which means "empty" as well
  as "ended". Called before the source ran, it closed the input after zero batches and silently
  truncated the result. The documented "call after source.run()" ordering is now enforced.
- `Fragment::relay_from()` skipped its schema check entirely when the target id was undeclared
  or the source was a result fragment, moving batches unchecked — the opposite of the "@throws
  on unknown stream id" contract. Both cases now throw.
- `Fragment::run()` left output streams neither closed nor failed when execution threw, so a
  peer parked in `wait()` blocked forever with no error anywhere (the S2/S3 hazard in the design
  doc). It now poisons every output before unwinding.
- `streaming_fragment` cleared the *whole* per-connection bind catalog on teardown and rebuild,
  discarding a peer fragment's declarations. It erases only the ids it declared, which is what
  the header always claimed. Adds `stream_bind_catalog::erase`.
- `Fragment::build()` silently ignored a partition mode when the fragment had one output, so a
  mistaken `declare_output_hash_key()` looked like it worked while every row went to
  destination 0. Rejected now, matching the sink's own rule.

**Consistency**

- Three copies of "resolve the catalog or throw" — in the bind function, the fragment, and the
  plan generator — collapse into one `exec::catalog_for()`. They had drifted to three different
  exception types (`std::runtime_error`, `duckdb::InternalException`,
  `sirius::invalid_input_exception`); the shared one throws the sirius exception its neighbours
  in `src/exec/` use.
- Dropped a comment referring to `integration::streaming_fragment::build`, a namespace that does
  not exist in this repo.
@aocsa
aocsa marked this pull request as draft August 18, 2026 02:21
Second round of review fixes on top of 4431213 — Copilot's latest pass plus an
independent review surfaced five more real issues, all confirmed by tracing the
actual failure paths before fixing.

**Correctness**

- `Fragment::Impl::declare_streams()` still called `stream_bind_catalog::clear()`,
  wiping every fragment's declarations on the shared connection — the exact bug
  class 4431213 fixed one file over in `streaming_fragment.cpp`, left open here.
  `declare()` already overwrites same-id entries, so the blanket clear() is
  removed rather than replaced with erase().
- The partition-mode guard in `Fragment::build()` only ran on the non-result
  branch, so it could never fire for a 0-output fragment — a caller could declare
  a hash key or broadcast mode and never call `declare_output()`, and build()
  would silently drop the routing spec. Moved the check above the is_result()
  split so it covers 0 and 1 output streams alike.
- `stream_bind_catalog::set_built()` silently overwrote the built-operator
  pointer when the same declared stream id was read twice in one plan (e.g. a
  self-join): the earlier leaf never got wired to the session, so its pipeline
  waited forever with no error. Now rejects the second bind.
- `streaming_fragment::run()` left output-poisoning-on-failure to the FFI
  wrapper one layer up; a direct caller of `streaming_fragment` got no such
  protection. `fail_output()` is idempotent, so the core class now poisons its
  own outputs too — safe to do at both layers.
- The transparent (normal SQL) connection path never installed a
  `stream_bind_catalog`, so `sirius_stream_source` and any `streaming_fragment`
  built outside the FFI's private embedded connection failed immediately. Now
  installed/removed alongside the existing per-connection state.

**Docs**

- `declare_output_broadcast()`'s comment still said "no-op" on a single output;
  it throws since 4431213. Fixed, and added the same note to
  `declare_output_hash_key()`.
- `stream_bind_catalog`'s class doc said "one fragment per connection at a
  time," which the relay_from chaining design directly contradicts.

**Tests**

- Adds test/cpp/exec/test_sirius_ffi_fragment.cpp, the first C++ unit coverage
  of the sirius::ffi::Fragment/Context surface: a failed build() rolls back its
  setup transaction cleanly enough that a second, independent Fragment on the
  same Context can attempt its own build() right after (both via a fresh
  Fragment and via one dropped mid-failure).
streaming-sessions.md's exec::streaming_fragment/Fragment FFI sections had grown
stale relative to the review-fix commits — the two-phase build() transaction, the
catalog erase/clear discipline, both partition-mode-guard cases, relay_from's two
preconditions, the self-join duplicate-read guard, and output poisoning at both
layers weren't covered. Gives that layer its own document instead of patching the
old sections further.

- Adds docs/super-sirius/streaming-fragments.md: stream_bind_catalog +
  sirius_stream_source (the bind-time/plan-time bridge), exec::streaming_fragment,
  and sirius::ffi::Fragment/Context, with a Tests table and a note on why this PR
  doesn't close sirius-db#839 (run() still blocks; no push/pull/wait — tracked in sirius-db#1590).
- Trims streaming-sessions.md's now-superseded sections to a pointer at the new
  doc; moves the fragment-layer test rows out of its Tests table.
- Adds the new doc to the README index, next to Streaming Sessions.
aocsa added a commit to aocsa/sirius that referenced this pull request Aug 18, 2026
Completes the Rust half of sirius-db#1396; the C++ half shipped in sirius-db#1481.

Binds the full `Fragment` surface `sirius_ffi.hpp` exports: the lifecycle
(`make_fragment`, `build`, `run`), input declaration (`declare_input_column`,
`declare_input_sender`, `close_input`), output declaration (`declare_output`,
`declare_output_broadcast`, `declare_output_hash_key`), and the data verbs
(`relay_from`, `result_to_arrow`, `output_batch_count`, `output_types`), plus
the `stream_view_name` convention helper.

Four of those — `declare_output_broadcast`, `declare_output_hash_key`,
`close_input`, `output_types` — postdate the original sirius-db#1396 draft and are bound
here for the first time.

**A fragment borrows the context; it does not own it.** That drives the shape.
Several fragments of one query are alive simultaneously — senders parked,
waiting on their receiver — so the factory has to be
`SiriusContext::fragment(&self)`; a `&mut self` factory would permit exactly
one, which is the one thing a multi-fragment query cannot live with. Every call
into C++ still needs `Pin<&mut Context>`, so the handle moves behind a
`RefCell<UniquePtr<Context>>`, with the borrow taken and released inside each
call and never held across one. `SiriusContext` is consequently neither `Send`
nor `Sync`, which is documented on the type.

BREAKING CHANGE: `SiriusContext::execute_substrait` and
`execute_substrait_result` now take `&self` rather than `&mut self` — they
cannot demand an exclusive borrow the fragment factory has already given away.

The doc comments mirror the preconditions sirius-db#1481 made explicit, so they surface
as `Err` rather than as a surprising exception: a partition mode on a
fragment with fewer than two outputs is rejected; `relay_from` requires the
source to have run, the target stream to be declared, and a source that is not
a result fragment; and `build` rejects a plan reading the same input stream id
twice, which used to strand the first reader's pipeline.

`Fragment<'ctx>` carries the context's lifetime, so a fragment cannot outlive
the engine it was built on.
Two formatting-only fixes reported by CI's pre-commit run.
aocsa added a commit to aocsa/sirius that referenced this pull request Aug 18, 2026
Completes the Rust half of sirius-db#1396; the C++ half shipped in sirius-db#1481.

Binds the full `Fragment` surface `sirius_ffi.hpp` exports: the lifecycle
(`make_fragment`, `build`, `run`), input declaration (`declare_input_column`,
`declare_input_sender`, `close_input`), output declaration (`declare_output`,
`declare_output_broadcast`, `declare_output_hash_key`), and the data verbs
(`relay_from`, `result_to_arrow`, `output_batch_count`, `output_types`), plus
the `stream_view_name` convention helper.

Four of those — `declare_output_broadcast`, `declare_output_hash_key`,
`close_input`, `output_types` — postdate the original sirius-db#1396 draft and are bound
here for the first time.

**A fragment borrows the context; it does not own it.** That drives the shape.
Several fragments of one query are alive simultaneously — senders parked,
waiting on their receiver — so the factory has to be
`SiriusContext::fragment(&self)`; a `&mut self` factory would permit exactly
one, which is the one thing a multi-fragment query cannot live with. Every call
into C++ still needs `Pin<&mut Context>`, so the handle moves behind a
`RefCell<UniquePtr<Context>>`, with the borrow taken and released inside each
call and never held across one. `SiriusContext` is consequently neither `Send`
nor `Sync`, which is documented on the type.

BREAKING CHANGE: `SiriusContext::execute_substrait` and
`execute_substrait_result` now take `&self` rather than `&mut self` — they
cannot demand an exclusive borrow the fragment factory has already given away.

The doc comments mirror the preconditions sirius-db#1481 made explicit, so they surface
as `Err` rather than as a surprising exception: a partition mode on a
fragment with fewer than two outputs is rejected; `relay_from` requires the
source to have run, the target stream to be declared, and a source that is not
a result fragment; and `build` rejects a plan reading the same input stream id
twice, which used to strand the first reader's pipeline.

`Fragment<'ctx>` carries the context's lifetime, so a fragment cannot outlive
the engine it was built on.
@aocsa
aocsa marked this pull request as ready for review August 18, 2026 04:12
aocsa added a commit to aocsa/sirius that referenced this pull request Aug 18, 2026
Completes the Rust half of sirius-db#1396; the C++ half shipped in sirius-db#1481.

Binds the full `Fragment` surface `sirius_ffi.hpp` exports: the lifecycle
(`make_fragment`, `build`, `run`), input declaration (`declare_input_column`,
`declare_input_sender`, `close_input`), output declaration (`declare_output`,
`declare_output_broadcast`, `declare_output_hash_key`), and the data verbs
(`relay_from`, `result_to_arrow`, `output_batch_count`, `output_types`), plus
the `stream_view_name` convention helper.

Four of those — `declare_output_broadcast`, `declare_output_hash_key`,
`close_input`, `output_types` — postdate the original sirius-db#1396 draft and are bound
here for the first time.

**A fragment borrows the context; it does not own it.** That drives the shape.
Several fragments of one query are alive simultaneously — senders parked,
waiting on their receiver — so the factory has to be
`SiriusContext::fragment(&self)`; a `&mut self` factory would permit exactly
one, which is the one thing a multi-fragment query cannot live with. Every call
into C++ still needs `Pin<&mut Context>`, so the handle moves behind a
`RefCell<UniquePtr<Context>>`, with the borrow never outliving the call that
takes it. `SiriusContext` is consequently neither `Send` nor `Sync`, and the
"one fragment between build() and run()" invariant is no longer enforced by the
type system — the engine rejects a violation at runtime instead. Both are
documented on the type.

BREAKING CHANGE: `SiriusContext::execute_substrait` and
`execute_substrait_result` now take `&self` rather than `&mut self` — they
cannot demand an exclusive borrow the fragment factory has already given away.

The doc comments record where each precondition is actually enforced, which is
not always the call that caused it: the two routing modes are mutually
exclusive at declaration time, but "a routing mode needs at least two
destinations" cannot be known until `build()` and is raised there, alongside
the rejection of a plan that reads one input stream id twice. `relay_from`
requires the source to have run, the target stream to be declared, and a source
that is not a result fragment.

Both Arrow-producing paths share one `drain_arrow` helper rather than two
copies of the raw-address sequence — it is the only place an address crosses
the FFI boundary, so it should exist once.

`Fragment<'ctx>` carries the context's lifetime, so a fragment cannot outlive
the engine it was built on.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants