explain: add WITH (projection pushdown) option to show source column demand

Author: DAlperin

src/adapter/src/explain/mir.rs

@@ -135,7 +135,12 @@ impl<'a> Explainable<'a, DataflowDescription<OptimizedMirRelationExpr>> {
             .iter_mut()
             .map(|(id, import)| {
                 let op = import.desc.arguments.operators.as_ref();
-                ExplainSource::new(*id, op, context.config.filter_pushdown)
+                ExplainSource::new(
+                    *id,
+                    op,
+                    context.config.filter_pushdown,
+                    context.config.projection_pushdown,
+                )
             })
             .collect::<Vec<_>>();
 

src/compute-types/src/explain.rs

@@ -71,7 +71,12 @@ impl<'a> DataflowDescription<Plan> {
             .iter_mut()
             .map(|(id, import)| {
                 let op = import.desc.arguments.operators.as_ref();
-                ExplainSource::new(*id, op, context.config.filter_pushdown)
+                ExplainSource::new(
+                    *id,
+                    op,
+                    context.config.filter_pushdown,
+                    context.config.projection_pushdown,
+                )
             })
             .collect::<Vec<_>>();
 
@@ -139,7 +144,12 @@ impl<'a> DataflowDescription<OptimizedMirRelationExpr> {
             .iter_mut()
             .map(|(id, import)| {
                 let op = import.desc.arguments.operators.as_ref();
-                ExplainSource::new(*id, op, context.config.filter_pushdown)
+                ExplainSource::new(
+                    *id,
+                    op,
+                    context.config.filter_pushdown,
+                    context.config.projection_pushdown,
+                )
             })
             .collect::<Vec<_>>();
 

src/expr/src/explain.rs

@@ -9,7 +9,7 @@
 
 //! `EXPLAIN` support for structures defined in this crate.
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, BTreeSet};
 use std::fmt::Formatter;
 use std::time::Duration;
 
@@ -18,7 +18,7 @@ use mz_repr::GlobalId;
 use mz_repr::explain::ExplainError::LinearChainsPlusRecursive;
 use mz_repr::explain::text::DisplayText;
 use mz_repr::explain::{
-    AnnotatedPlan, Explain, ExplainConfig, ExplainError, ExprHumanizer, ScalarOps,
+    AnnotatedPlan, Explain, ExplainConfig, ExplainError, ExprHumanizer, Indices, ScalarOps,
     UnsupportedFormat, UsedIndexes,
 };
 use mz_repr::optimize::OptimizerFeatures;
@@ -93,18 +93,51 @@ where
     }
 }
 
+/// Carries metadata about the columns demanded from a source decoder.
+/// (Only emitted when a context flag is enabled.)
+#[derive(Debug)]
+pub struct ProjectionInfo {
+    /// Input columns demanded by the source MFP (filter + map + project).
+    pub demand: Vec<usize>,
+    /// Total input arity, for deciding whether the demand is the identity.
+    pub input_arity: usize,
+}
+
+impl<'a, C, M> DisplayText<C> for HumanizedExpr<'a, ProjectionInfo, M>
+where
+    C: AsMut<Indent>,
+    M: HumanizerMode,
+{
+    fn fmt_text(&self, f: &mut Formatter<'_>, ctx: &mut C) -> std::fmt::Result {
+        let ProjectionInfo {
+            demand,
+            input_arity,
+        } = self.expr;
+
+        // Suppress the annotation when every column is demanded — that's the
+        // case where projection pushdown is a no-op.
+        if demand.len() == *input_arity {
+            return Ok(());
+        }
+
+        writeln!(f, "{}demand=({})", ctx.as_mut(), Indices(demand))
+    }
+}
+
 #[allow(missing_debug_implementations)]
 pub struct ExplainSource<'a> {
     pub id: GlobalId,
     pub op: Option<&'a MapFilterProject>,
     pub pushdown_info: Option<PushdownInfo<'a>>,
+    pub projection_info: Option<ProjectionInfo>,
 }
 
 impl<'a> ExplainSource<'a> {
     pub fn new(
         id: GlobalId,
         op: Option<&'a MapFilterProject>,
         filter_pushdown: bool,
+        projection_pushdown: bool,
     ) -> ExplainSource<'a> {
         let pushdown_info = if filter_pushdown {
             op.map(|op| {
@@ -121,10 +154,23 @@ impl<'a> ExplainSource<'a> {
             None
         };
 
+        let projection_info = if projection_pushdown {
+            op.filter(|op| !op.is_identity()).map(|op| {
+                let demand: BTreeSet<usize> = op.demand();
+                ProjectionInfo {
+                    demand: demand.into_iter().collect(),
+                    input_arity: op.input_arity,
+                }
+            })
+        } else {
+            None
+        };
+
         ExplainSource {
             id,
             op,
             pushdown_info,
+            projection_info,
         }
     }
 
@@ -155,6 +201,9 @@ where
             if let Some(pushdown_info) = &self.expr.pushdown_info {
                 self.child(pushdown_info).fmt_text(f, ctx)?;
             }
+            if let Some(projection_info) = &self.expr.projection_info {
+                self.child(projection_info).fmt_text(f, ctx)?;
+            }
             Ok(())
         })
     }

src/expr/src/explain/json.rs

@@ -11,7 +11,9 @@
 
 use mz_repr::explain::json::DisplayJson;
 
-use crate::explain::{ExplainMultiPlan, ExplainSinglePlan, ExplainSource, PushdownInfo};
+use crate::explain::{
+    ExplainMultiPlan, ExplainSinglePlan, ExplainSource, ProjectionInfo, PushdownInfo,
+};
 
 impl<'a, T: 'a> DisplayJson for ExplainSinglePlan<'a, T>
 where
