docs: deprecate pull_request_analysis_initial_snapshot pipe (IN-1217)

services/libs/tinybird/lambda-architecture.md

@@ -506,70 +506,72 @@ Before the Lambda Architecture can run continuously, we need to **create the fir
 Initial snapshot pipes:
 - Create the baseline/first snapshot in serving datasources
 - Run once at system startup or when resetting the pipeline
-- Use `COPY_MODE: replace` to overwrite the entire target datasource
 - Process current data to create a deterministic starting point
 - Enable subsequent merger copy pipes to work incrementally
 
+**Three independent pipe families serve different datasources:**
+1. **`activityRelations_enrich_initial_snapshot_*` (0, 1, 2)** — Bootstrap the unfiltered `activityRelations_enriched_deduplicated_bucket_*_ds` per-bucket serving layer. Uses `COPY_MODE replace` (atomic, safe to re-run). Replace mode makes these stateless — run all 3 at any time to bootstrap or recover.
+2. **`segmentId_aggregates_initial_snapshot`** — Bootstrap segment-level aggregates in `segmentsAggregatedMV`. Uses `COPY_MODE replace` (atomic). Run once at setup to initialize segment metrics.
+3. **`pull_request_analysis_initial_snapshot`** — Bootstrap PR lifecycle analysis in `pull_requests_analyzed`. Uses `COPY_MODE append` and **must be run bucketed** (see "How to run" below). This is the only append-mode initial snapshot — it splits the work into 10 pieces (`bucket_id=0..9, num_buckets=10`) to avoid timeout on the full dataset. The single-shot unbucketed invocation is deprecated (does not reliably finish).
+
 ### Examples
 
-#### 1. activityRelations_enrich_initial_snapshot_0 .. _2 (per bucket)
+#### 1. activityRelations_enrich_initial_snapshot_0 .. _2 (replace-mode, per bucket)
 
 ```
-Files: activityRelations_enrich_initial_snapshot_0.pipe .. _2.pipe
-
-TYPE: COPY
-COPY_MODE: replace
-COPY_SCHEDULE: @on-demand
-TARGET_DATASOURCE: activityRelations_enriched_deduplicated_bucket_<N>_ds
+Files: activityRelations_enrich_initial_snapshot_0.pipe, _1.pipe, _2.pipe
+TYPE: COPY, COPY_MODE: replace, COPY_SCHEDULE: @on-demand
+TARGET: activityRelations_enriched_deduplicated_bucket_<N>_ds
 
-What each one does:
-├─ Reads raw activityRelations (base table), filtered to cityHash64(segmentId) % 3 = N
+Each of the 3 pipes:
+├─ Reads raw activityRelations base table, filtered to cityHash64(segmentId) % 3 = N
 ├─ Enriches with country codes, org names, gitChangedLines buckets
-├─ Creates snapshotId: toStartOfInterval(now(), INTERVAL 1 day)
-└─ Replaces bucket N of the serving layer
+├─ Assigns snapshotId: toStartOfInterval(now(), INTERVAL 1 day)
+└─ Replaces (overwrites) bucket N of the serving layer — atomically safe to re-run
 
-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.
+When to run:
+  • First deployment: Run all 3 to bootstrap the unfiltered serving layer
+  • Recovery: Run a single bucket N if it fell behind >3 days (beyond MV retention)
+  • Replace mode guarantees atomicity — a failed run leaves the previous data intact
 ```
 
