feat(query): support explain insert with insert source (#15357)

TCeason · web-flow · commit 37820a81f44a · 2024-04-29T11:42:57.000Z
* feat(query): support explain insert with insert source

* fix explain delete err

* refactor explain_insert into plan/insert.rs explain

* fix clippy err
diff --git a/src/meta/app/src/principal/user_stage.rs b/src/meta/app/src/principal/user_stage.rs
@@ -487,6 +487,21 @@ pub struct CopyOptions {
     pub detailed_output: bool,
 }
 
+impl Display for CopyOptions {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "OnErrorMode {}", self.on_error)?;
+        write!(f, "SizeLimit {}", self.size_limit)?;
+        write!(f, "MaxFiles {}", self.max_files)?;
+        write!(f, "SplitSize {}", self.split_size)?;
+        write!(f, "Purge {}", self.purge)?;
+        write!(f, "DisableVariantCheck {}", self.disable_variant_check)?;
+        write!(f, "ReturnFailedOnly {}", self.return_failed_only)?;
+        write!(f, "MaxFileSize {}", self.max_file_size)?;
+        write!(f, "Single {}", self.single)?;
+        write!(f, "DetailedOutput {}", self.detailed_output)
+    }
+}
+
 impl CopyOptions {
     pub fn apply(&mut self, opts: &BTreeMap<String, String>, ignore_unknown: bool) -> Result<()> {
         if opts.is_empty() {
diff --git a/src/query/catalog/src/plan/datasource/datasource_info/stage.rs b/src/query/catalog/src/plan/datasource/datasource_info/stage.rs
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::fmt::Debug;
+use std::fmt::Display;
 use std::fmt::Formatter;
 use std::sync::Arc;
 
@@ -83,3 +84,15 @@ impl Debug for StageTableInfo {
         write!(f, "{:?}", self.stage_info)
     }
 }
+
+impl Display for StageTableInfo {
+    // Ignore the schema.
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "StageName {}", self.stage_info.stage_name)?;
+        write!(f, "StageType {}", self.stage_info.stage_type)?;
+        write!(f, "StageParam {}", self.stage_info.stage_params.storage)?;
+        write!(f, "IsTemporary {}", self.stage_info.is_temporary)?;
+        write!(f, "FileFormatParams {}", self.stage_info.file_format_params)?;
+        write!(f, "CopyOption {}", self.stage_info.copy_options)
+    }
+}
diff --git a/src/query/service/src/interpreters/interpreter_explain.rs b/src/query/service/src/interpreters/interpreter_explain.rs
@@ -32,7 +32,6 @@ use databend_common_sql::optimizer::ColumnSet;
 use databend_common_sql::plans::FunctionCall;
 use databend_common_sql::plans::UpdatePlan;
 use databend_common_sql::BindContext;
-use databend_common_sql::InsertInputSource;
 use databend_common_sql::MetadataRef;
 use databend_common_storages_result_cache::gen_result_cache_key;
 use databend_common_storages_result_cache::ResultCacheReader;
@@ -97,25 +96,7 @@ impl Interpreter for ExplainInterpreter {
                     self.explain_query(s_expr, metadata, bind_context, formatted_ast)
                         .await?
                 }
