| 993 | |
| 994 | |
| 995 | BlockIO InterpreterInsertQuery::execute() |
| 996 | { |
| 997 | auto context = getContext(); |
| 998 | const Settings & settings = context->getSettingsRef(); |
| 999 | auto & query = query_ptr->as<ASTInsertQuery &>(); |
| 1000 | |
| 1001 | StoragePtr table = getTable(query); |
| 1002 | setInsertContextValues(context, query, table); |
| 1003 | if (context->getServerSettings()[ServerSetting::disable_insertion_and_mutation] |
| 1004 | && query.table_id.database_name != DatabaseCatalog::SYSTEM_DATABASE |
| 1005 | && query.table_id.database_name != DatabaseCatalog::TEMPORARY_DATABASE) |
| 1006 | { |
| 1007 | /// Allow inserts into external table engines (object storage, message queues, external databases) |
| 1008 | /// as they don't create merge tasks on the server replica |
| 1009 | bool is_external_storage = |
| 1010 | table->isObjectStorage() || /// S3, Azure, GCS, HDFS, etc. |
| 1011 | table->isDataLake() || /// Iceberg, DeltaLake, Hudi |
| 1012 | table->isMessageQueue() || /// Kafka, RabbitMQ, NATS |
| 1013 | table->isExternalDatabase(); /// MySQL, PostgreSQL, MongoDB, Hive, YTsaurus |
| 1014 | |
| 1015 | if (!is_external_storage) |
| 1016 | throw Exception(ErrorCodes::QUERY_IS_PROHIBITED, "Insert queries are prohibited"); |
| 1017 | } |
| 1018 | |
| 1019 | if (context->getMessageQueueDisableInsertion() |
| 1020 | && table->isMessageQueue() |
| 1021 | && no_destination) |
| 1022 | { |
| 1023 | throw Exception(ErrorCodes::QUERY_IS_PROHIBITED, "Message queue insertion is disabled"); |
| 1024 | } |
| 1025 | |
| 1026 | checkStorageSupportsTransactionsIfNeeded(table, getContext()); |
| 1027 | |
| 1028 | if (query.partition_by && !table->supportsPartitionBy()) |
| 1029 | throw Exception(ErrorCodes::NOT_IMPLEMENTED, "PARTITION BY clause is not supported by storage"); |
| 1030 | |
| 1031 | auto table_lock = table->lockForShare(context->getInitialQueryId(), settings[Setting::lock_acquire_timeout]); |
| 1032 | |
| 1033 | table->updateExternalDynamicMetadataIfExists(context); |
| 1034 | auto metadata_snapshot = table->getInMemoryMetadataPtr(context, false); |
| 1035 | auto query_sample_block = getSampleBlock(query, table, metadata_snapshot, context, no_destination, allow_materialized); |
| 1036 | /// For table functions we check access while executing |
| 1037 | /// getTable() -> ITableFunction::execute(). |
| 1038 | if (!query.table_function) |
| 1039 | context->checkAccess(AccessType::INSERT, query.table_id, query_sample_block.getNames()); |
| 1040 | |
| 1041 | if (!allow_materialized) |
| 1042 | { |
| 1043 | for (const auto & column : metadata_snapshot->getColumns()) |
| 1044 | if (column.default_desc.kind == ColumnDefaultKind::Materialized && query_sample_block.has(column.name)) |
| 1045 | throw Exception(ErrorCodes::ILLEGAL_COLUMN, "Cannot insert column {}, because it is MATERIALIZED column.", column.name); |
| 1046 | } |
| 1047 | |
| 1048 | BlockIO res; |
| 1049 | if (query.select) |
| 1050 | { |
| 1051 | if (settings[Setting::parallel_distributed_insert_select]) |
| 1052 | { |
no test coverage detected