Scope job analytics stages join by pipeline ids (!244536) · Merge requests · GitLab.org / GitLab

What does this MR do and why?

Scopes the CI job analytics stages join by pipeline ids when a pipeline filter (source/ref) is present, so the by_traversal_path_pipeline_id projection (added in !244535 (merged)) can prune the stages scan instead of reading the container's entire stage history.

Changes:

SiphonFinishedBuildsFinder#filter_by_pipeline_attrs now hoists the pipelines dedup subquery into a named WITH pipelines AS (...) CTE. Both the builds side (commit_id IN) and the stages side (pipeline_id IN) reference it by name, so the subquery is emitted once.
#stage_join scopes the stages subquery by the pipelines CTE when one exists. This means filter_by_pipeline_attrs must be called before with_stages; Ci::JobAnalytics::QueryBuilder is reordered accordingly. Calling them the other way round still returns correct results, it just skips the stages-side pruning.
The stages join now also matches on partition_id (stage_id = stages.id AND partition_id = stages.partition_id). Joining on id alone would fan out build rows if the same stage id existed in another partition. Covered by new execution specs that insert the same id under a different partition_id for builds, stages and pipelines.
SiphonStagesFinder gains for_pipeline_ids. pipeline_id is immutable per stage row, so filtering before the dedup step is safe.
Without a pipeline filter, behavior is unchanged: stages scoped by traversal_path only, no pipelines CTE.

Production validation (test projection with this query shape, 90-day window, gitlab-org/gitlab): stages granules 7,514 -> 780 (about 10x), total rows_read 36.8M -> 23.1M (-37%). On 7-day (default) views the effect is proportionally larger since the stages scan was about 87% of rows_read.

Resolves the finder half of #605208 (closed).

References

Issue: #605208 (closed) (design, measurements, EXPLAIN plans)
Parent MR (projection, stacked): !244535 (merged)
Feature issue: #601321 (closed)

Database changes

No schema changes in this MR (the projection ships in !244535 (merged)). Emitted SQL changes:

pipelines dedup subquery becomes a named CTE (defined once, first in the WITH list)
stages subquery gains pipeline_id IN (SELECT id FROM pipelines) when a source/ref filter is present
stages join matches on both stage_id and partition_id

Query Plans

Before

Query

