Skip to content

perf: read the unfiltered buckets once in issue_analysis_copy_pipe #5165

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
epipav merged 1 commit into from
Nov 3, 2026
Merged

perf: read the unfiltered buckets once in issue_analysis_copy_pipe #5165

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
207 changes: 137 additions & 70 deletions services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@ DESCRIPTION >
each bucket's latest snapshot scans the bucket datasources and is too slow for the push-time check.
The first non-author response still comes from the bot-filtered `activityRelations_deduplicated_cleaned_bucket_union`, so an
automated bot comment does not count as a response.
The unfiltered buckets are read once per run: `issue_events` reads both event types together, and `issues` is referenced
only by the final node. A node referenced twice is computed twice, and reading the buckets takes most of the run.

NODE issue_events_delta
DESCRIPTION >
Expand All @@ -30,109 +32,174 @@ SQL >
ORDER BY updatedAt DESC
LIMIT 1 BY segmentId, timestamp, type, platform, channel, sourceId

NODE issues_opened_events
NODE issue_events
DESCRIPTION >
Issue opened events: the daily unfiltered buckets, with any activity that also appears in the MV replaced by its MV version.
Issue opened and closed events from the daily unfiltered buckets and the MV, in one read of the buckets. An opened event
that also appears in the MV is replaced by its MV version. A close keeps both versions, since only the earliest close per
issue is used.

SQL >
SELECT activityId, sourceId, timestamp, segmentId, channel, platform, memberId, organizationId
FROM activityRelations_enriched_deduplicated_bucket_union
WHERE
type = 'issues-opened'
AND activityId NOT IN (SELECT activityId FROM issue_events_delta WHERE type = 'issues-opened')
UNION ALL
SELECT activityId, sourceId, timestamp, segmentId, channel, platform, memberId, organizationId
FROM issue_events_delta
WHERE type = 'issues-opened'

NODE issues_opened
SQL >
SELECT
activityId as id,
activityId,
sourceId,
timestamp AS openedAt,
sourceParentId,
timestamp,
segmentId,
lower(channel) AS channel,
channel,
platform,
memberId,
organizationId
FROM issues_opened_events
organizationId,
type
FROM activityRelations_enriched_deduplicated_bucket_union
WHERE
toYear(timestamp) >= 1971
type IN ('issues-opened', 'issues-closed')
AND (
(
platform IN ('git', 'gerrit', 'github', 'gitlab')
AND (lower(channel), segmentId)
IN (SELECT lower(channel), segmentId FROM repos_to_channels)
)
OR platform NOT IN ('git', 'gerrit', 'github', 'gitlab')
type = 'issues-closed'
OR activityId
NOT IN (SELECT activityId FROM issue_events_delta WHERE type = 'issues-opened')
)
UNION ALL
SELECT
activityId,
sourceId,
sourceParentId,
timestamp,
segmentId,
channel,
platform,
memberId,
organizationId,
type
FROM issue_events_delta

NODE issues_closed
NODE issues
DESCRIPTION >
First close per issue from the daily buckets and the MV together. A close present in both counts the same under MIN, so no
deduplication is needed here.
One row per issue opened in an enabled repository, with the issue's first close from any repository. Opened events key
on `sourceId` and closes on `sourceParentId`, so grouping by that key pairs them without a join. `openerCount` is the
number of distinct non-empty openers per issue and `soleOpener` is the opener when there is one, for the response rule in
the final node.

SQL >
SELECT sourceParentId, MIN(timestamp) AS closedAt
SELECT
tupleElement(opened, 1) AS id,
tupleElement(opened, 2) AS sourceId,
tupleElement(opened, 3) AS openedAt,
tupleElement(opened, 4) AS segmentId,
tupleElement(opened, 5) AS channel,
tupleElement(opened, 6) AS platform,
tupleElement(opened, 7) AS memberId,
tupleElement(opened, 8) AS organizationId,
firstClosedAt,
openerCount,
soleOpener
FROM
(
SELECT sourceParentId, timestamp
FROM activityRelations_enriched_deduplicated_bucket_union
WHERE type = 'issues-closed'
UNION ALL
SELECT sourceParentId, timestamp
FROM issue_events_delta
WHERE type = 'issues-closed'
SELECT arrayJoin(openedRows) AS opened, firstClosedAt, openerCount, soleOpener
FROM
(
SELECT
if(type = 'issues-opened', sourceId, sourceParentId) AS issueKey,
minIf(
timestamp, type = 'issues-closed' AND sourceParentId != ''
) AS firstClosedAt,
uniqExactIf(memberId, isOpened AND memberId != '') AS openerCount,
anyIf(memberId, isOpened AND memberId != '') AS soleOpener,
groupArrayIf(
(
activityId,
sourceId,
timestamp,
segmentId,
lower(channel),
platform,
memberId,
organizationId
),
isOpened
) AS openedRows
FROM
(
SELECT
*,
type = 'issues-opened'
AND (
(
platform IN ('git', 'gerrit', 'github', 'gitlab')
AND (lower(channel), segmentId)
IN (SELECT lower(channel), segmentId FROM repos_to_channels)
)
OR platform NOT IN ('git', 'gerrit', 'github', 'gitlab')
) AS isOpened
FROM issue_events
WHERE toYear(timestamp) >= 1971
)
GROUP BY issueKey
)
)
WHERE sourceParentId != '' AND toYear(timestamp) >= 1971
GROUP BY sourceParentId

