Fix test for streams.

src/Processors/Executors/TreeExecutorBlockInputStream.cpp

@@ -164,7 +164,7 @@ void TreeExecutorBlockInputStream::execute(bool on_totals, bool on_extremes)
         }
     };
 
-    while (!stack.empty())
+    while (!stack.empty() && !is_cancelled)
     {
         IProcessor * node = stack.top();
 
@@ -295,7 +295,7 @@ void TreeExecutorBlockInputStream::initRowsBeforeLimit()
 
 Block TreeExecutorBlockInputStream::readImpl()
 {
-    while (true)
+    while (!is_cancelled)
     {
         if (input_port->isFinished())
         {
@@ -338,6 +338,8 @@ Block TreeExecutorBlockInputStream::readImpl()
 
         execute(false, false);
     }
+
+    return {};
 }
 
 void TreeExecutorBlockInputStream::setProgressCallback(const ProgressCallback & callback)
@@ -373,4 +375,12 @@ void TreeExecutorBlockInputStream::addTotalRowsApprox(size_t value)
         sources_with_progress.front()->addTotalRowsApprox(value);
 }
 
+void TreeExecutorBlockInputStream::cancel(bool kill)
+{
+    IBlockInputStream::cancel(kill);
+
+    for (auto & processor : processors)
+        processor->cancel();
+}
+
 }

src/Processors/Executors/TreeExecutorBlockInputStream.h

@@ -39,6 +39,8 @@ public:
     String getName() const override { return "TreeExecutor"; }
     Block getHeader() const override { return root->getOutputs().front().getHeader(); }
 
+    void cancel(bool kill) override;
+
     /// This methods does not affect TreeExecutor as IBlockInputStream itself.
     /// They just passed to all SourceWithProgress processors.
     void setProgressCallback(const ProgressCallback & callback) final;