ClickHouse查询优化

// src/Interpreters/executeQuery.cpp

static std::tuple<ASTPtr, BlockIO> executeQueryImpl()
{
    // 构造Parser
    ParserQuery parser(end, settings.allow_settings_after_format_in_insert);

    // 将SQL转为抽象语法树
    ast = parseQuery(parser, begin, end, "", max_query_size, settings.max_parser_depth);

    // 设置query的上下文，比如SETTINGS
    ...

    if (async_insert)
    {
        ...
    } else {
        // 生成interpreter实例
        interpreter = InterpreterFactory::get(ast, context, SelectQueryOptions(stage).setInternal(internal));

        // interpreter优化AST并返回执行计划
        res = interpreter->execute();
    }

    // 返回抽象语法树和执行计划
    return std::make_tuple(ast, std::move(res));
}

// src/Parsers/parseQuery.cpp

ASTPtr tryParseQuery()
{
    // 将SQL拆分为token流
    Tokens tokens(query_begin, all_queries_end, max_query_size);
    IParser::Pos token_iterator(tokens, max_parser_depth);

    // 将token流解析为语法树
    ASTPtr res;
    const bool parse_res = parser.parse(token_iterator, res, expected);

    return res;
}

// src/Parsers/ParserQuery.cpp

bool ParserQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
{
    ParserQueryWithOutput query_with_output_p(end, allow_settings_after_format_in_insert);
    ParserInsertQuery insert_p(end, allow_settings_after_format_in_insert);
    ParserUseQuery use_p;
    ParserSetQuery set_p;
    ParserSystemQuery system_p;
    ParserCreateUserQuery create_user_p;
    ParserCreateRoleQuery create_role_p;
    ParserCreateQuotaQuery create_quota_p;
    ParserCreateRowPolicyQuery create_row_policy_p;
    ParserCreateSettingsProfileQuery create_settings_profile_p;
    ParserCreateFunctionQuery create_function_p;
    ParserDropFunctionQuery drop_function_p;
    ParserDropAccessEntityQuery drop_access_entity_p;
    ParserGrantQuery grant_p;
    ParserSetRoleQuery set_role_p;
    ParserExternalDDLQuery external_ddl_p;
    ParserTransactionControl transaction_control_p;
    ParserBackupQuery backup_p;

    bool res = query_with_output_p.parse(pos, node, expected)
        || insert_p.parse(pos, node, expected)
        || use_p.parse(pos, node, expected)
        || set_role_p.parse(pos, node, expected)
        || set_p.parse(pos, node, expected)
        || system_p.parse(pos, node, expected)
        || create_user_p.parse(pos, node, expected)
        || create_role_p.parse(pos, node, expected)
        || create_quota_p.parse(pos, node, expected)
        || create_row_policy_p.parse(pos, node, expected)
        || create_settings_profile_p.parse(pos, node, expected)
        || create_function_p.parse(pos, node, expected)
        || drop_function_p.parse(pos, node, expected)
        || drop_access_entity_p.parse(pos, node, expected)
        || grant_p.parse(pos, node, expected)
        || external_ddl_p.parse(pos, node, expected)
        || transaction_control_p.parse(pos, node, expected)
        || backup_p.parse(pos, node, expected);

    return res;
}

// src/Interpreters/InterpreterFactory.cpp

std::unique_ptr<IInterpreter> InterpreterFactory::get()
{
    ...
    if (query->as<ASTSelectQuery>())
    {
        return std::make_unique<InterpreterSelectQuery>(query, context, options);
    }
    ...
}

graph TD
    A[初始化settings] --> B[优化with]
    B --> C[优化joins]
    C --> D[谓词下推将where下推到prewhere]
    D --> E{是否要再次优化}
    E --> |是| A
    E --> |否| F[检查storage权限]
    F --> G[生成analysis_result和result_header]

// src/Interpreters/InterpreterSelectQuery.cpp

