dbms: MATERIALIZED columns evaluation during insert, error handling. #[METR-12739]

dbms/include/DB/DataStreams/MaterializingBlockOutputStream.h

@@ -0,0 +1,37 @@
+#pragma once
+
+#include <DB/Columns/ColumnConst.h>
+#include <DB/DataStreams/IBlockOutputStream.h>
+#include <statdaemons/ext/range.hpp>
+
+
+namespace DB
+{
+
+/** Преобразует столбцы-константы в полноценные столбцы ("материализует" их).
+  */
+class MaterializingBlockOutputStream : public IBlockOutputStream
+{
+public:
+	MaterializingBlockOutputStream(const BlockOutputStreamPtr & output) : output{output}
+	{}
+
+	void write(const Block & original_block) override
+	{
+		/// copy block to get rid of const
+		auto block = original_block;
+
+		for (const auto i : ext::range(0, block.columns()))
+		{
+			ColumnPtr & col = block.getByPosition(i).column;
+			if (col->isConst())
+				col = dynamic_cast<IColumnConst &>(*col).convertToFullColumn();
+		}
+
+		output->write(block);
+	}
+
+	BlockOutputStreamPtr output;
+};
+
+}

dbms/include/DB/DataStreams/ProhibitColumnsBlockOutputStream.h

@@ -11,11 +11,12 @@ namespace DB
 class ProhibitColumnsBlockOutputStream : public IBlockOutputStream
 {
 public:
-	ProhibitColumnsBlockOutputStream(BlockOutputStreamPtr output, const NamesAndTypesList & columns)
+	ProhibitColumnsBlockOutputStream(const BlockOutputStreamPtr & output, const NamesAndTypesList & columns)
 		: output{output}, columns{columns}
 	{
 	}
 
+private:
 	void write(const Block & block) override
 	{
         for (const auto & column : columns)
@@ -25,12 +26,11 @@ public:
 		output->write(block);
 	}
 
-	void flush() { output->flush(); }
+	void flush() override { output->flush(); }
 
 	void writePrefix() override { output->writePrefix(); }
 	void writeSuffix() override { output->writeSuffix(); }
 
-private:
 	BlockOutputStreamPtr output;
 	NamesAndTypesList columns;
 };

dbms/include/DB/Functions/FunctionsMiscellaneous.h

@@ -123,7 +123,8 @@ public:
 		return new DataTypeString;
 	}
 
