Fix crash in EXPLAIN PIPELINE for Merge over Distributed

src/Interpreters/IInterpreterUnionOrSelectQuery.cpp

@@ -18,9 +18,12 @@ namespace DB
 QueryPipelineBuilder IInterpreterUnionOrSelectQuery::buildQueryPipeline()
 {
     QueryPlan query_plan;
+    return buildQueryPipeline(query_plan);
+}
 
+QueryPipelineBuilder IInterpreterUnionOrSelectQuery::buildQueryPipeline(QueryPlan & query_plan)
+{
     buildQueryPlan(query_plan);
-
     return std::move(*query_plan.buildQueryPipeline(
         QueryPlanOptimizationSettings::fromContext(context), BuildQueryPipelineSettings::fromContext(context)));
 }

src/Interpreters/IInterpreterUnionOrSelectQuery.h

@@ -35,6 +35,7 @@ class IInterpreterUnionOrSelectQuery : public IInterpreter
 
     virtual void buildQueryPlan(QueryPlan & query_plan) = 0;
     QueryPipelineBuilder buildQueryPipeline();
+    QueryPipelineBuilder buildQueryPipeline(QueryPlan & query_plan);
 
     virtual void ignoreWithTotals() = 0;
 

src/Interpreters/InterpreterSelectQueryAnalyzer.h

@@ -72,6 +72,7 @@ class InterpreterSelectQueryAnalyzer : public IInterpreter
     void setProperClientInfo(size_t replica_number, size_t count_participating_replicas);
 
     const Planner & getPlanner() const { return planner; }
+    Planner & getPlanner() { return planner; }
 
 private:
     ASTPtr query;

src/Storages/StorageMerge.cpp

@@ -649,14 +649,13 @@ QueryPipelineBuilderPtr ReadFromMerge::createSources(
         QueryProcessingStage::Complete,
         storage_snapshot,
         modified_query_info);
+
     if (processed_stage <= storage_stage || (allow_experimental_analyzer && processed_stage == QueryProcessingStage::FetchColumns))
     {
         /// If there are only virtual columns in query, you must request at least one other column.
         if (real_column_names.empty())
             real_column_names.push_back(ExpressionActions::getSmallestColumn(storage_snapshot->metadata->getColumns().getAllPhysical()).name);
 
-        /// Steps for reading from child tables should have the same lifetime as the current step
-        /// because `builder` can have references to them (mainly for EXPLAIN PIPELINE).
         QueryPlan & plan = child_plans.emplace_back();
 
         StorageView * view = dynamic_cast<StorageView *>(storage.get());
@@ -709,12 +708,15 @@ QueryPipelineBuilderPtr ReadFromMerge::createSources(
         modified_context->setSetting("max_threads", streams_num);
         modified_context->setSetting("max_streams_to_max_threads_ratio", 1);
 
+        QueryPlan & plan = child_plans.emplace_back();
+
         if (allow_experimental_analyzer)
         {
             InterpreterSelectQueryAnalyzer interpreter(modified_query_info.query_tree,
                 modified_context,
                 SelectQueryOptions(processed_stage).ignoreProjections());
             builder = std::make_unique<QueryPipelineBuilder>(interpreter.buildQueryPipeline());
+            plan = std::move(interpreter.getPlanner()).extractQueryPlan();
         }
         else
         {
@@ -723,7 +725,7 @@ QueryPipelineBuilderPtr ReadFromMerge::createSources(
             InterpreterSelectQuery interpreter{modified_query_info.query,
                 modified_context,
                 SelectQueryOptions(processed_stage).ignoreProjections()};
-            builder = std::make_unique<QueryPipelineBuilder>(interpreter.buildQueryPipeline());
+            builder = std::make_unique<QueryPipelineBuilder>(interpreter.buildQueryPipeline(plan));
         }
 
         /** Materialization is needed, since from distributed storage the constants come materialized.

src/Storages/StorageMerge.h

@@ -160,7 +160,7 @@ class ReadFromMerge final : public SourceStepWithFilter
     StorageSnapshotPtr merge_storage_snapshot;
 
     /// Store read plan for each child table.
-    /// It's needed to guarantee lifetime for child steps to be the same as for this step.
+    /// It's needed to guarantee lifetime for child steps to be the same as for this step (mainly for EXPLAIN PIPELINE).
     std::vector<QueryPlan> child_plans;
 
     SelectQueryInfo query_info;

tests/queries/0_stateless/02704_storage_merge_explain_graph_crash.sql

@@ -1,9 +1,16 @@
 DROP TABLE IF EXISTS foo;
+DROP TABLE IF EXISTS foo2;
+DROP TABLE IF EXISTS foo2_dist;
 DROP TABLE IF EXISTS merge1;
 
 CREATE TABLE foo (`Id` Int32, `Val` Int32) ENGINE = MergeTree ORDER BY Id;
 INSERT INTO foo SELECT number, number FROM numbers(100);
 
-CREATE TABLE merge1 AS foo ENGINE = Merge(currentDatabase(), '^foo');
+CREATE TABLE foo2 (`Id` Int32, `Val` Int32) ENGINE = MergeTree ORDER BY Id;
+INSERT INTO foo2 SELECT number, number FROM numbers(100);
+CREATE TABLE foo2_dist (`Id` UInt32, `Val` String) ENGINE = Distributed(test_shard_localhost, currentDatabase(), foo2);
+
+CREATE TABLE merge1 AS foo ENGINE = Merge(currentDatabase(), '^(foo|foo2_dist)$');
 
 EXPLAIN PIPELINE graph = 1, compact = 1 SELECT * FROM merge1 FORMAT Null;
+EXPLAIN PIPELINE graph = 1, compact = 1 SELECT * FROM merge1 FORMAT Null SETTINGS allow_experimental_analyzer=1;