Add debug name to LIR planning (MaterializeInc#11721)

Author: umanwizard

src/dataflow-types/src/plan/mod.rs

@@ -314,6 +314,25 @@ pub enum Plan<T = mz_repr::Timestamp> {
     },
 }
 
+/// Various bits of state to print along with error messages during LIR planning,
+/// to aid debugging.
+#[derive(Copy, Clone, Debug)]
+pub struct LirDebugInfo<'a> {
+    debug_name: &'a str,
+    id: GlobalId,
+    dataflow_uuid: uuid::Uuid,
+}
+
+impl<'a> std::fmt::Display for LirDebugInfo<'a> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "Debug name: {}; id: {}; dataflow UUID: {}",
+            self.debug_name, self.id, self.dataflow_uuid
+        )
+    }
+}
+
 impl<T: timely::progress::Timestamp> Plan<T> {
     /// Replace the plan with another one
     /// that has the collection in some additional forms.
@@ -379,6 +398,7 @@ impl<T: timely::progress::Timestamp> Plan<T> {
     pub fn from_mir(
         expr: &MirRelationExpr,
         arrangements: &mut BTreeMap<Id, AvailableCollections>,
+        debug_info: LirDebugInfo<'_>,
     ) -> Result<(Self, AvailableCollections), ()> {
         // This function is recursive and can overflow its stack, so grow it if
         // needed. The growth here is unbounded. Our general solution for this problem
@@ -389,12 +409,13 @@ impl<T: timely::progress::Timestamp> Plan<T> {
         // to allow the unbounded growth here. We are though somewhat protected by
         // higher levels enforcing their own limits on stack depth (in the parser,
         // transformer/desugarer, and planner).
-        mz_ore::stack::maybe_grow(|| Plan::from_mir_inner(expr, arrangements))
+        mz_ore::stack::maybe_grow(|| Plan::from_mir_inner(expr, arrangements, debug_info))
     }
 
     fn from_mir_inner(
         expr: &MirRelationExpr,
         arrangements: &mut BTreeMap<Id, AvailableCollections>,
+        debug_info: LirDebugInfo<'_>,
     ) -> Result<(Self, AvailableCollections), ()> {
         // Extract a maximally large MapFilterProject from `expr`.
         // We will then try and push this in to the resulting expression.
@@ -486,12 +507,12 @@ impl<T: timely::progress::Timestamp> Plan<T> {
 
                 // Plan the value using only the initial arrangements, but
                 // introduce any resulting arrangements bound to `id`.
-                let (value, v_keys) = Plan::from_mir(value, arrangements)?;
+                let (value, v_keys) = Plan::from_mir(value, arrangements, debug_info)?;
                 let pre_existing = arrangements.insert(Id::Local(*id), v_keys);
                 assert!(pre_existing.is_none());
                 // Plan the body using initial and `value` arrangements,
                 // and then remove reference to the value arrangements.
-                let (body, b_keys) = Plan::from_mir(body, arrangements)?;
+                let (body, b_keys) = Plan::from_mir(body, arrangements, debug_info)?;
                 arrangements.remove(&Id::Local(*id));
                 // Return the plan, and any `body` arrangements.
                 (
@@ -504,7 +525,7 @@ impl<T: timely::progress::Timestamp> Plan<T> {
                 )
             }
             MirRelationExpr::FlatMap { input, func, exprs } => {
-                let (input, keys) = Plan::from_mir(input, arrangements)?;
+                let (input, keys) = Plan::from_mir(input, arrangements, debug_info)?;
                 // This stage can absorb arbitrary MFP instances.
                 let mfp = mfp.take();
                 let mut exprs = exprs.clone();
@@ -549,7 +570,7 @@ impl<T: timely::progress::Timestamp> Plan<T> {
                 let mut input_keys = Vec::new();
                 let mut input_arities = Vec::new();
                 for input in inputs.iter() {
-                    let (plan, keys) = Plan::from_mir(input, arrangements)?;
+                    let (plan, keys) = Plan::from_mir(input, arrangements, debug_info)?;
                     input_arities.push(input.arity());
                     plans.push(plan);
                     input_keys.push(keys);
@@ -599,7 +620,8 @@ impl<T: timely::progress::Timestamp> Plan<T> {
                             // are available. Print an error message, to increase the chances that
                             // the user will tell us about this.
                             soft_panic_or_log!("Arrangements depended on by delta join alarmingly absent: {:?}
-This is not expected to cause incorrect results, but could indicate a performance issue in Materialize.", missing);
+Dataflow info: {}
+This is not expected to cause incorrect results, but could indicate a performance issue in Materialize.", missing, debug_info);
                         } else {
                             // It's fine and expected that linear joins don't have all their arrangements available up front,
                             // so no need to print an error here.
@@ -631,7 +653,7 @@ This is not expected to cause incorrect results, but could indicate a performanc
             } => {
                 let input_arity = input.arity();
                 let output_arity = group_key.len() + aggregates.len();
-                let (input, keys) = Self::from_mir(input, arrangements)?;
+                let (input, keys) = Self::from_mir(input, arrangements, debug_info)?;
                 let (input_key, permutation_and_new_arity) = if let Some((
                     input_key,
                     permutation,
@@ -674,7 +696,7 @@ This is not expected to cause incorrect results, but could indicate a performanc
                 monotonic,
             } => {
                 let arity = input.arity();
-                let (input, keys) = Self::from_mir(input, arrangements)?;
+                let (input, keys) = Self::from_mir(input, arrangements, debug_info)?;
 
                 let top_k_plan = TopKPlan::create_from(
                     group_key.clone(),
@@ -703,7 +725,7 @@ This is not expected to cause incorrect results, but could indicate a performanc
             }
             MirRelationExpr::Negate { input } => {
                 let arity = input.arity();
-                let (input, keys) = Self::from_mir(input, arrangements)?;
+                let (input, keys) = Self::from_mir(input, arrangements, debug_info)?;
 
                 // We don't have an MFP here -- install an operator to permute the
                 // input, if necessary.
@@ -722,7 +744,7 @@ This is not expected to cause incorrect results, but could indicate a performanc
             }
             MirRelationExpr::Threshold { input } => {
                 let arity = input.arity();
-                let (input, keys) = Self::from_mir(input, arrangements)?;
+                let (input, keys) = Self::from_mir(input, arrangements, debug_info)?;
                 // We don't have an MFP here -- install an operator to permute the
                 // input, if necessary.
                 let input = if !keys.raw {
@@ -759,10 +781,10 @@ This is not expected to cause incorrect results, but could indicate a performanc
             MirRelationExpr::Union { base, inputs } => {
                 let arity = base.arity();
                 let mut plans_keys = Vec::with_capacity(1 + inputs.len());
-                let (plan, keys) = Self::from_mir(base, arrangements)?;
+                let (plan, keys) = Self::from_mir(base, arrangements, debug_info)?;
                 plans_keys.push((plan, keys));
                 for input in inputs.iter() {
-                    let (plan, keys) = Self::from_mir(input, arrangements)?;
+                    let (plan, keys) = Self::from_mir(input, arrangements, debug_info)?;
                     plans_keys.push((plan, keys));
                 }
                 let plans = plans_keys
@@ -783,7 +805,7 @@ This is not expected to cause incorrect results, but could indicate a performanc
             }
             MirRelationExpr::ArrangeBy { input, keys } => {
                 let arity = input.arity();
-                let (input, mut input_keys) = Self::from_mir(input, arrangements)?;
+                let (input, mut input_keys) = Self::from_mir(input, arrangements, debug_info)?;
                 let keys = keys.iter().cloned().map(|k| {
                     let (permutation, thinning) = permutation_for_arrangement(&k, arity);
                     (k, permutation, thinning)
@@ -932,7 +954,15 @@ This is not expected to cause incorrect results, but could indicate a performanc
         // Build each object in order, registering the arrangements it forms.
         let mut objects_to_build = Vec::with_capacity(desc.objects_to_build.len());
         for build in desc.objects_to_build.into_iter() {
-            let (plan, keys) = Self::from_mir(&build.plan, &mut arrangements)?;
+            let (plan, keys) = Self::from_mir(
+                &build.plan,
+                &mut arrangements,
+                LirDebugInfo {
+                    debug_name: &desc.debug_name,
+                    id: build.id,
+                    dataflow_uuid: desc.id,
+                },
+            )?;
             arrangements.insert(Id::Global(build.id), keys);
             objects_to_build.push(crate::BuildDesc { id: build.id, plan });
         }