ClickHouse/dbms/src/Interpreters/InterpreterInsertQuery.cpp

157 lines
5.5 KiB
C++
Raw Normal View History

#include <IO/ConcatReadBuffer.h>
2017-07-13 20:58:19 +00:00
#include <Common/typeid_cast.h>
#include <DataStreams/AddingDefaultBlockOutputStream.h>
#include <DataStreams/CountingBlockOutputStream.h>
#include <DataStreams/ConvertingBlockInputStream.h>
#include <DataStreams/NullAndDoCopyBlockInputStream.h>
#include <DataStreams/PushingToViewsBlockOutputStream.h>
#include <DataStreams/SquashingBlockOutputStream.h>
#include <DataStreams/copyData.h>
#include <Parsers/ASTIdentifier.h>
#include <Parsers/ASTInsertQuery.h>
2018-02-25 06:34:20 +00:00
#include <Parsers/ASTSelectWithUnionQuery.h>
#include <Interpreters/InterpreterInsertQuery.h>
2018-02-25 06:34:20 +00:00
#include <Interpreters/InterpreterSelectWithUnionQuery.h>
2011-10-30 11:30:52 +00:00
#include <TableFunctions/TableFunctionFactory.h>
#include <Parsers/ASTFunction.h>
2011-10-30 11:30:52 +00:00
namespace DB
{
namespace ErrorCodes
{
extern const int NO_SUCH_COLUMN_IN_TABLE;
extern const int READONLY;
extern const int ILLEGAL_COLUMN;
}
2011-10-30 11:30:52 +00:00
2018-01-12 13:03:19 +00:00
InterpreterInsertQuery::InterpreterInsertQuery(
const ASTPtr & query_ptr_, const Context & context_, bool allow_materialized_)
: query_ptr(query_ptr_), context(context_), allow_materialized(allow_materialized_)
2011-10-30 11:30:52 +00:00
{
}
StoragePtr InterpreterInsertQuery::getTable(const ASTInsertQuery & query)
2011-10-30 11:30:52 +00:00
{
if (query.table_function)
{
auto table_function = typeid_cast<const ASTFunction *>(query.table_function.get());
const auto & factory = TableFunctionFactory::instance();
return factory.get(table_function->name, context)->execute(query.table_function, context);
}
/// Into what table to write.
return context.getTable(query.database, query.table);
2011-10-30 11:30:52 +00:00
}
Block InterpreterInsertQuery::getSampleBlock(const ASTInsertQuery & query, const StoragePtr & table)
{
Block table_sample_non_materialized = table->getSampleBlockNonMaterialized();
2013-10-25 14:56:47 +00:00
2017-04-02 17:37:49 +00:00
/// If the query does not include information about columns
if (!query.columns)
return table_sample_non_materialized;
Block table_sample = table->getSampleBlock();
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
Block res;
for (const auto & identifier : query.columns->children)
{
std::string current_name = identifier->getColumnName();
2013-10-25 14:56:47 +00:00
2017-04-02 17:37:49 +00:00
/// The table does not have a column with that name
if (!table_sample.has(current_name))
throw Exception("No such column " + current_name + " in table " + query.table, 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);
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
BlockIO InterpreterInsertQuery::execute()
2012-03-11 08:52:56 +00:00
{
ASTInsertQuery & query = typeid_cast<ASTInsertQuery &>(*query_ptr);
checkAccess(query);
StoragePtr table = getTable(query);
2012-03-11 08:52:56 +00:00
auto table_lock = table->lockStructure(true, __PRETTY_FUNCTION__);
2018-03-13 15:00:28 +00:00
NamesAndTypesList required_columns = table->getColumns().getAllPhysical();
2017-04-02 17:37:49 +00:00
/// We create a pipeline of several streams, into which we will write data.
BlockOutputStreamPtr out;
2016-07-07 01:57:48 +00:00
out = std::make_shared<PushingToViewsBlockOutputStream>(query.database, query.table, table, context, query_ptr, query.no_destination);
2016-07-07 01:57:48 +00:00
out = std::make_shared<AddingDefaultBlockOutputStream>(
out, getSampleBlock(query, table), required_columns, table->getColumns().defaults, context);
2016-07-07 01:57:48 +00:00
2018-03-12 19:02:54 +00:00
/// 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 (!(context.getSettingsRef().insert_distributed_sync && table->getName() == "Distributed"))
{
out = std::make_shared<SquashingBlockOutputStream>(
out, context.getSettingsRef().min_insert_block_size_rows, context.getSettingsRef().min_insert_block_size_bytes);
}
2012-03-11 08:52:56 +00:00
auto out_wrapper = std::make_shared<CountingBlockOutputStream>(out);
out_wrapper->setProcessListElement(context.getProcessListElement());
out = std::move(out_wrapper);
BlockIO res;
res.out = std::move(out);
2017-04-02 17:37:49 +00:00
/// What type of query: INSERT or INSERT SELECT?
if (query.select)
{
/// Passing 1 as subquery_depth will disable limiting size of intermediate result.
2018-02-26 09:05:06 +00:00
InterpreterSelectWithUnionQuery interpreter_select{query.select, context, {}, QueryProcessingStage::Complete, 1};
res.in = interpreter_select.execute().in;
2018-02-23 01:00:47 +00:00
res.in = std::make_shared<ConvertingBlockInputStream>(context, res.in, res.out->getHeader(), ConvertingBlockInputStream::MatchColumnsMode::Position);
res.in = std::make_shared<NullAndDoCopyBlockInputStream>(res.in, res.out);
res.out = nullptr;
if (!allow_materialized)
{
Block in_header = res.in->getHeader();
for (const auto & name_type : table->getColumns().materialized)
if (in_header.has(name_type.name))
throw Exception("Cannot insert column " + name_type.name + ", because it is MATERIALIZED column.", ErrorCodes::ILLEGAL_COLUMN);
}
}
return res;
2012-03-11 08:52:56 +00:00
}
void InterpreterInsertQuery::checkAccess(const ASTInsertQuery & query)
{
const Settings & settings = context.getSettingsRef();
auto readonly = settings.readonly;
2012-03-11 08:52:56 +00:00
if (!readonly || (query.database.empty() && context.tryGetExternalTable(query.table) && readonly >= 2))
{
return;
}
throw Exception("Cannot insert into table in readonly mode", ErrorCodes::READONLY);
}
2011-10-30 11:30:52 +00:00
}