From bb19e41728a4423a5e00a197c04b4b8287c48eb3 Mon Sep 17 00:00:00 2001
From: anilb <epipav@gmail.com>
Date: Sat, 25 Jul 2026 14:02:39 +0200
Subject: [PATCH 1/3] feat: bucketed lambda merger

Signed-off-by: anilb <epipav@gmail.com>
---
 services/libs/tinybird/README.md              |  16 +--
 ...Relations_enrich_snapshot_MV_ds.datasource |   2 +-
 ...iched_deduplicated_bucket_0_ds.datasource} |   5 +-
 ...riched_deduplicated_bucket_1_ds.datasource |  33 ++++++
 ...riched_deduplicated_bucket_2_ds.datasource |  33 ++++++
 services/libs/tinybird/lambda-architecture.md | 110 +++++++++++-------
 .../pipes/activities_daily_counts.pipe        |   9 +-
 .../pipes/activities_relations_filtered.pipe  |   5 +-
 ...yRelations_enrich_initial_snapshot_0.pipe} |  11 +-
 ...tyRelations_enrich_initial_snapshot_1.pipe |  45 +++++++
 ...tyRelations_enrich_initial_snapshot_2.pipe |  45 +++++++
 ...ns_enriched_deduplicated_bucket_union.pipe |  21 ++++
 ...ctivityRelations_snapshot_merger_copy.pipe |  35 ------
 ...ivityRelations_snapshot_merger_copy_0.pipe |  43 +++++++
 ...ivityRelations_snapshot_merger_copy_1.pipe |  43 +++++++
 ...ivityRelations_snapshot_merger_copy_2.pipe |  43 +++++++
 ...r_segment_aggregates_initial_snapshot.pipe |   3 +-
 ...n_segment_aggregates_initial_snapshot.pipe |   3 +-
 .../pipes/cdp_segment_metrics_copy_pipe.pipe  |  16 +--
 .../pipes/cdp_segment_metrics_total_sink.pipe |   5 +-
 .../tinybird/pipes/monitoring_entities.pipe   |  14 ++-
 ...ull_request_analysis_initial_snapshot.pipe |  45 +++----
 22 files changed, 434 insertions(+), 151 deletions(-)
 rename services/libs/tinybird/datasources/{activityRelations_enriched_deduplicated_ds.datasource => activityRelations_enriched_deduplicated_bucket_0_ds.datasource} (87%)
 create mode 100644 services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_bucket_1_ds.datasource
 create mode 100644 services/libs/tinybird/datasources/activityRelations_enriched_deduplicated_bucket_2_ds.datasource
 rename services/libs/tinybird/pipes/{activityRelations_enrich_initial_snapshot.pipe => activityRelations_enrich_initial_snapshot_0.pipe} (73%)
 create mode 100644 services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot_1.pipe
 create mode 100644 services/libs/tinybird/pipes/activityRelations_enrich_initial_snapshot_2.pipe
 create mode 100644 services/libs/tinybird/pipes/activityRelations_enriched_deduplicated_bucket_union.pipe
 delete mode 100644 services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy.pipe
 create mode 100644 services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_0.pipe
 create mode 100644 services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_1.pipe
 create mode 100644 services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_2.pipe

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..af6f1857fa 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,63 @@ 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)
+    ├─ 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 +265,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
 
