Method initializePipeline

src/QueryPlan/ReadStorageRowCountStep.cpp:38–76

    return std::make_shared<ReadStorageRowCountStep>(output_stream->header, query, agg_desc, num_rows, is_final_agg, database_and_table);
}
void ReadStorageRowCountStep::initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & context)
{
    const auto & func = agg_desc.function;
    const AggregateFunctionCount & agg_count = static_cast<const AggregateFunctionCount &>(*func);
    Block output_header;
    if (is_final_agg)
    {
        auto count_column = ColumnVector<UInt64>::create();
        count_column->insertValue(num_rows);
        output_header.insert({count_column->getPtr(), agg_count.getReturnType(), agg_desc.column_name});
    }
    else
    {
        std::vector<char> state(agg_count.sizeOfData());
        AggregateDataPtr place = state.data();

        agg_count.create(place);
        SCOPE_EXIT_MEMORY_SAFE(agg_count.destroy(place));

        agg_count.set(place, num_rows);
        auto column = ColumnAggregateFunction::create(func);
        column->insertFrom(place);

        // AggregateFunction's argument type must keep same. 
        output_header.insert(
            {std::move(column), std::make_shared<DataTypeAggregateFunction>(func, func->getArgumentTypes(), agg_desc.parameters), agg_desc.column_name});
    }

    auto istream = std::make_shared<OneBlockInputStream>(output_header);
    auto pipe = Pipe(std::make_shared<SourceFromInputStream>(istream));

    for (const auto & processor : pipe.getProcessors())
        processors.emplace_back(processor);

    pipeline.init(std::move(pipe));

    if (context.context)
        pipeline.addInterpreterContext(context.context);
}

void ReadStorageRowCountStep::toProto(Protos::ReadStorageRowCountStep & proto, bool) const
{