Method transformPipeline

src/QueryPlan/TableWriteStep.cpp:139–209

}

void TableWriteStep::transformPipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings)
{
    switch (target->getTargetType())
    {
        case TargetType::INSERT: {
            auto * insert_target = dynamic_cast<TableWriteStep::InsertTarget *>(target.get());
            auto target_storage = DatabaseCatalog::instance().getTable(insert_target->getStorageID(), settings.context);
            
            auto insert_target_header = getHeader(insert_target->getColumns());
            auto out_streams = createOutputStream(
                target_storage, settings, insert_target_header, settings.context->getSettingsRef().max_threads, false, insert_target->getQuery());

            if (out_streams.empty())
                throw Exception("No output stream when transform TableWriteStep", ErrorCodes::LOGICAL_ERROR);

            const auto & header = out_streams[0]->getHeader();
            auto actions_dag = ActionsDAG::makeConvertingActions(
                pipeline.getHeader().getColumnsWithTypeAndName(),
                header.getColumnsWithTypeAndName(),
                ActionsDAG::MatchColumnsMode::Position);
            auto actions = std::make_shared<ExpressionActions>(
                actions_dag, ExpressionActionsSettings::fromContext(settings.context, CompileExpressions::yes));

            pipeline.addSimpleTransform(
                [&](const Block & in_header) -> ProcessorPtr { return std::make_shared<ExpressionTransform>(in_header, actions); });

            size_t min_insert_block_size_rows = settings.context->getSettingsRef().min_insert_block_size_rows;
            size_t min_insert_block_size_bytes = settings.context->getSettingsRef().min_insert_block_size_bytes;
            /// It's important to squash blocks as early as possible (before other transforms),
            ///  because other transforms may work inefficient if block size is small.

            /// Do not squash blocks if it is a sync INSERT into Distributed, since it lead to double bufferization on client and server side.
            /// Client-side bufferization might cause excessive timeouts (especially in case of big blocks).
            if (!(settings.context->getSettingsRef().insert_distributed_sync && target_storage->isRemote()) && settings.context->getSettingsRef().enable_insert_squashing)
            {
                pipeline.addSimpleTransform(
                    [&](const Block & current_header) -> ProcessorPtr {
                        return std::make_shared<SimpleSquashingChunksTransform>(current_header, min_insert_block_size_rows, min_insert_block_size_bytes);}
                );
                LOG_INFO(&Poco::Logger::get("TableWriteStep"), fmt::format("squash min insert block size rows:{}, min insert block size bytes:{}", min_insert_block_size_rows, min_insert_block_size_bytes));
            }
            //LOG_DEBUG(&Poco::Logger::get("TableWriteStep"), fmt::format("output header: {}", stream->getHeader().dumpStructure()));
            pipeline.resize(out_streams.size());
            LOG_INFO(&Poco::Logger::get("TableWriteStep"), fmt::format("pipeline size: {}, out streams size {}", pipeline.getNumStreams(), out_streams.size()));

            if (insert_select_with_profiles)
            {
                pipeline.addSimpleTransform([&](const Block &, QueryPipeline::StreamType type) -> ProcessorPtr
                {
                    if (type != QueryPipeline::StreamType::Main)
                        return nullptr;

                    auto stream = std::move(out_streams.back());
                    out_streams.pop_back();

                    return std::make_shared<ProcessorToOutputStream>(std::move(stream));
                });
            }