Method executeFetchColumns

src/Interpreters/InterpreterSelectQuery.cpp:2716–2940

}

void InterpreterSelectQuery::executeFetchColumns(QueryProcessingStage::Enum processing_stage, QueryPlan & query_plan)
{
    auto & query = getSelectQuery();
    const Settings & settings = context->getSettingsRef();
    std::optional<UInt64> num_rows;

    /// Optimization for trivial query like SELECT count() FROM table.
    if (processing_stage == QueryProcessingStage::FetchColumns && (num_rows = getTrivialCount(settings[Setting::allow_experimental_parallel_reading_from_replicas])))
    {
        const auto & desc = query_analyzer->aggregates()[0];
        const auto & func = desc.function;
        const AggregateFunctionCount & agg_count = static_cast<const AggregateFunctionCount &>(*func);

        /// We will process it up to "WithMergeableState".
        std::vector<char> state(agg_count.sizeOfData());
        AggregateDataPtr place = state.data();

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

        AggregateFunctionCount::set(place, *num_rows);

        auto column = ColumnAggregateFunction::create(func);
        column->insertFrom(place);

        Block header = analysis_result.before_aggregation->dag.getResultColumns();
        size_t arguments_size = desc.argument_names.size();
        DataTypes argument_types(arguments_size);
        for (size_t j = 0; j < arguments_size; ++j)
            argument_types[j] = header.getByName(desc.argument_names[j]).type;

        auto block_with_count = std::make_shared<const Block>(Block{
            {std::move(column), std::make_shared<DataTypeAggregateFunction>(func, argument_types, desc.parameters), desc.column_name}});

        auto source = std::make_shared<SourceFromSingleChunk>(block_with_count);
        auto prepared_count = std::make_unique<ReadFromPreparedSource>(Pipe(std::move(source)));
        prepared_count->setStepDescription("Optimized trivial count");
        query_plan.addStep(std::move(prepared_count));
        from_stage = QueryProcessingStage::WithMergeableState;
        analysis_result.first_stage = false;
        return;
    }

    /// Limitation on the number of columns to read.
    /// It's not applied in 'only_analyze' mode, because the query could be analyzed without removal of unnecessary columns.
    if (!options.only_analyze && settings[Setting::max_columns_to_read] && required_columns.size() > settings[Setting::max_columns_to_read])
        throw Exception(
            ErrorCodes::TOO_MANY_COLUMNS,
            "Limit for number of columns to read exceeded. Requested: {}, maximum: {}",
            required_columns.size(),
            settings[Setting::max_columns_to_read].value);

    /// General limit for the number of threads.
    size_t max_threads_execute_query = getMaxThreadsForAvailableMemory(
        settings[Setting::max_threads], settings[Setting::max_threads_min_free_memory_per_thread]);

    /**
     * To simultaneously query more remote servers when async_socket_for_remote is off