-                Plan::Insert(insert_plan) => {
-                    let mut res = self.explain_plan(&self.plan)?;
-                    if let InsertInputSource::SelectPlan(plan) = &insert_plan.source {
-                        if let Plan::Query {
-                            s_expr,
-                            metadata,
-                            bind_context,
-                            formatted_ast,
-                            ..
-                        } = &**plan
-                        {
-                            let query = self
-                                .explain_query(s_expr, metadata, bind_context, formatted_ast)
-                                .await?;
-                            res.extend(query);
-                        }
-                    }
-                    vec![DataBlock::concat(&res)?]
-                }
+                Plan::Insert(insert_plan) => insert_plan.explain(self.config.verbose).await?,
                 Plan::CreateTable(plan) => match &plan.as_select {
                     Some(box Plan::Query {
                         s_expr,
@@ -557,6 +538,18 @@ impl ExplainInterpreter {
             result.extend(input);
         }
 
+        if result.is_empty() {
+            let table_name = format!(
+                "{}.{}.{}",
+                delete.catalog_name, delete.database_name, delete.table_name
+            );
+            let children = vec![FormatTreeNode::new(format!("table: {table_name}"))];
+            let formatted_plan = FormatTreeNode::with_children("DeletePlan:".to_string(), children)
+                .format_pretty()?;
+            let line_split_result: Vec<&str> = formatted_plan.lines().collect();
+            let formatted_plan = StringType::from_data(line_split_result);
+            result.push(DataBlock::new_from_columns(vec![formatted_plan]));
+        }
         Ok(vec![DataBlock::concat(&result)?])
     }
 }
diff --git a/src/query/sql/src/planner/plans/copy_into_table.rs b/src/query/sql/src/planner/plans/copy_into_table.rs
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::fmt::Debug;
+use std::fmt::Display;
 use std::fmt::Formatter;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -47,6 +48,17 @@ pub enum ValidationMode {
     ReturnAllErrors,
 }
 
+impl Display for ValidationMode {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            ValidationMode::None => write!(f, ""),
+            ValidationMode::ReturnNRows(v) => write!(f, "RETURN_ROWS={v}"),
+            ValidationMode::ReturnErrors => write!(f, "RETURN_ERRORS"),
+            ValidationMode::ReturnAllErrors => write!(f, "RETURN_ALL_ERRORS"),
+        }
+    }
+}
+
 impl FromStr for ValidationMode {
     type Err = String;
     fn from_str(s: &str) -> Result<Self, String> {
@@ -75,6 +87,21 @@ pub enum CopyIntoTableMode {
     Copy,
 }
 
+impl Display for CopyIntoTableMode {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            CopyIntoTableMode::Insert { overwrite } => {
+                if *overwrite {
+                    write!(f, "INSERT OVERWRITE")
+                } else {
+                    write!(f, "INSERT")
+                }
+            }
+            CopyIntoTableMode::Replace => write!(f, "REPLACE"),
+            CopyIntoTableMode::Copy => write!(f, "COPY"),
+        }
+    }
+}
 impl CopyIntoTableMode {
     pub fn is_overwrite(&self) -> bool {
         match self {
diff --git a/src/query/sql/src/planner/plans/insert.rs b/src/query/sql/src/planner/plans/insert.rs
@@ -14,7 +14,11 @@
 
 use std::sync::Arc;
 
+use databend_common_ast::ast::FormatTreeNode;
+use databend_common_expression::types::StringType;
+use databend_common_expression::DataBlock;
 use databend_common_expression::DataSchemaRef;
+use databend_common_expression::FromData;
 use databend_common_expression::Scalar;
 use databend_common_expression::TableSchemaRef;
 use databend_common_meta_app::principal::FileFormatParams;
@@ -26,6 +30,7 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use super::Plan;
+use crate::plans::CopyIntoTablePlan;
 
 #[derive(Clone, Debug, EnumAsInner)]
 pub enum InsertInputSource {
@@ -82,6 +87,152 @@ impl Insert {
     pub fn has_select_plan(&self) -> bool {
         matches!(&self.source, InsertInputSource::SelectPlan(_))
     }
+
+    #[async_backtrace::framed]
+    pub async fn explain(
+        &self,
+        verbose: bool,
+    ) -> databend_common_exception::Result<Vec<DataBlock>> {
+        let mut result = vec![];
+
+        let Insert {
+            catalog,
+            database,
+            table,
+            schema,
+            overwrite,
+            // table_info only used create table as select.
+            table_info: _,
+            source,
+        } = self;
+
+        let table_name = format!("{}.{}.{}", catalog, database, table);
+        let inserted_columns = schema
+            .fields
+            .iter()
+            .map(|field| format!("{}.{} (#{})", table, field.name, field.column_id))
+            .collect::<Vec<_>>()
+            .join(",");
+
+        let mut children = vec![
+            FormatTreeNode::new(format!("table: {table_name}")),
+            FormatTreeNode::new(format!("inserted columns: [{inserted_columns}]")),
+            FormatTreeNode::new(format!("overwrite: {overwrite}")),
+        ];
+
+        let mut formatted_plan = String::new();
+
+        match source {
+            InsertInputSource::SelectPlan(plan) => {
+                if let Plan::Query {
+                    s_expr, metadata, ..
+                } = &**plan
+                {
+                    let metadata = &*metadata.read();
+                    let sub_tree = s_expr.to_format_tree(metadata, verbose)?;
+                    children.push(sub_tree);
+                    formatted_plan = FormatTreeNode::with_children(
+                        "InsertPlan (subquery):".to_string(),
+                        children,
+                    )
+                    .format_pretty()?;
+                }
+            }
+            InsertInputSource::Values(values) => match values {
+                InsertValue::Values { .. } => {
+                    formatted_plan =
+                        FormatTreeNode::with_children("InsertPlan (values):".to_string(), children)
+                            .format_pretty()?;
+                }
+                InsertValue::RawValues { .. } => {
+                    formatted_plan = FormatTreeNode::with_children(
+                        "InsertPlan (rawvalues):".to_string(),
+                        children,
+                    )
+                    .format_pretty()?;
+                }
+            },
+            InsertInputSource::Stage(plan) => match *plan.clone() {
+                Plan::CopyIntoTable(copy_plan) => {
+                    let CopyIntoTablePlan {
+                        no_file_to_copy,
+                        from_attachment,
+                        required_values_schema,
+                        required_source_schema,
+                        write_mode,
+                        validation_mode,
+                        force,
+                        stage_table_info,
+                        enable_distributed,
+                        ..
+                    } = &*copy_plan;
+                    let required_values_schema = required_values_schema
+                        .fields()
+                        .iter()
+                        .map(|field| field.name().to_string())
+                        .collect::<Vec<_>>()
+                        .join(",");
+                    let required_source_schema = required_source_schema
+                        .fields()
+                        .iter()
+                        .map(|field| field.name().to_string())
+                        .collect::<Vec<_>>()
+                        .join(",");
+                    let stage_node = vec![
+                        FormatTreeNode::new(format!("no_file_to_copy: {no_file_to_copy}")),
+                        FormatTreeNode::new(format!("from_attachment: {from_attachment}")),
+                        FormatTreeNode::new(format!(
+                            "required_values_schema: [{required_values_schema}]"
+                        )),
+                        FormatTreeNode::new(format!(
+                            "required_source_schema: [{required_source_schema}]"
+                        )),
+                        FormatTreeNode::new(format!("write_mode: {write_mode}")),
+                        FormatTreeNode::new(format!("validation_mode: {validation_mode}")),
+                        FormatTreeNode::new(format!("force: {force}")),
+                        FormatTreeNode::new(format!("stage_table_info: {stage_table_info}")),
+                        FormatTreeNode::new(format!("enable_distributed: {enable_distributed}")),
+                    ];
+                    children.extend(stage_node);
+                    formatted_plan =
+                        FormatTreeNode::with_children("InsertPlan (stage):".to_string(), children)
+                            .format_pretty()?;
+                }
+                _ => unreachable!("plan in InsertInputSource::Stag must be CopyIntoTable"),
+            },
+            InsertInputSource::StreamingWithFileFormat {
+                format,
+                on_error_mode,
+                start,
+                ..
+            } => {
+                let stage_node = vec![
+                    FormatTreeNode::new(format!("format: {format}")),
+                    FormatTreeNode::new(format!("on_error_mode: {on_error_mode}")),
+                    FormatTreeNode::new(format!("start: {start}")),
+                ];
+                children.extend(stage_node);
+                formatted_plan = FormatTreeNode::with_children(
+                    "InsertPlan (StreamingWithFileFormat):".to_string(),
+                    children,
+                )
+                .format_pretty()?;
+            }
+            InsertInputSource::StreamingWithFormat(_, _, _) => {
+                // used in clickhouse handler only; will discard soon
+                formatted_plan = FormatTreeNode::with_children(
+                    "InsertPlan (StreamingWithFormat):".to_string(),
+                    children,
+                )
+                .format_pretty()?;
+            }
+        };
+
+        let line_split_result: Vec<&str> = formatted_plan.lines().collect();
+        let formatted_plan = StringType::from_data(line_split_result);
+        result.push(DataBlock::new_from_columns(vec![formatted_plan]));
+        Ok(vec![DataBlock::concat(&result)?])
+    }
 }
 
 impl std::fmt::Debug for Insert {
diff --git a/tests/sqllogictests/suites/mode/standalone/explain/delete.test b/tests/sqllogictests/suites/mode/standalone/explain/delete.test
@@ -16,6 +16,13 @@ insert into t1 values(1, 2), (2, 3), (3, 4), (8, 9);
 statement ok
 insert into t2 values(2, 3), (3, 4);
 
+query T
+explain delete from t1
+----
+DeletePlan:
+└── table: default.default.t1
+
+
 query T
 explain delete from t1 where a in (select a from t2);
 ----
diff --git a/tests/sqllogictests/suites/mode/standalone/explain/explain.test b/tests/sqllogictests/suites/mode/standalone/explain/explain.test
@@ -1703,16 +1703,17 @@ Filter
 query T
 explain insert into t2 select * from t1;
 ----
-Insert
-TableScan
-├── table: default.default.t1
-├── output columns: [a (#0), b (#1), c (#2)]
-├── read rows: 0
-├── read size: 0
-├── partitions total: 0
-├── partitions scanned: 0
-├── push downs: [filters: [], limit: NONE]
-└── estimated rows: 0.00
+InsertPlan (subquery):
+├── table: default.default.t2
+├── inserted columns: [t2.a (#0),t2.b (#1),t2.c (#2)]
+├── overwrite: false
+└── EvalScalar
+    ├── scalars: [t1.a (#0) AS (#0), t1.b (#1) AS (#1), t1.c (#2) AS (#2)]
+    └── Scan
+        ├── table: default.t1
+        ├── filters: []
+        ├── order by: []
+        └── limit: NONE
 
 
 statement ok
diff --git a/tests/sqllogictests/suites/mode/standalone/explain/insert.test b/tests/sqllogictests/suites/mode/standalone/explain/insert.test
