chore(tesseract): Collapse logical plan into Query + LogicalPlan root

rust/cube/cubesqlplanner/cubesqlplanner/.gitignore

@@ -11,3 +11,7 @@ node_modules
 /cubesql/egraph-debug-intermediate
 egraph-debug
 /cubesql/debug-qtrace
+
+# insta snapshot review artefacts
+*.snap.new
+*.pending-snap

rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/full_key_aggregate.rs

@@ -6,12 +6,20 @@ use typed_builder::TypedBuilder;
 
 /// Reference to a multi-stage CTE consumed by `FullKeyAggregate`:
 /// the CTE's name plus the symbols it exposes.
-#[derive(TypedBuilder)]
+#[derive(Clone, TypedBuilder)]
 pub struct MultiStageSubqueryRef {
     name: String,
     #[builder(default)]
     symbols: Vec<Rc<MemberSymbol>>,
     schema: Rc<LogicalSchema>,
+    /// True when the CTE behind this ref projects measures as ungrouped raw
+    /// columns (no aggregate wrap yet) — the consumer of this ref must
+    /// register an `ungrouped_measure_reference` for each measure symbol,
+    /// so its own outer SELECT wraps the column in the right aggregate.
+    /// Used by the aggregate-multiplied subquery shape: its MeasureSubquery
+    /// data input is ungrouped, while keys/regular-measure refs are not.
+    #[builder(default)]
+    is_ungrouped: bool,
 }
 
 impl MultiStageSubqueryRef {
@@ -26,6 +34,10 @@ impl MultiStageSubqueryRef {
     pub fn schema(&self) -> &Rc<LogicalSchema> {
         &self.schema
     }
+
+    pub fn is_ungrouped(&self) -> bool {
+        self.is_ungrouped
+    }
 }
 
 impl PrettyPrint for MultiStageSubqueryRef {
@@ -40,35 +52,47 @@ impl PrettyPrint for MultiStageSubqueryRef {
 }
 
 /// Top-level aggregating source that stitches together several
-/// multi-stage / multi-fact CTEs into one keyed result. The
-/// physical builder picks a join strategy from `multi_stage_subquery_refs`
-/// and `use_full_join_and_coalesce`.
+/// multi-stage / multi-fact CTEs into one keyed result. The physical
+/// builder picks a join strategy from `data_inputs` and
+/// `keys_subquery_ref` — when a keys CTE is present, joins go through
+/// it on `join_keys`; otherwise data inputs are stitched directly.
 #[derive(Clone, TypedBuilder)]
 pub struct FullKeyAggregate {
     schema: Rc<LogicalSchema>,
-    use_full_join_and_coalesce: bool,
     #[builder(default)]
-    multi_stage_subquery_refs: Vec<Rc<MultiStageSubqueryRef>>,
+    data_inputs: Vec<Rc<MultiStageSubqueryRef>>,
+    #[builder(default)]
+    keys_subquery_ref: Option<Rc<MultiStageSubqueryRef>>,
+    // Members used as the JOIN keys when stitching `data_queries` onto the
+    // keys source. When empty, defaults to `schema.all_dimensions()` —
+    // historical behaviour for the multi-stage flow, where outer dimensions
+    // are both projected and used as join columns. When non-empty,
+    // decouples "what to project" (schema) from "what to join on" — needed
+    // for the multiplied-measures flow where pk dimensions drive the join
+    // while outer dimensions ride along as payload.
+    #[builder(default)]
+    join_keys: Vec<Rc<MemberSymbol>>,
 }
 
 impl FullKeyAggregate {
     pub fn schema(&self) -> &Rc<LogicalSchema> {
         &self.schema
     }
 
-    /// When true, multi-fact branches are stitched together via a
-    /// FULL OUTER JOIN over keys with COALESCE on dimension columns;
-    /// otherwise an INNER JOIN is used.
-    pub fn use_full_join_and_coalesce(&self) -> bool {
-        self.use_full_join_and_coalesce
+    pub fn data_inputs(&self) -> &Vec<Rc<MultiStageSubqueryRef>> {
+        &self.data_inputs
     }
 
-    pub fn multi_stage_subquery_refs(&self) -> &Vec<Rc<MultiStageSubqueryRef>> {
-        &self.multi_stage_subquery_refs
+    pub fn keys_subquery_ref(&self) -> &Option<Rc<MultiStageSubqueryRef>> {
+        &self.keys_subquery_ref
+    }
+
+    pub fn join_keys(&self) -> &Vec<Rc<MemberSymbol>> {
+        &self.join_keys
     }
 
     pub fn is_empty(&self) -> bool {
-        self.multi_stage_subquery_refs.is_empty()
+        self.data_inputs.is_empty()
     }
 }
 
@@ -87,8 +111,9 @@ impl LogicalNode for FullKeyAggregate {
         Ok(Rc::new(
             Self::builder()
                 .schema(self.schema().clone())
-                .use_full_join_and_coalesce(self.use_full_join_and_coalesce())
-                .multi_stage_subquery_refs(self.multi_stage_subquery_refs().clone())
+                .data_inputs(self.data_inputs().clone())
+                .keys_subquery_ref(self.keys_subquery_ref().clone())
+                .join_keys(self.join_keys().clone())
                 .build(),
         ))
     }
@@ -112,18 +137,21 @@ impl PrettyPrint for FullKeyAggregate {
         let details_state = state.new_level();
         result.println(&format!("schema:"), &state);
         self.schema().pretty_print(result, &details_state);
-        result.println(
-            &format!(
-                "use_full_join_and_coalesce: {}",
-                self.use_full_join_and_coalesce()
-            ),
-            &state,
-        );
-        if !self.multi_stage_subquery_refs().is_empty() {
-            result.println("multi_stage_subquery_refs:", &state);
-            for subquery_ref in self.multi_stage_subquery_refs().iter() {
-                subquery_ref.pretty_print(result, &details_state);
+        if !self.data_inputs().is_empty() {
+            result.println("data_inputs:", &state);
+            for input in self.data_inputs().iter() {
+                input.pretty_print(result, &details_state);
             }
         }
+        if let Some(keys_ref) = self.keys_subquery_ref() {
+            result.println("keys_subquery_ref:", &state);
+            keys_ref.pretty_print(result, &details_state);
+        }
+        if !self.join_keys.is_empty() {
+            result.println(
+                &format!("join_keys: {}", print_symbols(self.join_keys())),
+                &state,
+            );
+        }
     }
 }