Better exception message for invalid pipeline

dbms/src/Processors/Executors/PipelineExecutor.cpp

@@ -40,7 +40,21 @@ PipelineExecutor::PipelineExecutor(Processors & processors_, QueryStatus * elem)
     , expand_pipeline_task(nullptr)
     , process_list_element(elem)
 {
-    buildGraph();
+    try
+    {
+        buildGraph();
+    }
+    catch (Exception & exception)
+    {
+        /// If exception was thrown while pipeline initialization, it means that query pipeline was not build correctly.
+        /// It is logical error, and we need more information about pipeline.
+        WriteBufferFromOwnString buf;
+        printPipeline(processors, buf);
+        buf.finalize();
+        exception.addMessage("Query pipeline:\n" + buf.str());
+
+        throw;
+    }
 }
 
 bool PipelineExecutor::addEdges(UInt64 node)

dbms/src/Processors/printPipeline.h

@@ -42,6 +42,9 @@ void printPipeline(const Processors & processors, const Statuses & statuses, Wri
     {
         for (const auto & port : processor->getOutputs())
         {
+            if (!port.isConnected())
+                continue;
+
             const IProcessor & curr = *processor;
             const IProcessor & next = port.getInputPort().getProcessor();
 

dbms/src/Processors/tests/gtest_exception_on_incorrect_pipeline.cpp

@@ -0,0 +1,69 @@
+#include <gtest/gtest.h>
+
+#include <Processors/Sources/SourceFromSingleChunk.h>
+#include <Processors/NullSink.h>
+#include <Processors/Executors/PipelineExecutor.h>
+
+#include <Columns/ColumnsNumber.h>
+#include <DataTypes/DataTypesNumber.h>
+
+using namespace DB;
+
+TEST(Processors, PortsConnected)
+{
+    auto col = ColumnUInt8::create(1, 1);
+    Columns columns;
+    columns.emplace_back(std::move(col));
+    Chunk chunk(std::move(columns), 1);
+
+    Block header = {ColumnWithTypeAndName(ColumnUInt8::create(), std::make_shared<DataTypeUInt8>(), "x")};
+
+    auto source = std::make_shared<SourceFromSingleChunk>(std::move(header), std::move(chunk));
+    auto sink = std::make_shared<NullSink>(source->getPort().getHeader());
+
+    connect(source->getPort(), sink->getPort());
+
+    Processors processors;
+    processors.emplace_back(std::move(source));
+    processors.emplace_back(std::move(sink));
+
+    PipelineExecutor executor(processors);
+    executor.execute(1);
+}
+
+TEST(Processors, PortsNotConnected)
+{
+    auto col = ColumnUInt8::create(1, 1);
+    Columns columns;
+    columns.emplace_back(std::move(col));
+    Chunk chunk(std::move(columns), 1);
+
+    Block header = {ColumnWithTypeAndName(ColumnUInt8::create(), std::make_shared<DataTypeUInt8>(), "x")};
+
+    auto source = std::make_shared<SourceFromSingleChunk>(std::move(header), std::move(chunk));
+    auto sink = std::make_shared<NullSink>(source->getPort().getHeader());
+
+    /// connect(source->getPort(), sink->getPort());
+
+    Processors processors;
+    processors.emplace_back(std::move(source));
+    processors.emplace_back(std::move(sink));
+
+    auto exec = [&]()
+    {
+
+        try
+        {
+            PipelineExecutor executor(processors);
+            executor.execute(1);
+        }
+        catch (DB::Exception & e)
+        {
+            std::cout << e.displayText() << std::endl;
+            ASSERT_TRUE(e.displayText().find("pipeline") != std::string::npos);
+            throw;
+        }
+    };
+
+    ASSERT_THROW(exec(), DB::Exception);
+}