2013-04-24 10:31:32 +00:00
|
|
|
|
#pragma once
|
|
|
|
|
|
2013-09-15 01:10:16 +00:00
|
|
|
|
#include <DB/IO/WriteBufferFromFile.h>
|
|
|
|
|
#include <DB/IO/CompressedWriteBuffer.h>
|
|
|
|
|
|
2014-03-09 17:36:01 +00:00
|
|
|
|
#include <DB/Storages/MergeTree/MergeTreeData.h>
|
2013-04-24 10:31:32 +00:00
|
|
|
|
|
2013-09-15 01:10:16 +00:00
|
|
|
|
|
2013-04-24 10:31:32 +00:00
|
|
|
|
namespace DB
|
|
|
|
|
{
|
2014-03-04 11:30:50 +00:00
|
|
|
|
class IMergedBlockOutputStream : public IBlockOutputStream
|
2013-04-24 10:31:32 +00:00
|
|
|
|
{
|
|
|
|
|
public:
|
2014-03-09 17:36:01 +00:00
|
|
|
|
IMergedBlockOutputStream(MergeTreeData & storage_) : storage(storage_), index_offset(0)
|
2013-04-24 10:31:32 +00:00
|
|
|
|
{
|
|
|
|
|
}
|
2013-08-24 08:01:19 +00:00
|
|
|
|
|
2014-03-04 11:30:50 +00:00
|
|
|
|
protected:
|
|
|
|
|
typedef std::set<std::string> OffsetColumns;
|
2013-04-24 10:31:32 +00:00
|
|
|
|
struct ColumnStream
|
|
|
|
|
{
|
|
|
|
|
ColumnStream(const String & data_path, const std::string & marks_path) :
|
|
|
|
|
plain(data_path, DBMS_DEFAULT_BUFFER_SIZE, O_TRUNC | O_CREAT | O_WRONLY),
|
|
|
|
|
compressed(plain),
|
|
|
|
|
marks(marks_path, 4096, O_TRUNC | O_CREAT | O_WRONLY) {}
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-04-24 10:31:32 +00:00
|
|
|
|
WriteBufferFromFile plain;
|
|
|
|
|
CompressedWriteBuffer compressed;
|
|
|
|
|
WriteBufferFromFile marks;
|
2013-09-15 01:10:16 +00:00
|
|
|
|
|
2013-09-26 19:16:43 +00:00
|
|
|
|
void finalize()
|
2013-09-15 01:10:16 +00:00
|
|
|
|
{
|
|
|
|
|
compressed.next();
|
2013-10-03 13:06:27 +00:00
|
|
|
|
plain.next();
|
|
|
|
|
marks.next();
|
2013-09-15 01:10:16 +00:00
|
|
|
|
}
|
2014-03-05 16:28:24 +00:00
|
|
|
|
|
|
|
|
|
void sync()
|
|
|
|
|
{
|
|
|
|
|
plain.sync();
|
|
|
|
|
marks.sync();
|
|
|
|
|
}
|
2013-04-24 10:31:32 +00:00
|
|
|
|
};
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-04-24 10:31:32 +00:00
|
|
|
|
typedef std::map<String, SharedPtr<ColumnStream> > ColumnStreams;
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
|
|
|
|
void addStream(const String & path, const String & name, const IDataType & type, size_t level = 0, String filename = "")
|
2013-04-24 10:31:32 +00:00
|
|
|
|
{
|
2014-03-04 11:30:50 +00:00
|
|
|
|
String escaped_column_name;
|
|
|
|
|
if (filename.size())
|
|
|
|
|
escaped_column_name = escapeForFileName(filename);
|
|
|
|
|
else
|
|
|
|
|
escaped_column_name = escapeForFileName(name);
|
|
|
|
|
|
2013-04-24 10:31:32 +00:00
|
|
|
|
/// Для массивов используются отдельные потоки для размеров.
|
|
|
|
|
if (const DataTypeArray * type_arr = dynamic_cast<const DataTypeArray *>(&type))
|
|
|
|
|
{
|
2013-08-07 19:45:47 +00:00
|
|
|
|
String size_name = DataTypeNested::extractNestedTableName(name)
|
|
|
|
|
+ ARRAY_SIZES_COLUMN_NAME_SUFFIX + toString(level);
|
|
|
|
|
String escaped_size_name = escapeForFileName(DataTypeNested::extractNestedTableName(name))
|
|
|
|
|
+ ARRAY_SIZES_COLUMN_NAME_SUFFIX + toString(level);
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-04-24 10:31:32 +00:00
|
|
|
|
column_streams[size_name] = new ColumnStream(
|
2014-03-04 11:30:50 +00:00
|
|
|
|
path + escaped_size_name + ".bin",
|
|
|
|
|
path + escaped_size_name + ".mrk");
|
|
|
|
|
|
|
|
|
|
addStream(path, name, *type_arr->getNestedType(), level + 1);
|
2013-04-24 10:31:32 +00:00
|
|
|
|
}
|
2013-07-12 13:35:05 +00:00
|
|
|
|
else if (const DataTypeNested * type_nested = dynamic_cast<const DataTypeNested *>(&type))
|
|
|
|
|
{
|
|
|
|
|
String size_name = name + ARRAY_SIZES_COLUMN_NAME_SUFFIX + toString(level);
|
|
|
|
|
String escaped_size_name = escaped_column_name + ARRAY_SIZES_COLUMN_NAME_SUFFIX + toString(level);
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-07-12 13:35:05 +00:00
|
|
|
|
column_streams[size_name] = new ColumnStream(
|
2014-03-04 11:30:50 +00:00
|
|
|
|
path + escaped_size_name + ".bin",
|
|
|
|
|
path + escaped_size_name + ".mrk");
|
2013-07-12 13:35:05 +00:00
|
|
|
|
|
|
|
|
|
const NamesAndTypesList & columns = *type_nested->getNestedTypesList();
|
|
|
|
|
for (NamesAndTypesList::const_iterator it = columns.begin(); it != columns.end(); ++it)
|
2014-03-04 11:30:50 +00:00
|
|
|
|
addStream(path, DataTypeNested::concatenateNestedName(name, it->first), *it->second, level + 1);
|
2013-07-12 13:35:05 +00:00
|
|
|
|
}
|
2013-04-24 10:31:32 +00:00
|
|
|
|
else
|
|
|
|
|
column_streams[name] = new ColumnStream(
|
2014-03-04 11:30:50 +00:00
|
|
|
|
path + escaped_column_name + ".bin",
|
|
|
|
|
path + escaped_column_name + ".mrk");
|
2013-04-24 10:31:32 +00:00
|
|
|
|
}
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
|
|
|
|
|
2013-04-24 10:31:32 +00:00
|
|
|
|
/// Записать данные одного столбца.
|
2013-08-08 13:12:28 +00:00
|
|
|
|
void writeData(const String & name, const IDataType & type, const IColumn & column, OffsetColumns & offset_columns, size_t level = 0)
|
2013-04-24 10:31:32 +00:00
|
|
|
|
{
|
|
|
|
|
size_t size = column.size();
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-04-24 10:31:32 +00:00
|
|
|
|
/// Для массивов требуется сначала сериализовать размеры, а потом значения.
|
|
|
|
|
if (const DataTypeArray * type_arr = dynamic_cast<const DataTypeArray *>(&type))
|
|
|
|
|
{
|
2013-08-07 19:45:47 +00:00
|
|
|
|
String size_name = DataTypeNested::extractNestedTableName(name)
|
|
|
|
|
+ ARRAY_SIZES_COLUMN_NAME_SUFFIX + toString(level);
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-08-08 13:12:28 +00:00
|
|
|
|
if (offset_columns.count(size_name) == 0)
|
2013-04-24 10:31:32 +00:00
|
|
|
|
{
|
2013-08-08 13:12:28 +00:00
|
|
|
|
offset_columns.insert(size_name);
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-08-08 13:12:28 +00:00
|
|
|
|
ColumnStream & stream = *column_streams[size_name];
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-08-08 13:12:28 +00:00
|
|
|
|
size_t prev_mark = 0;
|
|
|
|
|
while (prev_mark < size)
|
2013-04-24 10:31:32 +00:00
|
|
|
|
{
|
2013-08-08 13:12:28 +00:00
|
|
|
|
size_t limit = 0;
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-08-08 13:12:28 +00:00
|
|
|
|
/// Если есть index_offset, то первая засечка идёт не сразу, а после этого количества строк.
|
|
|
|
|
if (prev_mark == 0 && index_offset != 0)
|
|
|
|
|
{
|
|
|
|
|
limit = index_offset;
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
{
|
|
|
|
|
limit = storage.index_granularity;
|
|
|
|
|
writeIntBinary(stream.plain.count(), stream.marks);
|
|
|
|
|
writeIntBinary(stream.compressed.offset(), stream.marks);
|
|
|
|
|
}
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-08-08 13:12:28 +00:00
|
|
|
|
type_arr->serializeOffsets(column, stream.compressed, prev_mark, limit);
|
2013-12-23 03:10:12 +00:00
|
|
|
|
stream.compressed.nextIfAtEnd(); /// Чтобы вместо засечек, указывающих на конец сжатого блока, были засечки, указывающие на начало следующего.
|
2013-08-08 13:12:28 +00:00
|
|
|
|
prev_mark += limit;
|
2013-04-24 10:31:32 +00:00
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
2013-08-24 08:26:25 +00:00
|
|
|
|
else if (const DataTypeNested * type_nested = dynamic_cast<const DataTypeNested *>(&type))
|
2013-07-12 13:35:05 +00:00
|
|
|
|
{
|
|
|
|
|
String size_name = name + ARRAY_SIZES_COLUMN_NAME_SUFFIX + toString(level);
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-07-12 13:35:05 +00:00
|
|
|
|
ColumnStream & stream = *column_streams[size_name];
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-07-12 13:35:05 +00:00
|
|
|
|
size_t prev_mark = 0;
|
|
|
|
|
while (prev_mark < size)
|
|
|
|
|
{
|
|
|
|
|
size_t limit = 0;
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-07-12 13:35:05 +00:00
|
|
|
|
/// Если есть index_offset, то первая засечка идёт не сразу, а после этого количества строк.
|
|
|
|
|
if (prev_mark == 0 && index_offset != 0)
|
|
|
|
|
{
|
|
|
|
|
limit = index_offset;
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
{
|
|
|
|
|
limit = storage.index_granularity;
|
|
|
|
|
writeIntBinary(stream.plain.count(), stream.marks);
|
|
|
|
|
writeIntBinary(stream.compressed.offset(), stream.marks);
|
|
|
|
|
}
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-07-12 13:35:05 +00:00
|
|
|
|
type_nested->serializeOffsets(column, stream.compressed, prev_mark, limit);
|
2013-12-23 03:10:12 +00:00
|
|
|
|
stream.compressed.nextIfAtEnd(); /// Чтобы вместо засечек, указывающих на конец сжатого блока, были засечки, указывающие на начало следующего.
|
2013-07-12 13:35:05 +00:00
|
|
|
|
prev_mark += limit;
|
|
|
|
|
}
|
|
|
|
|
}
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-04-24 10:31:32 +00:00
|
|
|
|
{
|
|
|
|
|
ColumnStream & stream = *column_streams[name];
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-04-24 10:31:32 +00:00
|
|
|
|
size_t prev_mark = 0;
|
|
|
|
|
while (prev_mark < size)
|
|
|
|
|
{
|
|
|
|
|
size_t limit = 0;
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-04-24 10:31:32 +00:00
|
|
|
|
/// Если есть index_offset, то первая засечка идёт не сразу, а после этого количества строк.
|
|
|
|
|
if (prev_mark == 0 && index_offset != 0)
|
|
|
|
|
{
|
|
|
|
|
limit = index_offset;
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
{
|
|
|
|
|
limit = storage.index_granularity;
|
|
|
|
|
writeIntBinary(stream.plain.count(), stream.marks);
|
|
|
|
|
writeIntBinary(stream.compressed.offset(), stream.marks);
|
|
|
|
|
}
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2013-04-24 10:31:32 +00:00
|
|
|
|
type.serializeBinary(column, stream.compressed, prev_mark, limit);
|
2013-12-23 03:10:12 +00:00
|
|
|
|
stream.compressed.nextIfAtEnd(); /// Чтобы вместо засечек, указывающих на конец сжатого блока, были засечки, указывающие на начало следующего.
|
2013-04-24 10:31:32 +00:00
|
|
|
|
prev_mark += limit;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
2014-03-09 17:36:01 +00:00
|
|
|
|
MergeTreeData & storage;
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
|
|
|
|
ColumnStreams column_streams;
|
|
|
|
|
|
|
|
|
|
/// Смещение до первой строчки блока, для которой надо записать индекс.
|
|
|
|
|
size_t index_offset;
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
/** Для записи куска, полученного слиянием нескольких других.
|
|
|
|
|
* Данные уже отсортированы, относятся к одному месяцу, и пишутся в один кускок.
|
|
|
|
|
*/
|
|
|
|
|
class MergedBlockOutputStream : public IMergedBlockOutputStream
|
|
|
|
|
{
|
|
|
|
|
public:
|
2014-03-09 17:36:01 +00:00
|
|
|
|
MergedBlockOutputStream(MergeTreeData & storage_,
|
2014-03-04 11:30:50 +00:00
|
|
|
|
UInt16 min_date, UInt16 max_date, UInt64 min_part_id, UInt64 max_part_id, UInt32 level)
|
|
|
|
|
: IMergedBlockOutputStream(storage_), marks_count(0)
|
|
|
|
|
{
|
|
|
|
|
part_name = storage.getPartName(
|
|
|
|
|
DayNum_t(min_date), DayNum_t(max_date),
|
|
|
|
|
min_part_id, max_part_id, level);
|
|
|
|
|
|
|
|
|
|
part_tmp_path = storage.full_path + "tmp_" + part_name + "/";
|
|
|
|
|
part_res_path = storage.full_path + part_name + "/";
|
|
|
|
|
|
|
|
|
|
Poco::File(part_tmp_path).createDirectories();
|
|
|
|
|
|
|
|
|
|
index_stream = new WriteBufferFromFile(part_tmp_path + "primary.idx", DBMS_DEFAULT_BUFFER_SIZE, O_TRUNC | O_CREAT | O_WRONLY);
|
|
|
|
|
|
|
|
|
|
for (NamesAndTypesList::const_iterator it = storage.columns->begin(); it != storage.columns->end(); ++it)
|
|
|
|
|
addStream(part_tmp_path, it->first, *it->second);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
void write(const Block & block)
|
|
|
|
|
{
|
|
|
|
|
size_t rows = block.rows();
|
|
|
|
|
|
|
|
|
|
/// Сначала пишем индекс. Индекс содержит значение Primary Key для каждой index_granularity строки.
|
|
|
|
|
typedef std::vector<const ColumnWithNameAndType *> PrimaryColumns;
|
|
|
|
|
PrimaryColumns primary_columns;
|
|
|
|
|
|
|
|
|
|
for (size_t i = 0, size = storage.sort_descr.size(); i < size; ++i)
|
|
|
|
|
primary_columns.push_back(
|
|
|
|
|
!storage.sort_descr[i].column_name.empty()
|
|
|
|
|
? &block.getByName(storage.sort_descr[i].column_name)
|
|
|
|
|
: &block.getByPosition(storage.sort_descr[i].column_number));
|
|
|
|
|
|
|
|
|
|
for (size_t i = index_offset; i < rows; i += storage.index_granularity)
|
|
|
|
|
{
|
|
|
|
|
for (PrimaryColumns::const_iterator it = primary_columns.begin(); it != primary_columns.end(); ++it)
|
|
|
|
|
{
|
|
|
|
|
(*it)->type->serializeBinary((*(*it)->column)[i], *index_stream);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
++marks_count;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Множество записанных столбцов со смещениями, чтобы не писать общие для вложенных структур столбцы несколько раз
|
|
|
|
|
OffsetColumns offset_columns;
|
|
|
|
|
|
|
|
|
|
/// Теперь пишем данные.
|
|
|
|
|
for (NamesAndTypesList::const_iterator it = storage.columns->begin(); it != storage.columns->end(); ++it)
|
|
|
|
|
{
|
|
|
|
|
const ColumnWithNameAndType & column = block.getByName(it->first);
|
|
|
|
|
writeData(column.name, *column.type, *column.column, offset_columns);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
size_t written_for_last_mark = (storage.index_granularity - index_offset + rows) % storage.index_granularity;
|
|
|
|
|
index_offset = (storage.index_granularity - written_for_last_mark) % storage.index_granularity;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
void writeSuffix()
|
|
|
|
|
{
|
|
|
|
|
/// Заканчиваем запись.
|
|
|
|
|
index_stream->next();
|
|
|
|
|
index_stream = NULL;
|
|
|
|
|
|
|
|
|
|
for (ColumnStreams::iterator it = column_streams.begin(); it != column_streams.end(); ++it)
|
|
|
|
|
it->second->finalize();
|
|
|
|
|
|
|
|
|
|
column_streams.clear();
|
|
|
|
|
|
|
|
|
|
if (marks_count == 0)
|
|
|
|
|
{
|
|
|
|
|
/// Кусок пустой - все записи удалились.
|
|
|
|
|
Poco::File(part_tmp_path).remove(true);
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
{
|
|
|
|
|
/// Переименовываем кусок.
|
|
|
|
|
Poco::File(part_tmp_path).renameTo(part_res_path);
|
|
|
|
|
|
|
|
|
|
/// А добавление нового куска в набор (и удаление исходных кусков) сделает вызывающая сторона.
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
/// Сколько засечек уже записано.
|
|
|
|
|
size_t marksCount()
|
|
|
|
|
{
|
|
|
|
|
return marks_count;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private:
|
|
|
|
|
String part_name;
|
|
|
|
|
String part_tmp_path;
|
|
|
|
|
String part_res_path;
|
|
|
|
|
size_t marks_count;
|
|
|
|
|
|
|
|
|
|
SharedPtr<WriteBufferFromFile> index_stream;
|
2013-04-24 10:31:32 +00:00
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
typedef Poco::SharedPtr<MergedBlockOutputStream> MergedBlockOutputStreamPtr;
|
2014-03-04 11:30:50 +00:00
|
|
|
|
|
|
|
|
|
/// Записывает только те, столбцы, что лежат в block
|
|
|
|
|
class MergedColumnOnlyOutputStream : public IMergedBlockOutputStream
|
|
|
|
|
{
|
|
|
|
|
public:
|
2014-03-09 17:36:01 +00:00
|
|
|
|
MergedColumnOnlyOutputStream(MergeTreeData & storage_, String part_path_, bool sync_ = false) :
|
2014-03-05 16:28:24 +00:00
|
|
|
|
IMergedBlockOutputStream(storage_), part_path(part_path_), initialized(false), sync(sync_)
|
2014-03-04 11:30:50 +00:00
|
|
|
|
{
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
void write(const Block & block)
|
|
|
|
|
{
|
|
|
|
|
if (!initialized)
|
|
|
|
|
{
|
|
|
|
|
column_streams.clear();
|
|
|
|
|
for (size_t i = 0; i < block.columns(); ++i)
|
|
|
|
|
{
|
|
|
|
|
addStream(part_path, block.getByPosition(i).name,
|
|
|
|
|
*block.getByPosition(i).type, 0, prefix + block.getByPosition(i).name);
|
|
|
|
|
}
|
|
|
|
|
initialized = true;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
size_t rows = block.rows();
|
|
|
|
|
|
|
|
|
|
OffsetColumns offset_columns;
|
|
|
|
|
for (size_t i = 0; i < block.columns(); ++i)
|
|
|
|
|
{
|
|
|
|
|
const ColumnWithNameAndType & column = block.getByPosition(i);
|
|
|
|
|
writeData(column.name, *column.type, *column.column, offset_columns);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
size_t written_for_last_mark = (storage.index_granularity - index_offset + rows) % storage.index_granularity;
|
|
|
|
|
index_offset = (storage.index_granularity - written_for_last_mark) % storage.index_granularity;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
void writeSuffix()
|
|
|
|
|
{
|
|
|
|
|
for (auto & column_stream : column_streams)
|
|
|
|
|
{
|
|
|
|
|
column_stream.second->finalize();
|
2014-03-05 16:28:24 +00:00
|
|
|
|
if (sync)
|
|
|
|
|
column_stream.second->sync();
|
2014-03-04 14:09:19 +00:00
|
|
|
|
std::string column = escapeForFileName(column_stream.first);
|
|
|
|
|
Poco::File(part_path + prefix + column + ".bin").renameTo(part_path + column + ".bin");
|
|
|
|
|
Poco::File(part_path + prefix + column + ".mrk").renameTo(part_path + column + ".mrk");
|
2014-03-04 11:30:50 +00:00
|
|
|
|
}
|
|
|
|
|
column_streams.clear();
|
|
|
|
|
initialized = false;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
private:
|
|
|
|
|
String part_path;
|
|
|
|
|
|
|
|
|
|
bool initialized;
|
|
|
|
|
|
|
|
|
|
const std::string prefix = "tmp_";
|
2014-03-05 16:28:24 +00:00
|
|
|
|
|
|
|
|
|
bool sync;
|
2014-03-04 11:30:50 +00:00
|
|
|
|
};
|
2013-04-24 10:31:32 +00:00
|
|
|
|
|
2013-09-26 19:16:43 +00:00
|
|
|
|
}
|