dbms: development [#CONV-2944].

dbms/include/DB/Columns/ColumnConst.h

@@ -5,6 +5,7 @@
 #include <DB/Core/Field.h>
 #include <DB/Core/Exception.h>
 #include <DB/Core/ErrorCodes.h>
+#include <DB/Columns/ColumnVector.h>
 #include <DB/Columns/IColumn.h>
 
 
@@ -13,10 +14,19 @@ namespace DB
 
 using Poco::SharedPtr;
 
+
+class IColumnConst : public IColumn
+{
+public:
+	bool isConst() const { return true; }
+	virtual ColumnPtr convertToFullColumn() const = 0;
+};
+
+
 /** шаблон для столбцов-констант (столбцов одинаковых значений).
   */
 template <typename T>
-class ColumnConst : public IColumn
+class ColumnConst : public IColumnConst
 {
 public:
 	typedef T Type;
@@ -26,7 +36,6 @@ public:
 	std::string getName() const { return "ColumnConst<" + TypeName<T>::get() + ">"; }
 	bool isNumeric() const { return IsNumber<T>::value; }
 	size_t sizeOfField() const { return sizeof(T); }
-	bool isConst() const { return true; }
 	ColumnPtr cloneEmpty() const { return new ColumnConst(0, data); }
 	size_t size() const { return s; }
 	Field operator[](size_t n) const { return typename NearestFieldType<T>::Type(data); }
@@ -80,7 +89,7 @@ public:
 	const T & getData() const { return data; }
 
 	/** Преобразование из константы в полноценный столбец */
-//	virtual ColumnPtr convertToFullColumn() const = 0;
+	ColumnPtr convertToFullColumn() const;
 
 private:
 	size_t s;
@@ -90,4 +99,12 @@ private:
 
 typedef ColumnConst<String> ColumnConstString;
 
+
+template <typename T> ColumnPtr ColumnConst<T>::convertToFullColumn() const
+{
+	ColumnVector<T> * res = new ColumnVector<T>;
+	res->getData().assign(s, data);
+	return res;
+}
+
 }

dbms/include/DB/DataStreams/BlockInputStreamFromRowInputStream.h

@@ -21,6 +21,8 @@ public:
 		size_t max_block_size_ = DEFAULT_BLOCK_SIZE);
 
 	Block readImpl();
+	void readPrefix() { row_input->readPrefix(); }
+	void readSuffix() { row_input->readSuffix(); }
 
 	String getName() const { return "BlockInputStreamFromRowInputStream"; }
 

dbms/include/DB/DataStreams/BlockOutputStreamFromRowOutputStream.h

@@ -15,6 +15,8 @@ class BlockOutputStreamFromRowOutputStream : public IBlockOutputStream
 public:
 	BlockOutputStreamFromRowOutputStream(RowOutputStreamPtr row_output_);
 	void write(const Block & block);
+	void writePrefix() { row_output->writePrefix(); }
+	void writeSuffix() { row_output->writeSuffix(); }
 
 	BlockOutputStreamPtr clone() { return new BlockOutputStreamFromRowOutputStream(row_output); }
 

dbms/include/DB/DataStreams/MaterializingBlockInputStream.h

@@ -0,0 +1,48 @@
+#pragma once
+
+#include <DB/Columns/ColumnConst.h>
+
+#include <DB/DataStreams/IProfilingBlockInputStream.h>
+
+
+namespace DB
+{
+
+/** Преобразует столбцы-константы в полноценные столбцы ("материализует" их).
+  */
+class MaterializingBlockInputStream : public IProfilingBlockInputStream
+{
+public:
+	MaterializingBlockInputStream(BlockInputStreamPtr input_)
+		: input(input_)
+	{
+		children.push_back(input);
+	}
+
+	Block readImpl()
+	{
+		Block res = input->read();
+
+		if (!res)
+			return res;
+
+		size_t columns = res.columns();
+		for (size_t i = 0; i < columns; ++i)
+		{
+			ColumnPtr col = res.getByPosition(i).column;
+			if (col->isConst())
+				res.getByPosition(i).column = dynamic_cast<IColumnConst &>(*col).convertToFullColumn();
+		}
+
+		return res;
+	}
+
+	String getName() const { return "MaterializingBlockInputStream"; }
+
+	BlockInputStreamPtr clone() { return new MaterializingBlockInputStream(input); }
+
+private:
+	BlockInputStreamPtr input;
+};
+
+}

