Merge pull request ClickHouse#11330 from ClickHouse/fix-10241

alexey-milovidov · web-flow · commit 92ac60844728 · 2020-06-02T04:06:39.000+03:00
More parallel MV processing.
diff --git a/src/DataStreams/PushingToViewsBlockOutputStream.cpp b/src/DataStreams/PushingToViewsBlockOutputStream.cpp
@@ -153,7 +153,7 @@ void PushingToViewsBlockOutputStream::write(const Block & block)
     const Settings & settings = context.getSettingsRef();
     if (settings.parallel_view_processing && views.size() > 1)
     {
-        // Push to views concurrently if enabled, and more than one view is attached
+        // Push to views concurrently if enabled and more than one view is attached
         ThreadPool pool(std::min(size_t(settings.max_threads), views.size()));
         for (size_t view_num = 0; view_num < views.size(); ++view_num)
         {
@@ -208,6 +208,45 @@ void PushingToViewsBlockOutputStream::writeSuffix()
 
     std::exception_ptr first_exception;
 
+    const Settings & settings = context.getSettingsRef();
+    bool parallel_processing = false;
+
+    /// Run writeSuffix() for views in separate thread pool.
+    /// In could have been done in PushingToViewsBlockOutputStream::process, however
+    /// it is not good if insert into main table fail but into view succeed.
+    if (settings.parallel_view_processing && views.size() > 1)
+    {
+        parallel_processing = true;
+
+        // Push to views concurrently if enabled and more than one view is attached
+        ThreadPool pool(std::min(size_t(settings.max_threads), views.size()));
+        auto thread_group = CurrentThread::getGroup();
+
+        for (auto & view : views)
+        {
+            if (view.exception)
+                continue;
+
+            pool.scheduleOrThrowOnError([thread_group, &view]
+            {
+                setThreadName("PushingToViews");
+                if (thread_group)
+                    CurrentThread::attachToIfDetached(thread_group);
+
+                try
+                {
+                    view.out->writeSuffix();
+                }
+                catch (...)
+                {
+                    view.exception = std::current_exception();
+                }
+            });
+        }
+        // Wait for concurrent view processing
+        pool.wait();
+    }
+
     for (auto & view : views)
     {
         if (view.exception)
@@ -218,6 +257,9 @@ void PushingToViewsBlockOutputStream::writeSuffix()
             continue;
         }
 
+        if (parallel_processing)
+            continue;
+
         try
         {
             view.out->writeSuffix();
diff --git a/tests/performance/parallel_mv.xml b/tests/performance/parallel_mv.xml
@@ -0,0 +1,24 @@
+<test>
+    <settings>
+        <parallel_view_processing>1</parallel_view_processing>
+    </settings>
+
+    <create_query>create table main_table (number UInt64) engine = MergeTree order by tuple();</create_query>
+    <create_query>create materialized view mv_1 engine = MergeTree order by tuple() as
+        select number, toString(number) from main_table where number % 13 != 0;</create_query>
+    <create_query>create materialized view mv_2 engine = MergeTree order by tuple() as
+        select number, toString(number) from main_table where number % 13 != 1;</create_query>
+    <create_query>create materialized view mv_3 engine = MergeTree order by tuple() as
+        select number, toString(number) from main_table where number % 13 != 3;</create_query>
+    <create_query>create materialized view mv_4 engine = MergeTree order by tuple() as
+        select number, toString(number) from main_table where number % 13 != 4;</create_query>
+
+    <!--<query>insert into main_table select number from numbers(100000)</query>-->
+    <query>insert into main_table select number from numbers(1000000)</query>
+
+    <drop_query>drop table if exists main_table;</drop_query>
+    <drop_query>drop table if exists mv_1;</drop_query>
+    <drop_query>drop table if exists mv_2;</drop_query>
+    <drop_query>drop table if exists mv_3;</drop_query>
+    <drop_query>drop table if exists mv_4;</drop_query>
+</test>
