Merge branch 'main' into feat-list-func

Author: mergify[bot]

Cargo.lock

@@ -692,6 +692,7 @@ impl<'ast> Visitor<'ast> for AstFormatVisitor {
             ExplainKind::Raw => "Raw",
             ExplainKind::Plan => "Plan",
             ExplainKind::Memo(_) => "Memo",
+            ExplainKind::JOIN => "JOIN",
             ExplainKind::AnalyzePlan => "Analyze",
         });
         let format_ctx = AstFormatContext::with_children(name, 1);

src/query/ast/src/ast/format/ast_format.rs

@@ -25,6 +25,8 @@ pub enum ExplainKind {
     Raw,
     Plan,
 
+    JOIN,
+
     // Explain analyze plan
     AnalyzePlan,
 }

src/query/ast/src/ast/statements/explain.rs

@@ -253,7 +253,8 @@ impl Display for Statement {
                     ExplainKind::Raw => write!(f, " RAW")?,
                     ExplainKind::Plan => (),
                     ExplainKind::AnalyzePlan => write!(f, " ANALYZE")?,
-                    ExplainKind::Memo(_) => write!(f, "MEMO")?,
+                    ExplainKind::JOIN => write!(f, " JOIN")?,
+                    ExplainKind::Memo(_) => write!(f, " MEMO")?,
                 }
                 write!(f, " {query}")?;
             }

src/query/ast/src/ast/statements/statement.rs

