feat(pass2): share SQL table scans; a coarser grouping reads a finer grouping's pane chain, newest panes rolled up (#878 part 2)

crates/devtools/src/candidate_labels.rs

@@ -216,8 +216,18 @@ pub fn pass2_label(
         .collect();
     let shared = pass2.shared_pane_chains.iter().map(|shared| {
         let build = &candidates[shared.members[0]].build;
+        // The chain is one member's build in panes (a coarser member reads
+        // it rolled up).
         let chain = |node: &OperatorNode| {
-            is_pane_chain_of(node, build, shared.pane_width_ms, shared.pane_count, true)
+            shared.members.iter().any(|&m| {
+                is_pane_chain_of(
+                    node,
+                    &candidates[m].build,
+                    shared.pane_width_ms,
+                    shared.pane_count,
+                    true,
+                )
+            })
         };
         let mut text = format!(
             "one {} in {} × {} panes",

crates/executor/src/operators/mod.rs

@@ -183,6 +183,11 @@ enum Kind {
     SummaryMerge {
         state: usize,
         groups: Vec<usize>,
+        /// Whether a merge of no input state outputs one empty state, as a
+        /// global SQL aggregate has one row over no input rows. Only a merge
+        /// with no groups; `false` (no output row) by default.
+        #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+        empty_global_state: bool,
     },
     Evaluation {
         state: usize,

crates/executor/src/operators/summary/mod.rs

@@ -278,11 +278,34 @@ impl Operator {
             .collect::<Vec<_>>();
         fields.push(input.fields[state].clone());
         Ok(Self {
-            kind: Kind::SummaryMerge { state, groups },
+            kind: Kind::SummaryMerge {
+                state,
+                groups,
+                empty_global_state: false,
+            },
             inputs: vec![input],
             output: schema(fields),
         })
     }
+    /// A merge into no groups that outputs one empty state when it has no
+    /// input state: the global SQL aggregate a rollup of a relation's
+    /// grouped states to no keys stands for, whose one row is NULL over no
+    /// input rows.
+    pub fn with_empty_global_state(mut self) -> Result<Self, Error> {
+        match &mut self.kind {
+            Kind::SummaryMerge {
+                groups,
+                empty_global_state,
+                ..
+            } if groups.is_empty() => *empty_global_state = true,
+            _ => {
+                return Err(invalid(
+                    "an empty global state needs a merge into no groups",
+                ))
+            }
+        }
+        Ok(self)
+    }
     pub fn evaluation(
         input: SchemaRef,
         state: usize,
@@ -485,10 +508,20 @@ pub(super) fn execute_merge<'a>(
     let input_schema = operator.inputs[0].clone();
     let input = inputs.pop().ok_or_else(|| invalid("input missing"))?;
     Ok(futures::stream::once(async move {
-        let Kind::SummaryMerge { state, groups } = &operator.kind else {
+        let Kind::SummaryMerge {
+            state,
+            groups,
+            empty_global_state,
+        } = &operator.kind
+        else {
             unreachable!()
         };
-        merge_summary(input, *state, groups, &input_schema, &output, &context).await
+        let merge = Merge {
+            state_column: *state,
+            groups,
+            empty_global_state: *empty_global_state,
+        };
+        merge_summary(input, merge, &input_schema, &output, &context).await
     })
     .boxed_local())
 }
@@ -831,16 +864,27 @@ fn state_indices(batch: &Batch, state: usize) -> Result<(&StateColumn, &UInt32Ar
 }
 
 /// Merge the states of each group across all input rows.
+/// A `SummaryMerge`'s parameters.
+struct Merge<'a> {
+    state_column: usize,
+    groups: &'a [usize],
+    empty_global_state: bool,
+}
+
 async fn merge_summary(
     mut input: Input<'_, Batch>,
-    state_column: usize,
-    groups: &[usize],
+    merge: Merge<'_>,
     input_schema: &SchemaRef,
     output: &SchemaRef,
     context: &RunContext,
 ) -> Result<Batch, Error> {
     use crate::summary_kernels::exact::ScalarState;
     use crate::values::StateStorage;
+    let Merge {
+        state_column,
+        groups,
+        empty_global_state,
+    } = merge;
     let family = &input_schema.fields[state_column].dtype;
     let mut batches = Vec::new();
     let mut reservations = Vec::new();
@@ -861,6 +905,13 @@ async fn merge_summary(
                 .map(|(row, group)| (index, row, group)),
         );
     }
+    // A global SQL aggregate's one row over no input rows: one empty state.
+    if entries.is_empty() && empty_global_state {
+        table.add_empty_group()?;
+        let mut updater = StateColumnUpdater::new(family)?;
+        updater.add_group()?;
+        return grouped_states(&table, updater.seal(), input_schema, output);
+    }
     let columns = batches
         .iter()
         .map(|batch| state_indices(batch, state_column))

crates/executor/src/operators/unchecked.rs

@@ -190,8 +190,17 @@ impl TryFrom<UncheckedOperator> for Operator {
             Kind::KeyedEvaluation { state, k } => {
                 Operator::keyed_evaluation(input(0)?, state, k, output.clone())?
             }
-            Kind::SummaryMerge { state, groups } => {
-                Operator::summary_merge(input(0)?, state, groups)?
+            Kind::SummaryMerge {
+                state,
+                groups,
+                empty_global_state,
+            } => {
+                let merge = Operator::summary_merge(input(0)?, state, groups)?;
+                if empty_global_state {
+                    merge.with_empty_global_state()?
+                } else {
+                    merge
+                }
             }
             Kind::Evaluation { state, query } => Operator::evaluation(input(0)?, state, query)?,
         }

crates/executor/src/physical_planner/mod.rs

@@ -626,6 +626,21 @@ fn compile_internal(
                     continue;
                 }
             }
+            // A rollup of a relation's states to no keys is a global SQL
+            // aggregate: one row, an empty state, over no input rows.
+            if let Payload::ASAP(ASAPOp::SummaryMerge {
+                group_by: Some(keys),
+                ..
+            }) = &node.payload
+            {
+                if keys.is_empty() && built_from_relation(&restored[&id]) {
+                    let operator = compile_node(node, &schemas)
+                        .and_then(Operator::with_empty_global_state)
+                        .map_err(|error| invalid(format!("node {id}: {error}")))?;
+                    physical_dag.add(id, inputs, operator)?;
+                    continue;
+                }
+            }
             // A SQL table's trailing window (and a tumbling pane's shift
             // under it) runs as a filter on the table's time column against
             // the evaluation time.
@@ -654,6 +669,21 @@ fn compile_internal(
     Ok(physical_dag)
 }
 
+/// Whether every summary state `state` merges was built from a relation's
+/// rows (a SQL table's), where a global aggregate has a row over no input
+/// row; not from a PromQL vector's, where it has none.
+fn built_from_relation(state: &OperatorNode) -> bool {
+    match state.asap() {
+        Some(ASAPOp::SummaryMerge { children, .. }) => {
+            children.iter().all(|child| built_from_relation(child))
+        }
+        Some(ASAPOp::SummaryAgg { child, .. } | ASAPOp::PaneChain { child, .. }) => {
+            child.result_kind == planner_types::ir::OperatorResultKind::Relation
+        }
+        _ => false,
+    }
+}
+
 /// The native operator of `id` when it is a range `TimeRange` or a
 /// `TimeShift` over a SQL table's rows (a table `Scan` below any stacked
 /// `TimeShift`s), `None` for any other node.

crates/integration-tests/tests/sketch_rollup.rs

@@ -340,9 +340,9 @@ async fn quantiles_rolled_up_keep_the_sketch_bound() {
     }
 }
 
-/// A regrouping merge keys only by group keys of its input states, does
-/// not stand for a global SQL aggregate, and never merges counter states
-/// of different groups.
+/// A regrouping merge keys only by group keys of its input states (or by
+/// none, a global aggregate), and never merges counter states of different
+/// groups.
 #[tokio::test]
 async fn invalid_rollups_are_rejected() {
     let accuracy = AccuracyTarget::max_error(0.05)
@@ -383,9 +383,8 @@ async fn invalid_rollups_are_rejected() {
     for keys in [vec![2], vec![1, 1]] {
         assert!(regroup(keys.clone()).is_err(), "{keys:?}");
     }
-    assert!(regroup(vec![])
-        .unwrap_err()
-        .contains(&CoverageError::GlobalRelationRegroup.to_string()));
+    // No keys: the global build over the same rows.
+    assert!(regroup(vec![]).unwrap().coverage().is_some());
     // Both keys, swapped, are still group keys of the input.
     assert!(regroup(vec![1, 0]).unwrap().coverage().is_some());
     // Rate states merge consecutive windows of one series, not groups.