-	/// Выполнить функцию над блоком.
+	/** Выполнить функцию над блоком. convertToFullColumn вызывается для того, чтобы в случае
+	 *	распределенного выполнения запроса каждый сервер возвращал свое имя хоста. */
 	void execute(Block & block, const ColumnNumbers & arguments, size_t result)
 	{
 		block.getByPosition(result).column = ColumnConstString(

dbms/include/DB/Storages/ITableDeclaration.h

@@ -54,6 +54,7 @@ public:
 	/** То же самое, но в виде блока-образца.
 	  */
 	Block getSampleBlock() const;
+	Block getSampleBlockNonMaterialized() const;
 
 	/** Проверить, что все запрошенные имена есть в таблице и заданы корректно.
 	  * (список имён не пустой и имена не повторяются)

dbms/src/Interpreters/InterpreterInsertQuery.cpp

@@ -1,8 +1,8 @@
 #include <DB/IO/ConcatReadBuffer.h>
 
 #include <DB/DataStreams/ProhibitColumnsBlockOutputStream.h>
+#include <DB/DataStreams/MaterializingBlockOutputStream.h>
 #include <DB/DataStreams/AddingDefaultBlockOutputStream.h>
-#include <DB/DataStreams/MaterializingBlockInputStream.h>
 #include <DB/DataStreams/PushingToViewsBlockOutputStream.h>
 #include <DB/DataStreams/copyData.h>
 
@@ -35,11 +35,12 @@ StoragePtr InterpreterInsertQuery::getTable()
 Block InterpreterInsertQuery::getSampleBlock()
 {
 	ASTInsertQuery & query = typeid_cast<ASTInsertQuery &>(*query_ptr);
-	Block db_sample = getTable()->getSampleBlock();
 
 	/// Если в запросе не указана информация о столбцах
 	if (!query.columns)
-		return db_sample;
+		return getTable()->getSampleBlockNonMaterialized();
+
+	Block db_sample = getTable()->getSampleBlock();
 
 	/// Формируем блок, основываясь на именах столбцов из запроса
 	Block res;
@@ -70,7 +71,7 @@ void InterpreterInsertQuery::execute(ReadBuffer * remaining_data_istr)
 
 	/** @note looks suspicious, first we ask to create block from NamesAndTypesList (internally in ITableDeclaration),
 	 *  then we compose the same list from the resulting block */
-	NamesAndTypesListPtr required_columns = new NamesAndTypesList(table->getSampleBlock().getColumnsList());
+	NamesAndTypesListPtr required_columns = new NamesAndTypesList(table->getColumnsList());
 
 	/// Надо убедиться, что запрос идет в таблицу, которая поддерживает вставку.
 	/// TODO Плохо - исправить.
@@ -80,9 +81,12 @@ void InterpreterInsertQuery::execute(ReadBuffer * remaining_data_istr)
 	BlockOutputStreamPtr out{
 		new ProhibitColumnsBlockOutputStream{
 			new AddingDefaultBlockOutputStream{
-				new PushingToViewsBlockOutputStream{query.database, query.table, context, query_ptr},
+				new MaterializingBlockOutputStream{
+					new PushingToViewsBlockOutputStream{query.database, query.table, context, query_ptr}
+				},
 				required_columns, table->column_defaults, context
-			}, table->materialized_columns
+			},
+			table->materialized_columns
 		}
 	};
 
@@ -121,7 +125,6 @@ void InterpreterInsertQuery::execute(ReadBuffer * remaining_data_istr)
 	{
 		InterpreterSelectQuery interpreter_select(query.select, context);
 		BlockInputStreamPtr in{interpreter_select.execute()};
-		in = new MaterializingBlockInputStream(in);
 
 		copyData(*in, *out);
 	}
@@ -135,7 +138,7 @@ BlockIO InterpreterInsertQuery::execute()
 
 	auto table_lock = table->lockStructure(true);
 
-	NamesAndTypesListPtr required_columns = new NamesAndTypesList(table->getSampleBlock().getColumnsList());
+	NamesAndTypesListPtr required_columns = new NamesAndTypesList(table->getColumnsList());
 
 	/// Надо убедиться, что запрос идет в таблицу, которая поддерживает вставку.
 	/// TODO Плохо - исправить.
@@ -145,9 +148,12 @@ BlockIO InterpreterInsertQuery::execute()
 	BlockOutputStreamPtr out{
 		new ProhibitColumnsBlockOutputStream{
 			new AddingDefaultBlockOutputStream{
-				new PushingToViewsBlockOutputStream{query.database, query.table, context, query_ptr},
+				new MaterializingBlockOutputStream{
+					new PushingToViewsBlockOutputStream{query.database, query.table, context, query_ptr}
+				},
 				required_columns, table->column_defaults, context
-			}, table->materialized_columns
+			},
+			table->materialized_columns
 		}
 	};
 
@@ -163,7 +169,6 @@ BlockIO InterpreterInsertQuery::execute()
 	{
 		InterpreterSelectQuery interpreter_select(query.select, context);
 		BlockInputStreamPtr in = interpreter_select.execute();
-		in = new MaterializingBlockInputStream(in);
 		copyData(*in, *out);
 	}
 

dbms/src/Storages/ITableDeclaration.cpp

@@ -84,10 +84,6 @@ const DataTypePtr ITableDeclaration::getDataTypeByName(const String & column_nam
 		if (column.name == column_name)
 			return column.type;
 
-	for (const auto & column : materialized_columns)
-		if (column.name == column_name)
-			return column.type;
-
 	throw Exception("There is no column " + column_name + " in table.", ErrorCodes::NO_SUCH_COLUMN_IN_TABLE);
 }
 