@@ -493,24 +509,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_<N>_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,7 +581,7 @@ 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
 ```
@@ -606,11 +623,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..40b81c274b
--- /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..ca6e426dbd
--- /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..276a117126
--- /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..fcdf195275 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,15 +26,9 @@ SQL >
         countState(
             if(ar.timestamp >= now() - INTERVAL 30 DAY, CAST(1 AS UInt8), NULL)
         ) AS activitiesLast30DaysState
-    FROM activityRelations_enriched_deduplicated_ds AS ar
+    FROM activityRelations_enriched_deduplicated_bucket_union AS ar
     WHERE
-        snapshotId = (
-            SELECT snapshotId
-            FROM activityRelations_enriched_deduplicated_ds
-            ORDER BY snapshotId DESC
-            LIMIT 1
-        )
-        AND segmentId IS NOT NULL
+        segmentId IS NOT NULL
         AND segmentId != ''
     GROUP BY ar.segmentId
 
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..c6444af4b4 100644
--- a/services/libs/tinybird/pipes/cdp_segment_metrics_total_sink.pipe
+++ b/services/libs/tinybird/pipes/cdp_segment_metrics_total_sink.pipe
@@ -9,10 +9,9 @@ SQL >
     SELECT
         count() AS activitiesTotal,
         countIf(timestamp >= now() - INTERVAL 30 DAY) AS activitiesLast30Days
-    FROM activityRelations_enriched_deduplicated_ds
+    FROM activityRelations_enriched_deduplicated_bucket_union
     WHERE
-        snapshotId = (SELECT max(snapshotId) FROM activityRelations_enriched_deduplicated_ds)
-        AND segmentId IS NOT NULL
+        segmentId IS NOT NULL
         AND segmentId != ''
 
 NODE cdpDashboardMetricsTotal
diff --git a/services/libs/tinybird/pipes/monitoring_entities.pipe b/services/libs/tinybird/pipes/monitoring_entities.pipe
index 4509a6ca1a..799666c144 100644
--- a/services/libs/tinybird/pipes/monitoring_entities.pipe
+++ b/services/libs/tinybird/pipes/monitoring_entities.pipe
@@ -2,14 +2,22 @@ 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..89a149ac1b 100644
--- a/services/libs/tinybird/pipes/pull_request_analysis_initial_snapshot.pipe
+++ b/services/libs/tinybird/pipes/pull_request_analysis_initial_snapshot.pipe
@@ -17,10 +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'
         )
         {% if defined(bucket_id) %}
@@ -40,10 +39,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 +62,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 +88,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 +120,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 +148,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 +180,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 +206,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,10 +238,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'
+        type = 'patchset-created'
         AND sourceParentId != ''
         {% if defined(bucket_id) %}
             AND cityHash64(segmentId)

From 7b1934194e9e296724bf2e29187940c6f06f9456 Mon Sep 17 00:00:00 2001
From: anilb <epipav@gmail.com>
Date: Sat, 25 Jul 2026 16:41:04 +0200
Subject: [PATCH 2/3] fix: inclusive delta watermark

Signed-off-by: anilb <epipav@gmail.com>
---
 services/libs/tinybird/lambda-architecture.md | 20 ++++++++++---------
 ...ivityRelations_snapshot_merger_copy_0.pipe |  2 +-
 ...ivityRelations_snapshot_merger_copy_1.pipe |  2 +-
 ...ivityRelations_snapshot_merger_copy_2.pipe |  2 +-
 4 files changed, 14 insertions(+), 12 deletions(-)

diff --git a/services/libs/tinybird/lambda-architecture.md b/services/libs/tinybird/lambda-architecture.md
index af6f1857fa..a933adda10 100644
--- a/services/libs/tinybird/lambda-architecture.md
+++ b/services/libs/tinybird/lambda-architecture.md
@@ -124,7 +124,9 @@ This document explains the **Lambda Architecture** implementation used in our Ti
                      ↓
 
     What each bucket pipe does (bucketing key: cityHash64(segmentId) % 3):
-    ├─ Fetches: NEW deltas for its bucket from MV_ds (snapshotId > bucket's last snapshot)
+    ├─ 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
@@ -588,14 +590,14 @@ 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
diff --git a/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_0.pipe b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_0.pipe
index 40b81c274b..948eb3fc28 100644
--- a/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_0.pipe
+++ b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_0.pipe
@@ -16,7 +16,7 @@ SQL >
     FROM activityRelations_enrich_snapshot_MV_ds
     WHERE
         cityHash64(segmentId) % 3 = 0
-        AND snapshotId > lastSnapshotId
+        AND snapshotId >= lastSnapshotId
         AND snapshotId <= toStartOfInterval(now(), INTERVAL 1 day)
     ORDER BY updatedAt DESC
     LIMIT 1 BY segmentId, timestamp, type, platform, channel, sourceId
diff --git a/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_1.pipe b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_1.pipe
index ca6e426dbd..65cd809a03 100644
--- a/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_1.pipe
+++ b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_1.pipe
@@ -16,7 +16,7 @@ SQL >
     FROM activityRelations_enrich_snapshot_MV_ds
     WHERE
         cityHash64(segmentId) % 3 = 1
-        AND snapshotId > lastSnapshotId
+        AND snapshotId >= lastSnapshotId
         AND snapshotId <= toStartOfInterval(now(), INTERVAL 1 day)
     ORDER BY updatedAt DESC
     LIMIT 1 BY segmentId, timestamp, type, platform, channel, sourceId
diff --git a/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_2.pipe b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_2.pipe
index 276a117126..077fce0e60 100644
--- a/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_2.pipe
+++ b/services/libs/tinybird/pipes/activityRelations_snapshot_merger_copy_2.pipe
@@ -16,7 +16,7 @@ SQL >
     FROM activityRelations_enrich_snapshot_MV_ds
     WHERE
         cityHash64(segmentId) % 3 = 2
-        AND snapshotId > lastSnapshotId
+        AND snapshotId >= lastSnapshotId
         AND snapshotId <= toStartOfInterval(now(), INTERVAL 1 day)
     ORDER BY updatedAt DESC
     LIMIT 1 BY segmentId, timestamp, type, platform, channel, sourceId

From 747071b96f360e70cc57566b283c5845315f5156 Mon Sep 17 00:00:00 2001
From: anilb <epipav@gmail.com>
Date: Sat, 25 Jul 2026 16:52:54 +0200
Subject: [PATCH 3/3] chore: format and doc cleanup

Signed-off-by: anilb <epipav@gmail.com>
---
 services/libs/tinybird/lambda-architecture.md            | 7 ++++---
 .../tinybird/pipes/cdp_segment_metrics_copy_pipe.pipe    | 4 +---
 .../tinybird/pipes/cdp_segment_metrics_total_sink.pipe   | 4 +---
 services/libs/tinybird/pipes/monitoring_entities.pipe    | 9 ++++++---
 .../pipes/pull_request_analysis_initial_snapshot.pipe    | 7 ++-----
 5 files changed, 14 insertions(+), 17 deletions(-)

diff --git a/services/libs/tinybird/lambda-architecture.md b/services/libs/tinybird/lambda-architecture.md
index a933adda10..7a25fef1d1 100644
--- a/services/libs/tinybird/lambda-architecture.md
+++ b/services/libs/tinybird/lambda-architecture.md
@@ -285,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):
 ┌─────────┬─────────────┬──────────────────┐
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 fcdf195275..60cea51108 100644
--- a/services/libs/tinybird/pipes/cdp_segment_metrics_copy_pipe.pipe
+++ b/services/libs/tinybird/pipes/cdp_segment_metrics_copy_pipe.pipe
@@ -27,9 +27,7 @@ SQL >
             if(ar.timestamp >= now() - INTERVAL 30 DAY, CAST(1 AS UInt8), NULL)
         ) AS activitiesLast30DaysState
     FROM activityRelations_enriched_deduplicated_bucket_union AS ar
-    WHERE
-        segmentId IS NOT NULL
-        AND segmentId != ''
+    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 c6444af4b4..425ffaa4cd 100644
--- a/services/libs/tinybird/pipes/cdp_segment_metrics_total_sink.pipe
+++ b/services/libs/tinybird/pipes/cdp_segment_metrics_total_sink.pipe
@@ -10,9 +10,7 @@ SQL >
         count() AS activitiesTotal,
         countIf(timestamp >= now() - INTERVAL 30 DAY) AS activitiesLast30Days
     FROM activityRelations_enriched_deduplicated_bucket_union
-    WHERE
-        segmentId IS NOT NULL
-        AND segmentId != ''
+    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 799666c144..3ad7a08fd0 100644
--- a/services/libs/tinybird/pipes/monitoring_entities.pipe
+++ b/services/libs/tinybird/pipes/monitoring_entities.pipe
@@ -12,11 +12,14 @@ SQL >
         sum(c) AS value
     FROM
         (
-            SELECT count() AS c FROM activityRelations_enriched_deduplicated_bucket_0_ds
+            SELECT count() AS c
+            FROM activityRelations_enriched_deduplicated_bucket_0_ds
             UNION ALL
-            SELECT count() FROM activityRelations_enriched_deduplicated_bucket_1_ds
+            SELECT count()
+            FROM activityRelations_enriched_deduplicated_bucket_1_ds
             UNION ALL
-            SELECT count() FROM activityRelations_enriched_deduplicated_bucket_2_ds
+            SELECT count()
+            FROM activityRelations_enriched_deduplicated_bucket_2_ds
         )
 
 NODE members_total
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 89a149ac1b..fdc2d3116f 100644
--- a/services/libs/tinybird/pipes/pull_request_analysis_initial_snapshot.pipe
+++ b/services/libs/tinybird/pipes/pull_request_analysis_initial_snapshot.pipe
@@ -19,9 +19,7 @@ SQL >
         updatedAt
     FROM activityRelations_enriched_deduplicated_bucket_union
     WHERE
-        (
-            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) }}
@@ -240,8 +238,7 @@ SQL >
     SELECT sourceParentId, toInt64(COUNT(*)) AS numberOfPatchsets
     FROM activityRelations_enriched_deduplicated_bucket_union
     WHERE
-        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) }}
