Fix INSERT into Distributed() table with MATERIALIZED column

By just skipping MATERIALIZED columns during processing.

P.S. you cannot use insert_allow_materialized_columns since it works
only for Buffer() engine.

Fixes: ClickHouse#4015
Fixes: ClickHouse#3673
Fixes: 01501fa ("correct column list
for rewritten INSERT query into Distributed [#CLICKHOUSE-4161]")

dbms/src/Storages/StorageDistributed.cpp

@@ -84,17 +84,21 @@ ASTPtr rewriteSelectQuery(const ASTPtr & query, const std::string & database, co
 /// The columns list in the original INSERT query is incorrect because inserted blocks are transformed
 /// to the form of the sample block of the Distributed table. So we rewrite it and add all columns from
 /// the sample block instead.
-ASTPtr createInsertToRemoteTableQuery(const std::string & database, const std::string & table, const Block & sample_block)
+ASTPtr createInsertToRemoteTableQuery(const std::string & database, const std::string & table, const Block & sample_block, const ColumnsDescription & columns)
 {
     auto query = std::make_shared<ASTInsertQuery>();
     query->database = database;
     query->table = table;
 
-    auto columns = std::make_shared<ASTExpressionList>();
-    query->columns = columns;
-    query->children.push_back(columns);
+    auto block_columns = std::make_shared<ASTExpressionList>();
+    query->columns = block_columns;
+    query->children.push_back(block_columns);
     for (const auto & col : sample_block)
-        columns->children.push_back(std::make_shared<ASTIdentifier>(col.name));
+    {
+        if (columns.get(col.name).default_desc.kind == DB::ColumnDefaultKind::Materialized)
+            continue;
+        block_columns->children.push_back(std::make_shared<ASTIdentifier>(col.name));
+    }
 
     return query;
 }
@@ -331,9 +335,11 @@ BlockOutputStreamPtr StorageDistributed::write(const ASTPtr &, const Context & c
     bool insert_sync = settings.insert_distributed_sync || owned_cluster;
     auto timeout = settings.insert_distributed_timeout;
 
+    const ColumnsDescription & columns = getColumns();
+
     /// DistributedBlockOutputStream will not own cluster, but will own ConnectionPools of the cluster
     return std::make_shared<DistributedBlockOutputStream>(
-        context, *this, createInsertToRemoteTableQuery(remote_database, remote_table, getSampleBlock()), cluster,
+        context, *this, createInsertToRemoteTableQuery(remote_database, remote_table, getSampleBlock(), columns), cluster,
         insert_sync, timeout);
 }
 

dbms/tests/queries/0_stateless/00952_insert_into_distributed_with_materialized_column.reference

@@ -0,0 +1,3 @@
+2018-08-01
+2018-08-01
+2018-08-01	2017-08-01

dbms/tests/queries/0_stateless/00952_insert_into_distributed_with_materialized_column.sql

@@ -0,0 +1,12 @@
+DROP TABLE IF EXISTS test.test_local;
+DROP TABLE IF EXISTS test.test_distributed;
+
+CREATE TABLE test.test_local (date Date, value Date MATERIALIZED toDate('2017-08-01')) ENGINE = MergeTree(date, date, 8192);
+CREATE TABLE test.test_distributed AS test.test_local ENGINE = Distributed('test_shard_localhost', 'test', test_local, rand());
+
+SET insert_distributed_sync=1;
+
+INSERT INTO test.test_distributed VALUES ('2018-08-01');
+SELECT * FROM test.test_distributed;
+SELECT * FROM test.test_local;
+SELECT date, value FROM test.test_local;