@@ -95,16 +91,20 @@ const DataTypePtr ITableDeclaration::getDataTypeByName(const String & column_nam
 Block ITableDeclaration::getSampleBlock() const
 {
 	Block res;
-	const NamesAndTypesList & names_and_types = getColumnsList();
 
-	for (NamesAndTypesList::const_iterator it = names_and_types.begin(); it != names_and_types.end(); ++it)
-	{
-		ColumnWithNameAndType col;
-		col.name = it->name;
-		col.type = it->type;
-		col.column = col.type->createColumn();
-		res.insert(col);
-	}
+	for (const auto & col : getColumnsList())
+		res.insert({ col.type->createColumn(), col.type, col.name });
+
+	return res;
+}
+
+
+Block ITableDeclaration::getSampleBlockNonMaterialized() const
+{
+	Block res;
+
+	for (const auto & col : getColumnsListAsterisk())
+		res.insert({ col.type->createColumn(), col.type, col.name });
 
 	return res;
 }

dbms/tests/queries/0_stateless/00030_1_alter_table.reference

@@ -1,15 +1,15 @@
-CounterID	UInt32
-StartDate	Date
-UserID	UInt32
-VisitID	UInt32
-NestedColumn.A	Array(UInt8)
-NestedColumn.S	Array(String)
-ToDrop	UInt32
-Added0	UInt32
-Added1	UInt32
-Added2	UInt32
-AddedNested1.A	Array(UInt32)
-AddedNested1.B	Array(UInt64)
-AddedNested1.C	Array(String)
-AddedNested2.A	Array(UInt32)
-AddedNested2.B	Array(UInt64)
+CounterID	UInt32		
+StartDate	Date		
+UserID	UInt32		
+VisitID	UInt32		
+NestedColumn.A	Array(UInt8)		
+NestedColumn.S	Array(String)		
+ToDrop	UInt32		
+Added0	UInt32		
+Added1	UInt32		
+Added2	UInt32		
+AddedNested1.A	Array(UInt32)		
+AddedNested1.B	Array(UInt64)		
+AddedNested1.C	Array(String)		
+AddedNested2.A	Array(UInt32)		
+AddedNested2.B	Array(UInt64)		

dbms/tests/queries/0_stateless/00030_2_alter_table.reference

@@ -1,11 +1,11 @@
-CounterID	UInt32
-StartDate	Date
-UserID	UInt32
-VisitID	UInt32
-Added0	String
-Added1	UInt32
-Added2	UInt32
-AddedNested1.A	Array(UInt32)
-AddedNested1.C	Array(String)
-AddedNested2.A	Array(UInt32)
-AddedNested2.B	Array(UInt64)
+CounterID	UInt32		
+StartDate	Date		
+UserID	UInt32		
+VisitID	UInt32		
+Added0	String		
+Added1	UInt32		
+Added2	UInt32		
+AddedNested1.A	Array(UInt32)		
+AddedNested1.C	Array(String)		
+AddedNested2.A	Array(UInt32)		
+AddedNested2.B	Array(UInt64)		

dbms/tests/queries/0_stateless/00061_merge_tree_alter.reference

@@ -1,72 +1,72 @@
-d	Date
-k	UInt64
-i32	Int32
+d	Date		
+k	UInt64		
+i32	Int32		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	10	42
-d	Date
-k	UInt64
-i32	Int32
-dt	DateTime
+d	Date		
+k	UInt64		
+i32	Int32		
+dt	DateTime		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt DateTime) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	9	41	1992-01-01 08:00:00
 2015-01-01	10	42	0000-00-00 00:00:00
