Check num columns in chunk while pushing to port.

dbms/src/Processors/Chunk.cpp

@@ -1,5 +1,6 @@
 #include <Processors/Chunk.h>
 #include <IO/WriteHelpers.h>
+#include <IO/Operators.h>
 
 namespace DB
 {
@@ -144,6 +145,15 @@ UInt64 Chunk::allocatedBytes() const
     return res;
 }
 
+std::string Chunk::dumpStructure() const
+{
+    WriteBufferFromOwnString out;
+    for (auto & column : columns)
+        out << ' ' << column->dumpStructure();
+
+    return out.str();
+}
+
 
 void ChunkMissingValues::setBit(size_t column_idx, size_t row_idx)
 {

dbms/src/Processors/Chunk.h

@@ -53,6 +53,8 @@ public:
     UInt64 bytes() const;
     UInt64 allocatedBytes() const;
 
+    std::string dumpStructure() const;
+
 private:
     Columns columns;
     UInt64 num_rows = 0;

dbms/src/Processors/Port.h

@@ -255,6 +255,18 @@ public:
             ++(*version);
 
         assumeConnected();
+
+        if (chunk.getNumColumns() != header.columns())
+        {
+            String msg = "Invalid number of columns in chunk pushed to OutputPort. Expected"
+                    + std::to_string(header.columns()) + ", found " + std::to_string(chunk.getNumColumns()) + '\n';
+
+            msg += "Header: " + header.dumpStructure() + '\n';
+            msg += "Chunk: " + chunk.dumpStructure() + '\n';
+
+            throw Exception(msg, ErrorCodes::LOGICAL_ERROR);
+        }
+
         state->push(std::move(chunk));
     }
 

dbms/src/Processors/Transforms/ExpressionTransform.cpp

@@ -19,10 +19,9 @@ ExpressionTransform::ExpressionTransform(const Block & header, ExpressionActions
 
 void ExpressionTransform::transform(Chunk & chunk)
 {
-    auto num_rows = chunk.getNumRows();
     auto block = getInputPort().getHeader().cloneWithColumns(chunk.detachColumns());
     expression->execute(block);
-    num_rows = block.rows();
+    auto num_rows = block.rows();
     chunk.setColumns(block.getColumns(), num_rows);
 }