From cf0ebe76643103408f300ea15c0c008e7fa8ec39 Mon Sep 17 00:00:00 2001 From: Thiago Hora Date: Thu, 3 Sep 2026 16:30:05 +0200 Subject: [PATCH] [OPIK-8230] fix(traces): scope the delete mutation to the batch's own partitions A ClickHouse mutation is registered against every active part of the target table before its WHERE clause is even considered - the existing partition predicate (OPIK_6901) prunes which rows a matched part rewrites, not which parts the mutation visits at all. Post-cutover traces_local has ~1,900 partitions / ~3,650 parts, so an ordinary delete matching a handful of rows was registered against all of them, at ~19ms/part fixed overhead. That is what pushed deletes past max_execution_time (Code 159 TIMEOUT_EXCEEDED), where the statement errors and skips the TracesDeleted cascade and the deletion-events bridge write even though the mutation completes and the rows are in fact deleted. Scope each statement to the partitions its own batch resolves to via IN PARTITION, derived by WeeklyPartitions#groupByPartition (new; groups ids by partition rather than naming their union - IN PARTITION accepts exactly one partition per statement). One statement per partition, falling back to today's unbounded form whenever the batch contains an id whose partition cannot be derived exactly, or whenever the mutation's target table is not actually partitioned (the legacy pre-wrap traces has no partition key at all, so IN PARTITION against it is a ClickHouse syntax error - Code 248 INVALID_PARTITION_VALUE - not a no-op). The routing decision stays in exactly one place: deleteBatch calls tracesMutationTable() itself, resolved once and reused across every partition-scoped statement in the batch, never the wrap flag directly (TraceMutationRoutingArchTest enforces this). Both forms share one SQL template, DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS, with an IN PARTITION conditional - scoped and unbounded no longer duplicate their WHERE/SETTINGS body across two near-identical constants, matching the pre-diff code's own pattern for its now-removed fragment. executeDelete lost its sqlTemplate parameter along with the second constant, since there is now only one to pass. WeeklyPartitions#of, the flat-union predecessor this class shipped with, is removed - groupByPartition is its only production caller's need, and of() had none left once TraceDAO switched over. Its unique boundary-case coverage (ceiling/floor ids, largest UUIDv7 timestamp) was migrated to groupByPartition rather than dropped; the rest was redundant with groupByPartition's own tests using the same pinned ids. Every stale {@link WeeklyPartitions#of} javadoc reference left over from the rename now points at #groupByPartition. TracesLegacyTablePruningMutationTest's nonV7IdDisablesPruning assertion was still matching against the retired row-level predicate regex - a tautology that could never fail regardless of correctness, since the new SQL shape can't contain it either way. Migrated to the same doesNotContain("IN PARTITION") check its two sibling assertions in the same file already use, and the now-fully-dead PARTITION_PREDICATE / EMITTED_IN_CLAUSE / HONEST_WEEK / LEGACY_WEEK / boundPartitionsOf this left behind were removed with it. Two known, accepted tradeoffs, documented at their call sites rather than addressed here since both are out of this ticket's stated scope: - TraceService's delete->cascade coupling (TracesDeleted publish + deletion-events capture) is still all-or-nothing over a batch that can now complete several partitions' statements before one errors, so more of a batch's rows can be synchronously, confirmedly deleted without the cascade firing than before. Deletes are idempotent and this is the ticket's own explicit boundary ("removes the condition that trips it; does not restructure the coupling"). - The IN PARTITION gate is a name check (tracesMutationTable() == traces_local), not a runtime read of the table's actual partition key - an operational precondition the class already relies on for every other mutation, made explicit rather than newly introduced. Verified locally against real ClickHouse via testcontainers: - WeeklyPartitionsTest: 17/17 - TracesPartitionPruningMutationTest: 8/8, rewritten to assert on system.part_log MutatePart events rather than EXPLAIN (which reports read-planner pruning, a different layer from what a mutation actually visits - the gap the old coverage fell into) - TracesLegacyTablePruningMutationTest: 3/3, updated for the corrected pre-wrap behavior - TraceMutationRoutingArchTest: 3/3 - the routing decision is still made in exactly one place Not yet exercised: the TracesDeleted cascade and deletion_events_local write firing again once the statement stops erroring, and the e2e specs - none reachable without the full cascade path or the e2e suite running. Co-Authored-By: Claude Opus 5 --- .../java/com/comet/opik/domain/TraceDAO.java | 190 ++++++-- .../com/comet/opik/domain/TraceService.java | 16 +- .../comet/opik/utils/WeeklyPartitions.java | 101 ++-- .../TracesLegacyTablePruningMutationTest.java | 114 ++--- .../TracesPartitionPruningMutationTest.java | 443 +++++++++--------- .../opik/utils/WeeklyPartitionsTest.java | 210 +++++---- 6 files changed, 599 insertions(+), 475 deletions(-) diff --git a/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceDAO.java b/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceDAO.java index 99026c4cd53..a3e92dd596d 100644 --- a/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceDAO.java +++ b/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceDAO.java @@ -1973,22 +1973,57 @@ ORDER BY (workspace_id, project_id, trace_id, id) DESC, last_updated_at DESC * project_id}, so no delete is ever project-less - also required once {@code traces} is a Distributed table * (OPIK-7455). *

