0pt/tests/integration/materialization-drift-clickhouse.integration.test.ts

/**
 * 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<string, string> = CLICKHOUSE_AUTH
  ? { Authorization: `Basic ${Buffer.from(CLICKHOUSE_AUTH).toString('base64')}` }
  : {};

async function ch(query: string): Promise<string> {
  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<AggregateBucket[]> {
  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<number> {
  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<string, unknown> = {};
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);
});