[WIP] Trying to fix block structure mismatch

dbms/src/Interpreters/InterpreterSelectQuery.cpp

@@ -1236,6 +1236,8 @@ void InterpreterSelectQuery::executeMergeSorted(Pipeline & pipeline)
     /// If there are several streams, then we merge them into one
     if (pipeline.hasMoreThanOneStream())
     {
+        unifyStreams(pipeline);
+
         /** MergingSortedBlockInputStream reads the sources sequentially.
           * To make the data on the remote servers prepared in parallel, we wrap it in AsynchronousBlockInputStream.
           */
@@ -1290,16 +1292,7 @@ void InterpreterSelectQuery::executeUnion(Pipeline & pipeline)
     /// If there are still several streams, then we combine them into one
     if (pipeline.hasMoreThanOneStream())
     {
-        /// Unify streams in case they have different headers.
-        auto first_header = pipeline.streams.at(0)->getHeader();
-        for (size_t i = 1; i < pipeline.streams.size(); ++i)
-        {
-            auto & stream = pipeline.streams[i];
-            auto header = stream->getHeader();
-            auto mode = ConvertingBlockInputStream::MatchColumnsMode::Name;
-            if (!blocksHaveEqualStructure(first_header, header))
-                stream = std::make_shared<ConvertingBlockInputStream>(context, stream, first_header, mode);
-        }
+        unifyStreams(pipeline);
 
         pipeline.firstStream() = std::make_shared<UnionBlockInputStream<>>(pipeline.streams, pipeline.stream_with_non_joined_data, max_streams);
         pipeline.stream_with_non_joined_data = nullptr;
@@ -1429,6 +1422,23 @@ void InterpreterSelectQuery::executeSubqueriesInSetsAndJoins(Pipeline & pipeline
         SizeLimits(settings.max_rows_to_transfer, settings.max_bytes_to_transfer, settings.transfer_overflow_mode));
 }
 
+void InterpreterSelectQuery::unifyStreams(Pipeline & pipeline)
+{
+    if (pipeline.hasMoreThanOneStream())
+    {
+        /// Unify streams in case they have different headers.
+        auto first_header = pipeline.streams.at(0)->getHeader();
+        for (size_t i = 1; i < pipeline.streams.size(); ++i)
+        {
+            auto & stream = pipeline.streams[i];
+            auto header = stream->getHeader();
+            auto mode = ConvertingBlockInputStream::MatchColumnsMode::Name;
+            if (!blocksHaveEqualStructure(first_header, header))
+                stream = std::make_shared<ConvertingBlockInputStream>(context, stream, first_header, mode);
+        }
+    }
+}
+
 
 void InterpreterSelectQuery::ignoreWithTotals()
 {

dbms/src/Interpreters/InterpreterSelectQuery.h

@@ -190,6 +190,9 @@ class InterpreterSelectQuery : public IInterpreter
     void executeExtremes(Pipeline & pipeline);
     void executeSubqueriesInSetsAndJoins(Pipeline & pipeline, std::unordered_map<String, SubqueryForSet> & subqueries_for_sets);
 
+    /// If pipeline has several streams with different headers, add ConvertingBlockInputStream to first header.
+    void unifyStreams(Pipeline & pipeline);
+
     enum class Modificator
     {
         ROLLUP = 0,

dbms/tests/integration/test_block_structure_mismatch/configs/remote_servers.xml

@@ -0,0 +1,18 @@
+<yandex>
+    <remote_servers>
+        <testcluster>
+            <shard>
+                <replica>
+                    <host>node1</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+            <shard>
+                <replica>
+                    <host>node2</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </testcluster>
+    </remote_servers>
+</yandex>

dbms/tests/integration/test_block_structure_mismatch/test.py

@@ -0,0 +1,50 @@
+import time
+import pytest
+
+from contextlib import contextmanager
+from helpers.cluster import ClickHouseCluster
+
+cluster = ClickHouseCluster(__file__)
+
+node1 = cluster.add_instance('node1', main_configs=['configs/remote_servers.xml'], with_zookeeper=True)
+node2 = cluster.add_instance('node2', main_configs=['configs/remote_servers.xml'], with_zookeeper=True)
+
+#test reproducing issue https://github.com/yandex/ClickHouse/issues/3162
+@pytest.fixture(scope="module")
+def started_cluster():
+    try:
+        cluster.start()
+
+        for node in (node1, node2):
+            node.query('''
+CREATE TABLE local_test (
+  t UInt64,
+  date Date MATERIALIZED toDate(t/1000),
+  shard UInt64,
+  col1 String,
+  col2 String
+) ENGINE = MergeTree
+PARTITION BY toRelativeDayNum(date)
+ORDER BY (t)
+SETTINGS index_granularity=8192
+            ''')
+
+            node.query('''
+CREATE TABLE dist_test (
+  t UInt64,
+  shard UInt64,
+  date Date MATERIALIZED toDate(t/1000),
+  col1 String,
+  col2 String
+) Engine = Distributed(testcluster, default, local_test, shard)
+            ''')
+
+        yield cluster
+
+    finally:
+        cluster.shutdown()
+
+def test(started_cluster):
+    node1.query("INSERT INTO dist_test (t, shard, col1, col2) VALUES (1000, 1, 'foo', 'bar'), (1000, 2, 'x', 'y')")
+    #time.sleep(3)
+    assert node1.query("SELECT col1, col2 FROM dist_test WHERE (t < 3600000) AND (col1 = 'foo') ORDER BY t ASC") == "foo\tbar\n"