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 | |
25 | namespace DB |
26 | { |
27 | |
28 | namespace 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 | |
37 | InterpreterInsertQuery::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 | |
49 | StoragePtr 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 | |
63 | Block 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 | |
97 | BlockIO 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 = 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 | |
175 | void 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 | |
188 | std::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 | |