Fix tests.

src/Processors/QueryPipeline.cpp

@@ -262,6 +262,26 @@ QueryPipeline QueryPipeline::unitePipelines(
     return pipeline;
 }
 
+
+void QueryPipeline::addCreatingSetsTransform()
+{
+    pipeline.resize(1);
+
+    auto transform = std::make_shared<CreatingSetsTransform>(
+            pipeline.getHeader(),
+            getOutputStream().header,
+            std::move(subquery_for_set),
+            network_transfer_limits,
+            context));
+
+    InputPort * totals_port = nullptr;
+
+    if (pipe.getTotalsPort())
+        totals_port = transform->addTotalsPort();
+
+    pipe.addTransform(std::move(transform), totals_port, nullptr);
+}
+
 void QueryPipeline::addDelayingPipeline(QueryPipeline pipeline)
 {
     checkInitializedAndNotCompleted();

src/Processors/QueryPipeline.h

@@ -89,6 +89,8 @@ public:
     /// Pipeline must have same header.
     void addDelayingPipeline(QueryPipeline pipeline);
 
+    void addCreatingSetsTransform();
+
     PipelineExecutorPtr execute();
 
     size_t getNumStreams() const { return pipe.numOutputPorts(); }

src/Processors/QueryPlan/CreatingSetsStep.cpp

@@ -39,14 +39,7 @@ CreatingSetStep::CreatingSetStep(
 
 void CreatingSetStep::transformPipeline(QueryPipeline & pipeline)
 {
-    pipeline.resize(1);
-    pipeline.addTransform(
-        std::make_shared<CreatingSetsTransform>(
-            pipeline.getHeader(),
-            getOutputStream().header,
-            std::move(subquery_for_set),
-            network_transfer_limits,
-            context));
+    pipeline.addCreatingSetsTransform();
 }
 
 void CreatingSetStep::describeActions(FormatSettings & settings) const