void InterpreterSelectQuery::executeImpl()
{
    ...
    // 个人理解针对EXPLAIN PLAN，只构建执行计划不执行
    if (options.only_analyze)
    {
        ...
    }
    else
    {
        // 从磁盘读取所需列，注意这一行，后文跳转进去分析
        executeFetchColumns(from_stage, query_plan);
    }
    if (options.to_stage > QueryProcessingStage::FetchColumns)
    {
        // 在分布式执行Query时只在远程节点执行
        if (expressions.first_stage)
        {
            // 当storage不支持prewhere时添加FilterStep
            if (!query_info.projection && expressions.filter_info)
            {
                ...
            }
            if (expressions.before_array_join)
            {
                ...
            }
            if (expressions.array_join)
            {
                ...
            }
            if (expressions.before_join)
            {
                ...
            }
            // 可选步骤：将join key转为一致的supertype
            if (expressions.converting_join_columns)
            {
                ...
            }
            // 添加Join
            if (expressions.hasJoin())
            {
                ...
            }
            // 添加where
            if (!query_info.projection && expressions.hasWhere())
                executeWhere(query_plan, expressions.before_where, expressions.remove_where_filter);
            // 添加aggregation
            if (expressions.need_aggregate)
            {
                executeAggregation(
                    query_plan, expressions.before_aggregation, aggregate_overflow_row, aggregate_final, query_info.input_order_info);
                /// We need to reset input order info, so that executeOrder can't use it
                query_info.input_order_info.reset();
                if (query_info.projection)
                    query_info.projection->input_order_info.reset();
            }
            // 准备执行：
            // 1. before windows函数
            // 2. windows函数
            // 3. after windows函数
            // 4. 准备DISTINCT
            if (expressions.need_aggregate)
            {
                // 存在聚合函数，在windows函数/ORDER BY之前不执行
            }
            else
            {
                // 不存在聚合函数
                // 存在windows函数，应该在初始节点运行
                // 并且，ORDER BY和DISTINCT依赖于windows函数，这里也不能运行
                if (query_analyzer->hasWindow())
                {
                    executeExpression(query_plan, expressions.before_window, "Before window functions");
                }
                else
                {
                    // 没有windows函数，执行before ORDER BY、准备DISTINCT
                    assert(!expressions.before_window);
                    executeExpression(query_plan, expressions.before_order_by, "Before ORDER BY");
                    executeDistinct(query_plan, true, expressions.selected_columns, true);
                }
            }
            // 如果查询没有GROUP、HAVING，有ORDER或LIMIT，会在远程排序、LIMIT
            preliminary_sort();
        }
        // 在分布式执行Query时只在初始节点执行或optimize_distributed_group_by_sharding_key开启时
        if (expressions.second_stage || from_aggregation_stage)
        {
             if (from_aggregation_stage)
            {
                // 远程节点聚合过，这里啥也不干
            }
            else if (expressions.need_aggregate)
            {
                // 从不同节点拉取数据合并
                if (!expressions.first_stage)
                    executeMergeAggregated(query_plan, aggregate_overflow_row, aggregate_final);
                
                if (!aggregate_final)
                {
                    // 执行group by with totals/rollup/cube
                    ...
                }
                // 添加Having
                else if (expressions.hasHaving())
                    executeHaving(query_plan, expressions.before_having, expressions.remove_having_filter);
            }
            // 报个错
            else if (query.group_by_with_totals || query.group_by_with_rollup || query.group_by_with_cube)
                throw Exception("WITH TOTALS, ROLLUP or CUBE are not supported without aggregation", ErrorCodes::NOT_IMPLEMENTED);
            // 准备执行：
            // 1. before windows函数
            // 2. windows函数
            // 3. after windows函数
            // 4. 准备DISTINCT
            if (from_aggregation_stage)
            {
                if (query_analyzer->hasWindow())
                    throw Exception(
                        "Window functions does not support processing from WithMergeableStateAfterAggregation",
                        ErrorCodes::NOT_IMPLEMENTED);
            }
            else if (expressions.need_aggregate)
            {
                executeExpression(query_plan, expressions.before_window,
                    "Before window functions");
                executeWindow(query_plan);
                executeExpression(query_plan, expressions.before_order_by, "Before ORDER BY");
                executeDistinct(query_plan, true, expressions.selected_columns, true);
            }
            else
            {
                if (query_analyzer->hasWindow())
                {
                    executeWindow(query_plan);
                    executeExpression(query_plan, expressions.before_order_by, "Before ORDER BY");
                    executeDistinct(query_plan, true, expressions.selected_columns, true);
                }
                else
                {
                    // Neither aggregation nor windows, all expressions before
                    // ORDER BY executed on shards.
                }
            }
            // 添加order by
            if (expressions.has_order_by)
            {
                // 在分布式查询中，没有聚合函数却有order by，将会在远端节点order by
                ...
            }
            // 多source order by优化
            ...
            
            // 多条流时再次执行distinct
            if (!from_aggregation_stage && query.distinct)
                executeDistinct(query_plan, false, expressions.selected_columns, false);

            // 处理limit
            ...
            // 处理projection
            ...
            // 处理offset
            ...
        }
        
        // 需要子查询结果构建set
        if (!subqueries_for_sets.empty())
            executeSubqueriesInSetsAndJoins(query_plan, subqueries_for_sets);
    }
}

