From 3d35d31655edd271dcdc1b6e3a562edbcb8c17fd Mon Sep 17 00:00:00 2001 From: Igor Nikonov Date: Tue, 25 Jun 2024 16:11:04 +0000 Subject: [PATCH] More cleanup --- src/Processors/QueryPlan/ParallelReplicasLocalPlan.cpp | 10 +++++++--- src/Processors/QueryPlan/ReadFromMergeTree.cpp | 7 ++++--- src/Processors/QueryPlan/ReadFromMergeTree.h | 4 ++-- 3 files changed, 13 insertions(+), 8 deletions(-) diff --git a/src/Processors/QueryPlan/ParallelReplicasLocalPlan.cpp b/src/Processors/QueryPlan/ParallelReplicasLocalPlan.cpp index 67c10985408..5f48a12072b 100644 --- a/src/Processors/QueryPlan/ParallelReplicasLocalPlan.cpp +++ b/src/Processors/QueryPlan/ParallelReplicasLocalPlan.cpp @@ -69,9 +69,13 @@ std::unique_ptr createLocalPlanForParallelReplicas( chassert(reading); - ReadFromMergeTree * analyzed_merge_tree = nullptr; + ReadFromMergeTree::AnalysisResultPtr analyzed_result_ptr; if (analyzed_read_from_merge_tree.get()) - analyzed_merge_tree = typeid_cast(analyzed_read_from_merge_tree.get()); + { + auto * analyzed_merge_tree = typeid_cast(analyzed_read_from_merge_tree.get()); + if (analyzed_merge_tree) + analyzed_result_ptr = analyzed_merge_tree->getAnalyzedResult(); + } MergeTreeAllRangesCallback all_ranges_cb = [coordinator](InitialAllRangesAnnouncement announcement) { coordinator->handleInitialAllRangesAnnouncement(announcement); }; @@ -80,7 +84,7 @@ std::unique_ptr createLocalPlanForParallelReplicas( { return coordinator->handleRequest(std::move(req)); }; auto read_from_merge_tree_parallel_replicas - = reading->createLocalParallelReplicasReadingStep(analyzed_merge_tree, all_ranges_cb, read_task_cb); + = reading->createLocalParallelReplicasReadingStep(analyzed_result_ptr, all_ranges_cb, read_task_cb); node->step = std::move(read_from_merge_tree_parallel_replicas); addConvertingActions(*query_plan, header, /*has_missing_objects=*/false); diff --git a/src/Processors/QueryPlan/ReadFromMergeTree.cpp b/src/Processors/QueryPlan/ReadFromMergeTree.cpp index 0ed06639f8c..95c337e2150 100644 --- a/src/Processors/QueryPlan/ReadFromMergeTree.cpp +++ b/src/Processors/QueryPlan/ReadFromMergeTree.cpp @@ -337,11 +337,12 @@ ReadFromMergeTree::ReadFromMergeTree( } std::unique_ptr ReadFromMergeTree::createLocalParallelReplicasReadingStep( - const ReadFromMergeTree * analyzed_merge_tree, + AnalysisResultPtr analyzed_result_ptr_, std::optional all_ranges_callback_, std::optional read_task_callback_) { const auto number_of_local_replica = 0; + const bool enable_parallel_reading = true; return std::make_unique( prepared_parts, alter_conversions_for_parts, @@ -354,8 +355,8 @@ std::unique_ptr ReadFromMergeTree::createLocalParallelReplica requested_num_streams, max_block_numbers_to_read, log, - (analyzed_merge_tree ? analyzed_merge_tree->analyzed_result_ptr : nullptr), - true, + analyzed_result_ptr_, + enable_parallel_reading, all_ranges_callback_, read_task_callback_, number_of_local_replica); diff --git a/src/Processors/QueryPlan/ReadFromMergeTree.h b/src/Processors/QueryPlan/ReadFromMergeTree.h index 99205c1de1d..c9c6030d207 100644 --- a/src/Processors/QueryPlan/ReadFromMergeTree.h +++ b/src/Processors/QueryPlan/ReadFromMergeTree.h @@ -127,9 +127,9 @@ public: std::optional number_of_current_replica_ = std::nullopt); std::unique_ptr createLocalParallelReplicasReadingStep( - const ReadFromMergeTree * analyzed_merge_tree, + AnalysisResultPtr analyzed_result_ptr_, std::optional all_ranges_callback_, - std::optional read_task_callback); + std::optional read_task_callback_); static constexpr auto name = "ReadFromMergeTree"; String getName() const override { return name; }