## Coverage Report

#### Created: 2024-10-13 08:39

    /Users/andrewlamb/Software/datafusion/datafusion/physical-plan/src/explain.rs

    Line
    Count
    Source (jump to first uncovered line)
    1
    // Licensed to the Apache Software Foundation (ASF) under one
    2
    // or more contributor license agreements.  See the NOTICE file
    3
    // distributed with this work for additional information
    4
    // regarding copyright ownership.  The ASF licenses this file
    5
    // to you under the Apache License, Version 2.0 (the
    6
    // "License"); you may not use this file except in compliance
    7
    // with the License.  You may obtain a copy of the License at
    8
    //
    9
    //   http://www.apache.org/licenses/LICENSE-2.0
    10
    //
    11
    // Unless required by applicable law or agreed to in writing,
    12
    // software distributed under the License is distributed on an
    13
    // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
    14
    // KIND, either express or implied.  See the License for the
    15
    // specific language governing permissions and limitations
    16
    // under the License.
    17
    18
    //! Defines the EXPLAIN operator
    19
    20
    use std::any::Any;
    21
    use std::sync::Arc;
    22
    23
    use super::{DisplayAs, ExecutionMode, PlanProperties, SendableRecordBatchStream};
    24
    use crate::stream::RecordBatchStreamAdapter;
    25
    use crate::{DisplayFormatType, ExecutionPlan, Partitioning};
    26
    27
    use arrow::{array::StringBuilder, datatypes::SchemaRef, record_batch::RecordBatch};
    28
    use datafusion_common::display::StringifiedPlan;
    29
    use datafusion_common::{internal_err, Result};
    30
    use datafusion_execution::TaskContext;
    31
    use datafusion_physical_expr::EquivalenceProperties;
    32
    33
    use log::trace;
    34
    35
    /// Explain execution plan operator. This operator contains the string
    36
    /// values of the various plans it has when it is created, and passes
    37
    /// them to its output.
    38
    #[derive(Debug, Clone)]
    39
    pub struct ExplainExec {
    40
        /// The schema that this exec plan node outputs
    41
        schema: SchemaRef,
    42
        /// The strings to be printed
    43
        stringified_plans: Vec<StringifiedPlan>,
    44
        /// control which plans to print
    45
        verbose: bool,
    46
        cache: PlanProperties,
    47
    }
    48
    49
    impl ExplainExec {
    50
        /// Create a new ExplainExec
    51
    0
        pub fn new(
    52
    0
            schema: SchemaRef,
    53
    0
            stringified_plans: Vec<StringifiedPlan>,
    54
    0
            verbose: bool,
    55
    0
        ) -> Self {
    56
    0
            let cache = Self::compute_properties(Arc::clone(&schema));
    57
    0
            ExplainExec {
    58
    0
                schema,
    59
    0
                stringified_plans,
    60
    0
                verbose,
    61
    0
                cache,
    62
    0
            }
    63
    0
        }
    64
    65
        /// The strings to be printed
    66
    0
        pub fn stringified_plans(&self) -> &[StringifiedPlan] {
    67
    0
            &self.stringified_plans
    68
    0
        }
    69
    70
        /// access to verbose
    71
    0
        pub fn verbose(&self) -> bool {
    72
    0
            self.verbose
    73
    0
        }
    74
    75
        /// This function creates the cache object that stores the plan properties such as schema, equivalence properties, ordering, partitioning, etc.
    76
    0
        fn compute_properties(schema: SchemaRef) -> PlanProperties {
    77
    0
            let eq_properties = EquivalenceProperties::new(schema);
    78
    0
            PlanProperties::new(
    79
    0
                eq_properties,
    80
    0
                Partitioning::UnknownPartitioning(1),
    81
    0
                ExecutionMode::Bounded,
    82
    0
            )
    83
    0
        }
    84
    }
    85
    86
    impl DisplayAs for ExplainExec {
    87
    0
        fn fmt_as(
    88
    0
            &self,
    89
    0
            t: DisplayFormatType,
    90
    0
            f: &mut std::fmt::Formatter,
    91
    0
        ) -> std::fmt::Result {
    92
    0
            match t {
    93
                DisplayFormatType::Default | DisplayFormatType::Verbose => {
    94
    0
                    write!(f, "ExplainExec")
    95
    0
                }
    96
    0
            }
    97
    0
        }
    98
    }
    99
    100
    impl ExecutionPlan for ExplainExec {
    101
    0
        fn name(&self) -> &'static str {
    102
    0
            "ExplainExec"
    103
    0
        }
    104
    105
        /// Return a reference to Any that can be used for downcasting
    106
    0
        fn as_any(&self) -> &dyn Any {
    107
    0
            self
    108
    0
        }
    109
    110
    0
        fn properties(&self) -> &PlanProperties {
    111
    0
            &self.cache
    112
    0
        }
    113
    114
    0
        fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
    115
    0
            // this is a leaf node and has no children
    116
    0
            vec![]
    117
    0
        }
    118
    119
    0
        fn with_new_children(
    120
    0
            self: Arc<Self>,
    121
    0
            _: Vec<Arc<dyn ExecutionPlan>>,
    122
    0
        ) -> Result<Arc<dyn ExecutionPlan>> {
    123
    0
            Ok(self)
    124
    0
        }
    125
    126
    0
        fn execute(
    127
    0
            &self,
    128
    0
            partition: usize,
    129
    0
            context: Arc<TaskContext>,
    130
    0
        ) -> Result<SendableRecordBatchStream> {
    131
    0
            trace!("Start ExplainExec::execute for partition {} of context session_id {} and task_id {:?}", partition, context.session_id(), context.task_id());
    132
    0
            if 0 != partition {
    133
    0
                return internal_err!("ExplainExec invalid partition {partition}");
    134
    0
            }
    135
    0
    136
    0
            let mut type_builder =
    137
    0
                StringBuilder::with_capacity(self.stringified_plans.len(), 1024);
    138
    0
            let mut plan_builder =
    139
    0
                StringBuilder::with_capacity(self.stringified_plans.len(), 1024);
    140
    0
    141
    0
            let plans_to_print = self
    142
    0
                .stringified_plans
    143
    0
                .iter()
    144
    0
                .filter(|s| s.should_display(self.verbose));
    145
    0
    146
    0
            // Identify plans that are not changed
    147
    0
            let mut prev: Option<&StringifiedPlan> = None;
    148
    149
    0
            for p in plans_to_print {
    150
    0
                type_builder.append_value(p.plan_type.to_string());
    151
    0
                match prev {
    152
    0
                    Some(prev) if !should_show(prev, p) => {
    153
    0
                        plan_builder.append_value("SAME TEXT AS ABOVE");
    154
    0
                    }
    155
    0
                    Some(_) | None => {
    156
    0
                        plan_builder.append_value(&*p.plan);
    157
    0
                    }
    158
                }
    159
    0
                prev = Some(p);
    160
            }
    161
    162
    0
            let record_batch = RecordBatch::try_new(
    163
    0
                Arc::clone(&self.schema),
    164
    0
                vec![
    165
    0
                    Arc::new(type_builder.finish()),
    166
    0
                    Arc::new(plan_builder.finish()),
    167
    0
                ],
    168
    0
            )?;
    169
    170
    0
            trace!(
    171
    0
                "Before returning RecordBatchStream in ExplainExec::execute for partition {} of context session_id {} and task_id {:?}", partition, context.session_id(), context.task_id());
    172
    173
    0
            Ok(Box::pin(RecordBatchStreamAdapter::new(
    174
    0
                Arc::clone(&self.schema),
    175
    0
                futures::stream::iter(vec![Ok(record_batch)]),
    176
    0
            )))
    177
    0
        }
    178
    }
    179
    180
    /// If this plan should be shown, given the previous plan that was
    181
    /// displayed.
    182
    ///
    183
    /// This is meant to avoid repeating the same plan over and over again
    184
    /// in explain plans to make clear what is changing
    185
    0
    fn should_show(previous_plan: &StringifiedPlan, this_plan: &StringifiedPlan) -> bool {
    186
    0
        // if the plans are different, or if they would have been
    187
    0
        // displayed in the normal explain (aka non verbose) plan
    188
    0
        (previous_plan.plan != this_plan.plan) || this_plan.should_display(false)
    189
    0
    }