-d	Date
-k	UInt64
-i32	Int32
-dt	DateTime
-n.ui8	Array(UInt8)
-n.s	Array(String)
+d	Date		
+k	UInt64		
+i32	Int32		
+dt	DateTime		
+n.ui8	Array(UInt8)		
+n.s	Array(String)		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt DateTime,  `n.ui8` Array(UInt8),  `n.s` Array(String)) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	8	40	2012-12-12 12:12:12	[1,2,3]	['12','13','14']
 2015-01-01	9	41	1992-01-01 08:00:00	[]	[]
 2015-01-01	10	42	0000-00-00 00:00:00	[]	[]
-d	Date
-k	UInt64
-i32	Int32
-dt	DateTime
-n.ui8	Array(UInt8)
-n.s	Array(String)
-n.d	Array(Date)
+d	Date		
+k	UInt64		
+i32	Int32		
+dt	DateTime		
+n.ui8	Array(UInt8)		
+n.s	Array(String)		
+n.d	Array(Date)		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt DateTime,  `n.ui8` Array(UInt8),  `n.s` Array(String),  `n.d` Array(Date)) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	7	39	2014-07-14 13:26:50	[10,20,30]	['120','130','140']	['2000-01-01','2000-01-01','2000-01-03']
 2015-01-01	8	40	2012-12-12 12:12:12	[1,2,3]	['12','13','14']	['0000-00-00','0000-00-00','0000-00-00']
 2015-01-01	9	41	1992-01-01 08:00:00	[]	[]	[]
 2015-01-01	10	42	0000-00-00 00:00:00	[]	[]	[]
-d	Date
-k	UInt64
-i32	Int32
-dt	DateTime
-n.ui8	Array(UInt8)
-n.s	Array(String)
-n.d	Array(Date)
-s	String
+d	Date		
+k	UInt64		
+i32	Int32		
+dt	DateTime		
+n.ui8	Array(UInt8)		
+n.s	Array(String)		
+n.d	Array(Date)		
+s	String		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt DateTime,  `n.ui8` Array(UInt8),  `n.s` Array(String),  `n.d` Array(Date),  s String) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	6	38	2014-07-15 13:26:50	[10,20,30]	['asd','qwe','qwe']	['2000-01-01','2000-01-01','2000-01-03']	100500
 2015-01-01	7	39	2014-07-14 13:26:50	[10,20,30]	['120','130','140']	['2000-01-01','2000-01-01','2000-01-03']	
 2015-01-01	8	40	2012-12-12 12:12:12	[1,2,3]	['12','13','14']	['0000-00-00','0000-00-00','0000-00-00']	
 2015-01-01	9	41	1992-01-01 08:00:00	[]	[]	[]	
 2015-01-01	10	42	0000-00-00 00:00:00	[]	[]	[]	
-d	Date
-k	UInt64
-i32	Int32
-dt	DateTime
-n.ui8	Array(UInt8)
-n.s	Array(String)
-s	Int64
+d	Date		
+k	UInt64		
+i32	Int32		
+dt	DateTime		
+n.ui8	Array(UInt8)		
+n.s	Array(String)		
+s	Int64		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt DateTime,  `n.ui8` Array(UInt8),  `n.s` Array(String),  s Int64) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	6	38	2014-07-15 13:26:50	[10,20,30]	['asd','qwe','qwe']	100500
 2015-01-01	7	39	2014-07-14 13:26:50	[10,20,30]	['120','130','140']	0
 2015-01-01	8	40	2012-12-12 12:12:12	[1,2,3]	['12','13','14']	0
 2015-01-01	9	41	1992-01-01 08:00:00	[]	[]	0
 2015-01-01	10	42	0000-00-00 00:00:00	[]	[]	0
-d	Date
-k	UInt64
-i32	Int32
-dt	DateTime
-n.ui8	Array(UInt8)
-n.s	Array(String)
-s	UInt32
-n.d	Array(Date)
+d	Date		
+k	UInt64		
+i32	Int32		
+dt	DateTime		
+n.ui8	Array(UInt8)		
+n.s	Array(String)		
+s	UInt32		
+n.d	Array(Date)		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt DateTime,  `n.ui8` Array(UInt8),  `n.s` Array(String),  s UInt32,  `n.d` Array(Date)) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	6	38	2014-07-15 13:26:50	[10,20,30]	['asd','qwe','qwe']	100500	['0000-00-00','0000-00-00','0000-00-00']
 2015-01-01	7	39	2014-07-14 13:26:50	[10,20,30]	['120','130','140']	0	['0000-00-00','0000-00-00','0000-00-00']
