2021-03-02 12:28:09 +00:00
|
|
|
#include <DataStreams/narrowBlockInputStreams.h>
|
2021-04-01 11:31:46 +00:00
|
|
|
#include <DataStreams/OneBlockInputStream.h>
|
|
|
|
#include <Storages/StorageMerge.h>
|
|
|
|
#include <Storages/StorageFactory.h>
|
|
|
|
#include <Storages/VirtualColumnUtils.h>
|
|
|
|
#include <Storages/AlterCommands.h>
|
2020-11-06 14:07:56 +00:00
|
|
|
#include <Interpreters/Context.h>
|
2021-03-02 12:28:09 +00:00
|
|
|
#include <Interpreters/TreeRewriter.h>
|
2021-04-01 11:31:46 +00:00
|
|
|
#include <Interpreters/ExpressionActions.h>
|
2021-03-02 12:28:09 +00:00
|
|
|
#include <Interpreters/evaluateConstantExpression.h>
|
2021-04-01 11:31:46 +00:00
|
|
|
#include <Interpreters/InterpreterSelectQuery.h>
|
2021-04-01 11:21:36 +00:00
|
|
|
#include <Interpreters/IdentifierSemantic.h>
|
2020-10-14 12:19:29 +00:00
|
|
|
#include <Interpreters/getHeaderForProcessingStage.h>
|
2021-06-25 14:30:58 +00:00
|
|
|
#include <Interpreters/addTypeConversionToAST.h>
|
|
|
|
#include <Interpreters/replaceAliasColumnsInQuery.h>
|
2021-03-02 12:28:09 +00:00
|
|
|
#include <Parsers/ASTSelectQuery.h>
|
2021-04-01 11:31:46 +00:00
|
|
|
#include <Parsers/ASTLiteral.h>
|
|
|
|
#include <Parsers/ASTIdentifier.h>
|
|
|
|
#include <Parsers/ASTExpressionList.h>
|
|
|
|
#include <DataTypes/DataTypeString.h>
|
|
|
|
#include <Columns/ColumnString.h>
|
|
|
|
#include <Common/typeid_cast.h>
|
|
|
|
#include <Common/checkStackSize.h>
|
|
|
|
#include <Databases/IDatabase.h>
|
2021-06-15 19:55:21 +00:00
|
|
|
#include <common/range.h>
|
2021-04-01 11:31:46 +00:00
|
|
|
#include <algorithm>
|
2018-09-20 05:40:06 +00:00
|
|
|
#include <Parsers/queryToString.h>
|
2021-04-01 11:31:46 +00:00
|
|
|
#include <Processors/Transforms/MaterializingTransform.h>
|
2020-01-31 16:29:40 +00:00
|
|
|
#include <Processors/ConcatProcessor.h>
|
2020-11-17 17:16:55 +00:00
|
|
|
#include <Processors/Transforms/ExpressionTransform.h>
|
2014-12-30 18:04:53 +00:00
|
|
|
|
2012-05-30 05:53:09 +00:00
|
|
|
|
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
|
2016-01-11 21:46:36 +00:00
|
|
|
namespace ErrorCodes
|
|
|
|
{
|
2020-02-25 18:02:41 +00:00
|
|
|
extern const int LOGICAL_ERROR;
|
|
|
|
extern const int NOT_IMPLEMENTED;
|
2017-04-01 07:20:54 +00:00
|
|
|
extern const int ILLEGAL_PREWHERE;
|
2017-12-30 00:36:06 +00:00
|
|
|
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
|
2019-05-21 13:04:34 +00:00
|
|
|
extern const int SAMPLING_NOT_SUPPORTED;
|
2021-02-28 05:24:39 +00:00
|
|
|
extern const int ALTER_OF_COLUMN_IS_FORBIDDEN;
|
2016-01-11 21:46:36 +00:00
|
|
|
}
|
|
|
|
|
2014-09-30 03:08:47 +00:00
|
|
|
StorageMerge::StorageMerge(
|
2019-12-04 16:06:55 +00:00
|
|
|
const StorageID & table_id_,
|
2018-03-06 20:18:34 +00:00
|
|
|
const ColumnsDescription & columns_,
|
2021-04-23 12:18:23 +00:00
|
|
|
const String & comment,
|
2021-06-25 13:51:17 +00:00
|
|
|
const String & source_database_name_or_regexp_,
|
|
|
|
bool database_is_regexp_,
|
2021-06-25 08:00:30 +00:00
|
|
|
const DbToTableSetMap & source_databases_and_tables_,
|
2021-04-10 23:33:54 +00:00
|
|
|
ContextPtr context_)
|
2021-06-08 03:11:27 +00:00
|
|
|
: IStorage(table_id_)
|
|
|
|
, WithContext(context_->getGlobalContext())
|
2021-06-25 13:51:17 +00:00
|
|
|
, source_database_regexp(source_database_name_or_regexp_)
|
2021-06-08 03:11:27 +00:00
|
|
|
, source_databases_and_tables(source_databases_and_tables_)
|
2021-06-25 13:51:17 +00:00
|
|
|
, source_database_name_or_regexp(source_database_name_or_regexp_)
|
|
|
|
, database_is_regexp(database_is_regexp_)
|
2020-12-10 20:16:53 +00:00
|
|
|
{
|
|
|
|
StorageInMemoryMetadata storage_metadata;
|
|
|
|
storage_metadata.setColumns(columns_);
|
2021-04-23 12:18:23 +00:00
|
|
|
storage_metadata.setComment(comment);
|
2020-12-10 20:16:53 +00:00
|
|
|
setInMemoryMetadata(storage_metadata);
|
|
|
|
}
|
|
|
|
|
|
|
|
StorageMerge::StorageMerge(
|
|
|
|
const StorageID & table_id_,
|
|
|
|
const ColumnsDescription & columns_,
|
2021-04-23 12:18:23 +00:00
|
|
|
const String & comment,
|
2021-06-25 13:51:17 +00:00
|
|
|
const String & source_database_name_or_regexp_,
|
|
|
|
bool database_is_regexp_,
|
2020-12-10 20:16:53 +00:00
|
|
|
const String & source_table_regexp_,
|
2021-04-10 23:33:54 +00:00
|
|
|
ContextPtr context_)
|
2020-12-10 20:16:53 +00:00
|
|
|
: IStorage(table_id_)
|
2021-04-10 23:33:54 +00:00
|
|
|
, WithContext(context_->getGlobalContext())
|
2021-06-25 13:51:17 +00:00
|
|
|
, source_database_regexp(source_database_name_or_regexp_)
|
2020-12-10 20:16:53 +00:00
|
|
|
, source_table_regexp(source_table_regexp_)
|
2021-06-25 13:51:17 +00:00
|
|
|
, source_database_name_or_regexp(source_database_name_or_regexp_)
|
|
|
|
, database_is_regexp(database_is_regexp_)
|
2012-05-30 05:53:09 +00:00
|
|
|
{
|
2020-06-19 15:39:41 +00:00
|
|
|
StorageInMemoryMetadata storage_metadata;
|
|
|
|
storage_metadata.setColumns(columns_);
|
2021-04-23 12:18:23 +00:00
|
|
|
storage_metadata.setComment(comment);
|
2020-06-19 15:39:41 +00:00
|
|
|
setInMemoryMetadata(storage_metadata);
|
2012-05-30 05:53:09 +00:00
|
|
|
}
|
|
|
|
|
2018-07-05 19:56:51 +00:00
|
|
|
template <typename F>
|
|
|
|
StoragePtr StorageMerge::getFirstTable(F && predicate) const
|
2017-07-12 18:44:27 +00:00
|
|
|
{
|
2021-06-07 09:14:29 +00:00
|
|
|
auto database_table_iterators = getDatabaseIterators(getContext());
|
2017-07-12 18:44:27 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
for (auto & iterator : database_table_iterators)
|
2017-07-12 18:44:27 +00:00
|
|
|
{
|
2021-06-07 09:14:29 +00:00
|
|
|
while (iterator->isValid())
|
|
|
|
{
|
2021-06-13 08:03:19 +00:00
|
|
|
const auto & table = iterator->table();
|
|
|
|
if (table.get() != this && predicate(table))
|
2021-06-25 08:00:30 +00:00
|
|
|
return table;
|
2018-02-23 01:00:47 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
iterator->next();
|
|
|
|
}
|
2017-07-12 18:44:27 +00:00
|
|
|
}
|
2018-02-23 01:00:47 +00:00
|
|
|
|
2018-07-05 19:56:51 +00:00
|
|
|
return {};
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
bool StorageMerge::isRemote() const
|
|
|
|
{
|
2020-06-02 02:06:16 +00:00
|
|
|
auto first_remote_table = getFirstTable([](const StoragePtr & table) { return table && table->isRemote(); });
|
2018-07-05 19:56:51 +00:00
|
|
|
return first_remote_table != nullptr;
|
2017-07-12 18:44:27 +00:00
|
|
|
}
|
|
|
|
|
2017-07-21 20:59:01 +00:00
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
bool StorageMerge::mayBenefitFromIndexForIn(const ASTPtr & left_in_operand, ContextPtr query_context, const StorageMetadataPtr & /*metadata_snapshot*/) const
|
2018-03-16 09:00:04 +00:00
|
|
|
{
|
|
|
|
/// It's beneficial if it is true for at least one table.
|
2021-04-12 19:04:26 +00:00
|
|
|
StorageListWithLocks selected_tables = getSelectedTables(query_context);
|
2018-03-16 09:00:04 +00:00
|
|
|
|
|
|
|
size_t i = 0;
|
|
|
|
for (const auto & table : selected_tables)
|
|
|
|
{
|
2021-06-07 09:14:29 +00:00
|
|
|
const auto & storage_ptr = std::get<1>(table);
|
2020-06-17 09:38:47 +00:00
|
|
|
auto metadata_snapshot = storage_ptr->getInMemoryMetadataPtr();
|
|
|
|
if (storage_ptr->mayBenefitFromIndexForIn(left_in_operand, query_context, metadata_snapshot))
|
2018-03-16 09:00:04 +00:00
|
|
|
return true;
|
|
|
|
|
|
|
|
++i;
|
|
|
|
/// For simplicity reasons, check only first ten tables.
|
|
|
|
if (i > 10)
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
|
|
|
|
|
2021-04-22 13:32:17 +00:00
|
|
|
QueryProcessingStage::Enum StorageMerge::getQueryProcessingStage(
|
|
|
|
ContextPtr local_context,
|
|
|
|
QueryProcessingStage::Enum to_stage,
|
|
|
|
const StorageMetadataPtr &,
|
|
|
|
SelectQueryInfo & query_info) const
|
2018-04-19 14:47:09 +00:00
|
|
|
{
|
2020-11-13 22:56:25 +00:00
|
|
|
/// In case of JOIN the first stage (which includes JOIN)
|
|
|
|
/// should be done on the initiator always.
|
|
|
|
///
|
|
|
|
/// Since in case of JOIN query on shards will receive query w/o JOIN (and their columns).
|
2021-06-12 09:27:05 +00:00
|
|
|
/// (see removeJoin())
|
2020-11-13 22:56:25 +00:00
|
|
|
///
|
|
|
|
/// And for this we need to return FetchColumns.
|
2021-04-01 11:21:36 +00:00
|
|
|
if (const auto * select = query_info.query->as<ASTSelectQuery>(); select && hasJoin(*select))
|
2020-11-13 22:56:25 +00:00
|
|
|
return QueryProcessingStage::FetchColumns;
|
|
|
|
|
2018-04-19 14:47:09 +00:00
|
|
|
auto stage_in_source_tables = QueryProcessingStage::FetchColumns;
|
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
DatabaseTablesIterators database_table_iterators = getDatabaseIterators(local_context);
|
2018-04-19 14:47:09 +00:00
|
|
|
|
2018-09-08 11:29:23 +00:00
|
|
|
size_t selected_table_size = 0;
|
2018-04-19 14:47:09 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
for (const auto & iterator : database_table_iterators)
|
2018-04-19 14:47:09 +00:00
|
|
|
{
|
2021-06-07 09:14:29 +00:00
|
|
|
while (iterator->isValid())
|
2018-04-19 14:47:09 +00:00
|
|
|
{
|
2021-06-07 09:14:29 +00:00
|
|
|
const auto & table = iterator->table();
|
|
|
|
if (table && table.get() != this)
|
|
|
|
{
|
|
|
|
++selected_table_size;
|
|
|
|
stage_in_source_tables = std::max(
|
|
|
|
stage_in_source_tables,
|
|
|
|
table->getQueryProcessingStage(local_context, to_stage, table->getInMemoryMetadataPtr(), query_info));
|
|
|
|
}
|
2018-04-19 14:47:09 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
iterator->next();
|
|
|
|
}
|
2018-04-19 14:47:09 +00:00
|
|
|
}
|
|
|
|
|
2018-09-20 05:40:06 +00:00
|
|
|
return selected_table_size == 1 ? stage_in_source_tables : std::min(stage_in_source_tables, QueryProcessingStage::WithMergeableState);
|
2018-04-19 14:47:09 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
Pipe StorageMerge::read(
|
2017-04-01 07:20:54 +00:00
|
|
|
const Names & column_names,
|
2020-06-16 14:25:08 +00:00
|
|
|
const StorageMetadataPtr & metadata_snapshot,
|
2020-09-20 17:52:17 +00:00
|
|
|
SelectQueryInfo & query_info,
|
2021-04-10 23:33:54 +00:00
|
|
|
ContextPtr local_context,
|
2018-04-19 14:47:09 +00:00
|
|
|
QueryProcessingStage::Enum processed_stage,
|
2019-02-18 23:38:44 +00:00
|
|
|
const size_t max_block_size,
|
2019-07-07 00:20:38 +00:00
|
|
|
unsigned num_streams)
|
2012-05-30 05:53:09 +00:00
|
|
|
{
|
2020-08-06 12:24:05 +00:00
|
|
|
Pipes pipes;
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
bool has_database_virtual_column = false;
|
2018-02-23 01:00:47 +00:00
|
|
|
bool has_table_virtual_column = false;
|
|
|
|
Names real_column_names;
|
|
|
|
real_column_names.reserve(column_names.size());
|
|
|
|
|
2019-01-04 12:10:00 +00:00
|
|
|
for (const auto & column_name : column_names)
|
2018-02-23 01:00:47 +00:00
|
|
|
{
|
2021-06-07 09:14:29 +00:00
|
|
|
if (column_name == "_database" && isVirtualColumn(column_name, metadata_snapshot))
|
|
|
|
has_database_virtual_column = true;
|
|
|
|
else if (column_name == "_table" && isVirtualColumn(column_name, metadata_snapshot))
|
2018-02-23 01:00:47 +00:00
|
|
|
has_table_virtual_column = true;
|
2017-04-01 07:20:54 +00:00
|
|
|
else
|
2019-01-04 12:10:00 +00:00
|
|
|
real_column_names.push_back(column_name);
|
2018-02-23 01:00:47 +00:00
|
|
|
}
|
2017-04-01 07:20:54 +00:00
|
|
|
|
|
|
|
/** Just in case, turn off optimization "transfer to PREWHERE",
|
|
|
|
* since there is no certainty that it works when one of table is MergeTree and other is not.
|
|
|
|
*/
|
2021-04-10 23:33:54 +00:00
|
|
|
auto modified_context = Context::createCopy(local_context);
|
2021-05-08 20:54:49 +00:00
|
|
|
modified_context->setSetting("optimize_move_to_prewhere", Field{false});
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2018-02-23 01:00:47 +00:00
|
|
|
/// What will be result structure depending on query processed stage in source tables?
|
2021-04-10 23:33:54 +00:00
|
|
|
Block header = getHeaderForProcessingStage(*this, column_names, metadata_snapshot, query_info, local_context, processed_stage);
|
2018-09-18 11:09:21 +00:00
|
|
|
|
|
|
|
/** First we make list of selected tables to find out its size.
|
|
|
|
* This is necessary to correctly pass the recommended number of threads to each table.
|
|
|
|
*/
|
2021-06-25 08:00:30 +00:00
|
|
|
StorageListWithLocks selected_tables
|
|
|
|
= getSelectedTables(local_context, query_info.query, has_database_virtual_column, has_table_virtual_column);
|
2018-02-21 06:47:32 +00:00
|
|
|
|
2018-09-18 16:06:15 +00:00
|
|
|
if (selected_tables.empty())
|
2019-05-21 13:04:34 +00:00
|
|
|
/// FIXME: do we support sampling in this case?
|
2020-01-31 16:29:40 +00:00
|
|
|
return createSources(
|
2021-06-07 09:14:29 +00:00
|
|
|
{},
|
|
|
|
query_info,
|
|
|
|
processed_stage,
|
|
|
|
max_block_size,
|
|
|
|
header,
|
|
|
|
{},
|
2021-07-01 13:09:33 +00:00
|
|
|
{},
|
2021-06-07 09:14:29 +00:00
|
|
|
real_column_names,
|
|
|
|
modified_context,
|
|
|
|
0,
|
|
|
|
has_database_virtual_column,
|
|
|
|
has_table_virtual_column);
|
2018-09-18 16:06:15 +00:00
|
|
|
|
2017-04-01 07:20:54 +00:00
|
|
|
size_t tables_count = selected_tables.size();
|
2021-04-10 23:33:54 +00:00
|
|
|
Float64 num_streams_multiplier
|
|
|
|
= std::min(unsigned(tables_count), std::max(1U, unsigned(local_context->getSettingsRef().max_streams_multiplier_for_merge_tables)));
|
2019-07-07 00:20:38 +00:00
|
|
|
num_streams *= num_streams_multiplier;
|
|
|
|
size_t remaining_streams = num_streams;
|
2018-02-21 19:42:42 +00:00
|
|
|
|
2020-05-13 13:49:10 +00:00
|
|
|
InputOrderInfoPtr input_sorting_info;
|
|
|
|
if (query_info.order_optimizer)
|
2019-12-11 01:35:47 +00:00
|
|
|
{
|
|
|
|
for (auto it = selected_tables.begin(); it != selected_tables.end(); ++it)
|
|
|
|
{
|
2021-06-07 09:14:29 +00:00
|
|
|
auto storage_ptr = std::get<1>(*it);
|
2020-06-17 11:05:11 +00:00
|
|
|
auto storage_metadata_snapshot = storage_ptr->getInMemoryMetadataPtr();
|
2021-04-10 23:33:54 +00:00
|
|
|
auto current_info = query_info.order_optimizer->getInputOrder(storage_metadata_snapshot, local_context);
|
2019-12-11 01:35:47 +00:00
|
|
|
if (it == selected_tables.begin())
|
|
|
|
input_sorting_info = current_info;
|
2019-12-11 13:09:46 +00:00
|
|
|
else if (!current_info || (input_sorting_info && *current_info != *input_sorting_info))
|
2019-12-11 01:35:47 +00:00
|
|
|
input_sorting_info.reset();
|
|
|
|
|
|
|
|
if (!input_sorting_info)
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
|
2020-05-13 13:49:10 +00:00
|
|
|
query_info.input_order_info = input_sorting_info;
|
2019-12-11 01:35:47 +00:00
|
|
|
}
|
|
|
|
|
2021-06-24 23:25:06 +00:00
|
|
|
auto sample_block = getInMemoryMetadataPtr()->getSampleBlock();
|
|
|
|
Names required_columns;
|
|
|
|
|
2020-01-31 16:29:40 +00:00
|
|
|
for (const auto & table : selected_tables)
|
2017-04-01 07:20:54 +00:00
|
|
|
{
|
2018-11-26 00:56:50 +00:00
|
|
|
size_t current_need_streams = tables_count >= num_streams ? 1 : (num_streams / tables_count);
|
2018-09-18 11:09:21 +00:00
|
|
|
size_t current_streams = std::min(current_need_streams, remaining_streams);
|
|
|
|
remaining_streams -= current_streams;
|
2019-01-09 15:44:20 +00:00
|
|
|
current_streams = std::max(size_t(1), current_streams);
|
2017-04-01 07:20:54 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
const auto & storage = std::get<1>(table);
|
2017-07-21 20:59:01 +00:00
|
|
|
|
2019-05-21 13:04:34 +00:00
|
|
|
/// If sampling requested, then check that table supports it.
|
2020-03-23 02:12:31 +00:00
|
|
|
if (query_info.query->as<ASTSelectQuery>()->sampleSize() && !storage->supportsSampling())
|
2019-05-21 13:04:34 +00:00
|
|
|
throw Exception("Illegal SAMPLE: table doesn't support sampling", ErrorCodes::SAMPLING_NOT_SUPPORTED);
|
|
|
|
|
2021-06-24 23:25:06 +00:00
|
|
|
Aliases aliases;
|
2020-06-18 09:22:54 +00:00
|
|
|
auto storage_metadata_snapshot = storage->getInMemoryMetadataPtr();
|
2021-06-24 23:25:06 +00:00
|
|
|
auto storage_columns = storage_metadata_snapshot->getColumns();
|
|
|
|
|
|
|
|
if (processed_stage == QueryProcessingStage::FetchColumns && !storage_columns.getAliases().empty())
|
|
|
|
{
|
2021-06-25 14:30:58 +00:00
|
|
|
auto syntax_result = TreeRewriter(local_context).analyzeSelect(query_info.query, TreeRewriterResult({}, storage, storage_metadata_snapshot));
|
|
|
|
ASTPtr required_columns_expr_list = std::make_shared<ASTExpressionList>();
|
2021-06-24 23:25:06 +00:00
|
|
|
|
2021-06-25 14:30:58 +00:00
|
|
|
ASTPtr column_expr;
|
|
|
|
for (const auto & column : real_column_names)
|
|
|
|
{
|
|
|
|
const auto column_default = storage_columns.getDefault(column);
|
|
|
|
bool is_alias = column_default && column_default->kind == ColumnDefaultKind::Alias;
|
2021-06-24 23:25:06 +00:00
|
|
|
|
2021-06-25 14:30:58 +00:00
|
|
|
if (is_alias)
|
2021-06-24 23:25:06 +00:00
|
|
|
{
|
2021-06-25 14:30:58 +00:00
|
|
|
column_expr = column_default->expression->clone();
|
|
|
|
replaceAliasColumnsInQuery(column_expr, storage_metadata_snapshot->getColumns(),
|
|
|
|
syntax_result->array_join_result_to_source, local_context);
|
2021-06-25 07:00:15 +00:00
|
|
|
|
2021-06-25 14:30:58 +00:00
|
|
|
auto column_description = storage_columns.get(column);
|
|
|
|
column_expr = addTypeConversionToAST(std::move(column_expr), column_description.type->getName(),
|
|
|
|
storage_metadata_snapshot->getColumns().getAll(), local_context);
|
|
|
|
column_expr = setAlias(column_expr, column);
|
|
|
|
|
|
|
|
auto type = sample_block.getByName(column).type;
|
|
|
|
aliases.push_back({ .name = column, .type = type, .expression = column_expr->clone() });
|
2021-06-24 23:25:06 +00:00
|
|
|
}
|
|
|
|
else
|
2021-06-25 14:30:58 +00:00
|
|
|
column_expr = std::make_shared<ASTIdentifier>(column);
|
2021-06-24 23:25:06 +00:00
|
|
|
|
2021-06-25 14:30:58 +00:00
|
|
|
required_columns_expr_list->children.emplace_back(std::move(column_expr));
|
|
|
|
}
|
2021-06-24 23:25:06 +00:00
|
|
|
|
2021-06-25 14:30:58 +00:00
|
|
|
syntax_result = TreeRewriter(local_context).analyze(required_columns_expr_list, storage_columns.getAllPhysical(),
|
|
|
|
storage, storage_metadata_snapshot);
|
|
|
|
auto alias_actions = ExpressionAnalyzer(required_columns_expr_list, syntax_result, local_context).getActionsDAG(true);
|
|
|
|
required_columns = alias_actions->getRequiredColumns().getNames();
|
2021-06-24 23:25:06 +00:00
|
|
|
}
|
2020-06-18 09:22:54 +00:00
|
|
|
|
2020-08-06 12:24:05 +00:00
|
|
|
auto source_pipe = createSources(
|
2021-06-07 09:14:29 +00:00
|
|
|
storage_metadata_snapshot,
|
|
|
|
query_info,
|
|
|
|
processed_stage,
|
|
|
|
max_block_size,
|
|
|
|
header,
|
2021-07-01 13:09:33 +00:00
|
|
|
aliases,
|
2021-06-07 09:14:29 +00:00
|
|
|
table,
|
2021-07-01 13:09:33 +00:00
|
|
|
required_columns.empty() ? real_column_names : required_columns,
|
2021-06-07 09:14:29 +00:00
|
|
|
modified_context,
|
|
|
|
current_streams,
|
|
|
|
has_database_virtual_column,
|
|
|
|
has_table_virtual_column);
|
2020-08-06 12:24:05 +00:00
|
|
|
|
|
|
|
pipes.emplace_back(std::move(source_pipe));
|
2017-04-01 07:20:54 +00:00
|
|
|
}
|
|
|
|
|
2020-08-06 12:24:05 +00:00
|
|
|
auto pipe = Pipe::unitePipes(std::move(pipes));
|
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
if (!pipe.empty())
|
2020-10-12 09:58:09 +00:00
|
|
|
// It's possible to have many tables read from merge, resize(num_streams) might open too many files at the same time.
|
|
|
|
// Using narrowPipe instead.
|
2020-08-03 13:54:14 +00:00
|
|
|
narrowPipe(pipe, num_streams);
|
2018-02-22 21:35:17 +00:00
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
return pipe;
|
2012-05-30 05:53:09 +00:00
|
|
|
}
|
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
Pipe StorageMerge::createSources(
|
2020-06-16 15:51:29 +00:00
|
|
|
const StorageMetadataPtr & metadata_snapshot,
|
2020-09-20 17:52:17 +00:00
|
|
|
SelectQueryInfo & query_info,
|
2020-06-16 15:51:29 +00:00
|
|
|
const QueryProcessingStage::Enum & processed_stage,
|
|
|
|
const UInt64 max_block_size,
|
|
|
|
const Block & header,
|
2021-06-24 23:25:06 +00:00
|
|
|
const Aliases & aliases,
|
2020-06-16 15:51:29 +00:00
|
|
|
const StorageWithLockAndName & storage_with_lock,
|
2020-01-31 16:29:40 +00:00
|
|
|
Names & real_column_names,
|
2021-05-31 14:49:02 +00:00
|
|
|
ContextMutablePtr modified_context,
|
2020-06-16 15:51:29 +00:00
|
|
|
size_t streams_num,
|
2021-06-07 09:14:29 +00:00
|
|
|
bool has_database_virtual_column,
|
2020-06-16 15:51:29 +00:00
|
|
|
bool has_table_virtual_column,
|
2020-01-31 16:29:40 +00:00
|
|
|
bool concat_streams)
|
2014-02-11 18:38:21 +00:00
|
|
|
{
|
2021-06-07 09:14:29 +00:00
|
|
|
const auto & [database_name, storage, struct_lock, table_name] = storage_with_lock;
|
2018-11-08 17:28:52 +00:00
|
|
|
SelectQueryInfo modified_query_info = query_info;
|
2018-09-18 11:09:21 +00:00
|
|
|
modified_query_info.query = query_info.query->clone();
|
|
|
|
|
2020-08-14 09:38:18 +00:00
|
|
|
/// Original query could contain JOIN but we need only the first joined table and its columns.
|
2021-04-01 11:31:46 +00:00
|
|
|
auto & modified_select = modified_query_info.query->as<ASTSelectQuery &>();
|
2021-06-12 09:27:05 +00:00
|
|
|
|
|
|
|
TreeRewriterResult new_analyzer_res = *query_info.syntax_analyzer_result;
|
|
|
|
removeJoin(modified_select, new_analyzer_res, modified_context);
|
2021-04-01 11:21:36 +00:00
|
|
|
modified_query_info.syntax_analyzer_result = std::make_shared<TreeRewriterResult>(std::move(new_analyzer_res));
|
2020-08-14 09:38:18 +00:00
|
|
|
|
2019-12-27 19:30:22 +00:00
|
|
|
VirtualColumnUtils::rewriteEntityInAst(modified_query_info.query, "_table", table_name);
|
2021-06-07 09:14:29 +00:00
|
|
|
VirtualColumnUtils::rewriteEntityInAst(modified_query_info.query, "_database", database_name);
|
2018-09-19 10:16:30 +00:00
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
Pipe pipe;
|
2020-01-31 16:29:40 +00:00
|
|
|
|
2018-09-19 10:16:30 +00:00
|
|
|
if (!storage)
|
2020-01-31 16:29:40 +00:00
|
|
|
{
|
2020-08-06 12:24:05 +00:00
|
|
|
pipe = QueryPipeline::getPipe(InterpreterSelectQuery(
|
2021-04-10 23:33:54 +00:00
|
|
|
modified_query_info.query, modified_context,
|
2020-06-17 16:39:58 +00:00
|
|
|
std::make_shared<OneBlockInputStream>(header),
|
2020-08-06 12:24:05 +00:00
|
|
|
SelectQueryOptions(processed_stage).analyze()).execute().pipeline);
|
2020-06-17 16:39:58 +00:00
|
|
|
|
2020-02-26 14:13:41 +00:00
|
|
|
pipe.addInterpreterContext(modified_context);
|
2020-08-03 13:54:14 +00:00
|
|
|
return pipe;
|
2020-01-31 16:29:40 +00:00
|
|
|
}
|
2018-09-18 11:09:21 +00:00
|
|
|
|
2021-04-22 13:32:17 +00:00
|
|
|
auto storage_stage
|
2021-04-21 16:00:27 +00:00
|
|
|
= storage->getQueryProcessingStage(modified_context, QueryProcessingStage::Complete, metadata_snapshot, modified_query_info);
|
2020-04-01 18:38:01 +00:00
|
|
|
if (processed_stage <= storage_stage)
|
2018-09-18 11:09:21 +00:00
|
|
|
{
|
2018-09-18 16:06:15 +00:00
|
|
|
/// If there are only virtual columns in query, you must request at least one other column.
|
2019-12-30 18:20:43 +00:00
|
|
|
if (real_column_names.empty())
|
2020-06-17 16:39:58 +00:00
|
|
|
real_column_names.push_back(ExpressionActions::getSmallestColumn(metadata_snapshot->getColumns().getAllPhysical()));
|
2018-09-18 16:06:15 +00:00
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
pipe = storage->read(
|
|
|
|
real_column_names,
|
|
|
|
metadata_snapshot,
|
|
|
|
modified_query_info,
|
|
|
|
modified_context,
|
|
|
|
processed_stage,
|
|
|
|
max_block_size,
|
|
|
|
UInt32(streams_num));
|
2018-09-18 11:09:21 +00:00
|
|
|
}
|
2020-04-01 18:38:01 +00:00
|
|
|
else if (processed_stage > storage_stage)
|
2018-09-18 11:09:21 +00:00
|
|
|
{
|
2021-06-07 09:14:29 +00:00
|
|
|
modified_select.replaceDatabaseAndTable(database_name, table_name);
|
2018-09-18 11:09:21 +00:00
|
|
|
|
2018-09-20 05:40:06 +00:00
|
|
|
/// Maximum permissible parallelism is streams_num
|
2020-03-13 14:50:26 +00:00
|
|
|
modified_context->setSetting("max_threads", streams_num);
|
|
|
|
modified_context->setSetting("max_streams_to_max_threads_ratio", 1);
|
2018-09-20 05:40:06 +00:00
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
InterpreterSelectQuery interpreter{modified_query_info.query, modified_context, SelectQueryOptions(processed_stage)};
|
2020-02-17 11:50:53 +00:00
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
|
2020-08-06 12:24:05 +00:00
|
|
|
pipe = QueryPipeline::getPipe(interpreter.execute().pipeline);
|
2018-09-18 11:09:21 +00:00
|
|
|
|
|
|
|
/** Materialization is needed, since from distributed storage the constants come materialized.
|
|
|
|
* If you do not do this, different types (Const and non-Const) columns will be produced in different threads,
|
|
|
|
* And this is not allowed, since all code is based on the assumption that in the block stream all types are the same.
|
|
|
|
*/
|
2020-08-06 12:24:05 +00:00
|
|
|
pipe.addSimpleTransform([](const Block & stream_header) { return std::make_shared<MaterializingTransform>(stream_header); });
|
2018-09-18 11:09:21 +00:00
|
|
|
}
|
2014-02-11 18:38:21 +00:00
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
if (!pipe.empty())
|
2018-09-18 16:06:15 +00:00
|
|
|
{
|
2020-08-03 13:54:14 +00:00
|
|
|
if (concat_streams && pipe.numOutputPorts() > 1)
|
2020-10-12 09:58:09 +00:00
|
|
|
// It's possible to have many tables read from merge, resize(1) might open too many files at the same time.
|
|
|
|
// Using concat instead.
|
2020-08-03 13:54:14 +00:00
|
|
|
pipe.addTransform(std::make_shared<ConcatProcessor>(pipe.getHeader(), pipe.numOutputPorts()));
|
2018-09-18 11:09:21 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
if (has_database_virtual_column)
|
|
|
|
{
|
|
|
|
ColumnWithTypeAndName column;
|
|
|
|
column.name = "_database";
|
|
|
|
column.type = std::make_shared<DataTypeString>();
|
|
|
|
column.column = column.type->createColumnConst(0, Field(database_name));
|
|
|
|
|
|
|
|
auto adding_column_dag = ActionsDAG::makeAddingColumnActions(std::move(column));
|
|
|
|
auto adding_column_actions = std::make_shared<ExpressionActions>(
|
|
|
|
std::move(adding_column_dag),
|
|
|
|
ExpressionActionsSettings::fromContext(modified_context, CompileExpressions::yes));
|
|
|
|
|
|
|
|
pipe.addSimpleTransform([&](const Block & stream_header)
|
|
|
|
{
|
|
|
|
return std::make_shared<ExpressionTransform>(stream_header, adding_column_actions);
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
if (has_table_virtual_column)
|
2018-09-19 10:16:30 +00:00
|
|
|
{
|
2020-10-01 17:34:22 +00:00
|
|
|
ColumnWithTypeAndName column;
|
|
|
|
column.name = "_table";
|
|
|
|
column.type = std::make_shared<DataTypeString>();
|
|
|
|
column.column = column.type->createColumnConst(0, Field(table_name));
|
2021-02-04 14:25:11 +00:00
|
|
|
|
|
|
|
auto adding_column_dag = ActionsDAG::makeAddingColumnActions(std::move(column));
|
2021-03-04 17:38:12 +00:00
|
|
|
auto adding_column_actions = std::make_shared<ExpressionActions>(
|
|
|
|
std::move(adding_column_dag),
|
2021-05-19 14:32:07 +00:00
|
|
|
ExpressionActionsSettings::fromContext(modified_context, CompileExpressions::yes));
|
2021-02-04 14:25:11 +00:00
|
|
|
|
2020-10-01 17:34:22 +00:00
|
|
|
pipe.addSimpleTransform([&](const Block & stream_header)
|
2020-08-03 13:54:14 +00:00
|
|
|
{
|
2021-02-04 14:25:11 +00:00
|
|
|
return std::make_shared<ExpressionTransform>(stream_header, adding_column_actions);
|
2020-08-03 13:54:14 +00:00
|
|
|
});
|
|
|
|
}
|
2014-02-11 18:38:21 +00:00
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
/// Subordinary tables could have different but convertible types, like numeric types of different width.
|
|
|
|
/// We must return streams with structure equals to structure of Merge table.
|
2021-07-02 03:44:42 +00:00
|
|
|
convertingSourceStream(header, metadata_snapshot, aliases, modified_context, modified_query_info.query, pipe, processed_stage);
|
2020-02-26 14:13:41 +00:00
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
pipe.addTableLock(struct_lock);
|
2020-10-16 17:16:58 +00:00
|
|
|
pipe.addStorageHolder(storage);
|
2020-08-03 13:54:14 +00:00
|
|
|
pipe.addInterpreterContext(modified_context);
|
2018-09-18 11:09:21 +00:00
|
|
|
}
|
2018-09-19 10:16:30 +00:00
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
return pipe;
|
2014-02-11 18:38:21 +00:00
|
|
|
}
|
|
|
|
|
2020-04-09 18:10:27 +00:00
|
|
|
StorageMerge::StorageListWithLocks StorageMerge::getSelectedTables(
|
2021-06-25 08:00:30 +00:00
|
|
|
ContextPtr query_context,
|
|
|
|
const ASTPtr & query /* = nullptr */,
|
|
|
|
bool filter_by_database_virtual_column /* = false */,
|
|
|
|
bool filter_by_table_virtual_column /* = false */) const
|
2018-09-18 11:09:21 +00:00
|
|
|
{
|
2021-06-25 08:00:30 +00:00
|
|
|
assert(!filter_by_database_virtual_column || !filter_by_table_virtual_column || query);
|
2021-04-12 19:04:26 +00:00
|
|
|
|
|
|
|
const Settings & settings = query_context->getSettingsRef();
|
2018-09-18 11:09:21 +00:00
|
|
|
StorageListWithLocks selected_tables;
|
2021-06-07 09:14:29 +00:00
|
|
|
DatabaseTablesIterators database_table_iterators = getDatabaseIterators(getContext());
|
2018-09-18 11:09:21 +00:00
|
|
|
|
2021-06-25 08:00:30 +00:00
|
|
|
MutableColumnPtr database_name_virtual_column;
|
2021-04-12 19:04:26 +00:00
|
|
|
MutableColumnPtr table_name_virtual_column;
|
2021-06-25 08:00:30 +00:00
|
|
|
if (filter_by_database_virtual_column)
|
|
|
|
{
|
|
|
|
database_name_virtual_column = ColumnString::create();
|
|
|
|
}
|
|
|
|
|
|
|
|
if (filter_by_table_virtual_column)
|
|
|
|
{
|
2021-04-12 19:04:26 +00:00
|
|
|
table_name_virtual_column = ColumnString::create();
|
2021-06-25 08:00:30 +00:00
|
|
|
}
|
2018-09-18 11:09:21 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
for (const auto & iterator : database_table_iterators)
|
2018-09-18 11:09:21 +00:00
|
|
|
{
|
2021-06-25 08:00:30 +00:00
|
|
|
if (filter_by_database_virtual_column)
|
|
|
|
database_name_virtual_column->insert(iterator->databaseName());
|
2021-06-07 09:14:29 +00:00
|
|
|
while (iterator->isValid())
|
|
|
|
{
|
|
|
|
StoragePtr storage = iterator->table();
|
|
|
|
if (!storage)
|
|
|
|
continue;
|
2018-09-18 11:09:21 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
if (query && query->as<ASTSelectQuery>()->prewhere() && !storage->supportsPrewhere())
|
|
|
|
throw Exception("Storage " + storage->getName() + " doesn't support PREWHERE.", ErrorCodes::ILLEGAL_PREWHERE);
|
2018-09-18 11:09:21 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
if (storage.get() != this)
|
|
|
|
{
|
|
|
|
auto table_lock = storage->lockForShare(query_context->getCurrentQueryId(), settings.lock_acquire_timeout);
|
|
|
|
selected_tables.emplace_back(iterator->databaseName(), storage, std::move(table_lock), iterator->name());
|
2021-06-25 08:00:30 +00:00
|
|
|
if (filter_by_table_virtual_column)
|
2021-06-07 09:14:29 +00:00
|
|
|
table_name_virtual_column->insert(iterator->name());
|
|
|
|
}
|
2018-09-18 11:09:21 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
iterator->next();
|
|
|
|
}
|
2018-09-18 11:09:21 +00:00
|
|
|
}
|
|
|
|
|
2021-06-25 08:00:30 +00:00
|
|
|
if (filter_by_database_virtual_column)
|
|
|
|
{
|
|
|
|
/// Filter names of selected tables if there is a condition on "_database" virtual column in WHERE clause
|
|
|
|
Block virtual_columns_block
|
|
|
|
= Block{ColumnWithTypeAndName(std::move(database_name_virtual_column), std::make_shared<DataTypeString>(), "_database")};
|
|
|
|
VirtualColumnUtils::filterBlockWithQuery(query, virtual_columns_block, query_context);
|
|
|
|
auto values = VirtualColumnUtils::extractSingleValueFromBlock<String>(virtual_columns_block, "_database");
|
|
|
|
|
|
|
|
/// Remove unused databases from the list
|
|
|
|
selected_tables.remove_if([&](const auto & elem) { return values.find(std::get<0>(elem)) == values.end(); });
|
|
|
|
}
|
|
|
|
|
|
|
|
if (filter_by_table_virtual_column)
|
2018-09-18 11:09:21 +00:00
|
|
|
{
|
2021-04-12 19:04:26 +00:00
|
|
|
/// Filter names of selected tables if there is a condition on "_table" virtual column in WHERE clause
|
|
|
|
Block virtual_columns_block = Block{ColumnWithTypeAndName(std::move(table_name_virtual_column), std::make_shared<DataTypeString>(), "_table")};
|
|
|
|
VirtualColumnUtils::filterBlockWithQuery(query, virtual_columns_block, query_context);
|
2018-09-18 11:09:21 +00:00
|
|
|
auto values = VirtualColumnUtils::extractSingleValueFromBlock<String>(virtual_columns_block, "_table");
|
|
|
|
|
|
|
|
/// Remove unused tables from the list
|
2021-06-07 09:14:29 +00:00
|
|
|
selected_tables.remove_if([&](const auto & elem) { return values.find(std::get<3>(elem)) == values.end(); });
|
2018-09-18 11:09:21 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
return selected_tables;
|
|
|
|
}
|
|
|
|
|
2021-06-25 13:51:17 +00:00
|
|
|
DatabaseTablesIteratorPtr StorageMerge::getDatabaseIterator(const String & database_name, ContextPtr local_context) const
|
|
|
|
{
|
|
|
|
auto database = DatabaseCatalog::instance().getDatabase(database_name);
|
|
|
|
|
|
|
|
auto table_name_match = [this, &database_name](const String & table_name_) -> bool {
|
|
|
|
if (source_databases_and_tables)
|
|
|
|
{
|
|
|
|
const auto & source_tables = (*source_databases_and_tables).at(database_name);
|
|
|
|
return source_tables.count(table_name_);
|
|
|
|
}
|
|
|
|
else
|
|
|
|
return source_table_regexp->match(table_name_);
|
|
|
|
};
|
|
|
|
|
|
|
|
return database->getTablesIterator(local_context, table_name_match);
|
|
|
|
}
|
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
StorageMerge::DatabaseTablesIterators StorageMerge::getDatabaseIterators(ContextPtr local_context) const
|
2019-06-02 12:11:01 +00:00
|
|
|
{
|
2020-12-07 07:23:58 +00:00
|
|
|
try
|
|
|
|
{
|
|
|
|
checkStackSize();
|
|
|
|
}
|
|
|
|
catch (Exception & e)
|
|
|
|
{
|
|
|
|
e.addMessage("while getting table iterator of Merge table. Maybe caused by two Merge tables that will endlessly try to read each other's data");
|
|
|
|
throw;
|
|
|
|
}
|
2020-12-10 20:16:53 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
DatabaseTablesIterators database_table_iterators;
|
2020-12-10 20:16:53 +00:00
|
|
|
|
2021-06-25 13:51:17 +00:00
|
|
|
/// database_name argument is not a regexp
|
|
|
|
if (!database_is_regexp)
|
|
|
|
database_table_iterators.emplace_back(getDatabaseIterator(source_database_name_or_regexp, local_context));
|
2021-06-07 09:14:29 +00:00
|
|
|
|
2021-06-25 13:51:17 +00:00
|
|
|
/// database_name argument is a regexp
|
|
|
|
else
|
2020-12-10 20:16:53 +00:00
|
|
|
{
|
2021-06-25 13:51:17 +00:00
|
|
|
auto databases = DatabaseCatalog::instance().getDatabases();
|
|
|
|
|
|
|
|
for (const auto & db : databases)
|
2021-06-07 09:14:29 +00:00
|
|
|
{
|
2021-06-25 13:51:17 +00:00
|
|
|
if (source_database_regexp->match(db.first))
|
|
|
|
database_table_iterators.emplace_back(getDatabaseIterator(db.first, local_context));
|
2021-06-07 09:14:29 +00:00
|
|
|
}
|
|
|
|
}
|
2020-12-10 20:16:53 +00:00
|
|
|
|
2021-06-07 09:14:29 +00:00
|
|
|
return database_table_iterators;
|
2019-06-02 12:11:01 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
void StorageMerge::checkAlterIsPossible(const AlterCommands & commands, ContextPtr local_context) const
|
2019-12-26 18:17:05 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
auto name_deps = getDependentViewsByColumn(local_context);
|
2019-12-26 18:17:05 +00:00
|
|
|
for (const auto & command : commands)
|
|
|
|
{
|
|
|
|
if (command.type != AlterCommand::Type::ADD_COLUMN && command.type != AlterCommand::Type::MODIFY_COLUMN
|
|
|
|
&& command.type != AlterCommand::Type::DROP_COLUMN && command.type != AlterCommand::Type::COMMENT_COLUMN)
|
|
|
|
throw Exception(
|
|
|
|
"Alter of type '" + alterTypeToString(command.type) + "' is not supported by storage " + getName(),
|
|
|
|
ErrorCodes::NOT_IMPLEMENTED);
|
2021-04-30 05:02:32 +00:00
|
|
|
if (command.type == AlterCommand::Type::DROP_COLUMN && !command.clear)
|
2021-02-28 05:24:39 +00:00
|
|
|
{
|
2021-02-28 07:42:08 +00:00
|
|
|
const auto & deps_mv = name_deps[command.column_name];
|
2021-02-28 05:24:39 +00:00
|
|
|
if (!deps_mv.empty())
|
|
|
|
{
|
|
|
|
throw Exception(
|
|
|
|
"Trying to ALTER DROP column " + backQuoteIfNeed(command.column_name) + " which is referenced by materialized view "
|
|
|
|
+ toString(deps_mv),
|
|
|
|
ErrorCodes::ALTER_OF_COLUMN_IS_FORBIDDEN);
|
|
|
|
}
|
|
|
|
}
|
2019-12-26 18:17:05 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2019-03-05 10:12:20 +00:00
|
|
|
void StorageMerge::alter(
|
2021-04-10 23:33:54 +00:00
|
|
|
const AlterCommands & params, ContextPtr local_context, TableLockHolder &)
|
2013-09-23 12:01:19 +00:00
|
|
|
{
|
2019-12-03 16:25:32 +00:00
|
|
|
auto table_id = getStorageID();
|
2016-05-13 21:08:19 +00:00
|
|
|
|
2019-12-26 18:17:05 +00:00
|
|
|
StorageInMemoryMetadata storage_metadata = getInMemoryMetadata();
|
2021-04-10 23:33:54 +00:00
|
|
|
params.apply(storage_metadata, local_context);
|
|
|
|
DatabaseCatalog::instance().getDatabase(table_id.database_name)->alterTable(local_context, table_id, storage_metadata);
|
2020-06-15 16:55:33 +00:00
|
|
|
setInMemoryMetadata(storage_metadata);
|
2013-09-23 12:01:19 +00:00
|
|
|
}
|
2014-07-11 08:12:03 +00:00
|
|
|
|
2020-06-16 15:51:29 +00:00
|
|
|
void StorageMerge::convertingSourceStream(
|
|
|
|
const Block & header,
|
|
|
|
const StorageMetadataPtr & metadata_snapshot,
|
2021-06-24 23:25:06 +00:00
|
|
|
const Aliases & aliases,
|
2021-04-10 23:33:54 +00:00
|
|
|
ContextPtr local_context,
|
2020-06-16 15:51:29 +00:00
|
|
|
ASTPtr & query,
|
|
|
|
Pipe & pipe,
|
|
|
|
QueryProcessingStage::Enum processed_stage)
|
2018-09-19 10:16:30 +00:00
|
|
|
{
|
2020-01-31 16:29:40 +00:00
|
|
|
Block before_block_header = pipe.getHeader();
|
2020-11-17 17:16:55 +00:00
|
|
|
|
2021-06-24 23:25:06 +00:00
|
|
|
auto storage_sample_block = metadata_snapshot->getSampleBlock();
|
|
|
|
auto pipe_columns = pipe.getHeader().getNamesAndTypesList();
|
2020-11-17 17:16:55 +00:00
|
|
|
|
2021-06-24 23:25:06 +00:00
|
|
|
for (const auto & alias : aliases)
|
2020-08-03 13:54:14 +00:00
|
|
|
{
|
2021-06-24 23:25:06 +00:00
|
|
|
pipe_columns.emplace_back(NameAndTypePair(alias.name, alias.type));
|
|
|
|
ASTPtr expr = std::move(alias.expression);
|
|
|
|
auto syntax_result = TreeRewriter(local_context).analyze(expr, pipe_columns);
|
|
|
|
auto expression_analyzer = ExpressionAnalyzer{alias.expression, syntax_result, local_context};
|
|
|
|
|
|
|
|
auto dag = std::make_shared<ActionsDAG>(pipe_columns);
|
|
|
|
auto actions_dag = expression_analyzer.getActionsDAG(true, false);
|
|
|
|
auto actions = std::make_shared<ExpressionActions>(actions_dag, ExpressionActionsSettings::fromContext(local_context, CompileExpressions::yes));
|
|
|
|
|
|
|
|
pipe.addSimpleTransform([&](const Block & stream_header)
|
|
|
|
{
|
|
|
|
return std::make_shared<ExpressionTransform>(stream_header, actions);
|
|
|
|
});
|
|
|
|
}
|
2020-11-17 17:16:55 +00:00
|
|
|
|
2020-08-03 13:54:14 +00:00
|
|
|
{
|
2021-06-24 23:25:06 +00:00
|
|
|
auto convert_actions_dag = ActionsDAG::makeConvertingActions(pipe.getHeader().getColumnsWithTypeAndName(),
|
|
|
|
header.getColumnsWithTypeAndName(),
|
|
|
|
ActionsDAG::MatchColumnsMode::Name);
|
|
|
|
auto actions = std::make_shared<ExpressionActions>(convert_actions_dag, ExpressionActionsSettings::fromContext(local_context, CompileExpressions::yes));
|
|
|
|
pipe.addSimpleTransform([&](const Block & stream_header)
|
|
|
|
{
|
|
|
|
return std::make_shared<ExpressionTransform>(stream_header, actions);
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
2018-09-19 10:16:30 +00:00
|
|
|
|
2019-04-09 14:22:35 +00:00
|
|
|
auto where_expression = query->as<ASTSelectQuery>()->where();
|
2018-09-19 10:16:30 +00:00
|
|
|
|
2018-09-20 05:40:06 +00:00
|
|
|
if (!where_expression)
|
2018-09-19 10:16:30 +00:00
|
|
|
return;
|
|
|
|
|
2021-06-15 19:55:21 +00:00
|
|
|
for (size_t column_index : collections::range(0, header.columns()))
|
2018-09-19 10:16:30 +00:00
|
|
|
{
|
2018-09-20 05:40:06 +00:00
|
|
|
ColumnWithTypeAndName header_column = header.getByPosition(column_index);
|
|
|
|
ColumnWithTypeAndName before_column = before_block_header.getByName(header_column.name);
|
|
|
|
/// If the processed_stage greater than FetchColumns and the block structure between streams is different.
|
|
|
|
/// the where expression maybe invalid because of convertingBlockInputStream.
|
|
|
|
/// So we need to throw exception.
|
|
|
|
if (!header_column.type->equals(*before_column.type.get()) && processed_stage > QueryProcessingStage::FetchColumns)
|
2018-09-19 10:16:30 +00:00
|
|
|
{
|
2020-06-16 15:51:29 +00:00
|
|
|
NamesAndTypesList source_columns = metadata_snapshot->getSampleBlock().getNamesAndTypesList();
|
2020-04-27 15:38:35 +00:00
|
|
|
auto virtual_column = *getVirtuals().tryGetByName("_table");
|
2020-04-24 10:20:03 +00:00
|
|
|
source_columns.emplace_back(NameAndTypePair{virtual_column.name, virtual_column.type});
|
2021-04-10 23:33:54 +00:00
|
|
|
auto syntax_result = TreeRewriter(local_context).analyze(where_expression, source_columns);
|
|
|
|
ExpressionActionsPtr actions = ExpressionAnalyzer{where_expression, syntax_result, local_context}.getActions(false, false);
|
2018-09-20 05:40:06 +00:00
|
|
|
Names required_columns = actions->getRequiredColumns();
|
2018-09-19 10:16:30 +00:00
|
|
|
|
2019-01-04 12:10:00 +00:00
|
|
|
for (const auto & required_column : required_columns)
|
2018-09-20 05:40:06 +00:00
|
|
|
{
|
|
|
|
if (required_column == header_column.name)
|
|
|
|
throw Exception("Block structure mismatch in Merge Storage: different types:\n" + before_block_header.dumpStructure()
|
2020-07-28 18:35:18 +00:00
|
|
|
+ "\n" + header.dumpStructure(), ErrorCodes::LOGICAL_ERROR);
|
2018-09-20 05:40:06 +00:00
|
|
|
}
|
2018-09-19 10:16:30 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-10-08 20:39:24 +00:00
|
|
|
IStorage::ColumnSizeByName StorageMerge::getColumnSizes() const
|
|
|
|
{
|
|
|
|
|
|
|
|
auto first_materialize_mysql = getFirstTable([](const StoragePtr & table) { return table && table->getName() == "MaterializeMySQL"; });
|
|
|
|
if (!first_materialize_mysql)
|
|
|
|
return {};
|
|
|
|
return first_materialize_mysql->getColumnSizes();
|
|
|
|
}
|
2017-12-30 00:36:06 +00:00
|
|
|
|
|
|
|
void registerStorageMerge(StorageFactory & factory)
|
|
|
|
{
|
|
|
|
factory.registerStorage("Merge", [](const StorageFactory::Arguments & args)
|
|
|
|
{
|
|
|
|
/** In query, the name of database is specified as table engine argument which contains source tables,
|
|
|
|
* as well as regex for source-table names.
|
|
|
|
*/
|
|
|
|
|
|
|
|
ASTs & engine_args = args.engine_args;
|
|
|
|
|
|
|
|
if (engine_args.size() != 2)
|
|
|
|
throw Exception("Storage Merge requires exactly 2 parameters"
|
|
|
|
" - name of source database and regexp for table names.",
|
|
|
|
ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
|
|
|
|
|
2021-06-27 06:09:23 +00:00
|
|
|
auto [is_regexp, database_ast] = evaluateDatabaseNameForMergeEngine(engine_args[0], args.getLocalContext());
|
|
|
|
|
|
|
|
if (!is_regexp)
|
|
|
|
engine_args[0] = database_ast;
|
|
|
|
String source_database_name_or_regexp = database_ast->as<ASTLiteral &>().value.safeGet<String>();
|
2017-12-30 00:36:06 +00:00
|
|
|
|
2021-06-25 15:01:57 +00:00
|
|
|
engine_args[1] = evaluateConstantExpressionAsLiteral(engine_args[1], args.getLocalContext());
|
2019-03-15 17:09:14 +00:00
|
|
|
String table_name_regexp = engine_args[1]->as<ASTLiteral &>().value.safeGet<String>();
|
2017-12-30 00:36:06 +00:00
|
|
|
|
2021-06-04 14:48:48 +00:00
|
|
|
return StorageMerge::create(
|
2021-06-25 15:01:57 +00:00
|
|
|
args.table_id, args.columns, args.comment, source_database_name_or_regexp, is_regexp, table_name_regexp, args.getContext());
|
2017-12-30 00:36:06 +00:00
|
|
|
});
|
|
|
|
}
|
2020-04-28 10:38:57 +00:00
|
|
|
|
|
|
|
NamesAndTypesList StorageMerge::getVirtuals() const
|
2020-04-27 13:55:30 +00:00
|
|
|
{
|
2021-06-07 09:14:29 +00:00
|
|
|
NamesAndTypesList virtuals{{"_database", std::make_shared<DataTypeString>()}, {"_table", std::make_shared<DataTypeString>()}};
|
2020-04-28 10:38:57 +00:00
|
|
|
|
2020-06-02 02:06:16 +00:00
|
|
|
auto first_table = getFirstTable([](auto && table) { return table; });
|
2020-04-28 10:38:57 +00:00
|
|
|
if (first_table)
|
|
|
|
{
|
|
|
|
auto table_virtuals = first_table->getVirtuals();
|
|
|
|
virtuals.insert(virtuals.end(), table_virtuals.begin(), table_virtuals.end());
|
|
|
|
}
|
|
|
|
|
2020-04-27 17:46:51 +00:00
|
|
|
return virtuals;
|
2020-04-27 13:55:30 +00:00
|
|
|
}
|
2013-09-23 12:01:19 +00:00
|
|
|
}
|