Skip to content

[adapters] Regression in gather_output_to_host. #6647

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
blp merged 1 commit into from
Jul 16, 2026
+18 −0
Merged

[adapters] Regression in gather_output_to_host. #6647

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions crates/adapters/src/static_compile/catalog.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,24 @@ impl Catalog {
.shard_workers_accumulate(hosts[ordinal].workers.clone())
.into_parts();

// The output gather is collaborative: every host shards its slice of
// the view to the owning host's workers, which accumulate and emit
// it. The accumulator's `enable_count` is per-host and is enabled
// only when an output connector attaches, which in a multihost
// pipeline happens only on the owning host (connectors, including
// dynamic HTTP `listen`, are assigned to one host). A non-owning
// host would therefore leave its accumulator disabled and send an
// empty batch instead of its slice, silently dropping every view
// record computed on that host.
//
// Enable the accumulator on every host so the gather is always
// complete. Materialized views already get this via the integral's
// `into_enabled_stream`; do the same for the delta gather. This
// gathers a view even when it has no connector; a future
// optimization could propagate the owning host's connector state to
// the other hosts to skip that work.
enabled_count.enable();

let integral = if integrate {
Some(
stream.shard_workers_accumulate_integrate_trace(hosts[ordinal].workers.clone()),
Expand Down
Loading