dbms/include/DB/DataStreams/TabSeparatedRowInputStream.h

@@ -18,15 +18,21 @@ using Poco::SharedPtr;
 class TabSeparatedRowInputStream : public IRowInputStream
 {
 public:
-	TabSeparatedRowInputStream(ReadBuffer & istr_, const Block & sample_);
+	/** with_names - в первой строке заголовок с именами столбцов
+	  * with_types - на следующей строке заголовок с именами типов
+	  */
+	TabSeparatedRowInputStream(ReadBuffer & istr_, const Block & sample_, bool with_names_ = false, bool with_types_ = false);
 
 	Row read();
+	void readPrefix();
 
 	RowInputStreamPtr clone() { return new TabSeparatedRowInputStream(istr, sample); }
 
 private:
 	ReadBuffer & istr;
 	const Block & sample;
+	bool with_names;
+	bool with_types;
 	DataTypes data_types;
 };
 

dbms/include/DB/DataStreams/TabSeparatedRowOutputStream.h

@@ -18,17 +18,23 @@ using Poco::SharedPtr;
 class TabSeparatedRowOutputStream : public IRowOutputStream
 {
 public:
-	TabSeparatedRowOutputStream(WriteBuffer & ostr_, const Block & sample_);
+	/** with_names - выводить в первой строке заголовок с именами столбцов
+	  * with_types - выводить на следующей строке заголовок с именами типов
+	  */
+	TabSeparatedRowOutputStream(WriteBuffer & ostr_, const Block & sample_, bool with_names_ = false, bool with_types_ = false);
 
 	void writeField(const Field & field);
 	void writeFieldDelimiter();
 	void writeRowEndDelimiter();
+	void writePrefix();
 
 	RowOutputStreamPtr clone() { return new TabSeparatedRowOutputStream(ostr, sample); }
 
 private:
 	WriteBuffer & ostr;
 	const Block & sample;
+	bool with_names;
+	bool with_types;
 	DataTypes data_types;
 	size_t field_number;
 };

dbms/include/DB/DataStreams/copyData.h

@@ -1,5 +1,4 @@
-#ifndef DBMS_DATA_STREAMS_COPY_DATA_H
+#pragma once
-#define DBMS_DATA_STREAMS_COPY_DATA_H
 
 #include <DB/Core/Block.h>
 #include <DB/DataStreams/IBlockInputStream.h>
@@ -20,5 +19,3 @@ void copyData(IBlockInputStream & from, IRowOutputStream & to);
 void copyData(IRowInputStream & from, IBlockOutputStream & to, const Block & sample);
 
 }
-
-#endif

dbms/src/Columns/ColumnConst.cpp

@@ -0,0 +1,29 @@
+#include <DB/Columns/ColumnString.h>
+#include <DB/Columns/ColumnConst.h>
+
+
+namespace DB
+{
+
+template <> ColumnPtr ColumnConst<String>::convertToFullColumn() const
+{
+	ColumnString * res = new ColumnString;
+	ColumnString::Offsets_t & offsets = res->getOffsets();
+	ColumnUInt8::Container_t & vec = dynamic_cast<ColumnVector<UInt8> &>(res->getData()).getData();
+
+	size_t string_size = data.size() + 1;
+	size_t offset = 0;
+	offsets.resize(s);
+	vec.resize(s * string_size);
+
+	for (size_t i = 0; i < s; ++i)
+	{
+		memcpy(&vec[offset], data.data(), string_size);
+		offset += string_size;
+		offsets[i] = offset;
+	}
+
+	return res;
+}
+
+}

dbms/src/DataStreams/BlockInputStreamFromRowInputStream.cpp

