ClickHouse/src/Storages/StorageDistributed.cpp

1078 lines
39 KiB
C++
Raw Normal View History

2018-12-19 12:38:13 +00:00
#include <Storages/StorageDistributed.h>
#include <Databases/IDatabase.h>
#include <Disks/IDisk.h>
#include <DataTypes/DataTypeFactory.h>
#include <DataTypes/DataTypeUUID.h>
2018-12-19 12:38:13 +00:00
#include <DataTypes/DataTypesNumber.h>
2018-12-19 12:38:13 +00:00
#include <Storages/Distributed/DistributedBlockOutputStream.h>
#include <Storages/StorageFactory.h>
2018-12-25 23:14:39 +00:00
#include <Storages/AlterCommands.h>
2020-04-10 09:24:16 +00:00
#include <Columns/ColumnConst.h>
#include <Common/Macros.h>
#include <Common/escapeForFileName.h>
2017-07-13 20:58:19 +00:00
#include <Common/typeid_cast.h>
#include <Common/quoteString.h>
#include <Common/randomSeed.h>
#include <Common/formatReadable.h>
2018-12-19 12:38:13 +00:00
#include <Parsers/ASTDropQuery.h>
#include <Parsers/ASTExpressionList.h>
#include <Parsers/ASTIdentifier.h>
#include <Parsers/ASTInsertQuery.h>
2018-12-19 12:38:13 +00:00
#include <Parsers/ASTLiteral.h>
#include <Parsers/ASTSelectQuery.h>
2018-12-19 12:38:13 +00:00
#include <Parsers/ASTTablesInSelectQuery.h>
#include <Parsers/ParserAlterQuery.h>
2018-12-19 12:38:13 +00:00
#include <Parsers/TablePropertiesQueriesASTs.h>
#include <Parsers/parseQuery.h>
2018-12-19 12:38:13 +00:00
#include <Interpreters/ClusterProxy/SelectStreamFactory.h>
#include <Interpreters/ClusterProxy/executeQuery.h>
#include <Interpreters/Cluster.h>
2018-12-19 12:38:13 +00:00
#include <Interpreters/ExpressionAnalyzer.h>
#include <Interpreters/InterpreterAlterQuery.h>
#include <Interpreters/InterpreterDescribeQuery.h>
2018-12-19 12:38:13 +00:00
#include <Interpreters/InterpreterSelectQuery.h>
#include <Interpreters/TranslateQualifiedNamesVisitor.h>
#include <Interpreters/TreeRewriter.h>
#include <Interpreters/Context.h>
2018-12-19 12:38:13 +00:00
#include <Interpreters/createBlockSelector.h>
#include <Interpreters/evaluateConstantExpression.h>
#include <Interpreters/getClusterName.h>
#include <Interpreters/getTableExpressions.h>
#include <Functions/IFunction.h>
#include <Core/Field.h>
#include <Core/Settings.h>
2012-05-21 20:38:34 +00:00
2018-06-05 19:46:49 +00:00
#include <IO/ReadHelpers.h>
2020-11-09 19:07:38 +00:00
#include <IO/WriteBufferFromString.h>
#include <IO/Operators.h>
2018-06-05 19:46:49 +00:00
2016-12-12 03:33:34 +00:00
#include <Poco/DirectoryIterator.h>
2015-02-10 21:10:58 +00:00
#include <memory>
#include <filesystem>
#include <optional>
2020-09-18 19:25:56 +00:00
#include <cassert>
2016-12-12 03:33:34 +00:00
namespace
{
2020-03-09 01:03:43 +00:00
const UInt64 FORCE_OPTIMIZE_SKIP_UNUSED_SHARDS_HAS_SHARDING_KEY = 1;
const UInt64 FORCE_OPTIMIZE_SKIP_UNUSED_SHARDS_ALWAYS = 2;
const UInt64 DISTRIBUTED_GROUP_BY_NO_MERGE_AFTER_AGGREGATION = 2;
}
2012-05-21 20:38:34 +00:00
namespace DB
{
namespace ErrorCodes
{
2020-02-25 18:02:41 +00:00
extern const int NOT_IMPLEMENTED;
extern const int STORAGE_REQUIRES_PARAMETER;
extern const int BAD_ARGUMENTS;
extern const int NUMBER_OF_ARGUMENTS_DOESNT_MATCH;
extern const int INCORRECT_NUMBER_OF_COLUMNS;
extern const int INFINITE_LOOP;
2018-06-05 19:46:49 +00:00
extern const int TYPE_MISMATCH;
2018-12-19 12:38:13 +00:00
extern const int TOO_MANY_ROWS;
extern const int UNABLE_TO_SKIP_UNUSED_SHARDS;
extern const int INVALID_SHARD_ID;
extern const int ALTER_OF_COLUMN_IS_FORBIDDEN;
extern const int DISTRIBUTED_TOO_MANY_PENDING_BYTES;
}
namespace ActionLocks
{
2019-04-22 15:11:16 +00:00
extern const StorageActionBlockType DistributedSend;
}
2014-08-21 12:07:29 +00:00
namespace
{
2018-07-25 12:31:47 +00:00
/// select query has database, table and table function names as AST pointers
/// Creates a copy of query, changes database, table and table function names.
ASTPtr rewriteSelectQuery(const ASTPtr & query, const std::string & database, const std::string & table, ASTPtr table_function_ptr = nullptr)
{
2018-02-25 00:50:53 +00:00
auto modified_query_ast = query->clone();
ASTSelectQuery & select_query = modified_query_ast->as<ASTSelectQuery &>();
// Get rid of the settings clause so we don't send them to remote. Thus newly non-important
// settings won't break any remote parser. It's also more reasonable since the query settings
// are written into the query context and will be sent by the query pipeline.
select_query.setExpression(ASTSelectQuery::Expression::SETTINGS, {});
if (table_function_ptr)
select_query.addTableFunction(table_function_ptr);
else
select_query.replaceDatabaseAndTable(database, table);
/// Restore long column names (cause our short names are ambiguous).
/// TODO: aliased table functions & CREATE TABLE AS table function cases
if (!table_function_ptr)
{
RestoreQualifiedNamesVisitor::Data data;
data.distributed_table = DatabaseAndTableWithAlias(*getTableExpression(query->as<ASTSelectQuery &>(), 0));
data.remote_table.database = database;
data.remote_table.table = table;
data.rename = true;
RestoreQualifiedNamesVisitor(data).visit(modified_query_ast);
}
return modified_query_ast;
}
/// The columns list in the original INSERT query is incorrect because inserted blocks are transformed
/// to the form of the sample block of the Distributed table. So we rewrite it and add all columns from
/// the sample block instead.
ASTPtr createInsertToRemoteTableQuery(const std::string & database, const std::string & table, const Block & sample_block_non_materialized)
{
auto query = std::make_shared<ASTInsertQuery>();
2020-03-02 20:23:58 +00:00
query->table_id = StorageID(database, table);
auto columns = std::make_shared<ASTExpressionList>();
query->columns = columns;
query->children.push_back(columns);
for (const auto & col : sample_block_non_materialized)
columns->children.push_back(std::make_shared<ASTIdentifier>(col.name));
return query;
}
/// Calculate maximum number in file names in directory and all subdirectories.
/// To ensure global order of data blocks yet to be sent across server restarts.
UInt64 getMaximumFileNumber(const std::string & dir_path)
{
UInt64 res = 0;
std::filesystem::recursive_directory_iterator begin(dir_path);
std::filesystem::recursive_directory_iterator end;
for (auto it = begin; it != end; ++it)
{
const auto & file_path = it->path();
if (!std::filesystem::is_regular_file(*it) || !endsWith(file_path.filename().string(), ".bin"))
continue;
UInt64 num = 0;
try
{
num = parse<UInt64>(file_path.filename().stem().string());
}
catch (Exception & e)
{
e.addMessage("Unexpected file name " + file_path.filename().string() + " found at " + file_path.parent_path().string() + ", should have numeric base name.");
throw;
}
if (num > res)
res = num;
}
return res;
}
std::string makeFormattedListOfShards(const ClusterPtr & cluster)
{
2020-11-09 19:07:38 +00:00
WriteBufferFromOwnString buf;
bool head = true;
2020-11-09 19:07:38 +00:00
buf << "[";
for (const auto & shard_info : cluster->getShardsInfo())
{
2020-11-09 19:07:38 +00:00
(head ? buf : buf << ", ") << shard_info.shard_num;
head = false;
}
2020-11-09 19:07:38 +00:00
buf << "]";
2020-11-09 19:07:38 +00:00
return buf.str();
}
ExpressionActionsPtr buildShardingKeyExpression(const ASTPtr & sharding_key, const Context & context, const NamesAndTypesList & columns, bool project)
{
ASTPtr query = sharding_key;
auto syntax_result = TreeRewriter(context).analyze(query, columns);
return ExpressionAnalyzer(query, syntax_result, context).getActions(project);
}
bool isExpressionActionsDeterministics(const ExpressionActionsPtr & actions)
{
for (const auto & action : actions->getActions())
{
2020-11-10 14:54:59 +00:00
if (action.node->type != ActionsDAG::ActionType::FUNCTION)
continue;
2020-11-03 11:28:28 +00:00
if (!action.node->function_base->isDeterministic())
return false;
}
return true;
}
class ReplacingConstantExpressionsMatcher
{
public:
using Data = Block;
static bool needChildVisit(ASTPtr &, const ASTPtr &)
{
return true;
}
static void visit(ASTPtr & node, Block & block_with_constants)
{
if (!node->as<ASTFunction>())
return;
std::string name = node->getColumnName();
if (block_with_constants.has(name))
{
auto result = block_with_constants.getByName(name);
if (!isColumnConst(*result.column))
return;
2020-03-23 17:28:38 +00:00
node = std::make_shared<ASTLiteral>(assert_cast<const ColumnConst &>(*result.column).getField());
}
}
};
2020-03-23 17:28:38 +00:00
2020-06-18 09:08:24 +00:00
void replaceConstantExpressions(
ASTPtr & node,
const Context & context,
const NamesAndTypesList & columns,
ConstStoragePtr storage,
const StorageMetadataPtr & metadata_snapshot)
{
auto syntax_result = TreeRewriter(context).analyze(node, columns, storage, metadata_snapshot);
Block block_with_constants = KeyCondition::getBlockWithConstants(node, syntax_result, context);
InDepthNodeVisitor<ReplacingConstantExpressionsMatcher, true> visitor(block_with_constants);
visitor.visit(node);
}
/// Returns one of the following:
/// - QueryProcessingStage::Complete
/// - QueryProcessingStage::WithMergeableStateAfterAggregation
/// - none (in this case regular WithMergeableState should be used)
std::optional<QueryProcessingStage::Enum> getOptimizedQueryProcessingStage(const ASTPtr & query_ptr, bool extremes, const Block & sharding_key_block)
{
const auto & select = query_ptr->as<ASTSelectQuery &>();
auto sharding_block_has = [&](const auto & exprs, size_t limit = SIZE_MAX) -> bool
{
size_t i = 0;
for (auto & expr : exprs)
{
2020-09-03 01:39:36 +00:00
++i;
if (i > limit)
break;
auto id = expr->template as<ASTIdentifier>();
if (!id)
return false;
/// TODO: if GROUP BY contains multiIf()/if() it should contain only columns from sharding_key
if (!sharding_key_block.has(id->name()))
return false;
}
return true;
};
// GROUP BY qualifiers
// - TODO: WITH TOTALS can be implemented
// - TODO: WITH ROLLUP can be implemented (I guess)
if (select.group_by_with_totals || select.group_by_with_rollup || select.group_by_with_cube)
return {};
// TODO: extremes support can be implemented
if (extremes)
return {};
// DISTINCT
if (select.distinct)
{
if (!sharding_block_has(select.select()->children))
return {};
}
// GROUP BY
const ASTPtr group_by = select.groupBy();
if (!group_by)
{
if (!select.distinct)
return {};
}
else
{
if (!sharding_block_has(group_by->children, 1))
return {};
}
// ORDER BY
const ASTPtr order_by = select.orderBy();
if (order_by)
return QueryProcessingStage::WithMergeableStateAfterAggregation;
// LIMIT BY
// LIMIT
// OFFSET
if (select.limitBy() || select.limitLength() || select.limitOffset())
return QueryProcessingStage::WithMergeableStateAfterAggregation;
// Only simple SELECT FROM GROUP BY sharding_key can use Complete state.
return QueryProcessingStage::Complete;
}
size_t getClusterQueriedNodes(const Settings & settings, const ClusterPtr & cluster)
{
size_t num_local_shards = cluster->getLocalShardCount();
size_t num_remote_shards = cluster->getRemoteShardCount();
return (num_remote_shards * settings.max_parallel_replicas) + num_local_shards;
}
2020-03-26 18:25:26 +00:00
}
/// For destruction of std::unique_ptr of type that is incomplete in class definition.
StorageDistributed::~StorageDistributed() = default;
NamesAndTypesList StorageDistributed::getVirtuals() const
2020-04-27 13:55:30 +00:00
{
/// NOTE This is weird. Most of these virtual columns are part of MergeTree
/// tables info. But Distributed is general-purpose engine.
return NamesAndTypesList{
2020-04-27 13:55:30 +00:00
NameAndTypePair("_table", std::make_shared<DataTypeString>()),
NameAndTypePair("_part", std::make_shared<DataTypeString>()),
NameAndTypePair("_part_index", std::make_shared<DataTypeUInt64>()),
NameAndTypePair("_part_uuid", std::make_shared<DataTypeUUID>()),
2020-04-27 13:55:30 +00:00
NameAndTypePair("_partition_id", std::make_shared<DataTypeString>()),
NameAndTypePair("_sample_factor", std::make_shared<DataTypeFloat64>()),
NameAndTypePair("_shard_num", std::make_shared<DataTypeUInt32>()),
};
}
StorageDistributed::StorageDistributed(
2019-12-04 16:06:55 +00:00
const StorageID & id_,
const ColumnsDescription & columns_,
2019-08-24 21:20:20 +00:00
const ConstraintsDescription & constraints_,
const String & remote_database_,
const String & remote_table_,
const String & cluster_name_,
const Context & context_,
const ASTPtr & sharding_key_,
2020-07-23 14:10:48 +00:00
const String & storage_policy_name_,
2019-10-25 19:07:47 +00:00
const String & relative_data_path_,
const DistributedSettings & distributed_settings_,
bool attach_,
ClusterPtr owned_cluster_)
2020-04-27 13:55:30 +00:00
: IStorage(id_)
2019-12-04 16:06:55 +00:00
, remote_database(remote_database_)
, remote_table(remote_table_)
2020-11-06 14:07:56 +00:00
, global_context(context_.getGlobalContext())
2020-05-30 21:57:37 +00:00
, log(&Poco::Logger::get("StorageDistributed (" + id_.table_name + ")"))
, owned_cluster(std::move(owned_cluster_))
2020-11-06 14:07:56 +00:00
, cluster_name(global_context.getMacros()->expand(cluster_name_))
2019-12-04 16:06:55 +00:00
, has_sharding_key(sharding_key_)
, relative_data_path(relative_data_path_)
, distributed_settings(distributed_settings_)
, rng(randomSeed())
{
2020-06-19 15:39:41 +00:00
StorageInMemoryMetadata storage_metadata;
storage_metadata.setColumns(columns_);
storage_metadata.setConstraints(constraints_);
setInMemoryMetadata(storage_metadata);
2019-08-24 21:20:20 +00:00
2019-08-26 13:46:07 +00:00
if (sharding_key_)
{
2020-11-06 14:07:56 +00:00
sharding_key_expr = buildShardingKeyExpression(sharding_key_, global_context, storage_metadata.getColumns().getAllPhysical(), false);
2019-08-26 13:46:07 +00:00
sharding_key_column_name = sharding_key_->getColumnName();
sharding_key_is_deterministic = isExpressionActionsDeterministics(sharding_key_expr);
2019-08-26 13:46:07 +00:00
}
if (!relative_data_path.empty())
2020-07-23 14:10:48 +00:00
{
2020-11-06 14:07:56 +00:00
storage_policy = global_context.getStoragePolicy(storage_policy_name_);
data_volume = storage_policy->getVolume(0);
if (storage_policy->getVolumes().size() > 1)
LOG_WARNING(log, "Storage policy for Distributed table has multiple volumes. "
"Only {} volume will be used to store data. Other will be ignored.", data_volume->getName());
2020-07-23 14:10:48 +00:00
}
/// Sanity check. Skip check if the table is already created to allow the server to start.
2019-08-03 11:02:40 +00:00
if (!attach_ && !cluster_name.empty())
{
2020-11-06 14:07:56 +00:00
size_t num_local_shards = global_context.getCluster(cluster_name)->getLocalShardCount();
2019-12-04 16:06:55 +00:00
if (num_local_shards && remote_database == id_.database_name && remote_table == id_.table_name)
throw Exception("Distributed table " + id_.table_name + " looks at itself", ErrorCodes::INFINITE_LOOP);
}
}
StorageDistributed::StorageDistributed(
2019-12-04 16:06:55 +00:00
const StorageID & id_,
const ColumnsDescription & columns_,
2019-08-24 21:20:20 +00:00
const ConstraintsDescription & constraints_,
ASTPtr remote_table_function_ptr_,
const String & cluster_name_,
const Context & context_,
const ASTPtr & sharding_key_,
2020-07-23 14:10:48 +00:00
const String & storage_policy_name_,
2019-10-25 19:07:47 +00:00
const String & relative_data_path_,
const DistributedSettings & distributed_settings_,
bool attach,
ClusterPtr owned_cluster_)
: StorageDistributed(id_, columns_, constraints_, String{}, String{}, cluster_name_, context_, sharding_key_, storage_policy_name_, relative_data_path_, distributed_settings_, attach, std::move(owned_cluster_))
{
2019-12-04 16:06:55 +00:00
remote_table_function_ptr = std::move(remote_table_function_ptr_);
}
2020-11-07 21:30:40 +00:00
QueryProcessingStage::Enum StorageDistributed::getQueryProcessingStage(
const Context & context, QueryProcessingStage::Enum to_stage, SelectQueryInfo & query_info) const
{
const auto & settings = context.getSettingsRef();
auto metadata_snapshot = getInMemoryMetadataPtr();
ClusterPtr cluster = getCluster();
query_info.cluster = cluster;
/// Always calculate optimized cluster here, to avoid conditions during read()
/// (Anyway it will be calculated in the read())
if (settings.optimize_skip_unused_shards)
{
ClusterPtr optimized_cluster = getOptimizedCluster(context, metadata_snapshot, query_info.query);
if (optimized_cluster)
{
LOG_DEBUG(log, "Skipping irrelevant shards - the query will be sent to the following shards of the cluster (shard numbers): {}", makeFormattedListOfShards(optimized_cluster));
cluster = optimized_cluster;
query_info.cluster = cluster;
}
else
{
LOG_DEBUG(log, "Unable to figure out irrelevant shards from WHERE/PREWHERE clauses - the query will be sent to all shards of the cluster{}", has_sharding_key ? "" : " (no sharding key)");
}
}
if (settings.distributed_group_by_no_merge)
{
if (settings.distributed_group_by_no_merge == DISTRIBUTED_GROUP_BY_NO_MERGE_AFTER_AGGREGATION)
return QueryProcessingStage::WithMergeableStateAfterAggregation;
else
return QueryProcessingStage::Complete;
}
/// Nested distributed query cannot return Complete stage,
/// since the parent query need to aggregate the results after.
if (to_stage == QueryProcessingStage::WithMergeableState)
return QueryProcessingStage::WithMergeableState;
/// If there is only one node, the query can be fully processed by the
/// shard, initiator will work as a proxy only.
if (getClusterQueriedNodes(settings, cluster) == 1)
return QueryProcessingStage::Complete;
if (settings.optimize_skip_unused_shards &&
settings.optimize_distributed_group_by_sharding_key &&
has_sharding_key &&
(settings.allow_nondeterministic_optimize_skip_unused_shards || sharding_key_is_deterministic))
{
Block sharding_key_block = sharding_key_expr->getSampleBlock();
auto stage = getOptimizedQueryProcessingStage(query_info.query, settings.extremes, sharding_key_block);
if (stage)
{
LOG_DEBUG(log, "Force processing stage to {}", QueryProcessingStage::toString(*stage));
return *stage;
}
}
return QueryProcessingStage::WithMergeableState;
2020-03-18 00:57:00 +00:00
}
2020-08-03 13:54:14 +00:00
Pipe StorageDistributed::read(
2020-09-25 13:19:26 +00:00
const Names & column_names,
const StorageMetadataPtr & metadata_snapshot,
SelectQueryInfo & query_info,
2020-09-25 13:19:26 +00:00
const Context & context,
QueryProcessingStage::Enum processed_stage,
const size_t max_block_size,
const unsigned num_streams)
{
QueryPlan plan;
read(plan, column_names, metadata_snapshot, query_info, context, processed_stage, max_block_size, num_streams);
return plan.convertToPipe(QueryPlanOptimizationSettings(context.getSettingsRef()));
2020-09-25 13:19:26 +00:00
}
void StorageDistributed::read(
QueryPlan & query_plan,
const Names & column_names,
2020-06-17 14:37:21 +00:00
const StorageMetadataPtr & metadata_snapshot,
SelectQueryInfo & query_info,
const Context & context,
QueryProcessingStage::Enum processed_stage,
const size_t /*max_block_size*/,
const unsigned /*num_streams*/)
{
const auto & modified_query_ast = rewriteSelectQuery(
query_info.query, remote_database, remote_table, remote_table_function_ptr);
2019-08-09 13:37:42 +00:00
Block header =
InterpreterSelectQuery(query_info.query, context, SelectQueryOptions(processed_stage)).getSampleBlock();
2019-10-19 20:36:35 +00:00
const Scalars & scalars = context.hasQueryContext() ? context.getQueryContext().getScalars() : Scalars{};
bool has_virtual_shard_num_column = std::find(column_names.begin(), column_names.end(), "_shard_num") != column_names.end();
2020-06-17 14:37:21 +00:00
if (has_virtual_shard_num_column && !isVirtualColumn("_shard_num", metadata_snapshot))
has_virtual_shard_num_column = false;
2018-09-10 03:59:48 +00:00
ClusterProxy::SelectStreamFactory select_stream_factory = remote_table_function_ptr
? ClusterProxy::SelectStreamFactory(
header, processed_stage, remote_table_function_ptr, scalars, has_virtual_shard_num_column, context.getExternalTables())
2018-07-25 12:31:47 +00:00
: ClusterProxy::SelectStreamFactory(
2020-03-04 20:29:52 +00:00
header, processed_stage, StorageID{remote_database, remote_table}, scalars, has_virtual_shard_num_column, context.getExternalTables());
2015-11-06 17:44:01 +00:00
ClusterProxy::executeQuery(query_plan, select_stream_factory, log,
modified_query_ast, context, query_info);
2012-05-21 20:38:34 +00:00
}
BlockOutputStreamPtr StorageDistributed::write(const ASTPtr &, const StorageMetadataPtr & metadata_snapshot, const Context & context)
{
auto cluster = getCluster();
const auto & settings = context.getSettingsRef();
/// Ban an attempt to make async insert into the table belonging to DatabaseMemory
if (!storage_policy && !owned_cluster && !settings.insert_distributed_sync && !settings.insert_shard_id)
{
throw Exception("Storage " + getName() + " must have own data directory to enable asynchronous inserts",
ErrorCodes::BAD_ARGUMENTS);
}
auto shard_num = cluster->getLocalShardCount() + cluster->getRemoteShardCount();
/// If sharding key is not specified, then you can only write to a shard containing only one shard
if (!settings.insert_shard_id && !settings.insert_distributed_one_random_shard && !has_sharding_key && shard_num >= 2)
{
throw Exception(
"Method write is not supported by storage " + getName() + " with more than one shard and no sharding key provided",
ErrorCodes::STORAGE_REQUIRES_PARAMETER);
}
if (settings.insert_shard_id && settings.insert_shard_id > shard_num)
{
throw Exception("Shard id should be range from 1 to shard number", ErrorCodes::INVALID_SHARD_ID);
}
2014-08-21 12:07:29 +00:00
/// Force sync insertion if it is remote() table function
bool insert_sync = settings.insert_distributed_sync || settings.insert_shard_id || owned_cluster;
auto timeout = settings.insert_distributed_timeout;
/// DistributedBlockOutputStream will not own cluster, but will own ConnectionPools of the cluster
return std::make_shared<DistributedBlockOutputStream>(
context, *this, metadata_snapshot,
createInsertToRemoteTableQuery(
remote_database, remote_table, metadata_snapshot->getSampleBlockNonMaterialized()),
cluster, insert_sync, timeout);
}
void StorageDistributed::checkAlterIsPossible(const AlterCommands & commands, const Context & context) const
2013-09-23 12:01:19 +00:00
{
2021-02-28 07:42:08 +00:00
auto name_deps = getDependentViewsByColumn(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
&& command.type != AlterCommand::Type::RENAME_COLUMN)
2016-05-13 21:08:19 +00:00
2019-12-26 18:17:05 +00:00
throw Exception("Alter of type '" + alterTypeToString(command.type) + "' is not supported by storage " + getName(),
ErrorCodes::NOT_IMPLEMENTED);
if (command.type == AlterCommand::DROP_COLUMN)
{
2021-02-28 07:42:08 +00:00
const auto & deps_mv = name_deps[command.column_name];
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
}
}
2020-06-18 16:10:47 +00:00
void StorageDistributed::alter(const AlterCommands & params, const Context & context, TableLockHolder &)
2013-09-23 12:01:19 +00:00
{
auto table_id = getStorageID();
checkAlterIsPossible(params, context);
2020-06-09 17:28:29 +00:00
StorageInMemoryMetadata new_metadata = getInMemoryMetadata();
params.apply(new_metadata, context);
DatabaseCatalog::instance().getDatabase(table_id.database_name)->alterTable(context, table_id, new_metadata);
setInMemoryMetadata(new_metadata);
2013-09-23 12:01:19 +00:00
}
void StorageDistributed::startup()
{
2020-02-21 13:44:44 +00:00
if (remote_database.empty() && !remote_table_function_ptr)
2020-05-23 22:24:01 +00:00
LOG_WARNING(log, "Name of remote database is empty. Default database will be used implicitly.");
2020-02-21 13:44:44 +00:00
2020-07-23 14:10:48 +00:00
if (!storage_policy)
return;
for (const DiskPtr & disk : data_volume->getDisks())
createDirectoryMonitors(disk);
for (const String & path : getDataPaths())
{
UInt64 inc = getMaximumFileNumber(path);
if (inc > file_names_increment.value)
file_names_increment.value.store(inc);
}
2020-05-23 22:24:01 +00:00
LOG_DEBUG(log, "Auto-increment is {}", file_names_increment.value);
}
void StorageDistributed::shutdown()
{
monitors_blocker.cancelForever();
std::lock_guard lock(cluster_nodes_mutex);
Fix DROP TABLE for Distributed (racy with INSERT) <details> ``` drop() on T1275: 0 DB::StorageDistributed::drop (this=0x7f9ed34f0000) at ../contrib/libcxx/include/__hash_table:966 1 0x000000000d557242 in DB::DatabaseOnDisk::dropTable (this=0x7f9fc22706d8, context=..., table_name=...) at ../contrib/libcxx/include/new:340 2 0x000000000d6fcf7c in DB::InterpreterDropQuery::executeToTable (this=this@entry=0x7f9e42560dc0, query=...) at ../contrib/libcxx/include/memory:3826 3 0x000000000d6ff5ee in DB::InterpreterDropQuery::execute (this=0x7f9e42560dc0) at ../src/Interpreters/InterpreterDropQuery.cpp:50 4 0x000000000daa40c0 in DB::executeQueryImpl (begin=<optimized out>, end=<optimized out>, context=..., internal=<optimized out>, stage=DB::QueryProcessingStage::Complete, has_query_tail=false, istr=0x0) at ../src/Interpreters/executeQuery.cpp:420 5 0x000000000daa59df in DB::executeQuery (query=..., context=..., internal=internal@entry=false, stage=<optimized out>, may_have_embedded_data=<optimized out>) at ../contrib/libcxx/include/string:1487 6 0x000000000e1369e6 in DB::TCPHandler::runImpl (this=this@entry=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:254 7 0x000000000e1379c9 in DB::TCPHandler::run (this=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:1326 8 0x000000001086fac7 in Poco::Net::TCPServerConnection::start (this=this@entry=0x7f9ddf3a9000) at ../contrib/poco/Net/src/TCPServerConnection.cpp:43 9 0x000000001086ff2b in Poco::Net::TCPServerDispatcher::run (this=0x7f9e4eba5c00) at ../contrib/poco/Net/src/TCPServerDispatcher.cpp:114 10 0x00000000109dbe8e in Poco::PooledThread::run (this=0x7f9e4a2d2f80) at ../contrib/poco/Foundation/src/ThreadPool.cpp:199 11 0x00000000109d78f9 in Poco::ThreadImpl::runnableEntry (pThread=<optimized out>) at ../contrib/poco/Foundation/include/Poco/SharedPtr.h:401 12 0x00007f9fc3cccea7 in start_thread (arg=<optimized out>) at pthread_create.c:477 13 0x00007f9fc3bebeaf in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:95 StorageDistributedDirectoryMonitor on T166: 0 DB::StorageDistributedDirectoryMonitor::StorageDistributedDirectoryMonitor (this=0x7f9ea7ab1400, storage_=..., path_=..., pool_=..., monitor_blocker_=..., bg_pool_=...) at ../src/Storages/Distributed/DirectoryMonitor.cpp:81 1 0x000000000dbf684e in std::__1::make_unique<> () at ../contrib/libcxx/include/memory:3474 2 DB::StorageDistributed::requireDirectoryMonitor (this=0x7f9ed34f0000, disk=..., name=...) at ../src/Storages/StorageDistributed.cpp:682 3 0x000000000de3d5fa in DB::DistributedBlockOutputStream::writeToShard (this=this@entry=0x7f9ed39c7418, block=..., dir_names=...) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:634 4 0x000000000de3e214 in DB::DistributedBlockOutputStream::writeAsyncImpl (this=this@entry=0x7f9ed39c7418, block=..., shard_id=shard_id@entry=79) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:539 5 0x000000000de3e47b in DB::DistributedBlockOutputStream::writeSplitAsync (this=this@entry=0x7f9ed39c7418, block=...) at ../contrib/libcxx/include/vector:1546 6 0x000000000de3eab0 in DB::DistributedBlockOutputStream::writeAsync (block=..., this=0x7f9ed39c7418) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:141 7 DB::DistributedBlockOutputStream::write (this=0x7f9ed39c7418, block=...) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:135 8 0x000000000d73b376 in DB::PushingToViewsBlockOutputStream::write (this=this@entry=0x7f9ea7a8cf58, block=...) at ../src/DataStreams/PushingToViewsBlockOutputStream.cpp:157 9 0x000000000d7853eb in DB::AddingDefaultBlockOutputStream::write (this=0x7f9ed383d118, block=...) at ../contrib/libcxx/include/memory:3826 10 0x000000000d740790 in DB::SquashingBlockOutputStream::write (this=0x7f9ed383de18, block=...) at ../contrib/libcxx/include/memory:3826 11 0x000000000d68c308 in DB::CountingBlockOutputStream::write (this=0x7f9ea7ac6d60, block=...) at ../contrib/libcxx/include/memory:3826 12 0x000000000ddab449 in DB::StorageBuffer::writeBlockToDestination (this=this@entry=0x7f9fbd56a000, block=..., table=...) at ../src/Storages/StorageBuffer.cpp:747 13 0x000000000ddabfa6 in DB::StorageBuffer::flushBuffer (this=this@entry=0x7f9fbd56a000, buffer=..., check_thresholds=check_thresholds@entry=true, locked=locked@entry=false, reset_block_structure=reset_block_structure@entry=false) at ../src/Storages/StorageBuffer.cpp:661 14 0x000000000ddac415 in DB::StorageBuffer::flushAllBuffers (reset_blocks_structure=false, check_thresholds=true, this=0x7f9fbd56a000) at ../src/Storages/StorageBuffer.cpp:605 shutdown() on T1275: 0 DB::StorageDistributed::shutdown (this=0x7f9ed34f0000) at ../contrib/libcxx/include/atomic:1612 1 0x000000000d6fd938 in DB::InterpreterDropQuery::executeToTable (this=this@entry=0x7f98530c79a0, query=...) at ../src/Storages/TableLockHolder.h:12 2 0x000000000d6ff5ee in DB::InterpreterDropQuery::execute (this=0x7f98530c79a0) at ../src/Interpreters/InterpreterDropQuery.cpp:50 3 0x000000000daa40c0 in DB::executeQueryImpl (begin=<optimized out>, end=<optimized out>, context=..., internal=<optimized out>, stage=DB::QueryProcessingStage::Complete, has_query_tail=false, istr=0x0) at ../src/Interpreters/executeQuery.cpp:420 4 0x000000000daa59df in DB::executeQuery (query=..., context=..., internal=internal@entry=false, stage=<optimized out>, may_have_embedded_data=<optimized out>) at ../contrib/libcxx/include/string:1487 5 0x000000000e1369e6 in DB::TCPHandler::runImpl (this=this@entry=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:254 6 0x000000000e1379c9 in DB::TCPHandler::run (this=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:1326 7 0x000000001086fac7 in Poco::Net::TCPServerConnection::start (this=this@entry=0x7f9ddf3a9000) at ../contrib/poco/Net/src/TCPServerConnection.cpp:43 8 0x000000001086ff2b in Poco::Net::TCPServerDispatcher::run (this=0x7f9e4eba5c00) at ../contrib/poco/Net/src/TCPServerDispatcher.cpp:114 9 0x00000000109dbe8e in Poco::PooledThread::run (this=0x7f9e4a2d2f80) at ../contrib/poco/Foundation/src/ThreadPool.cpp:199 10 0x00000000109d78f9 in Poco::ThreadImpl::runnableEntry (pThread=<optimized out>) at ../contrib/poco/Foundation/include/Poco/SharedPtr.h:401 11 0x00007f9fc3cccea7 in start_thread (arg=<optimized out>) at pthread_create.c:477 12 0x00007f9fc3bebeaf in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:95 ``` </details>
2020-10-26 20:01:06 +00:00
LOG_DEBUG(log, "Joining background threads for async INSERT");
cluster_nodes_data.clear();
Fix DROP TABLE for Distributed (racy with INSERT) <details> ``` drop() on T1275: 0 DB::StorageDistributed::drop (this=0x7f9ed34f0000) at ../contrib/libcxx/include/__hash_table:966 1 0x000000000d557242 in DB::DatabaseOnDisk::dropTable (this=0x7f9fc22706d8, context=..., table_name=...) at ../contrib/libcxx/include/new:340 2 0x000000000d6fcf7c in DB::InterpreterDropQuery::executeToTable (this=this@entry=0x7f9e42560dc0, query=...) at ../contrib/libcxx/include/memory:3826 3 0x000000000d6ff5ee in DB::InterpreterDropQuery::execute (this=0x7f9e42560dc0) at ../src/Interpreters/InterpreterDropQuery.cpp:50 4 0x000000000daa40c0 in DB::executeQueryImpl (begin=<optimized out>, end=<optimized out>, context=..., internal=<optimized out>, stage=DB::QueryProcessingStage::Complete, has_query_tail=false, istr=0x0) at ../src/Interpreters/executeQuery.cpp:420 5 0x000000000daa59df in DB::executeQuery (query=..., context=..., internal=internal@entry=false, stage=<optimized out>, may_have_embedded_data=<optimized out>) at ../contrib/libcxx/include/string:1487 6 0x000000000e1369e6 in DB::TCPHandler::runImpl (this=this@entry=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:254 7 0x000000000e1379c9 in DB::TCPHandler::run (this=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:1326 8 0x000000001086fac7 in Poco::Net::TCPServerConnection::start (this=this@entry=0x7f9ddf3a9000) at ../contrib/poco/Net/src/TCPServerConnection.cpp:43 9 0x000000001086ff2b in Poco::Net::TCPServerDispatcher::run (this=0x7f9e4eba5c00) at ../contrib/poco/Net/src/TCPServerDispatcher.cpp:114 10 0x00000000109dbe8e in Poco::PooledThread::run (this=0x7f9e4a2d2f80) at ../contrib/poco/Foundation/src/ThreadPool.cpp:199 11 0x00000000109d78f9 in Poco::ThreadImpl::runnableEntry (pThread=<optimized out>) at ../contrib/poco/Foundation/include/Poco/SharedPtr.h:401 12 0x00007f9fc3cccea7 in start_thread (arg=<optimized out>) at pthread_create.c:477 13 0x00007f9fc3bebeaf in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:95 StorageDistributedDirectoryMonitor on T166: 0 DB::StorageDistributedDirectoryMonitor::StorageDistributedDirectoryMonitor (this=0x7f9ea7ab1400, storage_=..., path_=..., pool_=..., monitor_blocker_=..., bg_pool_=...) at ../src/Storages/Distributed/DirectoryMonitor.cpp:81 1 0x000000000dbf684e in std::__1::make_unique<> () at ../contrib/libcxx/include/memory:3474 2 DB::StorageDistributed::requireDirectoryMonitor (this=0x7f9ed34f0000, disk=..., name=...) at ../src/Storages/StorageDistributed.cpp:682 3 0x000000000de3d5fa in DB::DistributedBlockOutputStream::writeToShard (this=this@entry=0x7f9ed39c7418, block=..., dir_names=...) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:634 4 0x000000000de3e214 in DB::DistributedBlockOutputStream::writeAsyncImpl (this=this@entry=0x7f9ed39c7418, block=..., shard_id=shard_id@entry=79) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:539 5 0x000000000de3e47b in DB::DistributedBlockOutputStream::writeSplitAsync (this=this@entry=0x7f9ed39c7418, block=...) at ../contrib/libcxx/include/vector:1546 6 0x000000000de3eab0 in DB::DistributedBlockOutputStream::writeAsync (block=..., this=0x7f9ed39c7418) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:141 7 DB::DistributedBlockOutputStream::write (this=0x7f9ed39c7418, block=...) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:135 8 0x000000000d73b376 in DB::PushingToViewsBlockOutputStream::write (this=this@entry=0x7f9ea7a8cf58, block=...) at ../src/DataStreams/PushingToViewsBlockOutputStream.cpp:157 9 0x000000000d7853eb in DB::AddingDefaultBlockOutputStream::write (this=0x7f9ed383d118, block=...) at ../contrib/libcxx/include/memory:3826 10 0x000000000d740790 in DB::SquashingBlockOutputStream::write (this=0x7f9ed383de18, block=...) at ../contrib/libcxx/include/memory:3826 11 0x000000000d68c308 in DB::CountingBlockOutputStream::write (this=0x7f9ea7ac6d60, block=...) at ../contrib/libcxx/include/memory:3826 12 0x000000000ddab449 in DB::StorageBuffer::writeBlockToDestination (this=this@entry=0x7f9fbd56a000, block=..., table=...) at ../src/Storages/StorageBuffer.cpp:747 13 0x000000000ddabfa6 in DB::StorageBuffer::flushBuffer (this=this@entry=0x7f9fbd56a000, buffer=..., check_thresholds=check_thresholds@entry=true, locked=locked@entry=false, reset_block_structure=reset_block_structure@entry=false) at ../src/Storages/StorageBuffer.cpp:661 14 0x000000000ddac415 in DB::StorageBuffer::flushAllBuffers (reset_blocks_structure=false, check_thresholds=true, this=0x7f9fbd56a000) at ../src/Storages/StorageBuffer.cpp:605 shutdown() on T1275: 0 DB::StorageDistributed::shutdown (this=0x7f9ed34f0000) at ../contrib/libcxx/include/atomic:1612 1 0x000000000d6fd938 in DB::InterpreterDropQuery::executeToTable (this=this@entry=0x7f98530c79a0, query=...) at ../src/Storages/TableLockHolder.h:12 2 0x000000000d6ff5ee in DB::InterpreterDropQuery::execute (this=0x7f98530c79a0) at ../src/Interpreters/InterpreterDropQuery.cpp:50 3 0x000000000daa40c0 in DB::executeQueryImpl (begin=<optimized out>, end=<optimized out>, context=..., internal=<optimized out>, stage=DB::QueryProcessingStage::Complete, has_query_tail=false, istr=0x0) at ../src/Interpreters/executeQuery.cpp:420 4 0x000000000daa59df in DB::executeQuery (query=..., context=..., internal=internal@entry=false, stage=<optimized out>, may_have_embedded_data=<optimized out>) at ../contrib/libcxx/include/string:1487 5 0x000000000e1369e6 in DB::TCPHandler::runImpl (this=this@entry=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:254 6 0x000000000e1379c9 in DB::TCPHandler::run (this=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:1326 7 0x000000001086fac7 in Poco::Net::TCPServerConnection::start (this=this@entry=0x7f9ddf3a9000) at ../contrib/poco/Net/src/TCPServerConnection.cpp:43 8 0x000000001086ff2b in Poco::Net::TCPServerDispatcher::run (this=0x7f9e4eba5c00) at ../contrib/poco/Net/src/TCPServerDispatcher.cpp:114 9 0x00000000109dbe8e in Poco::PooledThread::run (this=0x7f9e4a2d2f80) at ../contrib/poco/Foundation/src/ThreadPool.cpp:199 10 0x00000000109d78f9 in Poco::ThreadImpl::runnableEntry (pThread=<optimized out>) at ../contrib/poco/Foundation/include/Poco/SharedPtr.h:401 11 0x00007f9fc3cccea7 in start_thread (arg=<optimized out>) at pthread_create.c:477 12 0x00007f9fc3bebeaf in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:95 ``` </details>
2020-10-26 20:01:06 +00:00
LOG_DEBUG(log, "Background threads for async INSERT joined");
}
void StorageDistributed::drop()
{
Fix DROP TABLE for Distributed (racy with INSERT) <details> ``` drop() on T1275: 0 DB::StorageDistributed::drop (this=0x7f9ed34f0000) at ../contrib/libcxx/include/__hash_table:966 1 0x000000000d557242 in DB::DatabaseOnDisk::dropTable (this=0x7f9fc22706d8, context=..., table_name=...) at ../contrib/libcxx/include/new:340 2 0x000000000d6fcf7c in DB::InterpreterDropQuery::executeToTable (this=this@entry=0x7f9e42560dc0, query=...) at ../contrib/libcxx/include/memory:3826 3 0x000000000d6ff5ee in DB::InterpreterDropQuery::execute (this=0x7f9e42560dc0) at ../src/Interpreters/InterpreterDropQuery.cpp:50 4 0x000000000daa40c0 in DB::executeQueryImpl (begin=<optimized out>, end=<optimized out>, context=..., internal=<optimized out>, stage=DB::QueryProcessingStage::Complete, has_query_tail=false, istr=0x0) at ../src/Interpreters/executeQuery.cpp:420 5 0x000000000daa59df in DB::executeQuery (query=..., context=..., internal=internal@entry=false, stage=<optimized out>, may_have_embedded_data=<optimized out>) at ../contrib/libcxx/include/string:1487 6 0x000000000e1369e6 in DB::TCPHandler::runImpl (this=this@entry=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:254 7 0x000000000e1379c9 in DB::TCPHandler::run (this=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:1326 8 0x000000001086fac7 in Poco::Net::TCPServerConnection::start (this=this@entry=0x7f9ddf3a9000) at ../contrib/poco/Net/src/TCPServerConnection.cpp:43 9 0x000000001086ff2b in Poco::Net::TCPServerDispatcher::run (this=0x7f9e4eba5c00) at ../contrib/poco/Net/src/TCPServerDispatcher.cpp:114 10 0x00000000109dbe8e in Poco::PooledThread::run (this=0x7f9e4a2d2f80) at ../contrib/poco/Foundation/src/ThreadPool.cpp:199 11 0x00000000109d78f9 in Poco::ThreadImpl::runnableEntry (pThread=<optimized out>) at ../contrib/poco/Foundation/include/Poco/SharedPtr.h:401 12 0x00007f9fc3cccea7 in start_thread (arg=<optimized out>) at pthread_create.c:477 13 0x00007f9fc3bebeaf in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:95 StorageDistributedDirectoryMonitor on T166: 0 DB::StorageDistributedDirectoryMonitor::StorageDistributedDirectoryMonitor (this=0x7f9ea7ab1400, storage_=..., path_=..., pool_=..., monitor_blocker_=..., bg_pool_=...) at ../src/Storages/Distributed/DirectoryMonitor.cpp:81 1 0x000000000dbf684e in std::__1::make_unique<> () at ../contrib/libcxx/include/memory:3474 2 DB::StorageDistributed::requireDirectoryMonitor (this=0x7f9ed34f0000, disk=..., name=...) at ../src/Storages/StorageDistributed.cpp:682 3 0x000000000de3d5fa in DB::DistributedBlockOutputStream::writeToShard (this=this@entry=0x7f9ed39c7418, block=..., dir_names=...) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:634 4 0x000000000de3e214 in DB::DistributedBlockOutputStream::writeAsyncImpl (this=this@entry=0x7f9ed39c7418, block=..., shard_id=shard_id@entry=79) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:539 5 0x000000000de3e47b in DB::DistributedBlockOutputStream::writeSplitAsync (this=this@entry=0x7f9ed39c7418, block=...) at ../contrib/libcxx/include/vector:1546 6 0x000000000de3eab0 in DB::DistributedBlockOutputStream::writeAsync (block=..., this=0x7f9ed39c7418) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:141 7 DB::DistributedBlockOutputStream::write (this=0x7f9ed39c7418, block=...) at ../src/Storages/Distributed/DistributedBlockOutputStream.cpp:135 8 0x000000000d73b376 in DB::PushingToViewsBlockOutputStream::write (this=this@entry=0x7f9ea7a8cf58, block=...) at ../src/DataStreams/PushingToViewsBlockOutputStream.cpp:157 9 0x000000000d7853eb in DB::AddingDefaultBlockOutputStream::write (this=0x7f9ed383d118, block=...) at ../contrib/libcxx/include/memory:3826 10 0x000000000d740790 in DB::SquashingBlockOutputStream::write (this=0x7f9ed383de18, block=...) at ../contrib/libcxx/include/memory:3826 11 0x000000000d68c308 in DB::CountingBlockOutputStream::write (this=0x7f9ea7ac6d60, block=...) at ../contrib/libcxx/include/memory:3826 12 0x000000000ddab449 in DB::StorageBuffer::writeBlockToDestination (this=this@entry=0x7f9fbd56a000, block=..., table=...) at ../src/Storages/StorageBuffer.cpp:747 13 0x000000000ddabfa6 in DB::StorageBuffer::flushBuffer (this=this@entry=0x7f9fbd56a000, buffer=..., check_thresholds=check_thresholds@entry=true, locked=locked@entry=false, reset_block_structure=reset_block_structure@entry=false) at ../src/Storages/StorageBuffer.cpp:661 14 0x000000000ddac415 in DB::StorageBuffer::flushAllBuffers (reset_blocks_structure=false, check_thresholds=true, this=0x7f9fbd56a000) at ../src/Storages/StorageBuffer.cpp:605 shutdown() on T1275: 0 DB::StorageDistributed::shutdown (this=0x7f9ed34f0000) at ../contrib/libcxx/include/atomic:1612 1 0x000000000d6fd938 in DB::InterpreterDropQuery::executeToTable (this=this@entry=0x7f98530c79a0, query=...) at ../src/Storages/TableLockHolder.h:12 2 0x000000000d6ff5ee in DB::InterpreterDropQuery::execute (this=0x7f98530c79a0) at ../src/Interpreters/InterpreterDropQuery.cpp:50 3 0x000000000daa40c0 in DB::executeQueryImpl (begin=<optimized out>, end=<optimized out>, context=..., internal=<optimized out>, stage=DB::QueryProcessingStage::Complete, has_query_tail=false, istr=0x0) at ../src/Interpreters/executeQuery.cpp:420 4 0x000000000daa59df in DB::executeQuery (query=..., context=..., internal=internal@entry=false, stage=<optimized out>, may_have_embedded_data=<optimized out>) at ../contrib/libcxx/include/string:1487 5 0x000000000e1369e6 in DB::TCPHandler::runImpl (this=this@entry=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:254 6 0x000000000e1379c9 in DB::TCPHandler::run (this=0x7f9ddf3a9000) at ../src/Server/TCPHandler.cpp:1326 7 0x000000001086fac7 in Poco::Net::TCPServerConnection::start (this=this@entry=0x7f9ddf3a9000) at ../contrib/poco/Net/src/TCPServerConnection.cpp:43 8 0x000000001086ff2b in Poco::Net::TCPServerDispatcher::run (this=0x7f9e4eba5c00) at ../contrib/poco/Net/src/TCPServerDispatcher.cpp:114 9 0x00000000109dbe8e in Poco::PooledThread::run (this=0x7f9e4a2d2f80) at ../contrib/poco/Foundation/src/ThreadPool.cpp:199 10 0x00000000109d78f9 in Poco::ThreadImpl::runnableEntry (pThread=<optimized out>) at ../contrib/poco/Foundation/include/Poco/SharedPtr.h:401 11 0x00007f9fc3cccea7 in start_thread (arg=<optimized out>) at pthread_create.c:477 12 0x00007f9fc3bebeaf in clone () at ../sysdeps/unix/sysv/linux/x86_64/clone.S:95 ``` </details>
2020-10-26 20:01:06 +00:00
// Some INSERT in-between shutdown() and drop() can call
// requireDirectoryMonitor() again, so call shutdown() to clear them, but
// when the drop() (this function) executed none of INSERT is allowed in
// parallel.
//
// And second time shutdown() should be fast, since none of
// DirectoryMonitor should do anything, because ActionBlocker is canceled
// (in shutdown()).
shutdown();
// Distributed table w/o sharding_key does not allows INSERTs
if (relative_data_path.empty())
return;
LOG_DEBUG(log, "Removing pending blocks for async INSERT from filesystem on DROP TABLE");
auto disks = data_volume->getDisks();
for (const auto & disk : disks)
disk->removeRecursive(relative_data_path);
LOG_DEBUG(log, "Removed");
}
Strings StorageDistributed::getDataPaths() const
{
Strings paths;
if (relative_data_path.empty())
return paths;
for (const DiskPtr & disk : data_volume->getDisks())
paths.push_back(disk->getPath() + relative_data_path);
return paths;
}
2020-06-18 16:10:47 +00:00
void StorageDistributed::truncate(const ASTPtr &, const StorageMetadataPtr &, const Context &, TableExclusiveLockHolder &)
2018-04-21 00:35:20 +00:00
{
2018-06-09 15:48:22 +00:00
std::lock_guard lock(cluster_nodes_mutex);
2018-04-21 00:35:20 +00:00
LOG_DEBUG(log, "Removing pending blocks for async INSERT from filesystem on TRUNCATE TABLE");
2018-06-09 15:48:22 +00:00
for (auto it = cluster_nodes_data.begin(); it != cluster_nodes_data.end();)
2018-04-21 00:35:20 +00:00
{
it->second.directory_monitor->shutdownAndDropAllData();
2018-06-09 15:48:22 +00:00
it = cluster_nodes_data.erase(it);
2018-04-21 00:35:20 +00:00
}
LOG_DEBUG(log, "Removed");
2018-04-21 00:35:20 +00:00
}
StoragePolicyPtr StorageDistributed::getStoragePolicy() const
{
2020-07-23 14:10:48 +00:00
return storage_policy;
}
void StorageDistributed::createDirectoryMonitors(const DiskPtr & disk)
{
const std::string path(disk->getPath() + relative_data_path);
Poco::File{path}.createDirectories();
std::filesystem::directory_iterator begin(path);
std::filesystem::directory_iterator end;
for (auto it = begin; it != end; ++it)
{
const auto & dir_path = it->path();
if (std::filesystem::is_directory(dir_path))
{
const auto & tmp_path = dir_path / "tmp";
/// "tmp" created by DistributedBlockOutputStream
if (std::filesystem::is_directory(tmp_path) && std::filesystem::is_empty(tmp_path))
std::filesystem::remove(tmp_path);
if (std::filesystem::is_empty(dir_path))
{
2020-11-22 17:13:40 +00:00
LOG_DEBUG(log, "Removing {} (used for async INSERT into Distributed)", dir_path.string());
/// Will be created by DistributedBlockOutputStream on demand.
std::filesystem::remove(dir_path);
}
else
{
requireDirectoryMonitor(disk, dir_path.filename().string());
}
}
}
}
StorageDistributedDirectoryMonitor& StorageDistributed::requireDirectoryMonitor(const DiskPtr & disk, const std::string & name)
{
const std::string & disk_path = disk->getPath();
const std::string key(disk_path + name);
std::lock_guard lock(cluster_nodes_mutex);
auto & node_data = cluster_nodes_data[key];
if (!node_data.directory_monitor)
{
2020-06-03 00:10:39 +00:00
node_data.connection_pool = StorageDistributedDirectoryMonitor::createPool(name, *this);
node_data.directory_monitor = std::make_unique<StorageDistributedDirectoryMonitor>(
*this, disk, relative_data_path + name,
node_data.connection_pool,
monitors_blocker,
global_context.getDistributedSchedulePool());
}
return *node_data.directory_monitor;
}
std::vector<StorageDistributedDirectoryMonitor::Status> StorageDistributed::getDirectoryMonitorsStatuses() const
{
std::vector<StorageDistributedDirectoryMonitor::Status> statuses;
std::lock_guard lock(cluster_nodes_mutex);
2020-06-06 15:57:52 +00:00
statuses.reserve(cluster_nodes_data.size());
2020-06-04 17:23:46 +00:00
for (const auto & node : cluster_nodes_data)
statuses.push_back(node.second.directory_monitor->getStatus());
return statuses;
}
std::optional<UInt64> StorageDistributed::totalBytes(const Settings &) const
{
UInt64 total_bytes = 0;
for (const auto & status : getDirectoryMonitorsStatuses())
total_bytes += status.bytes_count;
return total_bytes;
}
2015-09-18 13:36:10 +00:00
size_t StorageDistributed::getShardCount() const
{
return getCluster()->getShardCount();
}
ClusterPtr StorageDistributed::getCluster() const
{
2020-11-06 14:07:56 +00:00
return owned_cluster ? owned_cluster : global_context.getCluster(cluster_name);
}
ClusterPtr StorageDistributed::getOptimizedCluster(const Context & context, const StorageMetadataPtr & metadata_snapshot, const ASTPtr & query_ptr) const
{
ClusterPtr cluster = getCluster();
const Settings & settings = context.getSettingsRef();
bool sharding_key_is_usable = settings.allow_nondeterministic_optimize_skip_unused_shards || sharding_key_is_deterministic;
if (has_sharding_key && sharding_key_is_usable)
{
ClusterPtr optimized = skipUnusedShards(cluster, query_ptr, metadata_snapshot, context);
if (optimized)
return optimized;
}
UInt64 force = settings.force_optimize_skip_unused_shards;
if (force)
{
2020-11-09 19:07:38 +00:00
WriteBufferFromOwnString exception_message;
if (!has_sharding_key)
exception_message << "No sharding key";
else if (!sharding_key_is_usable)
exception_message << "Sharding key is not deterministic";
else
exception_message << "Sharding key " << sharding_key_column_name << " is not used";
if (force == FORCE_OPTIMIZE_SKIP_UNUSED_SHARDS_ALWAYS)
throw Exception(exception_message.str(), ErrorCodes::UNABLE_TO_SKIP_UNUSED_SHARDS);
if (force == FORCE_OPTIMIZE_SKIP_UNUSED_SHARDS_HAS_SHARDING_KEY && has_sharding_key)
throw Exception(exception_message.str(), ErrorCodes::UNABLE_TO_SKIP_UNUSED_SHARDS);
}
return cluster;
}
IColumn::Selector StorageDistributed::createSelector(const ClusterPtr cluster, const ColumnWithTypeAndName & result)
{
const auto & slot_to_shard = cluster->getSlotToShard();
// If result.type is DataTypeLowCardinality, do shard according to its dictionaryType
#define CREATE_FOR_TYPE(TYPE) \
if (typeid_cast<const DataType##TYPE *>(result.type.get())) \
return createBlockSelector<TYPE>(*result.column, slot_to_shard); \
else if (auto * type_low_cardinality = typeid_cast<const DataTypeLowCardinality *>(result.type.get())) \
if (typeid_cast<const DataType ## TYPE *>(type_low_cardinality->getDictionaryType().get())) \
return createBlockSelector<TYPE>(*result.column->convertToFullColumnIfLowCardinality(), slot_to_shard);
CREATE_FOR_TYPE(UInt8)
CREATE_FOR_TYPE(UInt16)
CREATE_FOR_TYPE(UInt32)
CREATE_FOR_TYPE(UInt64)
CREATE_FOR_TYPE(Int8)
CREATE_FOR_TYPE(Int16)
CREATE_FOR_TYPE(Int32)
CREATE_FOR_TYPE(Int64)
#undef CREATE_FOR_TYPE
throw Exception{"Sharding key expression does not evaluate to an integer type", ErrorCodes::TYPE_MISMATCH};
}
2018-12-19 12:38:13 +00:00
/// Returns a new cluster with fewer shards if constant folding for `sharding_key_expr` is possible
/// using constraints from "PREWHERE" and "WHERE" conditions, otherwise returns `nullptr`
ClusterPtr StorageDistributed::skipUnusedShards(
ClusterPtr cluster,
const ASTPtr & query_ptr,
const StorageMetadataPtr & metadata_snapshot,
const Context & context) const
2018-12-19 12:38:13 +00:00
{
const auto & select = query_ptr->as<ASTSelectQuery &>();
2018-12-19 12:38:13 +00:00
if (!select.prewhere() && !select.where())
{
2018-12-19 12:38:13 +00:00
return nullptr;
}
ASTPtr condition_ast;
if (select.prewhere() && select.where())
{
condition_ast = makeASTFunction("and", select.prewhere()->clone(), select.where()->clone());
}
else
{
condition_ast = select.prewhere() ? select.prewhere()->clone() : select.where()->clone();
}
2018-12-19 12:38:13 +00:00
2020-06-18 09:08:24 +00:00
replaceConstantExpressions(condition_ast, context, metadata_snapshot->getColumns().getAll(), shared_from_this(), metadata_snapshot);
const auto blocks = evaluateExpressionOverConstantCondition(condition_ast, sharding_key_expr);
2018-12-19 12:38:13 +00:00
// Can't get definite answer if we can skip any shards
if (!blocks)
{
return nullptr;
}
std::set<int> shards;
for (const auto & block : *blocks)
{
if (!block.has(sharding_key_column_name))
throw Exception("sharding_key_expr should evaluate as a single row", ErrorCodes::TOO_MANY_ROWS);
2020-03-18 03:27:32 +00:00
const ColumnWithTypeAndName & result = block.getByName(sharding_key_column_name);
2018-12-19 12:38:13 +00:00
const auto selector = createSelector(cluster, result);
shards.insert(selector.begin(), selector.end());
}
return cluster->getClusterWithMultipleShards({shards.begin(), shards.end()});
}
ActionLock StorageDistributed::getActionLock(StorageActionBlockType type)
{
2019-04-22 15:11:16 +00:00
if (type == ActionLocks::DistributedSend)
return monitors_blocker.cancel();
return {};
}
void StorageDistributed::flushClusterNodesAllData(const Context & context)
{
/// Sync SYSTEM FLUSH DISTRIBUTED with TRUNCATE
auto table_lock = lockForShare(context.getCurrentQueryId(), context.getSettingsRef().lock_acquire_timeout);
std::vector<std::shared_ptr<StorageDistributedDirectoryMonitor>> directory_monitors;
{
std::lock_guard lock(cluster_nodes_mutex);
directory_monitors.reserve(cluster_nodes_data.size());
for (auto & node : cluster_nodes_data)
directory_monitors.push_back(node.second.directory_monitor);
}
/// TODO: Maybe it should be executed in parallel
for (auto & node : directory_monitors)
node->flushAllData();
}
2020-04-07 14:05:51 +00:00
void StorageDistributed::rename(const String & new_path_to_table_data, const StorageID & new_table_id)
2019-12-19 19:39:49 +00:00
{
2020-09-18 19:25:56 +00:00
assert(relative_data_path != new_path_to_table_data);
if (!relative_data_path.empty())
renameOnDisk(new_path_to_table_data);
2020-04-07 14:05:51 +00:00
renameInMemory(new_table_id);
}
2020-07-23 14:10:48 +00:00
size_t StorageDistributed::getRandomShardIndex(const Cluster::ShardsInfo & shards)
{
UInt32 total_weight = 0;
for (const auto & shard : shards)
total_weight += shard.weight;
assert(total_weight > 0);
size_t res;
{
std::lock_guard lock(rng_mutex);
res = std::uniform_int_distribution<size_t>(0, total_weight - 1)(rng);
}
for (auto i = 0ul, s = shards.size(); i < s; ++i)
{
if (shards[i].weight > res)
return i;
res -= shards[i].weight;
}
__builtin_unreachable();
}
void StorageDistributed::renameOnDisk(const String & new_path_to_table_data)
{
for (const DiskPtr & disk : data_volume->getDisks())
2019-12-19 19:39:49 +00:00
{
2020-09-18 19:08:53 +00:00
disk->moveDirectory(relative_data_path, new_path_to_table_data);
2020-09-18 19:08:53 +00:00
auto new_path = disk->getPath() + new_path_to_table_data;
2020-05-23 22:24:01 +00:00
LOG_DEBUG(log, "Updating path to {}", new_path);
2019-12-19 19:39:49 +00:00
std::lock_guard lock(cluster_nodes_mutex);
for (auto & node : cluster_nodes_data)
node.second.directory_monitor->updatePath(new_path_to_table_data);
2019-12-19 19:39:49 +00:00
}
relative_data_path = new_path_to_table_data;
2019-12-19 19:39:49 +00:00
}
void StorageDistributed::throwInsertIfNeeded() const
{
if (!distributed_settings.bytes_to_throw_insert)
return;
UInt64 total_bytes = *totalBytes(global_context.getSettingsRef());
if (total_bytes > distributed_settings.bytes_to_throw_insert)
{
throw Exception(ErrorCodes::DISTRIBUTED_TOO_MANY_PENDING_BYTES,
"Too many bytes pending for async INSERT: {} (bytes_to_throw_insert={})",
formatReadableSizeWithBinarySuffix(total_bytes),
formatReadableSizeWithBinarySuffix(distributed_settings.bytes_to_throw_insert));
}
}
void registerStorageDistributed(StorageFactory & factory)
{
factory.registerStorage("Distributed", [](const StorageFactory::Arguments & args)
{
/** Arguments of engine is following:
* - name of cluster in configuration;
* - name of remote database;
* - name of remote table;
* - policy to store data in;
*
* Remote database may be specified in following form:
* - identifier;
* - constant expression with string result, like currentDatabase();
* -- string literal as specific case;
* - empty string means 'use default database from cluster'.
*
* Distributed engine also supports SETTINGS clause.
*/
ASTs & engine_args = args.engine_args;
if (engine_args.size() < 3 || engine_args.size() > 5)
throw Exception(
"Storage Distributed requires from 3 to 5 parameters - "
"name of configuration section with list of remote servers, "
"name of remote database, "
"name of remote table, "
"sharding key expression (optional), "
"policy to store data in (optional).",
ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
2020-02-21 13:44:44 +00:00
String cluster_name = getClusterNameAndMakeLiteral(engine_args[0]);
engine_args[1] = evaluateConstantExpressionOrIdentifierAsLiteral(engine_args[1], args.local_context);
engine_args[2] = evaluateConstantExpressionOrIdentifierAsLiteral(engine_args[2], args.local_context);
2019-03-15 17:09:14 +00:00
String remote_database = engine_args[1]->as<ASTLiteral &>().value.safeGet<String>();
String remote_table = engine_args[2]->as<ASTLiteral &>().value.safeGet<String>();
const auto & sharding_key = engine_args.size() >= 4 ? engine_args[3] : nullptr;
2020-07-23 12:46:06 +00:00
const auto & storage_policy = engine_args.size() >= 5 ? engine_args[4]->as<ASTLiteral &>().value.safeGet<String>() : "default";
/// Check that sharding_key exists in the table and has numeric type.
if (sharding_key)
{
auto sharding_expr = buildShardingKeyExpression(sharding_key, args.context, args.columns.getAllPhysical(), true);
const Block & block = sharding_expr->getSampleBlock();
if (block.columns() != 1)
throw Exception("Sharding expression must return exactly one column", ErrorCodes::INCORRECT_NUMBER_OF_COLUMNS);
auto type = block.getByPosition(0).type;
if (!type->isValueRepresentedByInteger())
throw Exception("Sharding expression has type " + type->getName() +
", but should be one of integer type", ErrorCodes::TYPE_MISMATCH);
}
/// TODO: move some arguments from the arguments to the SETTINGS.
DistributedSettings distributed_settings;
if (args.storage_def->settings)
{
distributed_settings.loadFromQuery(*args.storage_def);
}
return StorageDistributed::create(
2019-12-04 16:06:55 +00:00
args.table_id, args.columns, args.constraints,
remote_database, remote_table, cluster_name,
args.context,
sharding_key,
storage_policy,
args.relative_data_path,
distributed_settings,
args.attach);
},
{
.supports_settings = true,
.supports_parallel_insert = true,
.source_access_type = AccessType::REMOTE,
});
}
2012-05-21 20:38:34 +00:00
}