WITH finished_builds AS (
    SELECT
        `siphon_p_ci_builds`.`id`,
        `siphon_p_ci_builds`.`partition_id`,
        argMax(
            `siphon_p_ci_builds`.`stage_id`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS stage_id,
        argMax(
            `siphon_p_ci_builds`.`name`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS name,
        argMax(
            `siphon_p_ci_builds`.`started_at`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS started_at,
        argMax(
            `siphon_p_ci_builds`.`finished_at`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS finished_at,
        argMax(
            `siphon_p_ci_builds`.`_siphon_deleted`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS _siphon_deleted,
        argMax(
            `siphon_p_ci_builds`.`type`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS type,
        argMax(
            `siphon_p_ci_builds`.`status`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS status
    FROM
        `siphon_p_ci_builds`
    WHERE
        `siphon_p_ci_builds`.`traversal_path` = '1/9970/15846663/'
        AND `siphon_p_ci_builds`.`commit_id` IN (
            SELECT
                `pipelines`.`id`
            FROM
                (
                    SELECT
                        `siphon_p_ci_pipelines`.`id`,
                        `siphon_p_ci_pipelines`.`partition_id`,
                        argMax(
                            `siphon_p_ci_pipelines`.`traversal_path`,
                            `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                        ) AS traversal_path,
                        argMax(
                            `siphon_p_ci_pipelines`.`status`,
                            `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                        ) AS status,
                        argMax(
                            `siphon_p_ci_pipelines`.`source`,
                            `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                        ) AS source,
                        argMax(
                            `siphon_p_ci_pipelines`.`ref`,
                            `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                        ) AS ref,
                        argMax(
                            `siphon_p_ci_pipelines`.`started_at`,
                            `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                        ) AS started_at,
                        argMax(
                            `siphon_p_ci_pipelines`.`finished_at`,
                            `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                        ) AS finished_at,
                        argMax(
                            `siphon_p_ci_pipelines`.`duration`,
                            `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                        ) AS duration,
                        argMax(
                            `siphon_p_ci_pipelines`.`_siphon_deleted`,
                            `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                        ) AS _siphon_deleted
                    FROM
                        `siphon_p_ci_pipelines`
                    WHERE
                        `siphon_p_ci_pipelines`.`traversal_path` = '1/9970/15846663/'
                        AND `siphon_p_ci_pipelines`.`started_at` >= toDateTime64('2026-04-07 00:00:00', 6, 'UTC')
                        AND `siphon_p_ci_pipelines`.`started_at` < toDateTime64('2026-07-06 00:00:00', 6, 'UTC')
                    GROUP BY
                        id,
                        partition_id
                ) pipelines
            WHERE
                `pipelines`.`ref` = 'master'
                AND `pipelines`.`_siphon_deleted` = 'false'
        )
    GROUP BY
        id,
        partition_id
)
SELECT
    `finished_builds`.`name`,
    `stages`.`name` AS stage_name,
    round(
        (
            avg(
                if(
                    `finished_builds`.`started_at` IS NOT NULL
                    AND `finished_builds`.`finished_at` IS NOT NULL
                    AND `finished_builds`.`finished_at` > `finished_builds`.`started_at`,
                    age(
                        'ms',
                        `finished_builds`.`started_at`,
                        `finished_builds`.`finished_at`
                    ),
                    NULL
                )
            ) / 1000.0
        ),
        2
    ) AS mean_duration,
    round(
        (
            quantile(0.95)(
                if(
                    `finished_builds`.`started_at` IS NOT NULL
                    AND `finished_builds`.`finished_at` IS NOT NULL
                    AND `finished_builds`.`finished_at` > `finished_builds`.`started_at`,
                    age(
                        'ms',
                        `finished_builds`.`started_at`,
                        `finished_builds`.`finished_at`
                    ),
                    NULL
                )
            ) / 1000.0
        ),
        2
    ) AS p95_duration,
    count() AS total_count
FROM
    `finished_builds`
    LEFT OUTER JOIN (
        SELECT
            *
        FROM
            (
                SELECT
                    `siphon_p_ci_stages`.`id`,
                    `siphon_p_ci_stages`.`partition_id`,
                    argMax(
                        `siphon_p_ci_stages`.`traversal_path`,
                        `siphon_p_ci_stages`.`_siphon_replicated_at`
                    ) AS traversal_path,
                    argMax(
                        `siphon_p_ci_stages`.`pipeline_id`,
                        `siphon_p_ci_stages`.`_siphon_replicated_at`
                    ) AS pipeline_id,
                    argMax(
                        `siphon_p_ci_stages`.`name`,
                        `siphon_p_ci_stages`.`_siphon_replicated_at`
                    ) AS name,
                    argMax(
                        `siphon_p_ci_stages`.`_siphon_deleted`,
                        `siphon_p_ci_stages`.`_siphon_replicated_at`
                    ) AS _siphon_deleted
                FROM
                    `siphon_p_ci_stages`
                WHERE
                    `siphon_p_ci_stages`.`traversal_path` = '1/9970/15846663/'
                GROUP BY
                    id,
                    partition_id
            ) stages
        WHERE
            `stages`.`_siphon_deleted` = 0
    ) stages ON `finished_builds`.`stage_id` = `stages`.`id`
WHERE
    `finished_builds`.`_siphon_deleted` = 0
    AND `finished_builds`.`type` = 'Ci::Build'
    AND `finished_builds`.`status` IN ('success', 'failed', 'canceled', 'skipped')
GROUP BY
    `finished_builds`.`name`,
    `stages`.`name`
ORDER BY
    mean_duration DESC
LIMIT
    6

Query Plan

Elapsed: 5.143s Read: 243,849,382 rows (4.64 GB)

explain
Expression ((Project names + (Before ORDER BY + Projection) [lifted up part]))
Limit (preliminary LIMIT)
Sorting (Sorting for ORDER BY)
Expression ((Before ORDER BY + Projection))
Aggregating
Expression ((Before GROUP BY + ))
Join (JOIN FillRightFirst)
CreatingSets (Create sets before main query execution)
Filter (((WHERE + (JOIN actions + Change column names to column identifiers)) + (Project names + Projection)))
Aggregating
Expression (Before GROUP BY)
Expression ((WHERE + Change column names to column identifiers))
ReadFromMergeTree (gitlab_clickhouse_main_production.siphon_p_ci_builds)
Indexes:
PrimaryKey
Keys:
traversal_path
Condition: (traversal_path in ['1/9970/15846663/', '1/9970/15846663/'])
Parts: 22/24
Granules: 110367/5931343
Search Algorithm: binary search
Skip
Name: idx_commit_id_bf
Description: bloom_filter GRANULARITY 1
Parts: 22/22
Granules: 110367/110367
Ranges: 22
Expression ((JOIN actions + (Change column names to column identifiers + (Project names + Projection))))
Filter ((WHERE + (Change column names to column identifiers + (Project names + Projection))))
Aggregating
Expression (Before GROUP BY)
Expression ((WHERE + Change column names to column identifiers))
ReadFromMergeTree (gitlab_clickhouse_main_production.siphon_p_ci_stages)
Indexes:
PrimaryKey
Keys:
traversal_path
Condition: (traversal_path in ['1/9970/15846663/', '1/9970/15846663/'])
Parts: 15/17
Granules: 7761/2737713
Search Algorithm: binary search
Ranges: 15

After

Query

WITH pipelines AS (
    SELECT
        `pipelines`.`id`
    FROM
        (
            SELECT
                `siphon_p_ci_pipelines`.`id`,
                `siphon_p_ci_pipelines`.`partition_id`,
                argMax(
                    `siphon_p_ci_pipelines`.`traversal_path`,
                    `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                ) AS traversal_path,
                argMax(
                    `siphon_p_ci_pipelines`.`status`,
                    `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                ) AS status,
                argMax(
                    `siphon_p_ci_pipelines`.`source`,
                    `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                ) AS source,
                argMax(
                    `siphon_p_ci_pipelines`.`ref`,
                    `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                ) AS ref,
                argMax(
                    `siphon_p_ci_pipelines`.`started_at`,
                    `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                ) AS started_at,
                argMax(
                    `siphon_p_ci_pipelines`.`finished_at`,
                    `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                ) AS finished_at,
                argMax(
                    `siphon_p_ci_pipelines`.`duration`,
                    `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                ) AS duration,
                argMax(
                    `siphon_p_ci_pipelines`.`_siphon_deleted`,
                    `siphon_p_ci_pipelines`.`_siphon_replicated_at`
                ) AS _siphon_deleted
            FROM
                `siphon_p_ci_pipelines`
            WHERE
                `siphon_p_ci_pipelines`.`traversal_path` = '1/9970/15846663/'
                AND `siphon_p_ci_pipelines`.`started_at` >= toDateTime64('2026-04-07 00:00:00', 6, 'UTC')
                AND `siphon_p_ci_pipelines`.`started_at` < toDateTime64('2026-07-06 00:00:00', 6, 'UTC')
            GROUP BY
                id,
                partition_id
        ) pipelines
    WHERE
        `pipelines`.`ref` = 'master'
        AND `pipelines`.`_siphon_deleted` = 'false'
),
finished_builds AS (
    SELECT
        `siphon_p_ci_builds`.`id`,
        `siphon_p_ci_builds`.`partition_id`,
        argMax(
            `siphon_p_ci_builds`.`stage_id`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS stage_id,
        argMax(
            `siphon_p_ci_builds`.`name`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS name,
        argMax(
            `siphon_p_ci_builds`.`started_at`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS started_at,
        argMax(
            `siphon_p_ci_builds`.`finished_at`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS finished_at,
        argMax(
            `siphon_p_ci_builds`.`_siphon_deleted`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS _siphon_deleted,
        argMax(
            `siphon_p_ci_builds`.`type`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS type,
        argMax(
            `siphon_p_ci_builds`.`status`,
            `siphon_p_ci_builds`.`_siphon_replicated_at`
        ) AS status
    FROM
        `siphon_p_ci_builds`
    WHERE
        `siphon_p_ci_builds`.`traversal_path` = '1/9970/15846663/'
        AND `siphon_p_ci_builds`.`commit_id` IN (
            SELECT
                `pipelines`.`id`
            FROM
                `pipelines`
        )
    GROUP BY
        id,
        partition_id
)
SELECT
    `finished_builds`.`name`,
    `stages`.`name` AS stage_name,
    round(
        (
            avg(
                if(
                    `finished_builds`.`started_at` IS NOT NULL
                    AND `finished_builds`.`finished_at` IS NOT NULL
                    AND `finished_builds`.`finished_at` > `finished_builds`.`started_at`,
                    age(
                        'ms',
                        `finished_builds`.`started_at`,
                        `finished_builds`.`finished_at`
                    ),
                    NULL
                )
            ) / 1000.0
        ),
        2
    ) AS mean_duration,
    round(
        (
            quantile(0.95)(
                if(
                    `finished_builds`.`started_at` IS NOT NULL
                    AND `finished_builds`.`finished_at` IS NOT NULL
                    AND `finished_builds`.`finished_at` > `finished_builds`.`started_at`,
                    age(
                        'ms',
                        `finished_builds`.`started_at`,
                        `finished_builds`.`finished_at`
                    ),
                    NULL
                )
            ) / 1000.0
        ),
        2
    ) AS p95_duration,
    count() AS total_count
FROM
    `finished_builds`
    LEFT OUTER JOIN (
        SELECT
            *
        FROM
            (
                SELECT
                    `siphon_p_ci_stages`.`id`,
                    `siphon_p_ci_stages`.`partition_id`,
                    argMax(
                        `siphon_p_ci_stages`.`traversal_path`,
                        `siphon_p_ci_stages`.`_siphon_replicated_at`
                    ) AS traversal_path,
                    argMax(
                        `siphon_p_ci_stages`.`pipeline_id`,
                        `siphon_p_ci_stages`.`_siphon_replicated_at`
                    ) AS pipeline_id,
                    argMax(
                        `siphon_p_ci_stages`.`name`,
                        `siphon_p_ci_stages`.`_siphon_replicated_at`
                    ) AS name,
                    argMax(
                        `siphon_p_ci_stages`.`_siphon_deleted`,
                        `siphon_p_ci_stages`.`_siphon_replicated_at`
                    ) AS _siphon_deleted
                FROM
                    `siphon_p_ci_stages`
                WHERE
                    `siphon_p_ci_stages`.`traversal_path` = '1/9970/15846663/'
                    AND `siphon_p_ci_stages`.`pipeline_id` IN (
                        SELECT
                            `pipelines`.`id`
                        FROM
                            `pipelines`
                    )
                GROUP BY
                    id,
                    partition_id
            ) stages
        WHERE
            `stages`.`_siphon_deleted` = 0
    ) stages ON `finished_builds`.`stage_id` = `stages`.`id`
    AND `finished_builds`.`partition_id` = `stages`.`partition_id`
WHERE
    `finished_builds`.`_siphon_deleted` = 0
    AND `finished_builds`.`type` = 'Ci::Build'
    AND `finished_builds`.`status` IN ('success', 'failed', 'canceled', 'skipped')
GROUP BY
    `finished_builds`.`name`,
    `stages`.`name`
ORDER BY
    mean_duration DESC
LIMIT
    6

Query Plan

Elapsed: 2.576s Read: 230,047,266 rows (3.88 GB)

explain
Expression ((Project names + (Before ORDER BY + Projection) [lifted up part]))
Limit (preliminary LIMIT)
Sorting (Sorting for ORDER BY)
Expression ((Before ORDER BY + Projection))
Aggregating
Expression ((Before GROUP BY + ))
Join (JOIN FillRightFirst)
CreatingSets (Create sets before main query execution)
Filter (((WHERE + (JOIN actions + Change column names to column identifiers)) + (Project names + Projection)))
Aggregating
Expression (Before GROUP BY)
Expression ((WHERE + Change column names to column identifiers))
ReadFromMergeTree (gitlab_clickhouse_main_production.siphon_p_ci_builds)
Indexes:
PrimaryKey
Keys:
traversal_path
Condition: (traversal_path in ['1/9970/15846663/', '1/9970/15846663/'])
Parts: 21/24
Granules: 110368/5931443
Search Algorithm: binary search
Skip
Name: idx_commit_id_bf
Description: bloom_filter GRANULARITY 1
Parts: 21/21
Granules: 110368/110368
Ranges: 21
Expression ((JOIN actions + (Change column names to column identifiers + (Project names + Projection))))
CreatingSets (Create sets before main query execution)
Filter (((WHERE + Change column names to column identifiers) + (Project names + Projection)))
Aggregating
Expression
ReadFromMergeTree (by_traversal_path_pipeline_id)
Indexes:
PrimaryKey
Keys:
traversal_path
pipeline_id
Condition: and((traversal_path in ['1/9970/15846663/', '1/9970/15846663/']), (pipeline_id in 46488-element set))
Parts: 15/15
Granules: 771/2737742 ---reads less stages data and uses the projection
Search Algorithm: generic exclusion search
Ranges: 18

How to set up and validate locally

In rails console:

finder = ClickHouse::Finders::Ci::SiphonFinishedBuildsFinder
project = Project.first

sql = finder.for_container(project)
  .filter_by_pipeline_attrs(container: project, from_time: 7.days.ago, to_time: Time.current, ref: 'master')
  .with_stages(project).select(:name, :stage_name).mean_duration
  .to_sql

sql.index('WITH pipelines AS') < sql.index('finished_builds AS')            # => true (CTE order)
sql.include?('`commit_id` IN (SELECT `pipelines`.`id` FROM `pipelines`)')   # => true
sql.include?('`pipeline_id` IN (SELECT `pipelines`.`id` FROM `pipelines`)') # => true
sql.scan('FROM `siphon_p_ci_pipelines`').size                               # => 1 (defined once)

# Without a pipeline filter: unchanged path
sql2 = finder.for_container(project).with_stages(project).select(:name, :stage_name).to_sql
sql2.include?('WITH pipelines') # => false
sql2.include?('`pipeline_id` IN') # => false

MR acceptance checklist

Evaluate this MR against the MR acceptance checklist.

Edited Jul 27, 2026 by Narendran

Scope job analytics stages join by pipeline ids