ClickHouse/dbms/include/DB/Storages/StorageTinyLog.h

153 lines
4.2 KiB
C
Raw Normal View History

2012-06-25 00:17:19 +00:00
#pragma once
#include <set>
2012-06-25 00:17:19 +00:00
#include <Poco/File.h>
#include <DB/Core/NamesAndTypes.h>
#include <DB/IO/ReadBufferFromFile.h>
#include <DB/IO/WriteBufferFromFile.h>
#include <DB/IO/CompressedReadBuffer.h>
#include <DB/IO/CompressedWriteBuffer.h>
#include <DB/Storages/IStorage.h>
#include <DB/DataStreams/IProfilingBlockInputStream.h>
#include <DB/DataStreams/IBlockOutputStream.h>
2012-06-25 00:17:19 +00:00
namespace DB
{
class StorageTinyLog;
class TinyLogBlockInputStream : public IProfilingBlockInputStream
{
public:
TinyLogBlockInputStream(size_t block_size_, const Names & column_names_, StorageTinyLog & storage_);
2012-06-25 00:17:19 +00:00
String getName() const { return "TinyLogBlockInputStream"; }
String getID() const;
2012-10-20 02:10:47 +00:00
protected:
Block readImpl();
2012-06-25 00:17:19 +00:00
private:
size_t block_size;
Names column_names;
StorageTinyLog & storage;
2012-06-25 01:22:30 +00:00
bool finished;
2012-06-25 00:17:19 +00:00
struct Stream
{
Stream(const std::string & data_path)
: plain(data_path, std::min(static_cast<size_t>(DBMS_DEFAULT_BUFFER_SIZE), Poco::File(data_path).getSize())),
compressed(plain)
{
}
ReadBufferFromFile plain;
CompressedReadBuffer compressed;
};
2014-04-22 22:43:55 +00:00
typedef std::map<std::string, std::unique_ptr<Stream> > FileStreams;
2012-06-25 00:17:19 +00:00
FileStreams streams;
2012-08-29 18:49:54 +00:00
void addStream(const String & name, const IDataType & type, size_t level = 0);
void readData(const String & name, const IDataType & type, IColumn & column, size_t limit, size_t level = 0, bool read_offsets = true);
2012-06-25 00:17:19 +00:00
};
class TinyLogBlockOutputStream : public IBlockOutputStream
{
public:
TinyLogBlockOutputStream(StorageTinyLog & storage_);
2012-06-25 00:17:19 +00:00
void write(const Block & block);
2013-09-15 01:40:29 +00:00
void writeSuffix();
2012-06-25 00:17:19 +00:00
private:
StorageTinyLog & storage;
struct Stream
{
2014-03-28 14:36:24 +00:00
Stream(const std::string & data_path, size_t max_compress_block_size) :
plain(data_path, max_compress_block_size, O_APPEND | O_CREAT | O_WRONLY),
2012-06-25 00:17:19 +00:00
compressed(plain)
{
}
WriteBufferFromFile plain;
CompressedWriteBuffer compressed;
2013-09-15 01:40:29 +00:00
2013-09-26 19:16:43 +00:00
void finalize()
2013-09-15 01:40:29 +00:00
{
compressed.next();
2013-09-26 19:16:43 +00:00
plain.next();
2013-09-15 01:40:29 +00:00
}
2012-06-25 00:17:19 +00:00
};
2014-04-22 22:43:55 +00:00
typedef std::map<std::string, std::unique_ptr<Stream> > FileStreams;
2012-06-25 00:17:19 +00:00
FileStreams streams;
typedef std::set<std::string> OffsetColumns;
2012-08-29 18:49:54 +00:00
void addStream(const String & name, const IDataType & type, size_t level = 0);
void writeData(const String & name, const IDataType & type, const IColumn & column, OffsetColumns & offset_columns, size_t level = 0);
2012-06-25 00:17:19 +00:00
};
/** Реализует хранилище, подходящее для маленьких кусочков лога.
* Отличается от StorageLog отсутствием файлов с засечками.
*/
class StorageTinyLog : public IStorage
{
friend class TinyLogBlockInputStream;
friend class TinyLogBlockOutputStream;
public:
/** Подцепить таблицу с соответствующим именем, по соответствующему пути (с / на конце),
* (корректность имён и путей не проверяется)
* состоящую из указанных столбцов.
* Если не указано attach - создать директорию, если её нет.
2012-06-25 00:17:19 +00:00
*/
2014-03-28 14:36:24 +00:00
static StoragePtr create(const std::string & path_, const std::string & name_, NamesAndTypesListPtr columns_, bool attach, size_t max_compress_block_size_ = DEFAULT_MAX_COMPRESS_BLOCK_SIZE);
2012-06-25 00:17:19 +00:00
std::string getName() const { return "TinyLog"; }
std::string getTableName() const { return name; }
const NamesAndTypesList & getColumnsList() const { return *columns; }
BlockInputStreams read(
const Names & column_names,
ASTPtr query,
const Settings & settings,
2012-06-25 00:17:19 +00:00
QueryProcessingStage::Enum & processed_stage,
size_t max_block_size = DEFAULT_BLOCK_SIZE,
unsigned threads = 1);
BlockOutputStreamPtr write(
ASTPtr query);
2014-03-20 13:28:49 +00:00
void drop() override;
2012-06-25 00:17:19 +00:00
void rename(const String & new_path_to_db, const String & new_name);
private:
String path;
String name;
NamesAndTypesListPtr columns;
2014-03-28 14:36:24 +00:00
size_t max_compress_block_size;
2012-06-25 00:17:19 +00:00
/// Данные столбца
struct ColumnData
{
Poco::File data_file;
};
typedef std::map<String, ColumnData> Files_t;
Files_t files;
2012-08-29 18:49:54 +00:00
2014-03-28 14:36:24 +00:00
StorageTinyLog(const std::string & path_, const std::string & name_, NamesAndTypesListPtr columns_, bool attach, size_t max_compress_block_size_);
2012-08-29 18:49:54 +00:00
void addFile(const String & column_name, const IDataType & type, size_t level = 0);
2012-06-25 00:17:19 +00:00
};
}