#include #include #include #include #include #include #include #include #include namespace DB { InterpreterInsertQuery::InterpreterInsertQuery(ASTPtr query_ptr_, Context & context_) : query_ptr(query_ptr_), context(context_) { } StoragePtr InterpreterInsertQuery::getTable() { ASTInsertQuery & query = dynamic_cast(*query_ptr); /// В какую таблицу писать. return context.getTable(query.database, query.table); } Block InterpreterInsertQuery::getSampleBlock() { ASTInsertQuery & query = dynamic_cast(*query_ptr); Block dbSample = getTable()->getSampleBlock(); /// Если в запросе не указана информация о столбцах if (!query.columns) return dbSample; /// Строим мап из имени столбца в его тип const NamesAndTypesList & names_and_types = dbSample.getColumnsList(); std::map nameToType(names_and_types.begin(), names_and_types.end()); /// Формируем блок, основываясь на именах столбцов из запроса Block res; for (ASTs::iterator it = query.columns->children.begin(); it != query.columns->children.end(); it ++) { std::string currentName = (*it)->getColumnName(); /// В таблице нет столбца с таким именем if (nameToType.count(currentName) == 0) throw Exception("No such column in table: " + currentName, ErrorCodes::NO_SUCH_COLUMN_IN_TABLE); ColumnWithNameAndType col; col.name = currentName; col.type = nameToType[col.name]; col.column = col.type->createColumn(); res.insert(col); } return res; } void InterpreterInsertQuery::execute(ReadBuffer * remaining_data_istr) { ASTInsertQuery & query = dynamic_cast(*query_ptr); StoragePtr table = getTable(); BlockInputStreamPtr in; NamesAndTypesListPtr required_columns = new NamesAndTypesList; *required_columns = table->getSampleBlock().getColumnsList(); BlockOutputStreamPtr out = new AddingDefaultBlockOutputStream(table->write(query_ptr), required_columns); /// Какой тип запроса: INSERT VALUES | INSERT FORMAT | INSERT SELECT? if (!query.select) { String format = query.format; if (format.empty()) format = "Values"; /// Данные могут содержаться в распарсенной (query.data) и ещё не распарсенной (remaining_data_istr) части запроса. /// Если данных нет. bool has_remaining_data = remaining_data_istr && !remaining_data_istr->eof(); if (!query.data && !has_remaining_data) throw Exception("No data to insert", ErrorCodes::NO_DATA_TO_INSERT); ConcatReadBuffer::ReadBuffers buffers; ReadBuffer buf1(const_cast(query.data), query.data ? query.end - query.data : 0, 0); if (query.data) buffers.push_back(&buf1); if (has_remaining_data) buffers.push_back(remaining_data_istr); ConcatReadBuffer istr(buffers); Block sample = table->getSampleBlock(); in = context.getFormatFactory().getInput(format, istr, sample, context.getSettings().max_block_size, context.getDataTypeFactory()); copyData(*in, *out); } else { InterpreterSelectQuery interpreter_select(query.select, context); in = interpreter_select.execute(); in = new MaterializingBlockInputStream(in); copyData(*in, *out); } } BlockOutputStreamPtr InterpreterInsertQuery::execute() { ASTInsertQuery & query = dynamic_cast(*query_ptr); StoragePtr table = getTable(); NamesAndTypesListPtr required_columns = new NamesAndTypesList; *required_columns = table->getSampleBlock().getColumnsList(); BlockOutputStreamPtr out = new AddingDefaultBlockOutputStream(table->write(query_ptr), required_columns); /// Какой тип запроса: INSERT или INSERT SELECT? if (!query.select) return out; else { InterpreterSelectQuery interpreter_select(query.select, context); BlockInputStreamPtr in = interpreter_select.execute(); in = new MaterializingBlockInputStream(in); copyData(*in, *out); return NULL; } } }