@@ -56,7 +56,7 @@ pub enum CreateDatabaseOption {
 pub fn statement(i: Input) -> IResult<StatementMsg> {
     let explain = map_res(
         rule! {
-            EXPLAIN ~ ( AST | SYNTAX | PIPELINE | GRAPH | FRAGMENTS | RAW | MEMO )? ~ #statement
+            EXPLAIN ~ ( AST | SYNTAX | PIPELINE | JOIN | GRAPH | FRAGMENTS | RAW | MEMO )? ~ #statement
         },
         |(_, opt_kind, statement)| {
             Ok(Statement::Explain {
@@ -72,6 +72,7 @@ pub fn statement(i: Input) -> IResult<StatementMsg> {
                         ExplainKind::Syntax(pretty_stmt)
                     }
                     Some(TokenKind::PIPELINE) => ExplainKind::Pipeline,
+                    Some(TokenKind::JOIN) => ExplainKind::JOIN,
                     Some(TokenKind::GRAPH) => ExplainKind::Graph,
                     Some(TokenKind::FRAGMENTS) => ExplainKind::Fragments,
                     Some(TokenKind::RAW) => ExplainKind::Raw,

src/query/ast/src/parser/statement.rs

@@ -230,6 +230,15 @@ where T: Number + AsPrimitive<f64>
         let state = place.get::<QuantileState>();
         state.merge_result(builder, self.levels.clone())
     }
+
+    fn need_manual_drop_state(&self) -> bool {
+        true
+    }
+
+    unsafe fn drop_state(&self, place: StateAddr) {
+        let state = place.get::<QuantileState>();
+        std::ptr::drop_in_place(state);
+    }
 }
 
 impl<T> AggregateQuantileContFunction<T>

src/query/functions/src/aggregates/aggregate_quantile_cont.rs

@@ -46,6 +46,7 @@ common-arrow = { path = "../../common/arrow" }
 common-ast = { path = "../ast" }
 common-auth = { path = "../../common/auth" }
 common-base = { path = "../../common/base" }
+common-cache = { path = "../../common/cache" }
 common-catalog = { path = "../catalog" }
 common-compress = { path = "../../common/compress" }
 common-config = { path = "../config" }
@@ -162,6 +163,7 @@ num = "0.4.0"
 p256 = "0.13"
 pretty_assertions = "1.3.0"
 reqwest = { workspace = true }
+sysinfo = "0.28.3"
 temp-env = "0.3.0"
 tempfile = "3.4.0"
 toml = { version = "0.7.3", default-features = false }

src/query/service/Cargo.toml

@@ -82,6 +82,28 @@ impl Interpreter for ExplainInterpreter {
                 _ => self.explain_plan(&self.plan)?,
             },
 
+            ExplainKind::JOIN => match &self.plan {
+                Plan::Query {
+                    s_expr, metadata, ..
+                } => {
+                    let ctx = self.ctx.clone();
+                    let settings = ctx.get_settings();
+
+                    let enable_distributed_eval_index =
+                        settings.get_enable_distributed_eval_index()?;
+                    settings.set_enable_distributed_eval_index(false)?;
+                    scopeguard::defer! {
+                        let _ = settings.set_enable_distributed_eval_index(enable_distributed_eval_index);
+                    }
+                    let mut builder = PhysicalPlanBuilder::new(metadata.clone(), ctx);
+                    let plan = builder.build(s_expr).await?;
+                    self.explain_join_order(&plan, metadata)?
+                }
+                _ => Err(ErrorCode::Unimplemented(
+                    "Unsupported EXPLAIN JOIN statement",
+                ))?,
+            },
+
             ExplainKind::AnalyzePlan => match &self.plan {
                 Plan::Query {
                     s_expr,
@@ -185,6 +207,17 @@ impl ExplainInterpreter {
         Ok(vec![DataBlock::new_from_columns(vec![formatted_plan])])
     }
 
+    pub fn explain_join_order(
+        &self,
+        plan: &PhysicalPlan,
+        metadata: &MetadataRef,
+    ) -> Result<Vec<DataBlock>> {
+        let result = plan.format_join(metadata)?.format_pretty()?;
+        let line_split_result: Vec<&str> = result.lines().collect();
+        let formatted_plan = StringType::from_data(line_split_result);
+        Ok(vec![DataBlock::new_from_columns(vec![formatted_plan])])
+    }
+
     pub async fn explain_pipeline(
         &self,
         s_expr: SExpr,

src/query/service/src/interpreters/interpreter_explain.rs

@@ -12,6 +12,7 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+use common_arrow::parquet::metadata::ThriftFileMetaData;
 use common_exception::Result;
 use common_expression::DataBlock;
 use common_expression::FunctionContext;
@@ -56,13 +57,13 @@ impl<'a> BlockWriter<'a> {
         block: DataBlock,
         col_stats: StatisticsOfColumns,
         cluster_stats: Option<ClusterStatistics>,
-    ) -> Result<BlockMeta> {
+    ) -> Result<(BlockMeta, Option<ThriftFileMetaData>)> {
         let (location, block_id) = self.location_generator.gen_block_location();
 
         let data_accessor = &self.data_accessor;
         let row_count = block.num_rows() as u64;
         let block_size = block.memory_size() as u64;
-        let (bloom_filter_index_size, bloom_filter_index_location) = self
+        let (bloom_filter_index_size, bloom_filter_index_location, meta) = self
             .build_block_index(data_accessor, schema.clone(), &block, block_id)
             .await?;
 
@@ -88,7 +89,7 @@ impl<'a> BlockWriter<'a> {
             bloom_filter_index_size,
             Compression::Lz4Raw,
         );
-        Ok(block_meta)
+        Ok((block_meta, meta))
     }
 
     pub async fn build_block_index(
@@ -97,7 +98,7 @@ impl<'a> BlockWriter<'a> {
         schema: TableSchemaRef,
         block: &DataBlock,
         block_id: Uuid,
-    ) -> Result<(u64, Option<Location>)> {
+    ) -> Result<(u64, Option<Location>, Option<ThriftFileMetaData>)> {
         let location = self
             .location_generator
             .block_bloom_index_location(&block_id);
@@ -109,18 +110,16 @@ impl<'a> BlockWriter<'a> {
             let filter_schema = bloom_index.filter_schema;
             let mut data = Vec::with_capacity(DEFAULT_BLOCK_INDEX_BUFFER_SIZE);
             let index_block_schema = &filter_schema;
-            let (size, _) = blocks_to_parquet(
+            let (size, meta) = blocks_to_parquet(
                 index_block_schema,
                 vec![index_block],
                 &mut data,
                 TableCompression::None,
             )?;
-
             data_accessor.write(&location.0, data).await?;
-
-            Ok((size, Some(location)))
+            Ok((size, Some(location), Some(meta)))
         } else {
-            Ok((0u64, None))
+            Ok((0u64, None, None))
         }
     }
 }