Method initializePipeline

src/QueryPlan/TableScanStep.cpp:1149–1651

}

void TableScanStep::initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & build_context)
{
    if (is_null_source)
    {
        LOG_DEBUG(log, "Create NullSource from TableScanStep without storage");
        pipeline.init(Pipe(std::make_shared<NullSource>(output_stream->header)));
        return;
    }
    auto * query = query_info.query->as<ASTSelectQuery>();
    bool use_expand_pipe = build_context.is_expand;
    if (!build_context.is_expand && (query->getWhere() || query->getPrewhere() || query_info.partition_filter)
        && build_context.context->getSettingsRef().enable_runtime_filter_pipeline_poll)
    {
        std::vector<RuntimeFilterId> ids;
        if (query->getWhere())
        {
            auto where_ids = RuntimeFilterUtils::extractRuntimeFilterId(query->getWhere());
            ids.insert(ids.end(), where_ids.begin(), where_ids.end());
        }
        if (query->getPrewhere())
        {
            auto prewhere_ids = RuntimeFilterUtils::extractRuntimeFilterId(query->getPrewhere());
            ids.insert(ids.end(), prewhere_ids.begin(), prewhere_ids.end());
        }

        if (query_info.partition_filter)
        {
            auto prewhere_ids = RuntimeFilterUtils::extractRuntimeFilterId(query_info.partition_filter);
            ids.insert(ids.end(), prewhere_ids.begin(), prewhere_ids.end());
        }

        if (!ids.empty())
        {
            Pipe pipe(std::make_shared<MergeTreeSelectPrepareProcessor>(
                *this,
                build_context,
                table_output_stream.header,
                std::move(ids),
                build_context.context->getSettingsRef().wait_runtime_filter_timeout));
            pipeline.init(std::move(pipe));
            pipeline.addTransform(std::make_shared<ResizeProcessor>(
                table_output_stream.header, 1, build_context.context->getSettingsRef().max_threads));
            return;
        }
    }

    Stopwatch stage_watch, total_watch;
    total_watch.start();
    stage_watch.start();
    storage = DatabaseCatalog::instance().getTable(storage_id, build_context.context);

    auto * merge_tree_storage = dynamic_cast<MergeTreeMetaBase *>(storage.get());
    bool is_merge_tree = merge_tree_storage != nullptr;

    bool use_projection_index = build_context.context->getSettingsRef().optimizer_index_projection_support && is_merge_tree
        && build_context.context->getSettingsRef().enable_ab_index_optimization;

    bool use_optimizer_projection_selection