-#### 2. pull_request_analysis_initial_snapshot
+#### 2. pull_request_analysis_initial_snapshot (append-mode, bucketed)
 
 ```
 File: pull_request_analysis_initial_snapshot.pipe
+TYPE: COPY, COPY_MODE: append, COPY_SCHEDULE: @on-demand
+TARGET: pull_requests_analyzed
+Params: bucket_id (0..9), num_buckets (set to 10)
 
-TYPE: COPY
-COPY_MODE: replace
-COPY_SCHEDULE: @on-demand
-TARGET_DATASOURCE: pull_requests_analyzed
+Purpose: Bootstrap PR lifecycle analysis (opened → reviewed → approved → merged)
 
-What it does:
-├─ Reads from activityRelations_deduplicated_cleaned_ds (latest snapshot)
-├─ Extracts all PR lifecycle events (opened, assigned, reviewed, approved, closed, merged)
-├─ Joins lifecycle events by sourceId/sourceParentId
-├─ Computes duration metrics (assignedInSeconds, reviewedInSeconds, etc.)
-└─ Writes initial PR analysis baseline
+Critical note: APPEND mode. Single-shot invocation is deprecated (times out on full dataset).
+  ✗ DEPRECATED: tb pipe copy run pull_request_analysis_initial_snapshot --wait
+  ✓ CORRECT:   for N in $(seq 0 9); tb pipe copy run ... --param bucket_id=$N --param num_buckets=10
+
+Execution splits the scan into 10 buckets (cityHash64(segmentId) % 10 = bucket_id).
+Each invocation appends its results; rows accumulate into the full dataset.
 
-Usage: Run once to bootstrap pull_requests_analyzed serving layer
+⚠️  Before the first bucket run, ENSURE pull_requests_analyzed is empty (append mode does NOT deduplicate).
+
+This is separate from the hourly PR Merger (pull_request_analysis_snapshot_merger_copy.pipe),
+which increments pull_requests_analyzed via baseline-merge once the initial snapshot is populated.
 ```
 
-#### 3. segmentId_aggregates_initial_snapshot
+#### 3. segmentId_aggregates_initial_snapshot (replace-mode)
 
 ```
 File: segmentId_aggregates_initial_snapshot.pipe
-
-TYPE: COPY
-COPY_MODE: replace
-COPY_SCHEDULE: @on-demand
-TARGET_DATASOURCE: segmentsAggregatedMV
+TYPE: COPY, COPY_MODE: replace, COPY_SCHEDULE: @on-demand
+TARGET: segmentsAggregatedMV
 
 What it does:
-├─ Reads from activityRelations_deduplicated_cleaned_ds (latest snapshot)
+├─ Queries activityRelations_enriched_deduplicated_bucket_union at latest snapshot
 ├─ Groups by segmentId
 ├─ Counts distinct contributors (memberId) and organizations (organizationId)
-└─ Writes initial segment aggregates
+└─ Overwrites (replaces) segment-level aggregates — atomically safe to re-run
 
-Usage: Run once to bootstrap segment-level metrics
+When to run: Once at initial setup to populate segment metrics
 ```
 
 ### When to Run Initial Snapshots
@@ -585,21 +587,42 @@ Usage: Run once to bootstrap segment-level metrics
 ```bash
 # Via Tinybird CLI (assuming you have tb CLI configured)
 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
 ```
 
+**Important — Append-Mode Safety (PR Initial Snapshot):**
+
+`pull_request_analysis_initial_snapshot.pipe` uses `COPY_MODE append` and runs split into `num_buckets` pieces. **Before running the first bucket, ensure the target `pull_requests_analyzed` datasource is empty**, otherwise successive runs will append duplicate rows and violate the deduplication contract (consumers reading all rows will see stale/duplicate metrics). Replace-mode pipes above (`activityRelations_*` and `segmentId_aggregates`) are safe to re-run (they overwrite atomically), but this append-mode pipe requires a clean slate:
+
+```bash
+# Ensure pull_requests_analyzed is empty (or back it up before reset)
+# Then run all 10 buckets sequentially (append mode — rows accumulate):
+for N in $(seq 0 9); do
+  tb pipe copy run pull_request_analysis_initial_snapshot --param bucket_id=$N --param num_buckets=10 --mode append
+done
+```
+
+If you interrupted a mid-run or suspect duplicates exist, either clear the datasource before retrying or use the scheduled PR Merger (`pull_request_analysis_snapshot_merger_copy`) to replace the entire snapshot (but note it requires the existing baseline to compute deltas — an empty datasource will produce no output).
+
 ### Comparison: Initial vs Merger Copy Pipes
 
