From c3b7597c10b54d7951646285c21313e02fba8902 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 | 17 +++ .../allowReadingShardsFromSeveralReplicas.cpp | 84 ++++++++++++++ .../QueryPlan/Optimizations/optimizeTree.cpp | 21 ++++ src/Processors/QueryPlan/ReadFromRemote.cpp | 34 ++++++ 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, 322 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 adfdad727e793..26b359920c495 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -7535,6 +7535,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 f84b7538e6a41..c9153de1741f7 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 862e0cedb423e..bbd35fe409b46 100644 --- a/src/Processors/QueryPlan/Optimizations/Optimizations.h +++ b/src/Processors/QueryPlan/Optimizations/Optimizations.h @@ -243,8 +243,25 @@ void optimizeJoinLogical(QueryPlan::Node & node, QueryPlan::Nodes &, const Query /// A separate tree traverse to apply sorting properties after *InOrder optimizations. void applyOrder(const QueryPlanOptimizationSettings & optimization_settings, QueryPlan::Node & root); +<<<<<<< HEAD /// Returns the name of used projection or nullopt if no projection is used. std::optional optimizeUseAggregateProjections( +======= +/// A separate tree traverse that propagates the stream-disjointness property (no two output streams +/// carry the same key value). +void applyStreamDisjointness(const QueryPlanOptimizationSettings & optimization_settings, QueryPlan::Node & root); + +/// 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 21ae4c550d830..e8fb8259bb3b9 100644 --- a/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp +++ b/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp @@ -584,6 +584,27 @@ void optimizeTreeSecondPass( optimizeJoinByShards(root); considerEnablingParallelReplicas(optimization_settings, root, query_plan); +<<<<<<< HEAD +======= + + /// Run after every optimization that can rewrite aggregation, sorting, projections, + /// distributed fragments, or parallel replicas. This placement makes the pass a pure + /// admission check: no later optimization needs to retract the heap or its synthetic sort. + if (optimization_settings.enable_group_by_top_k_optimization) + { + traverseQueryPlan(stack, root, [&](auto & frame_node) { tryOptimizeGroupByTopK(&frame_node, nodes, extra_settings); }); + } + + /// 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 b41f36a4343eb..5892af480069b 100644 --- a/src/Processors/QueryPlan/ReadFromRemote.cpp +++ b/src/Processors/QueryPlan/ReadFromRemote.cpp @@ -842,6 +842,7 @@ void ReadFromRemote::addPipe( remote_query_executor->setRemoteFunction(is_remote_function); remote_query_executor->setShardCount(shard_count); +<<<<<<< HEAD if (context->canUseTaskBasedParallelReplicas() || parallel_replicas_disabled) { // when doing parallel reading from replicas (ParallelReplicasMode::READ_TASKS) on a shard: @@ -852,6 +853,24 @@ void ReadFromRemote::addPipe( // each will execute parallel reading from replicas, so the query result will be multiplied by the number of created coordinators // // In case parallel replicas are disabled, there also should be a single connection to each shard to prevent result duplication +======= + // 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: + // * with parallel reading from replicas (`ParallelReplicasMode::READ_TASKS`) the replica we + // 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 + // `automatic_parallel_replicas_mode`), a replica just executes the query over all of its + // data. + if (context->canUseOffsetParallelReplicas() && allow_reading_shard_from_several_replicas) + remote_query_executor->setPoolMode(PoolMode::GET_MANY); + else +>>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) remote_query_executor->setPoolMode(PoolMode::GET_ONE); } else @@ -885,7 +904,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 2e739e032be38..11f10d7959a4d 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 5f93e27cb188ea03cbff6672c6442ad2719a2bd0 Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Thu, 5 Nov 2026 20:35:34 +0100 Subject: [PATCH 2/2] Resolve conflicts in backport of #128865 Added the PR's one-replica condition to antalya-26.6's existing GET_ONE/GET_MANY branch in `ReadFromRemote::addPipe`, and dropped unrelated upstream code that came in through the conflict hunks. --- .../QueryPlan/Optimizations/Optimizations.h | 18 +-------- .../QueryPlan/Optimizations/optimizeTree.cpp | 18 --------- src/Processors/QueryPlan/ReadFromRemote.cpp | 38 +++---------------- src/Processors/QueryPlan/ReadFromRemote.h | 3 -- 4 files changed, 7 insertions(+), 70 deletions(-) diff --git a/src/Processors/QueryPlan/Optimizations/Optimizations.h b/src/Processors/QueryPlan/Optimizations/Optimizations.h index bbd35fe409b46..f0958d7b39df4 100644 --- a/src/Processors/QueryPlan/Optimizations/Optimizations.h +++ b/src/Processors/QueryPlan/Optimizations/Optimizations.h @@ -243,25 +243,11 @@ void optimizeJoinLogical(QueryPlan::Node & node, QueryPlan::Nodes &, const Query /// A separate tree traverse to apply sorting properties after *InOrder optimizations. void applyOrder(const QueryPlanOptimizationSettings & optimization_settings, QueryPlan::Node & root); -<<<<<<< HEAD -/// Returns the name of used projection or nullopt if no projection is used. -std::optional optimizeUseAggregateProjections( -======= -/// A separate tree traverse that propagates the stream-disjointness property (no two output streams -/// carry the same key value). -void applyStreamDisjointness(const QueryPlanOptimizationSettings & optimization_settings, QueryPlan::Node & root); - /// 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 e8fb8259bb3b9..3a5658bf9a7d3 100644 --- a/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp +++ b/src/Processors/QueryPlan/Optimizations/optimizeTree.cpp @@ -584,27 +584,9 @@ void optimizeTreeSecondPass( optimizeJoinByShards(root); considerEnablingParallelReplicas(optimization_settings, root, query_plan); -<<<<<<< HEAD -======= - - /// Run after every optimization that can rewrite aggregation, sorting, projections, - /// distributed fragments, or parallel replicas. This placement makes the pass a pure - /// admission check: no later optimization needs to retract the heap or its synthetic sort. - if (optimization_settings.enable_group_by_top_k_optimization) - { - traverseQueryPlan(stack, root, [&](auto & frame_node) { tryOptimizeGroupByTopK(&frame_node, nodes, extra_settings); }); - } - - /// 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 5892af480069b..b2da286e6e597 100644 --- a/src/Processors/QueryPlan/ReadFromRemote.cpp +++ b/src/Processors/QueryPlan/ReadFromRemote.cpp @@ -842,8 +842,7 @@ void ReadFromRemote::addPipe( remote_query_executor->setRemoteFunction(is_remote_function); remote_query_executor->setShardCount(shard_count); -<<<<<<< HEAD - if (context->canUseTaskBasedParallelReplicas() || parallel_replicas_disabled) + if (context->canUseTaskBasedParallelReplicas() || parallel_replicas_disabled || !allow_reading_shard_from_several_replicas) { // when doing parallel reading from replicas (ParallelReplicasMode::READ_TASKS) on a shard: // establish a connection to a replica on the shard, the replica will instantiate coordinator to manage parallel reading from replicas on the shard. @@ -853,24 +852,9 @@ void ReadFromRemote::addPipe( // each will execute parallel reading from replicas, so the query result will be multiplied by the number of created coordinators // // In case parallel replicas are disabled, there also should be a single connection to each shard to prevent result duplication -======= - // 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: - // * with parallel reading from replicas (`ParallelReplicasMode::READ_TASKS`) the replica we - // 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 - // `automatic_parallel_replicas_mode`), a replica just executes the query over all of its - // data. - if (context->canUseOffsetParallelReplicas() && allow_reading_shard_from_several_replicas) - remote_query_executor->setPoolMode(PoolMode::GET_MANY); - else ->>>>>>> 22a323d93d6 (Merge pull request #128865 from groeneai/offset-parallel-replicas-ordered-merges) + // + // With the offset based modes every replica returns its own part of the shard's result, and the replicas + // share one port, so a read whose consumer needs stream order uses one replica. remote_query_executor->setPoolMode(PoolMode::GET_ONE); } else @@ -904,22 +888,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 11f10d7959a4d..e4834dce114ed 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);