NODE issues_comment_non_author
NODE issue_first_commenters
DESCRIPTION >
The first two distinct commenters per issue from the bot-filtered buckets, sorted by each one's first comment. The first
non-author response is the first commenter's comment, or the second's when the first is the opener.

SQL >
SELECT c.sourceParentId, MIN(c.timestamp) AS respondedAt
FROM activityRelations_deduplicated_cleaned_bucket_union c
INNER JOIN issues_opened opened ON c.sourceParentId = opened.sourceId
WHERE
c.type = 'issue-comment'
AND c.sourceParentId != ''
AND c.memberId != ''
AND opened.memberId != ''
AND c.memberId != opened.memberId
AND toYear(c.timestamp) >= 1971
GROUP BY c.sourceParentId
SELECT
sourceParentId,
tupleElement(firstTwo[1], 1) AS firstCommentAt,
tupleElement(firstTwo[1], 2) AS firstCommenterId,
if(length(firstTwo) > 1, tupleElement(firstTwo[2], 1), NULL) AS secondCommentAt
FROM
(
SELECT sourceParentId, groupArraySorted(2)((firstCommentAt, memberId)) AS firstTwo
FROM
(
SELECT sourceParentId, memberId, MIN(timestamp) AS firstCommentAt
FROM activityRelations_deduplicated_cleaned_bucket_union
WHERE
type = 'issue-comment'
AND sourceParentId != ''
AND memberId != ''
AND toYear(timestamp) >= 1971
GROUP BY sourceParentId, memberId
)
GROUP BY sourceParentId
)

NODE issue_analysis_results_merged
DESCRIPTION >
A response comes from someone other than the opener. When an issue has two or more distinct openers, every commenter
differs from one of them, so the first comment counts.

SQL >
SELECT
opened.id,
opened.sourceId,
opened.segmentId as segmentId,
opened.channel,
opened.platform,
opened.memberId,
opened.organizationId,
opened.openedAt,
IF(
comment_non_author.respondedAt = toDateTime(0), NULL, comment_non_author.respondedAt
i.id,
i.sourceId,
i.segmentId as segmentId,
i.channel,
i.platform,
i.memberId,
i.organizationId,
i.openedAt,
multiIf(
i.openerCount = 0 OR c.sourceParentId = '',
NULL,
i.openerCount > 1 OR c.firstCommenterId != i.soleOpener,
c.firstCommentAt,
c.secondCommentAt
) AS commentedAt,
IF(closed.closedAt = toDateTime(0), NULL, closed.closedAt) AS closedAt,
IF(i.firstClosedAt = toDateTime(0), NULL, i.firstClosedAt) AS closedAt,
IF(
closedAt IS NULL,
NULL,
toUnixTimestamp(toDateTime(closed.closedAt)) - toUnixTimestamp(toDateTime(opened.openedAt))
toUnixTimestamp(toDateTime(i.firstClosedAt)) - toUnixTimestamp(toDateTime(i.openedAt))
) AS closedInSeconds,
IF(
commentedAt IS NULL,
NULL,
toUnixTimestamp(toDateTime(comment_non_author.respondedAt))
- toUnixTimestamp(toDateTime(opened.openedAt))
toUnixTimestamp(toDateTime(commentedAt)) - toUnixTimestamp(toDateTime(i.openedAt))
) AS respondedInSeconds
FROM issues_opened opened
LEFT JOIN issues_closed AS closed ON opened.sourceId = closed.sourceParentId
LEFT JOIN
issues_comment_non_author AS comment_non_author
ON opened.sourceId = comment_non_author.sourceParentId
FROM issues i
LEFT JOIN issue_first_commenters c ON i.sourceId = c.sourceParentId

TYPE COPY
TARGET_DATASOURCE issues_analyzed
Expand Down
Loading