fix

Signed-off-by: xufei <xufeixw@mail.ustc.edu.cn>
pingcap · Feb 1, 2023 · 82bc816 · 82bc816
1 parent 3b4da89
commit 82bc816
Show file tree

Hide file tree

Showing 3 changed files with 2 additions and 5 deletions.
diff --git a/dbms/src/Flash/Coprocessor/InterpreterUtils.cpp b/dbms/src/Flash/Coprocessor/InterpreterUtils.cpp
@@ -114,14 +114,13 @@ void orderStreams(
 
     if (enable_fine_grained_shuffle)
     {
-        size_t merge_sort_concurrency = enable_fine_grained_shuffle ? std::min(max_streams, pipeline.streams.size()) : 1;
         pipeline.transform([&](auto & stream) {
             stream = std::make_shared<MergeSortingBlockInputStream>(
                 stream,
                 order_descr,
                 settings.max_block_size,
                 limit,
-                getAverageThreshold(settings.max_bytes_before_external_sort, merge_sort_concurrency),
+                getAverageThreshold(settings.max_bytes_before_external_sort, std::min(max_streams, pipeline.streams.size())),
                 SpillConfig(context.getTemporaryPath(), fmt::format("{}_sort", log->identifier()), settings.max_spilled_size_per_spill, context.getFileProvider()),
                 log->identifier());
             stream->setExtraInfo(String(enableFineGrainedShuffleExtraInfo));

diff --git a/dbms/src/Interpreters/Aggregator.cpp b/dbms/src/Interpreters/Aggregator.cpp
@@ -824,9 +824,7 @@ bool Aggregator::executeOnBlock(
       * It allows you to make, in the subsequent, an effective merge - either economical from memory or parallel.
       */
     if (result.isConvertibleToTwoLevel() && worth_convert_to_two_level)
-    {
         result.convertToTwoLevel();
-    }
 
     /** Flush data to disk if too much RAM is consumed.
       * Data can only be flushed to disk if a two-level aggregation is supported.

diff --git a/dbms/src/Interpreters/Aggregator.h b/dbms/src/Interpreters/Aggregator.h
@@ -742,7 +742,7 @@ struct AggregatedDataVariants : private boost::noncopyable
     case Type::NAME:                                                                                       \
     {                                                                                                      \
         const auto * ptr = reinterpret_cast<const AggregationMethodName(NAME) *>(aggregation_method_impl); \
-        return ptr->data.getBufferSizeInBytes();                                                           \
+        bytes_count = ptr->data.getBufferSizeInBytes();                                                    \
     }
 
             APPLY_FOR_AGGREGATED_VARIANTS(M)