ClickHouse/dbms/src/Interpreters/InterpreterInsertQuery.cpp

110 lines
3.2 KiB
C++
Raw Normal View History

2011-10-30 11:30:52 +00:00
#include <DB/IO/ConcatReadBuffer.h>
#include <DB/DataStreams/ProhibitColumnsBlockOutputStream.h>
#include <DB/DataStreams/MaterializingBlockOutputStream.h>
2013-10-25 14:56:47 +00:00
#include <DB/DataStreams/AddingDefaultBlockOutputStream.h>
#include <DB/DataStreams/PushingToViewsBlockOutputStream.h>
#include <DB/DataStreams/NullAndDoCopyBlockInputStream.h>
2015-04-16 06:12:35 +00:00
#include <DB/DataStreams/FormatFactory.h>
2011-10-30 11:30:52 +00:00
#include <DB/DataStreams/copyData.h>
#include <DB/Parsers/ASTInsertQuery.h>
#include <DB/Parsers/ASTSelectQuery.h>
#include <DB/Parsers/ASTIdentifier.h>
#include <DB/Interpreters/InterpreterSelectQuery.h>
#include <DB/Interpreters/InterpreterInsertQuery.h>
namespace DB
{
2012-03-05 00:09:41 +00:00
InterpreterInsertQuery::InterpreterInsertQuery(ASTPtr query_ptr_, Context & context_)
: query_ptr(query_ptr_), context(context_)
2011-10-30 11:30:52 +00:00
{
ProfileEvents::increment(ProfileEvents::InsertQuery);
2011-10-30 11:30:52 +00:00
}
StoragePtr InterpreterInsertQuery::getTable()
{
ASTInsertQuery & query = typeid_cast<ASTInsertQuery &>(*query_ptr);
2011-10-30 11:30:52 +00:00
/// В какую таблицу писать.
2012-08-02 17:33:31 +00:00
return context.getTable(query.database, query.table);
2011-10-30 11:30:52 +00:00
}
2012-03-19 12:57:56 +00:00
Block InterpreterInsertQuery::getSampleBlock()
{
ASTInsertQuery & query = typeid_cast<ASTInsertQuery &>(*query_ptr);
2013-10-25 14:56:47 +00:00
/// Если в запросе не указана информация о столбцах
if (!query.columns)
return getTable()->getSampleBlockNonMaterialized();
Block table_sample = getTable()->getSampleBlock();
2013-10-25 14:56:47 +00:00
/// Формируем блок, основываясь на именах столбцов из запроса
Block res;
for (const auto & identifier : query.columns->children)
2013-10-25 14:56:47 +00:00
{
std::string current_name = identifier->getColumnName();
2013-10-25 14:56:47 +00:00
/// В таблице нет столбца с таким именем
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
2013-10-25 14:56:47 +00:00
ColumnWithNameAndType col;
col.name = current_name;
col.type = table_sample.getByName(current_name).type;
2013-10-25 14:56:47 +00:00
col.column = col.type->createColumn();
res.insert(col);
}
2013-10-25 14:56:47 +00:00
return res;
}
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);
2012-03-11 08:52:56 +00:00
StoragePtr table = getTable();
auto table_lock = table->lockStructure(true);
NamesAndTypesListPtr required_columns = new NamesAndTypesList(table->getColumnsList());
/// Создаем кортеж из нескольких стримов, в которые будем писать данные.
BlockOutputStreamPtr out{
new ProhibitColumnsBlockOutputStream{
new AddingDefaultBlockOutputStream{
new MaterializingBlockOutputStream{
new PushingToViewsBlockOutputStream{query.database, query.table, context, query_ptr}
},
required_columns, table->column_defaults, context, context.getSettingsRef().strict_insert_defaults
},
table->materialized_columns
}
};
2012-03-11 08:52:56 +00:00
BlockIO res;
res.out_sample = getSampleBlock();
2012-03-11 08:52:56 +00:00
/// Какой тип запроса: INSERT или INSERT SELECT?
if (!query.select)
{
res.out = out;
}
2012-03-11 08:52:56 +00:00
else
{
2014-11-12 10:37:47 +00:00
InterpreterSelectQuery interpreter_select{query.select, context};
2015-06-18 02:11:05 +00:00
BlockInputStreamPtr in{interpreter_select.execute().in};
2014-11-12 10:37:47 +00:00
res.in = new NullAndDoCopyBlockInputStream{in, out};
2012-03-11 08:52:56 +00:00
}
return res;
2012-03-11 08:52:56 +00:00
}
2011-10-30 11:30:52 +00:00
}