From fa0f4543b688861a97843a8767d895f52b6dee9d Mon Sep 17 00:00:00 2001 From: Igor Nikonov Date: Mon, 22 Jul 2024 13:44:25 +0000 Subject: [PATCH] Fix marks multiplier for local replica to avoid overflow with fuzzer --- src/Processors/QueryPlan/ReadFromMergeTree.cpp | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/src/Processors/QueryPlan/ReadFromMergeTree.cpp b/src/Processors/QueryPlan/ReadFromMergeTree.cpp index 6f0037684e9..5922550eaf3 100644 --- a/src/Processors/QueryPlan/ReadFromMergeTree.cpp +++ b/src/Processors/QueryPlan/ReadFromMergeTree.cpp @@ -384,7 +384,11 @@ Pipe ReadFromMergeTree::readFromPoolParallelReplicas(RangesInDataParts parts_wit /// We have a special logic for local replica. It has to read less data, because in some cases it should /// merge states of aggregate functions or do some other important stuff other than reading from Disk. - const auto multiplier = context->getSettingsRef().parallel_replicas_single_task_marks_count_multiplier; + auto multiplier = context->getSettingsRef().parallel_replicas_single_task_marks_count_multiplier; + // avoid using multiplier if min marks to read is already too big (to avoid overflow) + if (pool_settings.min_marks_for_concurrent_read >= std::numeric_limits::max()) + multiplier = 1.0f; + if (auto result = pool_settings.min_marks_for_concurrent_read * multiplier; canConvertTo(result)) pool_settings.min_marks_for_concurrent_read = static_cast(result); else @@ -553,7 +557,11 @@ Pipe ReadFromMergeTree::readInOrder( .number_of_current_replica = number_of_current_replica.value_or(client_info.number_of_current_replica), }; - const auto multiplier = context->getSettingsRef().parallel_replicas_single_task_marks_count_multiplier; + auto multiplier = context->getSettingsRef().parallel_replicas_single_task_marks_count_multiplier; + // avoid using multiplier if min marks to read is already too big (to avoid overflow) + if (pool_settings.min_marks_for_concurrent_read >= std::numeric_limits::max()) + multiplier = 1.0f; + if (auto result = pool_settings.min_marks_for_concurrent_read * multiplier; canConvertTo(result)) pool_settings.min_marks_for_concurrent_read = static_cast(result); else