[adapters] Fix ad-hoc processing for single-worker output on multihost.

blp · blp · commit ea1b6d1c4d74 · 2026-03-04T21:40:45.000Z
When an operator produces output only to a subset of workers, for a single
host we created only as many partitions for ad-hoc processing as outputs.
For multi-host output, if we used the same model, then the coordinator
would have to inquire about the numbers of outputs on each host.  Instead,
it's easier to simply assume that there is one output per worker and
produce an empty batch for the workers that produced no output.  This
commit implements that model.

This fixes a out-of-range panic for multihost ad-hoc queries for output
operators that have fewer outputs than workers.

Signed-off-by: Ben Pfaff &lt;blp@feldera.com&gt;
diff --git a/crates/adapters/src/adhoc/table.rs b/crates/adapters/src/adhoc/table.rs
@@ -332,7 +332,7 @@ struct AdHocQueryExecution {
     indexed: bool,
     table_schema: Arc<Schema>,
     projected_schema: Arc<Schema>,
-    readers: Option<Vec<Arc<dyn SerBatchReader>>>,
+    readers: Vec<Arc<dyn SerBatchReader>>,
     projection: Option<Vec<usize>>,
     plan_properties: PlanProperties,
     children: Vec<Arc<dyn ExecutionPlan>>,
@@ -367,7 +367,7 @@ impl AdHocQueryExecution {
             indexed,
             table_schema,
             projected_schema,
-            readers,
+            readers: readers.unwrap_or_default(),
             projection: projection.cloned(),
             plan_properties,
             children: vec![],
@@ -454,11 +454,10 @@ use `with ('materialized' = 'true')` for tables, or `create materialized view` f
             Ok(())
         }
 
-        if let Some(readers) = &self.readers {
+        if let Some(batch_reader) = self.readers.get(partition).cloned() {
             let mut builder =
                 RecordBatchReceiverStreamBuilder::new(self.projected_schema.clone(), 10);
             // Returns a single batch when the returned stream is polled
-            let batch_reader = readers[partition].clone();
             let schema = self.table_schema.clone();
             let tx = builder.tx();
             let projection = self.projection.clone();
@@ -543,8 +542,18 @@ use `with ('materialized' = 'true')` for tables, or `create materialized view` f
                 builder.build(),
             )))
         } else {
-            // The case of no readers can happen if the table has never received any input &
-            // the circuit has never stepped so the correct response is to send an empty batch
+            // We did not find a reader.  There are two causes:
+            //
+            // - There are no readers at all, because the table has never
+            //   received any input, and the circuit has never stepped.
+            //
+            // - There are some readers but not for every worker, because the
+            //   output operator only produces output for one worker.  (It might
+            //   not even be our worker, it might be on a different host in
+            //   multihost.)
+            //
+            // In either case, the correct thing to do is to produce an empty
+            // batch.
             let fut =
                 futures::future::ready(Ok(RecordBatch::new_empty(self.projected_schema.clone())));
             let stream = futures::stream::once(fut);
