From ef3ddb97004b047ad5d1c35860007b6dee5511f7 Mon Sep 17 00:00:00 2001 From: Joana Maia Date: Mon, 19 Oct 2026 14:50:48 +0100 Subject: [PATCH 1/6] fix: count bot-merged prs and bot-closed issues in development pipes Refs linuxfoundation/lfx-cdp-insights-ops#790 Signed-off-by: Joana Maia --- services/libs/tinybird/lambda-architecture.md | 25 ++++- .../pipes/issue_analysis_copy_pipe.pipe | 21 +++- .../libs/tinybird/pipes/issues_count.pipe | 91 ++++++++++++++++++ .../tinybird/pipes/pull_requests_count.pipe | 95 +++++++++++++++++++ 4 files changed, 228 insertions(+), 4 deletions(-) create mode 100644 services/libs/tinybird/pipes/issues_count.pipe create mode 100644 services/libs/tinybird/pipes/pull_requests_count.pipe diff --git a/services/libs/tinybird/lambda-architecture.md b/services/libs/tinybird/lambda-architecture.md index 82def14307..e8cf4a1ad2 100644 --- a/services/libs/tinybird/lambda-architecture.md +++ b/services/libs/tinybird/lambda-architecture.md @@ -651,12 +651,35 @@ If you interrupted a mid-run or suspect duplicates exist, either clear the datas **Symptom**: Same activity appears multiple times -**Fix**: Ensure you're filtering by `max(snapshotId)`: +**Fix** (append-mode datasources): Ensure you're filtering by `max(snapshotId)`: ```sql WHERE snapshotId = (SELECT max(snapshotId) FROM datasource_name) ``` +#### Duplicate rows in pull_requests_analyzed + +`pull_requests_analyzed` holds one row per PR and its consumers do not filter by snapshot, so a duplicated PR is counted and averaged twice in every PR metric (development widgets, health scores, leaderboards, org dashboards). Duplicates appear when a bucket of `pull_request_analysis_initial_snapshot` (append mode) is run more than once without clearing that bucket first. The hourly merger carries them forward, so they never clear on their own. + +**Check** duplicate rows per initial-snapshot bucket (the bucket key is `cityHash64(segmentId) % num_buckets`): + +```bash +tb sql "SELECT cityHash64(segmentId) % 10 AS bucket, count() AS rows, uniqExact(id) AS prs, rows - prs AS extra_rows FROM pull_requests_analyzed GROUP BY bucket ORDER BY bucket" +``` + +**Fix** a bucket with `extra_rows` well above 0 by rebuilding only that bucket (example for bucket 5). Run it on `lfx_insights_stg` first, then on `lfx_insights`, and echo the workspace back before each mutating command: + +```bash +tb workspace ls # confirm the workspace +tb pipe copy pause pull_request_analysis_snapshot_merger_copy # stop the hourly merger +tb datasource delete pull_requests_analyzed --sql-condition "cityHash64(segmentId) % 10 = 5" --wait +tb pipe copy run pull_request_analysis_initial_snapshot --param bucket_id=5 --param num_buckets=10 --mode append --wait +tb sql "SELECT count() AS rows, uniqExact(id) AS prs FROM pull_requests_analyzed WHERE cityHash64(segmentId) % 10 = 5" # rows = prs +tb pipe copy resume pull_request_analysis_snapshot_merger_copy +``` + +Run each bucket once. The initial snapshot rebuilds PRs from `activityRelations_enriched_deduplicated_bucket_union`, which the bucket mergers refresh daily at 01:30-01:38 UTC with events up to 00:00 UTC. PR events after that cut reached `pull_requests_analyzed` only through the hourly merger, so the rebuilt rows do not have them. Run the rebuild right after the daily bucket mergers finish to keep that gap to a couple of hours. While the bucket is being rebuilt its PRs are missing from PR metrics, and the PR materialized view keeps running, so a PR in that bucket that receives events during the window can get a partial row. Keep the window short. + ### Missing Data **Symptom**: Recent data not appearing diff --git a/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe b/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe index bc26c5639f..d05a7ddcd1 100644 --- a/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe +++ b/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe @@ -1,5 +1,10 @@ DESCRIPTION > Compacts activities from same issue into one, keeping necessary information in a single row. Helps to serve issue-wide widgets in the development tab. + Opened and closed events come from the unfiltered `activityRelations_enriched_deduplicated_bucket_union`, so issues opened or closed + by bots (e.g. stale bots) are included, the same way `pull_requests_analyzed` includes bot activity. Issues are still limited to + enabled repositories, matching the cleaned buckets. + 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. NODE issues_opened SQL > @@ -12,13 +17,23 @@ SQL > platform, memberId, organizationId - FROM activityRelations_deduplicated_cleaned_bucket_union - WHERE type = 'issues-opened' AND toYear(timestamp) >= 1971 + FROM activityRelations_enriched_deduplicated_bucket_union + WHERE + type = 'issues-opened' + AND toYear(timestamp) >= 1971 + 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') + ) NODE issues_closed SQL > SELECT sourceParentId, MIN(timestamp) AS closedAt - FROM activityRelations_deduplicated_cleaned_bucket_union + FROM activityRelations_enriched_deduplicated_bucket_union WHERE type = 'issues-closed' AND sourceParentId != '' AND toYear(timestamp) >= 1971 GROUP BY sourceParentId diff --git a/services/libs/tinybird/pipes/issues_count.pipe b/services/libs/tinybird/pipes/issues_count.pipe new file mode 100644 index 0000000000..92129bd99c --- /dev/null +++ b/services/libs/tinybird/pipes/issues_count.pipe @@ -0,0 +1,91 @@ +DESCRIPTION > + - `issues_count.pipe` serves the opened and closed counts of the "Development - Issues resolution" widget. + - Counts issues from `issues_analyzed`, which holds one row per issue, so each issue is counted once per state. + - Includes issues opened or closed by bots (e.g. stale bots), since `issues_analyzed` is built from the unfiltered activity source. + - `state` picks the lifecycle timestamp the count is based on: `openedAt` or `closedAt`. Issues without that timestamp are not counted. + - **When `granularity` is NOT provided, returns a single total count** across the entire time range. + - **When `granularity` is provided, returns time-series data** aggregated by period (daily, weekly, monthly, quarterly, yearly), using `generate_timeseries` so periods without issues return 0. + - The pipe scopes data either to a single project (via `segments_filtered`) or to every project in a + collection (via `segments_filtered_by_collection`). Exactly one of `project`/`collectionSlug` must be given. + - Parameters: + - `project`: Project slug (e.g., 'k8s', 'tensorflow') - inherited from `segments_filtered`. Required unless `collectionSlug` is given. + - `collectionSlug`: Collection slug (e.g., 'cncf') - inherited from `segments_filtered_by_collection`. Required unless `project` is given. + - `repos`: Optional array of repository URLs for filtering (e.g., ['https://github.com/kubernetes/kubernetes']) + - `state`: Optional string, one of 'opened' (default) or 'closed' + - `startDate`: Optional DateTime filter for the `state` timestamp after this value (e.g., '2024-01-01 00:00:00') + - `endDate`: Optional DateTime filter for the `state` timestamp before this value (e.g., '2024-12-31 23:59:59') + - `platform`: Optional string filter for source platform (e.g., 'github', 'gitlab') + - `granularity`: Optional string for time aggregation ('daily', 'weekly', 'monthly', 'quarterly', 'yearly') + - Response: + - Without granularity: `issueCount` (total count) + - With granularity: `startDate`, `endDate`, and `issueCount` for each time period + +TAGS "Development metrics", "Widget", "Issues" + +NODE issues_by_state +SQL > + % + SELECT + id, + {% if String( + state, 'opened', description="One of 'opened' or 'closed'", required=False + ) == 'closed' %} closedAt + {% else %} openedAt + {% end %} AS eventAt + FROM issues_analyzed + WHERE + {% if defined(collectionSlug) %} + segmentId IN (SELECT segmentId FROM segments_filtered_by_collection) + {% else %} segmentId = (SELECT segmentId FROM segments_filtered) + {% end %} AND channel IN (SELECT channel FROM repos_to_channels) AND isNotNull(eventAt) + {% if defined(platform) %} + AND platform + = {{ + String( + platform, + description="Filter by platform (e.g., 'github', 'gitlab')", + required=False, + ) + }} + {% end %} + {% if defined(startDate) %} + AND eventAt + > {{ DateTime(startDate, description="Filter state timestamp after", required=False) }} + {% end %} + {% if defined(endDate) %} + AND eventAt + < {{ DateTime(endDate, description="Filter state timestamp before", required=False) }} + {% end %} + +NODE timeseries_generation_for_issues_count +SQL > + % + {% if defined(granularity) %} + SELECT ds."startDate", ds."endDate", countIf(iss.id != '') AS "issueCount" + FROM generate_timeseries ds + LEFT JOIN + issues_by_state iss + ON CASE + WHEN {{ granularity }} = 'daily' + THEN toDate(iss.eventAt) + WHEN {{ granularity }} = 'weekly' + THEN toStartOfWeek(iss.eventAt) + WHEN {{ granularity }} = 'monthly' + THEN toStartOfMonth(iss.eventAt) + WHEN {{ granularity }} = 'quarterly' + THEN toStartOfQuarter(iss.eventAt) + WHEN {{ granularity }} = 'yearly' + THEN toStartOfYear(iss.eventAt) + END + = ds."startDate" + GROUP BY ds."startDate", ds."endDate" + ORDER BY ds."startDate" + {% else %} SELECT 1 + {% end %} + +NODE issues_count_merged +SQL > + % + {% if not defined(granularity) %} SELECT count() AS "issueCount" FROM issues_by_state + {% else %} SELECT * FROM timeseries_generation_for_issues_count + {% end %} diff --git a/services/libs/tinybird/pipes/pull_requests_count.pipe b/services/libs/tinybird/pipes/pull_requests_count.pipe new file mode 100644 index 0000000000..dcfa506ea3 --- /dev/null +++ b/services/libs/tinybird/pipes/pull_requests_count.pipe @@ -0,0 +1,95 @@ +DESCRIPTION > + - `pull_requests_count.pipe` serves the opened, merged and closed counts of the "Development - Pull requests" widget. + - Counts pull requests (GitHub PRs, GitLab MRs and Gerrit changesets) from `pull_requests_filtered`. `pull_requests_analyzed` holds one row per PR, so each PR is counted once per state. + - Includes PRs opened, merged or closed by bots. A PR merged by a bot account (e.g. Prow or a merge queue) counts as merged, since `pull_requests_analyzed` takes `mergedAt` from the merge event whoever performed it. + - `state` picks the lifecycle timestamp the count is based on: `openedAt`, `mergedAt` or `closedAt`. PRs without that timestamp are not counted. + - **When `granularity` is NOT provided, returns a single total count** across the entire time range. + - **When `granularity` is provided, returns time-series data** aggregated by period (daily, weekly, monthly, quarterly, yearly), using `generate_timeseries` so periods without PRs return 0. + - Parameters: + - `project`: Project slug (e.g., 'k8s', 'tensorflow') - inherited from `segments_filtered`. Required unless `collectionSlug` is given. + - `collectionSlug`: Collection slug (e.g., 'cncf') - inherited from `segments_filtered_by_collection`. Required unless `project` is given. + - `repos`: Optional array of repository URLs for filtering (e.g., ['https://github.com/kubernetes/kubernetes']) + - `state`: Optional string, one of 'opened' (default), 'merged' or 'closed' + - `startDate`: Optional DateTime filter for the `state` timestamp after this value (e.g., '2024-01-01 00:00:00') + - `endDate`: Optional DateTime filter for the `state` timestamp before this value (e.g., '2024-12-31 23:59:59') + - `platform`: Optional string filter for source platform (e.g., 'github', 'gitlab', 'gerrit') + - `granularity`: Optional string for time aggregation ('daily', 'weekly', 'monthly', 'quarterly', 'yearly') + - Response: + - Without granularity: `pullRequestCount` (total count) + - With granularity: `startDate`, `endDate`, and `pullRequestCount` for each time period + +TAGS "Development metrics", "Widget", "Pull requests" + +NODE pull_requests_by_state +SQL > + % + SELECT + prf.id, + {% if String( + state, + 'opened', + description="One of 'opened', 'merged' or 'closed'", + required=False, + ) == 'merged' %} prf.mergedAt + {% elif String( + state, + 'opened', + description="One of 'opened', 'merged' or 'closed'", + required=False, + ) == 'closed' %} prf.closedAt + {% else %} prf.openedAt + {% end %} AS eventAt + FROM pull_requests_filtered prf + WHERE + isNotNull(eventAt) + {% if defined(platform) %} + AND prf.platform + = {{ + String( + platform, + description="Filter by platform (e.g., 'gerrit', 'github', 'gitlab')", + required=False, + ) + }} + {% end %} + {% if defined(startDate) %} + AND eventAt + > {{ DateTime(startDate, description="Filter state timestamp after", required=False) }} + {% end %} + {% if defined(endDate) %} + AND eventAt + < {{ DateTime(endDate, description="Filter state timestamp before", required=False) }} + {% end %} + +NODE timeseries_generation_for_pull_requests_count +SQL > + % + {% if defined(granularity) %} + SELECT ds."startDate", ds."endDate", countIf(prs.id != '') AS "pullRequestCount" + FROM generate_timeseries ds + LEFT JOIN + pull_requests_by_state prs + ON CASE + WHEN {{ granularity }} = 'daily' + THEN toDate(prs.eventAt) + WHEN {{ granularity }} = 'weekly' + THEN toStartOfWeek(prs.eventAt) + WHEN {{ granularity }} = 'monthly' + THEN toStartOfMonth(prs.eventAt) + WHEN {{ granularity }} = 'quarterly' + THEN toStartOfQuarter(prs.eventAt) + WHEN {{ granularity }} = 'yearly' + THEN toStartOfYear(prs.eventAt) + END + = ds."startDate" + GROUP BY ds."startDate", ds."endDate" + ORDER BY ds."startDate" + {% else %} SELECT 1 + {% end %} + +NODE pull_requests_count_merged +SQL > + % + {% if not defined(granularity) %} SELECT count() AS "pullRequestCount" FROM pull_requests_by_state + {% else %} SELECT * FROM timeseries_generation_for_pull_requests_count + {% end %} From 1c194d99a371b4acd98d32ba5055e28cd4c16248 Mon Sep 17 00:00:00 2001 From: Joana Maia Date: Mon, 19 Oct 2026 15:58:52 +0100 Subject: [PATCH 2/6] docs: drop the pr duplicate rebuild steps from the lambda doc Refs linuxfoundation/lfx-cdp-insights-ops#790 Signed-off-by: Joana Maia --- services/libs/tinybird/lambda-architecture.md | 25 +------------------ 1 file changed, 1 insertion(+), 24 deletions(-) diff --git a/services/libs/tinybird/lambda-architecture.md b/services/libs/tinybird/lambda-architecture.md index e8cf4a1ad2..82def14307 100644 --- a/services/libs/tinybird/lambda-architecture.md +++ b/services/libs/tinybird/lambda-architecture.md @@ -651,35 +651,12 @@ If you interrupted a mid-run or suspect duplicates exist, either clear the datas **Symptom**: Same activity appears multiple times -**Fix** (append-mode datasources): Ensure you're filtering by `max(snapshotId)`: +**Fix**: Ensure you're filtering by `max(snapshotId)`: ```sql WHERE snapshotId = (SELECT max(snapshotId) FROM datasource_name) ``` -#### Duplicate rows in pull_requests_analyzed - -`pull_requests_analyzed` holds one row per PR and its consumers do not filter by snapshot, so a duplicated PR is counted and averaged twice in every PR metric (development widgets, health scores, leaderboards, org dashboards). Duplicates appear when a bucket of `pull_request_analysis_initial_snapshot` (append mode) is run more than once without clearing that bucket first. The hourly merger carries them forward, so they never clear on their own. - -**Check** duplicate rows per initial-snapshot bucket (the bucket key is `cityHash64(segmentId) % num_buckets`): - -```bash -tb sql "SELECT cityHash64(segmentId) % 10 AS bucket, count() AS rows, uniqExact(id) AS prs, rows - prs AS extra_rows FROM pull_requests_analyzed GROUP BY bucket ORDER BY bucket" -``` - -**Fix** a bucket with `extra_rows` well above 0 by rebuilding only that bucket (example for bucket 5). Run it on `lfx_insights_stg` first, then on `lfx_insights`, and echo the workspace back before each mutating command: - -```bash -tb workspace ls # confirm the workspace -tb pipe copy pause pull_request_analysis_snapshot_merger_copy # stop the hourly merger -tb datasource delete pull_requests_analyzed --sql-condition "cityHash64(segmentId) % 10 = 5" --wait -tb pipe copy run pull_request_analysis_initial_snapshot --param bucket_id=5 --param num_buckets=10 --mode append --wait -tb sql "SELECT count() AS rows, uniqExact(id) AS prs FROM pull_requests_analyzed WHERE cityHash64(segmentId) % 10 = 5" # rows = prs -tb pipe copy resume pull_request_analysis_snapshot_merger_copy -``` - -Run each bucket once. The initial snapshot rebuilds PRs from `activityRelations_enriched_deduplicated_bucket_union`, which the bucket mergers refresh daily at 01:30-01:38 UTC with events up to 00:00 UTC. PR events after that cut reached `pull_requests_analyzed` only through the hourly merger, so the rebuilt rows do not have them. Run the rebuild right after the daily bucket mergers finish to keep that gap to a couple of hours. While the bucket is being rebuilt its PRs are missing from PR metrics, and the PR materialized view keeps running, so a PR in that bucket that receives events during the window can get a partial row. Keep the window short. - ### Missing Data **Symptom**: Recent data not appearing From 362948130132486ccad46ae95f4ad2d7b63599b6 Mon Sep 17 00:00:00 2001 From: Joana Maia Date: Mon, 19 Oct 2026 17:16:42 +0100 Subject: [PATCH 3/6] fix: give the insights app token read access to the count pipes Refs linuxfoundation/lfx-cdp-insights-ops#790 Signed-off-by: Joana Maia --- services/libs/tinybird/pipes/issues_count.pipe | 2 ++ services/libs/tinybird/pipes/pull_requests_count.pipe | 2 ++ 2 files changed, 4 insertions(+) diff --git a/services/libs/tinybird/pipes/issues_count.pipe b/services/libs/tinybird/pipes/issues_count.pipe index 92129bd99c..58a06d20b0 100644 --- a/services/libs/tinybird/pipes/issues_count.pipe +++ b/services/libs/tinybird/pipes/issues_count.pipe @@ -20,6 +20,8 @@ DESCRIPTION > - Without granularity: `issueCount` (total count) - With granularity: `startDate`, `endDate`, and `issueCount` for each time period +TOKEN "insights-app-token" READ + TAGS "Development metrics", "Widget", "Issues" NODE issues_by_state diff --git a/services/libs/tinybird/pipes/pull_requests_count.pipe b/services/libs/tinybird/pipes/pull_requests_count.pipe index dcfa506ea3..2cf97f47ea 100644 --- a/services/libs/tinybird/pipes/pull_requests_count.pipe +++ b/services/libs/tinybird/pipes/pull_requests_count.pipe @@ -18,6 +18,8 @@ DESCRIPTION > - Without granularity: `pullRequestCount` (total count) - With granularity: `startDate`, `endDate`, and `pullRequestCount` for each time period +TOKEN "insights-app-token" READ + TAGS "Development metrics", "Widget", "Pull requests" NODE pull_requests_by_state From 0585ef5ba9a3e82564b500ad846e2a0400fefaef Mon Sep 17 00:00:00 2001 From: Joana Maia Date: Mon, 19 Oct 2026 17:16:42 +0100 Subject: [PATCH 4/6] fix: keep issue analysis hourly by adding the unmerged daily deltas Refs linuxfoundation/lfx-cdp-insights-ops#790 Signed-off-by: Joana Maia --- .../pipes/issue_analysis_copy_pipe.pipe | 77 +++++++++++++++++-- 1 file changed, 72 insertions(+), 5 deletions(-) diff --git a/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe b/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe index d05a7ddcd1..cc7e39f0de 100644 --- a/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe +++ b/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe @@ -1,11 +1,78 @@ DESCRIPTION > Compacts activities from same issue into one, keeping necessary information in a single row. Helps to serve issue-wide widgets in the development tab. - Opened and closed events come from the unfiltered `activityRelations_enriched_deduplicated_bucket_union`, so issues opened or closed - by bots (e.g. stale bots) are included, the same way `pull_requests_analyzed` includes bot activity. Issues are still limited to - enabled repositories, matching the cleaned buckets. + Opened and closed events are unfiltered, so issues opened or closed by bots (e.g. stale bots) are included, the same way + `pull_requests_analyzed` includes bot activity. Issues are still limited to enabled repositories, matching the cleaned buckets. + The unfiltered buckets are refreshed once a day, so `issue_events` adds the hourly deltas the daily bucket merge has not taken in + yet, using the same delta rule as `activityRelations_snapshot_merger_copy_0..2`. This keeps issues as fresh as the hourly + cleaned buckets. 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. +NODE issue_events_delta +DESCRIPTION > + Issue opened and closed events from the realtime MV that the daily bucket merge has not taken in yet: per bucket, rows with + snapshotId at or after that bucket's latest snapshot, keeping the most recently updated version of each activity. + +SQL > + WITH + ( + SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_bucket_0_ds + ) AS bucket0Snapshot, + ( + SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_bucket_1_ds + ) AS bucket1Snapshot, + ( + SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_bucket_2_ds + ) AS bucket2Snapshot + SELECT + activityId, + sourceId, + sourceParentId, + timestamp, + segmentId, + channel, + platform, + memberId, + organizationId, + type + FROM activityRelations_enrich_snapshot_MV_ds + WHERE + type IN ('issues-opened', 'issues-closed') + AND snapshotId >= multiIf( + cityHash64(segmentId) % 3 = 0, + bucket0Snapshot, + cityHash64(segmentId) % 3 = 1, + bucket1Snapshot, + bucket2Snapshot + ) + ORDER BY updatedAt DESC + LIMIT 1 BY segmentId, timestamp, type, platform, channel, sourceId + +NODE issue_events +DESCRIPTION > + Issue opened and closed events: the daily unfiltered buckets, with any activity that has a newer version in the delta replaced + by that version. Matches on activityId: a tuple NOT IN over every issue event is several times slower. + +SQL > + SELECT + activityId, + sourceId, + sourceParentId, + timestamp, + segmentId, + channel, + platform, + memberId, + organizationId, + type + FROM activityRelations_enriched_deduplicated_bucket_union + WHERE + type IN ('issues-opened', 'issues-closed') + AND activityId NOT IN (SELECT activityId FROM issue_events_delta) + UNION ALL + SELECT * + FROM issue_events_delta + NODE issues_opened SQL > SELECT @@ -17,7 +84,7 @@ SQL > platform, memberId, organizationId - FROM activityRelations_enriched_deduplicated_bucket_union + FROM issue_events WHERE type = 'issues-opened' AND toYear(timestamp) >= 1971 @@ -33,7 +100,7 @@ SQL > NODE issues_closed SQL > SELECT sourceParentId, MIN(timestamp) AS closedAt - FROM activityRelations_enriched_deduplicated_bucket_union + FROM issue_events WHERE type = 'issues-closed' AND sourceParentId != '' AND toYear(timestamp) >= 1971 GROUP BY sourceParentId From 29df5bf474aa3cc1e5d3a4fc09af43d5b6789aa8 Mon Sep 17 00:00:00 2001 From: Joana Maia Date: Tue, 20 Oct 2026 11:22:22 +0100 Subject: [PATCH 5/6] fix: read the whole issue mv delta so the copy pipe passes the push check Refs linuxfoundation/lfx-cdp-insights-ops#790 Signed-off-by: Joana Maia --- .../pipes/issue_analysis_copy_pipe.pipe | 77 ++++++++----------- 1 file changed, 31 insertions(+), 46 deletions(-) diff --git a/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe b/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe index cc7e39f0de..aa2e0cf011 100644 --- a/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe +++ b/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe @@ -2,28 +2,18 @@ DESCRIPTION > Compacts activities from same issue into one, keeping necessary information in a single row. Helps to serve issue-wide widgets in the development tab. Opened and closed events are unfiltered, so issues opened or closed by bots (e.g. stale bots) are included, the same way `pull_requests_analyzed` includes bot activity. Issues are still limited to enabled repositories, matching the cleaned buckets. - The unfiltered buckets are refreshed once a day, so `issue_events` adds the hourly deltas the daily bucket merge has not taken in - yet, using the same delta rule as `activityRelations_snapshot_merger_copy_0..2`. This keeps issues as fresh as the hourly - cleaned buckets. + The unfiltered buckets are refreshed once a day, so the opened and closed events also include the realtime MV + (`activityRelations_enrich_snapshot_MV_ds`), which holds the last few days of activity changes, including the ones the daily + bucket merge has not taken in yet. This keeps issues as fresh as the hourly cleaned buckets. The MV is read whole: looking up + 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. NODE issue_events_delta DESCRIPTION > - Issue opened and closed events from the realtime MV that the daily bucket merge has not taken in yet: per bucket, rows with - snapshotId at or after that bucket's latest snapshot, keeping the most recently updated version of each activity. + Issue opened and closed events from the realtime MV, keeping the most recently updated version of each activity. SQL > - WITH - ( - SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_bucket_0_ds - ) AS bucket0Snapshot, - ( - SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_bucket_1_ds - ) AS bucket1Snapshot, - ( - SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_bucket_2_ds - ) AS bucket2Snapshot SELECT activityId, sourceId, @@ -36,42 +26,25 @@ SQL > organizationId, type FROM activityRelations_enrich_snapshot_MV_ds - WHERE - type IN ('issues-opened', 'issues-closed') - AND snapshotId >= multiIf( - cityHash64(segmentId) % 3 = 0, - bucket0Snapshot, - cityHash64(segmentId) % 3 = 1, - bucket1Snapshot, - bucket2Snapshot - ) + WHERE type IN ('issues-opened', 'issues-closed') ORDER BY updatedAt DESC LIMIT 1 BY segmentId, timestamp, type, platform, channel, sourceId -NODE issue_events +NODE issues_opened_events DESCRIPTION > - Issue opened and closed events: the daily unfiltered buckets, with any activity that has a newer version in the delta replaced - by that version. Matches on activityId: a tuple NOT IN over every issue event is several times slower. + Issue opened events: the daily unfiltered buckets, with any activity that also appears in the MV replaced by its MV version. SQL > - SELECT - activityId, - sourceId, - sourceParentId, - timestamp, - segmentId, - channel, - platform, - memberId, - organizationId, - type + SELECT activityId, sourceId, timestamp, segmentId, channel, platform, memberId, organizationId FROM activityRelations_enriched_deduplicated_bucket_union WHERE - type IN ('issues-opened', 'issues-closed') - AND activityId NOT IN (SELECT activityId FROM issue_events_delta) + type = 'issues-opened' + AND activityId + NOT IN (SELECT activityId FROM issue_events_delta WHERE type = 'issues-opened') UNION ALL - SELECT * + SELECT activityId, sourceId, timestamp, segmentId, channel, platform, memberId, organizationId FROM issue_events_delta + WHERE type = 'issues-opened' NODE issues_opened SQL > @@ -84,10 +57,9 @@ SQL > platform, memberId, organizationId - FROM issue_events + FROM issues_opened_events WHERE - type = 'issues-opened' - AND toYear(timestamp) >= 1971 + toYear(timestamp) >= 1971 AND ( ( platform IN ('git', 'gerrit', 'github', 'gitlab') @@ -98,10 +70,23 @@ SQL > ) NODE issues_closed +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. + SQL > SELECT sourceParentId, MIN(timestamp) AS closedAt - FROM issue_events - WHERE type = 'issues-closed' AND sourceParentId != '' AND toYear(timestamp) >= 1971 + 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' + ) + WHERE sourceParentId != '' AND toYear(timestamp) >= 1971 GROUP BY sourceParentId NODE issues_comment_non_author From 90298060b5cb2bc06d6e6feb4c76739ae8c9dd7c Mon Sep 17 00:00:00 2001 From: Joana Maia Date: Tue, 20 Oct 2026 11:26:00 +0100 Subject: [PATCH 6/6] fix: formatting Signed-off-by: Joana Maia --- services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe b/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe index aa2e0cf011..be8ad3d9b2 100644 --- a/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe +++ b/services/libs/tinybird/pipes/issue_analysis_copy_pipe.pipe @@ -39,8 +39,7 @@ SQL > FROM activityRelations_enriched_deduplicated_bucket_union WHERE type = 'issues-opened' - AND activityId - NOT IN (SELECT activityId FROM issue_events_delta 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