- * {@code } adds the table's own weekly partition expression, bound as the set of partitions the - * batch's ids resolve to. It is emitted whenever every id in the batch is one whose partition can be derived - * exactly ({@link WeeklyPartitions#of}); otherwise the predicate is omitted and the statement is byte-identical to - * the previous unbounded form. That is what preserves the original guarantee — a row whose {@code id_at} cannot be - * trusted is still deleted, because no id in such a batch is used to derive a partition. + * The unbounded form: no partition scoping at all, so it is the correct statement whenever + * {@link WeeklyPartitions#groupByPartition} cannot derive an exact partition for every id in the batch (a non-UUIDv7 + * id, or one past the {@code DateTime64} ceiling — see that method's Javadoc). {@link #DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS} + * is preferred whenever it can be used; this is the fallback that stays correct on any batch, at the cost of the + * mutation visiting every part of the table (OPIK-8230). *

- * No schema flag gates it. {@link WeeklyPartitions} derives a value per {@code id_at} type the mutation may meet - * — the legacy 32-bit {@code DateTime} of {@code traces} as well as the {@code DateTime64(0)} of the partitioned - * successor — so one rendered statement is correct on both sides of the cutover EXCHANGE, in either direction, with - * nothing to flip and nothing to revert on rollback. + * The pairs are bound (never inlined) as two positional string arrays and zipped back into {@code (project_id, id)} + * tuples with {@code arrayZip}, so the query text is constant regardless of batch size and no value reaches the SQL + * as a literal. {@code arrayZip} is a deterministic function, not a subquery - ClickHouse rejects subqueries in + * delete mutations. Callers batch to keep each array within the driver's reliable bind size ({@link + * com.comet.opik.infrastructure.FilterUtils#ANALYTICS_DELETE_BATCH_SIZE}). + */ + /** + * Deletes by the full {@code (workspace_id, project_id, id)} sort key, matching on {@code (project_id, id)} tuples + * so a single statement can span several projects (e.g. a reused id resolved to all its owning projects, or a + * cross-project batch) instead of one delete per project (OPIK-7483). Every deleted row carries its {@code + * project_id}, so no delete is ever project-less - also required once {@code traces} is a Distributed table + * (OPIK-7455). + *

+ * {@code } adds {@code IN PARTITION } (OPIK-8230), which scopes the mutation itself + * — which parts it is registered against — rather than which rows its {@code WHERE} selects. That distinction is + * the entire fix. A ClickHouse mutation selects parts at the partition stage, before the {@code WHERE} + * clause is considered at all — a {@code (workspace_id, project_id, id)} predicate, however exact, prunes zero + * parts, only rows within whichever parts were already selected. Post-cutover {@code traces_local} is + * weekly-partitioned into ~1,900 partitions / ~3,650 parts; a delete matching a handful of rows was registered + * against every one of them regardless, at ~19 ms/part fixed overhead, which is what pushed ordinary deletes past + * the {@code max_execution_time} ceiling (`Code: 159 TIMEOUT_EXCEEDED`) that pre-cutover deletes (1.5–2.2 s) never + * approached. Omitted for the unbounded fallback ({@link TraceDAOImpl#deleteBatch}), whose statement is then + * byte-identical to the pre-OPIK-8230 form. + *

+ * One statement per partition the batch's ids resolve to ({@link WeeklyPartitions#groupByPartition}), each binding + * only the {@code (project_id, id)} pairs whose id belongs to that partition — {@code IN PARTITION} accepts exactly + * one partition per statement; ClickHouse's {@code IN PARTITION p1, p2, …} form is for composite partition + * keys and rejects a list of distinct partitions outright. Naming a partition that does not exist is a silent + * no-op, not an error, which is what makes it safe for a far-future id to be bound into two statements (see + * {@link WeeklyPartitions#groupByPartition}'s Javadoc) — the one naming the partition it is not actually in simply + * matches nothing there. *

- * Why it matters: a mutation selects parts at the partition stage, where the (workspace_id, project_id, id) - * predicate prunes nothing, so deleting a handful of rows rewrote every part of the table. Measured on prod-test - * (271.6 M rows, 3,928 parts): 12 ids rewrote 3,928 parts / 5.40 TiB. With this predicate the same batch - * selects 5 parts. An {@code id_at} range is not a substitute: on a batch spanning 1996 and 2200 a - * range still selected 2,644 parts, where the exact set selected 4. + * {@code } is interpolated by StringTemplate, exactly like {@code } and + * {@code } above it — {@code IN PARTITION {p:UInt32}} is a ClickHouse syntax error, so the value + * cannot be bound. Unlike the {@code } interpolation this carries no escaping question: the value is + * always a Java {@code long} produced by {@link WeeklyPartitions}, and a {@code long} cannot carry a quote or any + * other SQL-meaningful character. + *

+ * Kept as a single {@code DELETE FROM} statement rather than a hand-written {@code ALTER TABLE ... UPDATE ... IN + * PARTITION}: both compile to the same mutation, but the hand-written form would step outside + * {@link #tracesMutationTable()}'s routing guarantee ({@code TraceMutationRoutingArchTest}), outside the project's + * one {@code lightweight_deletes_sync} convention, and — decisively — outside {@code lightweight_delete_mode}: a + * {@code DELETE FROM} takes the patch-part path when that setting is enabled and the table's data-model columns + * allow it; a hand-written {@code ALTER ... UPDATE} does not; so staying on {@code DELETE FROM} means any future + * enabling of patch parts benefits this path for free, with no further code change here. *

* The pairs are bound (never inlined) as two positional string arrays and zipped back into {@code (project_id, id)} * tuples with {@code arrayZip}, so the query text is constant regardless of batch size and no value reaches the SQL @@ -1998,9 +2033,9 @@ ORDER BY (workspace_id, project_id, trace_id, id) DESC, last_updated_at DESC */ private static final String DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS = """ DELETE FROM + IN PARTITION WHERE workspace_id = :workspace_id AND (project_id, id) IN arrayZip(:project_ids, :trace_ids) - AND toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1))) IN :partitions SETTINGS log_comment = '' ; """; @@ -2328,7 +2363,7 @@ ORDER BY (workspace_id, project_id, id) DESC, last_updated_at DESC * {@code :min_id}'s week and inverted the window, so the fast pass matched nothing and every id fell through to * the unbounded pass. It now resolves them. *

- * The fallback remains load-bearing for what {@link com.comet.opik.utils.WeeklyPartitions#of} still cannot derive + * The fallback remains load-bearing for what {@link com.comet.opik.utils.WeeklyPartitions#groupByPartition} still cannot derive * exactly: an id at or past the end of {@code DateTime64}'s range, where {@code id_at} saturates to * {@code 2299-12-31} whatever the real week, so every such id collapses into one partition. Real data contains * them, so the bounded query is never a delete's sole resolver. @@ -3649,38 +3684,103 @@ public Mono delete(Set> projectIdTraceIdPairs, @NonNull C return makeMonoContextAware((userName, workspaceId) -> Flux .fromIterable(Lists.partition(List.copyOf(projectIdTraceIdPairs), ANALYTICS_DELETE_BATCH_SIZE)) - .concatMap(batch -> { - var template = getSTWithLogComment(DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS, "delete_traces", - workspaceId, - userName, "pairs_size=%s".formatted(batch.size())); - selectTracesMutationTable(template); - - var projectIds = batch.stream().map(pair -> pair.getLeft().toString()).toArray(String[]::new); - var traceIds = batch.stream().map(pair -> pair.getRight().toString()).toArray(String[]::new); - - // Prune to the batch's own partitions when every id in the batch allows it; otherwise emit the - // unbounded form. Needs no schema flag: WeeklyPartitions derives a value per id_at type the - // mutation may meet, so the set is correct on the legacy traces and on the partitioned successor. - var partitions = WeeklyPartitions.of(batch.stream().map(Pair::getRight).toList()); - // Flag only, exactly like distributed_wrap: the values reach ClickHouse via the bind below, - // never through the template, so the rendered SQL is constant regardless of batch contents. - partitions.ifPresent(_ -> template.add("partitions", true)); - - var statement = connection.createStatement(template.render()) - .bind("workspace_id", workspaceId) - .bind("project_ids", projectIds) - .bind("trace_ids", traceIds); + .concatMap(batch -> deleteBatch(batch, workspaceId, userName, connection)) + .then()); + } - if (partitions.isPresent()) { - statement = statement.bind("partitions", partitions.get().toArray(Long[]::new)); - } + /** + * Deletes one ANALYTICS_DELETE_BATCH_SIZE-sized batch, scoped to its own partitions when every id in it resolves + * exactly (OPIK-8230): one {@link #DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS} statement per partition the + * batch's ids resolve to ({@link WeeklyPartitions#groupByPartition}), each binding only the pairs whose id belongs + * to that partition. Falls back to a single {@link #DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS} statement over the whole + * batch either when {@code groupByPartition} cannot derive every id's partition exactly, or when the mutation's + * own target table is not actually partitioned — always correct, merely unscoped, and identical to what this + * method emitted before OPIK-8230. + *

+ * The second condition is load-bearing and easy to miss: {@link WeeklyPartitions} is deliberately schema-agnostic + * (see its own Javadoc) and derives a partition value regardless of which table {@code id_at} actually lives in — + * including the legacy, UNPARTITIONED {@code traces} that {@link #tracesMutationTable()} still targets before the + * sharding-readiness wrap is applied. {@code IN PARTITION } against a table with no partition key is not a + * no-op; ClickHouse rejects it outright (Code 248 {@code INVALID_PARTITION_VALUE}, "Wrong number of fields in the + * partition expression: 1, must be: 0"). So the scoped path is gated on {@link #tracesMutationTable()}'s + * OWN resolved name, not on whether a partition value merely COULD be derived: this method calls the + * resolver itself, exactly as {@link #selectTracesMutationTable} does for the template - never the wrap + * flag directly ({@code TraceMutationRoutingArchTest} enforces that {@link #tracesDistributedWrapEnabled()} + * is called only from {@link #tracesMutationTable()}, so no second place can get the read/mutate split + * wrong). Resolved once per batch and reused across every partition-scoped statement it emits: the table + * a mutation targets does not vary per partition, only per topology. + *

+ * The gate is a NAME check — {@code TRACES_LOCAL_TABLE.equals(table)} — not a runtime read of + * {@code traces_local}'s actual engine or partition key. That is an accepted operational precondition, not an + * oversight: it relies on the wrap being flipped only after the EXCHANGE that makes {@code traces_local} the + * partitioned successor has already run, which is the documented ordering the migration follows and the same + * assumption {@link #tracesMutationTable()} already makes for every other mutation in this class (a premature + * flip breaks inserts and reads the same way, table-missing rather than Code 248). Verifying the live schema + * before every batch would trade a real per-request cost for a misordering that procedure, not this code, is + * relied on to prevent. + *

+ * Sequential ({@code concatMap}), deliberately not concurrent: bounded concurrency was measured and rejected for + * this change (OPIK-8230) — a real gain in the rare many-partition tail, but a knob to hold in reserve rather than + * ship alongside the fix itself. + */ + private Mono deleteBatch(List> batch, String workspaceId, String userName, + Connection connection) { + // Resolved once, via the resolver itself - never the flag directly (see the method Javadoc for why). + var table = tracesMutationTable(); + var grouped = TRACES_LOCAL_TABLE.equals(table) + ? WeeklyPartitions.groupByPartition(batch.stream().map(Pair::getRight).toList()) + : Optional.>>empty(); + + if (grouped.isEmpty()) { + return executeDelete(table, batch, null, workspaceId, userName, connection); + } - var segment = startSegment("traces", "Clickhouse", "delete"); - return Mono.from(statement.execute()) - .doFinally(_ -> endSegment(segment)) - .then(); + // id -> its pairs, built once per batch rather than rescanning the whole batch once per partition: an id can + // map to more than one pair (the same trace id reused across projects, OPIK-7483), so this is a + // Collectors.groupingBy, not a plain lookup map. + var pairsById = batch.stream().collect(Collectors.groupingBy(Pair::getRight)); + + return Flux.fromIterable(grouped.get().entrySet()) + .concatMap(entry -> { + var partitionPairs = entry.getValue().stream() + .flatMap(id -> pairsById.get(id).stream()) + .toList(); + return executeDelete(table, partitionPairs, entry.getKey(), workspaceId, userName, connection); }) - .then()); + .then(); + } + + /** + * Renders and executes one {@link #DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS} statement — unbounded when + * {@code partition} is null, scoped to it otherwise (the weekly partition value the caller has already resolved + * every pair in {@code pairs} to belong to). Shares the bind/segment/logging shape in exactly one place rather + * than duplicated per call site in {@link #deleteBatch}. + *

+ * Takes {@code table} as a parameter rather than resolving it itself via {@link #selectTracesMutationTable}: it + * is already resolved once, up front, by {@link #deleteBatch} — the physical table does not vary per statement + * within one batch, only per topology — so re-resolving it here per call would be redundant, not incorrect. + */ + private Mono executeDelete(String table, List> pairs, Long partition, + String workspaceId, String userName, Connection connection) { + var template = getSTWithLogComment(DELETE_BY_PROJECT_ID_TRACE_ID_PAIRS, "delete_traces", workspaceId, + userName, "pairs_size=%s".formatted(pairs.size())); + template.add("traces_mutation_table", table); + if (partition != null) { + template.add("partition", partition); + } + + var projectIds = pairs.stream().map(pair -> pair.getLeft().toString()).toArray(String[]::new); + var traceIds = pairs.stream().map(pair -> pair.getRight().toString()).toArray(String[]::new); + + var statement = connection.createStatement(template.render()) + .bind("workspace_id", workspaceId) + .bind("project_ids", projectIds) + .bind("trace_ids", traceIds); + + var segment = startSegment("traces", "Clickhouse", "delete"); + return Mono.from(statement.execute()) + .doFinally(_ -> endSegment(segment)) + .then(); } /** diff --git a/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceService.java b/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceService.java index 3ab8e754f06..b6f83cded45 100644 --- a/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceService.java +++ b/apps/opik-backend/src/main/java/com/comet/opik/domain/TraceService.java @@ -537,7 +537,7 @@ public Mono delete(@NonNull Set ids, UUID projectId) { * Resolves every owning project for each id: a bounded fast pass, then an unbounded pass over only the ids the * bounded one leaves unresolved. Returns id -> owning projects; ids absent from the result have no live row. *

- * The bounded pass's week window can miss a row whose week {@link com.comet.opik.utils.WeeklyPartitions#of} + * The bounded pass's week window can miss a row whose week {@link com.comet.opik.utils.WeeklyPartitions#groupByPartition} * cannot derive exactly — an id at or past the end of {@code DateTime64}'s range, where {@code id_at} saturates * to {@code 2299-12-31} whatever the real week — so the unbounded pass re-resolves the miss set and the bounded * query is never a delete's sole resolver. A far-future timestamp short of that ceiling is no longer such a case: @@ -566,6 +566,20 @@ private Mono>> resolveOwningProjects(Set ids) { }); } + /** + * All-or-nothing over {@code projectIdTraceIdPairs}: {@code TracesDeleted} publishes and the deletion-events + * bridge captures only {@code .doOnSuccess}/{@code .then} of {@link TraceDAO#delete}, so any error anywhere in + * the batch skips both for the WHOLE batch, not just the pairs that did not complete. + *

+ * OPIK-8230 widens the window this can bite in, worth being explicit about here rather than only in that + * ticket: {@link TraceDAO#delete} can now emit several sequential statements per batch, one per partition + * (via {@code deleteBatch}), so an error partway through leaves the EARLIER statements' rows genuinely, + * synchronously deleted while the cascade still does not fire for any pair in the batch - not only for the one + * that errored. Restructuring this coupling to be per-statement aware is explicitly out of scope for that + * ticket ("this ticket removes the condition that trips it; it does not restructure the coupling") - deletes + * are idempotent, so a caller that retries converges, and the ticket's own fix (scoping each mutation so it + * completes well inside the statement timeout) is what makes hitting this at all much rarer than it is today. + */ private Mono delete(Set> projectIdTraceIdPairs, Connection connection) { return Mono.deferContextual(ctx -> { String workspaceId = ctx.get(RequestContext.WORKSPACE_ID); diff --git a/apps/opik-backend/src/main/java/com/comet/opik/utils/WeeklyPartitions.java b/apps/opik-backend/src/main/java/com/comet/opik/utils/WeeklyPartitions.java index 06ba82c0bba..0bafa6bd07e 100644 --- a/apps/opik-backend/src/main/java/com/comet/opik/utils/WeeklyPartitions.java +++ b/apps/opik-backend/src/main/java/com/comet/opik/utils/WeeklyPartitions.java @@ -9,14 +9,19 @@ import java.time.ZoneOffset; import java.time.temporal.TemporalAdjusters; import java.util.Collection; +import java.util.HashMap; import java.util.HashSet; +import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.UUID; +import java.util.stream.Collectors; /** - * Single derivation point for the {@code id_at} weekly partition values a batch of ids resolves to, so a mutation can - * name its own partitions instead of being planned against every part of the table. + * Single derivation point for the {@code id_at} weekly partition value(s) a batch of ids resolves to, so a mutation + * can name its own partitions instead of being planned against every part of the table. {@link #groupByPartition} + * groups ids by the partition each belongs to, for a caller emitting one statement per partition (OPIK-8230) that + * must bind only the ids belonging to it. * *

Mirrors the partition expression of {@code traces_local_v2} / {@code spans_local_v2} exactly — * {@code PARTITION BY toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))}, where {@code id_at} is @@ -55,9 +60,9 @@ *

Why the caller must treat an empty result as "no predicate", never as "no partitions"

* *

The whole value of the derivation is that the partitions a batch resolves to are the only places its rows - * can be; a set that is merely close is a silently skipped delete, not a slower one. So this returns a set only when - * every id in the batch is one it can derive exactly, and empty otherwise — leaving the caller to emit its unbounded - * form, which is always correct and merely slower. The two rejections are:

+ * can be; a grouping that is merely close is a silently skipped delete, not a slower one. So this returns non-empty + * only when every id in the batch is one it can derive exactly, and empty otherwise — leaving the caller to emit its + * unbounded form, which is always correct and merely slower. The two rejections are:

*
    *
  • Any id that is not a UUIDv7. Its high 48 bits are not a timestamp, and {@code UUIDv7ToDateTime} * returns {@code 1970-01-01} for it rather than throwing, so the row sits in the epoch partition while the bits @@ -101,10 +106,17 @@ public class WeeklyPartitions { private static final long ID_AT_LEGACY_MODULUS = 1L << 32; /** - * The weekly partition values the ids resolve to — under each {@code id_at} type the mutation may run against (see - * the class javadoc) — or empty if the batch contains an id whose partition cannot be derived exactly. In the empty - * case the caller must omit its partition predicate entirely. An empty batch yields empty for the same reason: - * there is nothing to bound the mutation to. + * Groups the ids by the weekly partition value(s) each resolves to — under each {@code id_at} type the mutation + * may run against (see the class javadoc) — or empty if the batch contains an id whose partition cannot be + * derived exactly. In the empty case the caller must omit its partition predicate entirely and emit the + * unbounded form. An empty batch yields empty for the same reason: there is nothing to bound the mutation to. + * {@code IN PARTITION} accepts exactly one partition per statement (OPIK-8230), so a caller emitting one + * statement per partition needs to know which ids go in each, not merely their union. + *

    + * An id past {@link #ID_AT_LEGACY_MODULUS} appears in two groups — one per {@code id_at} representation + * it resolves to. That is safe, not a widening bug: a statement scoped to the group that does not actually + * contain the id's row is a no-op there, never a wrong deletion, since the row-matching {@code (project_id, id)} + * predicate inside each statement is unchanged regardless of which partition the statement targets. *

    * A {@code null} batch throws rather than reading as empty, which is the one place this class is deliberately * intolerant. Empty is a documented answer — "this batch cannot be pruned, emit the unbounded form" — and a @@ -114,36 +126,61 @@ public class WeeklyPartitions { * * @throws NullPointerException if {@code ids} is null. */ - public static Optional> of(@NonNull Collection ids) { - var partitions = new HashSet(); + public static Optional>> groupByPartition(@NonNull Collection ids) { + var grouped = new HashMap>(); for (UUID id : ids) { - if (id == null || id.version() != 7) { - return Optional.empty(); - } - // UUIDv7: the high 48 bits are the unix epoch in milliseconds. `>>> 16` reads them unsigned, so the value - // is in [0, 2^48) — never negative, and never large enough for Instant.ofEpochSecond to overflow. That is - // also why neither derivation checks a floor: the smallest id_at any UUIDv7 can carry is the epoch, and - // even its Monday (1969-12-29) is comfortably inside Date32 on both schemas. - long epochMilli = id.getMostSignificantBits() >>> 16; - if (epochMilli >= ID_AT_CEILING) { + var idPartitions = partitionsOf(id); + if (idPartitions.isEmpty()) { return Optional.empty(); } - // Truncating to whole seconds is the column's own conversion (DateTime64(0) / DateTime both store seconds) - // and cannot move a value into an earlier day, so it never changes the week. - long epochSecond = epochMilli / 1_000L; - partitions.add(weeklyPartitionOf(epochSecond)); // DateTime64(0, 'UTC') — the partitioned successor - // Only past the 32-bit range do the two columns disagree; below it the modulo is the identity, so the - // branch is the invariant stated as code rather than an optimisation: an ordinary id CANNOT widen the set. - if (epochSecond >= ID_AT_LEGACY_MODULUS) { - partitions.add(weeklyPartitionOf(epochSecond % ID_AT_LEGACY_MODULUS)); // DateTime('UTC') — legacy + for (Long partition : idPartitions.get()) { + grouped.computeIfAbsent(partition, _ -> new HashSet<>()).add(id); } } - // Set.copyOf, not the working HashSet: what escapes here decides which partitions a DELETE mutation touches, so - // a caller holding a mutable reference could narrow the set after it was derived and turn a correct delete into - // a silent no-op. Immutable by default per apps/opik-backend/AGENTS.md, and the accumulator stays local. - return partitions.isEmpty() ? Optional.empty() : Optional.of(Set.copyOf(partitions)); + if (grouped.isEmpty()) { + return Optional.empty(); + } + // Immutable outer map AND immutable per-partition sets: what escapes here decides which partitions a DELETE + // mutation touches, so a caller holding a mutable reference could narrow it after derivation and turn a + // correct delete into a silent no-op. Collectors.toUnmodifiableMap's values must themselves be made + // unmodifiable explicitly; it does not do so for you. + return Optional.of(grouped.entrySet().stream() + .collect(Collectors.toUnmodifiableMap(Map.Entry::getKey, entry -> Set.copyOf(entry.getValue())))); + } + + /** + * The weekly partition value(s) a single id resolves to under each {@code id_at} type the mutation may meet (see + * class javadoc) — extracted from {@link #groupByPartition} so the per-id {@code id_at} math reads separately + * from the grouping it feeds. Empty exactly when the id cannot be derived exactly: not a UUIDv7, or an embedded + * timestamp at or past {@link #ID_AT_CEILING}. + */ + private static Optional> partitionsOf(UUID id) { + if (id == null || id.version() != 7) { + return Optional.empty(); + } + // UUIDv7: the high 48 bits are the unix epoch in milliseconds. `>>> 16` reads them unsigned, so the value + // is in [0, 2^48) — never negative, and never large enough for Instant.ofEpochSecond to overflow. That is + // also why neither derivation checks a floor: the smallest id_at any UUIDv7 can carry is the epoch, and + // even its Monday (1969-12-29) is comfortably inside Date32 on both schemas. + long epochMilli = id.getMostSignificantBits() >>> 16; + if (epochMilli >= ID_AT_CEILING) { + return Optional.empty(); + } + // Truncating to whole seconds is the column's own conversion (DateTime64(0) / DateTime both store seconds) + // and cannot move a value into an earlier day, so it never changes the week. + long epochSecond = epochMilli / 1_000L; + // DateTime64(0, 'UTC') — the partitioned successor. Only past the 32-bit range do the two columns disagree; + // below it the modulo is the identity, so the branch is the invariant stated as code rather than an + // optimisation: an ordinary id CANNOT widen the set. Immutable on the way out, for the same reason + // groupByPartition copies its own result — see there. Set.of throws on a duplicate, which is safe here + // rather than merely untested: the two arguments are weeks 2^32 seconds (~136 years) apart, so they cannot + // be the same week. + return Optional.of(epochSecond >= ID_AT_LEGACY_MODULUS + ? Set.of(weeklyPartitionOf(epochSecond), + weeklyPartitionOf(epochSecond % ID_AT_LEGACY_MODULUS)) // DateTime('UTC') — legacy + : Set.of(weeklyPartitionOf(epochSecond))); } /** diff --git a/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesLegacyTablePruningMutationTest.java b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesLegacyTablePruningMutationTest.java index fe67ab37cb5..ccc4dea6b41 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesLegacyTablePruningMutationTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesLegacyTablePruningMutationTest.java @@ -41,8 +41,6 @@ import java.util.Set; import java.util.UUID; import java.util.function.Consumer; -import java.util.regex.Pattern; -import java.util.stream.Collectors; import static com.comet.opik.api.resources.utils.AuthTestUtils.mockTargetWorkspace; import static org.assertj.core.api.Assertions.assertThat; @@ -84,40 +82,9 @@ class TracesLegacyTablePruningMutationTest { /** A UUIDv7 carrying id_at 2200-01-01 — the litellm shape, and the id the legacy 32-bit column wraps. */ private static final UUID FAR_FUTURE_ID = UUID.fromString("0699eb8a-59dd-7215-8000-03b8d2a8d5e2"); - /** - * The week that id partitions into under a {@code DateTime64} {@code id_at} — its honest one, and the only value the - * flag-gated predicate used to carry. Stated as a literal because it is what ClickHouse returned for - * {@code toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))} on that id, not something re-derived - * here. - */ - private static final long HONEST_WEEK = 21991230L; - - /** - * The week the same id partitions into on this table: {@code CAST(UUIDv7ToDateTime(...) AS DateTime('UTC'))} - * wraps 2200-01-01 to 2063-11-25, a Wednesday, whose Monday is 2063-11-19. This is the value that makes the delete - * below land, and the reason the predicate needs no flag. - */ - private static final long LEGACY_WEEK = 20631119L; - /** {@link UUID#randomUUID()} is a v4 by definition: no embedded timestamp to derive a partition from. */ private static final UUID NON_V7_ID = UUID.randomUUID(); - /** - * The partition-key fragment the DAO's template emits. Only ever compared against the statement read back - * from {@code system.query_log} — never spliced into a query this suite runs. - */ - private static final String PARTITION_PREDICATE = "toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))"; - - /** - * The {@code IN} clause the DAO emitted, captured to end of line: the predicate sits on its own line in the - * template with {@code SETTINGS log_comment} on the next, so the line boundary delimits it exactly. - */ - private static final Pattern EMITTED_IN_CLAUSE = Pattern.compile( - Pattern.quote(PARTITION_PREDICATE) + "\\s+IN\\s+([^\\n]*)"); - - /** A weekly partition value as it appears in SQL — {@code yyyyMMdd}, so always eight digits. */ - private static final Pattern PARTITION_VALUE = Pattern.compile("\\d{8}"); - private static final String INSERT_RAW_TRACE = """ INSERT INTO traces (workspace_id, project_id, id) VALUES (:workspace_id, :project_id, :id) @@ -213,45 +180,46 @@ void afterAll() { } @Test - @DisplayName("a far-future row on the legacy table is deleted by the same pruned statement") - void farFutureRowOnLegacyTableIsDeletedByAPrunedStatement() { - // The one test that shows the union is load-bearing rather than decorative. With only the honest week in the - // set, this delete matches nothing on this table and reports success - which is exactly the failure the - // configuration flag used to exist to avoid, now avoided by the predicate itself. + @DisplayName("a far-future row on the legacy table is still deleted, by the unbounded form") + void farFutureRowOnLegacyTableIsStillDeleted() { + // OPIK-8230 changes what this test proves. IN PARTITION cannot be used against a table with no partition key + // at all - ClickHouse rejects it outright (Code 248 INVALID_PARTITION_VALUE, "Wrong number of fields in the + // partition expression: 1, must be: 0") - so TraceDAO#deleteBatch now gates the scoped path on the mutation's + // OWN resolved table, not merely on whether WeeklyPartitions can derive a value. WeeklyPartitions itself stays + // schema-agnostic (it still derives BOTH representations for a far-future id, honest and legacy-wrapped, on + // purpose - see its Javadoc), but the DAO never spends that derivation here: on the legacy table every delete + // takes the unbounded form, correctly, and this is that guarantee restated for the topology it actually + // guards. // - // The project and its id come from the real ingestion path - create a trace through the endpoint, then read the - // project id back off it - so the delete runs against a project that exists and a genuine UUIDv7 project id, - // not a fabricated one. + // The project and its id come from the real ingestion path - create a trace through the endpoint, then read + // the project id back off it - so the delete runs against a project that exists and a genuine UUIDv7 project + // id, not a fabricated one. var projectId = projectIdOf(createTrace()); - // Only the far-future row is raw, because ingestion rejects it by design (24h window). A recent id would prove - // nothing here: the legacy id_at is accurate for one, so even an honest-week-only predicate would match it. + // Only the far-future row is raw, because ingestion rejects it by design (24h window). insertRawTrace(projectId, FAR_FUTURE_ID); assertThat(liveRowCount(projectId, FAR_FUTURE_ID)).as("the far-future row is seeded").isEqualTo("1"); delete(Set.of(Pair.of(projectId, FAR_FUTURE_ID))); assertThat(liveRowCount(projectId, FAR_FUTURE_ID)) - .as("it is deleted - so the predicate named the week THIS table filed it under, not only the honest one") + .as("it is deleted - the unbounded form is always correct, merely unscoped") .isEqualTo("0"); var sql = lastTraceDeleteSql(FAR_FUTURE_ID); assertThat(sql) - .as("the predicate is emitted here too - there is no schema flag holding it back any more") - .contains(PARTITION_PREDICATE); - // Asserted as an exact set, both ways round. Only the legacy week can match a row on this table, so a set - // missing it would fail the delete above; and a set missing the honest week would pass here while breaking the - // post-EXCHANGE suite, which is the pair this has to stay consistent with. - assertThat(boundPartitionsOf(sql)) - .as("both representations of the same id: the honest week and the one the 32-bit column wraps it to") - .containsExactlyInAnyOrder(HONEST_WEEK, LEGACY_WEEK); + .as("no id_at predicate of any kind, and no IN PARTITION clause: the legacy table has no partition" + + " key for either to name") + .doesNotContain("id_at") + .doesNotContain("IN PARTITION"); } @Test - @DisplayName("an ordinary row is deleted by a single-week statement, since both id_at types agree below 2106") - void ordinaryRowIsBoundedToOneWeek() { - // The counterweight: the union widens only where the two representations differ, so real traffic binds exactly - // what it bound before. A regression that added the wrapped week unconditionally would show up here. + @DisplayName("an ordinary row is also deleted by the unbounded form, same as a far-future one") + void ordinaryRowIsAlsoDeletedUnbounded() { + // The counterweight to the far-future case above: on the legacy table there is no "recent id gets the scoped + // form, far-future gets the unbounded one" split. Every delete here is unbounded, regardless of era - proving + // that ONLY with a far-future row would leave open whether an ordinary one still worked. var target = createTrace(); var projectId = projectIdOf(target); assertThat(liveRowCount(projectId, target.id())).as("the row is there").isEqualTo("1"); @@ -260,10 +228,10 @@ void ordinaryRowIsBoundedToOneWeek() { assertThat(liveRowCount(projectId, target.id())).as("and it is deleted").isEqualTo("0"); var sql = lastTraceDeleteSql(target.id()); - assertThat(sql).as("the mutation carries the partition predicate").contains(PARTITION_PREDICATE); - assertThat(boundPartitionsOf(sql)) - .as("one week, not two: a recent id_at is inside the 32-bit range, so the two derivations coincide") - .hasSize(1); + assertThat(sql) + .as("no id_at predicate of any kind, and no IN PARTITION clause") + .doesNotContain("id_at") + .doesNotContain("IN PARTITION"); } @Test @@ -291,9 +259,9 @@ void nonV7IdDisablesPruning() { assertThat(sql) .as("the unbounded form carries no id_at predicate of any kind") .doesNotContain("id_at"); - assertThat(EMITTED_IN_CLAUSE.matcher(sql).find()) - .as("and no partition IN clause: %s", sql) - .isFalse(); + assertThat(sql) + .as("and no IN PARTITION clause: %s", sql) + .doesNotContain("IN PARTITION"); } /** A trace through the real ingestion path, with the fields podam cannot fill sensibly cleared. */ @@ -306,26 +274,6 @@ private Trace createTrace() { return trace; } - /** - * The partition values actually bound into the emitted {@code IN} clause, so a test can assert the set is exact - * rather than merely inclusive. The driver substitutes bound values into the query text client-side, which is why - * they are readable here at all; the two assertions below are what make a change in that behaviour say so plainly - * instead of quietly turning every set assertion into a tautology on an empty set. - */ - private static Set boundPartitionsOf(String sql) { - var clause = EMITTED_IN_CLAUSE.matcher(sql); - assertThat(clause.find()) - .as("the delete SQL carries the partition predicate followed by an IN clause:%n%s", sql) - .isTrue(); - var bound = PARTITION_VALUE.matcher(clause.group(1)).results() - .map(match -> Long.parseLong(match.group())) - .collect(Collectors.toUnmodifiableSet()); - assertThat(bound) - .as("the IN clause carries inlined partition values — got '%s'", clause.group(1)) - .isNotEmpty(); - return bound; - } - /** Invokes the DAO under a workspace/user context, as {@code TraceService} does for the live delete path. */ private void delete(Set> projectIdTraceIdPairs) { template.nonTransaction(connection -> traceDAO.delete(projectIdTraceIdPairs, connection)) diff --git a/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesPartitionPruningMutationTest.java b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesPartitionPruningMutationTest.java index cde05fff7e6..84af1247613 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesPartitionPruningMutationTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/infrastructure/TracesPartitionPruningMutationTest.java @@ -20,17 +20,13 @@ import com.comet.opik.infrastructure.auth.RequestContext; import com.comet.opik.infrastructure.db.TransactionTemplateAsync; import com.comet.opik.podam.PodamFactoryUtils; -import com.comet.opik.utils.JsonUtils; import com.comet.opik.utils.WeeklyPartitions; import com.comet.opik.utils.template.TemplateUtils; -import com.fasterxml.jackson.annotation.JsonIgnoreProperties; -import com.fasterxml.jackson.annotation.JsonProperty; -import com.fasterxml.jackson.databind.JsonNode; import com.redis.testcontainers.RedisContainer; import io.r2dbc.spi.Statement; -import lombok.Builder; import org.apache.commons.lang3.RandomStringUtils; import org.apache.commons.lang3.tuple.Pair; +import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.DisplayName; @@ -51,6 +47,8 @@ import ru.vyarus.dropwizard.guice.test.jupiter.ext.TestDropwizardAppExtension; import uk.co.jemos.podam.api.PodamFactory; +import java.time.Duration; +import java.time.Instant; import java.time.LocalDate; import java.time.ZoneOffset; import java.util.ArrayList; @@ -85,7 +83,7 @@ * the DAO's predicate resolved to any partition other than the one ClickHouse filed a row under, the mutation would * select the wrong parts and that row would survive, so * {@link #deleteClearsEveryEraAndBindsExactlyThosePartitions} passing is the agreement between the migration's - * {@code PARTITION BY} as installed, the DAO's predicate, and {@link WeeklyPartitions#of}. + * {@code PARTITION BY} as installed, the DAO's predicate, and {@link WeeklyPartitions#groupByPartition}. * *

    Each test then pairs that with the SQL ClickHouse actually received, because rows alone cannot see pruning * silently stop — a delete that stopped bounding itself is still correct, just slow, and that is the regression this @@ -144,15 +142,15 @@ class TracesPartitionPruningMutationTest { private static final String USER = "user-" + RandomStringUtils.secure().nextAlphanumeric(32); /** - * The partition-key fragment the DAO's template emits. Only ever compared against the statement read back - * from {@code system.query_log} — never spliced into a query this suite runs. Verbatim comparison is safe because - * {@code query_log} stores the query as submitted, so both sides are the DAO's own template text. + * Matches the {@code IN PARTITION } clause the scoped delete template emits (OPIK-8230), capturing the + * eight-digit {@code yyyyMMdd} partition value. Only ever compared against the statement read back from + * {@code system.query_log} - never spliced into a query this suite runs. */ - private static final String PARTITION_PREDICATE = "toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))"; + private static final Pattern IN_PARTITION_CLAUSE = Pattern.compile("IN\\s+PARTITION\\s+(\\d{8})"); // The suite's whole SQL surface, per .agents/skills/opik-backend/SKILL.md "SQL Query Construction": one text block // per query, every varying value a :placeholder. There are no StringTemplate fragments and no interpolation at all, - // because nothing here re-implements the DAO's predicate - PARTITION_PREDICATE is only ever compared against the + // because nothing here re-implements the DAO's predicate - IN_PARTITION_CLAUSE is only ever compared against the // statement the DAO emitted, never spliced into a query of ours. // // The EXCHANGE/RENAME pair in installPartitionedSuccessorUnderTraces() stays as inline literals on purpose: they @@ -196,18 +194,26 @@ AND query LIKE concat('%', :trace_id, '%') """; /** - * The newest {@code delete_traces} statement carrying exactly {@code pairs_size} pairs. A request larger than - * {@link com.comet.opik.infrastructure.FilterUtils#ANALYTICS_DELETE_BATCH_SIZE} is chunked by the DAO into one - * statement per chunk, and pruning is derived per chunk — so a test that reads only one statement cannot see - * the second chunk at all. + * Every {@code delete_traces} statement finished since {@code since} whose query text mentions + * {@code projectId}, in submission order - the multi-statement counterpart of {@link #LAST_TRACE_DELETE}, since + * one {@code delete()} call can now emit more than one statement (OPIK-8230). + *

    + * Scoped by project id, not just by time: {@code WORKSPACE_ID} is one constant shared by every test in this + * class, so a time window alone is not exclusive to one test's own statements - a delete finished by a + * DIFFERENT test can still surface inside this window if system.query_log's buffered writes become visible only + * once a LATER test's own {@code SYSTEM FLUSH LOGS} call flushes them, even though its recorded + * {@code event_time_microseconds} is genuinely earlier. Each test mints its own fresh, collision-free project id + * ({@code ID_GENERATOR.generateId()}), which is literally embedded in the emitted query text, so filtering on it + * closes that gap the same way {@link #lastTraceDeleteSql} already does with a trace id. */ - private static final String DELETE_BY_PAIR_COUNT = """ + private static final String ALL_TRACE_DELETES_SINCE = """ SELECT query FROM system.query_log - WHERE log_comment LIKE concat('delete_traces:%pairs_size=', :pairs_size) + WHERE log_comment LIKE 'delete_traces:%' AND type = 'QueryFinish' - ORDER BY event_time_microseconds DESC - LIMIT 1 + AND event_time_microseconds >= :since + AND query LIKE concat('%', :project_id, '%') + ORDER BY event_time_microseconds """; /** @@ -224,46 +230,47 @@ SELECT toString(uniqExact(id)) """; /** - * The {@code IN} clause the DAO emitted, captured to end of line: the predicate sits on its own line in the - * template with {@code SETTINGS log_comment} on the next, so the line boundary delimits it exactly. Read this way - * rather than by matching a bracket style, because the driver's rendering of a {@code Long[]} is its own choice — - * what matters is which partition values are in the clause, not how it punctuates them. + * The physical part-level proof of pruning (OPIK-8230's own point): how many {@code MutatePart} events + * {@code table} logged in a window, and how many distinct partitions they span. {@code EXPLAIN} cannot see this - + * it reports what the read planner would select for a {@code SELECT}, a different layer from what a mutation is + * registered against, which is exactly the gap this suite used to fall into (see class Javadoc). Windowed by + * {@code event_time}, not correlated by {@code query_id}: a lightweight delete's mutation executes asynchronously + * under the mutations subsystem, so its {@code part_log} rows do not reliably carry the submitting statement's + * {@code query_id} - a finding from this ticket's own production investigation, not a guess. */ - private static final Pattern EMITTED_IN_CLAUSE = Pattern.compile( - Pattern.quote(PARTITION_PREDICATE) + "\\s+IN\\s+([^\\n]*)"); - - /** - * The emitted statement's shape, so its {@code WHERE} clause can be lifted verbatim and re-asked as a - * {@code SELECT}: {@code EXPLAIN} does not accept a mutation. Captures the target table too, since the DAO picks - * {@code traces} or {@code traces_local} depending on the wrap flag. - */ - private static final Pattern DELETE_SHAPE = Pattern.compile( - "DELETE\\s+FROM\\s+(\\S+)\\s+(WHERE\\b.*?)\\s+SETTINGS\\b", Pattern.DOTALL); + private static final String MUTATE_PART_EVENTS_SINCE = """ + SELECT toString(count()), toString(uniqExact(partition_id)) + FROM system.part_log + WHERE database = currentDatabase() + AND table = :table + AND event_type = 'MutatePart' + AND event_time >= :since + """; - /** {@code EXPLAIN} index entries that reflect partition-level part selection. */ - private static final Set PARTITION_INDEX_TYPES = Set.of("MinMax", "Partition"); + /** How many active parts {@code table} currently holds - the "everything" a fallback delete should visit. */ + private static final String ACTIVE_PART_COUNT = """ + SELECT toString(count()) + FROM system.parts + WHERE database = currentDatabase() + AND table = :table + AND active + """; /** - * Asks the planner how many parts the DAO's partition predicate selects. Declared once as a text block, per - * {@code .agents/skills/opik-backend/SKILL.md}: the table {@code } and the predicate - * {@code } are fragments and go through {@link TemplateUtils#newST}, the partition - * values are values and are bound — nothing is spliced with {@code .formatted(...)}. - *

    - * The predicate fragment is {@link #PARTITION_PREDICATE}, and using the constant does not re-author what is under - * test: every caller has already asserted the emitted statement contains that exact text, so the constant - * is pinned to the DAO's own SQL by assertion rather than by string surgery on it. The partition values come from - * the emitted statement too, parsed by {@link #boundPartitionsOf} and bound here. - *

    - * The DAO's {@code workspace_id} and {@code (project_id, id)} predicates are deliberately not reproduced. They are - * sort-key filters, not partition filters, so they cannot change partition selection — and leaving them out makes - * the unbounded case a full scan, which is the conservative direction for an assertion that the fallback prunes - * nothing. + * Whether {@code table} has any mutation still in flight. Used to settle the estate before starting a timed + * {@link #mutatePartActivitySince} window: a mutation from an EARLIER test in this suite can still be writing + * {@code MutatePart} rows to {@code system.part_log} after its submitting statement has already returned - this + * class runs every test against the same shared table (see class Javadoc), and nothing here sets + * {@code lightweight_deletes_sync} the way the retention sweep does, so a prior test's mutation completing late + * would otherwise land inside a LATER test's window and inflate its part/partition counts. Not a claim about + * production: production's real workload has no "between tests" moment to wait for. */ - private static final String EXPLAIN_SELECTED_PARTS = """ - EXPLAIN indexes = 1, json = 1 - SELECT id - FROM

    - WHERE IN :partitions + private static final String PENDING_MUTATIONS_COUNT = """ + SELECT toString(count()) + FROM system.mutations + WHERE database = currentDatabase() + AND table = :table + AND NOT is_done """; /** @@ -277,9 +284,6 @@ SELECT toString(uniqExact(id)) ENGINE = Distributed('{cluster}', '', 'traces_local', sipHash64(project_id)) """; - /** A weekly partition value as it appears in SQL — {@code yyyyMMdd}, so always eight digits. */ - private static final Pattern PARTITION_VALUE = Pattern.compile("\\d{8}"); - /** * One id per era the derivation has to get right, with the partition each resolves to. Not interchangeable samples: * {@code toMonday} agrees with the {@code Date32} expression across the ordinary calendar and diverges only @@ -449,23 +453,60 @@ private static long partitionNameOf(LocalDate monday) { } /** - * The partition values actually bound into the emitted {@code IN} clause, so a test can assert the set is exact - * rather than merely inclusive. The driver substitutes bound values into the query text client-side, which is why - * they are readable here at all; the two assertions below are what make a change in that behaviour say so plainly - * instead of quietly turning every set assertion into a tautology on an empty set. + * The single partition value an {@code IN PARTITION} statement names. {@code IN PARTITION} takes exactly one + * partition per statement (OPIK-8230) - unlike the old {@code WHERE ... IN (...)} predicate this replaces, there + * is never a set to parse, only ever one value or none. */ - private static Set boundPartitionsOf(String sql) { - var clause = EMITTED_IN_CLAUSE.matcher(sql); + private static long boundPartitionOf(String sql) { + var clause = IN_PARTITION_CLAUSE.matcher(sql); assertThat(clause.find()) - .as("the delete SQL carries the partition predicate followed by an IN clause:%n%s", sql) + .as("the delete SQL carries an IN PARTITION clause:%n%s", sql) .isTrue(); - var bound = PARTITION_VALUE.matcher(clause.group(1)).results() - .map(match -> Long.parseLong(match.group())) - .collect(Collectors.toUnmodifiableSet()); - assertThat(bound) - .as("the IN clause carries inlined partition values — got '%s'", clause.group(1)) - .isNotEmpty(); - return bound; + return Long.parseLong(clause.group(1)); + } + + /** + * How many {@code MutatePart} events {@code table} logged since {@code since}, and how many distinct partitions + * they span - the part-level proof that a delete actually scoped its mutation, not merely that its {@code WHERE} + * clause looks right. See {@link #MUTATE_PART_EVENTS_SINCE}'s Javadoc for why this is windowed rather than + * correlated by {@code query_id}. + *

    + * {@code SYSTEM FLUSH LOGS} first: {@code system.part_log} is buffered and flushes on its own schedule, so a read + * immediately after a delete can race the write - the same reason {@link #lastTraceDeleteSql} flushes before + * reading {@code system.query_log}. + */ + private MutatePartActivity mutatePartActivitySince(String table, Instant since) { + execute("SYSTEM FLUSH LOGS", _ -> { + }); + return template.nonTransaction(connection -> { + var statement = connection.createStatement(MUTATE_PART_EVENTS_SINCE) + .bind("table", table) + .bind("since", since.atOffset(ZoneOffset.UTC).toLocalDateTime()); + return Mono.from(statement.execute()) + .flatMap(result -> Mono.from(result.map((row, _) -> new MutatePartActivity( + Integer.parseInt(row.get(0, String.class)), + Integer.parseInt(row.get(1, String.class)))))); + }).block(); + } + + private int activePartCountOf(String table) { + return Integer.parseInt(queryOneString(ACTIVE_PART_COUNT, statement -> statement.bind("table", table))); + } + + /** + * Blocks until {@code table} has no in-flight mutation, polling {@link #PENDING_MUTATIONS_COUNT}. See that + * constant's Javadoc for why a timed MutatePart window needs this settle point. + */ + private void waitForMutationsToSettle(String table) { + Awaitility.await() + .atMost(Duration.ofSeconds(30)) + .pollInterval(Duration.ofMillis(200)) + .until(() -> "0".equals(queryOneString(PENDING_MUTATIONS_COUNT, + statement -> statement.bind("table", table)))); + } + + /** {@code MutatePart} events observed in a window: how many parts were touched, and how many partitions. */ + private record MutatePartActivity(int parts, int partitions) { } /** Invokes the DAO under a workspace/user context, as {@code TraceService} does for the live delete path. */ @@ -505,24 +546,22 @@ private String lastTraceDeleteSql(UUID traceId) { } /** - * The newest {@code delete_traces} statement that carried exactly {@code pairsSize} pairs. Chunks are identified by - * their pair count rather than by a contained id, because the point is to inspect a specific chunk of one - * request — and the DAO stamps each chunk's size into its {@code log_comment}. - *

    - * The returned text is truncated for large statements. ClickHouse caps {@code query_log.query} at - * {@code log_queries_cut_to_length} (100,000 bytes by default), and a full 10,000-pair chunk inlines to ~762 KiB, - * so anything at the tail of such a statement — the partition predicate included — is simply absent. Only assert on - * the text of statements small enough to be recorded whole. + * Every {@code delete_traces} statement ClickHouse finished since {@code since} whose query text mentions + * {@code projectId}. A single {@code delete()} call can now emit more than one statement - one per partition the + * batch resolves to (OPIK-8230) - so a caller that must see the whole set can no longer rely on "the newest + * one" the way {@link #lastTraceDeleteSql} does. See {@link #ALL_TRACE_DELETES_SINCE}'s Javadoc for why this is + * ALSO scoped by project id, not just by time. */ - private String deleteSqlForChunkOf(int pairsSize) { + private List deleteSqlsSince(Instant since, UUID projectId) { execute("SYSTEM FLUSH LOGS", _ -> { }); - var sql = queryOneString(DELETE_BY_PAIR_COUNT, - statement -> statement.bind("pairs_size", String.valueOf(pairsSize))); - assertThat(sql) - .as("query_log holds a delete_traces statement with pairs_size=%s", pairsSize) - .isNotBlank(); - return sql; + return template.stream(connection -> { + var statement = connection.createStatement(ALL_TRACE_DELETES_SINCE) + .bind("since", since.atOffset(ZoneOffset.UTC).toLocalDateTime()) + .bind("project_id", projectId.toString()); + return Flux.from(statement.execute()) + .flatMap(result -> result.map((row, _) -> row.get("query", String.class))); + }).collectList().block(); } /** {@code "1"} while a live (non-lightweight-deleted) row exists for the id, {@code "0"} once it is gone. */ @@ -670,70 +709,6 @@ private void insertRawTrace(UUID projectId, UUID id) { .bind("id", id.toString())); } - /** - * Parts the planner selects for the DAO's own statement, or empty when it reports no partition index at all. - *

    - * {@code SELECT id} rather than {@code count()}, so no trivial-count optimisation can answer from metadata without - * selecting parts at all. - *

    - * Only {@code MinMax} and {@code Partition} entries are considered, and {@code PrimaryKey} is deliberately - * excluded: the DAO's {@code WHERE} also filters {@code workspace_id} and {@code (project_id, id)}, which are the - * sort key, so {@code PrimaryKey} prunes parts for the unbounded statement too — counting it would make the - * fallback look pruned and destroy the discrimination this test rests on. Across the entries that do qualify it - * takes the smallest selected count and the largest initial count, so it does not depend on which of the two - * reports the pruning. - *

    - * Empty is a meaningful answer rather than a failure: the {@code Indexes} block carries a partition entry only when - * the query filters on the partition key, so its absence is exactly what the fallback should produce. - */ - private Optional partsSelectedBy(String daoDeleteSql) { - var shape = DELETE_SHAPE.matcher(daoDeleteSql); - assertThat(shape.find()) - .as("the emitted statement has the expected DELETE shape:%n%s", daoDeleteSql) - .isTrue(); - Set bound = EMITTED_IN_CLAUSE.matcher(daoDeleteSql).find() - ? boundPartitionsOf(daoDeleteSql) - : Set.of(); - - var explainSql = TemplateUtils.newST(EXPLAIN_SELECTED_PARTS) - .add("table", shape.group(1)); - if (!bound.isEmpty()) { - explainSql.add("partition_expression", PARTITION_PREDICATE); - } - var sql = explainSql.render(); - - var explainRows = template.stream(connection -> { - var statement = connection.createStatement(sql); - if (!bound.isEmpty()) { - statement.bind("partitions", bound.toArray(Long[]::new)); - } - return Flux.from(statement.execute()) - .flatMap(result -> result.map((row, _) -> row.get("explain", String.class))); - }) - .collectList() - .block(); - var explain = String.join("\n", explainRows); - - var indexes = JsonUtils.getJsonNodeFromString(explain).findValue("Indexes"); - if (indexes == null) { - return Optional.empty(); - } - SelectedParts partition = null; - for (JsonNode index : indexes) { - if (!PARTITION_INDEX_TYPES.contains(index.path("Type").asText()) || !index.has("Selected Parts")) { - continue; - } - var entry = JsonUtils.treeToValue(index, SelectedParts.class); - partition = partition == null - ? entry - : partition.toBuilder() - .selected(Math.min(partition.selected(), entry.selected())) - .total(Math.max(partition.total(), entry.total())) - .build(); - } - return Optional.ofNullable(partition); - } - /** * First column of the first row, as a string. Every read in this suite is a single scalar, so this is the only * mapper needed; values go in as binds. @@ -755,17 +730,6 @@ private void execute(String sql, Consumer binder) { }).block(); } - /** - * The part counts {@code EXPLAIN indexes = 1, json = 1} reports for one index entry: how many parts the query - * started from, and how many survived pruning. - */ - @Builder(toBuilder = true) - @JsonIgnoreProperties(ignoreUnknown = true) - private record SelectedParts( - @JsonProperty("Selected Parts") int selected, - @JsonProperty("Initial Parts") int total) { - } - @Test @DisplayName("traces is a mutation-rejecting Distributed wrapper, so these deletes ran on traces_local") void distributedTracesRejectsDirectMutation() { @@ -796,23 +760,19 @@ void allUuidV7DeletePrunesAndRemovesTheTargetRow() { .doesNotContain(target.id()) .contains(bystander.id()); var sql = lastTraceDeleteSql(target.id()); - assertThat(sql).as("the mutation carries the partition predicate").contains(PARTITION_PREDICATE); - // Asserted as the SIZE, not as a value read back from WeeklyPartitions: the DAO derives its set from the same - // method, so an expectation taken from it would move with any regression and pass regardless. The value itself - // is already pinned two independent ways - the row above had to go away, which it only does if the predicate - // named the partition ClickHouse filed it under, and deleteClearsEveryEraAndBindsExactlyThosePartitions states - // its expectations as literals. What is left to say here is the property those cannot: an id the endpoint just - // minted is inside the 32-bit range, where the two id_at types agree, so it must widen the set to nothing. - assertThat(boundPartitionsOf(sql)) - .as("bounded to one partition, not two: a recent id_at is a week both id_at types agree on") - .hasSize(1); + // A recent id_at is inside the 32-bit range, where the two id_at types agree, so WeeklyPartitions.groupByPartition + // resolves it to exactly one partition and the DAO emits the scoped IN PARTITION form - not the unbounded + // fallback. boundPartitionOf asserts the clause is present; there is nothing further to assert about its + // cardinality now that IN PARTITION accepts exactly one value per statement by construction. + assertThat(sql).as("the mutation carries an IN PARTITION clause").contains("IN PARTITION"); + boundPartitionOf(sql); } @Test @DisplayName("the DAO's own delete clears every era and binds exactly those partitions") void deleteClearsEveryEraAndBindsExactlyThosePartitions() { // The three-way agreement - the migration's PARTITION BY as installed, the DAO's predicate, and - // WeeklyPartitions.of - asserted through the DAO's own delete rather than by re-evaluating the expression in + // WeeklyPartitions.groupByPartition - asserted through the DAO's own delete rather than by re-evaluating the expression in // test SQL. If the predicate resolved to any partition other than the one ClickHouse filed a row under, the // mutation would select the wrong parts and that row would SURVIVE. So "every row is gone" IS the agreement. // @@ -826,17 +786,24 @@ void deleteClearsEveryEraAndBindsExactlyThosePartitions() { assertThat(ids.stream().map(id -> liveRowCount(projectId, id))) .as("every era is seeded").containsOnly("1"); + var since = Instant.now(); delete(ids.stream().map(id -> Pair.of(projectId, id)).collect(Collectors.toUnmodifiableSet())); assertThat(ids.stream().map(id -> liveRowCount(projectId, id))) .as("every era's row is gone, so the predicate named the partition each was actually filed under") .containsOnly("0"); - var sql = lastTraceDeleteSql(ids.getFirst()); - assertThat(sql).as("the mutation carries the partition predicate").contains(PARTITION_PREDICATE); - // Four values for three eras: 1996 and 2025 are inside the 32-bit range and name one week each, and the 2199 - // id names its legacy week too. Still an exact set, and still not a range across two centuries - which is the - // property under test; the extra value is the batch's own second representation, not a widening of its span. - assertThat(boundPartitionsOf(sql)) + + // One statement per partition the batch resolves to (OPIK-8230), not one statement carrying all four values: + // 1996 and 2025 are inside the 32-bit range and name one week each; the 2199 id names two (its own week AND + // its legacy wrap), so it is the id bound into two DIFFERENT statements rather than the batch widening to a + // fourth id. Four statements, each pairs_size=1 - still an exact set of partitions, and still not a range + // across two centuries, which is the property under test. + var sqls = deleteSqlsSince(since, projectId); + assertThat(sqls) + .as("one statement per partition the batch resolves to: %s", sqls) + .hasSize(4) + .allSatisfy(sql -> assertThat(sql).contains("pairs_size=1")); + assertThat(sqls.stream().map(TracesPartitionPruningMutationTest::boundPartitionOf)) .as("exactly the partitions the batch resolves to, not a range across two centuries") .containsExactlyInAnyOrder( partitionNameOf(ERA_MONDAYS.get(0)), @@ -846,48 +813,59 @@ void deleteClearsEveryEraAndBindsExactlyThosePartitions() { } @Test - @DisplayName("the planner actually prunes, and the fallback provably does not") + @DisplayName("a scoped delete touches far fewer parts than the table holds, and the fallback touches them all") void pruningReachesThePlannerAndTheFallbackDoesNot() { // Correctness and pruning are different claims, and this is the only test that makes the second one. Deletes - // were already correct before OPIK-6901 - what the change buys is parts touched (3,928/3,928 -> 5/3,928 on - // prod-test), so a suite that cannot see pruning stop does not test what this change exists to do. + // were already correct before OPIK-8230 - what the change buys is parts touched (~3,650/~3,650 -> ~1/~3,650 on + // production), so a suite that cannot see pruning stop does not test what this change exists to do. // - // The regression it guards is specific: a migration rewrites the partition expression to something semantically - // identical but textually different, the planner stops recognising the DAO's predicate as the partition key, - // pruning silently stops - and values still agree, so every row is still deleted and every other assertion in - // this suite stays green. That is the property the removed AST pin covered; this asks the planner directly - // instead of inferring it from text. - // - // EXPLAIN does not accept a mutation, so the WHERE clause is lifted verbatim out of the DAO's own emitted - // DELETE and put behind a SELECT - predicate and bound partition values included. Only the verb changes; the - // statement being explained is still the DAO's. Same instrument and record shape as - // TracesLocalV2PartitioningTest. - // One row per era, so the table holds several partitions for the planner to prune between. + // Asked of system.part_log's MutatePart events directly - the layer a MUTATION is actually registered + // against - rather than of EXPLAIN, which reports what the READ planner would select for a SELECT. That + // distinction IS the regression OPIK-8230 exists to fix: this suite's own predecessor asserted EXPLAIN + // pruning and stayed green while every delete still rewrote every part, because EXPLAIN cannot see a + // mutation's part selection at all. See the class Javadoc. + var table = "traces_local"; var projectId = ID_GENERATOR.generateId(); var ids = ERA_MONDAYS.stream().map(TracesPartitionPruningMutationTest::idInWeekOf).toList(); ids.forEach(id -> insertRawTrace(projectId, id)); - // Bounded: one derivable id, so the predicate names a single one of those partitions. + // Bounded: one derivable id, so the mutation is registered against only the partition it resolves to. + waitForMutationsToSettle(table); + var boundedSince = Instant.now(); delete(Set.of(Pair.of(projectId, ids.getFirst()))); - var bounded = partsSelectedBy(lastTraceDeleteSql(ids.getFirst())) - .orElseThrow(() -> new AssertionError( - "EXPLAIN reported no partition index for the bounded delete")); - - // Unbounded: a non-v7 id in the batch, so no predicate at all. Its partner is a different era, so the query_log - // lookup finds this statement rather than the one above. - var partner = ids.get(1); - delete(Set.of(Pair.of(projectId, partner), Pair.of(projectId, NON_V7_ID))); - var unbounded = partsSelectedBy(lastTraceDeleteSql(partner)); - - assertThat(bounded.selected()) - .as("the bounded delete selects fewer parts than the table holds: %s", bounded) - .isLessThan(bounded.total()); - // Shown to discriminate, or it proves nothing - the same trap as `.contains(partition)` and - // `doesNotContain("toDayOfWeek")`. The fallback must not prune: either the planner reports no partition index at - // all, because nothing filters on the key, or it reports every part still selected. - assertThat(unbounded.map(parts -> parts.selected() == parts.total()).orElse(true)) - .as("the fallback prunes nothing: %s", unbounded) - .isTrue(); + var bounded = mutatePartActivitySince(table, boundedSince); + + assertThat(bounded.partitions()) + .as("the bounded delete touched at least the one partition it resolves to: %s", bounded) + .isGreaterThanOrEqualTo(1); + + // Unbounded: a non-v7 id in the batch, so no IN PARTITION at all - the mutation must fall back to visiting + // every part, exactly as it did before OPIK-8230. Its partner is a different era so the two deletes touch + // disjoint rows, but that is incidental here; what is asserted is part count, not row identity. + waitForMutationsToSettle(table); + var totalPartsBeforeUnbounded = activePartCountOf(table); + var unboundedSince = Instant.now(); + delete(Set.of(Pair.of(projectId, ids.get(1)), Pair.of(projectId, NON_V7_ID))); + var unbounded = mutatePartActivitySince(table, unboundedSince); + + assertThat(unbounded.parts()) + .as("the fallback's mutation touches at least every part that existed when it was submitted: %s parts" + + " touched, %s existed", unbounded.parts(), totalPartsBeforeUnbounded) + .isGreaterThanOrEqualTo(totalPartsBeforeUnbounded); + + // The contrast itself, compared directly rather than each measurement against a separately-queried "total + // active parts" snapshot: this class runs every test PER_CLASS against the same shared table (see class + // Javadoc), and that denominator drifts with whatever residue earlier tests left in OTHER partitions - + // ClickHouse can also apply a pending mutation opportunistically during an unrelated background merge, which + // showed up here as bounded.parts() occasionally equalling a stale "total" snapshot even after + // waitForMutationsToSettle saw no mutation still is_done=0. Comparing the two measurements taken moments + // apart in this SAME test, under the SAME shared-state conditions, is not subject to that drift: the + // unbounded mutation's footprint is a superset of whatever existed when it was submitted (proven above), so + // it can only be smaller than the bounded one if pruning had stopped working entirely. + assertThat(bounded.parts()) + .as("the bounded delete touches far fewer parts than the unbounded fallback: bounded=%s, unbounded=%s", + bounded, unbounded) + .isLessThan(unbounded.parts()); } @Test @@ -948,7 +926,7 @@ void topologySetupIsANoOpOnceTheEstateProvidesIt() { .isEqualTo("0"); assertThat(lastTraceDeleteSql(id)) .as("and the delete is still pruned") - .contains(PARTITION_PREDICATE); + .contains("IN PARTITION"); } @Test @@ -988,6 +966,7 @@ void requestSpanningTwoChunksPrunesEachChunkIndependently() { ordered.add(secondChunkRow); ordered.add(secondChunkCompanion); + var since = Instant.now(); delete(ordered.stream().map(id -> Pair.of(projectId, id)).collect(Collectors.toCollection( LinkedHashSet::new))); @@ -998,15 +977,31 @@ void requestSpanningTwoChunksPrunesEachChunkIndependently() { .as("and so is the row in the chunk that pruned") .isEqualTo("0"); - // The full chunk's statement is only checked to exist - that is what shows the request was split at all. - // Nothing about its text can be asserted, for the truncation reason above. - deleteSqlForChunkOf(ANALYTICS_DELETE_BATCH_SIZE); + var sqls = deleteSqlsSince(since, projectId); + // Chunk one fell back to exactly one unbounded statement: the non-v7 id disables pruning for the whole + // chunk, so it never fans out per-partition the way chunk two does below. Identified by the ABSENCE of an + // IN PARTITION clause, not by its pairs_size: ClickHouse truncates query_log.query at + // log_queries_cut_to_length (100,000 bytes), and a full 10,000-pair chunk inlines to ~762 KiB, so + // "pairs_size=10000" - stamped at the tail, in the SETTINGS clause - is not reliably present in what comes + // back. "IN PARTITION" sits right after "DELETE FROM

    ", at the very front of the statement, so its + // absence is readable regardless of truncation. + var chunkOneSqls = sqls.stream().filter(sql -> !sql.contains("IN PARTITION")).toList(); + assertThat(chunkOneSqls) + .as("the full chunk fell back to exactly one unbounded statement: %s", sqls) + .hasSize(1); - var secondChunkSql = deleteSqlForChunkOf(2); - assertThat(secondChunkSql) - .as("the all-derivable chunk prunes even though an earlier chunk could not") - .contains(PARTITION_PREDICATE); - assertThat(boundPartitionsOf(secondChunkSql)) + // Chunk two: one statement per partition its two ids resolve to - the far-future row names two (its own week + // and its legacy wrap) and the companion names one, so three statements total, each pairs_size=1. That makes + // the test bite on the refactor it exists to catch: hoisting the derivation out of the per-chunk concatMap, + // so the whole request is derived once, would let the non-v7 id in chunk one strip pruning from chunk two as + // well - collapsing it back to one unbounded statement, which the size assertion below would catch. + var chunkTwoSqls = sqls.stream().filter(sql -> !chunkOneSqls.contains(sql)).toList(); + assertThat(chunkTwoSqls) + .as("the all-derivable chunk prunes even though an earlier chunk could not - one statement per" + + " partition: %s", chunkTwoSqls) + .hasSize(3) + .allSatisfy(sql -> assertThat(sql).contains("pairs_size=1")); + assertThat(chunkTwoSqls.stream().map(TracesPartitionPruningMutationTest::boundPartitionOf)) .as("bounded to exactly its own two weeks, plus the legacy representation of the far-future one") .containsExactlyInAnyOrder(partitionNameOf(ERA_MONDAYS.getLast()), partitionNameOf(ERA_MONDAYS.get(1)), @@ -1046,9 +1041,9 @@ void underivableIdDisablesPruning(String cause, UUID underivableId) { assertThat(sql) .as("the unbounded form for a %s batch carries no id_at predicate of any kind", cause) .doesNotContain("id_at"); - assertThat(EMITTED_IN_CLAUSE.matcher(sql).find()) - .as("and no partition IN clause: %s", sql) - .isFalse(); + assertThat(sql) + .as("and no IN PARTITION clause: %s", sql) + .doesNotContain("IN PARTITION"); } private static Stream underivableIdDisablesPruning() { diff --git a/apps/opik-backend/src/test/java/com/comet/opik/utils/WeeklyPartitionsTest.java b/apps/opik-backend/src/test/java/com/comet/opik/utils/WeeklyPartitionsTest.java index d95bcba84ae..c47eb5a0927 100644 --- a/apps/opik-backend/src/test/java/com/comet/opik/utils/WeeklyPartitionsTest.java +++ b/apps/opik-backend/src/test/java/com/comet/opik/utils/WeeklyPartitionsTest.java @@ -7,6 +7,7 @@ import org.junit.jupiter.params.provider.MethodSource; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.UUID; import java.util.stream.Stream; @@ -16,8 +17,8 @@ import static org.junit.jupiter.params.provider.Arguments.arguments; /** - * Covers {@link WeeklyPartitions#of}, which derives the weekly partition values a delete batch resolves to so the - * mutation can prune instead of rewriting every part. + * Covers {@link WeeklyPartitions#groupByPartition}, which groups a delete batch's ids by the weekly partition each + * resolves to, so the mutation can prune instead of rewriting every part (OPIK-8230). *

    * The expected values are not hand-computed: each is what ClickHouse itself returned for * {@code toYYYYMMDD(toDate32(id_at) - toIntervalDay(toDayOfWeek(id_at, 1)))} for that id, evaluated once with @@ -32,9 +33,9 @@ class WeeklyPartitionsTest { @ParameterizedTest(name = "{0}") @MethodSource - @DisplayName("matches the partitions ClickHouse computed, under each id_at type") + @DisplayName("groupByPartition matches the partitions ClickHouse computed, under each id_at type") void matchesThePartitionsClickHouseComputed(String era, UUID id, Set expectedPartitions) { - assertThat(WeeklyPartitions.of(List.of(id))).contains(expectedPartitions); + assertThat(WeeklyPartitions.groupByPartition(List.of(id)).map(Map::keySet)).contains(expectedPartitions); } /** @@ -47,8 +48,9 @@ private static Stream matchesThePartitionsClickHouseComputed() { return Stream.of( // The ordinary case: id_at 2026-08-19 (a Wednesday) -> Monday 2026-08-17. Inside the 32-bit DateTime // range, so both column types store the same instant and the id contributes a SINGLE value. That is - // every id real traffic produces, and it is why naming both representations costs nothing: the set a - // normal delete binds is exactly the set it bound before the legacy one was added. + // every id real traffic produces, and it is why naming both representations costs nothing: the + // partitions a normal delete resolves to are exactly the ones it resolved to before the legacy one + // was added. arguments("ordinary id", UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"), Set.of(20260817L)), // Long before Opik existed but well after the Unix epoch, and well inside Date32's 1900 floor: an id // this old prunes like any other, and likewise fits both column types. id_at 1996-02-09 -> 1996-02-05. @@ -56,85 +58,20 @@ private static Stream matchesThePartitionsClickHouseComputed() { // Far-future ids are supported, not excluded, and are the only ones that contribute two values: past // 2106 the two column types disagree, so the batch has to name both weeks or be wrong on one schema - // which is what removed the cutover flag. DateTime64 stores 2200-01-01 -> Monday 2199-12-30; the legacy - // 32-bit DateTime wraps the same id to 2063-11-25 -> Monday 2063-11-19. A set carrying only one of the - // two is a delete that reports success and removes nothing on the other schema. 4.1% of rows on + // 32-bit DateTime wraps the same id to 2063-11-25 -> Monday 2063-11-19. A grouping carrying only one of + // the two is a delete that reports success and removes nothing on the other schema. 4.1% of rows on // prod-test look like this. arguments("far-future id", UUID.fromString("0699eb8a-59dd-7215-8000-03b8d2a8d5e2"), Set.of(21991230L, 20631119L))); } - @Test - @DisplayName("a scattered batch yields the exact set, not a range") - void scatteredBatch() { - // 1996 and 2026 in one batch. An id_at RANGE over this span selected 2,644 of 3,928 parts on prod-test; - // the exact set selected 4. This is the reason the predicate is a set. - assertThat(WeeklyPartitions.of(List.of( - UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"), - UUID.fromString("00bfd451-fa93-7c10-9923-88a219a974c8")))) - .contains(Set.of(20260817L, 19960205L)); - } - - @Test - @DisplayName("a non-v7 id disables pruning for the whole batch") - void nonV7DisablesPruning() { - // All-or-nothing on purpose: deriving a partition from a non-v7 id reads whatever sits in the timestamp - // field, and a wrong partition is a SILENTLY skipped delete. Omitting the predicate keeps the old behaviour. - assertThat(WeeklyPartitions.of(List.of( - UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"), - UUID.fromString("9f527bac-527a-4f92-8875-0fa8af8e4f22")))) // v4 - .isEmpty(); - } - @Test @DisplayName("a single non-v7 id yields no partitions") void singleNonV7() { - assertThat(WeeklyPartitions.of(List.of(UUID.fromString("9f527bac-527a-4f92-8875-0fa8af8e4f22")))) + assertThat(WeeklyPartitions.groupByPartition(List.of(UUID.fromString("9f527bac-527a-4f92-8875-0fa8af8e4f22")))) .isEmpty(); } - @Test - @DisplayName("duplicate ids in the same week collapse to one partition") - void duplicatesCollapse() { - assertThat(WeeklyPartitions.of(List.of( - UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"), - UUID.fromString("01a01a75-6f8e-7f22-9279-ee4f7ca7810d"), - UUID.fromString("01a01a75-609d-7935-8d22-2dd8dfeb2454")))) - .contains(Set.of(20260817L)); - } - - @Test - @DisplayName("an empty batch yields no partitions") - void emptyBatch() { - assertThat(WeeklyPartitions.of(List.of())).isEmpty(); - } - - @Test - @DisplayName("the returned set is immutable, so a delete's partitions cannot be narrowed after derivation") - void returnedSetIsImmutable() { - // Not a general hygiene assertion: this set IS the partition list a DELETE binds, so a caller that removed an - // entry would turn a correct delete into one that matches nothing and reports success. - var partitions = WeeklyPartitions.of(List.of( - UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"), - UUID.fromString("00bfd451-fa93-7c10-9923-88a219a974c8"))) - .orElseThrow(); - - assertThatThrownBy(() -> partitions.remove(20260817L)) - .isInstanceOf(UnsupportedOperationException.class); - assertThat(partitions).containsExactlyInAnyOrder(20260817L, 19960205L); - } - - @Test - @DisplayName("a null batch throws rather than reading as an unprunable one") - void nullBatchThrows() { - // The one intolerant case, and deliberately not folded into the empty result above: empty is a documented - // answer ("emit the unbounded form"), so a caller that lost its batch would get a valid-looking answer, issue a - // correct-but-unbounded mutation, and never learn it had a bug. Matches every other collection-taking method in - // this package, all of which are @NonNull. - assertThatThrownBy(() -> WeeklyPartitions.of(null)) - .isInstanceOf(NullPointerException.class) - .hasMessageContaining("ids"); - } - @Test @DisplayName("the last id_at the column can store still prunes") void lastRepresentableIdStillPrunes() { @@ -142,7 +79,8 @@ void lastRepresentableIdStillPrunes() { // truncates to 23:59:59, whose Date32 is 2299-12-31 (a Sunday) -> Monday 2299-12-25. The legacy DateTime wraps // the same id to 2027-10-18, a Monday. The ceiling check must be exclusive at exactly this point, hence a case // sitting on it. - assertThat(WeeklyPartitions.of(List.of(UUID.fromString("0978a65f-77ff-7abc-8000-000000000001")))) + assertThat(WeeklyPartitions.groupByPartition(List.of(UUID.fromString("0978a65f-77ff-7abc-8000-000000000001"))) + .map(Map::keySet)) .contains(Set.of(22991225L, 20271018L)); } @@ -154,7 +92,7 @@ void firstUnrepresentableIdDisablesPruning() { // itself a Monday, giving 23000101) is a partition the row is NOT in. Pruning off rather than clamped: unlike // the legacy 32-bit wrap, which is a modulo this reproduces exactly, matching ClickHouse here would mean // reproducing its saturation semantics, for ids that should not exist. - assertThat(WeeklyPartitions.of(List.of(UUID.fromString("0978a65f-7800-7abc-8000-000000000001")))) + assertThat(WeeklyPartitions.groupByPartition(List.of(UUID.fromString("0978a65f-7800-7abc-8000-000000000001")))) .isEmpty(); } @@ -164,29 +102,121 @@ void largestUuidV7TimestampDisablesPruning() { // All 48 timestamp bits set: 10889-08-02, the furthest future any UUIDv7 can encode. Read unsigned it is still // only ~2.8e14 ms, far inside Instant's range — so the guard is what excludes it, not an exception, and there // is no input on which this can throw. - assertThat(WeeklyPartitions.of(List.of(UUID.fromString("ffffffff-ffff-7abc-8000-000000000001")))) + assertThat(WeeklyPartitions.groupByPartition(List.of(UUID.fromString("ffffffff-ffff-7abc-8000-000000000001")))) + .isEmpty(); + } + + @Test + @DisplayName("the earliest id a UUIDv7 can carry is inside Date32, so there is no floor to guard") + void earliestUuidV7TimestampIsInRange() { + // All 48 timestamp bits clear: id_at 1970-01-01, the earliest any UUIDv7 can encode (the field is unsigned). + // Its Monday, 1969-12-29, is 70 years above Date32's 1900 floor, so a below-1900 id_at is unreachable by + // construction rather than merely untested — which is why the derivation guards only the ceiling. It is also + // the floor of the legacy 32-bit DateTime, so the wrap leaves it untouched and it contributes one value. + assertThat(WeeklyPartitions.groupByPartition(List.of(UUID.fromString("00000000-0000-7abc-8000-000000000001"))) + .map(Map::keySet)) + .contains(Set.of(19691229L)); + } + + @Test + @DisplayName("groupByPartition puts each ordinary id under its own single partition") + void groupByPartitionOrdinaryBatch() { + var ordinary = UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"); // -> 20260817L + var from1996 = UUID.fromString("00bfd451-fa93-7c10-9923-88a219a974c8"); // -> 19960205L + + assertThat(WeeklyPartitions.groupByPartition(List.of(ordinary, from1996))) + .contains(Map.of( + 20260817L, Set.of(ordinary), + 19960205L, Set.of(from1996))); + } + + @Test + @DisplayName("groupByPartition puts a far-future id under both of its partitions") + void groupByPartitionFarFutureIdAppearsTwice() { + // Same id as matchesThePartitionsClickHouseComputed's far-future case: id_at 2200-01-01 partitions as + // 21991230 on the successor and as the legacy 32-bit wrap's 20631119. of() unions these into one set; here + // the id must appear as a MEMBER of both groups, since a statement scoped to 21991230 and one scoped to + // 20631119 each need this id in their own bound (project_id, id) pairs to actually delete the row wherever + // it physically lives. + var farFuture = UUID.fromString("0699eb8a-59dd-7215-8000-03b8d2a8d5e2"); + + assertThat(WeeklyPartitions.groupByPartition(List.of(farFuture))) + .contains(Map.of( + 21991230L, Set.of(farFuture), + 20631119L, Set.of(farFuture))); + } + + @Test + @DisplayName("groupByPartition mixes a far-future id into an ordinary id's group when they share a partition") + void groupByPartitionSharedPartitionMergesIntoOneGroup() { + // Two DIFFERENT ordinary ids whose legacy 1996 id lands in the same week (19960205) as the far-future id's + // legacy-wrapped value — this is the case that would break a naive "one group per id" implementation: the + // far-future id's SECOND value must merge into an existing group, not create a fresh singleton group that + // happens to collide. + var from1996 = UUID.fromString("00bfd451-fa93-7c10-9923-88a219a974c8"); // -> 19960205L only + var farFuture = UUID.fromString("0699eb8a-59dd-7215-8000-03b8d2a8d5e2"); // -> 21991230L, 20631119L + + var grouped = WeeklyPartitions.groupByPartition(List.of(from1996, farFuture)).orElseThrow(); + + assertThat(grouped.keySet()).containsExactlyInAnyOrder(19960205L, 21991230L, 20631119L); + assertThat(grouped.get(19960205L)).containsExactlyInAnyOrder(from1996); + assertThat(grouped.get(21991230L)).containsExactlyInAnyOrder(farFuture); + assertThat(grouped.get(20631119L)).containsExactlyInAnyOrder(farFuture); + } + + @Test + @DisplayName("groupByPartition collapses duplicate ids in the same partition into one entry via the Set") + void groupByPartitionDuplicatesCollapseWithinAGroup() { + var id = UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"); + + assertThat(WeeklyPartitions.groupByPartition(List.of(id, id))) + .contains(Map.of(20260817L, Set.of(id))); + } + + @Test + @DisplayName("groupByPartition disables pruning for the whole batch on a non-v7 id, same as of()") + void groupByPartitionNonV7DisablesPruning() { + assertThat(WeeklyPartitions.groupByPartition(List.of( + UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"), + UUID.fromString("9f527bac-527a-4f92-8875-0fa8af8e4f22")))) // v4 .isEmpty(); } @Test - @DisplayName("one out-of-range id disables pruning for the whole batch") - void outOfRangeIdDisablesPruningForTheWholeBatch() { - // Same all-or-nothing rule as a non-v7 id, for the same reason: a set derived from the rest of the batch is a - // set this row is not in. - assertThat(WeeklyPartitions.of(List.of( + @DisplayName("groupByPartition disables pruning for the whole batch on an out-of-range id, same as of()") + void groupByPartitionOutOfRangeIdDisablesPruning() { + assertThat(WeeklyPartitions.groupByPartition(List.of( UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"), UUID.fromString("ffffffff-ffff-7abc-8000-000000000001")))) .isEmpty(); } @Test - @DisplayName("the earliest id a UUIDv7 can carry is inside Date32, so there is no floor to guard") - void earliestUuidV7TimestampIsInRange() { - // All 48 timestamp bits clear: id_at 1970-01-01, the earliest any UUIDv7 can encode (the field is unsigned). - // Its Monday, 1969-12-29, is 70 years above Date32's 1900 floor, so a below-1900 id_at is unreachable by - // construction rather than merely untested — which is why `of` guards only the ceiling. It is also the floor of - // the legacy 32-bit DateTime, so the wrap leaves it untouched and it contributes one value. - assertThat(WeeklyPartitions.of(List.of(UUID.fromString("00000000-0000-7abc-8000-000000000001")))) - .contains(Set.of(19691229L)); + @DisplayName("groupByPartition on an empty batch yields no groups") + void groupByPartitionEmptyBatch() { + assertThat(WeeklyPartitions.groupByPartition(List.of())).isEmpty(); + } + + @Test + @DisplayName("groupByPartition on a null batch throws rather than reading as unprunable") + void groupByPartitionNullBatchThrows() { + assertThatThrownBy(() -> WeeklyPartitions.groupByPartition(null)) + .isInstanceOf(NullPointerException.class) + .hasMessageContaining("ids"); + } + + @Test + @DisplayName("groupByPartition's map and its per-partition sets are both immutable") + void groupByPartitionResultIsImmutable() { + // The map AND its values are load-bearing here: a caller that narrowed either would emit a DELETE ... IN + // PARTITION statement missing one of its own bound ids — a correct-looking statement that deletes less than + // it should, silently. + var ordinary = UUID.fromString("01a01a75-76de-785e-ae84-8870ed5e6db3"); + var grouped = WeeklyPartitions.groupByPartition(List.of(ordinary)).orElseThrow(); + + assertThatThrownBy(() -> grouped.remove(20260817L)) + .isInstanceOf(UnsupportedOperationException.class); + assertThatThrownBy(() -> grouped.get(20260817L).remove(ordinary)) + .isInstanceOf(UnsupportedOperationException.class); } }