comet-ml
diff --git a/‎apps/opik-backend/data-migrations/traces-local-v2-cutover/README.md‎
Lines changed: 676 additions & 0 deletions b/‎apps/opik-backend/data-migrations/traces-local-v2-cutover/README.md‎
Lines changed: 676 additions & 0 deletions
diff --git a/‎apps/opik-backend/data-migrations/traces-local-v2-cutover/scripts/backfill.sh‎
Lines changed: 319 additions & 0 deletions b/‎apps/opik-backend/data-migrations/traces-local-v2-cutover/scripts/backfill.sh‎
Lines changed: 319 additions & 0 deletions
diff --git a/‎apps/opik-backend/data-migrations/traces-local-v2-cutover/scripts/db-app-analytics/000001_backfill_traces_local_v2.sql‎
Lines changed: 79 additions & 0 deletions b/‎apps/opik-backend/data-migrations/traces-local-v2-cutover/scripts/db-app-analytics/000001_backfill_traces_local_v2.sql‎
Lines changed: 79 additions & 0 deletions
diff --git a/‎apps/opik-backend/data-migrations/traces-local-v2-cutover/scripts/db-app-analytics/000002_delta_and_deletion_replay.sql‎
Lines changed: 196 additions & 0 deletions b/‎apps/opik-backend/data-migrations/traces-local-v2-cutover/scripts/db-app-analytics/000002_delta_and_deletion_replay.sql‎
Lines changed: 196 additions & 0 deletions
diff --git a/‎apps/opik-backend/data-migrations/traces-local-v2-cutover/scripts/db-app-analytics/000003_exchange_and_wrap.sql‎
Lines changed: 56 additions & 0 deletions b/‎apps/opik-backend/data-migrations/traces-local-v2-cutover/scripts/db-app-analytics/000003_exchange_and_wrap.sql‎
Lines changed: 56 additions & 0 deletions
@@ -0,0 +1,79 @@
+-- runbook traces-local-v2-cutover — step 1 of 3: backfill (reference statement)
+-- The gate test TracesLocalV2CutoverTest reimplements this statement inline; keep the two in step (see its Javadoc).
+--
+-- This file is the SINGLE source of the backfill INSERT; ../backfill.sh reads it, substitutes the ${...} placeholders
+-- (database, window bounds, block size) and runs it once per time sub-window — so the script and this reference never
+-- drift. Run the migration through backfill.sh, never this file by hand. WINDOW_LO/WINDOW_HI are a created_at half-open
+-- range the driver picks so each INSERT stays under its --max-rows-per-insert bound (see README "Batching and throttling").
+--
+-- Slicing rationale (created_at, not id / not workspace), delta and replay design: see ../../README.md.
+-- Notes on the statement:
+--   * The SOURCE is sliced by created_at (immutable across upserts, backed by a minmax skip index). The DESTINATION's
+--     id_at partition is derived from each row's id independently of the slice.
+--   * end_time and ttft are the two denullified columns: coalesce them to their sentinels (epoch / NaN).
+--   * is_deleted is omitted so the new column defaults to 0.
+--   * apply_deleted_mask stays at its default 1, so rows already lightweight-deleted on the source are skipped.
+--   * No explicit ORDER BY: omitted deliberately to avoid a full per-window sort (memory). A parallel SELECT gives no
+--     output-order guarantee, so inserted blocks may span/interleave partitions; the destination ReplacingMergeTree
+--     dedups regardless of insert order and background merges compact the parts. This is NOT a claim that rows arrive
+--     in sort-key order — do not rely on it (see README "Why slice by created_at").
+--   * SETTINGS max_insert_block_size bounds the rows per part-forming block; peak insert memory is a small multiple of
+--     the smaller of that and min_insert_block_size_bytes (256 MB default), which dominates for wide trace rows.
+
+INSERT INTO ${ANALYTICS_DB_DATABASE_NAME}.traces_local_v2 (
+    id,
+    workspace_id,
+    project_id,
+    name,
+    start_time,
+    end_time,
+    input,
+    output,
+    metadata,
+    tags,
+    created_at,
+    last_updated_at,
+    created_by,
+    last_updated_by,
+    error_info,
+    thread_id,
+    visibility_mode,
+    truncation_threshold,
+    input_slim,
+    output_slim,
+    ttft,
+    source,
+    environment
+)
+SELECT
+    id,
+    workspace_id,
+    project_id,
+    name,
+    start_time,
+    coalesce(end_time, toDateTime64('1970-01-01 00:00:00', 6)) AS end_time,
+    input,
+    output,
+    metadata,
+    tags,
+    created_at,
+    last_updated_at,
+    created_by,
+    last_updated_by,
+    error_info,
+    thread_id,
+    visibility_mode,
+    truncation_threshold,
+    input_slim,
+    output_slim,
+    coalesce(ttft, toFloat64('nan')) AS ttft,
+    source,
+    environment
+FROM ${ANALYTICS_DB_DATABASE_NAME}.traces
+WHERE created_at >= toDateTime64('${WINDOW_LO}', 9, 'UTC')
+  AND created_at <  toDateTime64('${WINDOW_HI}', 9, 'UTC')
+SETTINGS max_insert_block_size = ${MAX_INSERT_BLOCK_SIZE},
+         log_comment = 'traces_local_v2_backfill:${WINDOW_LO}:${WINDOW_HI}';
+
+-- Per-window reconciliation is automated by backfill.sh (uniqExact of the dedup key, aborting on > 0.01% divergence);
+-- fidelity QA across the whole copy is 000005 via verify.sh. Rollback before the EXCHANGE: rollback.sh --stage A.
@@ -0,0 +1,196 @@
+-- runbook traces-local-v2-cutover — step 2 of 3: delta-insert + deletion replay
+-- The gate test TracesLocalV2CutoverTest reimplements these statements inline; keep the two in step (see its Javadoc).
+-- Run this only after the whole backfill (step 1) is complete and reconciled.
+
+-- Step 0: The SQL below (delta-insert + deletion replay) is the single source driven by ../delta_replay.sh, which reads
+-- this file, substitutes the placeholders and runs it — never run this file by hand:
+--   ../delta_replay.sh --database opik --backfill-start '2025-06-01 12:00:00.000000'
+-- The surrounding config operations (buffer raise/restore) and the go/no-go checkpoint stay with the operator, where
+-- situational awareness matters most — those are config/judgement, not SQL. clickhouse-client prints each statement's
+-- elapsed time, which is the replay measurement in step 5.
+
+-- Step 1: BACKFILL_START is the timestamp captured BEFORE the backfill began. backfill.sh prints it at startup
+-- ("RECORD backfill_start=..."); if you ran the backfill manually, use the now64(6) you captured before the first
+-- INSERT. The delta and the replay both key off this single anchor, so writes during the whole backfill window are
+-- covered.
+
+-- Step 2: Raise the async-insert buffer ceiling so the buffer can absorb the cutover window. Set
+-- databaseAnalytics.asyncInsertBusyTimeoutMaxMs ~= 10000 (env ANALYTICS_DB_ASYNC_INSERT_BUSY_TIMEOUT_MAX_MS) and roll
+-- it out (config push + rolling restart, OR a session-level SET on a dedicated cutover connection). Because
+-- async_insert_use_adaptive_busy_timeout=1, this only widens the buffer while rows are queued. VERIFY the widening
+-- took effect before proceeding — see README.
+
+-- Step 3: Delta-insert — re-copy every row written during the backfill window. Anchored on
+-- created_at OR last_updated_at >= backfill_start (NOT last_updated_at alone): last_updated_at is client-supplied on the
+-- batch-ingest path, so it is not a reliable "changed since" signal by itself. Every trace write sets EITHER a fresh
+-- server created_at (batch-ingest path) OR a fresh server last_updated_at (create/update merge paths), so the union is
+-- complete. ReplacingMergeTree dedups the re-copied rows against the backfilled ones (newest last_updated_at wins).
+-- Uses ${BACKFILL_START}. SETTINGS max_insert_block_size bounds per-block memory as in step 1.
+-- BATCHING: the delta covers only writes during the backfill window, not the whole table, so it is normally one
+-- statement. If the backfill ran for days on a busy system and the delta is large, run it as two batched passes to keep
+-- each INSERT bounded (both columns have a minmax skip index, so each pass prunes):
+--   (a) created_at >= backfill_start                                  -- batch by created_at sub-windows
+--   (b) last_updated_at >= backfill_start AND created_at < backfill_start  -- the updates-to-old-rows arm; batch by
+--       last_updated_at sub-windows. (a) ∪ (b) equals the OR below, with no overlap.
+-- >>> BEGIN delta-insert
+INSERT INTO ${ANALYTICS_DB_DATABASE_NAME}.traces_local_v2 (
+    id,
+    workspace_id,
+    project_id,
+    name,
+    start_time,
+    end_time,
+    input,
+    output,
+    metadata,
+    tags,
+    created_at,
+    last_updated_at,
+    created_by,
+    last_updated_by,
+    error_info,
+    thread_id,
+    visibility_mode,
+    truncation_threshold,
+    input_slim,
+    output_slim,
+    ttft,
+    source,
+    environment
+)
+SELECT
+    id,
+    workspace_id,
+    project_id,
+    name,
+    start_time,
+    coalesce(end_time, toDateTime64('1970-01-01 00:00:00', 6)) AS end_time,
+    input,
+    output,
+    metadata,
+    tags,
+    created_at,
+    last_updated_at,
+    created_by,
+    last_updated_by,
+    error_info,
+    thread_id,
+    visibility_mode,
+    truncation_threshold,
+    input_slim,
+    output_slim,
+    coalesce(ttft, toFloat64('nan')) AS ttft,
+    source,
+    environment
+FROM ${ANALYTICS_DB_DATABASE_NAME}.traces
+WHERE created_at >= toDateTime64('${BACKFILL_START}', 6)
+   OR last_updated_at >= toDateTime64('${BACKFILL_START}', 6)
+SETTINGS max_insert_block_size = ${MAX_INSERT_BLOCK_SIZE},
+         log_comment = 'traces_local_v2_cutover:delta_insert';
+-- >>> END delta-insert
+
+-- Step 4: Deletion replay — remove from the destination every row that was deleted on the source since backfill_start
+-- AND is still deleted there. Two branches, mirroring the product's two delete paths (TraceService.delete): a delete
+-- resolves each trace's owning project and deletes per project; ids it cannot resolve fall back to a workspace-scoped
+-- delete (TraceDAO DELETE_BY_ID with no project filter). The bridge records the first with the project and the second
+-- with an EMPTY project_id (DeletionEventDAO: "project_id is empty for workspace-scoped source tables"). So:
+--   * events WITH a project -> match the FULL key (workspace_id, project_id, id). Exact, prunes on the destination
+--     primary key, and correct even when an id is reused across projects (ids are not globally unique).
+--   * events WITHOUT a project -> match (workspace_id, id). A faithful mirror of the source's workspace-scoped delete.
+-- RESURRECTION GUARD (the `NOT IN traces` arm): a trace can be deleted and then re-created/updated under the same id
+-- during the window (client-supplied ids; the delete is a mask, a newer insert wins under FINAL). Such an id is bridged
+-- as deleted but is LIVE again on the source, and the backfill/delta already copied its live version. Deleting it by key
+-- would drop a row that is live on the source — silent data loss. So each branch deletes only ids that are NOT currently
+-- live on the source (mask-honored). The `id IN (deleted_ids since anchor)` bound keeps the deleted-id set tiny
+-- (retention is off, so these are user-scale deletes); `traces` has no id skip index (000088 indexes only
+-- created_at/last_updated_at — id minmax/bloom indexes exist only on traces_local_v2), so this source lookup is a
+-- bounded id-filtered read of that tiny set, not a value-indexed prune of the ~4 TB table.
+-- KNOWN RESIDUAL (workspace-scoped arm only): its guard keys on (workspace_id, id), so an id live in ONE project shields
+-- the deletion of that id's now-deleted copies in OTHER projects (ids are not globally unique). Requires cross-project id
+-- reuse + a workspace-scoped delete + a resurrection in the window — rare. The 000005 FINAL fingerprint flags it as an
+-- extra destination row (ok=0) rather than passing silently. OPIK-7483 removes the workspace-scoped delete path at the
+-- source (deletes always carry project_id), retiring this arm and the residual.
+-- allow_nondeterministic_mutations: a lightweight DELETE with cross-table subqueries is flagged nondeterministic, but
+-- deletion_events_local and traces are replicated and identical on every node and the window predicate is fixed, so the
+-- subqueries resolve to the same set on every replica. Idempotent (never masks a live-on-source id, so re-runs converge).
+-- lightweight_deletes_sync = 2: block until the delete mutation has completed on EVERY replica, not just the one that
+-- accepted it. The mutation is otherwise asynchronous, so without this the verify step (and the EXCHANGE) could run
+-- against a replica where the mask is not yet applied — a false mismatch, or worse an incomplete cutover.
+-- Uses ${BACKFILL_START}. Retention is disabled everywhere (see step 6), so this is user-scale volume — a single
+-- mutation. If it is ever large (e.g. retention enabled), bound each mutation by a partition predicate and loop the
+-- weeks, e.g. AND toMonday(id_at) = toDate('<week>').
+-- length(...) = 36 guards: toFixedString(x, 36) THROWS on a value longer than 36 bytes, which would abort the whole
+-- replay on a single malformed bridge row. For source_table='traces' the ids are 36-char UUIDs, so this is latent — but
+-- a malformed (non-36-char) deleted_id/project_id can't match a real trace id anyway, so skipping it via the length
+-- guard loses nothing and turns a hard abort mid-cutover into a benign no-op. Same guards in the reverse-replay.
+-- >>> BEGIN deletion-replay
+DELETE FROM ${ANALYTICS_DB_DATABASE_NAME}.traces_local_v2
+WHERE (
+    (workspace_id, project_id, id) IN (
+        SELECT
+            workspace_id,
+            toFixedString(project_id, 36),
+            toFixedString(deleted_id, 36)
+        FROM ${ANALYTICS_DB_DATABASE_NAME}.deletion_events_local
+        WHERE source_table = 'traces'
+          AND event_time >= toDateTime64('${BACKFILL_START}', 6)
+          AND project_id != ''
+          AND length(project_id) = 36
+          AND length(deleted_id) = 36
+    )
+    AND (workspace_id, project_id, id) NOT IN (
+        SELECT
+            workspace_id,
+            project_id,
+            id
+        FROM ${ANALYTICS_DB_DATABASE_NAME}.traces
+        WHERE id IN (
+            SELECT toFixedString(deleted_id, 36)
+            FROM ${ANALYTICS_DB_DATABASE_NAME}.deletion_events_local
+            WHERE source_table = 'traces'
+              AND event_time >= toDateTime64('${BACKFILL_START}', 6)
+              AND length(deleted_id) = 36
+        )
+    )
+)
+OR (
+    (workspace_id, id) IN (
+        SELECT
+            workspace_id,
+            toFixedString(deleted_id, 36)
+        FROM ${ANALYTICS_DB_DATABASE_NAME}.deletion_events_local
+        WHERE source_table = 'traces'
+          AND event_time >= toDateTime64('${BACKFILL_START}', 6)
+          AND project_id = ''
+          AND length(deleted_id) = 36
+    )
+    AND (workspace_id, id) NOT IN (
+        SELECT
+            workspace_id,
+            id
+        FROM ${ANALYTICS_DB_DATABASE_NAME}.traces
+        WHERE id IN (
+            SELECT toFixedString(deleted_id, 36)
+            FROM ${ANALYTICS_DB_DATABASE_NAME}.deletion_events_local
+            WHERE source_table = 'traces'
+              AND event_time >= toDateTime64('${BACKFILL_START}', 6)
+              AND length(deleted_id) = 36
+        )
+    )
+)
+SETTINGS allow_nondeterministic_mutations = 1,
+         lightweight_deletes_sync = 2,
+         log_comment = 'traces_local_v2_cutover:deletion_replay';
+-- >>> END deletion-replay
+
+-- Step 5: Measure the replay. Compare its wall time against the buffer window (must fit with margin — acceptance
+-- criterion). Re-run steps 3-4 if new rows/deletes accumulated during the replay itself; convergence is fast because
+-- the buffer is holding new writes.
+
+-- Step 6 (retention — see README): Data Retention is disabled in every deployment (RETENTION_ENABLED=false), so the
+-- retention delete path does not fire during the cutover. The only deletes in this window are user-initiated, and those
+-- ARE captured by the bridge. If retention is ever enabled, pause it for the window (or land retention-path capture).
+
+-- rollback: none for the delta-insert (it only adds newest versions that ReplacingMergeTree dedups); the replay is
+--           idempotent. If aborting the cutover here, TRUNCATE traces_local_v2 (step 1 rollback) and restore the buffer
+--           ceiling (step 2, reverse). The live `traces` table is still untouched until the EXCHANGE in step 3.
@@ -0,0 +1,56 @@
+-- runbook traces-local-v2-cutover — step 3 of 3: EXCHANGE + Distributed wrap (reference statements)
+-- The gate test TracesLocalV2CutoverTest reimplements these statements inline; keep the two in step (see its Javadoc).
+--
+-- ../exchange_and_wrap.sh drives this: it records cutover_start, runs the `exchange` block, and (unless --skip-wrap)
+-- the `wrap` block. Run it right after step 2's delta + replay, while the async-insert buffer is still holding writes.
+-- Do NOT run this whole file wholesale — the driver runs one marked block at a time. Buffer knob: raise
+-- databaseAnalytics.asyncInsertBusyTimeoutMaxMs before the cutover and unset it after (a backend-config action, not SQL).
+--
+-- cutover_start is a now64(6) captured RIGHT BEFORE the EXCHANGE; a rollback after this point replays deletes that fired
+-- on the new live table since then. exchange_and_wrap.sh captures and prints it; record it for the rollback.
+
+-- >>> BEGIN exchange
+-- The atomic swap: `traces` now refers to the partitioned data. The displaced old data lands under `traces_local_v2`
+-- momentarily, then is renamed to `traces_pre_cutover_backup` so its name marks it as the retained pre-cutover backup,
+-- not the "v2" successor (rationale: README "Naming and the parked backup"). Requires an Atomic database (default). If
+-- the Liquibase ClickHouse extension cannot execute EXCHANGE ON CLUSTER in the downtime-based path, use the fallback
+-- RENAME sequence in the README instead.
+-- log_comment tags these DDL statements in system.query_log for cutover attribution (DDL takes it via a leading SET,
+-- not a trailing SETTINGS clause).
+SET log_comment = 'traces_local_v2_cutover:exchange';
+EXCHANGE TABLES ${ANALYTICS_DB_DATABASE_NAME}.traces AND ${ANALYTICS_DB_DATABASE_NAME}.traces_local_v2 ON CLUSTER '{cluster}';
+
+RENAME TABLE ${ANALYTICS_DB_DATABASE_NAME}.traces_local_v2 TO ${ANALYTICS_DB_DATABASE_NAME}.traces_pre_cutover_backup ON CLUSTER '{cluster}';
+-- >>> END exchange
+
+-- >>> BEGIN wrap
+-- Sharding-ready wrap: move the partitioned table under *_local and front it with a Distributed table keyed on
+-- sipHash64(project_id). Transparent on a single shard; switching on sharding later is config-only. The {cluster} macro
+-- (not the literal 'cluster') keeps the DDL portable; it is resolved server-side.
+-- HARD PREREQUISITE: a Distributed table supports SELECT and INSERT but NOT mutations — a lightweight DELETE returns
+-- "DELETE query is not supported" (code 36) and ALTER ... DELETE returns "Distributed doesn't support mutations"
+-- (code 48). So the product's delete-by-id AND retention deletes both break the moment this wrap is applied. Do NOT run
+-- the wrap until those DAO paths target `traces_local` (see README "The Distributed wrap"). The EXCHANGE above is the
+-- data cutover and leaves `traces` a MergeTree where deletes still work; the wrap is a separate, gated step.
+--
+-- GAPLESS per node: build the Distributed wrapper under a temp name FIRST (its 'traces_local' target need not exist
+-- yet — Distributed resolves it lazily), then a SINGLE atomic multi-target RENAME rotates the data to `traces_local`
+-- and the wrapper into `traces` (the name freed by the first clause). So `traces` transitions MergeTree->Distributed
+-- with no window where the name is absent — unlike a RENAME-then-CREATE, which leaves `traces` missing in between.
+-- (A cross-node ON CLUSTER propagation skew still exists, as for any ON CLUSTER DDL; the driver's --confirm-maintenance
+-- gate covers it.) Partial-failure recovery: if the RENAME fails after the CREATE, `traces` is untouched (still the
+-- successor MergeTree, live) and only the temp wrapper lingers — drop it and retry:
+--   DROP TABLE IF EXISTS ${ANALYTICS_DB_DATABASE_NAME}.traces_dist ON CLUSTER '{cluster}' SYNC;
+SET log_comment = 'traces_local_v2_cutover:wrap';
+CREATE TABLE ${ANALYTICS_DB_DATABASE_NAME}.traces_dist ON CLUSTER '{cluster}' AS ${ANALYTICS_DB_DATABASE_NAME}.traces
+    ENGINE = Distributed('{cluster}', '${ANALYTICS_DB_DATABASE_NAME}', 'traces_local', sipHash64(project_id));
+
+RENAME TABLE
+    ${ANALYTICS_DB_DATABASE_NAME}.traces TO ${ANALYTICS_DB_DATABASE_NAME}.traces_local,
+    ${ANALYTICS_DB_DATABASE_NAME}.traces_dist TO ${ANALYTICS_DB_DATABASE_NAME}.traces
+    ON CLUSTER '{cluster}';
+-- >>> END wrap
+
+-- After the wrap: restore the buffer ceiling (unset asyncInsertBusyTimeoutMaxMs), verify (README "Verifying the
+-- migration"), and keep `traces_pre_cutover_backup` (the parked old data) until the soak completes. Rollback: the
+-- 000004_rollback_* files via ../rollback.sh.