Merge pull request #11200 from ClickHouse/fix-strict-resize-finishing

KochetovNicolai · web-flow · commit 361b92f08320 · 2020-05-26T21:31:44.000+03:00
Fix StrictResize
diff --git a/src/Processors/ResizeProcessor.cpp b/src/Processors/ResizeProcessor.cpp
@@ -347,11 +347,16 @@ IProcessor::Status StrictResizeProcessor::prepare(const PortNumbers & updated_in
 
         auto & waiting_output = output_ports[input_with_data.waiting_output];
 
-        if (waiting_output.status != OutputStatus::NeedData)
-            throw Exception("Invalid status for associated output.", ErrorCodes::LOGICAL_ERROR);
+        if (waiting_output.status == OutputStatus::NotActive)
+            throw Exception("Invalid status NotActive for associated output.", ErrorCodes::LOGICAL_ERROR);
 
-        waiting_output.port->pushData(input_with_data.port->pullData(/* set_not_needed = */ true));
-        waiting_output.status = OutputStatus::NotActive;
+        if (waiting_output.status != OutputStatus::Finished)
+        {
+            waiting_output.port->pushData(input_with_data.port->pullData(/* set_not_needed = */ true));
+            waiting_output.status = OutputStatus::NotActive;
+        }
+        else
+            abandoned_chunks.emplace_back(input_with_data.port->pullData(/* set_not_needed = */ true));
 
         if (input_with_data.port->isFinished())
         {
@@ -370,6 +375,18 @@ IProcessor::Status StrictResizeProcessor::prepare(const PortNumbers & updated_in
         return Status::Finished;
     }
 
+    /// Process abandoned chunks if any.
+    while (!abandoned_chunks.empty() && !waiting_outputs.empty())
+    {
+        auto & waiting_output = output_ports[waiting_outputs.front()];
+        waiting_outputs.pop();
+
+        waiting_output.port->pushData(std::move(abandoned_chunks.back()));
+        abandoned_chunks.pop_back();
+
+        waiting_output.status = OutputStatus::NotActive;
+    }
+
     /// Enable more inputs if needed.
     while (!disabled_input_ports.empty() && !waiting_outputs.empty())
     {
@@ -383,6 +400,7 @@ IProcessor::Status StrictResizeProcessor::prepare(const PortNumbers & updated_in
         waiting_outputs.pop();
     }
 
+    /// Close all other waiting for data outputs (there is no corresponding input for them).
     while (!waiting_outputs.empty())
     {
        auto & output = output_ports[waiting_outputs.front()];
diff --git a/src/Processors/ResizeProcessor.h b/src/Processors/ResizeProcessor.h
@@ -128,6 +128,9 @@ class StrictResizeProcessor : public IProcessor
 
     std::vector<InputPortWithStatus> input_ports;
     std::vector<OutputPortWithStatus> output_ports;
+    /// This field contained chunks which were read for output which had became finished while reading was happening.
+    /// They will be pushed to any next waiting output.
+    std::vector<Port::Data> abandoned_chunks;
 };
 
 }
diff --git a/tests/queries/0_stateless/001283_strict_resize_bug.reference b/tests/queries/0_stateless/001283_strict_resize_bug.reference
@@ -0,0 +1 @@
+49999995000000
diff --git a/tests/queries/0_stateless/001283_strict_resize_bug.sql b/tests/queries/0_stateless/001283_strict_resize_bug.sql
@@ -0,0 +1,7 @@
+drop table if exists num_10m;
+create table num_10m (number UInt64) engine = MergeTree order by tuple();
+insert into num_10m select * from numbers(10000000);
+
+select * from (select sum(number) from num_10m union all select sum(number) from num_10m) limit 1 settings max_block_size = 1024;
+
+drop table if exists num_1m;

Original file line number	Diff line number	Diff line change
`@@ -128,6 +128,9 @@ class StrictResizeProcessor : public IProcessor`
`128`	`128`
`129`	`129`	`std::vector<InputPortWithStatus> input_ports;`
`130`	`130`	`std::vector<OutputPortWithStatus> output_ports;`
	`131`	`+ /// This field contained chunks which were read for output which had became finished while reading was happening.`
	`132`	`+ /// They will be pushed to any next waiting output.`
	`133`	`+ std::vector<Port::Data> abandoned_chunks;`
`131`	`134`	`};`
`132`	`135`
`133`	`136`	`}`