graph TD
    A[Partition Key MinMax Index] --> B[Partition]
    B --> C[Primary Key Sparse Index]
    C --> D[Data Skipping Indexes]

// src/Processors/QueryPlan/ReadFromMergeTree.cpp

MergeTreeDataSelectAnalysisResultPtr ReadFromMergeTree::selectRangesToRead()
{
    ...
    try
    {
        // 使用partition by选取需要parts
        MergeTreeDataSelectExecutor::filterPartsByPartition(...);
        // 处理抽样
        ...
        // 使用主键索引和跳表索引
        result.parts_with_ranges = MergeTreeDataSelectExecutor::filterPartsByPrimaryKeyAndSkipIndexes(...);
    }
    catch(...)
    {
        ...
    }
    ...
}

SELECT
    database,
    table,
    count() AS parts,
    uniqExact(partition_id) AS partition_cnt,
    sum(rows),
    formatReadableSize(sum(data_compressed_bytes) AS comp_bytes) AS comp,
    formatReadableSize(sum(data_uncompressed_bytes) AS uncomp_bytes) AS uncomp,
    uncomp_bytes / comp_bytes AS ratio,
    formatReadableSize(sum(marks_bytes) AS mark_sum) AS marks,
    mark_sum / uncomp_bytes AS mark_ratio
FROM clusterAllReplicas(default_cluster, system.parts)
WHERE active
GROUP BY
    database,
    table
ORDER BY comp_bytes DESC

SELECT count() FROM table WHERE ... AND cityHash64(some_high_card_key) % 10 = 0; -- Deterministic
SELECT count() FROM table WHERE ... AND rand() % 10 = 0; -- Non-deterministic

<yandex>
	<profile>
  	<root>
      <max_insert_block_size>16777216</max_insert_block_size>
      <input_format_parallel_parsing>0</input_format_parallel_parsing>
      <min_insert_block_size_rows>10000000</min_insert_block_size_rows>
      <min_insert_block_size_bytes>1024000000</min_insert_block_size_bytes>
    </root>
  </profile>
</yandex>

// src/Interpreters/SquashingTransform.cpp

bool SquashingTransform::isEnoughSize(size_t rows, size_t bytes) const
{
    return (!min_block_size_rows && !min_block_size_bytes)
        || (min_block_size_rows && rows >= min_block_size_rows)
        || (min_block_size_bytes && bytes >= min_block_size_bytes);
}

<yandex>
    <profiles>
        <query>
            <max_memory_usage>10000000000</max_memory_usage>
            <max_memory_usage_for_all_queries>100000000000</max_memory_usage_for_all_queries>
            <max_rows_to_read>1000000000</max_rows_to_read>
            <max_bytes_to_read>100000000000</max_bytes_to_read>
            <max_rows_to_group_by>1000000</max_rows_to_group_by>
            <group_by_overflow_mode>any</group_by_overflow_mode>
            <max_rows_to_sort>1000000</max_rows_to_sort>
            <max_bytes_to_sort>1000000000</max_bytes_to_sort>
            <max_result_rows>100000</max_result_rows>
            <max_result_bytes>100000000</max_result_bytes>
            <result_overflow_mode>break</result_overflow_mode>
            <max_execution_time>60</max_execution_time>
            <min_execution_speed>1000000</min_execution_speed>
            <timeout_before_checking_execution_speed>15</timeout_before_checking_execution_speed>
            <max_columns_to_read>25</max_columns_to_read>
            <max_temporary_columns>100</max_temporary_columns>
            <max_temporary_non_const_columns>50</max_temporary_non_const_columns>
            <max_subquery_depth>2</max_subquery_depth>
            <max_pipeline_depth>25</max_pipeline_depth>
            <max_ast_depth>50</max_ast_depth>
            <max_ast_elements>100</max_ast_elements>
            <readonly>1</readonly>
        </query>
    </profiles>
</yandex>