2012-06-25 00:17:19 +00:00
|
|
|
|
#include <map>
|
2013-02-26 11:28:00 +00:00
|
|
|
|
#include <Poco/Path.h>
|
2012-06-25 00:17:19 +00:00
|
|
|
|
|
|
|
|
|
#include <DB/Common/escapeForFileName.h>
|
|
|
|
|
|
|
|
|
|
#include <DB/Core/Exception.h>
|
|
|
|
|
#include <DB/Core/ErrorCodes.h>
|
|
|
|
|
|
|
|
|
|
#include <DB/IO/ReadHelpers.h>
|
|
|
|
|
#include <DB/IO/WriteHelpers.h>
|
|
|
|
|
|
2012-08-29 18:49:54 +00:00
|
|
|
|
#include <DB/DataTypes/DataTypeArray.h>
|
|
|
|
|
|
|
|
|
|
#include <DB/Columns/ColumnArray.h>
|
|
|
|
|
|
2012-06-25 00:17:19 +00:00
|
|
|
|
#include <DB/Storages/StorageTinyLog.h>
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
#define DBMS_STORAGE_LOG_DATA_FILE_EXTENSION ".bin"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
namespace DB
|
|
|
|
|
{
|
|
|
|
|
|
|
|
|
|
using Poco::SharedPtr;
|
|
|
|
|
|
|
|
|
|
|
2013-01-23 17:38:03 +00:00
|
|
|
|
TinyLogBlockInputStream::TinyLogBlockInputStream(size_t block_size_, const Names & column_names_, StoragePtr owned_storage)
|
2013-02-21 09:11:18 +00:00
|
|
|
|
: IProfilingBlockInputStream(owned_storage), block_size(block_size_), column_names(column_names_), storage(dynamic_cast<StorageTinyLog &>(*owned_storage)), finished(false)
|
2012-06-25 00:17:19 +00:00
|
|
|
|
{
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Block TinyLogBlockInputStream::readImpl()
|
|
|
|
|
{
|
|
|
|
|
Block res;
|
|
|
|
|
|
2012-06-25 01:22:30 +00:00
|
|
|
|
if (finished || (!streams.empty() && streams.begin()->second->compressed.eof()))
|
|
|
|
|
{
|
|
|
|
|
/** Закрываем файлы (ещё до уничтожения объекта).
|
|
|
|
|
* Чтобы при создании многих источников, но одновременном чтении только из нескольких,
|
|
|
|
|
* буферы не висели в памяти.
|
|
|
|
|
*/
|
|
|
|
|
finished = true;
|
|
|
|
|
streams.clear();
|
|
|
|
|
return res;
|
|
|
|
|
}
|
|
|
|
|
|
2012-06-25 00:17:19 +00:00
|
|
|
|
/// Если файлы не открыты, то открываем их.
|
|
|
|
|
if (streams.empty())
|
|
|
|
|
for (Names::const_iterator it = column_names.begin(); it != column_names.end(); ++it)
|
2012-08-29 18:49:54 +00:00
|
|
|
|
addStream(*it, *storage.getDataTypeByName(*it));
|
2012-06-25 00:17:19 +00:00
|
|
|
|
|
|
|
|
|
for (Names::const_iterator it = column_names.begin(); it != column_names.end(); ++it)
|
|
|
|
|
{
|
|
|
|
|
ColumnWithNameAndType column;
|
|
|
|
|
column.name = *it;
|
|
|
|
|
column.type = storage.getDataTypeByName(*it);
|
|
|
|
|
column.column = column.type->createColumn();
|
2012-08-29 18:49:54 +00:00
|
|
|
|
readData(*it, *column.type, *column.column);
|
2012-06-25 00:17:19 +00:00
|
|
|
|
|
|
|
|
|
if (column.column->size())
|
|
|
|
|
res.insert(column);
|
|
|
|
|
}
|
|
|
|
|
|
2012-06-25 00:59:06 +00:00
|
|
|
|
if (!res || streams.begin()->second->compressed.eof())
|
2012-06-25 00:17:19 +00:00
|
|
|
|
{
|
2012-06-25 01:22:30 +00:00
|
|
|
|
finished = true;
|
2012-06-25 00:17:19 +00:00
|
|
|
|
streams.clear();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return res;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
2012-08-29 18:49:54 +00:00
|
|
|
|
void TinyLogBlockInputStream::addStream(const String & name, const IDataType & type, size_t level)
|
|
|
|
|
{
|
|
|
|
|
/// Для массивов используются отдельные потоки для размеров.
|
|
|
|
|
if (const DataTypeArray * type_arr = dynamic_cast<const DataTypeArray *>(&type))
|
|
|
|
|
{
|
|
|
|
|
String size_name = name + ARRAY_SIZES_COLUMN_NAME_SUFFIX + Poco::NumberFormatter::format(level);
|
|
|
|
|
streams.insert(std::make_pair(size_name, new Stream(storage.files[size_name].data_file.path())));
|
|
|
|
|
|
|
|
|
|
addStream(name, *type_arr->getNestedType(), level + 1);
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
streams.insert(std::make_pair(name, new Stream(storage.files[name].data_file.path())));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
void TinyLogBlockInputStream::readData(const String & name, const IDataType & type, IColumn & column, size_t level)
|
|
|
|
|
{
|
|
|
|
|
/// Для массивов требуется сначала десериализовать размеры, а потом значения.
|
|
|
|
|
if (const DataTypeArray * type_arr = dynamic_cast<const DataTypeArray *>(&type))
|
|
|
|
|
{
|
|
|
|
|
type_arr->deserializeOffsets(
|
|
|
|
|
column,
|
|
|
|
|
streams[name + ARRAY_SIZES_COLUMN_NAME_SUFFIX + Poco::NumberFormatter::format(level)]->compressed,
|
|
|
|
|
block_size);
|
|
|
|
|
|
2012-08-30 20:35:02 +00:00
|
|
|
|
if (column.size())
|
|
|
|
|
{
|
|
|
|
|
IColumn & nested_column = dynamic_cast<ColumnArray &>(column).getData();
|
|
|
|
|
readData(name, *type_arr->getNestedType(), nested_column, level + 1);
|
|
|
|
|
|
|
|
|
|
if (nested_column.size() != dynamic_cast<ColumnArray &>(column).getOffsets()[column.size() - 1])
|
|
|
|
|
throw Exception("Cannot read array data for all offsets", ErrorCodes::CANNOT_READ_ALL_DATA);
|
|
|
|
|
}
|
2012-08-29 18:49:54 +00:00
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
type.deserializeBinary(column, streams[name]->compressed, block_size);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
2013-01-23 17:38:03 +00:00
|
|
|
|
TinyLogBlockOutputStream::TinyLogBlockOutputStream(StoragePtr owned_storage)
|
2013-02-21 09:11:18 +00:00
|
|
|
|
: IBlockOutputStream(owned_storage), storage(dynamic_cast<StorageTinyLog &>(*owned_storage))
|
2012-06-25 00:17:19 +00:00
|
|
|
|
{
|
|
|
|
|
for (NamesAndTypesList::const_iterator it = storage.columns->begin(); it != storage.columns->end(); ++it)
|
2012-08-29 18:49:54 +00:00
|
|
|
|
addStream(it->first, *it->second);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
void TinyLogBlockOutputStream::addStream(const String & name, const IDataType & type, size_t level)
|
|
|
|
|
{
|
|
|
|
|
/// Для массивов используются отдельные потоки для размеров.
|
|
|
|
|
if (const DataTypeArray * type_arr = dynamic_cast<const DataTypeArray *>(&type))
|
|
|
|
|
{
|
|
|
|
|
String size_name = name + ARRAY_SIZES_COLUMN_NAME_SUFFIX + Poco::NumberFormatter::format(level);
|
|
|
|
|
streams.insert(std::make_pair(size_name, new Stream(storage.files[size_name].data_file.path())));
|
|
|
|
|
|
|
|
|
|
addStream(name, *type_arr->getNestedType(), level + 1);
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
streams.insert(std::make_pair(name, new Stream(storage.files[name].data_file.path())));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
void TinyLogBlockOutputStream::writeData(const String & name, const IDataType & type, const IColumn & column, size_t level)
|
|
|
|
|
{
|
|
|
|
|
/// Для массивов требуется сначала сериализовать размеры, а потом значения.
|
|
|
|
|
if (const DataTypeArray * type_arr = dynamic_cast<const DataTypeArray *>(&type))
|
|
|
|
|
{
|
|
|
|
|
type_arr->serializeOffsets(
|
|
|
|
|
column,
|
|
|
|
|
streams[name + ARRAY_SIZES_COLUMN_NAME_SUFFIX + Poco::NumberFormatter::format(level)]->compressed);
|
|
|
|
|
|
|
|
|
|
writeData(name, *type_arr->getNestedType(), dynamic_cast<const ColumnArray &>(column).getData(), level + 1);
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
type.serializeBinary(column, streams[name]->compressed);
|
2012-06-25 00:17:19 +00:00
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
void TinyLogBlockOutputStream::write(const Block & block)
|
|
|
|
|
{
|
2013-02-25 10:23:31 +00:00
|
|
|
|
storage.check(block, true);
|
2012-06-25 00:17:19 +00:00
|
|
|
|
|
|
|
|
|
for (size_t i = 0; i < block.columns(); ++i)
|
|
|
|
|
{
|
|
|
|
|
const ColumnWithNameAndType & column = block.getByPosition(i);
|
2012-08-29 18:49:54 +00:00
|
|
|
|
writeData(column.name, *column.type, *column.column);
|
2012-06-25 00:17:19 +00:00
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
2013-01-17 20:21:03 +00:00
|
|
|
|
StorageTinyLog::StorageTinyLog(const std::string & path_, const std::string & name_, NamesAndTypesListPtr columns_, bool attach)
|
2012-06-25 00:17:19 +00:00
|
|
|
|
: path(path_), name(name_), columns(columns_)
|
|
|
|
|
{
|
|
|
|
|
if (columns->empty())
|
|
|
|
|
throw Exception("Empty list of columns passed to StorageTinyLog constructor", ErrorCodes::EMPTY_LIST_OF_COLUMNS_PASSED);
|
2013-01-17 20:21:03 +00:00
|
|
|
|
|
|
|
|
|
if (!attach)
|
|
|
|
|
{
|
|
|
|
|
/// создаём файлы, если их нет
|
|
|
|
|
String full_path = path + escapeForFileName(name) + '/';
|
|
|
|
|
if (0 != mkdir(full_path.c_str(), S_IRWXU | S_IRWXG | S_IRWXO) && errno != EEXIST)
|
|
|
|
|
throwFromErrno("Cannot create directory " + full_path, ErrorCodes::CANNOT_CREATE_DIRECTORY);
|
|
|
|
|
}
|
2012-06-25 00:17:19 +00:00
|
|
|
|
|
|
|
|
|
for (NamesAndTypesList::const_iterator it = columns->begin(); it != columns->end(); ++it)
|
2012-08-29 18:49:54 +00:00
|
|
|
|
addFile(it->first, *it->second);
|
|
|
|
|
}
|
|
|
|
|
|
2013-02-06 11:26:35 +00:00
|
|
|
|
StoragePtr StorageTinyLog::create(const std::string & path_, const std::string & name_, NamesAndTypesListPtr columns_, bool attach)
|
|
|
|
|
{
|
|
|
|
|
return (new StorageTinyLog(path_, name_, columns_, attach))->thisPtr();
|
|
|
|
|
}
|
|
|
|
|
|
2012-08-29 18:49:54 +00:00
|
|
|
|
|
|
|
|
|
void StorageTinyLog::addFile(const String & column_name, const IDataType & type, size_t level)
|
|
|
|
|
{
|
|
|
|
|
if (files.end() != files.find(column_name))
|
|
|
|
|
throw Exception("Duplicate column with name " + column_name + " in constructor of StorageTinyLog.",
|
|
|
|
|
ErrorCodes::DUPLICATE_COLUMN);
|
|
|
|
|
|
|
|
|
|
if (const DataTypeArray * type_arr = dynamic_cast<const DataTypeArray *>(&type))
|
2012-06-25 00:17:19 +00:00
|
|
|
|
{
|
2012-08-29 18:49:54 +00:00
|
|
|
|
String size_column_suffix = ARRAY_SIZES_COLUMN_NAME_SUFFIX + Poco::NumberFormatter::format(level);
|
2012-06-25 00:17:19 +00:00
|
|
|
|
|
|
|
|
|
ColumnData column_data;
|
2012-08-29 18:49:54 +00:00
|
|
|
|
files.insert(std::make_pair(column_name + size_column_suffix, column_data));
|
|
|
|
|
files[column_name + size_column_suffix].data_file = Poco::File(
|
|
|
|
|
path + escapeForFileName(name) + '/' + escapeForFileName(column_name) + size_column_suffix + DBMS_STORAGE_LOG_DATA_FILE_EXTENSION);
|
|
|
|
|
|
|
|
|
|
addFile(column_name, *type_arr->getNestedType(), level + 1);
|
|
|
|
|
}
|
|
|
|
|
else
|
|
|
|
|
{
|
|
|
|
|
ColumnData column_data;
|
|
|
|
|
files.insert(std::make_pair(column_name, column_data));
|
|
|
|
|
files[column_name].data_file = Poco::File(
|
|
|
|
|
path + escapeForFileName(name) + '/' + escapeForFileName(column_name) + DBMS_STORAGE_LOG_DATA_FILE_EXTENSION);
|
2012-06-25 00:17:19 +00:00
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
void StorageTinyLog::rename(const String & new_path_to_db, const String & new_name)
|
|
|
|
|
{
|
|
|
|
|
/// Переименовываем директорию с данными.
|
|
|
|
|
Poco::File(path + escapeForFileName(name)).renameTo(new_path_to_db + escapeForFileName(new_name));
|
|
|
|
|
|
|
|
|
|
path = new_path_to_db;
|
|
|
|
|
name = new_name;
|
|
|
|
|
|
2012-08-29 18:49:54 +00:00
|
|
|
|
for (Files_t::iterator it = files.begin(); it != files.end(); ++it)
|
2013-02-26 11:37:49 +00:00
|
|
|
|
it->second.data_file = Poco::File(path + escapeForFileName(name) + '/' + Poco::Path(it->second.data_file.path()).getFileName());
|
2012-06-25 00:17:19 +00:00
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
BlockInputStreams StorageTinyLog::read(
|
|
|
|
|
const Names & column_names,
|
|
|
|
|
ASTPtr query,
|
2013-02-01 19:02:04 +00:00
|
|
|
|
const Settings & settings,
|
2012-06-25 00:17:19 +00:00
|
|
|
|
QueryProcessingStage::Enum & processed_stage,
|
|
|
|
|
size_t max_block_size,
|
|
|
|
|
unsigned threads)
|
|
|
|
|
{
|
|
|
|
|
check(column_names);
|
|
|
|
|
processed_stage = QueryProcessingStage::FetchColumns;
|
2013-01-23 17:38:03 +00:00
|
|
|
|
return BlockInputStreams(1, new TinyLogBlockInputStream(max_block_size, column_names, thisPtr()));
|
2012-06-25 00:17:19 +00:00
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
BlockOutputStreamPtr StorageTinyLog::write(
|
|
|
|
|
ASTPtr query)
|
|
|
|
|
{
|
2013-01-23 17:38:03 +00:00
|
|
|
|
return new TinyLogBlockOutputStream(thisPtr());
|
2012-06-25 00:17:19 +00:00
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
2013-01-23 17:43:19 +00:00
|
|
|
|
void StorageTinyLog::dropImpl()
|
2012-06-25 00:17:19 +00:00
|
|
|
|
{
|
|
|
|
|
for (Files_t::iterator it = files.begin(); it != files.end(); ++it)
|
|
|
|
|
if (it->second.data_file.exists())
|
|
|
|
|
it->second.data_file.remove();
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|