@@ -23,7 +23,6 @@ Block BlockInputStreamFromRowInputStream::readImpl()
 {
 	Block res;
 
-	row_input->readPrefix();
 	for (size_t rows = 0; rows < max_block_size; ++rows)
 	{
 		if (rows != 0)
@@ -43,7 +42,6 @@ Block BlockInputStreamFromRowInputStream::readImpl()
 		for (size_t i = 0; i < row.size(); ++i)
 			res.getByPosition(i).column->insert(row[i]);
 	}
-	row_input->readSuffix();
 
 	return res;
 }

dbms/src/DataStreams/BlockOutputStreamFromRowOutputStream.cpp

@@ -12,7 +12,6 @@ void BlockOutputStreamFromRowOutputStream::write(const Block & block)
 	size_t rows = block.rows();
 	size_t columns = block.columns();
 
-	row_output->writePrefix();
 	for (size_t i = 0; i < rows; ++i)
 	{
 		if (i != 0)
@@ -29,7 +28,6 @@ void BlockOutputStreamFromRowOutputStream::write(const Block & block)
 		
 		row_output->writeRowEndDelimiter();
 	}
-	row_output->writeSuffix();
 }
 
 }

dbms/src/DataStreams/FormatFactory.cpp

@@ -4,6 +4,7 @@
 #include <DB/DataStreams/TabSeparatedRowOutputStream.h>
 #include <DB/DataStreams/ValuesRowInputStream.h>
 #include <DB/DataStreams/ValuesRowOutputStream.h>
+#include <DB/DataStreams/TabSeparatedBlockOutputStream.h>
 #include <DB/DataStreams/BlockInputStreamFromRowInputStream.h>
 #include <DB/DataStreams/BlockOutputStreamFromRowOutputStream.h>
 #include <DB/DataStreams/FormatFactory.h>
