Method buildInsertPipeline

src/Interpreters/InterpreterInsertQuery.cpp:744–852

QueryPipeline InterpreterInsertQuery::buildInsertPipeline(ASTInsertQuery & query, StoragePtr table)
{
    auto context = getContext();

    // disable parallel replicas for inserts if enabled
    // the insert can trigger update for dependent materialized views
    // using parallel replicas in this context is unnecessary
    if (context->canUseParallelReplicasOnInitiator())
    {
        auto mutable_context = Context::createCopy(context);
        mutable_context->setSetting("enable_parallel_replicas", Field{0});
        context = mutable_context;
    }

    const Settings & settings = context->getSettingsRef();
    auto metadata_snapshot = table->getInMemoryMetadataPtr(context, false);
    auto query_sample_block
        = std::make_shared<const Block>(getSampleBlock(query, table, metadata_snapshot, context, no_destination, allow_materialized));
    if (query_sample_block->empty())
        throw Exception(ErrorCodes::EMPTY_LIST_OF_COLUMNS_PASSED, "Empty list of columns to insert");

    // when insert is initiated from FileLog or similar storages
    // they are allowed to expose its virtuals columns to the dependent views
    auto insert_dependencies = InsertDependenciesBuilder::create(
        table,
        query_ptr,
        query_sample_block,
        async_insert,
        /*skip_destination_table*/ no_destination,
        /*max_insert_threads*/ 1,
        context);

    auto chains = insert_dependencies->createChainWithDependenciesForAllStreams();
    chassert(chains.size() == 1);
    auto chain = std::move(chains.front());
    bool squash_with_strict_limits = settings[Setting::use_strict_insert_block_limits] && !async_insert;

    if (squash_with_strict_limits)
    {
        chain.addSource(
            std::make_shared<AddDeduplicationInfoTransform>(
                insert_dependencies,
                insert_dependencies->getRootViewID(),
                settings[Setting::insert_deduplication_token].value,
                context->getServerSettings()[ServerSetting::insert_deduplication_version].value,
                chain.getInputSharedHeader())
        );
    }

    if (shouldAddSquashingForStorage(table, context) && !no_squash)
    {
        auto applying = std::make_shared<ApplySquashingTransform>(chain.getInputSharedHeader());
        chain.addSource(std::move(applying));
    }

    if (shouldAddSquashingForStorage(table, context) && !no_squash)
    {
        bool table_prefers_large_blocks = table->prefersLargeBlocks();