2019-10-22 10:31:28 +00:00
|
|
|
#include <Interpreters/InterpreterInsertQuery.h>
|
2017-07-13 20:58:19 +00:00
|
|
|
|
2021-10-31 08:51:20 +00:00
|
|
|
#include <Access/Common/AccessFlags.h>
|
2022-03-29 17:25:34 +00:00
|
|
|
#include <Access/EnabledQuota.h>
|
2020-12-23 14:48:14 +00:00
|
|
|
#include <AggregateFunctions/AggregateFunctionFactory.h>
|
2021-10-07 08:26:08 +00:00
|
|
|
#include <Columns/ColumnNullable.h>
|
2021-10-08 14:03:54 +00:00
|
|
|
#include <Processors/Transforms/buildPushingToViewsChain.h>
|
2021-10-07 08:26:08 +00:00
|
|
|
#include <DataTypes/DataTypeNullable.h>
|
2019-10-22 10:31:28 +00:00
|
|
|
#include <Interpreters/InterpreterSelectWithUnionQuery.h>
|
2020-04-25 11:33:47 +00:00
|
|
|
#include <Interpreters/InterpreterWatchQuery.h>
|
2021-10-07 08:26:08 +00:00
|
|
|
#include <Interpreters/QueryLog.h>
|
|
|
|
#include <Interpreters/TranslateQualifiedNamesVisitor.h>
|
|
|
|
#include <Interpreters/addMissingDefaults.h>
|
|
|
|
#include <Interpreters/getTableExpressions.h>
|
|
|
|
#include <Interpreters/processColumnTransformers.h>
|
2022-12-15 12:03:09 +00:00
|
|
|
#include <Interpreters/InterpreterSelectQueryAnalyzer.h>
|
2019-10-22 10:31:28 +00:00
|
|
|
#include <Parsers/ASTFunction.h>
|
2017-04-01 09:19:00 +00:00
|
|
|
#include <Parsers/ASTInsertQuery.h>
|
2020-03-19 16:51:09 +00:00
|
|
|
#include <Parsers/ASTSelectQuery.h>
|
2018-02-25 06:34:20 +00:00
|
|
|
#include <Parsers/ASTSelectWithUnionQuery.h>
|
2020-07-07 09:24:49 +00:00
|
|
|
#include <Parsers/ASTTablesInSelectQuery.h>
|
2021-09-26 14:54:59 +00:00
|
|
|
#include <Processors/Sinks/EmptySink.h>
|
2021-10-07 08:26:08 +00:00
|
|
|
#include <Processors/Transforms/CheckConstraintsTransform.h>
|
|
|
|
#include <Processors/Transforms/CountingTransform.h>
|
2020-11-17 17:16:55 +00:00
|
|
|
#include <Processors/Transforms/ExpressionTransform.h>
|
2021-04-15 14:34:55 +00:00
|
|
|
#include <Processors/Transforms/MaterializingTransform.h>
|
2021-09-01 18:41:50 +00:00
|
|
|
#include <Processors/Transforms/SquashingChunksTransform.h>
|
2021-10-07 08:26:08 +00:00
|
|
|
#include <Processors/Transforms/getSourceFromASTInsertQuery.h>
|
2022-05-20 19:49:31 +00:00
|
|
|
#include <Processors/QueryPlan/QueryPlan.h>
|
|
|
|
#include <QueryPipeline/QueryPipelineBuilder.h>
|
2020-03-19 16:51:09 +00:00
|
|
|
#include <Storages/StorageDistributed.h>
|
2021-01-05 03:22:06 +00:00
|
|
|
#include <Storages/StorageMaterializedView.h>
|
2022-05-11 08:47:08 +00:00
|
|
|
#include <Storages/WindowView/StorageWindowView.h>
|
2017-11-02 14:01:11 +00:00
|
|
|
#include <TableFunctions/TableFunctionFactory.h>
|
2023-04-07 13:29:51 +00:00
|
|
|
#include <Common/ThreadStatus.h>
|
2019-10-22 10:31:28 +00:00
|
|
|
#include <Common/checkStackSize.h>
|
2016-10-24 02:02:37 +00:00
|
|
|
|
2018-02-19 03:00:16 +00:00
|
|
|
|
2011-10-30 11:30:52 +00:00
|
|
|
namespace DB
|
|
|
|
{
|
2018-02-19 03:00:16 +00:00
|
|
|
|
2016-01-11 21:46:36 +00:00
|
|
|
namespace ErrorCodes
|
|
|
|
{
|
2021-07-15 08:23:59 +00:00
|
|
|
extern const int NOT_IMPLEMENTED;
|
2016-01-11 21:46:36 +00:00
|
|
|
extern const int NO_SUCH_COLUMN_IN_TABLE;
|
2018-02-19 00:45:32 +00:00
|
|
|
extern const int ILLEGAL_COLUMN;
|
2019-11-08 09:57:32 +00:00
|
|
|
extern const int DUPLICATE_COLUMN;
|
2016-01-11 21:46:36 +00:00
|
|
|
}
|
|
|
|
|
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_)
|
2021-04-10 23:33:54 +00:00
|
|
|
: WithContext(context_)
|
|
|
|
, query_ptr(query_ptr_)
|
2019-10-22 10:31:28 +00:00
|
|
|
, 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();
|
2022-03-29 17:25:34 +00:00
|
|
|
if (auto quota = getContext()->getQuota())
|
|
|
|
quota->checkExceeded(QuotaType::WRITTEN_BYTES);
|
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
|
|
|
{
|
2022-12-15 12:03:09 +00:00
|
|
|
auto current_context = getContext();
|
|
|
|
|
2017-11-02 14:01:11 +00:00
|
|
|
if (query.table_function)
|
|
|
|
{
|
|
|
|
const auto & factory = TableFunctionFactory::instance();
|
2022-12-15 12:03:09 +00:00
|
|
|
TableFunctionPtr table_function_ptr = factory.get(query.table_function, current_context);
|
2022-02-18 16:19:42 +00:00
|
|
|
|
|
|
|
/// If table function needs structure hint from select query
|
|
|
|
/// we can create a temporary pipeline and get the header.
|
|
|
|
if (query.select && table_function_ptr->needStructureHint())
|
|
|
|
{
|
2022-12-15 12:03:09 +00:00
|
|
|
Block header_block;
|
|
|
|
auto select_query_options = SelectQueryOptions(QueryProcessingStage::Complete, 1);
|
|
|
|
|
|
|
|
if (current_context->getSettingsRef().allow_experimental_analyzer)
|
|
|
|
{
|
2023-01-18 09:44:40 +00:00
|
|
|
InterpreterSelectQueryAnalyzer interpreter_select(query.select, current_context, select_query_options);
|
2022-12-15 12:03:09 +00:00
|
|
|
header_block = interpreter_select.getSampleBlock();
|
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
|
|
|
InterpreterSelectWithUnionQuery interpreter_select{
|
|
|
|
query.select, current_context, select_query_options};
|
|
|
|
auto tmp_pipeline = interpreter_select.buildQueryPipeline();
|
|
|
|
header_block = tmp_pipeline.getHeader();
|
|
|
|
}
|
|
|
|
|
|
|
|
ColumnsDescription structure_hint{header_block.getNamesAndTypesList()};
|
2022-02-18 16:19:42 +00:00
|
|
|
table_function_ptr->setStructureHint(structure_hint);
|
|
|
|
}
|
|
|
|
|
2022-12-15 12:03:09 +00:00
|
|
|
return table_function_ptr->execute(query.table_function, current_context, table_function_ptr->getName(),
|
2023-01-03 14:25:04 +00:00
|
|
|
/* cached_columns */ {}, /* use_global_context */ false, /* is_insert_query */true);
|
2017-11-02 14:01:11 +00:00
|
|
|
}
|
|
|
|
|
2021-10-15 08:41:25 +00:00
|
|
|
if (query.table_id)
|
2021-10-12 23:51:11 +00:00
|
|
|
{
|
2022-12-15 12:03:09 +00:00
|
|
|
query.table_id = current_context->resolveStorageID(query.table_id);
|
2021-10-12 23:51:11 +00:00
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
2021-11-15 10:05:24 +00:00
|
|
|
/// 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());
|
2022-12-15 12:03:09 +00:00
|
|
|
query.table_id = current_context->resolveStorageID(local_table_id);
|
2021-10-12 23:51:11 +00:00
|
|
|
}
|
|
|
|
|
2022-12-15 12:03:09 +00:00
|
|
|
return DatabaseCatalog::instance().getTable(query.table_id, current_context);
|
2011-10-30 11:30:52 +00:00
|
|
|
}
|
|
|
|
|
2020-06-16 12:48:10 +00:00
|
|
|
Block InterpreterInsertQuery::getSampleBlock(
|
|
|
|
const ASTInsertQuery & query,
|
|
|
|
const StoragePtr & table,
|
|
|
|
const StorageMetadataPtr & metadata_snapshot) const
|
2017-11-02 14:01:11 +00:00
|
|
|
{
|
2017-04-02 17:37:49 +00:00
|
|
|
/// If the query does not include information about columns
|
2013-10-25 14:56:47 +00:00
|
|
|
if (!query.columns)
|
2018-09-20 12:59:33 +00:00
|
|
|
{
|
2022-05-11 08:47:08 +00:00
|
|
|
if (auto * window_view = dynamic_cast<StorageWindowView *>(table.get()))
|
2022-05-14 09:21:54 +00:00
|
|
|
return window_view->getInputHeader();
|
2022-05-11 08:47:08 +00:00
|
|
|
else if (no_destination)
|
2020-06-16 12:58:05 +00:00
|
|
|
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)
|
2013-10-25 14:56:47 +00:00
|
|
|
{
|
2015-04-10 00:57:42 +00:00
|
|
|
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
|
|
|
|
{
|
2022-02-13 17:42:59 +00:00
|
|
|
Block table_sample_physical = metadata_snapshot->getSampleBlock();
|
|
|
|
Block table_sample_insertable = metadata_snapshot->getSampleBlockInsertable();
|
2021-09-15 19:35:48 +00:00
|
|
|
Block res;
|
|
|
|
for (const auto & current_name : names)
|
|
|
|
{
|
2019-11-08 09:57:32 +00:00
|
|
|
if (res.has(current_name))
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::DUPLICATE_COLUMN, "Column {} specified more than once", current_name);
|
2014-06-26 00:58:14 +00:00
|
|
|
|
2022-02-13 17:42:59 +00:00
|
|
|
/// Column is not ordinary or ephemeral
|
|
|
|
if (!table_sample_insertable.has(current_name))
|
|
|
|
{
|
|
|
|
/// Column is materialized
|
|
|
|
if (table_sample_physical.has(current_name))
|
|
|
|
{
|
|
|
|
if (!allow_materialized)
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::ILLEGAL_COLUMN, "Cannot insert column {}, because it is MATERIALIZED column.",
|
|
|
|
current_name);
|
2022-02-13 17:42:59 +00:00
|
|
|
res.insert(ColumnWithTypeAndName(table_sample_physical.getByName(current_name).type, current_name));
|
|
|
|
}
|
|
|
|
else /// The table does not have a column with that name
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::NO_SUCH_COLUMN_IN_TABLE, "No such column {} in table {}",
|
|
|
|
current_name, table->getStorageID().getNameForLogs());
|
2022-02-13 17:42:59 +00:00
|
|
|
}
|
|
|
|
else
|
|
|
|
res.insert(ColumnWithTypeAndName(table_sample_insertable.getByName(current_name).type, current_name));
|
2018-02-19 00:45:32 +00:00
|
|
|
}
|
2013-10-25 14:56:47 +00:00
|
|
|
return res;
|
|
|
|
}
|
2012-03-19 12:57:56 +00:00
|
|
|
|
2020-12-23 14:48:14 +00:00
|
|
|
static bool hasAggregateFunctions(const IAST * ast)
|
|
|
|
{
|
|
|
|
if (const auto * func = typeid_cast<const ASTFunction *>(ast))
|
2022-06-16 15:41:04 +00:00
|
|
|
if (AggregateUtils::isAggregateFunction(*func))
|
2020-12-23 14:48:14 +00:00
|
|
|
return true;
|
|
|
|
|
|
|
|
for (const auto & child : ast->children)
|
|
|
|
if (hasAggregateFunctions(child.get()))
|
|
|
|
return true;
|
2011-10-30 11:30:52 +00:00
|
|
|
|
2020-12-23 14:48:14 +00:00
|
|
|
return false;
|
|
|
|
}
|
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()
|
2020-12-23 14:48:14 +00:00
|
|
|
&& !select_query->limitBy()
|
|
|
|
&& !hasAggregateFunctions(select_query));
|
2020-11-02 05:28:37 +00:00
|
|
|
}
|
|
|
|
/// This query is ASTSelectWithUnionQuery subquery
|
|
|
|
return false;
|
2022-05-16 18:59:27 +00:00
|
|
|
}
|
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,
|
2022-09-21 18:37:40 +00:00
|
|
|
ThreadStatusesHolderPtr thread_status_holder,
|
2021-09-21 10:35:41 +00:00
|
|
|
std::atomic_uint64_t * elapsed_counter_ms)
|
2021-09-15 19:35:48 +00:00
|
|
|
{
|
2023-04-07 13:29:51 +00:00
|
|
|
ThreadGroupPtr running_group;
|
|
|
|
if (current_thread)
|
|
|
|
running_group = current_thread->getThreadGroup();
|
|
|
|
if (!running_group)
|
|
|
|
running_group = std::make_shared<ThreadGroup>(getContext());
|
|
|
|
|
2021-09-21 10:35:41 +00:00
|
|
|
auto sample = getSampleBlock(columns, table, metadata_snapshot);
|
2023-04-07 13:29:51 +00:00
|
|
|
return buildChainImpl(table, metadata_snapshot, sample, thread_status_holder, running_group, 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,
|
2022-09-21 18:37:40 +00:00
|
|
|
ThreadStatusesHolderPtr thread_status_holder,
|
2023-04-07 13:29:51 +00:00
|
|
|
ThreadGroupPtr running_group,
|
2021-09-21 10:35:41 +00:00
|
|
|
std::atomic_uint64_t * elapsed_counter_ms)
|
2021-09-15 19:35:48 +00:00
|
|
|
{
|
2022-09-22 17:57:04 +00:00
|
|
|
ThreadStatus * thread_status = current_thread;
|
|
|
|
|
|
|
|
if (!thread_status_holder)
|
|
|
|
thread_status = nullptr;
|
|
|
|
|
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;
|
|
|
|
|
2021-12-13 10:50:46 +00:00
|
|
|
/// Keep a reference to the context to make sure it stays alive until the chain is executed and destroyed
|
|
|
|
out.addInterpreterContext(context_ptr);
|
|
|
|
|
2021-09-15 19:35:48 +00:00
|
|
|
/// 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);
|
2022-09-22 17:57:04 +00:00
|
|
|
sink->setRuntimeData(thread_status, elapsed_counter_ms);
|
2021-09-15 19:35:48 +00:00
|
|
|
out.addSource(std::move(sink));
|
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
2023-04-07 13:29:51 +00:00
|
|
|
out = buildPushingToViewsChain(table, metadata_snapshot, context_ptr,
|
|
|
|
query_ptr, no_destination,
|
|
|
|
thread_status_holder, running_group, 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).
|
2022-11-06 22:56:26 +00:00
|
|
|
if (!(settings.insert_distributed_sync && table->isRemote()) && !async_insert && !no_squash && !(query && query->watch))
|
2021-09-15 19:35:48 +00:00
|
|
|
{
|
|
|
|
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,
|
2022-10-07 10:46:45 +00:00
|
|
|
table_prefers_large_blocks ? settings.min_insert_block_size_bytes : 0ULL));
|
2021-09-15 19:35:48 +00:00
|
|
|
}
|
|
|
|
|
2022-09-22 17:57:04 +00:00
|
|
|
auto counting = std::make_shared<CountingTransform>(out.getInputHeader(), thread_status, getContext()->getQuota());
|
2021-09-17 12:05:54 +00:00
|
|
|
counting->setProcessListElement(context_ptr->getProcessListElement());
|
2022-10-05 14:54:56 +00:00
|
|
|
counting->setProgressCallback(context_ptr->getProgressCallback());
|
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;
|
2022-05-20 19:49:31 +00:00
|
|
|
std::optional<QueryPipeline> distributed_pipeline;
|
|
|
|
QueryPlanResourceHolder resources;
|
2012-03-11 08:52:56 +00:00
|
|
|
|
2021-09-17 12:59:40 +00:00
|
|
|
StoragePtr table = getTable(query);
|
2022-01-31 22:27:55 +00:00
|
|
|
checkStorageSupportsTransactionsIfNeeded(table, getContext());
|
|
|
|
|
2022-01-04 10:27:53 +00:00
|
|
|
StoragePtr inner_table;
|
|
|
|
if (const auto * mv = dynamic_cast<const StorageMaterializedView *>(table.get()))
|
|
|
|
inner_table = mv->getTargetTable();
|
|
|
|
|
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);
|
2022-02-28 14:09:46 +00:00
|
|
|
|
|
|
|
/// For table functions we check access while executing
|
|
|
|
/// getTable() -> ITableFunction::execute().
|
2021-09-17 12:59:40 +00:00
|
|
|
if (!query.table_function)
|
|
|
|
getContext()->checkAccess(AccessType::INSERT, query.table_id, query_sample_block.getNames());
|
2020-01-24 16:20:36 +00:00
|
|
|
|
2023-01-09 12:30:32 +00:00
|
|
|
if (query.select && settings.parallel_distributed_insert_select)
|
2020-03-19 16:51:09 +00:00
|
|
|
// Distributed INSERT SELECT
|
2022-05-20 19:49:31 +00:00
|
|
|
distributed_pipeline = table->distributedWrite(query, getContext());
|
2019-12-12 10:49:15 +00:00
|
|
|
|
2021-09-17 12:59:40 +00:00
|
|
|
std::vector<Chain> out_chains;
|
2022-05-20 19:49:31 +00:00
|
|
|
if (!distributed_pipeline || query.watch)
|
2019-12-12 10:49:15 +00:00
|
|
|
{
|
2020-03-19 16:51:09 +00:00
|
|
|
size_t out_streams_size = 1;
|
2021-09-15 19:35:48 +00:00
|
|
|
|
2020-03-19 16:51:09 +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;
|
2020-07-07 09:24:49 +00:00
|
|
|
|
2020-12-24 10:11:07 +00:00
|
|
|
/// ASTSelectWithUnionQuery is not normalized now, so it may pass some queries which can be Trivial select queries
|
2022-08-30 10:09:01 +00:00
|
|
|
const auto mode_is_all = [](const auto & mode) { return mode == SelectUnionMode::UNION_ALL; };
|
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.
|
|
|
|
*/
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
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);
|
|
|
|
|
2021-05-12 09:40:39 +00:00
|
|
|
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
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
auto new_context = Context::createCopy(context);
|
|
|
|
new_context->setSettings(new_settings);
|
2023-04-13 17:35:32 +00:00
|
|
|
new_context->setInsertionTable(getContext()->getInsertionTable());
|
2020-08-24 14:29:31 +00:00
|
|
|
|
2022-12-15 12:03:09 +00:00
|
|
|
auto select_query_options = SelectQueryOptions(QueryProcessingStage::Complete, 1);
|
|
|
|
|
|
|
|
if (settings.allow_experimental_analyzer)
|
|
|
|
{
|
2023-01-18 09:44:40 +00:00
|
|
|
InterpreterSelectQueryAnalyzer interpreter_select_analyzer(query.select, new_context, select_query_options);
|
2023-01-20 11:19:16 +00:00
|
|
|
pipeline = interpreter_select_analyzer.buildQueryPipeline();
|
2022-12-15 12:03:09 +00:00
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
2023-01-21 11:19:09 +00:00
|
|
|
InterpreterSelectWithUnionQuery interpreter_select(query.select, new_context, select_query_options);
|
2022-12-15 12:03:09 +00:00
|
|
|
pipeline = interpreter_select.buildQueryPipeline();
|
|
|
|
}
|
2020-07-07 09:24:49 +00:00
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
2023-01-21 11:19:09 +00:00
|
|
|
/// Passing 1 as subquery_depth will disable limiting size of intermediate result.
|
2022-12-15 12:03:09 +00:00
|
|
|
auto select_query_options = SelectQueryOptions(QueryProcessingStage::Complete, 1);
|
|
|
|
|
|
|
|
if (settings.allow_experimental_analyzer)
|
|
|
|
{
|
2023-01-18 09:44:40 +00:00
|
|
|
InterpreterSelectQueryAnalyzer interpreter_select_analyzer(query.select, getContext(), select_query_options);
|
2023-01-20 11:19:16 +00:00
|
|
|
pipeline = interpreter_select_analyzer.buildQueryPipeline();
|
2022-12-15 12:03:09 +00:00
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
2023-01-21 11:19:09 +00:00
|
|
|
InterpreterSelectWithUnionQuery interpreter_select(query.select, getContext(), select_query_options);
|
2022-12-15 12:03:09 +00:00
|
|
|
pipeline = interpreter_select.buildQueryPipeline();
|
|
|
|
}
|
2020-07-07 09:24:49 +00:00
|
|
|
}
|
2019-12-12 10:49:15 +00:00
|
|
|
|
2021-09-15 19:35:48 +00:00
|
|
|
pipeline.dropTotalsAndExtremes();
|
2021-05-06 17:30:24 +00:00
|
|
|
|
2020-03-19 16:51:09 +00:00
|
|
|
if (table->supportsParallelInsert() && settings.max_insert_threads > 1)
|
2022-04-18 08:18:31 +00:00
|
|
|
out_streams_size = std::min(static_cast<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 18:10:56 +00:00
|
|
|
|
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();
|
2021-04-23 09:50:56 +00:00
|
|
|
const auto & output_columns = metadata_snapshot->getColumns();
|
2021-04-22 18:10:56 +00:00
|
|
|
|
2021-04-23 19:08:13 +00:00
|
|
|
if (input_columns.size() == query_columns.size())
|
2021-04-22 18:10:56 +00:00
|
|
|
{
|
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
|
2022-05-09 19:13:02 +00:00
|
|
|
/// default column values (in AddingDefaultsTransform), so all values will be cast correctly.
|
2023-02-08 19:14:28 +00:00
|
|
|
if (isNullableOrLowCardinalityNullable(input_columns[col_idx].type) && !isNullableOrLowCardinalityNullable(query_columns[col_idx].type) && output_columns.has(query_columns[col_idx].name))
|
|
|
|
query_sample_block.setColumn(col_idx, ColumnWithTypeAndName(makeNullableOrLowCardinalityNullable(query_columns[col_idx].column), makeNullableOrLowCardinalityNullable(query_columns[col_idx].type), query_columns[col_idx].name));
|
2021-04-23 19:08:13 +00:00
|
|
|
}
|
2021-04-22 18:10:56 +00:00
|
|
|
}
|
|
|
|
}
|
2020-03-19 16:51:09 +00:00
|
|
|
}
|
2020-04-25 11:33:47 +00:00
|
|
|
else if (query.watch)
|
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
InterpreterWatchQuery interpreter_watch{ query.watch, getContext() };
|
2021-09-16 17:40:42 +00:00
|
|
|
pipeline = interpreter_watch.buildQueryPipeline();
|
2020-04-25 11:33:47 +00:00
|
|
|
}
|
2018-09-20 11:40:04 +00:00
|
|
|
|
2023-04-07 13:29:51 +00:00
|
|
|
ThreadGroupPtr running_group;
|
|
|
|
if (current_thread)
|
|
|
|
running_group = current_thread->getThreadGroup();
|
|
|
|
if (!running_group)
|
|
|
|
running_group = std::make_shared<ThreadGroup>(getContext());
|
2021-12-20 12:55:07 +00:00
|
|
|
for (size_t i = 0; i < out_streams_size; ++i)
|
2019-12-12 10:49:15 +00:00
|
|
|
{
|
2023-04-07 13:29:51 +00:00
|
|
|
auto out = buildChainImpl(table, metadata_snapshot, query_sample_block,
|
|
|
|
/* thread_status_holder= */ nullptr,
|
|
|
|
running_group,
|
|
|
|
/* elapsed_counter_ms= */ nullptr);
|
2021-09-01 18:41:50 +00:00
|
|
|
out_chains.emplace_back(std::move(out));
|
2020-03-19 16:51:09 +00:00
|
|
|
}
|
2019-12-12 10:49:15 +00:00
|
|
|
}
|
2014-03-19 10:45:13 +00:00
|
|
|
|
2021-09-19 18:53:36 +00:00
|
|
|
BlockIO res;
|
|
|
|
|
2020-04-25 11:33:47 +00:00
|
|
|
/// What type of query: INSERT or INSERT SELECT or INSERT WATCH?
|
2022-05-20 19:49:31 +00:00
|
|
|
if (distributed_pipeline)
|
2020-05-27 18:20:26 +00:00
|
|
|
{
|
2022-05-20 19:49:31 +00:00
|
|
|
res.pipeline = std::move(*distributed_pipeline);
|
2020-05-27 18:20:26 +00:00
|
|
|
}
|
|
|
|
else if (query.select || query.watch)
|
2012-03-11 08:52:56 +00:00
|
|
|
{
|
2021-09-01 18:41:50 +00:00
|
|
|
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
|
|
|
|
2021-12-08 15:29:00 +00:00
|
|
|
/// We need to convert Sparse columns to full, because it's destination storage
|
2022-04-28 18:51:13 +00:00
|
|
|
/// may not support it or may have different settings for applying Sparse serialization.
|
2021-09-29 14:37:07 +00:00
|
|
|
pipeline.addSimpleTransform([&](const Block & in_header) -> ProcessorPtr
|
2021-04-15 14:34:55 +00:00
|
|
|
{
|
|
|
|
return std::make_shared<MaterializingTransform>(in_header);
|
|
|
|
});
|
|
|
|
|
2021-10-06 18:17:22 +00:00
|
|
|
size_t num_select_threads = pipeline.getNumThreads();
|
2022-05-20 19:49:31 +00:00
|
|
|
|
|
|
|
for (auto & chain : out_chains)
|
|
|
|
resources = chain.detachResources();
|
|
|
|
|
2021-09-15 19:35:48 +00:00
|
|
|
pipeline.addChains(std::move(out_chains));
|
2016-08-25 12:38:47 +00:00
|
|
|
|
2022-07-02 17:02:21 +00:00
|
|
|
if (!settings.parallel_view_processing)
|
|
|
|
{
|
|
|
|
/// Don't use more threads for INSERT than for SELECT to reduce memory consumption.
|
|
|
|
if (pipeline.getNumThreads() > num_select_threads)
|
|
|
|
pipeline.setMaxThreads(num_select_threads);
|
|
|
|
}
|
|
|
|
else if (pipeline.getNumThreads() < settings.max_threads)
|
|
|
|
{
|
|
|
|
/// It is possible for query to have max_threads=1, due to optimize_trivial_insert_select,
|
|
|
|
/// however in case of parallel_view_processing and multiple views, views can still be processed in parallel.
|
|
|
|
///
|
|
|
|
/// Note, number of threads will be limited by buildPushingToViewsChain() to max_threads.
|
|
|
|
pipeline.setMaxThreads(settings.max_threads);
|
|
|
|
}
|
2021-10-06 18:17:22 +00:00
|
|
|
|
2021-09-15 19:35:48 +00:00
|
|
|
pipeline.setSinks([&](const Block & cur_header, QueryPipelineBuilder::StreamType) -> ProcessorPtr
|
2021-09-01 18:41:50 +00:00
|
|
|
{
|
2021-09-26 14:54:59 +00:00
|
|
|
return std::make_shared<EmptySink>(cur_header);
|
2020-05-27 18:20:26 +00:00
|
|
|
});
|
2018-02-19 00:45:32 +00:00
|
|
|
|
|
|
|
if (!allow_materialized)
|
|
|
|
{
|
2020-06-17 16:39:58 +00:00
|
|
|
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))
|
2023-01-23 21:13:58 +00:00
|
|
|
throw Exception(ErrorCodes::ILLEGAL_COLUMN, "Cannot insert column {}, because it is MATERIALIZED column.", column.name);
|
2018-02-19 00:45:32 +00:00
|
|
|
}
|
2021-09-15 19:35:48 +00:00
|
|
|
|
2022-05-24 20:06:08 +00:00
|
|
|
res.pipeline = QueryPipelineBuilder::getPipeline(std::move(pipeline));
|
2019-02-07 13:18:04 +00:00
|
|
|
}
|
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
|
2023-02-22 21:52:49 +00:00
|
|
|
auto format = getInputFormatFromASTInsertQuery(query_ptr, true, query_sample_block, getContext(), nullptr);
|
|
|
|
for (auto && buffer : owned_buffers)
|
|
|
|
format->addBuffer(std::move(buffer));
|
|
|
|
|
|
|
|
auto pipe = getSourceFromInputFormat(query_ptr, std::move(format), getContext(), nullptr);
|
2021-09-15 19:35:48 +00:00
|
|
|
res.pipeline.complete(std::move(pipe));
|
|
|
|
}
|
2020-10-14 19:25:31 +00:00
|
|
|
}
|
2014-03-20 10:59:45 +00:00
|
|
|
|
2022-05-20 19:49:31 +00:00
|
|
|
res.pipeline.addResources(std::move(resources));
|
|
|
|
|
2021-12-11 07:57:23 +00:00
|
|
|
res.pipeline.addStorageHolder(table);
|
2022-01-04 10:27:53 +00:00
|
|
|
if (inner_table)
|
|
|
|
res.pipeline.addStorageHolder(inner_table);
|
2021-12-11 07:57:23 +00:00
|
|
|
|
2014-03-20 10:59:45 +00:00
|
|
|
return res;
|
2012-03-11 08:52:56 +00:00
|
|
|
}
|
|
|
|
|
2018-02-19 00:45:32 +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
|
|
|
}
|
|
|
|
|
2021-12-28 14:47:19 +00:00
|
|
|
void InterpreterInsertQuery::extendQueryLogElemImpl(QueryLogElement & elem, ContextPtr context_)
|
2020-12-14 03:30:39 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
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());
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-12-28 14:47:19 +00:00
|
|
|
void InterpreterInsertQuery::extendQueryLogElemImpl(QueryLogElement & elem, const ASTPtr &, ContextPtr context_) const
|
|
|
|
{
|
|
|
|
extendQueryLogElemImpl(elem, context_);
|
|
|
|
}
|
|
|
|
|
2011-10-30 11:30:52 +00:00
|
|
|
}
|