Merge pull request #10757 from ClickHouse/fix-parallel-mv

alesapin · web-flow · commit ef1c7da4af1a · 2020-05-15T14:03:09.000+03:00
Fix parallel MV
diff --git a/src/DataStreams/PushingToViewsBlockOutputStream.cpp b/src/DataStreams/PushingToViewsBlockOutputStream.cpp
@@ -90,7 +90,7 @@ PushingToViewsBlockOutputStream::PushingToViewsBlockOutputStream(
         else
             out = std::make_shared<PushingToViewsBlockOutputStream>(dependent_table, *views_context, ASTPtr());
 
-        views.emplace_back(ViewInfo{std::move(query), database_table, std::move(out)});
+        views.emplace_back(ViewInfo{std::move(query), database_table, std::move(out), nullptr});
     }
 
     /// Do not push to destination table if the flag is set
@@ -162,7 +162,12 @@ void PushingToViewsBlockOutputStream::write(const Block & block)
     {
         // Process sequentially
         for (size_t view_num = 0; view_num < views.size(); ++view_num)
+        {
             process(block, view_num);
+
+            if (views[view_num].exception)
+                std::rethrow_exception(views[view_num].exception);
+        }
     }
 }
 
@@ -190,8 +195,18 @@ void PushingToViewsBlockOutputStream::writeSuffix()
     if (output)
         output->writeSuffix();
 
+    std::exception_ptr first_exception;
+
     for (auto & view : views)
     {
+        if (view.exception)
+        {
+            if (!first_exception)
+                first_exception = view.exception;
+
+            continue;
+        }
+
         try
         {
             view.out->writeSuffix();
@@ -202,6 +217,9 @@ void PushingToViewsBlockOutputStream::writeSuffix()
             throw;
         }
     }
+
+    if (first_exception)
+        std::rethrow_exception(first_exception);
 }
 
 void PushingToViewsBlockOutputStream::flush()
@@ -270,7 +288,11 @@ void PushingToViewsBlockOutputStream::process(const Block & block, size_t view_n
     catch (Exception & ex)
     {
         ex.addMessage("while pushing to view " + view.table_id.getNameForLogs());
-        throw;
+        view.exception = std::current_exception();
+    }
+    catch (...)
+    {
+        view.exception = std::current_exception();
     }
 }
 
diff --git a/src/DataStreams/PushingToViewsBlockOutputStream.h b/src/DataStreams/PushingToViewsBlockOutputStream.h
@@ -40,6 +40,7 @@ class PushingToViewsBlockOutputStream : public IBlockOutputStream
         ASTPtr query;
         StorageID table_id;
         BlockOutputStreamPtr out;
+        std::exception_ptr exception;
     };
 
     std::vector<ViewInfo> views;
diff --git a/tests/queries/0_stateless/01275_parallel_mv.reference b/tests/queries/0_stateless/01275_parallel_mv.reference
@@ -0,0 +1,4 @@
+10
+10
+0
+10
diff --git a/tests/queries/0_stateless/01275_parallel_mv.sql b/tests/queries/0_stateless/01275_parallel_mv.sql
@@ -0,0 +1,18 @@
+drop table if exists testX;
+drop table if exists testXA;
+drop table if exists testXB;
+drop table if exists testXC;
+
+create table testX (A Int64) engine=MergeTree order by tuple();
+
+create materialized view testXA engine=MergeTree order by tuple() as select sleep(1) from testX;
+create materialized view testXB engine=MergeTree order by tuple() as select sleep(2), throwIf(A=1) from testX;
+create materialized view testXC engine=MergeTree order by tuple() as select sleep(1) from testX;
+
+set parallel_view_processing=1;
+insert into testX select number from numbers(10); -- {serverError 395}
+
+select count() from testX;
+select count() from testXA;
+select count() from testXB;
+select count() from testXC;

Original file line number	Diff line number	Diff line change
`@@ -90,7 +90,7 @@ PushingToViewsBlockOutputStream::PushingToViewsBlockOutputStream(`
`90`	`90`	`else`
`91`	`91`	`out = std::make_shared<PushingToViewsBlockOutputStream>(dependent_table, *views_context, ASTPtr());`
`92`	`92`
`93`		`- views.emplace_back(ViewInfo{std::move(query), database_table, std::move(out)});`
	`93`	`+ views.emplace_back(ViewInfo{std::move(query), database_table, std::move(out), nullptr});`
`94`	`94`	`}`
`95`	`95`
`96`	`96`	`/// Do not push to destination table if the flag is set`
`@@ -162,7 +162,12 @@ void PushingToViewsBlockOutputStream::write(const Block & block)`
`162`	`162`	`{`
`163`	`163`	`// Process sequentially`
`164`	`164`	`for (size_t view_num = 0; view_num < views.size(); ++view_num)`
	`165`	`+ {`
`165`	`166`	`process(block, view_num);`
	`167`	`+`
	`168`	`+ if (views[view_num].exception)`
	`169`	`+ std::rethrow_exception(views[view_num].exception);`
	`170`	`+ }`
`166`	`171`	`}`
`167`	`172`	`}`
`168`	`173`
`@@ -190,8 +195,18 @@ void PushingToViewsBlockOutputStream::writeSuffix()`
`190`	`195`	`if (output)`
`191`	`196`	`output->writeSuffix();`
`192`	`197`
	`198`	`+ std::exception_ptr first_exception;`
	`199`	`+`
`193`	`200`	`for (auto & view : views)`
`194`	`201`	`{`
	`202`	`+ if (view.exception)`
	`203`	`+ {`
	`204`	`+ if (!first_exception)`
	`205`	`+ first_exception = view.exception;`
	`206`	`+`
	`207`	`+ continue;`
	`208`	`+ }`
	`209`	`+`
`195`	`210`	`try`
`196`	`211`	`{`
`197`	`212`	`view.out->writeSuffix();`
`@@ -202,6 +217,9 @@ void PushingToViewsBlockOutputStream::writeSuffix()`
`202`	`217`	`throw;`
`203`	`218`	`}`
`204`	`219`	`}`
	`220`	`+`
	`221`	`+ if (first_exception)`
	`222`	`+ std::rethrow_exception(first_exception);`
`205`	`223`	`}`
`206`	`224`
`207`	`225`	`void PushingToViewsBlockOutputStream::flush()`
`@@ -270,7 +288,11 @@ void PushingToViewsBlockOutputStream::process(const Block & block, size_t view_n`
`270`	`288`	`catch (Exception & ex)`
`271`	`289`	`{`
`272`	`290`	`ex.addMessage("while pushing to view " + view.table_id.getNameForLogs());`
`273`		`- throw;`
	`291`	`+ view.exception = std::current_exception();`
	`292`	`+ }`
	`293`	`+ catch (...)`
	`294`	`+ {`
	`295`	`+ view.exception = std::current_exception();`
`274`	`296`	`}`
`275`	`297`	`}`
`276`	`298`
-Original file line number
+Diff line change
@@ @@ -0,0 +1,4 @@ @@
 +10
 +10
 +0
 +10