-| 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 |
+| Aspect | activityRelations Initial Snapshot | segmentId_aggregates Initial Snapshot | PR Initial Snapshot | Bucket Merger (activityRelations) | PR Merger |
+|--------|-----------------------------------|--------------------------------------|---------------------|-----------------------------------|-----------|
+| **Pipe Name(s)** | `activityRelations_enrich_initial_snapshot_0..2` | `segmentId_aggregates_initial_snapshot` | `pull_request_analysis_initial_snapshot` | `activityRelations_snapshot_merger_copy_0..2` | `pull_request_analysis_snapshot_merger_copy` |
+| **Status** | Active | Active | Active (bucketed only; single-shot deprecated) | Active | Active |
+| **Schedule** | @on-demand (manual) | @on-demand (manual) | @on-demand (manual) | Daily (01:30/01:34/01:38 UTC) | Hourly (0 * * * *) |
+| **Mode** | replace (atomic) | replace (atomic) | append (splits into `num_buckets` runs) | replace (atomic per-bucket) | replace |
+| **Purpose** | Bootstrap activityRelations serving layer (3 buckets) | Bootstrap segment aggregates | Bootstrap PR analytics (full rebuild, run with params) | Incremental merge of activityRelations MV deltas | Incremental merge of PR events |
+| **Source** | Base `activityRelations` table | Latest snapshot (query-time) | Base `activityRelations_enriched_deduplicated_bucket_union` | MV output + own bucket (carry-forward) | PR MV output + target |
+| **Target Datasource** | `activityRelations_enriched_deduplicated_bucket_<N>_ds` | `segmentsAggregatedMV` | `pull_requests_analyzed` | `activityRelations_enriched_deduplicated_bucket_<N>_ds` | `pull_requests_analyzed` |
+| **Frequency** | Once at setup (or rarely to recover a bucket) | Once at setup | Once at setup (split as: `for N in 0..9; tb pipe copy run ... --param bucket_id=$N --param num_buckets=10`) | Continuous (daily, one run per bucket) | Continuous (hourly) |
+
+**Key Distinctions:**
+- **replace mode** (activityRelations + segmentId + PR mergers): Atomic, safe to re-run; overwrites entire target
+- **append mode** (PR initial snapshot bucketed): Runs split into `num_buckets` pieces; **target must be empty before first run** to avoid duplicates
+- All pipes are documented and active; **only** the single-shot (unbucketed) invocation of `pull_request_analysis_initial_snapshot` is deprecated
 
+---
 
 ## Troubleshooting
 

services/libs/tinybird/pipes/pull_request_analysis_baseline_merge_MV.pipe

@@ -9,9 +9,10 @@ DESCRIPTION >
     that copied everything at once." It referenced a datasource deleted in January's bucketing
     migration (`activityRelations_deduplicated_cleaned_ds` -> `..._bucket_union`) and had been
     disabled/not running since 2026-08-06, which is why the reference had gone stale unnoticed.
-    This MV + pull_request_analysis_snapshot_merger_copy.pipe are the real hourly path; this MV +
-    pull_request_analysis_initial_snapshot.pipe (bootstrap) are the only two places the non-author
-    filter needs to live.
+    This MV + pull_request_analysis_snapshot_merger_copy.pipe are the real hourly path
+    for ongoing deduplication once pull_requests_analyzed is populated. They do not
+    perform the initial bootstrap — pull_request_analysis_initial_snapshot.pipe is still
+    used for that (run bucketed via bucket_id/num_buckets; see lambda-architecture.md).
 
 NODE snapshot_resolver
 DESCRIPTION >

services/libs/tinybird/pipes/pull_request_analysis_initial_snapshot.pipe

@@ -1,7 +1,8 @@
 DESCRIPTION >
-    Compacts activities from same PR into one, keeping necessary information in a single row. Helps to serve PR-wide widgets in the development tab.
-    Run with bucket_id (0-4) and num_buckets (default 5) to process a subset of segments at a time, avoiding memory limits.
-    After all buckets complete, the hourly snapshot merger takes over.
+    Bootstrap/full-rebuild pipe for pull_requests_analyzed. Do not run single-shot
+    (`--wait` with no bucket params) — it does not finish against the full dataset.
+    Run bucketed instead, one bucket_id at a time (see lambda-architecture.md):
+    tb pipe copy run pull_request_analysis_initial_snapshot --param bucket_id=N --param num_buckets=10 --mode append
 
 NODE pull_request_opened
 SQL >