Skip to content
Merged
Show file tree
Hide file tree
Changes from 26 commits
Commits
Show all changes
28 commits
Select commit Hold shift + click to select a range
da8f0d4
[OPIK-6901] [BE] test: buffered cutover deletion-bridge E2E gate + ru…
andrescrz Jul 16, 2026
4ad4529
test(cutover): address Baz review — safe wrap default, input validati…
andrescrz Jul 22, 2026
b3cf335
test(cutover): validate the resumed backfill_start anchor content
andrescrz Jul 22, 2026
a8d910a
test(cutover): address second Baz review — wrap-only split-state guar…
andrescrz Jul 22, 2026
64f627e
test(cutover): address third Baz review — numeric arg validation + co…
andrescrz Jul 22, 2026
36f075f
test(cutover): self-heal leaked topology in resetTables
andrescrz Jul 22, 2026
9793a9b
test(cutover): make resetTables normalize from any leaked topology
andrescrz Jul 22, 2026
049b668
test(cutover): address Thiago review — drop-size guard, UTC bounds, l…
andrescrz Jul 24, 2026
2959f33
test(cutover): make stage-C rollback gapless; enforce --wrap-only mai…
andrescrz Jul 24, 2026
33851e6
test(cutover): make stage-B rollback atomic; clarify buffer prereq is…
andrescrz Jul 24, 2026
fdf8b7b
test(cutover): make the Distributed wrap gapless per node
andrescrz Jul 24, 2026
ee97d5f
test(cutover): stage-C rollback drops the ex-wrapper by an unambiguou…
andrescrz Jul 24, 2026
c7b4eed
test(cutover): enforce wrap prerequisites, tag DDL, correct buffer-co…
andrescrz Jul 24, 2026
2447cc0
test(cutover): enforce the async-insert buffer precondition on the EX…
andrescrz Jul 24, 2026
dcf6208
test(cutover): address Baz review — arg guards, query tags, verify sk…
andrescrz Jul 24, 2026
a5c9b7a
test(cutover): address JetoPistola review (clear wins) — align test S…
andrescrz Jul 27, 2026
54729ab
test(cutover): address JetoPistola review (remaining) — replay fail-s…
andrescrz Jul 27, 2026
1e888a1
docs(cutover): cross-reference the gate test from the reference SQL i…
andrescrz Jul 27, 2026
c578713
test(cutover): address Baz — bad-id partition msg, liveCount FINAL, e…
andrescrz Jul 27, 2026
936f107
test(cutover): final deletion replay before EXCHANGE; version-collaps…
andrescrz Jul 27, 2026
9f85b74
test(cutover): server-clock resurrection instant + explicit container…
andrescrz Jul 27, 2026
c731512
fix(cutover): gate EXCHANGE/rollback on retention-paused; make delete…
andrescrz Jul 27, 2026
adfd0f1
fix(cutover): resume only on dst>=src; residual full-compare gate; te…
andrescrz Jul 27, 2026
8755d6a
fix(cutover): gate rollback stages B/C on --accept-post-cutover-write…
andrescrz Jul 27, 2026
91b7983
fix(cutover): server-clock resurrection stamp; scope loss claim; conf…
andrescrz Jul 27, 2026
90f5fba
fix(cutover): add --reverse-replay-only restart path; guide partial-E…
andrescrz Jul 27, 2026
ff6e8fc
fix(cutover): drop flaky replay-time gate; disclose workspace-replay …
andrescrz Jul 27, 2026
994479e
docs(cutover): fix grammar in workspace-scoped replay explanation
andrescrz Jul 27, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
670 changes: 670 additions & 0 deletions apps/opik-backend/data-migrations/traces-local-v2-cutover/README.md

Large diffs are not rendered by default.

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
@@ -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,
Comment thread
andrescrz marked this conversation as resolved.
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,
Comment thread
andrescrz marked this conversation as resolved.
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.
Original file line number Diff line number Diff line change
@@ -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,
Comment thread
andrescrz marked this conversation as resolved.
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:
Comment thread
andrescrz marked this conversation as resolved.
-- * 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),
Comment thread
andrescrz marked this conversation as resolved.
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 (
Comment thread
andrescrz marked this conversation as resolved.
SELECT
workspace_id,
id
FROM ${ANALYTICS_DB_DATABASE_NAME}.traces
Comment thread
andrescrz marked this conversation as resolved.
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).

Comment thread
andrescrz marked this conversation as resolved.
-- 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.
Original file line number Diff line number Diff line change
@@ -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}';
Comment thread
andrescrz marked this conversation as resolved.
-- >>> 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}';
Comment thread
andrescrz marked this conversation as resolved.
-- >>> 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.
Loading
Loading