@@ -19,6 +20,10 @@ BlockInputStreamPtr FormatFactory::getInput(const String & name, ReadBuffer & bu
 		return new NativeBlockInputStream(buf, data_type_factory);
 	else if (name == "TabSeparated")
 		return new BlockInputStreamFromRowInputStream(new TabSeparatedRowInputStream(buf, sample), sample, max_block_size);
+	else if (name == "TabSeparatedWithNames")
+		return new BlockInputStreamFromRowInputStream(new TabSeparatedRowInputStream(buf, sample, true), sample, max_block_size);
+	else if (name == "TabSeparatedWithNamesAndTypes")
+		return new BlockInputStreamFromRowInputStream(new TabSeparatedRowInputStream(buf, sample, true, true), sample, max_block_size);
 	else if (name == "Values")
 		return new BlockInputStreamFromRowInputStream(new ValuesRowInputStream(buf, sample), sample, max_block_size);
 	else
@@ -33,6 +38,12 @@ BlockOutputStreamPtr FormatFactory::getOutput(const String & name, WriteBuffer &
 		return new NativeBlockOutputStream(buf);
 	else if (name == "TabSeparated")
 		return new BlockOutputStreamFromRowOutputStream(new TabSeparatedRowOutputStream(buf, sample));
+	else if (name == "TabSeparatedWithNames")
+		return new BlockOutputStreamFromRowOutputStream(new TabSeparatedRowOutputStream(buf, sample, true));
+	else if (name == "TabSeparatedWithNamesAndTypes")
+		return new BlockOutputStreamFromRowOutputStream(new TabSeparatedRowOutputStream(buf, sample, true, true));
+	else if (name == "BlockTabSeparated")
+		return new TabSeparatedBlockOutputStream(buf);
 	else if (name == "Values")
 		return new BlockOutputStreamFromRowOutputStream(new ValuesRowOutputStream(buf, sample));
 	else

dbms/src/DataStreams/TabSeparatedRowInputStream.cpp

@@ -8,8 +8,8 @@ namespace DB
 
 using Poco::SharedPtr;
 
-TabSeparatedRowInputStream::TabSeparatedRowInputStream(ReadBuffer & istr_, const Block & sample_)
+TabSeparatedRowInputStream::TabSeparatedRowInputStream(ReadBuffer & istr_, const Block & sample_, bool with_names_, bool with_types_)
-	: istr(istr_), sample(sample_)
+	: istr(istr_), sample(sample_), with_names(with_names_), with_types(with_types_)
 {
 	size_t columns = sample.columns();
 	data_types.resize(columns);
@@ -18,6 +18,31 @@ TabSeparatedRowInputStream::TabSeparatedRowInputStream(ReadBuffer & istr_, const
 }
 
 
+void TabSeparatedRowInputStream::readPrefix()
+{
+	size_t columns = sample.columns();
+	String tmp;
+
+	if (with_names)
+	{
+		for (size_t i = 0; i < columns; ++i)
+		{
+			readEscapedString(tmp, istr);
+			assertString(i == columns - 1 ? "\n" : "\t", istr);
+		}
+	}
+
+	if (with_types)
+	{
+		for (size_t i = 0; i < columns; ++i)
+		{
+			readEscapedString(tmp, istr);
+			assertString(i == columns - 1 ? "\n" : "\t", istr);
+		}
+	}
+}
+
+
 Row TabSeparatedRowInputStream::read()
 {
 	Row res;

dbms/src/DataStreams/TabSeparatedRowOutputStream.cpp

@@ -9,8 +9,8 @@ namespace DB
 using Poco::SharedPtr;
 
 
-TabSeparatedRowOutputStream::TabSeparatedRowOutputStream(WriteBuffer & ostr_, const Block & sample_)
+TabSeparatedRowOutputStream::TabSeparatedRowOutputStream(WriteBuffer & ostr_, const Block & sample_, bool with_names_, bool with_types_)
-	: ostr(ostr_), sample(sample_), field_number(0)
+	: ostr(ostr_), sample(sample_), with_names(with_names_), with_types(with_types_), field_number(0)
 {
 	size_t columns = sample.columns();
 	data_types.resize(columns);
@@ -19,6 +19,30 @@ TabSeparatedRowOutputStream::TabSeparatedRowOutputStream(WriteBuffer & ostr_, co
 }
 
 
+void TabSeparatedRowOutputStream::writePrefix()
+{
+	size_t columns = sample.columns();
+	
+	if (with_names)
+	{
+		for (size_t i = 0; i < columns; ++i)
+		{
+			writeEscapedString(sample.getByPosition(i).name, ostr);
+			writeChar(i == columns - 1 ? '\n' : '\t', ostr);
+		}
+	}
+
+	if (with_types)
+	{
+		for (size_t i = 0; i < columns; ++i)
+		{
+			writeEscapedString(sample.getByPosition(i).type->getName(), ostr);
+			writeChar(i == columns - 1 ? '\n' : '\t', ostr);
+		}
+	}
+}
+
+
 void TabSeparatedRowOutputStream::writeField(const Field & field)
 {
 	data_types[field_number]->serializeTextEscaped(field, ostr);

dbms/src/Interpreters/InterpreterCreateQuery.cpp

@@ -6,6 +6,7 @@
 #include <DB/IO/WriteBufferFromString.h>
 #include <DB/IO/WriteHelpers.h>
 
+#include <DB/DataStreams/MaterializingBlockInputStream.h>
 #include <DB/DataStreams/copyData.h>
 
 #include <DB/Parsers/ASTCreateQuery.h>
@@ -167,7 +168,10 @@ StoragePtr InterpreterCreateQuery::execute()
 
 	/// Если запрос CREATE SELECT, то вставим в таблицу данные
 	if (create.select)
-		copyData(*interpreter_select->execute(), *res->write(query_ptr));
+	{
+		BlockInputStreamPtr from = new MaterializingBlockInputStream(interpreter_select->execute());
+		copyData(*from, *res->write(query_ptr));
+	}
 	
 	return res;
 }

dbms/src/Interpreters/InterpreterInsertQuery.cpp

@@ -1,5 +1,6 @@
 #include <DB/IO/ConcatReadBuffer.h>
 
+#include <DB/DataStreams/MaterializingBlockInputStream.h>
 #include <DB/DataStreams/FormatFactory.h>
 #include <DB/DataStreams/copyData.h>
 
@@ -79,6 +80,7 @@ void InterpreterInsertQuery::execute(SharedPtr<ReadBuffer> remaining_data_istr)
 	{
 		InterpreterSelectQuery interpreter_select(query.select, context, max_block_size);
 		in = interpreter_select.execute();
+		in = new MaterializingBlockInputStream(in);
 		copyData(*in, *out);
 	}
 }