ClickHouse/src/Storages/StorageMemory.cpp

370 lines
11 KiB
C++
Raw Normal View History

2020-11-23 17:45:59 +00:00
#include <cassert>
#include <Common/Exception.h>
2011-10-31 17:55:06 +00:00
#include <DataStreams/IBlockInputStream.h>
2015-01-18 08:25:56 +00:00
2020-09-22 09:23:46 +00:00
#include <Interpreters/MutationsInterpreter.h>
#include <Storages/StorageFactory.h>
2020-09-22 09:23:46 +00:00
#include <Storages/StorageMemory.h>
2021-02-07 01:41:31 +00:00
#include <Storages/MemorySettings.h>
2011-10-31 17:55:06 +00:00
2018-06-05 19:46:49 +00:00
#include <IO/WriteHelpers.h>
#include <Processors/Sources/SourceWithProgress.h>
#include <Processors/Pipe.h>
2018-06-05 19:46:49 +00:00
2011-10-31 17:55:06 +00:00
namespace DB
{
namespace ErrorCodes
{
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
}
class MemorySource : public SourceWithProgress
2011-10-31 17:55:06 +00:00
{
2021-02-12 00:25:00 +00:00
using InitializerFunc = std::function<void(std::shared_ptr<const Blocks> &)>;
2015-01-18 08:25:56 +00:00
public:
MemorySource(
Names column_names_,
const StorageMemory & storage,
const StorageMetadataPtr & metadata_snapshot,
2021-02-12 00:25:00 +00:00
std::shared_ptr<const Blocks> data_,
2020-12-13 21:21:25 +00:00
std::shared_ptr<std::atomic<size_t>> parallel_execution_index_,
InitializerFunc initializer_func_ = {})
2020-06-19 17:17:13 +00:00
: SourceWithProgress(metadata_snapshot->getSampleBlockForColumns(column_names_, storage.getVirtuals(), storage.getStorageID()))
, column_names_and_types(metadata_snapshot->getColumns().getAllWithSubcolumns().addTypes(std::move(column_names_)))
2020-10-08 11:27:13 +00:00
, data(data_)
2020-12-13 21:21:25 +00:00
, parallel_execution_index(parallel_execution_index_)
, initializer_func(std::move(initializer_func_))
{
}
2011-10-31 17:55:06 +00:00
String getName() const override { return "Memory"; }
2011-10-31 17:55:06 +00:00
2015-01-18 08:25:56 +00:00
protected:
Chunk generate() override
{
2020-12-13 21:21:25 +00:00
if (initializer_func)
2020-09-04 08:36:47 +00:00
{
2020-12-13 21:21:25 +00:00
initializer_func(data);
initializer_func = {};
2020-09-04 08:36:47 +00:00
}
2020-12-13 21:21:25 +00:00
size_t current_index = getAndIncrementExecutionIndex();
2020-12-13 21:50:55 +00:00
2021-04-06 10:41:48 +00:00
if (!data || current_index >= data->size())
2020-12-13 21:50:55 +00:00
{
return {};
2020-12-13 21:21:25 +00:00
}
2021-02-12 00:25:00 +00:00
const Block & src = (*data)[current_index];
Columns columns;
columns.reserve(columns.size());
/// Add only required columns to `res`.
for (const auto & elem : column_names_and_types)
{
2021-02-12 00:25:00 +00:00
auto current_column = src.getByName(elem.getNameInStorage()).column;
current_column = current_column->decompress();
if (elem.isSubcolumn())
2020-12-22 15:03:48 +00:00
columns.emplace_back(elem.getTypeInStorage()->getSubcolumn(elem.getSubcolumnName(), *current_column));
2020-08-25 23:12:16 +00:00
else
columns.emplace_back(std::move(current_column));
}
2020-08-25 19:46:47 +00:00
2021-02-12 00:25:00 +00:00
return Chunk(std::move(columns), src.rows());
}
2015-01-18 08:25:56 +00:00
private:
2020-12-13 21:21:25 +00:00
size_t getAndIncrementExecutionIndex()
{
if (parallel_execution_index)
{
return (*parallel_execution_index)++;
}
else
{
2020-12-13 21:50:55 +00:00
return execution_index++;
2020-12-13 21:21:25 +00:00
}
}
2020-09-04 08:36:47 +00:00
const NamesAndTypesList column_names_and_types;
2020-12-13 21:21:25 +00:00
size_t execution_index = 0;
2021-02-12 00:25:00 +00:00
std::shared_ptr<const Blocks> data;
2020-12-13 21:21:25 +00:00
std::shared_ptr<std::atomic<size_t>> parallel_execution_index;
InitializerFunc initializer_func;
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:
2021-02-10 17:48:39 +00:00
MemoryBlockOutputStream(
StorageMemory & storage_,
const StorageMetadataPtr & metadata_snapshot_)
: storage(storage_)
, metadata_snapshot(metadata_snapshot_)
2021-02-10 17:48:39 +00:00
{
}
Block getHeader() const override { return metadata_snapshot->getSampleBlock(); }
void write(const Block & block) override
{
2020-06-17 14:32:25 +00:00
metadata_snapshot->check(block, true);
2021-02-07 02:40:06 +00:00
2021-02-12 00:25:00 +00:00
if (storage.compress)
2021-02-07 02:40:06 +00:00
{
2021-02-12 00:25:00 +00:00
Block compressed_block;
2021-02-12 21:26:12 +00:00
for (const auto & elem : block)
2021-02-12 00:25:00 +00:00
compressed_block.insert({ elem.column->compress(), elem.type, elem.name });
2021-02-07 02:40:06 +00:00
2021-02-12 00:25:00 +00:00
new_blocks.emplace_back(compressed_block);
}
else
{
new_blocks.emplace_back(block);
}
2021-02-07 02:40:06 +00:00
}
2021-02-07 02:40:06 +00:00
2021-02-07 02:40:06 +00:00
void writeSuffix() override
{
2021-02-12 00:25:00 +00:00
size_t inserted_bytes = 0;
size_t inserted_rows = 0;
for (const auto & block : new_blocks)
{
inserted_bytes += block.allocatedBytes();
inserted_rows += block.rows();
}
2021-02-07 02:40:06 +00:00
std::lock_guard lock(storage.mutex);
2021-02-12 00:25:00 +00:00
auto new_data = std::make_unique<Blocks>(*(storage.data.get()));
2021-02-07 02:40:06 +00:00
new_data->insert(new_data->end(), new_blocks.begin(), new_blocks.end());
storage.data.set(std::move(new_data));
storage.total_size_bytes.fetch_add(inserted_bytes, std::memory_order_relaxed);
storage.total_size_rows.fetch_add(inserted_rows, std::memory_order_relaxed);
}
2021-02-07 02:40:06 +00:00
2015-01-18 08:25:56 +00:00
private:
2021-02-12 00:25:00 +00:00
Blocks new_blocks;
2021-02-07 02:40:06 +00:00
StorageMemory & storage;
StorageMetadataPtr metadata_snapshot;
2015-01-18 08:25:56 +00:00
};
2011-10-31 17:55:06 +00:00
2021-02-07 01:41:31 +00:00
StorageMemory::StorageMemory(
const StorageID & table_id_,
ColumnsDescription columns_description_,
ConstraintsDescription constraints_,
2021-04-23 12:18:23 +00:00
const String & comment,
2021-02-07 01:41:31 +00:00
bool compress_)
2021-02-12 00:25:00 +00:00
: IStorage(table_id_), data(std::make_unique<const Blocks>()), compress(compress_)
{
2020-06-19 15:39:41 +00:00
StorageInMemoryMetadata storage_metadata;
storage_metadata.setColumns(std::move(columns_description_));
storage_metadata.setConstraints(std::move(constraints_));
2021-04-23 12:18:23 +00:00
storage_metadata.setComment(comment);
2020-06-19 15:39:41 +00:00
setInMemoryMetadata(storage_metadata);
}
2020-08-03 13:54:14 +00:00
Pipe StorageMemory::read(
const Names & column_names,
const StorageMetadataPtr & metadata_snapshot,
SelectQueryInfo & /*query_info*/,
ContextPtr /*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
{
2020-06-19 17:17:13 +00:00
metadata_snapshot->check(column_names, getVirtuals(), getStorageID());
2012-05-30 04:45:49 +00:00
2020-09-04 08:36:47 +00:00
if (delay_read_for_global_subqueries)
{
/// Note: for global subquery we use single source.
/// Mainly, the reason is that at this point table is empty,
/// and we don't know the number of blocks are going to be inserted into it.
///
/// It may seem to be not optimal, but actually data from such table is used to fill
/// set for IN or hash table for JOIN, which can't be done concurrently.
/// Since no other manipulation with data is done, multiple sources shouldn't give any profit.
2020-10-08 07:19:37 +00:00
return Pipe(std::make_shared<MemorySource>(
column_names,
*this,
metadata_snapshot,
2020-12-13 21:21:25 +00:00
nullptr /* data */,
nullptr /* parallel execution index */,
2021-02-12 00:25:00 +00:00
[this](std::shared_ptr<const Blocks> & data_to_initialize)
2020-10-08 11:16:53 +00:00
{
2020-12-13 21:21:25 +00:00
data_to_initialize = data.get();
2020-10-08 07:19:37 +00:00
}));
2020-09-04 08:36:47 +00:00
}
2020-10-08 11:15:11 +00:00
auto current_data = data.get();
2020-10-04 16:28:36 +00:00
size_t size = current_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
Pipes pipes;
2012-05-30 04:45:49 +00:00
2020-12-13 21:21:25 +00:00
auto parallel_execution_index = std::make_shared<std::atomic<size_t>>(0);
2020-08-25 17:42:35 +00:00
2017-06-02 15:54:39 +00:00
for (size_t stream = 0; stream < num_streams; ++stream)
{
2020-12-13 21:21:25 +00:00
pipes.emplace_back(std::make_shared<MemorySource>(column_names, *this, metadata_snapshot, current_data, parallel_execution_index));
}
2020-08-06 12:24:05 +00:00
return Pipe::unitePipes(std::move(pipes));
2011-10-31 17:55:06 +00:00
}
BlockOutputStreamPtr StorageMemory::write(const ASTPtr & /*query*/, const StorageMetadataPtr & metadata_snapshot, ContextPtr /*context*/)
2011-10-31 17:55:06 +00:00
{
return std::make_shared<MemoryBlockOutputStream>(*this, metadata_snapshot);
2011-10-31 17:55:06 +00:00
}
2011-11-05 23:31:19 +00:00
2020-01-22 11:30:11 +00:00
void StorageMemory::drop()
2011-11-05 23:31:19 +00:00
{
2021-02-12 00:25:00 +00:00
data.set(std::make_unique<Blocks>());
total_size_bytes.store(0, std::memory_order_relaxed);
total_size_rows.store(0, std::memory_order_relaxed);
2011-11-05 23:31:19 +00:00
}
2021-02-12 00:25:00 +00:00
static inline void updateBlockData(Block & old_block, const Block & new_block)
2020-09-22 09:23:46 +00:00
{
2021-02-12 00:25:00 +00:00
for (const auto & it : new_block)
2020-09-22 09:23:46 +00:00
{
2021-02-12 00:25:00 +00:00
auto col_name = it.name;
auto & col_with_type_name = old_block.getByName(col_name);
col_with_type_name.column = it.column;
2020-09-22 09:23:46 +00:00
}
}
void StorageMemory::checkMutationIsPossible(const MutationCommands & /*commands*/, const Settings & /*settings*/) const
{
/// Some validation will be added
}
void StorageMemory::mutate(const MutationCommands & commands, ContextPtr context)
2020-09-22 09:23:46 +00:00
{
2020-10-08 07:19:37 +00:00
std::lock_guard lock(mutex);
2020-09-22 09:29:57 +00:00
auto metadata_snapshot = getInMemoryMetadataPtr();
auto storage = getStorageID();
auto storage_ptr = DatabaseCatalog::instance().getTable(storage, context);
2021-05-19 08:56:08 +00:00
2021-06-28 17:02:22 +00:00
/// When max_threads > 1, the order of returning blocks is uncertain,
2021-05-19 08:56:08 +00:00
/// which will lead to inconsistency after updateBlockData.
auto new_context = Context::createCopy(context);
new_context->setSetting("max_streams_to_max_threads_ratio", 1);
new_context->setSetting("max_threads", 1);
auto interpreter = std::make_unique<MutationsInterpreter>(storage_ptr, metadata_snapshot, commands, new_context, true);
2020-09-22 09:23:46 +00:00
auto in = interpreter->execute();
in->readPrefix();
2021-02-12 00:25:00 +00:00
Blocks out;
2021-02-07 01:41:31 +00:00
while (Block block = in->read())
2020-09-22 09:23:46 +00:00
{
2021-02-12 00:25:00 +00:00
if (compress)
for (auto & elem : block)
elem.column = elem.column->compress();
2021-02-07 01:41:31 +00:00
2021-02-12 00:25:00 +00:00
out.push_back(block);
2020-09-22 09:23:46 +00:00
}
in->readSuffix();
2021-02-12 00:25:00 +00:00
std::unique_ptr<Blocks> new_data;
2020-11-23 05:30:36 +00:00
2021-02-12 00:25:00 +00:00
// all column affected
2020-09-22 09:23:46 +00:00
if (interpreter->isAffectingAllColumns())
{
2021-02-12 00:25:00 +00:00
new_data = std::make_unique<Blocks>(out);
2020-09-22 09:23:46 +00:00
}
else
{
2021-02-12 00:25:00 +00:00
/// just some of the column affected, we need update it with new column
new_data = std::make_unique<Blocks>(*(data.get()));
2020-10-04 16:28:36 +00:00
auto data_it = new_data->begin();
2020-09-22 09:23:46 +00:00
auto out_it = out.begin();
2020-11-26 19:22:26 +00:00
2020-11-23 17:45:59 +00:00
while (data_it != new_data->end())
2020-09-22 09:23:46 +00:00
{
2021-02-12 00:25:00 +00:00
/// Mutation does not change the number of blocks
2020-11-23 17:45:59 +00:00
assert(out_it != out.end());
2021-02-12 00:25:00 +00:00
updateBlockData(*data_it, *out_it);
2020-09-22 09:23:46 +00:00
++data_it;
++out_it;
}
2020-11-23 17:45:59 +00:00
assert(out_it == out.end());
2020-09-22 09:23:46 +00:00
}
2020-11-23 05:30:36 +00:00
2021-02-12 00:25:00 +00:00
size_t rows = 0;
2020-11-23 05:30:36 +00:00
size_t bytes = 0;
for (const auto & buffer : *new_data)
{
rows += buffer.rows();
bytes += buffer.bytes();
}
2021-05-17 10:24:01 +00:00
total_size_bytes.store(bytes, std::memory_order_relaxed);
total_size_rows.store(rows, std::memory_order_relaxed);
2020-11-23 05:30:36 +00:00
data.set(std::move(new_data));
2020-09-22 09:23:46 +00:00
}
2020-06-18 10:29:13 +00:00
void StorageMemory::truncate(
const ASTPtr &, const StorageMetadataPtr &, ContextPtr, TableExclusiveLockHolder &)
2018-04-21 00:35:20 +00:00
{
2021-02-12 00:25:00 +00:00
data.set(std::make_unique<Blocks>());
total_size_bytes.store(0, std::memory_order_relaxed);
total_size_rows.store(0, std::memory_order_relaxed);
2018-04-21 00:35:20 +00:00
}
2020-11-25 13:47:32 +00:00
std::optional<UInt64> StorageMemory::totalRows(const Settings &) const
{
/// All modifications of these counters are done under mutex which automatically guarantees synchronization/consistency
/// When run concurrently we are fine with any value: "before" or "after"
return total_size_rows.load(std::memory_order_relaxed);
}
2020-11-25 13:47:32 +00:00
std::optional<UInt64> StorageMemory::totalBytes(const Settings &) const
{
return total_size_bytes.load(std::memory_order_relaxed);
}
void registerStorageMemory(StorageFactory & factory)
{
factory.registerStorage("Memory", [](const StorageFactory::Arguments & args)
{
if (!args.engine_args.empty())
2021-02-07 01:41:31 +00:00
throw Exception(ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH,
"Engine {} doesn't support any arguments ({} given)",
args.engine_name, args.engine_args.size());
bool has_settings = args.storage_def->settings;
MemorySettings settings;
if (has_settings)
settings.loadFromQuery(*args.storage_def);
2021-04-23 12:18:23 +00:00
return StorageMemory::create(args.table_id, args.columns, args.constraints, args.comment, settings.compress);
},
{
2021-02-12 21:26:12 +00:00
.supports_settings = true,
.supports_parallel_insert = true,
});
}
2011-10-31 17:55:06 +00:00
}