diff --git a/evidence/bi-p105/materialization-drift.json b/evidence/bi-p105/materialization-drift.json new file mode 100644 index 0000000..669368a --- /dev/null +++ b/evidence/bi-p105/materialization-drift.json @@ -0,0 +1,75 @@ +{ + "clickhouse_version": "26.3.10.60", + "recorded_at": "2026-09-07T01:55:42.821Z", + "source_rows": 2000000, + "after_backfill": { + "verdict": "match", + "buckets_compared": 48, + "revenue_total": { + "source": 801571002, + "aggregate": 801571002, + "delta": 0 + } + }, + "performance": { + "note": "ClickHouse caches marks and pages between repetitions, so these are WARM numbers on one host with nothing else running. They describe this shape of query on this data, not a system property.", + "repetitions": 9, + "base_table_ms": { + "n": 9, + "min": 48.70366200000012, + "p50": 76.67811500000016, + "p95": 115.04416399999991, + "max": 115.04416399999991, + "spread_ratio": 2.36 + }, + "pre_aggregate_ms": { + "n": 9, + "min": 15.281603999999788, + "p50": 22.727642999999944, + "p95": 84.41279499999973, + "max": 84.41279499999973, + "spread_ratio": 5.52 + }, + "speedup_p50": 3.4, + "speedup_range_min_max": [0.6, 7.5] + }, + "gdpr_erasure_drill": { + "mutation": "ALTER TABLE sales DELETE WHERE tenant_id='t_alpha' AND customer_id < 250", + "verdict": "drifted", + "buckets_compared": 48, + "revenue_true_after_erasure": 761580554, + "revenue_pre_aggregate_would_report": 801571002, + "revenue_overstatement": 39990448, + "revenue_overstatement_pct": 5.251, + "worst_bucket": { + "key": "2026-08-28|FI", + "metric": "revenue", + "source_value": 15830779, + "aggregate_value": 16699145, + "absolute_delta": 868366, + "relative_delta": 0.05485301765630106 + }, + "guarded_router_decision": { + "source": "base_table", + "refusal": "state_not_serving", + "detail": "materialization state is \"drifted\"" + }, + "staleness_only_router_would_serve": true, + "staleness_seconds_at_decision": 1, + "note": "The aggregate was NOT stale by any clock: it had just been refreshed, and every freshness check passed. Only the source comparison caught it. This is why staleness alone is not a correctness guarantee." + }, + "after_rebuild": { + "verdict": "match", + "revenue_total": { + "source": 761580554, + "aggregate": 761580554, + "delta": 0 + }, + "routing": { + "source": "materialization", + "materialization_id": "mv_sales_daily", + "target_ref": "opt_p105_mtql9p2f.sales_daily", + "staleness_seconds": 0 + } + } +} diff --git a/src/analytics/materialization/drift-detector.ts b/src/analytics/materialization/drift-detector.ts new file mode 100644 index 0000000..e5f348a --- /dev/null +++ b/src/analytics/materialization/drift-detector.ts @@ -0,0 +1,170 @@ +/** + * src/analytics/materialization/drift-detector.ts + * ================================================ + * BI-P105 — is the pre-aggregate still telling the truth? + * + * The comparison is bucket-by-bucket between the aggregate and a recomputation + * from the source over the same window. Three things make it a real detector + * rather than a ceremony: + * + * 1. A MISSING BUCKET IS DRIFT. The nastiest pre-aggregate failure is not a + * wrong value, it is an absent row: late-arriving data, a materialized view + * that was created after the backfill, a refresh that failed silently. Every + * value it DOES hold is correct, so a value-only comparison passes while the + * total is wrong. Set difference is checked first, before any arithmetic. + * + * 2. ZERO BUCKETS IS `insufficient_data`, NEVER `match`. A comparison over an + * empty window finds no discrepancies — the same false green BI-P025 found + * in a quality check over an empty table. The router treats + * `insufficient_data` as "do not serve", not as "fine". + * + * 3. RELATIVE AND ABSOLUTE TOLERANCE BOTH HAVE TO HOLD. A 0.1% tolerance on a + * revenue bucket of 40M is 40 000 kr of invisible error; an absolute-only + * tolerance flags every rounding difference on a large bucket. Requiring + * both means a discrepancy has to be small in kronor AND small in + * proportion. + */ + +import type { DriftTolerance, DriftVerdict, MaterializationStatus } from './materialization'; + +/** One aggregated row, keyed by its dimension tuple. */ +export interface AggregateBucket { + /** Stable key for the dimension tuple, e.g. "2026-09-01|SE". */ + key: string; + /** metric name -> value. */ + values: Record; +} + +export interface DriftDiscrepancy { + key: string; + metric: string; + source_value: number; + aggregate_value: number; + absolute_delta: number; + /** |delta| / |source|, or Infinity when the source value is 0 and the aggregate is not. */ + relative_delta: number; +} + +export interface DriftReport { + verdict: DriftVerdict; + checked_at: string; + buckets_compared: number; + /** In the source recomputation, absent from the aggregate. Always drift. */ + missing_in_aggregate: string[]; + /** In the aggregate, absent from the source. Always drift (deleted rows). */ + extra_in_aggregate: string[]; + discrepancies: DriftDiscrepancy[]; + worst?: DriftDiscrepancy; + tolerance: DriftTolerance; + /** + * What a caller would have been told if it had trusted the aggregate: the + * total the aggregate reports vs the total the source computes, per metric. + * This is the number that makes a drift report actionable — "0.4% off" means + * nothing until it is "1 284 500 kr reported, 1 271 900 kr true". + */ + totals: Record; +} + +const MAX_REPORTED_DISCREPANCIES = 50; + +function relative(sourceValue: number, delta: number): number { + if (sourceValue === 0) return delta === 0 ? 0 : Number.POSITIVE_INFINITY; + return Math.abs(delta / sourceValue); +} + +/** + * Compare a recomputation from source against the pre-aggregate. + * + * `fromSource` is the truth. Both sides must cover the same window; the caller + * is responsible for that, and `missing_in_aggregate` will catch it loudly if + * they do not. + */ +export function detectDrift(args: { + fromSource: AggregateBucket[]; + fromAggregate: AggregateBucket[]; + tolerance: DriftTolerance; + metrics: string[]; + now?: Date; +}): DriftReport { + const { fromSource, fromAggregate, tolerance, metrics } = args; + const checked_at = (args.now ?? new Date()).toISOString(); + const sourceByKey = new Map(fromSource.map((b) => [b.key, b])); + const aggByKey = new Map(fromAggregate.map((b) => [b.key, b])); + + const missing_in_aggregate = [...sourceByKey.keys()].filter((k) => !aggByKey.has(k)).sort(); + const extra_in_aggregate = [...aggByKey.keys()].filter((k) => !sourceByKey.has(k)).sort(); + + const discrepancies: DriftDiscrepancy[] = []; + let buckets_compared = 0; + const totals: DriftReport['totals'] = {}; + for (const m of metrics) totals[m] = { source: 0, aggregate: 0, delta: 0 }; + + for (const [key, sourceBucket] of sourceByKey) { + for (const m of metrics) totals[m].source += sourceBucket.values[m] ?? 0; + const aggBucket = aggByKey.get(key); + if (!aggBucket) continue; + buckets_compared += 1; + for (const m of metrics) { + const sv = sourceBucket.values[m] ?? 0; + const av = aggBucket.values[m] ?? 0; + const delta = av - sv; + const rel = relative(sv, delta); + if (Math.abs(delta) > tolerance.absolute || rel > tolerance.relative) { + discrepancies.push({ + key, + metric: m, + source_value: sv, + aggregate_value: av, + absolute_delta: delta, + relative_delta: rel, + }); + } + } + } + for (const bucket of aggByKey.values()) { + for (const m of metrics) totals[m].aggregate += bucket.values[m] ?? 0; + } + for (const m of metrics) totals[m].delta = totals[m].aggregate - totals[m].source; + + discrepancies.sort((a, b) => Math.abs(b.absolute_delta) - Math.abs(a.absolute_delta)); + + // A comparison that compared nothing has not proved anything. This is the + // BI-P025 false green, restated for pre-aggregates. + const verdict: DriftVerdict = + buckets_compared === 0 && missing_in_aggregate.length === 0 && extra_in_aggregate.length === 0 + ? 'insufficient_data' + : discrepancies.length > 0 || missing_in_aggregate.length > 0 || extra_in_aggregate.length > 0 + ? 'drifted' + : 'match'; + + const report: DriftReport = { + verdict, + checked_at, + buckets_compared, + missing_in_aggregate, + extra_in_aggregate, + discrepancies: discrepancies.slice(0, MAX_REPORTED_DISCREPANCIES), + tolerance, + totals, + }; + if (discrepancies.length > 0) report.worst = discrepancies[0]; + return report; +} + +/** + * Fold a drift report into the runtime status. A drifted aggregate is taken + * OUT of service here rather than merely annotated: leaving it `serving` with + * a bad verdict relies on every future reader remembering to check the + * verdict, and one that forgets serves the wrong number. + * + * `insufficient_data` also stops service. It means the check could not prove + * anything, and an unproven aggregate is not a current one. + */ +export function applyDriftReport(status: MaterializationStatus, report: DriftReport): MaterializationStatus { + return { + ...status, + state: report.verdict === 'match' ? status.state : 'drifted', + last_drift_check_at: report.checked_at, + last_drift_verdict: report.verdict, + }; +} diff --git a/src/analytics/materialization/materialization.ts b/src/analytics/materialization/materialization.ts new file mode 100644 index 0000000..77b5f86 --- /dev/null +++ b/src/analytics/materialization/materialization.ts @@ -0,0 +1,309 @@ +/** + * src/analytics/materialization/materialization.ts + * ================================================= + * BI-P105 — MATERIALIZATION / PRE-AGGREGATION. + * + * A pre-aggregate is the one optimisation that fails SILENTLY. A slow query is + * a complaint; a pre-aggregate that has drifted from its source answers + * instantly, with confidence, and is wrong — no error, no timeout, nothing in + * a log. So this module is built around one rule: + * + * A MATERIALIZATION SERVES ONLY WHILE IT IS PROVABLY CURRENT AND PROVABLY + * UNDRIFTED. Every unknown routes to the source table. + * + * Six refusal reasons, each of which is a real way the aggregate becomes a + * confident liar: + * + * coverage_miss it does not carry the metric/dimension/grain + * asked for (a filter on a dimension the + * aggregate collapsed cannot be applied after + * the fact, and applying it anyway is how a + * pre-aggregate returns another slice's number) + * watermark_unknown nobody knows how current it is. BI-P026 already + * refuses to render UNKNOWN age as current; the + * same doctrine applies here. + * staleness_exceeded it is older than its own freshness contract + * window_beyond_watermark the question reaches past the newest data the + * aggregate contains — it would silently answer + * with a shorter period + * drift_never_verified it has never been compared to its source + * drift_verification_stale the last comparison is older than the interval + * the spec requires + * drift_detected the last comparison FAILED + * + * `drift_never_verified` is the one people leave out, and it is the one that + * matters most on day one: a brand-new materialization is exactly as unproven + * as a drifted one, and a router that defaults new aggregates to "serving" + * ships the bug on the day the feature launches. + * + * The DoD also asks for measured improvement and documented freshness + * semantics: scripts/bi-p105-materialization-drift.mjs measures both against a + * live ClickHouse and writes evidence/bi-p105/materialization-drift.json, + * including what happens WITHOUT this router (the drifted aggregate served). + */ +import { SiaxError } from '../../platform/errors'; +import type { AnalyticsQueryRequest, TimeGrain } from '../contract/query-request'; + +export const MATERIALIZATION_STRATEGIES = [ + 'clickhouse_materialized_view', + 'clickhouse_projection', + 'dbt_incremental', + 'cube_preaggregation', +] as const; +export type MaterializationStrategy = (typeof MATERIALIZATION_STRATEGIES)[number]; + +export const MATERIALIZATION_STATES = ['building', 'serving', 'drifted', 'disabled'] as const; +export type MaterializationState = (typeof MATERIALIZATION_STATES)[number]; + +/** + * Grains ordered coarsest-last. An aggregate at `day` can answer `day` and + * anything coarser by re-aggregating; it can NEVER answer `hour`, because the + * detail was thrown away when the rows were rolled up. + */ +const GRAIN_ORDER: readonly TimeGrain[] = ['hour', 'day', 'week', 'month', 'quarter', 'year']; + +/** + * `week` does not divide `month` and `month` does not divide `quarter` by row + * count, but both roll up cleanly from `day`. Grains that CANNOT be derived by + * re-aggregating a coarser one are listed here so the router refuses rather + * than quietly returning a mis-bucketed series. + */ +function canRollUp(from: TimeGrain, to: TimeGrain): boolean { + if (from === to) return true; + const fi = GRAIN_ORDER.indexOf(from); + const ti = GRAIN_ORDER.indexOf(to); + if (fi < 0 || ti < 0 || ti < fi) return false; + // Weekly buckets do not compose into calendar months/quarters/years. + if (from === 'week' && to !== 'week') return false; + return true; +} + +export interface DriftTolerance { + /** Absolute delta allowed per metric per bucket. */ + absolute: number; + /** Relative delta allowed (0.001 = 0.1%). Both must hold. */ + relative: number; +} + +/** What a materialization IS. Declarative; the engine adapters read it. */ +export interface MaterializationSpec { + materialization_id: string; + tenant_id: string; + project_id: string; + /** The dataset it pre-aggregates (BI-P003). */ + dataset: string; + strategy: MaterializationStrategy; + /** Physical target (ClickHouse table, dbt model, Cube pre-agg name). */ + target_ref: string; + metrics: string[]; + dimensions: string[]; + time_dimension?: string; + grain?: TimeGrain; + /** The freshness contract. Older than this and it stops serving. */ + max_staleness_seconds: number; + /** How often the aggregate must be re-proved against its source. */ + drift_check_interval_seconds: number; + tolerance: DriftTolerance; +} + +export const DRIFT_VERDICTS = ['match', 'drifted', 'insufficient_data'] as const; +export type DriftVerdict = (typeof DRIFT_VERDICTS)[number]; + +/** Runtime state. Separate from the spec: the spec is config, this is fact. */ +export interface MaterializationStatus { + materialization_id: string; + state: MaterializationState; + /** + * The newest source timestamp the aggregate represents. UNDEFINED means + * UNKNOWN, and unknown is never treated as current. + */ + data_watermark?: string; + last_refresh_at?: string; + last_drift_check_at?: string; + last_drift_verdict?: DriftVerdict; +} + +export const ROUTING_REFUSALS = [ + 'coverage_miss', + 'watermark_unknown', + 'staleness_exceeded', + 'window_beyond_watermark', + 'drift_never_verified', + 'drift_verification_stale', + 'drift_detected', + 'state_not_serving', +] as const; +export type RoutingRefusal = (typeof ROUTING_REFUSALS)[number]; + +export type RoutingDecision = + | { + source: 'materialization'; + materialization_id: string; + target_ref: string; + /** Reported so a caller can show it; never inferred, always measured. */ + staleness_seconds: number; + } + | { source: 'base_table'; refusal: RoutingRefusal; detail: string }; + +function refuse(refusal: RoutingRefusal, detail: string): RoutingDecision { + return { source: 'base_table', refusal, detail }; +} + +/** + * Staleness of an aggregate, in seconds, or `undefined` when it cannot be + * computed. Deliberately NOT defaulted to 0: "we could not tell" and "it is + * current" are different facts, and conflating them is the bug. + */ +export function materializationStaleness(status: MaterializationStatus, now: Date): number | undefined { + if (!status.data_watermark) return undefined; + const t = Date.parse(status.data_watermark); + if (Number.isNaN(t)) return undefined; + return Math.max(0, Math.floor((now.getTime() - t) / 1000)); +} + +/** + * Decide whether a request may be answered from the pre-aggregate. + * + * Fail-closed by construction: the function returns `materialization` only + * after every check has passed, and every unknown is a refusal with a named + * reason. Callers get the reason so it can be logged and counted (BI-P103 + * reads the counts). + */ +export function selectQuerySource(args: { + request: AnalyticsQueryRequest; + spec: MaterializationSpec; + status: MaterializationStatus; + now: Date; +}): RoutingDecision { + const { request, spec, status, now } = args; + + if (request.dataset !== spec.dataset) { + return refuse('coverage_miss', `materialization covers "${spec.dataset}", not "${request.dataset}"`); + } + if (status.state !== 'serving') { + return refuse('state_not_serving', `materialization state is "${status.state}"`); + } + + // ── coverage ──────────────────────────────────────────────────────────── + const metrics = new Set(spec.metrics); + for (const m of request.metrics) { + if (!metrics.has(m)) return refuse('coverage_miss', `metric "${m}" is not in the pre-aggregate`); + } + const dimensions = new Set(spec.dimensions); + for (const d of request.dimensions) { + if (!dimensions.has(d)) return refuse('coverage_miss', `dimension "${d}" is not in the pre-aggregate`); + } + for (const f of request.filters) { + if (!dimensions.has(f.dimension)) { + // The aggregate summed across this dimension. Filtering afterwards would + // return a number computed over rows the filter should have excluded. + return refuse( + 'coverage_miss', + `filter on "${f.dimension}" cannot be applied: the pre-aggregate has already summed across it`, + ); + } + } + const requestedGrain = request.timeRange?.grain; + if (requestedGrain) { + if (!spec.grain) return refuse('coverage_miss', 'the pre-aggregate has no time grain'); + if (!canRollUp(spec.grain, requestedGrain)) { + return refuse( + 'coverage_miss', + `a "${spec.grain}" pre-aggregate cannot answer at "${requestedGrain}" grain`, + ); + } + } + + // ── freshness ─────────────────────────────────────────────────────────── + const staleness = materializationStaleness(status, now); + if (staleness === undefined) { + return refuse( + 'watermark_unknown', + 'the pre-aggregate has no data watermark — unknown age is not current age', + ); + } + if (staleness > spec.max_staleness_seconds) { + return refuse( + 'staleness_exceeded', + `${staleness}s stale, contract allows ${spec.max_staleness_seconds}s`, + ); + } + const to = request.timeRange?.to; + if (to) { + const toMs = Date.parse(to); + const watermarkMs = Date.parse(status.data_watermark!); + if (!Number.isNaN(toMs) && toMs > watermarkMs) { + return refuse( + 'window_beyond_watermark', + `the question runs to ${to} but the pre-aggregate only contains data to ${status.data_watermark}`, + ); + } + } + + // ── drift ─────────────────────────────────────────────────────────────── + if (!status.last_drift_check_at || !status.last_drift_verdict) { + return refuse( + 'drift_never_verified', + 'the pre-aggregate has never been compared to its source; a new aggregate is as unproven as a drifted one', + ); + } + if (status.last_drift_verdict !== 'match') { + return refuse('drift_detected', `last drift check returned "${status.last_drift_verdict}"`); + } + const checkedAgo = Math.floor((now.getTime() - Date.parse(status.last_drift_check_at)) / 1000); + if (Number.isNaN(checkedAgo)) { + return refuse('drift_never_verified', 'last_drift_check_at is not a timestamp'); + } + if (checkedAgo > spec.drift_check_interval_seconds) { + return refuse( + 'drift_verification_stale', + `last verified ${checkedAgo}s ago, spec requires every ${spec.drift_check_interval_seconds}s`, + ); + } + + return { + source: 'materialization', + materialization_id: spec.materialization_id, + target_ref: spec.target_ref, + staleness_seconds: staleness, + }; +} + +/** + * The routing decision as a freshness envelope for the result (BI-P026 shape): + * a consumer must be able to see WHICH source answered and how old it was. + * A result that does not say is a result nobody can audit. + */ +export function describeRouting(decision: RoutingDecision): { + served_from: 'materialization' | 'base_table'; + materialization_id?: string; + staleness_seconds?: number; + refusal?: RoutingRefusal; + detail?: string; +} { + return decision.source === 'materialization' + ? { + served_from: 'materialization', + materialization_id: decision.materialization_id, + staleness_seconds: decision.staleness_seconds, + } + : { served_from: 'base_table', refusal: decision.refusal, detail: decision.detail }; +} + +export function assertSpec(spec: MaterializationSpec): void { + const invalid = (d: string): never => { + throw new SiaxError('INVALID_REQUEST', '0pt.analytics.materialization', d); + }; + if (!spec.materialization_id?.trim()) invalid('materialization_id is required'); + if (!spec.tenant_id?.trim() || !spec.project_id?.trim()) invalid('a materialization is tenant-scoped'); + if (spec.metrics.length === 0) invalid('a materialization with no metrics pre-aggregates nothing'); + if (spec.max_staleness_seconds <= 0) invalid('max_staleness_seconds must be positive'); + if (spec.drift_check_interval_seconds <= 0) { + invalid( + 'drift_check_interval_seconds must be positive — an aggregate that is never re-verified is one ' + + 'source mutation away from serving wrong numbers forever', + ); + } + if (spec.tolerance.absolute < 0 || spec.tolerance.relative < 0) invalid('tolerance must be non-negative'); + if (spec.grain && !GRAIN_ORDER.includes(spec.grain)) invalid('grain is not a time grain'); +} diff --git a/tests/analytics/materialization-drift.test.ts b/tests/analytics/materialization-drift.test.ts new file mode 100644 index 0000000..dadef13 --- /dev/null +++ b/tests/analytics/materialization-drift.test.ts @@ -0,0 +1,327 @@ +/** + * tests/analytics/materialization-drift.test.ts + * ============================================== + * BI-P105 — MATERIALIZATION / PRE-AGGREGATION. + * + * The DoD asks for measured improvement and documented freshness semantics. + * The failure mode it does not name, and the one that actually costs money, is + * a pre-aggregate that returns a confident wrong number. So every test here is + * written as an attack on the router: + * + * - a drifted aggregate must NOT serve, and the counterfactual (a naive + * router that only checks staleness) must serve it — otherwise the guard + * is not known to be doing anything; + * - a MISSING bucket must count as drift even though every value present is + * correct; + * - a comparison over an empty window must be `insufficient_data`, never + * `match`; + * - an aggregate that has never been verified must not serve, because on + * launch day it is exactly as unproven as a drifted one. + */ +import { describe, expect, it } from 'vitest'; +import { + applyDriftReport, + detectDrift, + type AggregateBucket, +} from '@/src/analytics/materialization/drift-detector'; +import { + assertSpec, + describeRouting, + materializationStaleness, + selectQuerySource, + type MaterializationSpec, + type MaterializationStatus, +} from '@/src/analytics/materialization/materialization'; +import type { AnalyticsQueryRequest } from '@/src/analytics/contract/query-request'; + +const NOW = new Date('2026-09-07T12:00:00.000Z'); +const TOLERANCE = { absolute: 0.5, relative: 0.001 }; + +const SPEC: MaterializationSpec = { + materialization_id: 'mv_sales_daily', + tenant_id: 't_alpha', + project_id: 'p_main', + dataset: 'ds_sales', + strategy: 'clickhouse_materialized_view', + target_ref: 'opt_sales_daily', + metrics: ['revenue', 'orders'], + dimensions: ['order_date', 'region'], + time_dimension: 'order_date', + grain: 'day', + max_staleness_seconds: 3600, + drift_check_interval_seconds: 86_400, + tolerance: TOLERANCE, +}; + +/** Fresh, recently verified, matching. The only state that may serve. */ +function healthyStatus(): MaterializationStatus { + return { + materialization_id: SPEC.materialization_id, + state: 'serving', + data_watermark: '2026-09-07T11:45:00.000Z', + last_refresh_at: '2026-09-07T11:46:00.000Z', + last_drift_check_at: '2026-09-07T06:00:00.000Z', + last_drift_verdict: 'match', + }; +} + +const REQUEST: AnalyticsQueryRequest = { + dataset: 'ds_sales', + metrics: ['revenue'], + dimensions: ['order_date'], + filters: [{ dimension: 'region', operator: 'eq', values: ['SE'] }], + timeRange: { from: '2026-09-01T00:00:00.000Z', to: '2026-09-07T00:00:00.000Z', grain: 'day' }, +}; + +/** + * The router someone writes when they have thought about staleness but not + * about correctness. It exists here as the counterfactual: it must SERVE the + * drifted aggregate that `selectQuerySource` refuses. + */ +function naiveRouter(status: MaterializationStatus, now: Date): 'materialization' | 'base_table' { + const staleness = materializationStaleness(status, now) ?? 0; + return staleness <= SPEC.max_staleness_seconds ? 'materialization' : 'base_table'; +} + +function buckets(spec: Record): AggregateBucket[] { + return Object.entries(spec).map(([key, [revenue, orders]]) => ({ key, values: { revenue, orders } })); +} + +const SOURCE = buckets({ + '2026-09-01|SE': [120_000, 410], + '2026-09-02|SE': [131_500, 442], + '2026-09-03|SE': [98_250, 305], + '2026-09-04|SE': [140_000, 480], +}); + +describe('BI-P105 drift detection', () => { + it('an identical aggregate matches', () => { + const report = detectDrift({ + fromSource: SOURCE, + fromAggregate: SOURCE, + tolerance: TOLERANCE, + metrics: ['revenue', 'orders'], + now: NOW, + }); + expect(report.verdict).toBe('match'); + expect(report.buckets_compared).toBe(4); + expect(report.totals.revenue.delta).toBe(0); + }); + + it('a value beyond tolerance is drift, and the report says what a caller would have been told', () => { + const drifted = buckets({ + '2026-09-01|SE': [120_000, 410], + '2026-09-02|SE': [131_500, 442], + '2026-09-03|SE': [98_250, 305], + // A source row was deleted (GDPR erasure) but the aggregate still counts it. + '2026-09-04|SE': [152_400, 495], + }); + const report = detectDrift({ + fromSource: SOURCE, + fromAggregate: drifted, + tolerance: TOLERANCE, + metrics: ['revenue', 'orders'], + now: NOW, + }); + expect(report.verdict).toBe('drifted'); + expect(report.worst?.key).toBe('2026-09-04|SE'); + expect(report.worst?.absolute_delta).toBe(12_400); + expect(report.totals.revenue.aggregate - report.totals.revenue.source).toBe(12_400); + }); + + it('a MISSING bucket is drift even though every value present is correct', () => { + const lateArriving = SOURCE.filter((b) => b.key !== '2026-09-04|SE'); + const report = detectDrift({ + fromSource: SOURCE, + fromAggregate: lateArriving, + tolerance: TOLERANCE, + metrics: ['revenue', 'orders'], + now: NOW, + }); + expect(report.discrepancies).toHaveLength(0); // every compared value agrees + expect(report.verdict).toBe('drifted'); // and it is still wrong + expect(report.missing_in_aggregate).toEqual(['2026-09-04|SE']); + expect(report.totals.revenue.delta).toBe(-140_000); + }); + + it('an EXTRA bucket is drift — the source rows are gone, the aggregate is not', () => { + const stale = [...SOURCE, { key: '2026-08-31|SE', values: { revenue: 90_000, orders: 300 } }]; + const report = detectDrift({ + fromSource: SOURCE, + fromAggregate: stale, + tolerance: TOLERANCE, + metrics: ['revenue', 'orders'], + now: NOW, + }); + expect(report.verdict).toBe('drifted'); + expect(report.extra_in_aggregate).toEqual(['2026-08-31|SE']); + }); + + it('an EMPTY comparison is insufficient_data, not match', () => { + const report = detectDrift({ + fromSource: [], + fromAggregate: [], + tolerance: TOLERANCE, + metrics: ['revenue'], + now: NOW, + }); + expect(report.verdict).toBe('insufficient_data'); + expect(report.buckets_compared).toBe(0); + // And it must not be allowed to serve on the strength of that. + const status = applyDriftReport(healthyStatus(), report); + expect(status.state).toBe('drifted'); + }); + + it('both tolerances must hold — a large bucket cannot hide behind a relative bound', () => { + const source = buckets({ 'b|SE': [40_000_000, 1] }); + const off = buckets({ 'b|SE': [40_020_000, 1] }); // 0.05%: inside relative, far outside absolute + const report = detectDrift({ + fromSource: source, + fromAggregate: off, + tolerance: { absolute: 1000, relative: 0.001 }, + metrics: ['revenue'], + now: NOW, + }); + expect(report.verdict).toBe('drifted'); + expect(report.worst?.relative_delta).toBeCloseTo(0.0005, 6); + }); + + it('a bucket that should be zero and is not is drift, not a division error', () => { + const report = detectDrift({ + fromSource: buckets({ 'b|SE': [0, 0] }), + fromAggregate: buckets({ 'b|SE': [12, 1] }), + tolerance: TOLERANCE, + metrics: ['revenue'], + now: NOW, + }); + expect(report.verdict).toBe('drifted'); + expect(report.worst?.relative_delta).toBe(Number.POSITIVE_INFINITY); + }); +}); + +describe('BI-P105 routing — a pre-aggregate serves only while it is provably current and undrifted', () => { + it('serves when everything is proved', () => { + const decision = selectQuerySource({ request: REQUEST, spec: SPEC, status: healthyStatus(), now: NOW }); + expect(decision.source).toBe('materialization'); + expect(describeRouting(decision)).toMatchObject({ + served_from: 'materialization', + materialization_id: 'mv_sales_daily', + staleness_seconds: 900, + }); + }); + + it('a DRIFTED aggregate is refused — and the naive router serves it', () => { + const drifted = applyDriftReport(healthyStatus(), { + verdict: 'drifted', + checked_at: '2026-09-07T11:00:00.000Z', + buckets_compared: 4, + missing_in_aggregate: [], + extra_in_aggregate: [], + discrepancies: [], + tolerance: TOLERANCE, + totals: {}, + }); + + const guarded = selectQuerySource({ request: REQUEST, spec: SPEC, status: drifted, now: NOW }); + expect(guarded.source).toBe('base_table'); + expect(guarded).toMatchObject({ refusal: 'state_not_serving' }); + + // The counterfactual: the same state, routed by staleness alone. + expect(naiveRouter(drifted, NOW)).toBe('materialization'); + }); + + it('a NEVER-VERIFIED aggregate is refused — and the naive router serves it', () => { + const brandNew: MaterializationStatus = { + materialization_id: SPEC.materialization_id, + state: 'serving', + data_watermark: '2026-09-07T11:59:00.000Z', + last_refresh_at: '2026-09-07T11:59:00.000Z', + }; + const guarded = selectQuerySource({ request: REQUEST, spec: SPEC, status: brandNew, now: NOW }); + expect(guarded).toMatchObject({ source: 'base_table', refusal: 'drift_never_verified' }); + expect(naiveRouter(brandNew, NOW)).toBe('materialization'); + }); + + it('a verification older than the required interval is refused', () => { + const status = { ...healthyStatus(), last_drift_check_at: '2026-09-05T06:00:00.000Z' }; + expect(selectQuerySource({ request: REQUEST, spec: SPEC, status, now: NOW })).toMatchObject({ + refusal: 'drift_verification_stale', + }); + }); + + it('an UNKNOWN watermark is refused rather than treated as current', () => { + const status = { ...healthyStatus(), data_watermark: undefined }; + expect(materializationStaleness(status, NOW)).toBeUndefined(); + expect(selectQuerySource({ request: REQUEST, spec: SPEC, status, now: NOW })).toMatchObject({ + refusal: 'watermark_unknown', + }); + }); + + it('an aggregate older than its freshness contract is refused', () => { + const status = { ...healthyStatus(), data_watermark: '2026-09-07T09:00:00.000Z' }; + expect(selectQuerySource({ request: REQUEST, spec: SPEC, status, now: NOW })).toMatchObject({ + refusal: 'staleness_exceeded', + }); + }); + + it('a question that reaches past the watermark is refused, not silently shortened', () => { + const request = { ...REQUEST, timeRange: { ...REQUEST.timeRange, to: '2026-09-07T12:00:00.000Z' } }; + expect(selectQuerySource({ request, spec: SPEC, status: healthyStatus(), now: NOW })).toMatchObject({ + refusal: 'window_beyond_watermark', + }); + }); + + it('a filter on a dimension the aggregate summed across is refused', () => { + const request = { + ...REQUEST, + filters: [{ dimension: 'sales_channel', operator: 'eq' as const, values: ['web'] }], + }; + const decision = selectQuerySource({ request, spec: SPEC, status: healthyStatus(), now: NOW }); + expect(decision).toMatchObject({ source: 'base_table', refusal: 'coverage_miss' }); + expect((decision as { detail: string }).detail).toMatch(/already summed across it/); + }); + + it('a metric or dimension the aggregate does not carry is refused', () => { + expect( + selectQuerySource({ + request: { ...REQUEST, metrics: ['margin'] }, + spec: SPEC, + status: healthyStatus(), + now: NOW, + }), + ).toMatchObject({ refusal: 'coverage_miss' }); + expect( + selectQuerySource({ + request: { ...REQUEST, dimensions: ['product'] }, + spec: SPEC, + status: healthyStatus(), + now: NOW, + }), + ).toMatchObject({ refusal: 'coverage_miss' }); + }); + + it('a daily aggregate cannot answer an hourly question, but can answer a monthly one', () => { + const hourly = { ...REQUEST, timeRange: { ...REQUEST.timeRange, grain: 'hour' as const } }; + expect( + selectQuerySource({ request: hourly, spec: SPEC, status: healthyStatus(), now: NOW }), + ).toMatchObject({ refusal: 'coverage_miss' }); + + const monthly = { ...REQUEST, timeRange: { ...REQUEST.timeRange, grain: 'month' as const } }; + expect( + selectQuerySource({ request: monthly, spec: SPEC, status: healthyStatus(), now: NOW }).source, + ).toBe('materialization'); + }); + + it('a WEEKLY aggregate cannot answer a monthly question — weeks do not compose into months', () => { + const weekly = { ...SPEC, grain: 'week' as const }; + const monthly = { ...REQUEST, timeRange: { ...REQUEST.timeRange, grain: 'month' as const } }; + expect( + selectQuerySource({ request: monthly, spec: weekly, status: healthyStatus(), now: NOW }), + ).toMatchObject({ refusal: 'coverage_miss' }); + }); + + it('a spec with no drift check interval is refused at declaration time', () => { + expect(() => assertSpec({ ...SPEC, drift_check_interval_seconds: 0 })).toThrow(/never re-verified/); + expect(() => assertSpec(SPEC)).not.toThrow(); + }); +}); diff --git a/tests/integration/materialization-drift-clickhouse.integration.test.ts b/tests/integration/materialization-drift-clickhouse.integration.test.ts new file mode 100644 index 0000000..a8f70b3 --- /dev/null +++ b/tests/integration/materialization-drift-clickhouse.integration.test.ts @@ -0,0 +1,348 @@ +/** + * tests/integration/materialization-drift-clickhouse.integration.test.ts + * ======================================================================= + * BI-P105 against a LIVE ClickHouse. Skips honestly when CLICKHOUSE_URL is + * unset; with it set and the server down, it FAILS rather than skipping. + * + * BI-P105's DoD is "performance improvement measured and freshness semantics + * documented", and both halves are done here against real data rather than + * asserted. What the DoD does not say, and what this file is really about, is + * that a pre-aggregate fails SILENTLY: it answers instantly and wrongly. + * + * The drill drives a real ClickHouse materialized view through the exact + * sequence that produces a confident wrong number in production: + * + * 1. 2M source rows, a SummingMergeTree target and a materialized view. + * 2. Backfill, verify with the SHIPPED detector: verdict `match`. + * 3. Measure both paths repeatedly, and report the DISTRIBUTION — a single + * speed-up figure quoted as a system property is misleading (BI-P104). + * 4. `ALTER TABLE ... DELETE` on the source: a GDPR erasure, the single most + * common real cause of pre-aggregate drift. A ClickHouse materialized + * view is an INSERT trigger; it does not see mutations, so the target + * keeps the deleted revenue forever and nothing errors. + * 5. The detector must catch it, and `selectQuerySource` must refuse to + * serve. The counterfactual — a router that checks only staleness — is + * run on the same state and MUST serve, with the wrong number recorded + * in kronor. + * + * Everything measured lands in evidence/bi-p105/materialization-drift.json. + * + * docker run -d --name ch -p 28123:8123 -e CLICKHOUSE_PASSWORD=test \ + * clickhouse/clickhouse-server:26.3 + * CLICKHOUSE_URL=http://127.0.0.1:28123 CLICKHOUSE_AUTH=default:test \ + * pnpm vitest run tests/integration/materialization-drift-clickhouse.integration.test.ts + * + * It CREATEs and DROPs its own database. Never point it at the b0ts + * ClickHouse or any quarantined market data. + */ +import fs from 'node:fs'; +import path from 'node:path'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { + applyDriftReport, + detectDrift, + type AggregateBucket, + type DriftReport, +} from '@/src/analytics/materialization/drift-detector'; +import { + materializationStaleness, + selectQuerySource, + type MaterializationSpec, + type MaterializationStatus, +} from '@/src/analytics/materialization/materialization'; +import type { AnalyticsQueryRequest } from '@/src/analytics/contract/query-request'; + +const CLICKHOUSE_URL = process.env.CLICKHOUSE_URL; +const CLICKHOUSE_AUTH = process.env.CLICKHOUSE_AUTH; +const run = CLICKHOUSE_URL ? describe : describe.skip; +const DB = `opt_p105_${Date.now().toString(36)}`; +const ROWS = 2_000_000; +const REPETITIONS = 9; + +const headers: Record = CLICKHOUSE_AUTH + ? { Authorization: `Basic ${Buffer.from(CLICKHOUSE_AUTH).toString('base64')}` } + : {}; + +async function ch(query: string): Promise { + const res = await fetch(CLICKHOUSE_URL ?? 'http://skipped.invalid', { + method: 'POST', + headers, + body: query, + }); + const text = await res.text(); + if (!res.ok) throw new Error(`ClickHouse ${res.status}: ${text.slice(0, 400)}`); + return text; +} + +/** Rows as buckets keyed "date|region", which is what the detector compares. */ +async function bucketsFrom(query: string): Promise { + const text = await ch(`${query} FORMAT JSONEachRow`); + return text + .trim() + .split('\n') + .filter(Boolean) + .map((line) => { + const r = JSON.parse(line) as { d: string; region: string; revenue: number; orders: number }; + return { key: `${r.d}|${r.region}`, values: { revenue: Number(r.revenue), orders: Number(r.orders) } }; + }); +} + +/** Median and the spread around it. A single number is not a system property. */ +function distribution(samples: number[]): { + n: number; + min: number; + p50: number; + p95: number; + max: number; + spread_ratio: number; +} { + const s = [...samples].sort((a, b) => a - b); + const at = (q: number) => s[Math.min(s.length - 1, Math.floor(q * s.length))]; + return { + n: s.length, + min: s[0], + p50: at(0.5), + p95: at(0.95), + max: s[s.length - 1], + spread_ratio: Number((s[s.length - 1] / Math.max(s[0], 1e-9)).toFixed(2)), + }; +} + +async function timeQuery(query: string): Promise { + const t0 = performance.now(); + await ch(`${query} FORMAT JSONEachRow`); + return performance.now() - t0; +} + +const SOURCE_AGG = ` + SELECT toString(order_date) AS d, region, sum(amount) AS revenue, count() AS orders + FROM ${DB}.sales WHERE tenant_id = 't_alpha' GROUP BY d, region ORDER BY d, region`; +const TARGET_AGG = ` + SELECT toString(order_date) AS d, region, sum(revenue) AS revenue, sum(orders) AS orders + FROM ${DB}.sales_daily WHERE tenant_id = 't_alpha' GROUP BY d, region ORDER BY d, region`; + +const SPEC: MaterializationSpec = { + materialization_id: 'mv_sales_daily', + tenant_id: 't_alpha', + project_id: 'p_main', + dataset: 'ds_sales', + strategy: 'clickhouse_materialized_view', + target_ref: `${DB}.sales_daily`, + metrics: ['revenue', 'orders'], + dimensions: ['order_date', 'region'], + time_dimension: 'order_date', + grain: 'day', + max_staleness_seconds: 3600, + drift_check_interval_seconds: 86_400, + tolerance: { absolute: 0.5, relative: 0.0001 }, +}; + +const REQUEST: AnalyticsQueryRequest = { + dataset: 'ds_sales', + metrics: ['revenue'], + dimensions: ['order_date'], + filters: [], + timeRange: { grain: 'day' }, +}; + +const evidence: Record = {}; +let status: MaterializationStatus; + +run('BI-P105 materialization against a LIVE ClickHouse', () => { + beforeAll(async () => { + evidence.clickhouse_version = (await ch('SELECT version()')).trim(); + evidence.recorded_at = new Date().toISOString(); + evidence.source_rows = ROWS; + + await ch(`CREATE DATABASE IF NOT EXISTS ${DB}`); + await ch(` + CREATE TABLE ${DB}.sales ( + tenant_id LowCardinality(String), + order_date Date, + region LowCardinality(String), + customer_id UInt32, + amount Decimal(18, 2) + ) ENGINE = MergeTree ORDER BY (tenant_id, order_date, region)`); + await ch(` + INSERT INTO ${DB}.sales + SELECT + if(number % 5 = 0, 't_beta', 't_alpha') AS tenant_id, + toDate('2026-08-01') + toIntervalDay(number % 30) AS order_date, + ['SE','NO','DK','FI'][(number % 4) + 1] AS region, + toUInt32(number % 5000) AS customer_id, + toDecimal64((number % 977) + 13, 2) AS amount + FROM numbers(${ROWS})`); + + await ch(` + CREATE TABLE ${DB}.sales_daily ( + tenant_id LowCardinality(String), + order_date Date, + region LowCardinality(String), + revenue Decimal(38, 2), + orders UInt64 + ) ENGINE = SummingMergeTree ORDER BY (tenant_id, order_date, region)`); + await ch(` + CREATE MATERIALIZED VIEW ${DB}.sales_daily_mv TO ${DB}.sales_daily AS + SELECT tenant_id, order_date, region, sum(amount) AS revenue, count() AS orders + FROM ${DB}.sales GROUP BY tenant_id, order_date, region`); + // The MV is an INSERT trigger: it saw nothing that was already there, so + // the backfill is explicit. (This is drift cause #1, and it is why the + // first check below has to be able to fail.) + await ch(` + INSERT INTO ${DB}.sales_daily + SELECT tenant_id, order_date, region, sum(amount) AS revenue, count() AS orders + FROM ${DB}.sales GROUP BY tenant_id, order_date, region`); + await ch(`OPTIMIZE TABLE ${DB}.sales_daily FINAL`); + + status = { + materialization_id: SPEC.materialization_id, + state: 'serving', + data_watermark: new Date().toISOString(), + last_refresh_at: new Date().toISOString(), + }; + }, 300_000); + + afterAll(async () => { + if (!CLICKHOUSE_URL) return; + try { + await ch(`DROP DATABASE IF EXISTS ${DB}`); + } finally { + const dir = path.join(process.cwd(), 'evidence/bi-p105'); + fs.mkdirSync(dir, { recursive: true }); + fs.writeFileSync( + path.join(dir, 'materialization-drift.json'), + `${JSON.stringify(evidence, null, 2)}\n`, + ); + } + }, 60_000); + + it('the freshly backfilled aggregate MATCHES its source', async () => { + const report = detectDrift({ + fromSource: await bucketsFrom(SOURCE_AGG), + fromAggregate: await bucketsFrom(TARGET_AGG), + tolerance: SPEC.tolerance, + metrics: ['revenue', 'orders'], + }); + expect(report.buckets_compared).toBeGreaterThan(0); // an empty compare proves nothing + expect(report.verdict).toBe('match'); + + status = applyDriftReport(status, report); + evidence.after_backfill = { + verdict: report.verdict, + buckets_compared: report.buckets_compared, + revenue_total: report.totals.revenue, + }; + }, 120_000); + + it('measures the improvement as a DISTRIBUTION, not a single number', async () => { + const base: number[] = []; + const agg: number[] = []; + for (let i = 0; i < REPETITIONS; i += 1) { + // Interleaved so a machine that gets busy halfway through penalises both. + base.push(await timeQuery(SOURCE_AGG)); + agg.push(await timeQuery(TARGET_AGG)); + } + const baseD = distribution(base); + const aggD = distribution(agg); + evidence.performance = { + note: + 'ClickHouse caches marks and pages between repetitions, so these are WARM numbers on one host ' + + 'with nothing else running. They describe this shape of query on this data, not a system property.', + repetitions: REPETITIONS, + base_table_ms: baseD, + pre_aggregate_ms: aggD, + speedup_p50: Number((baseD.p50 / Math.max(aggD.p50, 1e-9)).toFixed(1)), + speedup_range_min_max: [ + Number((baseD.min / Math.max(aggD.max, 1e-9)).toFixed(1)), + Number((baseD.max / Math.max(aggD.min, 1e-9)).toFixed(1)), + ], + }; + // The claim under test is only that the pre-aggregate is not SLOWER; the + // magnitude is reported, never asserted as a threshold. + expect(aggD.p50).toBeLessThanOrEqual(baseD.p50); + }, 300_000); + + it('a GDPR erasure on the source DRIFTS the aggregate, and the drift is caught', async () => { + const before = await bucketsFrom(TARGET_AGG); + + // The mutation a right-to-erasure request produces. A materialized view is + // an insert trigger — it never sees this. + await ch( + `ALTER TABLE ${DB}.sales DELETE WHERE tenant_id = 't_alpha' AND customer_id < 250 SETTINGS mutations_sync = 2`, + ); + + const fromSource = await bucketsFrom(SOURCE_AGG); + const fromAggregate = await bucketsFrom(TARGET_AGG); + expect(fromAggregate).toEqual(before); // the aggregate genuinely did not move + + const report: DriftReport = detectDrift({ + fromSource, + fromAggregate, + tolerance: SPEC.tolerance, + metrics: ['revenue', 'orders'], + }); + expect(report.verdict).toBe('drifted'); + expect(report.worst).toBeDefined(); + + const drifted = applyDriftReport(status, report); + const guarded = selectQuerySource({ request: REQUEST, spec: SPEC, status: drifted, now: new Date() }); + expect(guarded.source).toBe('base_table'); + + // COUNTERFACTUAL: a router that checks only staleness serves it, and this + // is the number a dashboard would have shown. + const stalenessOnlyServes = + (materializationStaleness(drifted, new Date()) ?? 0) <= SPEC.max_staleness_seconds; + expect(stalenessOnlyServes).toBe(true); + + evidence.gdpr_erasure_drill = { + mutation: "ALTER TABLE sales DELETE WHERE tenant_id='t_alpha' AND customer_id < 250", + verdict: report.verdict, + buckets_compared: report.buckets_compared, + revenue_true_after_erasure: report.totals.revenue.source, + revenue_pre_aggregate_would_report: report.totals.revenue.aggregate, + revenue_overstatement: report.totals.revenue.delta, + revenue_overstatement_pct: Number( + ((report.totals.revenue.delta / report.totals.revenue.source) * 100).toFixed(3), + ), + worst_bucket: report.worst, + guarded_router_decision: guarded, + staleness_only_router_would_serve: stalenessOnlyServes, + staleness_seconds_at_decision: materializationStaleness(drifted, new Date()), + note: + 'The aggregate was NOT stale by any clock: it had just been refreshed, and every freshness check ' + + 'passed. Only the source comparison caught it. This is why staleness alone is not a correctness ' + + 'guarantee.', + }; + status = drifted; + }, 300_000); + + it('and a refresh that re-derives the aggregate clears the drift', async () => { + await ch(`TRUNCATE TABLE ${DB}.sales_daily`); + await ch(` + INSERT INTO ${DB}.sales_daily + SELECT tenant_id, order_date, region, sum(amount) AS revenue, count() AS orders + FROM ${DB}.sales GROUP BY tenant_id, order_date, region`); + await ch(`OPTIMIZE TABLE ${DB}.sales_daily FINAL`); + + const report = detectDrift({ + fromSource: await bucketsFrom(SOURCE_AGG), + fromAggregate: await bucketsFrom(TARGET_AGG), + tolerance: SPEC.tolerance, + metrics: ['revenue', 'orders'], + }); + expect(report.verdict).toBe('match'); + + const healed: MaterializationStatus = applyDriftReport( + { ...status, state: 'serving', data_watermark: new Date().toISOString() }, + report, + ); + const decision = selectQuerySource({ request: REQUEST, spec: SPEC, status: healed, now: new Date() }); + expect(decision.source).toBe('materialization'); + + evidence.after_rebuild = { + verdict: report.verdict, + revenue_total: report.totals.revenue, + routing: decision, + }; + }, 300_000); +});