thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-24 02:30:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Metadata in read and write methods of IStorage

Browse Source
This commit is contained in:
alesapin 2020-06-15 22:08:58 +03:00
parent b47a7327fd
commit 36ba0192df
84 changed files with 208 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/Core/ExternalTable.cpp
View File

@ -167,7 +167,7 @@ void ExternalTablesHandler::handlePart(const Poco::Net::MessageHeader & header,
auto temporary_table = TemporaryTableHolder(context, ColumnsDescription{columns}, {});
auto storage = temporary_table.getTable();
context.addExternalTable(data->table_name, std::move(temporary_table));
BlockOutputStreamPtr output = storage->write(ASTPtr(), context);
BlockOutputStreamPtr output = storage->write(ASTPtr(), storage->getInMemoryMetadataPtr(), context);
/// Write data
auto sink = std::make_shared<SinkToOutputStream>(std::move(output));

2
src/DataStreams/CreatingSetsBlockInputStream.cpp
View File

@ -101,7 +101,7 @@ void CreatingSetsBlockInputStream::createOne(SubqueryForSet & subquery)
BlockOutputStreamPtr table_out;
if (subquery.table)
table_out = subquery.table->write({}, context);
table_out = subquery.table->write({}, subquery.table->getInMemoryMetadataPtr(), context);
bool done_with_set = !subquery.set;
bool done_with_join = !subquery.join;

2
src/DataStreams/PushingToViewsBlockOutputStream.cpp
View File

@ -107,7 +107,7 @@ PushingToViewsBlockOutputStream::PushingToViewsBlockOutputStream(
/// Do not push to destination table if the flag is set
if (!no_destination)
{
output = storage->write(query_ptr, context);
output = storage->write(query_ptr, storage->getInMemoryMetadataPtr(), context);
replicated_output = dynamic_cast<ReplicatedMergeTreeBlockOutputStream *>(output.get());
}
}

7
src/DataStreams/RemoteQueryExecutor.cpp
View File

@ -319,12 +319,15 @@ void RemoteQueryExecutor::sendExternalTables()
for (const auto & table : external_tables)
{
StoragePtr cur = table.second;
auto metadata_snapshot = cur->getInMemoryMetadataPtr();
QueryProcessingStage::Enum read_from_table_stage = cur->getQueryProcessingStage(context);
Pipes pipes;
pipes = cur->read(cur->getColumns().getNamesOfPhysical(), {}, context,
read_from_table_stage, DEFAULT_BLOCK_SIZE, 1);
pipes = cur->read(
cur->getColumns().getNamesOfPhysical(),
metadata_snapshot, {}, context,
read_from_table_stage, DEFAULT_BLOCK_SIZE, 1);
auto data = std::make_unique<ExternalTableData>();
data->table_name = table.first;

3
src/Interpreters/InterpreterInsertQuery.cpp
View File

@ -117,6 +117,7 @@ BlockIO InterpreterInsertQuery::execute()
StoragePtr table = getTable(query);
auto table_lock = table->lockStructureForShare(
true, context.getInitialQueryId(), context.getSettingsRef().lock_acquire_timeout);
auto metadata_snapshot = table->getInMemoryMetadataPtr();
auto query_sample_block = getSampleBlock(query, table);
if (!query.table_function)
@ -226,7 +227,7 @@ BlockIO InterpreterInsertQuery::execute()
/// NOTE: we explicitly ignore bound materialized views when inserting into Kafka Storage.
/// Otherwise we'll get duplicates when MV reads same rows again from Kafka.
if (table->noPushingToViews() && !no_destination)
out = table->write(query_ptr, context);
out = table->write(query_ptr, metadata_snapshot, context);
else
out = std::make_shared<PushingToViewsBlockOutputStream>(table, context, query_ptr, no_destination);

4
src/Interpreters/InterpreterSelectQuery.cpp
View File

@ -255,6 +255,7 @@ InterpreterSelectQuery::InterpreterSelectQuery(
table_lock = storage->lockStructureForShare(
false, context->getInitialQueryId(), context->getSettingsRef().lock_acquire_timeout);
table_id = storage->getStorageID();
metadata_snapshot = storage->getInMemoryMetadataPtr();
}
if (has_input || !joined_tables.resolveTables())
@ -1293,7 +1294,6 @@ void InterpreterSelectQuery::executeFetchColumns(
else if (storage)
{
/// Table.
if (max_streams == 0)
throw Exception("Logical error: zero number of streams requested", ErrorCodes::LOGICAL_ERROR);
@ -1324,7 +1324,7 @@ void InterpreterSelectQuery::executeFetchColumns(
query_info.input_order_info = query_info.order_optimizer->getInputOrder(storage);
}
Pipes pipes = storage->read(required_columns, query_info, *context, processing_stage, max_block_size, max_streams);
Pipes pipes = storage->read(required_columns, metadata_snapshot, query_info, *context, processing_stage, max_block_size, max_streams);
if (pipes.empty())
{

1
src/Interpreters/InterpreterSelectQuery.h
View File

@ -184,6 +184,7 @@ private:
StoragePtr storage;
StorageID table_id = StorageID::createEmpty(); /// Will be initialized if storage is not nullptr
TableStructureReadLockHolder table_lock;
StorageMetadataPtr metadata_snapshot;
/// Used when we read from prepared input, not table or subquery.
BlockInputStreamPtr input;

2
src/Processors/Transforms/CreatingSetsTransform.cpp
View File

@ -66,7 +66,7 @@ void CreatingSetsTransform::startSubquery(SubqueryForSet & subquery)
elapsed_nanoseconds = 0;
if (subquery.table)
table_out = subquery.table->write({}, context);
table_out = subquery.table->write({}, subquery.table->getInMemoryMetadataPtr(), context);
done_with_set = !subquery.set;
done_with_join = !subquery.join;

3
src/Server/TCPHandler.cpp
View File

@ -952,8 +952,9 @@ bool TCPHandler::receiveData(bool scalar)
storage = temporary_table.getTable();
query_context->addExternalTable(temporary_id.table_name, std::move(temporary_table));
}
auto metadata_snapshot = storage->getInMemoryMetadataPtr();
/// The data will be written directly to the table.
state.io.out = storage->write(ASTPtr(), *query_context);
state.io.out = storage->write(ASTPtr(), metadata_snapshot, *query_context);
}
if (state.need_receive_data_for_input)
state.block_for_input = block;

3
src/Storages/IStorage.h
View File

@ -155,6 +155,7 @@ public: /// thread-unsafe part. lockStructure must be acquired
bool hasSelectQuery() const;
StorageInMemoryMetadata getInMemoryMetadata() const { return *metadata; }
StorageMetadataPtr getInMemoryMetadataPtr() const { return metadata; }
void setInMemoryMetadata(const StorageInMemoryMetadata & metadata_) { metadata = std::make_shared<StorageInMemoryMetadata>(metadata_); }
Block getSampleBlock() const; /// ordinary + materialized.
@ -292,6 +293,7 @@ public:
*/
virtual Pipes read(
const Names & /*column_names*/,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & /*context*/,
QueryProcessingStage::Enum /*processed_stage*/,
@ -309,6 +311,7 @@ public:
*/
virtual BlockOutputStreamPtr write(
const ASTPtr & /*query*/,
const StorageMetadataPtr & /*metadata_snapshot*/,
const Context & /*context*/)
{
throw Exception("Method write is not supported by storage " + getName(), ErrorCodes::NOT_IMPLEMENTED);

3
src/Storages/Kafka/StorageKafka.cpp
View File

@ -201,6 +201,7 @@ String StorageKafka::getDefaultClientId(const StorageID & table_id_)
Pipes StorageKafka::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /* query_info */,
const Context & context,
QueryProcessingStage::Enum /* processed_stage */,
@ -231,7 +232,7 @@ Pipes StorageKafka::read(
}
BlockOutputStreamPtr StorageKafka::write(const ASTPtr &, const Context & context)
BlockOutputStreamPtr StorageKafka::write(const ASTPtr &, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context)
{
auto modified_context = std::make_shared<Context>(context);
modified_context->applySettingsChanges(settings_adjustments);

2
src/Storages/Kafka/StorageKafka.h
View File

@ -39,6 +39,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
@ -47,6 +48,7 @@ public:
BlockOutputStreamPtr write(
const ASTPtr & query,
const StorageMetadataPtr & /*metadata_snapshot*/,
const Context & context) override;
void pushReadBuffer(ConsumerBufferPtr buf);

1
src/Storages/LiveView/StorageBlocks.h
View File

@ -37,6 +37,7 @@ public:
Pipes read(
const Names & /*column_names*/,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & /*context*/,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/LiveView/StorageLiveView.cpp
View File

@ -520,6 +520,7 @@ void StorageLiveView::refresh(const Context & context)
Pipes StorageLiveView::read(
const Names & /*column_names*/,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & /*context*/,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/LiveView/StorageLiveView.h
View File

@ -126,6 +126,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

1
src/Storages/MergeTree/StorageFromMergeTreeDataPart.h
View File

@ -21,6 +21,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,

13
src/Storages/StorageBuffer.cpp
View File

@ -145,6 +145,7 @@ QueryProcessingStage::Enum StorageBuffer::getQueryProcessingStage(const Context
Pipes StorageBuffer::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
@ -157,6 +158,7 @@ Pipes StorageBuffer::read(
{
auto destination = DatabaseCatalog::instance().getTable(destination_id, context);
auto destination_metadata_snapshot = destination->getInMemoryMetadataPtr();
if (destination.get() == this)
throw Exception("Destination table is myself. Read will cause infinite loop.", ErrorCodes::INFINITE_LOOP);
@ -177,7 +179,9 @@ Pipes StorageBuffer::read(
query_info.input_order_info = query_info.order_optimizer->getInputOrder(destination);
/// The destination table has the same structure of the requested columns and we can simply read blocks from there.
pipes_from_dst = destination->read(column_names, query_info, context, processed_stage, max_block_size, num_streams);
pipes_from_dst = destination->read(
column_names, destination_metadata_snapshot, query_info,
context, processed_stage, max_block_size, num_streams);
}
else
{
@ -210,7 +214,10 @@ Pipes StorageBuffer::read(
}
else
{
pipes_from_dst = destination->read(columns_intersection, query_info, context, processed_stage, max_block_size, num_streams);
pipes_from_dst = destination->read(
columns_intersection, destination_metadata_snapshot, query_info,
context, processed_stage, max_block_size, num_streams);
for (auto & pipe : pipes_from_dst)
{
pipe.addSimpleTransform(std::make_shared<AddingMissedTransform>(
@ -425,7 +432,7 @@ private:
};
BlockOutputStreamPtr StorageBuffer::write(const ASTPtr & /*query*/, const Context & /*context*/)
BlockOutputStreamPtr StorageBuffer::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & /*context*/)
{
return std::make_shared<BufferBlockOutputStream>(*this);
}

3
src/Storages/StorageBuffer.h
View File

@ -57,13 +57,14 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
void startup() override;
/// Flush all buffers into the subordinate table and stop background thread.

1
src/Storages/StorageDictionary.cpp
View File

@ -113,6 +113,7 @@ void StorageDictionary::checkTableCanBeDropped() const
Pipes StorageDictionary::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,

4
src/Storages/StorageDictionary.h
View File

@ -16,7 +16,9 @@ public:
void checkTableCanBeDropped() const override;
Pipes read(const Names & column_names,
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

3
src/Storages/StorageDistributed.cpp
View File

@ -464,6 +464,7 @@ QueryProcessingStage::Enum StorageDistributed::getQueryProcessingStage(const Con
Pipes StorageDistributed::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
@ -510,7 +511,7 @@ Pipes StorageDistributed::read(
}
BlockOutputStreamPtr StorageDistributed::write(const ASTPtr &, const Context & context)
BlockOutputStreamPtr StorageDistributed::write(const ASTPtr &, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context)
{
auto cluster = getCluster();
const auto & settings = context.getSettingsRef();

3
src/Storages/StorageDistributed.h
View File

@ -70,13 +70,14 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
/// Removes temporary data in local filesystem.
void truncate(const ASTPtr &, const Context &, TableStructureWriteLockHolder &) override;

2
src/Storages/StorageFile.cpp
View File

@ -377,6 +377,7 @@ private:
Pipes StorageFile::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
@ -479,6 +480,7 @@ private:
BlockOutputStreamPtr StorageFile::write(
const ASTPtr & /*query*/,
const StorageMetadataPtr & /*metadata_snapshot*/,
const Context & context)
{
if (format_name == "Distributed")

2
src/Storages/StorageFile.h
View File

@ -26,6 +26,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
@ -34,6 +35,7 @@ public:
BlockOutputStreamPtr write(
const ASTPtr & query,
const StorageMetadataPtr & /*metadata_snapshot*/,
const Context & context) override;
void truncate(const ASTPtr & /*query*/, const Context & /* context */, TableStructureWriteLockHolder &) override;

1
src/Storages/StorageGenerateRandom.cpp
View File

@ -429,6 +429,7 @@ void registerStorageGenerateRandom(StorageFactory & factory)
Pipes StorageGenerateRandom::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/StorageGenerateRandom.h
View File

@ -17,6 +17,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

5
src/Storages/StorageHDFS.cpp
View File

@ -264,9 +264,10 @@ Strings LSWithRegexpMatching(const String & path_for_ls, const HDFSFSPtr & fs, c
Pipes StorageHDFS::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & context_,
QueryProcessingStage::Enum /*processed_stage*/,
QueryProcessingStage::Enum /*processed_stage*/,
size_t max_block_size,
unsigned num_streams)
{
@ -300,7 +301,7 @@ Pipes StorageHDFS::read(
return pipes;
}
BlockOutputStreamPtr StorageHDFS::write(const ASTPtr & /*query*/, const Context & /*context*/)
BlockOutputStreamPtr StorageHDFS::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & /*context*/)
{
return std::make_shared<HDFSBlockOutputStream>(uri,
format_name,

6
src/Storages/StorageHDFS.h
View File

@ -19,14 +19,16 @@ class StorageHDFS final : public ext::shared_ptr_helper<StorageHDFS>, public ISt
public:
String getName() const override { return "HDFS"; }
Pipes read(const Names & column_names,
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
NamesAndTypesList getVirtuals() const override;

4
src/Storages/StorageInput.cpp
View File

@ -58,7 +58,9 @@ void StorageInput::setInputStream(BlockInputStreamPtr input_stream_)
}
Pipes StorageInput::read(const Names & /*column_names*/,
Pipes StorageInput::read(
const Names & /*column_names*/,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/StorageInput.h
View File

@ -19,6 +19,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

1
src/Storages/StorageJoin.cpp
View File

@ -435,6 +435,7 @@ private:
// TODO: multiple stream read and index read
Pipes StorageJoin::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & /*context*/,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/StorageJoin.h
View File

@ -38,6 +38,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

4
src/Storages/StorageLog.cpp
View File

@ -577,6 +577,7 @@ const StorageLog::Marks & StorageLog::getMarksWithRealRowCount() const
Pipes StorageLog::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
@ -620,8 +621,7 @@ Pipes StorageLog::read(
return pipes;
}
BlockOutputStreamPtr StorageLog::write(
const ASTPtr & /*query*/, const Context & /*context*/)
BlockOutputStreamPtr StorageLog::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & /*context*/)
{
loadMarks();
return std::make_shared<LogBlockOutputStream>(*this);

3
src/Storages/StorageLog.h
View File

@ -26,13 +26,14 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
void rename(const String & new_path_to_table_data, const StorageID & new_table_id) override;

11
src/Storages/StorageMaterializedView.cpp
View File

@ -108,6 +108,7 @@ QueryProcessingStage::Enum StorageMaterializedView::getQueryProcessingStage(cons
Pipes StorageMaterializedView::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
@ -117,11 +118,12 @@ Pipes StorageMaterializedView::read(
auto storage = getTargetTable();
auto lock = storage->lockStructureForShare(
false, context.getCurrentQueryId(), context.getSettingsRef().lock_acquire_timeout);
auto metadata_snapshot = storage->getInMemoryMetadataPtr();
if (query_info.order_optimizer)
query_info.input_order_info = query_info.order_optimizer->getInputOrder(storage);
Pipes pipes = storage->read(column_names, query_info, context, processed_stage, max_block_size, num_streams);
Pipes pipes = storage->read(column_names, metadata_snapshot, query_info, context, processed_stage, max_block_size, num_streams);
for (auto & pipe : pipes)
pipe.addTableLock(lock);
@ -129,12 +131,15 @@ Pipes StorageMaterializedView::read(
return pipes;
}
BlockOutputStreamPtr StorageMaterializedView::write(const ASTPtr & query, const Context & context)
BlockOutputStreamPtr StorageMaterializedView::write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context)
{
auto storage = getTargetTable();
auto lock = storage->lockStructureForShare(
true, context.getCurrentQueryId(), context.getSettingsRef().lock_acquire_timeout);
auto stream = storage->write(query, context);
auto metadata_snapshot = storage->getInMemoryMetadataPtr();
auto stream = storage->write(query, metadata_snapshot, context);
stream->addTableLock(lock);
return stream;
}

3
src/Storages/StorageMaterializedView.h
View File

@ -31,7 +31,7 @@ public:
return getTargetTable()->mayBenefitFromIndexForIn(left_in_operand, query_context);
}
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
void drop() override;
@ -63,6 +63,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

4
src/Storages/StorageMemory.cpp
View File

@ -87,6 +87,7 @@ StorageMemory::StorageMemory(const StorageID & table_id_, ColumnsDescription col
Pipes StorageMemory::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & /*context*/,
QueryProcessingStage::Enum /*processed_stage*/,
@ -119,8 +120,7 @@ Pipes StorageMemory::read(
}
BlockOutputStreamPtr StorageMemory::write(
const ASTPtr & /*query*/, const Context & /*context*/)
BlockOutputStreamPtr StorageMemory::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & /*context*/)
{
return std::make_shared<MemoryBlockOutputStream>(*this);
}

3
src/Storages/StorageMemory.h
View File

@ -30,13 +30,14 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
void drop() override;

4
src/Storages/StorageMerge.cpp
View File

@ -129,6 +129,7 @@ QueryProcessingStage::Enum StorageMerge::getQueryProcessingStage(const Context &
Pipes StorageMerge::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
@ -243,6 +244,7 @@ Pipes StorageMerge::createSources(const SelectQueryInfo & query_info, const Quer
return pipes;
}
auto metadata_snapshot = storage->getInMemoryMetadataPtr();
auto storage_stage = storage->getQueryProcessingStage(*modified_context, QueryProcessingStage::Complete, query_info.query);
if (processed_stage <= storage_stage)
{
@ -250,7 +252,7 @@ Pipes StorageMerge::createSources(const SelectQueryInfo & query_info, const Quer
if (real_column_names.empty())
real_column_names.push_back(ExpressionActions::getSmallestColumn(storage->getColumns().getAllPhysical()));
pipes = storage->read(real_column_names, modified_query_info, *modified_context, processed_stage, max_block_size, UInt32(streams_num));
pipes = storage->read(real_column_names, metadata_snapshot, modified_query_info, *modified_context, processed_stage, max_block_size, UInt32(streams_num));
}
else if (processed_stage > storage_stage)
{

1
src/Storages/StorageMerge.h
View File

@ -31,6 +31,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

3
src/Storages/StorageMergeTree.cpp
View File

@ -177,6 +177,7 @@ StorageMergeTree::~StorageMergeTree()
Pipes StorageMergeTree::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
@ -196,7 +197,7 @@ std::optional<UInt64> StorageMergeTree::totalBytes() const
return getTotalActiveSizeInBytes();
}
BlockOutputStreamPtr StorageMergeTree::write(const ASTPtr & /*query*/, const Context & context)
BlockOutputStreamPtr StorageMergeTree::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context)
{
return std::make_shared<MergeTreeBlockOutputStream>(*this, context.getSettingsRef().max_partitions_per_insert_block);
}

3
src/Storages/StorageMergeTree.h
View File

@ -39,6 +39,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
@ -48,7 +49,7 @@ public:
std::optional<UInt64> totalRows() const override;
std::optional<UInt64> totalBytes() const override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
/** Perform the next step in combining the parts.
*/

4
src/Storages/StorageMySQL.cpp
View File

@ -65,6 +65,7 @@ StorageMySQL::StorageMySQL(
Pipes StorageMySQL::read(
const Names & column_names_,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info_,
const Context & context_,
QueryProcessingStage::Enum /*processed_stage*/,
@ -198,8 +199,7 @@ private:
};
BlockOutputStreamPtr StorageMySQL::write(
const ASTPtr & /*query*/, const Context & context)
BlockOutputStreamPtr StorageMySQL::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context)
{
return std::make_shared<StorageMySQLBlockOutputStream>(*this, remote_database_name, remote_table_name, pool.get(), context.getSettingsRef().mysql_max_rows_to_insert);
}

3
src/Storages/StorageMySQL.h
View File

@ -39,13 +39,14 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
private:
friend class StorageMySQLBlockOutputStream;

3
src/Storages/StorageNull.h
View File

@ -24,6 +24,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo &,
const Context & /*context*/,
QueryProcessingStage::Enum /*processing_stage*/,
@ -35,7 +36,7 @@ public:
return pipes;
}
BlockOutputStreamPtr write(const ASTPtr &, const Context &) override
BlockOutputStreamPtr write(const ASTPtr &, const StorageMetadataPtr & /*metadata_snapshot*/, const Context &) override
{
return std::make_shared<NullBlockOutputStream>(getSampleBlock());
}

3
src/Storages/StorageReplicatedMergeTree.cpp
View File

@ -3381,6 +3381,7 @@ ReplicatedMergeTreeQuorumAddedParts::PartitionIdToMaxBlock StorageReplicatedMerg
Pipes StorageReplicatedMergeTree::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
@ -3442,7 +3443,7 @@ void StorageReplicatedMergeTree::assertNotReadonly() const
}
BlockOutputStreamPtr StorageReplicatedMergeTree::write(const ASTPtr & /*query*/, const Context & context)
BlockOutputStreamPtr StorageReplicatedMergeTree::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context)
{
const auto storage_settings_ptr = getSettings();
assertNotReadonly();

3
src/Storages/StorageReplicatedMergeTree.h
View File

@ -89,6 +89,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
@ -98,7 +99,7 @@ public:
std::optional<UInt64> totalRows() const override;
std::optional<UInt64> totalBytes() const override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
bool optimize(const ASTPtr & query, const ASTPtr & partition, bool final, bool deduplicate, const Context & query_context) override;

3
src/Storages/StorageS3.cpp
View File

@ -285,6 +285,7 @@ Strings listFilesWithRegexpMatching(Aws::S3::S3Client & client, const S3::URI &
Pipes StorageS3::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
@ -320,7 +321,7 @@ Pipes StorageS3::read(
return narrowPipes(std::move(pipes), num_streams);
}
BlockOutputStreamPtr StorageS3::write(const ASTPtr & /*query*/, const Context & /*context*/)
BlockOutputStreamPtr StorageS3::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & /*context*/)
{
return std::make_shared<StorageS3BlockOutputStream>(
format_name, min_upload_part_size, getSampleBlock(), context_global,

3
src/Storages/StorageS3.h
View File

@ -48,13 +48,14 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
NamesAndTypesList getVirtuals() const override;

2
src/Storages/StorageSet.cpp
View File

@ -81,7 +81,7 @@ void SetOrJoinBlockOutputStream::writeSuffix()
}
BlockOutputStreamPtr StorageSetOrJoinBase::write(const ASTPtr & /*query*/, const Context & /*context*/)
BlockOutputStreamPtr StorageSetOrJoinBase::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & /*context*/)
{
UInt64 id = ++increment;
return std::make_shared<SetOrJoinBlockOutputStream>(*this, path, path + "tmp/", toString(id) + ".bin");

2
src/Storages/StorageSet.h
View File

@ -21,7 +21,7 @@ class StorageSetOrJoinBase : public IStorage
public:
void rename(const String & new_path_to_table_data, const StorageID & new_table_id) override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
Strings getDataPaths() const override { return {path}; }

4
src/Storages/StorageStripeLog.cpp
View File

@ -253,6 +253,7 @@ void StorageStripeLog::rename(const String & new_path_to_table_data, const Stora
Pipes StorageStripeLog::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
@ -299,8 +300,7 @@ Pipes StorageStripeLog::read(
}
BlockOutputStreamPtr StorageStripeLog::write(
const ASTPtr & /*query*/, const Context & /*context*/)
BlockOutputStreamPtr StorageStripeLog::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & /*context*/)
{
return std::make_shared<StripeLogBlockOutputStream>(*this);
}

3
src/Storages/StorageStripeLog.h
View File

@ -27,13 +27,14 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
void rename(const String & new_path_to_table_data, const StorageID & new_table_id) override;

4
src/Storages/StorageTinyLog.cpp
View File

@ -394,6 +394,7 @@ void StorageTinyLog::rename(const String & new_path_to_table_data, const Storage
Pipes StorageTinyLog::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
@ -413,8 +414,7 @@ Pipes StorageTinyLog::read(
}
BlockOutputStreamPtr StorageTinyLog::write(
const ASTPtr & /*query*/, const Context & /*context*/)
BlockOutputStreamPtr StorageTinyLog::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & /*context*/)
{
return std::make_shared<TinyLogBlockOutputStream>(*this);
}

3
src/Storages/StorageTinyLog.h
View File

@ -26,13 +26,14 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
void rename(const String & new_path_to_table_data, const StorageID & new_table_id) override;

6
src/Storages/StorageURL.cpp
View File

@ -155,7 +155,9 @@ std::function<void(std::ostream &)> IStorageURLBase::getReadPOSTDataCallback(con
}
Pipes IStorageURLBase::read(const Names & column_names,
Pipes IStorageURLBase::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
@ -183,7 +185,7 @@ Pipes IStorageURLBase::read(const Names & column_names,
return pipes;
}
BlockOutputStreamPtr IStorageURLBase::write(const ASTPtr & /*query*/, const Context & /*context*/)
BlockOutputStreamPtr IStorageURLBase::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & /*context*/)
{
return std::make_shared<StorageURLBlockOutputStream>(
uri, format_name, getSampleBlock(), context_global,

3
src/Storages/StorageURL.h
View File

@ -21,13 +21,14 @@ class IStorageURLBase : public IStorage
public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
protected:
IStorageURLBase(

1
src/Storages/StorageValues.cpp
View File

@ -23,6 +23,7 @@ StorageValues::StorageValues(
Pipes StorageValues::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & /*context*/,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/StorageValues.h
View File

@ -17,6 +17,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

1
src/Storages/StorageView.cpp
View File

@ -54,6 +54,7 @@ StorageView::StorageView(
Pipes StorageView::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/StorageView.h
View File

@ -23,6 +23,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

8
src/Storages/StorageXDBC.cpp
View File

@ -82,7 +82,9 @@ std::function<void(std::ostream &)> StorageXDBC::getReadPOSTDataCallback(const N
return [query](std::ostream & os) { os << "query=" << query; };
}
Pipes StorageXDBC::read(const Names & column_names,
Pipes StorageXDBC::read(
const Names & column_names,
const StorageMetadataPtr & metadata_snapshot,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
@ -92,10 +94,10 @@ Pipes StorageXDBC::read(const Names & column_names,
check(column_names);
bridge_helper->startBridgeSync();
return IStorageURLBase::read(column_names, query_info, context, processed_stage, max_block_size, num_streams);
return IStorageURLBase::read(column_names, metadata_snapshot, query_info, context, processed_stage, max_block_size, num_streams);
}
BlockOutputStreamPtr StorageXDBC::write(const ASTPtr & /*query*/, const Context & context)
BlockOutputStreamPtr StorageXDBC::write(const ASTPtr & /*query*/, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context)
{
bridge_helper->startBridgeSync();

17
src/Storages/StorageXDBC.h
View File

@ -15,13 +15,14 @@ namespace DB
class StorageXDBC : public IStorageURLBase
{
public:
Pipes read(const Names & column_names,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
StorageXDBC(const StorageID & table_id_,
const std::string & remote_database_name,
@ -29,7 +30,7 @@ public:
const ColumnsDescription & columns_,
const Context & context_, BridgeHelperPtr bridge_helper_);
BlockOutputStreamPtr write(const ASTPtr & query, const Context & context) override;
BlockOutputStreamPtr write(const ASTPtr & query, const StorageMetadataPtr & /*metadata_snapshot*/, const Context & context) override;
private:

4
src/Storages/System/IStorageSystemOneBlock.h
View File

@ -28,7 +28,9 @@ public:
setInMemoryMetadata(metadata_);
}
Pipes read(const Names & column_names,
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/System/StorageSystemColumns.cpp
View File

@ -242,6 +242,7 @@ private:
Pipes StorageSystemColumns::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/System/StorageSystemColumns.h
View File

@ -19,6 +19,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

13
src/Storages/System/StorageSystemDetachedParts.cpp
View File

@ -46,12 +46,13 @@ protected:
}
Pipes read(
const Names & /* column_names */,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
const size_t /*max_block_size*/,
const unsigned /*num_streams*/) override
const Names & /* column_names */,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
const size_t /*max_block_size*/,
const unsigned /*num_streams*/) override
{
StoragesInfoStream stream(query_info, context);

1
src/Storages/System/StorageSystemDisks.cpp
View File

@ -28,6 +28,7 @@ StorageSystemDisks::StorageSystemDisks(const std::string & name_)
Pipes StorageSystemDisks::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/System/StorageSystemDisks.h
View File

@ -22,6 +22,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

1
src/Storages/System/StorageSystemNumbers.cpp
View File

@ -125,6 +125,7 @@ StorageSystemNumbers::StorageSystemNumbers(const StorageID & table_id, bool mult
Pipes StorageSystemNumbers::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo &,
const Context & /*context*/,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/System/StorageSystemNumbers.h
View File

@ -31,6 +31,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

1
src/Storages/System/StorageSystemOne.cpp
View File

@ -22,6 +22,7 @@ StorageSystemOne::StorageSystemOne(const std::string & name_)
Pipes StorageSystemOne::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo &,
const Context & /*context*/,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/System/StorageSystemOne.h
View File

@ -23,6 +23,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

13
src/Storages/System/StorageSystemPartsBase.cpp
View File

@ -224,12 +224,13 @@ StoragesInfo StoragesInfoStream::next()
}
Pipes StorageSystemPartsBase::read(
const Names & column_names,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
const size_t /*max_block_size*/,
const unsigned /*num_streams*/)
const Names & column_names,
const StorageMetadataPtr & /*metadata_*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
const size_t /*max_block_size*/,
const unsigned /*num_streams*/)
{
bool has_state_column = hasStateColumn(column_names);

13
src/Storages/System/StorageSystemPartsBase.h
View File

@ -56,12 +56,13 @@ class StorageSystemPartsBase : public IStorage
{
public:
Pipes read(
const Names & column_names,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
const Names & column_names,
const StorageMetadataPtr & metadata_,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
NamesAndTypesList getVirtuals() const override;

1
src/Storages/System/StorageSystemReplicas.cpp
View File

@ -59,6 +59,7 @@ StorageSystemReplicas::StorageSystemReplicas(const std::string & name_)
Pipes StorageSystemReplicas::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/System/StorageSystemReplicas.h
View File

@ -20,6 +20,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

13
src/Storages/System/StorageSystemStoragePolicies.cpp
View File

@ -31,12 +31,13 @@ StorageSystemStoragePolicies::StorageSystemStoragePolicies(const std::string & n
}
Pipes StorageSystemStoragePolicies::read(
const Names & column_names,
const SelectQueryInfo & /*query_info*/,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
const size_t /*max_block_size*/,
const unsigned /*num_streams*/)
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & /*query_info*/,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,
const size_t /*max_block_size*/,
const unsigned /*num_streams*/)
{
check(column_names);

13
src/Storages/System/StorageSystemStoragePolicies.h
View File

@ -21,12 +21,13 @@ public:
std::string getName() const override { return "SystemStoragePolicies"; }
Pipes read(
const Names & column_names,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
protected:
StorageSystemStoragePolicies(const std::string & name_);

1
src/Storages/System/StorageSystemTables.cpp
View File

@ -448,6 +448,7 @@ private:
Pipes StorageSystemTables::read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum /*processed_stage*/,

1
src/Storages/System/StorageSystemTables.h
View File

@ -20,6 +20,7 @@ public:
Pipes read(
const Names & column_names,
const StorageMetadataPtr & /*metadata_*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,

13
src/Storages/System/StorageSystemZeros.cpp
View File

@ -91,12 +91,13 @@ StorageSystemZeros::StorageSystemZeros(const StorageID & table_id_, bool multith
}
Pipes StorageSystemZeros::read(
const Names & column_names,
const SelectQueryInfo &,
const Context & /*context*/,
QueryProcessingStage::Enum /*processed_stage*/,
size_t max_block_size,
unsigned num_streams)
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo &,
const Context & /*context*/,
QueryProcessingStage::Enum /*processed_stage*/,
size_t max_block_size,
unsigned num_streams)
{
check(column_names);

13
src/Storages/System/StorageSystemZeros.h
View File

@ -21,12 +21,13 @@ public:
std::string getName() const override { return "SystemZeros"; }
Pipes read(
const Names & column_names,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
const Names & column_names,
const StorageMetadataPtr & /*metadata_snapshot*/,
const SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
size_t max_block_size,
unsigned num_streams) override;
bool hasEvenlyDistributedRead() const override { return true; }

6
src/Storages/tests/gtest_storage_log.cpp
View File

@ -71,6 +71,7 @@ TYPED_TEST_SUITE(StorageLogTest, DiskImplementations);
std::string writeData(int rows, DB::StoragePtr & table, const DB::Context & context)
{
using namespace DB;
auto metadata_snapshot = table->getInMemoryMetadataPtr();
std::string data;
@ -97,7 +98,7 @@ std::string writeData(int rows, DB::StoragePtr & table, const DB::Context & cont
block.insert(column);
}
BlockOutputStreamPtr out = table->write({}, context);
BlockOutputStreamPtr out = table->write({}, metadata_snapshot, context);
out->write(block);
return data;
@ -107,13 +108,14 @@ std::string writeData(int rows, DB::StoragePtr & table, const DB::Context & cont
std::string readData(DB::StoragePtr & table, const DB::Context & context)
{
using namespace DB;
auto metadata_snapshot = table->getInMemoryMetadataPtr();
Names column_names;
column_names.push_back("a");
QueryProcessingStage::Enum stage = table->getQueryProcessingStage(context);
BlockInputStreamPtr in = std::make_shared<TreeExecutorBlockInputStream>(std::move(table->read(column_names, {}, context, stage, 8192, 1)[0]));
BlockInputStreamPtr in = std::make_shared<TreeExecutorBlockInputStream>(std::move(table->read(column_names, metadata_snapshot, {}, context, stage, 8192, 1)[0]));
Block sample;
{