@@ -47,6 +49,7 @@ where
                      id,
                      op,
                      pushdown_info,
+                     projection_info,
                  }| {
                     let mut json = serde_json::json!({
                         "id": id,
@@ -58,6 +61,15 @@ where
                         object.insert("pushdown".to_owned(), serde_json::json!(pushdown));
                     }
 
+                    if let Some(ProjectionInfo {
+                        demand,
+                        input_arity: _,
+                    }) = projection_info
+                    {
+                        let object = json.as_object_mut().unwrap();
+                        object.insert("demand".to_owned(), serde_json::json!(demand));
+                    }
+
                     json
                 },
             )

src/repr/src/explain.rs

@@ -199,6 +199,8 @@ pub struct ExplainConfig {
     pub timing: bool,
     /// Show MFP pushdown information.
     pub filter_pushdown: bool,
+    /// Show source column demand pushed into persist source decoders.
+    pub projection_pushdown: bool,
 
     /// Optimizer feature flags.
     pub features: OptimizerFeatureOverrides,
@@ -213,6 +215,7 @@ impl Default for ExplainConfig {
             cardinality: false,
             column_names: false,
             filter_pushdown: false,
+            projection_pushdown: false,
             humanized_exprs: false,
             join_impls: true,
             keys: false,
@@ -1017,6 +1020,7 @@ mod tests {
             cardinality: false,
             column_names: false,
             filter_pushdown: false,
+            projection_pushdown: false,
             humanized_exprs: false,
             join_impls: false,
             keys: false,

src/sql-parser/src/ast/defs/statement.rs

@@ -3982,6 +3982,7 @@ pub enum ExplainPlanOptionName {
     NoFastPath,
     NoNotices,
     NodeIdentifiers,
+    ProjectionPushdown,
     RawPlans,
     RawSyntax,
     Raw, // Listed after the `Raw~` variants to keep the parser happy!
@@ -4019,6 +4020,7 @@ impl WithOptionName for ExplainPlanOptionName {
             | Self::NoFastPath
             | Self::NoNotices
             | Self::NodeIdentifiers
+            | Self::ProjectionPushdown
             | Self::RawPlans
             | Self::RawSyntax
             | Self::Raw

src/sql/src/plan/statement/dml.rs

@@ -572,6 +572,7 @@ generate_extracted_config!(
     (NonNegative, bool, Default(false)),
     (NoNotices, bool, Default(false)),
     (NodeIdentifiers, bool, Default(false)),
+    (ProjectionPushdown, Option<bool>, Default(None)),
     (Raw, bool, Default(false)),
     (RawPlans, bool, Default(false)),
     (RawSyntax, bool, Default(false)),
@@ -613,6 +614,7 @@ impl TryFrom<ExplainPlanOptionExtracted> for ExplainConfig {
             cardinality: v.cardinality,
             column_names: v.column_names,
             filter_pushdown: v.filter_pushdown.unwrap_or(enable_on_prod),
+            projection_pushdown: v.projection_pushdown.unwrap_or(enable_on_prod),
             humanized_exprs: !v.raw_plans && (v.humanized_expressions.unwrap_or(enable_on_prod)),
             join_impls: v.join_implementations,
             keys: v.keys,

test/sqllogictest/projection-pushdown.slt

@@ -0,0 +1,77 @@
+# Copyright Materialize, Inc. and contributors. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+mode cockroach
+
+# Verify that `EXPLAIN ... WITH (projection pushdown)` surfaces the column
+# demand the persist source decoder will receive. The `demand=(...)` set is the
+# union of columns read by the source MFP's filter, map, and projection. It is
+# suppressed when every input column is demanded (the no-op case).
+
+statement ok
+CREATE TABLE t (a int, b int, c int, d int);
+
+# Pure projection: only `a` is demanded.
+
+statement ok
+CREATE MATERIALIZED VIEW mv_project AS SELECT a FROM t;
+
+query T multiline
+EXPLAIN PHYSICAL PLAN WITH(humanized expressions, projection pushdown) AS VERBOSE TEXT FOR MATERIALIZED VIEW mv_project
+----
+materialize.public.mv_project:
+  Get::Collection materialize.public.t
+    raw=true
+
+Source materialize.public.t
+  project=(#0)
+  demand=(#0)
+
+Target cluster: quickstart
+
+EOF
+
+# Projection plus a filter on a different column: demand is the union.
+
+statement ok
+CREATE MATERIALIZED VIEW mv_project_filter AS SELECT a FROM t WHERE c > 5;
+
+query T multiline
+EXPLAIN PHYSICAL PLAN WITH(humanized expressions, projection pushdown) AS VERBOSE TEXT FOR MATERIALIZED VIEW mv_project_filter
+----
+materialize.public.mv_project_filter:
+  Get::Collection materialize.public.t
+    raw=true
+
+Source materialize.public.t
+  project=(#0)
+  filter=((#2{c} > 5))
+  demand=(#0, #2)
+
+Target cluster: quickstart
+
+EOF
+
+# Identity projection: no demand annotation is emitted.
+
+statement ok
+CREATE MATERIALIZED VIEW mv_identity AS SELECT * FROM t;
+
+query T multiline
+EXPLAIN PHYSICAL PLAN WITH(humanized expressions, projection pushdown) AS VERBOSE TEXT FOR MATERIALIZED VIEW mv_identity
+----
+materialize.public.mv_identity:
+  Get::PassArrangements materialize.public.t
+    raw=true
+
+Source materialize.public.t
+
+Target cluster: quickstart
+
+EOF