dbms: less copying when reading fixed columns in large blocks [#CONV-2944].

dbms/include/DB/DataTypes/IDataTypeNumberFixed.h

@@ -55,7 +55,7 @@ public:
 	{
 		typename ColumnType::Container_t & x =  dynamic_cast<ColumnType &>(column).getData();
 		x.resize(limit);
-		size_t size = istr.read(reinterpret_cast<char*>(&x[0]), sizeof(typename ColumnType::value_type) * limit);
+		size_t size = istr.readBig(reinterpret_cast<char*>(&x[0]), sizeof(typename ColumnType::value_type) * limit);
 		x.resize(size / sizeof(typename ColumnType::value_type));
 	}
 

dbms/include/DB/IO/CompressedReadBuffer.h

@@ -24,7 +24,9 @@ private:
 	std::vector<char> compressed_buffer;
 	qlz_state_decompress * qlz_state;
 
-	bool nextImpl()
+
+	/// Прочитать сжатые данные в compressed_buffer. Достать из их заголовка размер разжатых данных. Проверить чексумму.
+	bool readCompressedData(size_t & size_decompressed)
 	{
 		if (in.eof())
 			return false;
@@ -38,29 +40,43 @@ private:
 		if (size_compressed > DBMS_MAX_COMPRESSED_SIZE)
 			throw Exception("Too large size_compressed. Most likely corrupted data.", ErrorCodes::TOO_LARGE_SIZE_COMPRESSED);
 
-		size_t size_decompressed = qlz_size_decompressed(&compressed_buffer[0]);
+		size_decompressed = qlz_size_decompressed(&compressed_buffer[0]);
 
 		compressed_buffer.resize(size_compressed);
-		
-		memory.resize(size_decompressed);
-		internal_buffer = Buffer(&memory[0], &memory[size_decompressed]);
-		working_buffer = Buffer(&memory[0], &memory[size_decompressed]);
-
 		in.readStrict(&compressed_buffer[QUICKLZ_HEADER_SIZE], size_compressed - QUICKLZ_HEADER_SIZE);
 
 		if (checksum != CityHash128(&compressed_buffer[0], size_compressed))
 			throw Exception("Checksum doesnt match: corrupted data.", ErrorCodes::CHECKSUM_DOESNT_MATCH);
 
+		return true;
+	}
+
+	void decompress(char * to, size_t size_decompressed)
+	{
 		/// Старший бит первого байта определяет использованный метод сжатия.
 		if ((compressed_buffer[0] & 0x80) == 0)
 		{
 			if (!qlz_state)
 				qlz_state = new qlz_state_decompress;
 
-			qlz_decompress(&compressed_buffer[0], working_buffer.begin(), qlz_state);
+			qlz_decompress(&compressed_buffer[0], to, qlz_state);
 		}
 		else
-			LZ4_uncompress(&compressed_buffer[QUICKLZ_HEADER_SIZE], working_buffer.begin(), size_decompressed);
+			LZ4_uncompress(&compressed_buffer[QUICKLZ_HEADER_SIZE], to, size_decompressed);
+	}
+
+	bool nextImpl()
+	{
+		size_t size_decompressed = 0;
+
+		if (!readCompressedData(size_decompressed))
+			return false;
+
+		memory.resize(size_decompressed);
+		internal_buffer = Buffer(&memory[0], &memory[size_decompressed]);
+		working_buffer = Buffer(&memory[0], &memory[size_decompressed]);
+
+		decompress(working_buffer.begin(), size_decompressed);
 
 		return true;
 	}
@@ -79,6 +95,54 @@ public:
 		if (qlz_state)
 			delete qlz_state;
 	}
+
+
+	size_t readBig(char * to, size_t n)
+	{
+		size_t bytes_read = 0;
+
+		/// Если в буфере есть непрочитанные байты, то скопируем сколько надо в to.
+		if (pos < working_buffer.end())
+			bytes_read += read(to, std::min(static_cast<size_t>(working_buffer.end() - pos), n));
+
+		if (bytes_read < n)
+			bytes += offset();
+
+		/// Если надо ещё прочитать - будем, по возможности, разжимать сразу в to.
+		while (bytes_read < n)
+		{
+			size_t size_decompressed = 0;
+
+			if (!readCompressedData(size_decompressed))
+			{
+				working_buffer.resize(0);
+				pos = working_buffer.begin();
+				return bytes_read;
+			}
+
+			/// Если разжатый блок помещается целиком туда, куда его надо скопировать.
+			if (size_decompressed <= n - bytes_read)
+			{
+				decompress(to + bytes_read, size_decompressed);
+				bytes_read += size_decompressed;
+				bytes += size_decompressed;
+			}
+			else
+			{
+				memory.resize(size_decompressed);
+				internal_buffer = Buffer(&memory[0], &memory[size_decompressed]);
+				working_buffer = Buffer(&memory[0], &memory[size_decompressed]);
+				pos = working_buffer.begin();
+
+				decompress(working_buffer.begin(), size_decompressed);
+
+				bytes_read += read(to + bytes_read, n - bytes_read);
+				break;
+			}
+		}
+
+		return bytes_read;
+	}
 };
 
 }

dbms/include/DB/IO/ReadBuffer.h

@@ -110,6 +110,17 @@ public:
 			throw Exception("Cannot read all data", ErrorCodes::CANNOT_READ_ALL_DATA);
 	}
 
+	/** Метод, который может быть более эффективно реализован в наследниках, в случае чтения достаточно больших блоков.
+	  * Реализация может читать данные сразу в to, без лишнего копирования, если в to есть достаточно места для работы.
+	  * Например, CompressedReadBuffer может разжимать данные сразу в to, если весь разжатый блок туда помещается.
+	  * По-умолчанию - то же, что и read.
+	  * Для маленьких чтений использовать не нужно.
+	  */
+	virtual size_t readBig(char * to, size_t n)
+	{
+		return read(to, n);
+	}
+
 private:
 	/** Прочитать следующие данные и заполнить ими буфер.
 	  * Вернуть false в случае конца, true иначе.

dbms/src/DataTypes/DataTypeFixedString.cpp

@@ -55,7 +55,7 @@ void DataTypeFixedString::deserializeBinary(IColumn & column, ReadBuffer & istr,
 
 	size_t max_bytes = limit * n;
 	data.resize(max_bytes);
-	size_t read_bytes = istr.read(reinterpret_cast<char *>(&data[0]), max_bytes);
+	size_t read_bytes = istr.readBig(reinterpret_cast<char *>(&data[0]), max_bytes);
 
 	if (read_bytes % n != 0)
 		throw Exception("Cannot read all data of type FixedString",