1#include <Interpreters/InterpreterInsertQuery.h>
2
3#include <DataStreams/AddingDefaultBlockOutputStream.h>
4#include <DataStreams/AddingDefaultsBlockInputStream.h>
5#include <DataStreams/CheckConstraintsBlockOutputStream.h>
6#include <DataStreams/ConvertingBlockInputStream.h>
7#include <DataStreams/CountingBlockOutputStream.h>
8#include <DataStreams/InputStreamFromASTInsertQuery.h>
9#include <DataStreams/NullAndDoCopyBlockInputStream.h>
10#include <DataStreams/OwningBlockInputStream.h>
11#include <DataStreams/PushingToViewsBlockOutputStream.h>
12#include <DataStreams/SquashingBlockOutputStream.h>
13#include <DataStreams/copyData.h>
14#include <IO/ConcatReadBuffer.h>
15#include <IO/ReadBufferFromMemory.h>
16#include <Interpreters/InterpreterSelectWithUnionQuery.h>
17#include <Parsers/ASTFunction.h>
18#include <Parsers/ASTInsertQuery.h>
19#include <Parsers/ASTSelectWithUnionQuery.h>
20#include <Storages/Kafka/StorageKafka.h>
21#include <TableFunctions/TableFunctionFactory.h>
22#include <Common/checkStackSize.h>
23
24
25namespace DB
26{
27
28namespace ErrorCodes
29{
30 extern const int NO_SUCH_COLUMN_IN_TABLE;
31 extern const int READONLY;
32 extern const int ILLEGAL_COLUMN;
33 extern const int DUPLICATE_COLUMN;
34}
35
36
37InterpreterInsertQuery::InterpreterInsertQuery(
38 const ASTPtr & query_ptr_, const Context & context_, bool allow_materialized_, bool no_squash_, bool no_destination_)
39 : query_ptr(query_ptr_)
40 , context(context_)
41 , allow_materialized(allow_materialized_)
42 , no_squash(no_squash_)
43 , no_destination(no_destination_)
44{
45 checkStackSize();
46}
47
48
49StoragePtr InterpreterInsertQuery::getTable(const ASTInsertQuery & query)
50{
51 if (query.table_function)
52 {
53 const auto * table_function = query.table_function->as<ASTFunction>();
54 const auto & factory = TableFunctionFactory::instance();
55 TableFunctionPtr table_function_ptr = factory.get(table_function->name, context);
56 return table_function_ptr->execute(query.table_function, context, table_function_ptr->getName());
57 }
58
59 /// Into what table to write.
60 return context.getTable(query.database, query.table);
61}
62
63Block InterpreterInsertQuery::getSampleBlock(const ASTInsertQuery & query, const StoragePtr & table)
64{
65 Block table_sample_non_materialized = table->getSampleBlockNonMaterialized();
66 /// If the query does not include information about columns
67 if (!query.columns)
68 {
69 if (no_destination)
70 return table->getSampleBlockWithVirtuals();
71 else
72 return table_sample_non_materialized;
73 }
74
75 Block table_sample = table->getSampleBlock();
76 /// Form the block based on the column names from the query
77 Block res;
78 for (const auto & identifier : query.columns->children)
79 {
80 std::string current_name = identifier->getColumnName();
81
82 /// The table does not have a column with that name
83 if (!table_sample.has(current_name))
84 throw Exception("No such column " + current_name + " in table " + query.table, ErrorCodes::NO_SUCH_COLUMN_IN_TABLE);
85
86 if (!allow_materialized && !table_sample_non_materialized.has(current_name))
87 throw Exception("Cannot insert column " + current_name + ", because it is MATERIALIZED column.", ErrorCodes::ILLEGAL_COLUMN);
88 if (res.has(current_name))
89 throw Exception("Column " + current_name + " specified more than once", ErrorCodes::DUPLICATE_COLUMN);
90
91 res.insert(ColumnWithTypeAndName(table_sample.getByName(current_name).type, current_name));
92 }
93 return res;
94}
95
96
97BlockIO InterpreterInsertQuery::execute()
98{
99 const auto & query = query_ptr->as<ASTInsertQuery &>();
100 checkAccess(query);
101
102 StoragePtr table = getTable(query);
103
104 auto table_lock = table->lockStructureForShare(true, context.getInitialQueryId());
105
106 /// We create a pipeline of several streams, into which we will write data.
107 BlockOutputStreamPtr out;
108
109 /// NOTE: we explicitly ignore bound materialized views when inserting into Kafka Storage.
110 /// Otherwise we'll get duplicates when MV reads same rows again from Kafka.
111 if (table->noPushingToViews() && !no_destination)
112 out = table->write(query_ptr, context);
113 else
114 out = std::make_shared<PushingToViewsBlockOutputStream>(query.database, query.table, table, context, query_ptr, no_destination);
115
116 /// Do not squash blocks if it is a sync INSERT into Distributed, since it lead to double bufferization on client and server side.
117 /// Client-side bufferization might cause excessive timeouts (especially in case of big blocks).
118 if (!(context.getSettingsRef().insert_distributed_sync && table->isRemote()) && !no_squash)
119 {
120 out = std::make_shared<SquashingBlockOutputStream>(
121 out, out->getHeader(), context.getSettingsRef().min_insert_block_size_rows, context.getSettingsRef().min_insert_block_size_bytes);
122 }
123 auto query_sample_block = getSampleBlock(query, table);
124
125 /// Actually we don't know structure of input blocks from query/table,
126 /// because some clients break insertion protocol (columns != header)
127 out = std::make_shared<AddingDefaultBlockOutputStream>(
128 out, query_sample_block, out->getHeader(), table->getColumns().getDefaults(), context);
129
130 if (const auto & constraints = table->getConstraints(); !constraints.empty())
131 out = std::make_shared<CheckConstraintsBlockOutputStream>(query.table,
132 out, query_sample_block, table->getConstraints(), context);
133
134 auto out_wrapper = std::make_shared<CountingBlockOutputStream>(out);
135 out_wrapper->setProcessListElement(context.getProcessListElement());
136 out = std::move(out_wrapper);
137
138 BlockIO res;
139
140 /// What type of query: INSERT or INSERT SELECT?
141 if (query.select)
142 {
143 /// Passing 1 as subquery_depth will disable limiting size of intermediate result.
144 InterpreterSelectWithUnionQuery interpreter_select{query.select, context, SelectQueryOptions(QueryProcessingStage::Complete, 1)};
145
146 /// BlockIO may hold StoragePtrs to temporary tables
147 res = interpreter_select.execute();
148 res.out = nullptr;
149
150 res.in = std::make_shared<ConvertingBlockInputStream>(context, res.in, out->getHeader(), ConvertingBlockInputStream::MatchColumnsMode::Position);
151 res.in = std::make_shared<NullAndDoCopyBlockInputStream>(res.in, out);
152
153 if (!allow_materialized)
154 {
155 Block in_header = res.in->getHeader();
156 for (const auto & column : table->getColumns())
157 if (column.default_desc.kind == ColumnDefaultKind::Materialized && in_header.has(column.name))
158 throw Exception("Cannot insert column " + column.name + ", because it is MATERIALIZED column.", ErrorCodes::ILLEGAL_COLUMN);
159 }
160 }
161 else if (query.data && !query.has_tail) /// can execute without additional data
162 {
163 res.in = std::make_shared<InputStreamFromASTInsertQuery>(query_ptr, nullptr, query_sample_block, context, nullptr);
164 res.in = std::make_shared<NullAndDoCopyBlockInputStream>(res.in, out);
165 }
166 else
167 res.out = std::move(out);
168
169 res.pipeline.addStorageHolder(table);
170
171 return res;
172}
173
174
175void InterpreterInsertQuery::checkAccess(const ASTInsertQuery & query)
176{
177 const Settings & settings = context.getSettingsRef();
178 auto readonly = settings.readonly;
179
180 if (!readonly || (query.database.empty() && context.tryGetExternalTable(query.table) && readonly >= 2))
181 {
182 return;
183 }
184
185 throw Exception("Cannot insert into table in readonly mode", ErrorCodes::READONLY);
186}
187
188std::pair<String, String> InterpreterInsertQuery::getDatabaseTable() const
189{
190 const auto & query = query_ptr->as<ASTInsertQuery &>();
191 return {query.database, query.table};
192}
193
194}
195