ClickHouse/src/Interpreters/InterpreterInsertQuery.cpp

470 lines
19 KiB
C++
Raw Normal View History

#include <Interpreters/InterpreterInsertQuery.h>
2017-07-13 20:58:19 +00:00
#include <Access/Common/AccessFlags.h>
#include <Columns/ColumnNullable.h>
2021-10-08 14:03:54 +00:00
#include <Processors/Transforms/buildPushingToViewsChain.h>
#include <DataTypes/DataTypeNullable.h>
#include <IO/ConnectionTimeoutsContext.h>
#include <Interpreters/InterpreterSelectWithUnionQuery.h>
#include <Interpreters/InterpreterWatchQuery.h>
#include <Interpreters/QueryLog.h>
#include <Interpreters/TranslateQualifiedNamesVisitor.h>
#include <Interpreters/addMissingDefaults.h>
#include <Interpreters/getTableExpressions.h>
#include <Interpreters/processColumnTransformers.h>
#include <Parsers/ASTFunction.h>
#include <Parsers/ASTInsertQuery.h>
#include <Parsers/ASTSelectQuery.h>
2018-02-25 06:34:20 +00:00
#include <Parsers/ASTSelectWithUnionQuery.h>
#include <Parsers/ASTTablesInSelectQuery.h>
2021-09-26 14:54:59 +00:00
#include <Processors/Sinks/EmptySink.h>
#include <Processors/Transforms/CheckConstraintsTransform.h>
#include <Processors/Transforms/CountingTransform.h>
2020-11-17 17:16:55 +00:00
#include <Processors/Transforms/ExpressionTransform.h>
#include <Processors/Transforms/SquashingChunksTransform.h>
#include <Processors/Transforms/getSourceFromASTInsertQuery.h>
#include <Storages/StorageDistributed.h>
2021-01-05 03:22:06 +00:00
#include <Storages/StorageMaterializedView.h>
#include <TableFunctions/TableFunctionFactory.h>
#include <Common/checkStackSize.h>
2011-10-30 11:30:52 +00:00
namespace DB
{
namespace ErrorCodes
{
2021-07-15 08:23:59 +00:00
extern const int NOT_IMPLEMENTED;
extern const int NO_SUCH_COLUMN_IN_TABLE;
extern const int ILLEGAL_COLUMN;
extern const int DUPLICATE_COLUMN;
}
2018-01-12 13:03:19 +00:00
InterpreterInsertQuery::InterpreterInsertQuery(
2021-09-19 20:15:10 +00:00
const ASTPtr & query_ptr_, ContextPtr context_, bool allow_materialized_, bool no_squash_, bool no_destination_, bool async_insert_)
: WithContext(context_)
, query_ptr(query_ptr_)
, allow_materialized(allow_materialized_)
, no_squash(no_squash_)
, no_destination(no_destination_)
2021-09-19 20:15:10 +00:00
, async_insert(async_insert_)
2011-10-30 11:30:52 +00:00
{
2019-08-10 17:51:47 +00:00
checkStackSize();
2011-10-30 11:30:52 +00:00
}
2021-09-17 12:59:40 +00:00
2020-01-24 16:20:36 +00:00
StoragePtr InterpreterInsertQuery::getTable(ASTInsertQuery & query)
2011-10-30 11:30:52 +00:00
{
if (query.table_function)
{
const auto & factory = TableFunctionFactory::instance();
TableFunctionPtr table_function_ptr = factory.get(query.table_function, getContext());
return table_function_ptr->execute(query.table_function, getContext(), table_function_ptr->getName());
}
2021-10-15 08:41:25 +00:00
if (query.table_id)
2021-10-12 23:51:11 +00:00
{
query.table_id = getContext()->resolveStorageID(query.table_id);
}
else
{
/// Insert query parser does not fill table_id because table and
2021-11-11 13:28:18 +00:00
/// database can be parameters and be filled after parsing.
2021-10-12 23:51:11 +00:00
StorageID local_table_id(query.getDatabase(), query.getTable());
query.table_id = getContext()->resolveStorageID(local_table_id);
}
return DatabaseCatalog::instance().getTable(query.table_id, getContext());
2011-10-30 11:30:52 +00:00
}
Block InterpreterInsertQuery::getSampleBlock(
const ASTInsertQuery & query,
const StoragePtr & table,
const StorageMetadataPtr & metadata_snapshot) const
{
2017-04-02 17:37:49 +00:00
/// If the query does not include information about columns
if (!query.columns)
2018-09-20 12:59:33 +00:00
{
if (no_destination)
return metadata_snapshot->getSampleBlockWithVirtuals(table->getVirtuals());
2018-09-20 12:59:33 +00:00
else
2021-09-15 19:35:48 +00:00
return metadata_snapshot->getSampleBlockNonMaterialized();
2018-09-20 12:59:33 +00:00
}
2013-10-25 14:56:47 +00:00
2017-04-02 17:37:49 +00:00
/// Form the block based on the column names from the query
2021-09-15 19:35:48 +00:00
Names names;
const auto columns_ast = processColumnTransformers(getContext()->getCurrentDatabase(), table, metadata_snapshot, query.columns);
2020-09-03 17:51:16 +00:00
for (const auto & identifier : columns_ast->children)
{
std::string current_name = identifier->getColumnName();
2021-09-15 19:35:48 +00:00
names.emplace_back(std::move(current_name));
}
2013-10-25 14:56:47 +00:00
2021-09-15 19:35:48 +00:00
return getSampleBlock(names, table, metadata_snapshot);
}
Block InterpreterInsertQuery::getSampleBlock(
const Names & names,
const StoragePtr & table,
const StorageMetadataPtr & metadata_snapshot) const
{
Block table_sample = metadata_snapshot->getSampleBlock();
Block table_sample_non_materialized = metadata_snapshot->getSampleBlockNonMaterialized();
Block res;
for (const auto & current_name : names)
{
2017-04-02 17:37:49 +00:00
/// The table does not have a column with that name
if (!table_sample.has(current_name))
2021-01-13 13:55:34 +00:00
throw Exception("No such column " + current_name + " in table " + table->getStorageID().getNameForLogs(),
2020-08-24 14:29:31 +00:00
ErrorCodes::NO_SUCH_COLUMN_IN_TABLE);
2012-03-19 12:57:56 +00:00
if (!allow_materialized && !table_sample_non_materialized.has(current_name))
throw Exception("Cannot insert column " + current_name + ", because it is MATERIALIZED column.", ErrorCodes::ILLEGAL_COLUMN);
if (res.has(current_name))
throw Exception("Column " + current_name + " specified more than once", ErrorCodes::DUPLICATE_COLUMN);
res.insert(ColumnWithTypeAndName(table_sample.getByName(current_name).type, current_name));
}
return res;
2013-10-25 14:56:47 +00:00
}
2012-03-19 12:57:56 +00:00
2011-10-30 11:30:52 +00:00
2020-08-24 14:29:31 +00:00
/** A query that just reads all data without any complex computations or filetering.
* If we just pipe the result to INSERT, we don't have to use too many threads for read.
*/
2020-11-02 05:28:37 +00:00
static bool isTrivialSelect(const ASTPtr & select)
2020-08-24 14:29:31 +00:00
{
2020-11-02 08:02:35 +00:00
if (auto * select_query = select->as<ASTSelectQuery>())
2020-11-02 05:28:37 +00:00
{
const auto & tables = select_query->tables();
2020-08-24 14:29:31 +00:00
2020-11-02 05:28:37 +00:00
if (!tables)
return false;
2020-08-24 14:29:31 +00:00
2020-11-02 05:28:37 +00:00
const auto & tables_in_select_query = tables->as<ASTTablesInSelectQuery &>();
2020-08-24 14:29:31 +00:00
2020-11-02 05:28:37 +00:00
if (tables_in_select_query.children.size() != 1)
return false;
2020-08-24 14:29:31 +00:00
2020-11-02 05:28:37 +00:00
const auto & child = tables_in_select_query.children.front();
const auto & table_element = child->as<ASTTablesInSelectQueryElement &>();
const auto & table_expr = table_element.table_expression->as<ASTTableExpression &>();
2020-08-24 14:29:31 +00:00
2020-11-02 05:28:37 +00:00
if (table_expr.subquery)
return false;
2020-08-24 14:29:31 +00:00
2020-11-02 05:28:37 +00:00
/// Note: how to write it in more generic way?
return (!select_query->distinct
&& !select_query->limit_with_ties
&& !select_query->prewhere()
&& !select_query->where()
&& !select_query->groupBy()
&& !select_query->having()
&& !select_query->orderBy()
&& !select_query->limitBy());
}
/// This query is ASTSelectWithUnionQuery subquery
return false;
2020-08-24 14:29:31 +00:00
};
2021-09-15 19:35:48 +00:00
Chain InterpreterInsertQuery::buildChain(
const StoragePtr & table,
const StorageMetadataPtr & metadata_snapshot,
const Names & columns,
2021-09-21 10:35:41 +00:00
ThreadStatus * thread_status,
std::atomic_uint64_t * elapsed_counter_ms)
2021-09-15 19:35:48 +00:00
{
2021-09-21 10:35:41 +00:00
auto sample = getSampleBlock(columns, table, metadata_snapshot);
return buildChainImpl(table, metadata_snapshot, std::move(sample) , thread_status, elapsed_counter_ms);
2021-09-15 19:35:48 +00:00
}
Chain InterpreterInsertQuery::buildChainImpl(
const StoragePtr & table,
const StorageMetadataPtr & metadata_snapshot,
const Block & query_sample_block,
2021-09-21 10:35:41 +00:00
ThreadStatus * thread_status,
std::atomic_uint64_t * elapsed_counter_ms)
2021-09-15 19:35:48 +00:00
{
2021-09-17 12:05:54 +00:00
auto context_ptr = getContext();
2021-09-15 19:35:48 +00:00
const ASTInsertQuery * query = nullptr;
if (query_ptr)
query = query_ptr->as<ASTInsertQuery>();
2021-09-17 12:05:54 +00:00
const Settings & settings = context_ptr->getSettingsRef();
bool null_as_default = query && query->select && context_ptr->getSettingsRef().insert_null_as_default;
2021-09-15 19:35:48 +00:00
/// We create a pipeline of several streams, into which we will write data.
Chain out;
/// NOTE: we explicitly ignore bound materialized views when inserting into Kafka Storage.
/// Otherwise we'll get duplicates when MV reads same rows again from Kafka.
if (table->noPushingToViews() && !no_destination)
{
2021-09-17 12:05:54 +00:00
auto sink = table->write(query_ptr, metadata_snapshot, context_ptr);
2021-09-21 10:35:41 +00:00
sink->setRuntimeData(thread_status, elapsed_counter_ms);
2021-09-15 19:35:48 +00:00
out.addSource(std::move(sink));
}
else
{
2021-09-22 10:57:00 +00:00
out = buildPushingToViewsChain(table, metadata_snapshot, context_ptr, query_ptr, no_destination, thread_status, elapsed_counter_ms);
2021-09-15 19:35:48 +00:00
}
/// Note that we wrap transforms one on top of another, so we write them in reverse of data processing order.
/// Checking constraints. It must be done after calculation of all defaults, so we can check them on calculated columns.
if (const auto & constraints = metadata_snapshot->getConstraints(); !constraints.empty())
out.addSource(std::make_shared<CheckConstraintsTransform>(
2021-09-17 12:05:54 +00:00
table->getStorageID(), out.getInputHeader(), metadata_snapshot->getConstraints(), context_ptr));
2021-09-15 19:35:48 +00:00
auto adding_missing_defaults_dag = addMissingDefaults(
query_sample_block,
out.getInputHeader().getNamesAndTypesList(),
metadata_snapshot->getColumns(),
2021-09-17 12:05:54 +00:00
context_ptr,
2021-09-15 19:35:48 +00:00
null_as_default);
auto adding_missing_defaults_actions = std::make_shared<ExpressionActions>(adding_missing_defaults_dag);
/// Actually we don't know structure of input blocks from query/table,
/// because some clients break insertion protocol (columns != header)
out.addSource(std::make_shared<ConvertingTransform>(query_sample_block, adding_missing_defaults_actions));
/// It's important to squash blocks as early as possible (before other transforms),
/// because other transforms may work inefficient if block size is small.
/// Do not squash blocks if it is a sync INSERT into Distributed, since it lead to double bufferization on client and server side.
/// Client-side bufferization might cause excessive timeouts (especially in case of big blocks).
if (!(settings.insert_distributed_sync && table->isRemote()) && !no_squash && !(query && query->watch))
{
bool table_prefers_large_blocks = table->prefersLargeBlocks();
out.addSource(std::make_shared<SquashingChunksTransform>(
out.getInputHeader(),
table_prefers_large_blocks ? settings.min_insert_block_size_rows : settings.max_block_size,
table_prefers_large_blocks ? settings.min_insert_block_size_bytes : 0));
}
2021-09-21 10:35:41 +00:00
auto counting = std::make_shared<CountingTransform>(out.getInputHeader(), thread_status);
2021-09-17 12:05:54 +00:00
counting->setProcessListElement(context_ptr->getProcessListElement());
2021-09-15 19:35:48 +00:00
out.addSource(std::move(counting));
return out;
}
2020-08-24 14:29:31 +00:00
2021-09-17 12:59:40 +00:00
BlockIO InterpreterInsertQuery::execute()
2012-03-11 08:52:56 +00:00
{
2021-09-17 12:59:40 +00:00
const Settings & settings = getContext()->getSettingsRef();
auto & query = query_ptr->as<ASTInsertQuery &>();
QueryPipelineBuilder pipeline;
2012-03-11 08:52:56 +00:00
2021-09-17 12:59:40 +00:00
StoragePtr table = getTable(query);
2021-07-14 08:49:05 +00:00
if (query.partition_by && !table->supportsPartitionBy())
2021-07-15 08:23:59 +00:00
throw Exception(ErrorCodes::NOT_IMPLEMENTED, "PARTITION BY clause is not supported by storage");
2021-07-14 08:49:05 +00:00
2021-09-17 12:59:40 +00:00
auto table_lock = table->lockForShare(getContext()->getInitialQueryId(), settings.lock_acquire_timeout);
auto metadata_snapshot = table->getInMemoryMetadataPtr();
auto query_sample_block = getSampleBlock(query, table, metadata_snapshot);
if (!query.table_function)
getContext()->checkAccess(AccessType::INSERT, query.table_id, query_sample_block.getNames());
2020-01-24 16:20:36 +00:00
bool is_distributed_insert_select = false;
2021-09-17 12:59:40 +00:00
2020-03-19 17:04:59 +00:00
if (query.select && table->isRemote() && settings.parallel_distributed_insert_select)
{
// Distributed INSERT SELECT
if (auto maybe_pipeline = table->distributedWrite(query, getContext()))
2020-01-24 09:09:47 +00:00
{
2021-09-15 19:35:48 +00:00
pipeline = std::move(*maybe_pipeline);
is_distributed_insert_select = true;
2020-01-24 09:09:47 +00:00
}
}
2019-12-12 10:49:15 +00:00
2021-09-17 12:59:40 +00:00
std::vector<Chain> out_chains;
if (!is_distributed_insert_select || query.watch)
2019-12-12 10:49:15 +00:00
{
size_t out_streams_size = 1;
2021-09-15 19:35:48 +00:00
if (query.select)
{
2020-08-24 14:29:31 +00:00
bool is_trivial_insert_select = false;
if (settings.optimize_trivial_insert_select)
2020-08-02 05:35:58 +00:00
{
2020-11-10 06:42:38 +00:00
const auto & select_query = query.select->as<ASTSelectWithUnionQuery &>();
const auto & selects = select_query.list_of_selects->children;
const auto & union_modes = select_query.list_of_modes;
/// ASTSelectWithUnionQuery is not normalized now, so it may pass some queries which can be Trivial select queries
const auto mode_is_all = [](const auto & mode) { return mode == SelectUnionMode::ALL; };
SYSTEM RESTORE REPLICA replica [ON CLUSTER cluster] (#13652) * initial commit: add setting and stub * typo * added test stub * fix * wip merging new integration test and code proto * adding steps interpreters * adding firstly proposed solution (moving parts etc) * added checking zookeeper path existence * fixing the include * fixing and sorting includes * fixing outdated struct * fix the name * added ast ptr as level of indirection * fix ref * updating the changes * working on test stub * fix iterator -> reference * revert rocksdb submodule update * fixed show privileges test * updated the test stub * replaced rand() with thread_local_rng(), updated the tests updated the test fixed test config path test fix removed error messages fixed the test updated the test fixed string literal fixed literal typo: = * fixed the empty replica error message * updated the test and the code with logs * updated the possible test cases, updated * added the code/test milestone comments * updated the test (added more testcases) * replaced native assert with CH one * individual replicas recursive delete fix * updated the AS db.name AST * two small logging fixes * manually generated AST fixes * Updated the test, added the possible algo change * Some thoughts about optimizing the solution: ALTER MOVE PARTITION .. TO TABLE -> move to detached/ + ALTER ... ATTACH * fix * Removed the replica sync in test as it's invalid * Some test tweaks * tmp * Rewrote the algo by using the executeQuery instead of hand-crafting the ASTPtr. Two questions still active. * tr: logging active parts * Extracted the parts moving algo into a separate helper function * Fixed the test data and the queries slightly * Replaced query to system.parts to direct invocation, started building the test that breaks on various parts. * Added the case for tables when at least one replica is alive * Updated the test to test replicas restoration by detaching/attaching * Altered the test to check restoration without replica restart * Added the tables swap in the start if the server failed last time * Hotfix when only /replicas/replica... path was deleted * Restore ZK paths while creating a replicated MergeTree table * Updated the docs, fixed the algo for individual replicas restoration case * Initial parts table storage fix, tests sync fix * Reverted individual replica restoration to general algo * Slightly optimised getDataParts * Trying another solution with parts detaching * Rewrote algo without any steps, added ON CLUSTER support * Attaching parts from other replica on restoration * Getting part checksums from ZK * Removed ON CLUSTER, finished working solution * Multiple small changes after review * Fixing parallel test * Supporting rewritten form on cluster * Test fix * Moar logging * Using source replica as checksum provider * improve test, remove some code from parser * Trying solution with move to detached + forget * Moving all parts (not only Committed) to detached * Edited docs for RESTORE REPLICA * Re-merging * minor fixes Co-authored-by: Alexander Tokmakov <avtokmakov@yandex-team.ru>
2021-06-20 08:24:43 +00:00
is_trivial_insert_select =
std::all_of(union_modes.begin(), union_modes.end(), std::move(mode_is_all))
&& std::all_of(selects.begin(), selects.end(), isTrivialSelect);
2020-08-24 14:29:31 +00:00
}
if (is_trivial_insert_select)
{
/** When doing trivial INSERT INTO ... SELECT ... FROM table,
* don't need to process SELECT with more than max_insert_threads
* and it's reasonable to set block size for SELECT to the desired block size for INSERT
* to avoid unnecessary squashing.
*/
Settings new_settings = getContext()->getSettings();
2020-08-24 14:29:31 +00:00
new_settings.max_threads = std::max<UInt64>(1, settings.max_insert_threads);
if (table->prefersLargeBlocks())
{
if (settings.min_insert_block_size_rows)
new_settings.max_block_size = settings.min_insert_block_size_rows;
if (settings.min_insert_block_size_bytes)
new_settings.preferred_block_size_bytes = settings.min_insert_block_size_bytes;
}
2020-08-24 14:29:31 +00:00
auto new_context = Context::createCopy(context);
new_context->setSettings(new_settings);
2020-08-24 14:29:31 +00:00
InterpreterSelectWithUnionQuery interpreter_select{
query.select, new_context, SelectQueryOptions(QueryProcessingStage::Complete, 1)};
2021-09-15 19:35:48 +00:00
pipeline = interpreter_select.buildQueryPipeline();
}
else
{
2020-08-24 14:29:31 +00:00
/// Passing 1 as subquery_depth will disable limiting size of intermediate result.
InterpreterSelectWithUnionQuery interpreter_select{
query.select, getContext(), SelectQueryOptions(QueryProcessingStage::Complete, 1)};
2021-09-15 19:35:48 +00:00
pipeline = interpreter_select.buildQueryPipeline();
}
2019-12-12 10:49:15 +00:00
2021-09-15 19:35:48 +00:00
pipeline.dropTotalsAndExtremes();
if (table->supportsParallelInsert() && settings.max_insert_threads > 1)
2021-09-15 19:35:48 +00:00
out_streams_size = std::min(size_t(settings.max_insert_threads), pipeline.getNumStreams());
2020-05-27 18:20:26 +00:00
2021-09-15 19:35:48 +00:00
pipeline.resize(out_streams_size);
2021-04-22 21:30:31 +00:00
/// Allow to insert Nullable into non-Nullable columns, NULL values will be added as defaults values.
2021-04-22 22:13:07 +00:00
if (getContext()->getSettingsRef().insert_null_as_default)
2021-04-22 21:30:31 +00:00
{
2021-09-15 19:35:48 +00:00
const auto & input_columns = pipeline.getHeader().getColumnsWithTypeAndName();
2021-09-17 12:59:40 +00:00
const auto & query_columns = query_sample_block.getColumnsWithTypeAndName();
const auto & output_columns = metadata_snapshot->getColumns();
2021-04-23 19:08:13 +00:00
if (input_columns.size() == query_columns.size())
{
2021-04-23 19:08:13 +00:00
for (size_t col_idx = 0; col_idx < query_columns.size(); ++col_idx)
{
/// Change query sample block columns to Nullable to allow inserting nullable columns, where NULL values will be substituted with
/// default column values (in AddingDefaultBlockOutputStream), so all values will be cast correctly.
if (input_columns[col_idx].type->isNullable() && !query_columns[col_idx].type->isNullable() && output_columns.hasDefault(query_columns[col_idx].name))
2021-09-17 12:59:40 +00:00
query_sample_block.setColumn(col_idx, ColumnWithTypeAndName(makeNullable(query_columns[col_idx].column), makeNullable(query_columns[col_idx].type), query_columns[col_idx].name));
2021-04-23 19:08:13 +00:00
}
}
}
}
else if (query.watch)
{
InterpreterWatchQuery interpreter_watch{ query.watch, getContext() };
2021-09-16 17:40:42 +00:00
pipeline = interpreter_watch.buildQueryPipeline();
}
2018-09-20 11:40:04 +00:00
for (size_t i = 0; i < out_streams_size; i++)
2019-12-12 10:49:15 +00:00
{
2021-09-21 10:35:41 +00:00
auto out = buildChainImpl(table, metadata_snapshot, query_sample_block, nullptr, nullptr);
out_chains.emplace_back(std::move(out));
}
2019-12-12 10:49:15 +00:00
}
2021-09-19 18:53:36 +00:00
BlockIO res;
/// What type of query: INSERT or INSERT SELECT or INSERT WATCH?
2021-09-17 12:59:40 +00:00
if (is_distributed_insert_select)
2020-05-27 18:20:26 +00:00
{
2021-09-15 19:35:48 +00:00
res.pipeline = QueryPipelineBuilder::getPipeline(std::move(pipeline));
2020-05-27 18:20:26 +00:00
}
else if (query.select || query.watch)
{
const auto & header = out_chains.at(0).getInputHeader();
2020-11-17 17:16:55 +00:00
auto actions_dag = ActionsDAG::makeConvertingActions(
2021-09-15 19:35:48 +00:00
pipeline.getHeader().getColumnsWithTypeAndName(),
2020-11-17 17:16:55 +00:00
header.getColumnsWithTypeAndName(),
ActionsDAG::MatchColumnsMode::Position);
2021-05-19 14:32:07 +00:00
auto actions = std::make_shared<ExpressionActions>(actions_dag, ExpressionActionsSettings::fromContext(getContext(), CompileExpressions::yes));
2020-05-27 18:20:26 +00:00
2021-09-15 19:35:48 +00:00
pipeline.addSimpleTransform([&](const Block & in_header) -> ProcessorPtr
2019-12-12 10:49:15 +00:00
{
2020-11-17 17:16:55 +00:00
return std::make_shared<ExpressionTransform>(in_header, actions);
2020-05-27 18:20:26 +00:00
});
2019-12-12 10:49:15 +00:00
size_t num_select_threads = pipeline.getNumThreads();
size_t num_insert_threads = std::max_element(out_chains.begin(), out_chains.end(), [&](const auto &a, const auto &b)
{
return a.getNumThreads() < b.getNumThreads();
})->getNumThreads();
2021-09-15 19:35:48 +00:00
pipeline.addChains(std::move(out_chains));
pipeline.setMaxThreads(num_insert_threads);
/// Don't use more threads for insert then for select to reduce memory consumption.
if (!settings.parallel_view_processing && pipeline.getNumThreads() > num_select_threads)
pipeline.setMaxThreads(num_select_threads);
2021-09-15 19:35:48 +00:00
pipeline.setSinks([&](const Block & cur_header, QueryPipelineBuilder::StreamType) -> ProcessorPtr
{
2021-09-26 14:54:59 +00:00
return std::make_shared<EmptySink>(cur_header);
2020-05-27 18:20:26 +00:00
});
if (!allow_materialized)
{
for (const auto & column : metadata_snapshot->getColumns())
2020-05-27 18:20:26 +00:00
if (column.default_desc.kind == ColumnDefaultKind::Materialized && header.has(column.name))
throw Exception("Cannot insert column " + column.name + ", because it is MATERIALIZED column.", ErrorCodes::ILLEGAL_COLUMN);
}
2021-09-15 19:35:48 +00:00
res.pipeline = QueryPipelineBuilder::getPipeline(std::move(pipeline));
}
2019-12-16 13:52:32 +00:00
else
2021-09-10 12:38:30 +00:00
{
2021-09-15 19:35:48 +00:00
res.pipeline = QueryPipeline(std::move(out_chains.at(0)));
res.pipeline.setNumThreads(std::min<size_t>(res.pipeline.getNumThreads(), settings.max_threads));
2020-05-27 18:20:26 +00:00
2021-09-19 20:15:10 +00:00
if (query.hasInlinedData() && !async_insert)
2021-09-15 19:35:48 +00:00
{
/// can execute without additional data
2021-09-17 17:52:26 +00:00
auto pipe = getSourceFromASTInsertQuery(query_ptr, true, query_sample_block, getContext(), nullptr);
2021-09-15 19:35:48 +00:00
res.pipeline.complete(std::move(pipe));
}
}
Fix table lifetime in case of parallel DROP TABLE and INSERT Stress tests founds [1]: ==527==WARNING: MemorySanitizer: use-of-uninitialized-value 0 0x37078ffd in unsigned long std::__1::__cxx_atomic_fetch_add<unsigned long>(std::__1::__cxx_atomic_base_impl<unsigned long>*, unsigned long, std::__1::memory_order) obj-x86_64-linux-gnu/../contrib/libcxx/include/atomic:1050:12 1 0x37078ffd in std::__1::__atomic_base<unsigned long, true>::fetch_add(unsigned long, std::__1::memory_order) obj-x86_64-linux-gnu/../contrib/libcxx/include/atomic:1719:17 2 0x37078ffd in std::__1::__atomic_base<unsigned long, true>::operator++() obj-x86_64-linux-gnu/../contrib/libcxx/include/atomic:1756:57 3 0x37078ffd in SimpleIncrement::get() obj-x86_64-linux-gnu/../src/Common/SimpleIncrement.h:20:16 4 0x37078ffd in DB::MergeTreeDataWriter::writeTempPart(DB::BlockWithPartition&, std::__1::shared_ptr<DB::StorageInMemoryMetadata const> const&, std::__1::shared_ptr<DB::Context const>) obj-x86_64-linux-gnu/../src/Storages/MergeTree/MergeTreeDataWriter.cpp:276:46 5 0x373c446c in DB::MergeTreeSink::consume(DB::Chunk) obj-x86_64-linux-gnu/../src/Storages/MergeTree/MergeTreeSink.cpp:27:65 Uninitialized value was created by a heap deallocation 6 0x32d481e8 in DB::DatabaseCatalog::TableMarkedAsDropped::~TableMarkedAsDropped() obj-x86_64-linux-gnu/../src/Interpreters/DatabaseCatalog.h:248:12 7 0x32d3c134 in DB::DatabaseCatalog::dropTableDataTask() obj-x86_64-linux-gnu/../src/Interpreters/DatabaseCatalog.cpp:908:1 [1]: https://s3.amazonaws.com/clickhouse-test-reports/32534/fa6090f588dbf4cbb5f28bd2210847b070bb8218/stress_test__memory__actions_.html The query was CREATE MATERIALIZED VIEW ... POPULATE AS SELECT ... from 00040_aggregating_materialized_view test.
2021-12-11 07:57:23 +00:00
res.pipeline.addStorageHolder(table);
if (const auto * mv = dynamic_cast<const StorageMaterializedView *>(table.get()))
{
if (auto inner_table = mv->tryGetTargetTable())
res.pipeline.addStorageHolder(inner_table);
}
return res;
2012-03-11 08:52:56 +00:00
}
2020-03-02 20:23:58 +00:00
StorageID InterpreterInsertQuery::getDatabaseTable() const
2018-07-16 14:52:02 +00:00
{
2020-03-02 20:23:58 +00:00
return query_ptr->as<ASTInsertQuery &>().table_id;
2018-07-16 14:52:02 +00:00
}
2020-12-14 03:30:39 +00:00
void InterpreterInsertQuery::extendQueryLogElemImpl(QueryLogElement & elem, const ASTPtr &, ContextPtr context_) const
2020-12-14 03:30:39 +00:00
{
elem.query_kind = "Insert";
const auto & insert_table = context_->getInsertionTable();
2020-12-14 03:30:39 +00:00
if (!insert_table.empty())
{
elem.query_databases.insert(insert_table.getDatabaseName());
elem.query_tables.insert(insert_table.getFullNameNotQuoted());
}
}
2011-10-30 11:30:52 +00:00
}