From 83a2242e2d574beca2c7c31bfa0d0f86c9229760 Mon Sep 17 00:00:00 2001
From: Nikolai Kochetov <nik-kochetov@yandex-team.ru>
Date: Mon, 8 Apr 2019 13:37:09 +0300
Subject: [PATCH] Check num columns in chunk while pushing to port.

---
 dbms/src/Processors/Chunk.cpp                        | 10 ++++++++++
 dbms/src/Processors/Chunk.h                          |  2 ++
 dbms/src/Processors/Port.h                           | 12 ++++++++++++
 .../Processors/Transforms/ExpressionTransform.cpp    |  3 +--
 4 files changed, 25 insertions(+), 2 deletions(-)

diff --git a/dbms/src/Processors/Chunk.cpp b/dbms/src/Processors/Chunk.cpp
index 5d3f79ec1fa..1a9db9e9ca8 100644
--- a/dbms/src/Processors/Chunk.cpp
+++ b/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)
 {
diff --git a/dbms/src/Processors/Chunk.h b/dbms/src/Processors/Chunk.h
index 9bfcf8c3a3a..efe6a594b74 100644
--- a/dbms/src/Processors/Chunk.h
+++ b/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;
diff --git a/dbms/src/Processors/Port.h b/dbms/src/Processors/Port.h
index 3127a191cc0..711f5bc644a 100644
--- a/dbms/src/Processors/Port.h
+++ b/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));
     }
 
diff --git a/dbms/src/Processors/Transforms/ExpressionTransform.cpp b/dbms/src/Processors/Transforms/ExpressionTransform.cpp
index cd49946a136..4e1084b9f85 100644
--- a/dbms/src/Processors/Transforms/ExpressionTransform.cpp
+++ b/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);
 }