diff --git a/services/libs/tinybird/lambda-architecture.md b/services/libs/tinybird/lambda-architecture.md index 7a25fef1d1..da18b22bcb 100644 --- a/services/libs/tinybird/lambda-architecture.md +++ b/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__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__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__ds` | `segmentsAggregatedMV` | `pull_requests_analyzed` | `activityRelations_enriched_deduplicated_bucket__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 diff --git a/services/libs/tinybird/pipes/pull_request_analysis_baseline_merge_MV.pipe b/services/libs/tinybird/pipes/pull_request_analysis_baseline_merge_MV.pipe index 83658857a7..9c200b8cbd 100644 --- a/services/libs/tinybird/pipes/pull_request_analysis_baseline_merge_MV.pipe +++ b/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 > 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 d5b933268a..d6ceacd72b 100644 --- a/services/libs/tinybird/pipes/pull_request_analysis_initial_snapshot.pipe +++ b/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 >