From d6a374d8de919c0ebff2cd90c56a343fb506829e Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 15 Jul 2026 23:11:23 -0700 Subject: [PATCH] [adapters] Regression in gather_output_to_host. Switching to a streaming exchange in gather_output_to_host introduced a regression where the exchange was disabled on all but one host (the one that the stream is assigned to). As a result these hosts didn't collaborate in delivering outputs to the assigned host. Signed-off-by: Leonid Ryzhyk --- crates/adapters/src/static_compile/catalog.rs | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/crates/adapters/src/static_compile/catalog.rs b/crates/adapters/src/static_compile/catalog.rs index c7a2e4585a2..2189233d93e 100644 --- a/crates/adapters/src/static_compile/catalog.rs +++ b/crates/adapters/src/static_compile/catalog.rs @@ -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()),