ClickHouse/dbms/src/Storages/StorageMemory.cpp

153 lines
3.8 KiB
C++
Raw Normal View History

#include <Common/Exception.h>
2011-10-31 17:55:06 +00:00
#include <DataStreams/IBlockInputStream.h>
2015-01-18 08:25:56 +00:00
#include <Storages/StorageMemory.h>
#include <Storages/StorageFactory.h>
2011-10-31 17:55:06 +00:00
2018-06-05 19:46:49 +00:00
#include <IO/WriteHelpers.h>
2011-10-31 17:55:06 +00:00
namespace DB
{
namespace ErrorCodes
{
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
}
class MemoryBlockInputStream : public IBlockInputStream
2011-10-31 17:55:06 +00:00
{
2015-01-18 08:25:56 +00:00
public:
MemoryBlockInputStream(const Names & column_names_, BlocksList::iterator begin_, BlocksList::iterator end_, const StorageMemory & storage_)
: column_names(column_names_), begin(begin_), end(end_), it(begin), storage(storage_) {}
2011-10-31 17:55:06 +00:00
String getName() const override { return "Memory"; }
2011-10-31 17:55:06 +00:00
Block getHeader() const override { return storage.getSampleBlockForColumns(column_names); }
2015-01-18 08:25:56 +00:00
protected:
Block readImpl() override
{
if (it == end)
{
return Block();
}
else
{
Block src = *it;
Block res;
/// Add only required columns to `res`.
for (size_t i = 0, size = column_names.size(); i < size; ++i)
res.insert(src.getByName(column_names[i]));
++it;
return res;
}
}
2015-01-18 08:25:56 +00:00
private:
Names column_names;
BlocksList::iterator begin;
BlocksList::iterator end;
BlocksList::iterator it;
const StorageMemory & storage;
2015-01-18 08:25:56 +00:00
};
2011-10-31 17:55:06 +00:00
2015-01-18 08:25:56 +00:00
class MemoryBlockOutputStream : public IBlockOutputStream
2011-10-31 17:55:06 +00:00
{
2015-01-18 08:25:56 +00:00
public:
explicit MemoryBlockOutputStream(StorageMemory & storage_) : storage(storage_) {}
Block getHeader() const override { return storage.getSampleBlock(); }
void write(const Block & block) override
{
storage.check(block, true);
2019-01-02 06:44:36 +00:00
std::lock_guard lock(storage.mutex);
storage.data.push_back(block);
}
2015-01-18 08:25:56 +00:00
private:
StorageMemory & storage;
2015-01-18 08:25:56 +00:00
};
2011-10-31 17:55:06 +00:00
2019-12-04 16:06:55 +00:00
StorageMemory::StorageMemory(const StorageID & table_id_, ColumnsDescription columns_description_, ConstraintsDescription constraints_)
: IStorage(table_id_)
{
2019-08-24 21:20:20 +00:00
setColumns(std::move(columns_description_));
setConstraints(std::move(constraints_));
}
2012-01-09 19:20:48 +00:00
BlockInputStreams StorageMemory::read(
const Names & column_names,
2017-12-01 21:13:25 +00:00
const SelectQueryInfo & /*query_info*/,
const Context & /*context*/,
QueryProcessingStage::Enum /*processed_stage*/,
2017-12-01 21:13:25 +00:00
size_t /*max_block_size*/,
2017-06-02 15:54:39 +00:00
unsigned num_streams)
2011-10-31 17:55:06 +00:00
{
check(column_names);
2012-05-30 04:45:49 +00:00
2019-01-02 06:44:36 +00:00
std::lock_guard lock(mutex);
size_t size = data.size();
2017-06-02 15:54:39 +00:00
if (num_streams > size)
num_streams = size;
2012-05-30 04:45:49 +00:00
BlockInputStreams res;
2012-05-30 04:45:49 +00:00
2017-06-02 15:54:39 +00:00
for (size_t stream = 0; stream < num_streams; ++stream)
{
BlocksList::iterator begin = data.begin();
BlocksList::iterator end = data.begin();
2017-06-02 15:54:39 +00:00
std::advance(begin, stream * size / num_streams);
std::advance(end, (stream + 1) * size / num_streams);
res.push_back(std::make_shared<MemoryBlockInputStream>(column_names, begin, end, *this));
}
return res;
2011-10-31 17:55:06 +00:00
}
2011-10-31 17:55:06 +00:00
BlockOutputStreamPtr StorageMemory::write(
const ASTPtr & /*query*/, const Context & /*context*/)
2011-10-31 17:55:06 +00:00
{
return std::make_shared<MemoryBlockOutputStream>(*this);
2011-10-31 17:55:06 +00:00
}
2011-11-05 23:31:19 +00:00
2019-08-27 20:43:08 +00:00
void StorageMemory::drop(TableStructureWriteLockHolder &)
2011-11-05 23:31:19 +00:00
{
2019-01-02 06:44:36 +00:00
std::lock_guard lock(mutex);
data.clear();
2011-11-05 23:31:19 +00:00
}
2019-08-27 20:43:08 +00:00
void StorageMemory::truncate(const ASTPtr &, const Context &, TableStructureWriteLockHolder &)
2018-04-21 00:35:20 +00:00
{
2019-01-02 06:44:36 +00:00
std::lock_guard lock(mutex);
2018-04-21 00:35:20 +00:00
data.clear();
}
void registerStorageMemory(StorageFactory & factory)
{
factory.registerStorage("Memory", [](const StorageFactory::Arguments & args)
{
if (!args.engine_args.empty())
throw Exception(
"Engine " + args.engine_name + " doesn't support any arguments (" + toString(args.engine_args.size()) + " given)",
ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
2019-12-04 16:06:55 +00:00
return StorageMemory::create(args.table_id, args.columns, args.constraints);
});
}
2011-10-31 17:55:06 +00:00
}