From d0658d72530adb05b3c199d2a3584e2333455476 Mon Sep 17 00:00:00 2001 From: Alexey Milovidov Date: Tue, 3 Nov 2026 15:27:28 +0000 Subject: [PATCH 1/2] Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges Fix wrong `ORDER BY` results and `pushed bucket twice` with offset based parallel replicas Conflicts committed as-is; the next commit resolves them: - src/Processors/QueryPlan/Optimizations/Optimizations.h (both modified) - src/Processors/QueryPlan/Optimizations/optimizeTree.cpp (both modified) - src/Processors/QueryPlan/ReadFromRemote.cpp (both modified) - src/Processors/QueryPlan/ReadFromRemote.h (both modified) --- src/Core/Settings.cpp | 2 + .../QueryPlan/MergingAggregatedStep.h | 1 + .../QueryPlan/Optimizations/Optimizations.h | 13 +++ .../allowReadingShardsFromSeveralReplicas.cpp | 84 ++++++++++++++ .../QueryPlan/Optimizations/optimizeTree.cpp | 13 +++ src/Processors/QueryPlan/ReadFromRemote.cpp | 22 ++++ src/Processors/QueryPlan/ReadFromRemote.h | 6 + .../test_parallel_replicas_custom_key/test.py | 2 + ...parallel_replicas_ordered_merges.reference | 51 +++++++++ ...ffset_parallel_replicas_ordered_merges.sql | 104 ++++++++++++++++++ 10 files changed, 298 insertions(+) create mode 100644 src/Processors/QueryPlan/Optimizations/allowReadingShardsFromSeveralReplicas.cpp create mode 100644 tests/queries/0_stateless/05483_offset_parallel_replicas_ordered_merges.reference create mode 100644 tests/queries/0_stateless/05483_offset_parallel_replicas_ordered_merges.sql diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index 881f3b59613f2..4150173e8bb28 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -8228,6 +8228,8 @@ Possible values: This options will produce different results depending on the settings used. +With `parallel_replicas_mode` `sampling_key`, or a custom key mode on a cluster with more than one shard, each replica returns its own part of the result of the shard. The replicas of a shard are used together only when the initiator aggregates or fully sorts these parts itself (for example an aggregation without `GROUP BY` keys). Otherwise, for example for `GROUP BY` with `distributed_aggregation_memory_efficient` or for `ORDER BY`, the shard is read from one replica. + ### Parallel processing using `SAMPLE` key A query may be processed faster if it is executed on several servers in parallel. But the query performance may degrade in the following cases: diff --git a/src/Processors/QueryPlan/MergingAggregatedStep.h b/src/Processors/QueryPlan/MergingAggregatedStep.h index 09e629676789e..6d79b91ae5234 100644 --- a/src/Processors/QueryPlan/MergingAggregatedStep.h +++ b/src/Processors/QueryPlan/MergingAggregatedStep.h @@ -38,6 +38,7 @@ class MergingAggregatedStep : public ITransformingStep const SortDescription & getGroupBySortDescription() const { return group_by_sort_description; } bool memoryBoundMergingWillBeUsed() const; + bool isMemoryEfficient() const { return memory_efficient_aggregation; } bool isGroupingSets() const { return !grouping_sets_params.empty(); } const auto & getGroupingSetsParamsList() const { return grouping_sets_params; } diff --git a/src/Processors/QueryPlan/Optimizations/Optimizations.h b/src/Processors/QueryPlan/Optimizations/Optimizations.h index e54b0544287e9..a9c58f87e6cd1 100644 --- a/src/Processors/QueryPlan/Optimizations/Optimizations.h +++ b/src/Processors/QueryPlan/Optimizations/Optimizations.h @@ -330,8 +330,21 @@ void applyOrder(const QueryPlanOptimizationSettings & optimization_settings, Que /// carry the same key value). void applyStreamDisjointness(const QueryPlanOptimizationSettings & optimization_settings, QueryPlan::Node & root); +<<<<<<< HEAD /// Returns the name of used projection or nullopt if no projection is used. std::optional optimizeUseAggregateProjections( +======= +/// Lets an offset based parallel replicas read use several replicas of a shard only if no consumer needs stream order. +void allowReadingShardsFromSeveralReplicas(QueryPlan::Node & root); + +struct UseProjectionsResult +{ + std::optional applied_projection; + std::unordered_map projection_reject_reasons; +}; + +UseProjectionsResult optimizeUseAggregateProjections( +>>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) QueryPlan::Node & node, QueryPlan::Nodes & nodes, const QueryPlanOptimizationSettings & optimization_settings); diff --git a/src/Processors/QueryPlan/Optimizations/allowReadingShardsFromSeveralReplicas.cpp b/src/Processors/QueryPlan/Optimizations/allowReadingShardsFromSeveralReplicas.cpp new file mode 100644 index 0000000000000..7f288d392db59 --- /dev/null +++ b/src/Processors/QueryPlan/Optimizations/allowReadingShardsFromSeveralReplicas.cpp @@ -0,0 +1,84 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +namespace DB::QueryPlanOptimizations +{ + +namespace +{ + +enum class StreamOrderUse : uint8_t +{ + /// The step keeps its input streams apart and in order, so its own consumer decides. + PassedOn, + Ignored, + Required, +}; + +StreamOrderUse getStreamOrderUse(const IQueryPlanStep & step) +{ + if (typeid_cast(&step) || typeid_cast(&step) || typeid_cast(&step)) + return StreamOrderUse::PassedOn; + + if (const auto * merging_aggregated = typeid_cast(&step)) + { + /// Without keys the result is never two-level, so the memory efficient merge waits for every input. + if (!merging_aggregated->memoryBoundMergingWillBeUsed() + && (!merging_aggregated->isMemoryEfficient() || merging_aggregated->getParams().keys.empty())) + return StreamOrderUse::Ignored; + return StreamOrderUse::Required; + } + + if (const auto * aggregating = typeid_cast(&step)) + return aggregating->inOrder() ? StreamOrderUse::Required : StreamOrderUse::Ignored; + + if (const auto * sorting = typeid_cast(&step)) + return sorting->getType() == SortingStep::Type::Full ? StreamOrderUse::Ignored : StreamOrderUse::Required; + + return StreamOrderUse::Required; +} + +} + +void allowReadingShardsFromSeveralReplicas(QueryPlan::Node & root) +{ + std::unordered_map allowed; + Stack stack; + + traverseQueryPlan(stack, root, [&](QueryPlan::Node & node) + { + auto * reading = typeid_cast(node.step.get()); + if (!reading) + return; + + /// The consumer of the plan root is outside of the plan. + bool allow = false; + for (size_t i = stack.size() - 1; i > 0; --i) + { + const auto use = getStreamOrderUse(*stack[i - 1].node->step); + if (use == StreamOrderUse::PassedOn) + continue; + allow = use == StreamOrderUse::Ignored; + break; + } + + auto [it, inserted] = allowed.emplace(reading, allow); + if (!inserted) + it->second = it->second && allow; + }); + + for (const auto & [reading, allow] : allowed) + reading->allowReadingShardFromSeveralReplicas(allow); +} + +} diff --git a/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp b/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp index 9556da9e95500..f28b993352a66 100644 --- a/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp +++ b/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp @@ -813,6 +813,19 @@ void optimizeTreeSecondPass( { traverseQueryPlan(stack, root, [&](auto & frame_node) { tryOptimizeGroupByTopK(&frame_node, nodes, extra_settings); }); } +<<<<<<< HEAD +======= + + /// Runs behind every rewrite of the HAVING filter it reads, and behind the pass above, whose `top_k` it refuses. + if (optimization_settings.aggregation_having_prefilter) + { + traverseQueryPlan( + stack, root, [&](auto & frame_node) { tryPushHavingPrefilterIntoAggregation(&frame_node, nodes, extra_settings); }); + } + + /// Last: it reads the final consumers of every remote read. + allowReadingShardsFromSeveralReplicas(root); +>>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) } void addStepsToBuildSets( diff --git a/src/Processors/QueryPlan/ReadFromRemote.cpp b/src/Processors/QueryPlan/ReadFromRemote.cpp index 338b072fc01c0..69dec01617b46 100644 --- a/src/Processors/QueryPlan/ReadFromRemote.cpp +++ b/src/Processors/QueryPlan/ReadFromRemote.cpp @@ -861,6 +861,7 @@ void ReadFromRemote::addPipe( // Several connections to a shard are correct only when every replica reads its own part of the data, // which is the case only for the offset based modes (`SAMPLING_KEY`, `CUSTOM_KEY_SAMPLING`, // `CUSTOM_KEY_RANGE`), where the query sent to a replica carries the corresponding filter. + // The replicas then share one port, so a read whose consumer needs stream order uses one replica. // // In every other case a replica executes the whole query, so there should be a single connection // to a shard, otherwise the result of the shard is multiplied by the number of the connections: @@ -868,9 +869,15 @@ void ReadFromRemote::addPipe( // connect to instantiates the coordinator which manages the reading on the whole shard and // returns the result of the shard, so several connections mean several coordinators; // * with parallel replicas disabled, or not applicable for any other reason (e.g. by +<<<<<<< HEAD // `automatic_parallel_replicas_mode` or `parallel_replicas_only_with_analyzer`), a replica // just executes the query over all of its data. if (context->canUseOffsetParallelReplicas()) +======= + // `automatic_parallel_replicas_mode`), a replica just executes the query over all of its + // data. + if (context->canUseOffsetParallelReplicas() && allow_reading_shard_from_several_replicas) +>>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) remote_query_executor->setPoolMode(PoolMode::GET_MANY); else remote_query_executor->setPoolMode(PoolMode::GET_ONE); @@ -903,7 +910,22 @@ Pipes ReadFromRemote::addPipes(const ClusterProxy::SelectStreamFactory::Shards & void ReadFromRemote::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) { +<<<<<<< HEAD Pipes pipes = addPipes(shards, output_header); +======= + if (context->canUseOffsetParallelReplicas() && !allow_reading_shard_from_several_replicas) + LOG_DEBUG(log, "Reading every shard from one replica, the consumers of the read may depend on the order of its streams"); + + Pipes pipes = addPipes(shards, output_header, unavailable_shard_tracker); + + /// One pipe per executor, so the built pipes are this query's complete unit count. + if (unavailable_shard_tracker) + { + if (pipes.size() > shards.size()) + unavailable_shard_tracker->registerExtraUnits(pipes.size() - shards.size()); + unavailable_shard_tracker->seal(); + } +>>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) auto pipe = Pipe::unitePipes(std::move(pipes)); diff --git a/src/Processors/QueryPlan/ReadFromRemote.h b/src/Processors/QueryPlan/ReadFromRemote.h index 1be0509f9ac8f..c859becca1911 100644 --- a/src/Processors/QueryPlan/ReadFromRemote.h +++ b/src/Processors/QueryPlan/ReadFromRemote.h @@ -55,6 +55,8 @@ class ReadFromRemote final : public SourceStepWithFilterBase bool hasSerializedPlan() const; + void allowReadingShardFromSeveralReplicas(bool allow) { allow_reading_shard_from_several_replicas = allow; } + private: ClusterProxy::SelectStreamFactory::Shards shards; QueryProcessingStage::Enum stage; @@ -70,7 +72,11 @@ class ReadFromRemote final : public SourceStepWithFilterBase const String cluster_name; UnavailableShardTrackerPtr unavailable_shard_tracker; std::optional priority_func_factory; +<<<<<<< HEAD bool is_remote_function = false; +======= + bool allow_reading_shard_from_several_replicas = false; +>>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) Pipes addPipes(const ClusterProxy::SelectStreamFactory::Shards & used_shards, const SharedHeader & out_header); diff --git a/tests/integration/test_parallel_replicas_custom_key/test.py b/tests/integration/test_parallel_replicas_custom_key/test.py index 4bf3fe51dba6c..59f014ef4cc72 100644 --- a/tests/integration/test_parallel_replicas_custom_key/test.py +++ b/tests/integration/test_parallel_replicas_custom_key/test.py @@ -89,6 +89,8 @@ def test_parallel_replicas_custom_key_distributed( "parallel_replicas_mode": parallel_replicas_mode, "prefer_localhost_replica": 0, "serialize_query_plan": 0, + # with a memory efficient merge every shard is read from one replica + "distributed_aggregation_memory_efficient": 0, }, ) == expected_result diff --git a/tests/queries/0_stateless/05483_offset_parallel_replicas_ordered_merges.reference b/tests/queries/0_stateless/05483_offset_parallel_replicas_ordered_merges.reference new file mode 100644 index 0000000000000..ff72383de13aa --- /dev/null +++ b/tests/queries/0_stateless/05483_offset_parallel_replicas_ordered_merges.reference @@ -0,0 +1,51 @@ +30000 60000 2 2 +30000 60000 2 2 +30000 30000 1 1 +0 +1 +2 +3 +4 +5 +6 +7 +8 +9 +0 +0 +1 +1 +2 +2 +3 +3 +4 +4 +30000 +30000 +30000 30000 +0 +1 +2 +3 +4 +5 +6 +7 +8 +9 +449985000 +15000 +224985000 +899970000 +offset_pr_aggregating 3 +offset_pr_count 3 +offset_pr_expression 3 +offset_pr_filter 3 +offset_pr_full_sort 3 +offset_pr_group_by 1 +offset_pr_limit 1 +offset_pr_memory_bound 1 +offset_pr_plain_merge 3 +offset_pr_root 1 +offset_pr_union 6 diff --git a/tests/queries/0_stateless/05483_offset_parallel_replicas_ordered_merges.sql b/tests/queries/0_stateless/05483_offset_parallel_replicas_ordered_merges.sql new file mode 100644 index 0000000000000..b574f58176505 --- /dev/null +++ b/tests/queries/0_stateless/05483_offset_parallel_replicas_ordered_merges.sql @@ -0,0 +1,104 @@ +-- Offset based parallel replicas: every replica of a shard returns its own part of the result, which the +-- initiator must still merge correctly when it relies on the order of the streams. + +DROP TABLE IF EXISTS t_offset_pr_key; +DROP TABLE IF EXISTS t_offset_pr_sample; + +CREATE TABLE t_offset_pr_key (k UInt64) ENGINE = MergeTree ORDER BY k; +INSERT INTO t_offset_pr_key SELECT number FROM numbers(30000); + +CREATE TABLE t_offset_pr_sample (k UInt64) ENGINE = MergeTree ORDER BY intHash32(k) SAMPLE BY intHash32(k); +INSERT INTO t_offset_pr_sample SELECT number FROM numbers(30000); + +SET enable_parallel_replicas = 1, max_parallel_replicas = 3, automatic_parallel_replicas_mode = 0, + parallel_replicas_for_non_replicated_merge_tree = 1, prefer_localhost_replica = 0; +-- Two-level results, merged bucket by bucket. +SET distributed_aggregation_memory_efficient = 1, group_by_two_level_threshold = 1, group_by_two_level_threshold_bytes = 1, + enable_producing_buckets_out_of_order_in_aggregation = 0, optimize_aggregation_in_order = 0; + +SELECT count(), sum(c), min(c), max(c) +FROM (SELECT k, count() AS c FROM cluster('test_cluster_two_shard_three_replicas_localhost', currentDatabase(), t_offset_pr_key) GROUP BY k) +SETTINGS parallel_replicas_mode = 'custom_key_sampling', parallel_replicas_custom_key = 'k'; + +SELECT count(), sum(c), min(c), max(c) +FROM (SELECT k, count() AS c FROM cluster('test_cluster_two_shard_three_replicas_localhost', currentDatabase(), t_offset_pr_key) GROUP BY k) +SETTINGS parallel_replicas_mode = 'custom_key_range', parallel_replicas_custom_key = 'cityHash64(k)'; + +SELECT count(), sum(c), min(c), max(c) +FROM (SELECT k, count() AS c FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample) GROUP BY k) +SETTINGS parallel_replicas_mode = 'sampling_key'; + +SELECT k FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample) ORDER BY k LIMIT 10 +SETTINGS parallel_replicas_mode = 'sampling_key'; + +SELECT k FROM cluster('test_cluster_two_shard_three_replicas_localhost', currentDatabase(), t_offset_pr_key) ORDER BY k LIMIT 10 +SETTINGS parallel_replicas_mode = 'custom_key_sampling', parallel_replicas_custom_key = 'k'; + +-- Without GROUP BY keys the parts are merged in any order, so the replicas still share the shard. +SELECT count() FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample) +SETTINGS parallel_replicas_mode = 'sampling_key', use_hedged_requests = 0, log_queries = 1, log_comment = 'offset_pr_count'; + +SELECT count() FROM (SELECT k FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample) GROUP BY k) +SETTINGS parallel_replicas_mode = 'sampling_key', use_hedged_requests = 0, log_queries = 1, log_comment = 'offset_pr_group_by'; + +-- Merged all at once instead of bucket by bucket, so the replicas still share the shard. +SELECT count(), sum(c) FROM (SELECT k, count() AS c FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample) GROUP BY k) +SETTINGS parallel_replicas_mode = 'sampling_key', distributed_aggregation_memory_efficient = 0, + enable_memory_bound_merging_of_aggregation_results = 0, use_hedged_requests = 0, log_queries = 1, log_comment = 'offset_pr_plain_merge'; + +-- With a local shard and aggregation in order the states are merged in order, so the remote shard is read from one replica. +SELECT k, count() FROM cluster('test_cluster_two_shard_three_replicas_localhost', currentDatabase(), t_offset_pr_key) GROUP BY k +SETTINGS parallel_replicas_mode = 'custom_key_sampling', parallel_replicas_custom_key = 'k', prefer_localhost_replica = 1, + optimize_aggregation_in_order = 1, distributed_aggregation_memory_efficient = 0, enable_memory_bound_merging_of_aggregation_results = 1, + use_hedged_requests = 0, log_queries = 1, log_comment = 'offset_pr_memory_bound' +FORMAT Null; + +-- A full sort and an aggregation that is not in order take the parts in any order too. +SELECT k FROM (SELECT k FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample)) ORDER BY k LIMIT 10 +SETTINGS parallel_replicas_mode = 'sampling_key', use_hedged_requests = 0, log_queries = 1, log_comment = 'offset_pr_full_sort'; + +SELECT sum(k) FROM (SELECT k FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample)) +SETTINGS parallel_replicas_mode = 'sampling_key', use_hedged_requests = 0, log_queries = 1, log_comment = 'offset_pr_aggregating'; + +-- Expressions, filters and unions between the read and such a consumer are looked through. +SELECT sum(k % 2) FROM (SELECT k FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample)) +SETTINGS parallel_replicas_mode = 'sampling_key', use_hedged_requests = 0, log_queries = 1, log_comment = 'offset_pr_expression'; + +SELECT sum(k) FROM (SELECT k FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample)) WHERE k % 2 = 0 +SETTINGS parallel_replicas_mode = 'sampling_key', use_hedged_requests = 0, log_queries = 1, log_comment = 'offset_pr_filter'; + +SELECT sum(k) FROM +( + SELECT k FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample) + UNION ALL + SELECT k FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample) +) +SETTINGS parallel_replicas_mode = 'sampling_key', use_hedged_requests = 0, log_queries = 1, log_comment = 'offset_pr_union'; + +-- Any other consumer, and the plan root, read a shard from one replica. +-- The limit is above the row count, so no secondary query is cancelled. +SELECT k FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample) LIMIT 100000 +SETTINGS parallel_replicas_mode = 'sampling_key', use_hedged_requests = 0, log_queries = 1, log_comment = 'offset_pr_limit' +FORMAT Null; + +SELECT k FROM remote('127.0.0.{1|2|3}', currentDatabase(), t_offset_pr_sample) +SETTINGS parallel_replicas_mode = 'sampling_key', use_hedged_requests = 0, log_queries = 1, log_comment = 'offset_pr_root' +FORMAT Null; + +SYSTEM FLUSH LOGS query_log; + +SELECT initial.log_comment, count() +FROM system.query_log AS secondary +INNER JOIN +( + SELECT query_id, log_comment FROM system.query_log + WHERE event_date >= yesterday() AND current_database = currentDatabase() AND is_initial_query AND type = 'QueryFinish' + AND log_comment IN ('offset_pr_count', 'offset_pr_group_by', 'offset_pr_plain_merge', 'offset_pr_full_sort', 'offset_pr_aggregating', 'offset_pr_expression', 'offset_pr_filter', 'offset_pr_union', 'offset_pr_limit', 'offset_pr_root', 'offset_pr_memory_bound') +) AS initial ON secondary.initial_query_id = initial.query_id +WHERE secondary.event_date >= yesterday() AND NOT secondary.is_initial_query AND secondary.type = 'QueryFinish' +GROUP BY initial.log_comment +ORDER BY initial.log_comment +SETTINGS enable_parallel_replicas = 0; + +DROP TABLE t_offset_pr_sample; +DROP TABLE t_offset_pr_key; From 462538532924950fdff2781dda0f969c9ec20af8 Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Thu, 5 Nov 2026 17:39:43 +0100 Subject: [PATCH 2/2] Resolve conflicts in backport of #128865 Kept the PR's new last optimization pass and the `allow_reading_shard_from_several_replicas` flag, and dropped master-only code that came in through the conflicts (`UseProjectionsResult`, the HAVING prefilter pass, the unavailable-shard tracker in `addPipes`). --- .../QueryPlan/Optimizations/Optimizations.h | 14 ++----------- .../QueryPlan/Optimizations/optimizeTree.cpp | 10 ---------- src/Processors/QueryPlan/ReadFromRemote.cpp | 20 +------------------ src/Processors/QueryPlan/ReadFromRemote.h | 3 --- 4 files changed, 3 insertions(+), 44 deletions(-) diff --git a/src/Processors/QueryPlan/Optimizations/Optimizations.h b/src/Processors/QueryPlan/Optimizations/Optimizations.h index a9c58f87e6cd1..f76b2ad5cd198 100644 --- a/src/Processors/QueryPlan/Optimizations/Optimizations.h +++ b/src/Processors/QueryPlan/Optimizations/Optimizations.h @@ -330,21 +330,11 @@ void applyOrder(const QueryPlanOptimizationSettings & optimization_settings, Que /// carry the same key value). void applyStreamDisjointness(const QueryPlanOptimizationSettings & optimization_settings, QueryPlan::Node & root); -<<<<<<< HEAD -/// Returns the name of used projection or nullopt if no projection is used. -std::optional optimizeUseAggregateProjections( -======= /// Lets an offset based parallel replicas read use several replicas of a shard only if no consumer needs stream order. void allowReadingShardsFromSeveralReplicas(QueryPlan::Node & root); -struct UseProjectionsResult -{ - std::optional applied_projection; - std::unordered_map projection_reject_reasons; -}; - -UseProjectionsResult optimizeUseAggregateProjections( ->>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) +/// Returns the name of used projection or nullopt if no projection is used. +std::optional optimizeUseAggregateProjections( QueryPlan::Node & node, QueryPlan::Nodes & nodes, const QueryPlanOptimizationSettings & optimization_settings); diff --git a/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp b/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp index f28b993352a66..75a9779cc99d1 100644 --- a/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp +++ b/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp @@ -813,19 +813,9 @@ void optimizeTreeSecondPass( { traverseQueryPlan(stack, root, [&](auto & frame_node) { tryOptimizeGroupByTopK(&frame_node, nodes, extra_settings); }); } -<<<<<<< HEAD -======= - - /// Runs behind every rewrite of the HAVING filter it reads, and behind the pass above, whose `top_k` it refuses. - if (optimization_settings.aggregation_having_prefilter) - { - traverseQueryPlan( - stack, root, [&](auto & frame_node) { tryPushHavingPrefilterIntoAggregation(&frame_node, nodes, extra_settings); }); - } /// Last: it reads the final consumers of every remote read. allowReadingShardsFromSeveralReplicas(root); ->>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) } void addStepsToBuildSets( diff --git a/src/Processors/QueryPlan/ReadFromRemote.cpp b/src/Processors/QueryPlan/ReadFromRemote.cpp index 69dec01617b46..a2e5897b3189b 100644 --- a/src/Processors/QueryPlan/ReadFromRemote.cpp +++ b/src/Processors/QueryPlan/ReadFromRemote.cpp @@ -869,15 +869,9 @@ void ReadFromRemote::addPipe( // connect to instantiates the coordinator which manages the reading on the whole shard and // returns the result of the shard, so several connections mean several coordinators; // * with parallel replicas disabled, or not applicable for any other reason (e.g. by -<<<<<<< HEAD // `automatic_parallel_replicas_mode` or `parallel_replicas_only_with_analyzer`), a replica // just executes the query over all of its data. - if (context->canUseOffsetParallelReplicas()) -======= - // `automatic_parallel_replicas_mode`), a replica just executes the query over all of its - // data. if (context->canUseOffsetParallelReplicas() && allow_reading_shard_from_several_replicas) ->>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) remote_query_executor->setPoolMode(PoolMode::GET_MANY); else remote_query_executor->setPoolMode(PoolMode::GET_ONE); @@ -910,22 +904,10 @@ Pipes ReadFromRemote::addPipes(const ClusterProxy::SelectStreamFactory::Shards & void ReadFromRemote::initializePipeline(QueryPipelineBuilder & pipeline, const BuildQueryPipelineSettings &) { -<<<<<<< HEAD - Pipes pipes = addPipes(shards, output_header); -======= if (context->canUseOffsetParallelReplicas() && !allow_reading_shard_from_several_replicas) LOG_DEBUG(log, "Reading every shard from one replica, the consumers of the read may depend on the order of its streams"); - Pipes pipes = addPipes(shards, output_header, unavailable_shard_tracker); - - /// One pipe per executor, so the built pipes are this query's complete unit count. - if (unavailable_shard_tracker) - { - if (pipes.size() > shards.size()) - unavailable_shard_tracker->registerExtraUnits(pipes.size() - shards.size()); - unavailable_shard_tracker->seal(); - } ->>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) + Pipes pipes = addPipes(shards, output_header); auto pipe = Pipe::unitePipes(std::move(pipes)); diff --git a/src/Processors/QueryPlan/ReadFromRemote.h b/src/Processors/QueryPlan/ReadFromRemote.h index c859becca1911..2f14d1c56f99a 100644 --- a/src/Processors/QueryPlan/ReadFromRemote.h +++ b/src/Processors/QueryPlan/ReadFromRemote.h @@ -72,11 +72,8 @@ class ReadFromRemote final : public SourceStepWithFilterBase const String cluster_name; UnavailableShardTrackerPtr unavailable_shard_tracker; std::optional priority_func_factory; -<<<<<<< HEAD bool is_remote_function = false; -======= bool allow_reading_shard_from_several_replicas = false; ->>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) Pipes addPipes(const ClusterProxy::SelectStreamFactory::Shards & used_shards, const SharedHeader & out_header);