@@ -78,58 +78,58 @@ CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt DateTime,  `n.ui8`
 2015-01-01	8	40	2012-12-12 12:12:12	[1,2,3]	['12','13','14']	0	['0000-00-00','0000-00-00','0000-00-00']
 2015-01-01	9	41	1992-01-01 08:00:00	[]	[]	0	[]
 2015-01-01	10	42	0000-00-00 00:00:00	[]	[]	0	[]
-d	Date
-k	UInt64
-i32	Int32
-dt	DateTime
-n.s	Array(String)
-s	UInt32
+d	Date		
+k	UInt64		
+i32	Int32		
+dt	DateTime		
+n.s	Array(String)		
+s	UInt32		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt DateTime,  `n.s` Array(String),  s UInt32) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	6	38	2014-07-15 13:26:50	['asd','qwe','qwe']	100500
 2015-01-01	7	39	2014-07-14 13:26:50	['120','130','140']	0
 2015-01-01	8	40	2012-12-12 12:12:12	['12','13','14']	0
 2015-01-01	9	41	1992-01-01 08:00:00	[]	0
 2015-01-01	10	42	0000-00-00 00:00:00	[]	0
-d	Date
-k	UInt64
-i32	Int32
-dt	DateTime
-s	UInt32
+d	Date		
+k	UInt64		
+i32	Int32		
+dt	DateTime		
+s	UInt32		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt DateTime,  s UInt32) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	6	38	2014-07-15 13:26:50	100500
 2015-01-01	7	39	2014-07-14 13:26:50	0
 2015-01-01	8	40	2012-12-12 12:12:12	0
 2015-01-01	9	41	1992-01-01 08:00:00	0
 2015-01-01	10	42	0000-00-00 00:00:00	0
-d	Date
-k	UInt64
-i32	Int32
-dt	DateTime
-s	UInt32
-n.s	Array(String)
-n.d	Array(Date)
+d	Date		
+k	UInt64		
+i32	Int32		
+dt	DateTime		
+s	UInt32		
+n.s	Array(String)		
+n.d	Array(Date)		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt DateTime,  s UInt32,  `n.s` Array(String),  `n.d` Array(Date)) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	6	38	2014-07-15 13:26:50	100500	[]	[]
 2015-01-01	7	39	2014-07-14 13:26:50	0	[]	[]
 2015-01-01	8	40	2012-12-12 12:12:12	0	[]	[]
 2015-01-01	9	41	1992-01-01 08:00:00	0	[]	[]
 2015-01-01	10	42	0000-00-00 00:00:00	0	[]	[]
-d	Date
-k	UInt64
-i32	Int32
-dt	DateTime
-s	UInt32
+d	Date		
+k	UInt64		
+i32	Int32		
+dt	DateTime		
+s	UInt32		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt DateTime,  s UInt32) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	6	38	2014-07-15 13:26:50	100500
 2015-01-01	7	39	2014-07-14 13:26:50	0
 2015-01-01	8	40	2012-12-12 12:12:12	0
 2015-01-01	9	41	1992-01-01 08:00:00	0
 2015-01-01	10	42	0000-00-00 00:00:00	0
-d	Date
-k	UInt64
-i32	Int32
-dt	Date
-s	DateTime
+d	Date		
+k	UInt64		
+i32	Int32		
+dt	Date		
+s	DateTime		
 CREATE TABLE test.alter ( d Date,  k UInt64,  i32 Int32,  dt Date,  s DateTime) ENGINE = MergeTree(d, k, 8192)
 2015-01-01	6	38	2014-07-15	1970-01-02 06:55:00
 2015-01-01	7	39	2014-07-14	0000-00-00 00:00:00