impl

docker/packager/packager

@@ -331,7 +331,7 @@ if __name__ == "__main__":
             "clang-14-freebsd",
             "gcc-11",
         ),
-        default="clang-13",
+        default="clang-14",
         help="a compiler to use",
     )
     parser.add_argument(

src/Processors/Transforms/AggregatingTransform.h

@@ -70,6 +70,21 @@ struct ManyAggregatedData
         for (auto & mut : mutexes)
             mut = std::make_unique<std::mutex>();
     }
+
+    ~ManyAggregatedData()
+    {
+        // Aggregation states destruction may be very time-consuming.
+        // In the case of a query with LIMIT, most states won't be destroyed during conversion to blocks.
+        // Without the following code, they would be destroyed in the destructor of AggregatedDataVariants in the current thread (i.e. sequentially).
+        const auto pool = std::make_unique<ThreadPool>(variants.size());
+
+        for (auto && variant : variants)
+        {
+            // It doesn't make sense to spawn a thread if the variant is not going to actually destroy anything.
+            if (variant->aggregator)
+                pool->trySchedule([variant = std::move(variant)]() {});
+        }
+    }
 };
 
 using AggregatingTransformParamsPtr = std::shared_ptr<AggregatingTransformParams>;

tests/performance/destroy_aggregate_states.xml

@@ -0,0 +1,8 @@
+<test>
+  <settings>
+    <max_memory_usage>100000000000</max_memory_usage>
+  </settings>
+
+  <query>select number, uniq(number) from numbers_mt(1e7) group by number limit 100 format Null</query>
+  <query>select number, uniq(number) from numbers_mt(1e8) group by number limit 100 format Null</query>
+</test>