diff --git a/services/libs/tinybird/README.md b/services/libs/tinybird/README.md index b0f8bcc622..d015832333 100644 --- a/services/libs/tinybird/README.md +++ b/services/libs/tinybird/README.md @@ -32,7 +32,7 @@ See [dataflow](./dataflow.md) for a visual diagram showing how data flows from C We use **two parallel architectures** to process activityRelations data: ### Lambda Architecture -1) Deduplicates activityRelations without any filtering. Mainly consumed in CDP, and monitoring pipes. Output: `activityRelations_enriched_deduplicated_ds` +1) Deduplicates activityRelations without any filtering. Mainly consumed in CDP, and monitoring pipes. Output: `activityRelations_enriched_deduplicated_bucket_0-2_ds`, queried via `activityRelations_enriched_deduplicated_bucket_union` 2) Used for ingesting pull request event data and merging with existing events. Output: `pull_requests_analyzed` @@ -59,20 +59,20 @@ The following table compares the two parallel architectures processing activityR | Aspect | Lambda Architecture | Bucketing Architecture | |--------|---------------------|------------------------| | **Primary Use Case** | Pull requests, CDP, monitoring, member management | Insights API queries | -| **Output Datasource** | pull_requests_analyzed, activityRelations_enriched_deduplicated_ds | activityRelations_deduplicated_cleaned_bucket_0-9_ds | +| **Output Datasource** | pull_requests_analyzed, activityRelations_enriched_deduplicated_bucket_0-2_ds | activityRelations_deduplicated_cleaned_bucket_0-9_ds | | **Data Filtering** | UNFILTERED (includes bots, all repos) | FILTERED (valid members, enabled repos) | -| **Partitioning Strategy** | Single datasource, snapshot-based | 10 parallel buckets, hash-based | -| **Copy Mode** | Append (creates new snapshots) | Replace (hourly full refresh) | -| **Query Pattern** | Filter by max(snapshotId) | Union all buckets or route to specific bucket | -| **TTL** | 6 hours (keeps ~6 snapshots) | No TTL on buckets (replace mode) | -| **Scalability** | Vertical (single large datasource) | Horizontal (add more buckets) | +| **Partitioning Strategy** | 3 parallel buckets, hash-based | 10 parallel buckets, hash-based | +| **Copy Mode** | Replace (daily per-bucket incremental merge) | Replace (hourly full refresh) | +| **Query Pattern** | Union all buckets (no snapshot filter) | Union all buckets or route to specific bucket | +| **TTL** | No TTL on buckets (replace mode) | No TTL on buckets (replace mode) | +| **Scalability** | Horizontal (add more buckets) | Horizontal (add more buckets) | | **Dependencies** | Single-table triggers work well | Multi-table dependencies (members, repos) | **Which activityRelations output to use:** - Use Bucketing Architecture output (`activityRelations_deduplicated_cleaned_bucket_*_ds`) for: Insights API, project-specific analytics, filtered queries - since each bucket contains a subset of project data, main use-case is project-specific widgets -- **Use Lambda Architecture output** (`activityRelations_enriched_deduplicated_ds`) for: CDP operations, monitoring, any use case requiring complete unfiltered data, where we can not use the buckets +- **Use Lambda Architecture output** (`activityRelations_enriched_deduplicated_bucket_union`) for: CDP operations, monitoring, any use case requiring complete unfiltered data, where we can not use the filtered buckets ## Making changes to resources 1. Install the **tb client** for classic tinybird diff --git a/services/libs/tinybird/datasources/activityRelations_enrich_snapshot_MV_ds.datasource b/services/libs/tinybird/datasources/activityRelations_enrich_snapshot_MV_ds.datasource index 83cd63c922..2bb3be2381 100644 --- a/services/libs/tinybird/datasources/activityRelations_enrich_snapshot_MV_ds.datasource +++ b/services/libs/tinybird/datasources/activityRelations_enrich_snapshot_MV_ds.datasource @@ -31,4 +31,4 @@ SCHEMA > ENGINE ReplacingMergeTree ENGINE_PARTITION_KEY toYYYYMM(snapshotId) ENGINE_SORTING_KEY snapshotId, segmentId, timestamp, type, platform, channel, sourceId -ENGINE_TTL toDateTime(snapshotId) + toIntervalDay(1) +ENGINE_TTL toDateTime(snapshotId) + toIntervalDay(3) diff --git a/services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_ds.datasource b/services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_bucket_0_ds.datasource similarity index 87% rename from services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_ds.datasource rename to services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_bucket_0_ds.datasource index 24547e43ef..3ff076b875 100644 --- a/services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_ds.datasource +++ b/services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_bucket_0_ds.datasource @@ -1,5 +1,3 @@ -TOKEN "delete_fork_repos_activities_script" READ - SCHEMA > `activityId` String, `conversationId` String, @@ -31,6 +29,5 @@ SCHEMA > `snapshotId` DateTime ENGINE MergeTree -ENGINE_PARTITION_KEY snapshotId +ENGINE_PARTITION_KEY toYear(timestamp) ENGINE_SORTING_KEY segmentId, timestamp, type, platform, memberId, organizationId -ENGINE_TTL toDateTime(snapshotId) + toIntervalDay(2) diff --git a/services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_bucket_1_ds.datasource b/services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_bucket_1_ds.datasource new file mode 100644 index 0000000000..3ff076b875 --- /dev/null +++ b/services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_bucket_1_ds.datasource @@ -0,0 +1,33 @@ +SCHEMA > + `activityId` String, + `conversationId` String, + `createdAt` DateTime64(3), + `updatedAt` DateTime64(3), + `memberId` String, + `objectMemberId` String, + `objectMemberUsername` String, + `organizationId` String, + `parentId` String, + `platform` LowCardinality(String), + `segmentId` String, + `username` String, + `sourceId` String, + `type` LowCardinality(String), + `timestamp` DateTime64(3), + `sourceParentId` String, + `channel` String, + `sentimentScore` Int8, + `gitInsertions` UInt32, + `gitDeletions` UInt32, + `score` Int8, + `isContribution` UInt8, + `pullRequestReviewState` LowCardinality(String), + `gitChangedLines` UInt64, + `gitChangedLinesBucket` String, + `organizationCountryCode` LowCardinality(String), + `organizationName` String, + `snapshotId` DateTime + +ENGINE MergeTree +ENGINE_PARTITION_KEY toYear(timestamp) +ENGINE_SORTING_KEY segmentId, timestamp, type, platform, memberId, organizationId diff --git a/services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_bucket_2_ds.datasource b/services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_bucket_2_ds.datasource new file mode 100644 index 0000000000..3ff076b875 --- /dev/null +++ b/services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_bucket_2_ds.datasource @@ -0,0 +1,33 @@ +SCHEMA > + `activityId` String, + `conversationId` String, + `createdAt` DateTime64(3), + `updatedAt` DateTime64(3), + `memberId` String, + `objectMemberId` String, + `objectMemberUsername` String, + `organizationId` String, + `parentId` String, + `platform` LowCardinality(String), + `segmentId` String, + `username` String, + `sourceId` String, + `type` LowCardinality(String), + `timestamp` DateTime64(3), + `sourceParentId` String, + `channel` String, + `sentimentScore` Int8, + `gitInsertions` UInt32, + `gitDeletions` UInt32, + `score` Int8, + `isContribution` UInt8, + `pullRequestReviewState` LowCardinality(String), + `gitChangedLines` UInt64, + `gitChangedLinesBucket` String, + `organizationCountryCode` LowCardinality(String), + `organizationName` String, + `snapshotId` DateTime + +ENGINE MergeTree +ENGINE_PARTITION_KEY toYear(timestamp) +ENGINE_SORTING_KEY segmentId, timestamp, type, platform, memberId, organizationId diff --git a/services/libs/tinybird/lambda-architecture.md b/services/libs/tinybird/lambda-architecture.md index 2ed5535a3b..7a25fef1d1 100644 --- a/services/libs/tinybird/lambda-architecture.md +++ b/services/libs/tinybird/lambda-architecture.md @@ -4,7 +4,7 @@ - [Overview](#overview) - [Main Activity Relations Pipeline](#main-activity-relations-pipeline-lambda-architecture---produces-unfiltered-data) -- [Downstream Consumers](#downstream-consumers-of-activityrelations_enriched_deduplicated_ds) +- [Downstream Consumers](#downstream-consumers-of-activityrelations_enriched_deduplicated_bucket_union) - [Timeline View](#timeline-view-hourly-execution) - [Snapshot-Based Deduplication](#snapshot-based-deduplication-strategy) - [Pull Requests Pipeline](#pull-requests-pipeline-primary-lambda-architecture-use-case) @@ -115,51 +115,65 @@ This document explains the **Lambda Architecture** implementation used in our Ti (see below) (future: issues, commits, etc.) -[4] Merge Layer - Snapshot Merger Copy Pipe +[4] Merge Layer - Bucketed Snapshot Merger Copy Pipes ┌────────────────────────────────────────┐ │ activityRelations_snapshot_ │ - │ merger_copy │ - │ (TYPE: COPY, every day at 1 AM) │ + │ merger_copy_0 .. _2 │ + │ (TYPE: COPY, daily, staggered) │ └────────────────────────────────────────┘ ↓ - What it does: - ├─ Fetches: NEW data from MV result datasource (latest snapshotId + 1 hour) - ├─ Fetches: OLD data from serving layer (current max snapshotId) - ├─ Merges: UNION ALL → creates new snapshot - ├─ Mode: append - └─ Schedule: 0 1 * * * (every day at 1 AM UTC) - - -[5] Final Datasource + What each bucket pipe does (bucketing key: cityHash64(segmentId) % 3): + ├─ Fetches: NEW deltas for its bucket from MV_ds (snapshotId >= bucket's last + │ snapshot — inclusive, so rows arriving late for an already-merged day are + │ replayed on the next run; the dedup below makes the replay idempotent) + ├─ Collapses multi-day deltas: ORDER BY updatedAt DESC LIMIT 1 BY dedup key (no FINAL) + ├─ Fetches: OLD data from its own bucket datasource (carry-forward, NOT IN delta keys) + ├─ Merges: UNION ALL → one fresh snapshot for the bucket + ├─ Mode: replace (atomic — a failed run leaves the previous bucket data intact) + └─ Schedule: daily, staggered at 01:30/01:34/01:38 UTC + + Why replace-mode buckets (vs the old single append pipe + TTL): + ├─ Old: one 380GB+ copy per day, running at 40-75% of the 3600s copy timeout, + │ with a memory-heavy NOT IN over the full realtime snapshot + ├─ Old: TTL (snapshotId + 2 days) expired the last good snapshot one hour BEFORE + │ the next scheduled run — a single failed run emptied the datasource, and + │ the empty state self-perpetuated (max(snapshotId) = epoch matches no MV + │ rows), requiring a manual full re-bootstrap + └─ New: 1/3rd-sized copies far from timeout/memory limits; a failure loses + nothing; a bucket self-heals for up to ~2 missed days (MV delta + retention is 3 days); longer gaps → run that bucket's initial snapshot + + +[5] Final Datasources + Union ┌────────────────────────────────────────┐ │ activityRelations_enriched │ - │ _deduplicated_ds │ + │ _deduplicated_bucket_0_ds .. _2_ds │ │ (UNFILTERED - used by CDP, monitoring)│ └────────────────────────────────────────┘ ↓ - • TYPE: MergeTree - • Partitioned by: snapshotId + • TYPE: MergeTree (each bucket) + • Partitioned by: toYear(timestamp) • Sorting Key: segmentId, timestamp, type, platform, memberId, organizationId - • TTL: snapshotId + 6 hours (keeps last ~6 snapshots) - • Query Pattern: WHERE snapshotId = (SELECT max(snapshotId) FROM ...) - - Data structure: - ├─ Record A, snapshotId: 14:00 - ├─ Record A, snapshotId: 15:00 (same record, updated snapshot) - ├─ Record A, snapshotId: 16:00 (same record, latest snapshot) - ├─ Record B, snapshotId: 15:00 - └─ Record B, snapshotId: 16:00 - - ⚠️ Multiple copies exist physically - ✅ Deduplication happens at query time via max(snapshotId) filter + • No TTL — replace-mode copies keep exactly ONE snapshot per bucket + • Query via: activityRelations_enriched_deduplicated_bucket_union (UNION ALL of 3) + • Bucket count: 3 for now — increasing it later remaps cityHash64(segmentId) % N, + so add the new datasources/pipes, update the union, and re-run ALL initial + snapshot pipes afterwards (replace mode makes the re-bootstrap safe) + + ⚠️ Do NOT filter by max(snapshotId) across the union: buckets may legitimately + sit on different snapshot days after a partial failure, and a global max + filter would silently drop the lagging buckets. Each bucket already holds + exactly one snapshot, so no snapshot filtering is needed at all. ``` --- -## Downstream Consumers of activityRelations_enriched_deduplicated_ds +## Downstream Consumers of activityRelations_enriched_deduplicated_bucket_union -The unfiltered lambda architecture output is consumed by: +The unfiltered lambda architecture output is consumed through the +`activityRelations_enriched_deduplicated_bucket_union` pipe (never by querying +bucket datasources directly) by: ### 1. Pull Requests Pipeline (Primary consumer - detailed section below) - Analyzes PR lifecycle: opened → reviewed → approved → merged @@ -253,7 +267,11 @@ Instead of using FINAL in copy pipes or query time, our approach uses **snapshot 1. **Physical Storage**: Multiple copies of the same record exist across different snapshots 2. **Logical View**: Queries filter by `max(snapshotId)` to see only the latest version -3. **Automatic Cleanup**: TTL removes old snapshots (keeps last 6 hours) +3. **Automatic Cleanup**: TTL removes old snapshots + +> **Note**: This applies to append-mode pipelines. The unfiltered activityRelations +> serving layer now uses replace-mode bucket copies instead — each bucket holds exactly +> one snapshot, so neither TTL cleanup nor query-time snapshot filtering is involved. ### Example @@ -267,10 +285,11 @@ Instead of using FINAL in copy pipes or query time, our approach uses **snapshot │ 2 │ some_value │ 2024-01-01 15:00 │ └─────────┴─────────────┴──────────────────┘ --- Query with snapshot filter: +-- Query with snapshot filter (generic pattern — substitute your append-mode, +-- snapshot-stamped datasource): SELECT * -FROM activityRelations_enriched_deduplicated_ds -WHERE snapshotId = (SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_ds) +FROM your_snapshot_ds +WHERE snapshotId = (SELECT max(snapshotId) FROM your_snapshot_ds) -- Result (deduplicated logical view): ┌─────────┬─────────────┬──────────────────┐ @@ -493,24 +512,25 @@ Initial snapshot pipes: ### Examples -#### 1. activityRelations_enrich_initial_snapshot +#### 1. activityRelations_enrich_initial_snapshot_0 .. _2 (per bucket) ``` -File: activityRelations_enrich_initial_snapshot.pipe +Files: activityRelations_enrich_initial_snapshot_0.pipe .. _2.pipe TYPE: COPY COPY_MODE: replace COPY_SCHEDULE: @on-demand -TARGET_DATASOURCE: activityRelations_enriched_deduplicated_ds +TARGET_DATASOURCE: activityRelations_enriched_deduplicated_bucket__ds -What it does: -├─ Reads raw activityRelations (base table) +What each one does: +├─ Reads raw activityRelations (base table), filtered to cityHash64(segmentId) % 3 = N ├─ Enriches with country codes, org names, gitChangedLines buckets -├─ Filters valid members, repos, segments -├─ Creates snapshotId: toStartOfInterval(now(), INTERVAL 1 hour) -└─ Writes initial baseline to serving datasource +├─ Creates snapshotId: toStartOfInterval(now(), INTERVAL 1 day) +└─ Replaces bucket N of the serving layer -Usage: Run once to bootstrap activityRelations serving layer +Usage: Run all 3 to bootstrap the serving layer, or a single one to rebuild a +bucket that fell behind further than the MV delta retention (3 days). Replace +mode makes them safe to re-run. ``` #### 2. pull_request_analysis_initial_snapshot @@ -564,21 +584,21 @@ Usage: Run once to bootstrap segment-level metrics **How to run:** ```bash # Via Tinybird CLI (assuming you have tb CLI configured) -tb pipe copy run activityRelations_enrich_initial_snapshot --wait +for N in 0 1 2; do tb pipe copy run activityRelations_enrich_initial_snapshot_$N --wait; done tb pipe copy run pull_request_analysis_initial_snapshot --wait tb pipe copy run segmentId_aggregates_initial_snapshot --wait ``` ### Comparison: Initial vs Merger Copy Pipes -| Aspect | Initial Snapshot | Merger Copy Pipe | -|--------|------------------|------------------| -| **Schedule** | @on-demand (manual) | Hourly (10 * * * * or 0 * * * *) | -| **Mode** | replace (overwrites all) | append (adds new snapshot) | -| **Purpose** | Bootstrap/reset | Incremental updates | -| **Source** | Base tables or latest snapshot | MV output + existing serving data | -| **Frequency** | Once (or rarely) | Continuous (hourly) | -| **Snapshot Strategy** | Create first snapshot | Create new snapshots, merge with old | +| Aspect | Initial Snapshot | Bucket Merger (activityRelations) | PR Merger | +|--------|------------------|-----------------------------------|-----------| +| **Schedule** | @on-demand (manual) | Daily (01:30/01:34/01:38 UTC) | Hourly (0 * * * *) | +| **Mode** | replace (overwrites all) | replace (atomic per-bucket swap) | replace | +| **Purpose** | Bootstrap/reset | Incremental merge of MV deltas | Incremental merge of PR events | +| **Source** | Base tables | MV output + own bucket (carry-forward) | MV output + own target | +| **Frequency** | Once (or rarely) | Continuous (daily) | Continuous (hourly) | +| **Snapshot Strategy** | Create first snapshot | One snapshot per bucket, re-stamped each run | Single snapshot, replaced each run | ## Troubleshooting @@ -606,11 +626,16 @@ WHERE snapshotId = (SELECT max(snapshotId) FROM datasource_name) **Symptom**: Recent data not appearing **Possible Causes**: -1. **TTL deleted it**: Old snapshots auto-deleted (6-hour window for activities, 1-day for MV output) +1. **TTL deleted MV deltas**: MV output retention is 3 days — a bucket that missed more + than 2 daily runs can no longer catch up incrementally and needs its initial + snapshot pipe re-run (the bucket keeps serving its last good data meanwhile) 2. **MV behind**: Check latest `snapshotId` in MV_ds -3. **Copy pipe not run**: Runs at :10 past the hour (activities) or :00 (PRs) +3. **Copy pipe not run**: Bucket mergers run daily at 01:30-01:38 UTC; PR merger hourly at :00 4. **Filtering**: Check member/repo/segment filters in MV +5. **A bucket is lagging**: Compare `max(snapshotId)` per bucket datasource — with + replace-mode buckets a failed run leaves old (not missing) data, so look for + buckets whose snapshotId is older than the others ## TLDR -This architecture moves heavyweight operations (enrichment, filtering, joins) to ingestion time via Materialized Views. Copy pipes only handle snapshot merging (UNION ALL of new + old data). Deduplication happens at query time by filtering `WHERE snapshotId = max(snapshotId)`, not in copy pipes or with FINAL. This makes hourly copies fast and lightweight, while queries remain fast by reading pre-processed snapshots. \ No newline at end of file +This architecture moves heavyweight operations (enrichment, filtering, joins) to ingestion time via Materialized Views. Copy pipes only handle snapshot merging (UNION ALL of new + old data). For the unfiltered activityRelations serving layer the merge is split across 3 hash buckets with replace-mode copies — each bucket holds exactly one snapshot, so a failed copy leaves the previous data intact and queries need no snapshot filter (read the bucket union). Append-mode pipelines (e.g. historical patterns described above) deduplicate at query time by filtering `WHERE snapshotId = max(snapshotId)`, not in copy pipes or with FINAL. This keeps scheduled copies fast and lightweight, while queries remain fast by reading pre-processed snapshots. \ No newline at end of file diff --git a/services/libs/tinybird/pipes/activities_daily_counts.pipe b/services/libs/tinybird/pipes/activities_daily_counts.pipe index 74b22da688..236d967c7f 100644 --- a/services/libs/tinybird/pipes/activities_daily_counts.pipe +++ b/services/libs/tinybird/pipes/activities_daily_counts.pipe @@ -13,7 +13,7 @@ DESCRIPTION > - `count`: number of unique `activityId` per day, after filters. - Performance: - With ~1–2k rows/day and ≤30 days, `uniqExact` is typically fine. `uniqCombined` offers lower latency with negligible error for this scale. - - Filters are pushed into `PREWHERE` to minimize I/O (segment/time first). + - Filters push down into each bucket of the union via `WHERE` (segment/time first). TAGS "Activity metrics" @@ -21,16 +21,13 @@ NODE daily_counts SQL > % SELECT toStartOfDay(timestamp) AS date, uniq(activityId) AS count - FROM - activityRelations_enriched_deduplicated_ds - PREWHERE + FROM activityRelations_enriched_deduplicated_bucket_union + WHERE "segmentId" IN {{ Array(segmentIds, 'String', required=True, description="Segment IDs") }} {% if defined(after) %} AND timestamp >= parseDateTimeBestEffort({{ String(after) }}) {% end %} {% if defined(before) %} AND timestamp <= parseDateTimeBestEffort({{ String(before) }}) {% end %} - WHERE - snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) {% if defined(platform) %} AND platform = {{ String(platform) }} {% end %} GROUP BY date ORDER BY date ASC diff --git a/services/libs/tinybird/pipes/activities_relations_filtered.pipe b/services/libs/tinybird/pipes/activities_relations_filtered.pipe index ba0b6cde73..100a290115 100644 --- a/services/libs/tinybird/pipes/activities_relations_filtered.pipe +++ b/services/libs/tinybird/pipes/activities_relations_filtered.pipe @@ -44,10 +44,9 @@ SQL > ar.sourceParentId AS sourceParentId, ar.timestamp AS timestamp, ar.type AS type - FROM activityRelations_enriched_deduplicated_ds AS ar + FROM activityRelations_enriched_deduplicated_bucket_union AS ar WHERE - snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) - AND (length(segments_arr) = 0 OR ar.segmentId IN segments_arr) + (length(segments_arr) = 0 OR ar.segmentId IN segments_arr) {% if defined(startDate) %} AND ar.timestamp > parseDateTimeBestEffort({{ String(startDate) }}) {% end %} diff --git a/services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot.pipe b/services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot_0.pipe similarity index 73% rename from services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot.pipe rename to services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot_0.pipe index 438ea83625..277cedcec3 100644 --- a/services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot.pipe +++ b/services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot_0.pipe @@ -1,8 +1,14 @@ +DESCRIPTION > + Bootstraps bucket 0/3 (cityHash64(segmentId) % 3 = 0) of the unfiltered enriched + activityRelations serving layer from the raw activityRelations datasource. Run on demand: + on first deploy, or to rebuild a bucket after an outage longer than the MV delta + retention (3 days). Replace mode — safe to re-run. + NODE country_mapping_array SQL > SELECT groupArray((country, country_code, timezone_offset)) AS country_data FROM country_mapping_ds -NODE activityRelations_deduplicated_cleaned_denormalized +NODE activityRelations_deduplicated_enriched_bucket_0 SQL > WITH upperUTF8(o.location) AS search_u, @@ -31,8 +37,9 @@ SQL > toStartOfInterval(now(), INTERVAL 1 day) as snapshotId from activityRelations final left join organizations o final on o.id = activityRelations.organizationId + where cityHash64(activityRelations.segmentId) % 3 = 0 TYPE COPY -TARGET_DATASOURCE activityRelations_enriched_deduplicated_ds +TARGET_DATASOURCE activityRelations_enriched_deduplicated_bucket_0_ds COPY_MODE replace COPY_SCHEDULE @on-demand diff --git a/services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot_1.pipe b/services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot_1.pipe new file mode 100644 index 0000000000..3c134c9694 --- /dev/null +++ b/services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot_1.pipe @@ -0,0 +1,45 @@ +DESCRIPTION > + Bootstraps bucket 1/3 (cityHash64(segmentId) % 3 = 1) of the unfiltered enriched + activityRelations serving layer from the raw activityRelations datasource. Run on demand: + on first deploy, or to rebuild a bucket after an outage longer than the MV delta + retention (3 days). Replace mode — safe to re-run. + +NODE country_mapping_array +SQL > + SELECT groupArray((country, country_code, timezone_offset)) AS country_data FROM country_mapping_ds + +NODE activityRelations_deduplicated_enriched_bucket_1 +SQL > + WITH + upperUTF8(o.location) AS search_u, + (SELECT arrayMap(x -> upperUTF8(x .1), country_data) FROM country_mapping_array) AS names_u, + (SELECT country_data FROM country_mapping_array) AS mapping, + multiSearchFirstIndexUTF8(search_u, names_u) AS idx, + if(idx = 0, ('Unknown', 'XX', 0), mapping[idx]) AS country_data + SELECT + activityRelations.*, + (gitInsertions + gitDeletions) as gitChangedLines, + case + when gitChangedLines > 0 and gitChangedLines < 10 + then '1-9' + when gitChangedLines > 9 and gitChangedLines < 60 + then '10-59' + when gitChangedLines > 59 and gitChangedLines < 100 + then '60-99' + when gitChangedLines > 99 and gitChangedLines < 500 + then '100-499' + when gitChangedLines > 499 + then '500+' + else '' + end as "gitChangedLinesBucket", + CAST(country_data .2 AS LowCardinality(String)) AS organizationCountryCode, + o.displayName as "organizationName", + toStartOfInterval(now(), INTERVAL 1 day) as snapshotId + from activityRelations final + left join organizations o final on o.id = activityRelations.organizationId + where cityHash64(activityRelations.segmentId) % 3 = 1 + +TYPE COPY +TARGET_DATASOURCE activityRelations_enriched_deduplicated_bucket_1_ds +COPY_MODE replace +COPY_SCHEDULE @on-demand diff --git a/services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot_2.pipe b/services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot_2.pipe new file mode 100644 index 0000000000..db5561b747 --- /dev/null +++ b/services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot_2.pipe @@ -0,0 +1,45 @@ +DESCRIPTION > + Bootstraps bucket 2/3 (cityHash64(segmentId) % 3 = 2) of the unfiltered enriched + activityRelations serving layer from the raw activityRelations datasource. Run on demand: + on first deploy, or to rebuild a bucket after an outage longer than the MV delta + retention (3 days). Replace mode — safe to re-run. + +NODE country_mapping_array +SQL > + SELECT groupArray((country, country_code, timezone_offset)) AS country_data FROM country_mapping_ds + +NODE activityRelations_deduplicated_enriched_bucket_2 +SQL > + WITH + upperUTF8(o.location) AS search_u, + (SELECT arrayMap(x -> upperUTF8(x .1), country_data) FROM country_mapping_array) AS names_u, + (SELECT country_data FROM country_mapping_array) AS mapping, + multiSearchFirstIndexUTF8(search_u, names_u) AS idx, + if(idx = 0, ('Unknown', 'XX', 0), mapping[idx]) AS country_data + SELECT + activityRelations.*, + (gitInsertions + gitDeletions) as gitChangedLines, + case + when gitChangedLines > 0 and gitChangedLines < 10 + then '1-9' + when gitChangedLines > 9 and gitChangedLines < 60 + then '10-59' + when gitChangedLines > 59 and gitChangedLines < 100 + then '60-99' + when gitChangedLines > 99 and gitChangedLines < 500 + then '100-499' + when gitChangedLines > 499 + then '500+' + else '' + end as "gitChangedLinesBucket", + CAST(country_data .2 AS LowCardinality(String)) AS organizationCountryCode, + o.displayName as "organizationName", + toStartOfInterval(now(), INTERVAL 1 day) as snapshotId + from activityRelations final + left join organizations o final on o.id = activityRelations.organizationId + where cityHash64(activityRelations.segmentId) % 3 = 2 + +TYPE COPY +TARGET_DATASOURCE activityRelations_enriched_deduplicated_bucket_2_ds +COPY_MODE replace +COPY_SCHEDULE @on-demand diff --git a/services/libs/tinybird/pipes/activityRelations_enriched_deduplicated_bucket_union.pipe b/services/libs/tinybird/pipes/activityRelations_enriched_deduplicated_bucket_union.pipe new file mode 100644 index 0000000000..0cae88b901 --- /dev/null +++ b/services/libs/tinybird/pipes/activityRelations_enriched_deduplicated_bucket_union.pipe @@ -0,0 +1,21 @@ +DESCRIPTION > + Unions the 3 unfiltered enriched activityRelations buckets. This replaces direct reads of + activityRelations_enriched_deduplicated_ds. Do NOT filter by max(snapshotId) here or in + consumers: each bucket holds exactly one snapshot (replace-mode copies), and buckets may + legitimately sit on different snapshot days after a partial failure — a global max filter + would silently drop lagging buckets. + Also: never run a bare count() over this union — it cannot use the trivial-count + optimization and times out. For whole-table counts, count() per bucket datasource and + sum the results (see monitoring_entities for the pattern). Filtered/column-reading + queries push down into each bucket normally and perform fine. + +NODE activityRelations_enriched_deduplicated_BUCKET_UNION_0 +SQL > + SELECT * + FROM activityRelations_enriched_deduplicated_bucket_0_ds + UNION ALL + SELECT * + FROM activityRelations_enriched_deduplicated_bucket_1_ds + UNION ALL + SELECT * + FROM activityRelations_enriched_deduplicated_bucket_2_ds diff --git a/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy.pipe b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy.pipe deleted file mode 100644 index 9986be2413..0000000000 --- a/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy.pipe +++ /dev/null @@ -1,35 +0,0 @@ -DESCRIPTION > - Merges realtime snapshots (coming from MV) and historical snapshot (coming from datasource) to create another snapshot - -NODE realtime_snapshot -SQL > - WITH (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) as maxSnapshotId - SELECT activityRelations_enrich_snapshot_MV_ds.* - FROM activityRelations_enrich_snapshot_MV_ds FINAL - where activityRelations_enrich_snapshot_MV_ds.snapshotId = maxSnapshotId + INTERVAL 1 day - -NODE historical_snapshot -SQL > - WITH - ( - Select max(snapshotId)::DateTime from activityRelations_enriched_deduplicated_ds - ) as maxSnapshotID - SELECT * REPLACE (maxSnapshotID::DateTime + interval 1 day AS snapshotId) - FROM activityRelations_enriched_deduplicated_ds - where - snapshotId = maxSnapshotID - and (segmentId, timestamp, type, platform, channel, sourceId) - NOT IN (SELECT segmentId, timestamp, type, platform, channel, sourceId FROM realtime_snapshot) - -NODE merged_snapshots -SQL > - select * - from realtime_snapshot - union all - select * - from historical_snapshot - -TYPE COPY -TARGET_DATASOURCE activityRelations_enriched_deduplicated_ds -COPY_MODE append -COPY_SCHEDULE 0 1 * * * diff --git a/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_0.pipe b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_0.pipe new file mode 100644 index 0000000000..948eb3fc28 --- /dev/null +++ b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_0.pipe @@ -0,0 +1,43 @@ +DESCRIPTION > + Bucket 0/3 of the activityRelations lambda merge layer (bucketing key: cityHash64(segmentId) % 3). + Merges realtime enriched deltas from activityRelations_enrich_snapshot_MV_ds with the bucket's + current data and atomically replaces the bucket datasource. Replace mode means a failed run + leaves the previous bucket data intact — there is no TTL cliff. A bucket that missed up to 2 + daily runs self-heals on the next run (MV delta retention is 3 days); for longer gaps run + activityRelations_enrich_initial_snapshot_0 on demand. + +NODE realtime_delta +SQL > + WITH + ( + SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_bucket_0_ds + ) AS lastSnapshotId + SELECT * REPLACE (toStartOfInterval(now(), INTERVAL 1 day) AS snapshotId) + FROM activityRelations_enrich_snapshot_MV_ds + WHERE + cityHash64(segmentId) % 3 = 0 + AND snapshotId >= lastSnapshotId + AND snapshotId <= toStartOfInterval(now(), INTERVAL 1 day) + ORDER BY updatedAt DESC + LIMIT 1 BY segmentId, timestamp, type, platform, channel, sourceId + +NODE historical_carryforward +SQL > + SELECT * REPLACE (toStartOfInterval(now(), INTERVAL 1 day) AS snapshotId) + FROM activityRelations_enriched_deduplicated_bucket_0_ds + WHERE + (segmentId, timestamp, type, platform, channel, sourceId) + NOT IN (SELECT segmentId, timestamp, type, platform, channel, sourceId FROM realtime_delta) + +NODE merged_snapshot +SQL > + select * + from realtime_delta + union all + select * + from historical_carryforward + +TYPE COPY +TARGET_DATASOURCE activityRelations_enriched_deduplicated_bucket_0_ds +COPY_MODE replace +COPY_SCHEDULE 30 1 * * * diff --git a/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_1.pipe b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_1.pipe new file mode 100644 index 0000000000..65cd809a03 --- /dev/null +++ b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_1.pipe @@ -0,0 +1,43 @@ +DESCRIPTION > + Bucket 1/3 of the activityRelations lambda merge layer (bucketing key: cityHash64(segmentId) % 3). + Merges realtime enriched deltas from activityRelations_enrich_snapshot_MV_ds with the bucket's + current data and atomically replaces the bucket datasource. Replace mode means a failed run + leaves the previous bucket data intact — there is no TTL cliff. A bucket that missed up to 2 + daily runs self-heals on the next run (MV delta retention is 3 days); for longer gaps run + activityRelations_enrich_initial_snapshot_1 on demand. + +NODE realtime_delta +SQL > + WITH + ( + SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_bucket_1_ds + ) AS lastSnapshotId + SELECT * REPLACE (toStartOfInterval(now(), INTERVAL 1 day) AS snapshotId) + FROM activityRelations_enrich_snapshot_MV_ds + WHERE + cityHash64(segmentId) % 3 = 1 + AND snapshotId >= lastSnapshotId + AND snapshotId <= toStartOfInterval(now(), INTERVAL 1 day) + ORDER BY updatedAt DESC + LIMIT 1 BY segmentId, timestamp, type, platform, channel, sourceId + +NODE historical_carryforward +SQL > + SELECT * REPLACE (toStartOfInterval(now(), INTERVAL 1 day) AS snapshotId) + FROM activityRelations_enriched_deduplicated_bucket_1_ds + WHERE + (segmentId, timestamp, type, platform, channel, sourceId) + NOT IN (SELECT segmentId, timestamp, type, platform, channel, sourceId FROM realtime_delta) + +NODE merged_snapshot +SQL > + select * + from realtime_delta + union all + select * + from historical_carryforward + +TYPE COPY +TARGET_DATASOURCE activityRelations_enriched_deduplicated_bucket_1_ds +COPY_MODE replace +COPY_SCHEDULE 34 1 * * * diff --git a/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_2.pipe b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_2.pipe new file mode 100644 index 0000000000..077fce0e60 --- /dev/null +++ b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_2.pipe @@ -0,0 +1,43 @@ +DESCRIPTION > + Bucket 2/3 of the activityRelations lambda merge layer (bucketing key: cityHash64(segmentId) % 3). + Merges realtime enriched deltas from activityRelations_enrich_snapshot_MV_ds with the bucket's + current data and atomically replaces the bucket datasource. Replace mode means a failed run + leaves the previous bucket data intact — there is no TTL cliff. A bucket that missed up to 2 + daily runs self-heals on the next run (MV delta retention is 3 days); for longer gaps run + activityRelations_enrich_initial_snapshot_2 on demand. + +NODE realtime_delta +SQL > + WITH + ( + SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_bucket_2_ds + ) AS lastSnapshotId + SELECT * REPLACE (toStartOfInterval(now(), INTERVAL 1 day) AS snapshotId) + FROM activityRelations_enrich_snapshot_MV_ds + WHERE + cityHash64(segmentId) % 3 = 2 + AND snapshotId >= lastSnapshotId + AND snapshotId <= toStartOfInterval(now(), INTERVAL 1 day) + ORDER BY updatedAt DESC + LIMIT 1 BY segmentId, timestamp, type, platform, channel, sourceId + +NODE historical_carryforward +SQL > + SELECT * REPLACE (toStartOfInterval(now(), INTERVAL 1 day) AS snapshotId) + FROM activityRelations_enriched_deduplicated_bucket_2_ds + WHERE + (segmentId, timestamp, type, platform, channel, sourceId) + NOT IN (SELECT segmentId, timestamp, type, platform, channel, sourceId FROM realtime_delta) + +NODE merged_snapshot +SQL > + select * + from realtime_delta + union all + select * + from historical_carryforward + +TYPE COPY +TARGET_DATASOURCE activityRelations_enriched_deduplicated_bucket_2_ds +COPY_MODE replace +COPY_SCHEDULE 38 1 * * * diff --git a/services/libs/tinybird/pipes/cdp_member_segment_aggregates_initial_snapshot.pipe b/services/libs/tinybird/pipes/cdp_member_segment_aggregates_initial_snapshot.pipe index cd6f260052..d05e81ad6f 100644 --- a/services/libs/tinybird/pipes/cdp_member_segment_aggregates_initial_snapshot.pipe +++ b/services/libs/tinybird/pipes/cdp_member_segment_aggregates_initial_snapshot.pipe @@ -11,8 +11,7 @@ SQL > avgState(sentimentScore) AS averageSentimentState, maxState(act.updatedAt) as lastActivityUpdatedAtState, max(act.updatedAt) as updatedAt - FROM activityRelations_enriched_deduplicated_ds act - WHERE snapshotId = (SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_ds) + FROM activityRelations_enriched_deduplicated_bucket_union act GROUP BY segmentId, memberId TYPE COPY diff --git a/services/libs/tinybird/pipes/cdp_organization_segment_aggregates_initial_snapshot.pipe b/services/libs/tinybird/pipes/cdp_organization_segment_aggregates_initial_snapshot.pipe index 4dfc7419cd..49e198d4d5 100644 --- a/services/libs/tinybird/pipes/cdp_organization_segment_aggregates_initial_snapshot.pipe +++ b/services/libs/tinybird/pipes/cdp_organization_segment_aggregates_initial_snapshot.pipe @@ -12,8 +12,7 @@ SQL > avgState(score) AS avgContributorEngagement, maxState(act.updatedAt) as lastActivityUpdatedAtState, max(act.updatedAt) as updatedAt - FROM activityRelations_enriched_deduplicated_ds act - WHERE snapshotId = (SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_ds) + FROM activityRelations_enriched_deduplicated_bucket_union act GROUP BY segmentId, organizationId TYPE COPY diff --git a/services/libs/tinybird/pipes/cdp_segment_metrics_copy_pipe.pipe b/services/libs/tinybird/pipes/cdp_segment_metrics_copy_pipe.pipe index 2abcb8e01b..60cea51108 100644 --- a/services/libs/tinybird/pipes/cdp_segment_metrics_copy_pipe.pipe +++ b/services/libs/tinybird/pipes/cdp_segment_metrics_copy_pipe.pipe @@ -1,9 +1,9 @@ DESCRIPTION > Daily batch job that builds segment-level aggregate states from existing member and organization segment aggregate datasources. - Activities computed from activityRelations_enriched_deduplicated_ds (latest snapshotId). - Optimized: compute only for "valid segments" early, compute latest snapshot once, - and reuse empty states as constants. + Activities computed from activityRelations_enriched_deduplicated_bucket_union + (each bucket holds exactly one snapshot, so no snapshot filtering is needed). + Optimized: compute only for "valid segments" early and reuse empty states as constants. NODE validSegments SQL > @@ -26,16 +26,8 @@ SQL > countState( if(ar.timestamp >= now() - INTERVAL 30 DAY, CAST(1 AS UInt8), NULL) ) AS activitiesLast30DaysState - FROM activityRelations_enriched_deduplicated_ds AS ar - WHERE - snapshotId = ( - SELECT snapshotId - FROM activityRelations_enriched_deduplicated_ds - ORDER BY snapshotId DESC - LIMIT 1 - ) - AND segmentId IS NOT NULL - AND segmentId != '' + FROM activityRelations_enriched_deduplicated_bucket_union AS ar + WHERE segmentId IS NOT NULL AND segmentId != '' GROUP BY ar.segmentId NODE memberPerSegment diff --git a/services/libs/tinybird/pipes/cdp_segment_metrics_total_sink.pipe b/services/libs/tinybird/pipes/cdp_segment_metrics_total_sink.pipe index 2625da3d07..425ffaa4cd 100644 --- a/services/libs/tinybird/pipes/cdp_segment_metrics_total_sink.pipe +++ b/services/libs/tinybird/pipes/cdp_segment_metrics_total_sink.pipe @@ -9,11 +9,8 @@ SQL > SELECT count() AS activitiesTotal, countIf(timestamp >= now() - INTERVAL 30 DAY) AS activitiesLast30Days - FROM activityRelations_enriched_deduplicated_ds - WHERE - snapshotId = (SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_ds) - AND segmentId IS NOT NULL - AND segmentId != '' + FROM activityRelations_enriched_deduplicated_bucket_union + WHERE segmentId IS NOT NULL AND segmentId != '' NODE cdpDashboardMetricsTotal SQL > diff --git a/services/libs/tinybird/pipes/monitoring_entities.pipe b/services/libs/tinybird/pipes/monitoring_entities.pipe index 4509a6ca1a..3ad7a08fd0 100644 --- a/services/libs/tinybird/pipes/monitoring_entities.pipe +++ b/services/libs/tinybird/pipes/monitoring_entities.pipe @@ -2,14 +2,25 @@ TAGS "Monitoring" NODE activityRelations_total SQL > + -- count() per bucket (metadata-fast), then sum: a bare count() over the bucket + -- union subquery cannot use the trivial-count optimization and times out SELECT 'rows_total' AS name, 'Total rows in table' AS help, 'gauge' AS type, map('table', 'activityRelations') AS labels, - count() AS value - FROM activityRelations_enriched_deduplicated_ds - where snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) + sum(c) AS value + FROM + ( + SELECT count() AS c + FROM activityRelations_enriched_deduplicated_bucket_0_ds + UNION ALL + SELECT count() + FROM activityRelations_enriched_deduplicated_bucket_1_ds + UNION ALL + SELECT count() + FROM activityRelations_enriched_deduplicated_bucket_2_ds + ) NODE members_total SQL > diff --git a/services/libs/tinybird/pipes/pull_request_analysis_initial_snapshot.pipe b/services/libs/tinybird/pipes/pull_request_analysis_initial_snapshot.pipe index b5ffe50676..fdc2d3116f 100644 --- a/services/libs/tinybird/pipes/pull_request_analysis_initial_snapshot.pipe +++ b/services/libs/tinybird/pipes/pull_request_analysis_initial_snapshot.pipe @@ -17,12 +17,9 @@ SQL > organizationId, platform, updatedAt - FROM activityRelations_enriched_deduplicated_ds + FROM activityRelations_enriched_deduplicated_bucket_union WHERE - snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) - AND ( - type = 'pull_request-opened' OR type = 'merge_request-opened' OR type = 'changeset-created' - ) + (type = 'pull_request-opened' OR type = 'merge_request-opened' OR type = 'changeset-created') {% if defined(bucket_id) %} AND cityHash64(segmentId) % {{ UInt8(num_buckets, 5, description="Total number of buckets", required=False) }} @@ -40,10 +37,9 @@ NODE pull_request_first_assigned SQL > % SELECT sourceParentId, argMin(updatedAt, timestamp) AS updatedAt, min(timestamp) AS assignedAt - FROM activityRelations_enriched_deduplicated_ds + FROM activityRelations_enriched_deduplicated_bucket_union WHERE - snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) - AND type IN ('pull_request-assigned', 'merge_request-assigned') + type IN ('pull_request-assigned', 'merge_request-assigned') {% if defined(bucket_id) %} AND cityHash64(segmentId) % {{ UInt8(num_buckets, 5, description="Total number of buckets", required=False) }} @@ -64,10 +60,9 @@ SQL > % SELECT sourceParentId, argMin(updatedAt, timestamp) AS updatedAt, MIN(timestamp) AS reviewRequestedAt - FROM activityRelations_enriched_deduplicated_ds + FROM activityRelations_enriched_deduplicated_bucket_union WHERE - snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) - AND (type = 'pull_request-review-requested' OR type = 'merge_request-review-requested') + (type = 'pull_request-review-requested' OR type = 'merge_request-review-requested') {% if defined(bucket_id) %} AND cityHash64(segmentId) % {{ UInt8(num_buckets, 5, description="Total number of buckets", required=False) }} @@ -91,10 +86,9 @@ SQL > ) AS sourceParentId, argMin(updatedAt, timestamp) AS updatedAt, MIN(timestamp) AS reviewedAt - FROM activityRelations_enriched_deduplicated_ds + FROM activityRelations_enriched_deduplicated_bucket_union WHERE - snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) - AND sourceParentId <> '' + sourceParentId <> '' and ( type = 'pull_request-reviewed' OR type = 'merge_request-review-changes-requested' @@ -124,10 +118,9 @@ SQL > ) AS sourceParentId, argMin(updatedAt, timestamp) AS updatedAt, MIN(timestamp) AS approvedAt - FROM activityRelations_enriched_deduplicated_ds + FROM activityRelations_enriched_deduplicated_bucket_union WHERE - snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) - AND ( + ( (type = 'pull_request-reviewed' and pullRequestReviewState = 'APPROVED') OR type = 'merge_request-review-approved' OR type = 'patchset_approval-created' @@ -153,10 +146,9 @@ SQL > if(type = 'changeset-abandoned', sourceId, sourceParentId) AS sourceParentId, argMin(updatedAt, timestamp) AS updatedAt, MIN(timestamp) AS closedAt - FROM activityRelations_enriched_deduplicated_ds + FROM activityRelations_enriched_deduplicated_bucket_union WHERE - snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) - AND ( + ( type = 'pull_request-closed' OR type = 'merge_request-closed' OR type = 'changeset-closed' @@ -186,10 +178,9 @@ SQL > if(type = 'changeset-merged', sourceId, sourceParentId) AS sourceParentId, argMin(updatedAt, timestamp) AS updatedAt, MIN(timestamp) AS mergedAt - FROM activityRelations_enriched_deduplicated_ds + FROM activityRelations_enriched_deduplicated_bucket_union WHERE - snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) - AND (type = 'pull_request-merged' OR type = 'merge_request-merged' OR type = 'changeset-merged') + (type = 'pull_request-merged' OR type = 'merge_request-merged' OR type = 'changeset-merged') {% if defined(bucket_id) %} AND cityHash64(segmentId) % {{ UInt8(num_buckets, 5, description="Total number of buckets", required=False) }} @@ -213,10 +204,9 @@ SQL > ) AS sourceParentId, argMin(updatedAt, timestamp) AS updatedAt, MIN(timestamp) AS resolvedAt - FROM activityRelations_enriched_deduplicated_ds + FROM activityRelations_enriched_deduplicated_bucket_union WHERE - snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) - AND ( + ( type = 'pull_request-closed' OR type = 'pull_request-merged' OR type = 'merge_request-closed' @@ -246,11 +236,9 @@ DESCRIPTION > SQL > % SELECT sourceParentId, toInt64(COUNT(*)) AS numberOfPatchsets - FROM activityRelations_enriched_deduplicated_ds + FROM activityRelations_enriched_deduplicated_bucket_union WHERE - snapshotId = (select max(snapshotId) from activityRelations_enriched_deduplicated_ds) - AND type = 'patchset-created' - AND sourceParentId != '' + type = 'patchset-created' AND sourceParentId != '' {% if defined(bucket_id) %} AND cityHash64(segmentId) % {{ UInt8(num_buckets, 5, description="Total number of buckets", required=False) }}