From 78c982e205892f73accd8306f14e050778ec375f Mon Sep 17 00:00:00 2001
From: Alexey Milovidov <milovidov@yandex-team.ru>
Date: Mon, 24 May 2010 16:52:58 +0000
Subject: [PATCH] dbms: development.

---
 dbms/include/DB/Columns/ColumnString.h        |  2 +-
 .../BlockInputStreamFromRowInputStream.h      |  2 +
 .../BlockInputStreamFromRowInputStream.cpp    | 18 ++++--
 .../tests/block_row_transforms.cpp            | 59 +++++++++++++++++++
 4 files changed, 75 insertions(+), 6 deletions(-)
 create mode 100644 dbms/src/DataStreams/tests/block_row_transforms.cpp

diff --git a/dbms/include/DB/Columns/ColumnString.h b/dbms/include/DB/Columns/ColumnString.h
index c3e08882c4f..9c50e93b547 100644
--- a/dbms/include/DB/Columns/ColumnString.h
+++ b/dbms/include/DB/Columns/ColumnString.h
@@ -41,7 +41,7 @@ public:
 
 	void insert(const Field & x)
 	{
-		String & s = boost::get<String &>(x);
+		const String & s = boost::get<const String &>(x);
 		size_t old_size = char_data.size();
 		size_t size_to_append = s.size() + 1;
 		char_data.resize(old_size + size_to_append);
diff --git a/dbms/include/DB/DataStreams/BlockInputStreamFromRowInputStream.h b/dbms/include/DB/DataStreams/BlockInputStreamFromRowInputStream.h
index 51ccd6e53e1..17a285800ca 100644
--- a/dbms/include/DB/DataStreams/BlockInputStreamFromRowInputStream.h
+++ b/dbms/include/DB/DataStreams/BlockInputStreamFromRowInputStream.h
@@ -33,6 +33,8 @@ private:
 	IRowInputStream & row_input;
 	const Block & sample;
 	size_t max_block_size;
+
+	void initBlock(Block & res);
 };
 
 }
diff --git a/dbms/src/DataStreams/BlockInputStreamFromRowInputStream.cpp b/dbms/src/DataStreams/BlockInputStreamFromRowInputStream.cpp
index c1a58ae346e..0b4a03ce216 100644
--- a/dbms/src/DataStreams/BlockInputStreamFromRowInputStream.cpp
+++ b/dbms/src/DataStreams/BlockInputStreamFromRowInputStream.cpp
@@ -3,6 +3,7 @@
 
 #include <DB/DataStreams/BlockInputStreamFromRowInputStream.h>
 
+#include <iostream>
 
 namespace DB
 {
@@ -19,10 +20,8 @@ BlockInputStreamFromRowInputStream::BlockInputStreamFromRowInputStream(
 }
 
 
-Block BlockInputStreamFromRowInputStream::read()
+void BlockInputStreamFromRowInputStream::initBlock(Block & res)
 {
-	Block res;
-
 	for (size_t i = 0; i < sample.columns(); ++i)
 	{
 		const ColumnWithNameAndType & sample_elem = sample.getByPosition(i);
@@ -34,13 +33,22 @@ Block BlockInputStreamFromRowInputStream::read()
 
 		res.insert(res_elem);
 	}
+}
+
+
+Block BlockInputStreamFromRowInputStream::read()
+{
+	Block res;
 
 	for (size_t rows = 0; rows < max_block_size; ++rows)
 	{
 		Row row = row_input.read();
 
 		if (row.empty())
-			return sample;
+			return res;
+
+		if (!res)
+			initBlock(res);
 
 		if (row.size() != sample.columns())
 			throw Exception("Number of columns doesn't match", ErrorCodes::NUMBER_OF_COLUMNS_DOESNT_MATCH);
@@ -48,7 +56,7 @@ Block BlockInputStreamFromRowInputStream::read()
 		for (size_t i = 0; i < row.size(); ++i)
 			res.getByPosition(i).column->insert(row[i]);
 	}
-	
+
 	return res;
 }
 
diff --git a/dbms/src/DataStreams/tests/block_row_transforms.cpp b/dbms/src/DataStreams/tests/block_row_transforms.cpp
new file mode 100644
index 00000000000..eee7ba73fb9
--- /dev/null
+++ b/dbms/src/DataStreams/tests/block_row_transforms.cpp
@@ -0,0 +1,59 @@
+#include <string>
+
+#include <iostream>
+#include <fstream>
+
+#include <Poco/Stopwatch.h>
+#include <Poco/SharedPtr.h>
+
+#include <DB/Core/Block.h>
+#include <DB/Core/ColumnWithNameAndType.h>
+
+#include <DB/DataTypes/DataTypesNumberFixed.h>
+#include <DB/DataTypes/DataTypeString.h>
+
+#include <DB/DataStreams/TabSeparatedRowInputStream.h>
+#include <DB/DataStreams/TabSeparatedRowOutputStream.h>
+#include <DB/DataStreams/BlockInputStreamFromRowInputStream.h>
+#include <DB/DataStreams/copyData.h>
+
+
+int main(int argc, char ** argv)
+{
+	try
+	{
+		Poco::SharedPtr<DB::DataTypes> data_types = new DB::DataTypes;
+		data_types->push_back(new DB::DataTypeUInt64);
+		data_types->push_back(new DB::DataTypeString);
+
+		DB::Block sample;
+
+		DB::ColumnWithNameAndType col1;
+		col1.name = "col1";
+		col1.type = data_types->at(0);
+		col1.column = col1.type->createColumn();
+		sample.insert(col1);
+
+		DB::ColumnWithNameAndType col2;
+		col2.name = "col2";
+		col2.type = data_types->at(1);
+		col2.column = col2.type->createColumn();
+		sample.insert(col2);
+
+		std::ifstream istr("test_in");
+		std::ofstream ostr("test_out");
+
+		DB::TabSeparatedRowInputStream row_input(istr, data_types);
+		DB::BlockInputStreamFromRowInputStream block_input(row_input, sample);
+		DB::TabSeparatedRowOutputStream row_output(ostr, data_types);
+
+		DB::copyData(block_input, row_output);
+	}
+	catch (const DB::Exception & e)
+	{
+		std::cerr << e.what() << ", " << e.message() << std::endl;
+		return 1;
+	}
+
+	return 0;
+}