diff --git a/src/Usage/Adapter.php b/src/Usage/Adapter.php index b3473e1..9fe6902 100644 --- a/src/Usage/Adapter.php +++ b/src/Usage/Adapter.php @@ -218,4 +218,19 @@ abstract public function sumDaily(string $tenant, array $queries = [], string $a * @return array Metric name => sum value */ abstract public function sumDailyBatch(string $tenant, array $metrics, array $queries = []): array; + + /** + * Backfill the pre-aggregated daily events table from the raw events + * table for the half-open, UTC-midnight-aligned window [$from, $to). + * + * Only meaningful on adapters that maintain a daily rollup; the base + * refuses so a mis-wired caller fails loudly instead of silently + * backfilling nothing. + * + * @throws \Exception + */ + public function backfillDaily(string $from, string $to, bool $force = false): void + { + throw new \Exception(static::class . ' does not maintain a daily rollup to backfill.'); + } } diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index 4500ac9..e6f0b08 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -1152,6 +1152,23 @@ private function decodeTotal(string $result): int * * @var array}> */ + /** + * Latest-value slate: keyed (tenant, metric, dims) with NO time key, one + * argMax state per series. Serves unwindowed grouped latest-value reads + * (the billing prefetch shape) at one row per group — production + * measurement without it: ~300k rows read per grouped gauge read, and + * force_optimize_projection refuses the time-keyed slate for these + * queries (PROJECTION_NOT_USED). Existing installs must + * ALTER TABLE ... MATERIALIZE PROJECTION each p_latest_* once (setup() + * only attaches them for parts written afterwards). + */ + private const GAUGE_LATEST_PROJECTIONS = [ + ['name' => 'p_latest_by_service', 'dims' => ['service']], + ['name' => 'p_latest_by_resourceType', 'dims' => ['resourceType']], + ['name' => 'p_latest_by_resourceId', 'dims' => ['resourceId']], + ['name' => 'p_latest_by_resourceType_resourceId', 'dims' => ['resourceType', 'resourceId']], + ]; + private const GAUGE_PROJECTIONS = [ ['name' => 'p_by_service', 'dims' => ['service']], ['name' => 'p_by_resourceType', 'dims' => ['resourceType']], @@ -1224,6 +1241,15 @@ public function setup(): void 'argMax(value, time) AS value' ); } + foreach (self::GAUGE_LATEST_PROJECTIONS as $projection) { + $this->addProjection( + $this->getGaugesTableName(), + $projection['name'], + $projection['dims'], + 'argMax(value, time) AS value', + includeTimeKey: false, + ); + } } /** @@ -1386,16 +1412,27 @@ private function ensureDimColumns(string $tableName, array $columns, string $typ * * @param array $dims */ - private function addProjection(string $baseTable, string $name, array $dims, string $aggregateExpr): void + private function addProjection(string $baseTable, string $name, array $dims, string $aggregateExpr, bool $includeTimeKey = true): void { $escapedTable = $this->escapeIdentifier($this->database) . '.' . $this->escapeIdentifier($baseTable); - $selectParts = ['metric', 'time']; - $groupParts = ['metric', 'time']; + // Two key shapes, one per read family. With `time` in the keys the + // projection has one row per sample — narrow-column but 1:1 with the + // base table — which is what windowed grouped reads need (the time + // predicate must be expressible on the projection). WITHOUT `time` + // the projection holds one aggregate state per series, so an + // unwindowed latest-value read (the billing prefetch shape) collapses + // to one row per group instead of re-aggregating every sample. + $selectParts = $includeTimeKey ? ['metric', 'time'] : []; + $groupParts = $includeTimeKey ? ['metric', 'time'] : []; if ($this->sharedTables) { $selectParts[] = 'tenant'; $groupParts[] = 'tenant'; } + if (!$includeTimeKey) { + $selectParts[] = 'metric'; + $groupParts[] = 'metric'; + } foreach ($dims as $dim) { $selectParts[] = $this->escapeIdentifier($dim); $groupParts[] = $this->escapeIdentifier($dim); @@ -1643,6 +1680,99 @@ private function createDailyMaterializedView(): void $this->query($this->qualifyDdl($statement->query, $dailyMvName, $dailyTableName)); } + /** + * Backfill the daily rollup from the raw events table for [$from, $to). + * + * The daily materialized view only aggregates rows inserted after it was + * created; windows before that are absent from the rollup, so every read + * routed to it under-counts. This runs the exact aggregation the view + * body runs, bounded to the window, making backfilled days + * indistinguishable from view-produced ones. + * + * Both bounds must be UTC midnights: SummingMergeTree folds duplicate + * keys by adding them, so a partial day written here would double-count + * against rows the view produced for that day. For the same reason the + * window must not overlap rows already in the rollup — the call refuses + * when it does unless $force is passed, after the caller has cleared the + * range itself. + * + * Single-operator migration primitive, not a concurrent API: the overlap + * check and the INSERT are separate statements with no lock, so two + * concurrent calls for the same empty window can both pass the check and + * double the rollup. Run one backfill at a time, and verify with a + * raw-vs-rollup parity read before routing traffic at the window. + * + * @throws Exception + */ + public function backfillDaily(string $from, string $to, bool $force = false): void + { + try { + // The constructor's timezone argument is ignored when the string + // carries its own offset, and the bindings are formatted without + // one — so normalize to UTC first. A bound like + // '2026-01-01 00:00:00+05:00' is midnight only in its own zone; + // after normalization it is 19:00 UTC and correctly refused. + $fromDt = (new DateTime($from, new DateTimeZone('UTC')))->setTimezone(new DateTimeZone('UTC')); + $toDt = (new DateTime($to, new DateTimeZone('UTC')))->setTimezone(new DateTimeZone('UTC')); + } catch (Throwable $e) { + throw new Exception("backfillDaily() bounds must be valid datetimes: {$e->getMessage()}", 0, $e); + } + + if (!$this->isDayAligned($fromDt) || !$this->isDayAligned($toDt)) { + throw new Exception('backfillDaily() bounds must be UTC midnights: a partial day double-counts against rows the daily view already produced for that day.'); + } + if ($fromDt >= $toDt) { + throw new Exception('backfillDaily() needs an ascending half-open window.'); + } + + $this->setOperationContext('backfillDaily()'); + + $eventsTable = $this->buildTableReference($this->getEventsTableName()); + $dailyTable = $this->buildTableReference($this->getEventsDailyTableName()); + + $bindings = [ + 'bf_from' => $fromDt->format('Y-m-d H:i:s.v'), + 'bf_to' => $toDt->format('Y-m-d H:i:s.v'), + ]; + $window = 'time >= {bf_from:DateTime64(3)} AND time < {bf_to:DateTime64(3)}'; + + if (!$force) { + $existing = $this->decodeTotal($this->query( + "SELECT count() AS total FROM {$dailyTable} WHERE {$window} FORMAT JSON", + $bindings, + )); + if ($existing > 0) { + throw new Exception("backfillDaily() window already holds {$existing} rollup rows and inserting again would double-count. Clear the range first, or pass force after doing so."); + } + } + + // Mirrors createDailyMaterializedView()'s body exactly, plus the bound. + $dimensions = 'resourceType, resourceId, resourceInternalId, teamId, teamInternalId'; + + if ($this->sharedTables) { + $columns = "metric, value, time, tenant, {$dimensions}"; + $innerSelect = "metric, tenant, {$dimensions}, sum(value) as value, toStartOfDay(time, 'UTC') as d"; + $innerGroupBy = "metric, tenant, {$dimensions}, d"; + $outerSelect = "metric, value, d as time, tenant, {$dimensions}"; + } else { + $columns = "metric, value, time, {$dimensions}"; + $innerSelect = "metric, {$dimensions}, sum(value) as value, toStartOfDay(time, 'UTC') as d"; + $innerGroupBy = "metric, {$dimensions}, d"; + $outerSelect = "metric, value, d as time, {$dimensions}"; + } + + $sql = "INSERT INTO {$dailyTable} ({$columns})" + . " SELECT {$outerSelect}" + . " FROM (" + . " SELECT {$innerSelect}" + . " FROM {$eventsTable}" + . " WHERE {$window}" + . " GROUP BY {$innerGroupBy}" + . " )"; + + $this->query($sql, $bindings); + } + /** * Validate that an attribute name exists in the schema for a given type. * @@ -3047,6 +3177,11 @@ private function routedSum(string $tenant, array $queries, string $operation): i $this->maybeDualRead($tenant, $queries, $route, $plan, $total); return $total; } + if ($route === 'split') { + $total = $this->sumSplitDailyAndRaw($tenant, $queries, $plan); + $this->maybeDualRead($tenant, $queries, $route, $plan, $total); + return $total; + } return $this->sumFromTable($tenant, $queries, 'value', Usage::TYPE_EVENT); } @@ -3055,7 +3190,7 @@ private function routedSum(string $tenant, array $queries, string $operation): i * Snapshot of the parsed query shape relevant for routing. * * @param array $queries - * @return array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns: array, hasCursor: bool} + * @return array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns: array, hasCursor: bool, hasIrregularTimeFilter: bool, endInclusive: bool} */ private function extractRoutingPlan(array $queries): array { @@ -3067,6 +3202,8 @@ private function extractRoutingPlan(array $queries): array $interval = null; $orderColumns = []; $hasCursor = false; + $hasIrregularTimeFilter = false; + $endInclusive = false; foreach ($queries as $query) { $method = $query->getMethod(); @@ -3128,10 +3265,27 @@ private function extractRoutingPlan(array $queries): array if ($method === Method::GreaterThanEqual || $method === Method::GreaterThan) { $start = $this->tightenLowerBound($start, $this->stringifyTime($values[0] ?? null)); } elseif ($method === Method::LessThanEqual || $method === Method::LessThan) { - $end = $this->tightenUpperBound($end, $this->stringifyTime($values[0] ?? null)); + $candidate = $this->stringifyTime($values[0] ?? null); + $tightened = $this->tightenUpperBound($end, $candidate); + if ($tightened !== $end) { + $endInclusive = $method === Method::LessThanEqual; + } + $end = $tightened; } elseif ($method === Method::Between) { $start = $this->tightenLowerBound($start, $this->stringifyTime($values[0] ?? null)); - $end = $this->tightenUpperBound($end, $this->stringifyTime($values[1] ?? null)); + $candidate = $this->stringifyTime($values[1] ?? null); + $tightened = $this->tightenUpperBound($end, $candidate); + if ($tightened !== $end) { + // Between's upper bound is inclusive. + $endInclusive = true; + } + $end = $tightened; + } else { + // A time filter that is not a window bound (notBetween, + // equal, ...) carves shapes a day-granularity rollup + // cannot honor: a mid-day hole is invisible at day rows, + // and the split route's interior would drop it entirely. + $hasIrregularTimeFilter = true; } } } @@ -3145,6 +3299,8 @@ private function extractRoutingPlan(array $queries): array 'interval' => $interval, 'orderColumns' => $orderColumns, 'hasCursor' => $hasCursor, + 'hasIrregularTimeFilter' => $hasIrregularTimeFilter, + 'endInclusive' => $endInclusive, ]; } @@ -3161,7 +3317,7 @@ private function extractRoutingPlan(array $queries): array * table and the ClickHouse optimizer transparently picks the * matching projection. * - * @param array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns?: array, hasCursor?: bool} $plan + * @param array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns?: array, hasCursor?: bool, hasIrregularTimeFilter?: bool, endInclusive?: bool} $plan */ private function selectAggregateSource(array $plan): string { @@ -3177,6 +3333,10 @@ private function selectAggregateSource(array $plan): string return 'raw'; } + if (!empty($plan['hasIrregularTimeFilter'])) { + return 'raw'; + } + if (in_array('id', $plan['filterColumns'], true) || in_array('value', $plan['filterColumns'], true)) { return 'raw'; } @@ -3204,20 +3364,64 @@ private function selectAggregateSource(array $plan): string } if (!$this->isDayAligned($startDt)) { - return 'raw'; + return $this->splitInterior($startDt, $endDt, $boundaryDt) !== null ? 'split' : 'raw'; } if ($endDt >= $boundaryDt) { return 'hybrid'; } - if (!$this->isDayAligned($endDt)) { - return 'raw'; + if (!$this->isDayAligned($endDt) || !empty($plan['endInclusive'])) { + // A day-aligned but INCLUSIVE upper bound covers the midnight + // instant itself, and a daily row cannot represent it: the bucket + // for that day holds the whole day, so 'daily' would either drop + // the instant (translating <= to <) or over-count the day. Split + // reads the interior from the rollup and that instant from raw. + return $this->splitInterior($startDt, $endDt, $boundaryDt) !== null ? 'split' : 'raw'; } return 'daily'; } + /** + * Interior [from, to) of a window: the whole UTC days the daily rollup can + * answer. From = the start ceiled to the next midnight (null start = the + * rollup's full history), to = the end floored to midnight and capped at + * today (the running day always reads raw). Null when no whole day fits — + * a sub-day window has no rollup-answerable interior. + * + * This is what makes routing reach the windows production actually uses: + * billing cycles are anchored at the moment a team upgraded (0 of 451k + * production teams have midnight-aligned invoice dates), so day-aligned + * routes alone never fire for billing. + * + * @return array{0: ?string, 1: string}|null [interiorFrom|null, interiorTo) + */ + private function splitInterior(?DateTime $startDt, DateTime $endDt, DateTime $boundaryDt): ?array + { + $from = null; + if ($startDt !== null) { + $from = (clone $startDt); + if ($from->format('H:i:s.u') !== '00:00:00.000000') { + $from->setTime(0, 0, 0, 0)->modify('+1 day'); + } + } + + $to = (clone $endDt)->setTime(0, 0, 0, 0); + if ($to > $boundaryDt) { + $to = clone $boundaryDt; + } + + if ($from !== null && $from >= $to) { + return null; + } + + return [ + $from?->format('Y-m-d H:i:s.v'), + $to->format('Y-m-d H:i:s.v'), + ]; + } + /** * Returns true when the timestamp falls exactly on a UTC midnight. */ @@ -3528,7 +3732,7 @@ private function buildDailyTimeQueries(array $timeQueries): array } /** - * @param array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns?: array, hasCursor?: bool} $plan + * @param array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns?: array, hasCursor?: bool, hasIrregularTimeFilter?: bool, endInclusive?: bool} $plan */ private function recordRoute(string $operation, array $plan, string $route): void { @@ -3567,7 +3771,7 @@ private function appendRouteLogEntry(array $entry): void * * @param array $queries * @param string $route - * @param array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns?: array, hasCursor?: bool} $plan + * @param array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns?: array, hasCursor?: bool, hasIrregularTimeFilter?: bool, endInclusive?: bool} $plan */ private function maybeDualRead(string $tenant, array $queries, string $route, array $plan, int $rolledTotal): void { @@ -3664,6 +3868,147 @@ private function sumHybridDailyAndRaw(string $tenant, array $queries, array $pla return $this->decodeTotal($result); } + /** + * Split flat-sum: the interior whole days from the daily rollup, the + * sub-day head and tail edges from the raw events table, folded by an + * outer SUM over UNION ALL. This is the route production windows + * actually take — billing cycles are anchored at upgrade time, never + * midnight, so the day-aligned 'daily'/'hybrid' routes cannot fire for + * them. Measured on the largest production tenant (30-day window): + * 282M rows raw vs 6.7M rows split, identical totals. + * + * @param array $queries + * @param array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns?: array, hasCursor?: bool, hasIrregularTimeFilter?: bool, endInclusive?: bool} $plan + */ + private function sumSplitDailyAndRaw(string $tenant, array $queries, array $plan): int + { + [$branches, $bindings] = $this->splitBranchStatements($tenant, $queries, $plan, null); + + $sql = " + SELECT sum(total) AS total FROM ( + " . implode("\n UNION ALL\n ", $branches) . " + ) + FORMAT JSON + "; + + return $this->decodeTotal($this->query($sql, $bindings)); + } + + /** + * Split batched event totals: the batch shape of sumSplitDailyAndRaw(), + * grouped by metric on every branch and re-folded per metric outside. + * + * @param array $metrics + * @param array $queries + * @param array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns?: array, hasCursor?: bool, hasIrregularTimeFilter?: bool, endInclusive?: bool} $plan + * @return array + */ + private function totalBatchSplit(string $tenant, array $metrics, array $queries, array $plan): array + { + [$branches, $bindings] = $this->splitBranchStatements($tenant, $queries, $plan, $metrics); + + $sql = " + SELECT `metric`, sum(`total`) AS `agg_val` FROM ( + " . implode("\n UNION ALL\n ", $branches) . " + ) + GROUP BY `metric` + FORMAT JSON + "; + + $totalsByMetric = []; + foreach ($this->decodeRows($this->query($sql, $bindings)) as $row) { + $totalsByMetric[self::toStr($row['metric'] ?? null)] = self::toInt($row['agg_val'] ?? null); + } + + return $totalsByMetric; + } + + /** + * Compile the split branches: the daily rollup over the interior whole + * days, and the raw events table over the sub-day edges. The edges keep + * the caller's original time filters (preserving their inclusivity) with + * the interior boundary ANDed on, so head ∪ interior ∪ tail partitions + * the window exactly. Each branch's bindings are prefixed so the merged + * statement has no placeholder collisions. A null $metrics compiles the + * flat-sum shape; a list compiles the grouped-by-metric batch shape. + * + * @param array $queries + * @param array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns?: array, hasCursor?: bool, hasIrregularTimeFilter?: bool, endInclusive?: bool} $plan + * @param array|null $metrics + * @return array{0: array, 1: array} + * @throws Exception + */ + private function splitBranchStatements(string $tenant, array $queries, array $plan, ?array $metrics): array + { + $boundaryDt = new DateTime('today', new DateTimeZone('UTC')); + $endDt = new DateTime((string) $plan['end'], new DateTimeZone('UTC')); + $startDt = $plan['start'] !== null ? new DateTime($plan['start'], new DateTimeZone('UTC')) : null; + + $interior = $this->splitInterior($startDt, $endDt, $boundaryDt); + if ($interior === null) { + // Unreachable: selectAggregateSource() only returns 'split' when + // an interior exists. Fail loudly rather than mis-aggregate. + throw new Exception('Split route selected for a window with no whole interior day.'); + } + [$interiorFrom, $interiorTo] = $interior; + + $split = $this->splitTimeQueries($queries); + + $branchQueryLists = []; + + $dailyQueries = $split['nonTime']; + if ($interiorFrom !== null) { + $dailyQueries[] = Query::greaterThanEqual('time', $interiorFrom); + } + $dailyQueries[] = Query::lessThan('time', $interiorTo); + $branchQueryLists[] = [true, $dailyQueries]; + + if ($interiorFrom !== null && $startDt !== null && $startDt->format('Y-m-d H:i:s.v') !== $interiorFrom) { + $branchQueryLists[] = [false, array_merge($queries, [Query::lessThan('time', $interiorFrom)])]; + } + + // The tail carries the caller's own upper bound, so an inclusive + // bound resolves to exactly the boundary instant ([midnight, midnight]) + // — the rows the interior's half-open upper edge excludes. + if ($endDt->format('Y-m-d H:i:s.v') !== $interiorTo || !empty($plan['endInclusive'])) { + $branchQueryLists[] = [false, array_merge($queries, [Query::greaterThanEqual('time', $interiorTo)])]; + } + + $dailyTableName = $this->getEventsDailyTableName(); + $eventsTableName = $this->getEventsTableName(); + + $branches = []; + $bindings = []; + foreach ($branchQueryLists as $index => [$isDaily, $branchQueries]) { + $tableName = $isDaily ? $dailyTableName : $eventsTableName; + $parsed = $this->parseQueries($tenant, $branchQueries, Usage::TYPE_EVENT); + + $builder = $this->newBuilder(Usage::TYPE_EVENT)->from($tableName); + if ($metrics === null) { + $builder->sum('value', 'total'); + } else { + $builder + ->select(['metric']) + ->selectRaw('SUM(`value`) AS `total`') + ->filter([Query::equal('metric', $metrics)]) + ->groupByRaw('`metric`'); + } + $this->applyFilters($builder, $tenant, $parsed); + + $statement = $builder->build(); + [$branchSql, $branchBindings] = $this->prefixNamedBindings( + $this->qualifyDdl($statement->query, $tableName), + $statement->namedBindings ?? [], + 'b' . $index . '_', + ); + + $branches[] = $branchSql; + $bindings = array_merge($bindings, $branchBindings); + } + + return [$branches, $bindings]; + } + /** * Sum metric values from a specific table. * @@ -4244,37 +4589,18 @@ public function getTotalBatch(string $tenant, array $metrics, array $queries = [ } foreach ($typesToQuery as $queryType) { - $tableName = $this->getTableForType($queryType); - - $parsed = $this->parseQueries($tenant, $queries, $queryType); - - $valueExpr = $queryType === Usage::TYPE_EVENT - ? 'SUM(`value`) AS `agg_val`' - : 'argMax(`value`, `time`) AS `agg_val`'; - - $builder = $this->newBuilder($queryType) - ->from($tableName) - ->select(['metric']) - ->selectRaw($valueExpr) - ->filter([Query::equal('metric', $metrics)]) - ->groupByRaw('`metric`'); - - $this->applyFilters($builder, $tenant, $parsed); - - $statement = $builder->build(); - $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - - $result = $this->query($sql, $statement->namedBindings ?? []); - $rows = $this->decodeRows($result); - - foreach ($rows as $row) { - $metricName = self::toStr($row['metric'] ?? null); - + // Event totals route like sum()/getTotal(): closed-day windows + // read the daily rollup, open ones read rollup + today's raw tail. + // Gauges have no rollup and always read their own table. + $typeTotals = $queryType === Usage::TYPE_EVENT + ? $this->routedTotalBatch($tenant, $metrics, $queries) + : $this->totalBatchFromTable($tenant, $metrics, $queries, $queryType); + + foreach ($typeTotals as $metricName => $rowValue) { if (!isset($totals[$metricName])) { continue; } - $rowValue = self::toInt($row['agg_val'] ?? null); if ($rowValue === 0) { continue; } @@ -4296,6 +4622,203 @@ public function getTotalBatch(string $tenant, array $metrics, array $queries = [ return $totals; } + /** + * One grouped total per metric from a single physical table: SUM for the + * raw events table, latest-sample argMax for gauges. + * + * @param array $metrics + * @param array $queries + * @return array + * @throws Exception + */ + private function totalBatchFromTable(string $tenant, array $metrics, array $queries, string $type): array + { + $tableName = $this->getTableForType($type); + + $parsed = $this->parseQueries($tenant, $queries, $type); + + $valueExpr = $type === Usage::TYPE_EVENT + ? 'SUM(`value`) AS `agg_val`' + : 'argMax(`value`, `time`) AS `agg_val`'; + + $builder = $this->newBuilder($type) + ->from($tableName) + ->select(['metric']) + ->selectRaw($valueExpr) + ->filter([Query::equal('metric', $metrics)]) + ->groupByRaw('`metric`'); + + $this->applyFilters($builder, $tenant, $parsed); + + $statement = $builder->build(); + $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; + + $totalsByMetric = []; + foreach ($this->decodeRows($this->query($sql, $statement->namedBindings ?? [])) as $row) { + $totalsByMetric[self::toStr($row['metric'] ?? null)] = self::toInt($row['agg_val'] ?? null); + } + + return $totalsByMetric; + } + + /** + * Routed batched event totals: the same source selection sum()/getTotal() + * use, applied to one grouped-by-metric query, with the decision recorded + * in the route log under `getTotalBatch`. + * + * @param array $metrics + * @param array $queries + * @return array + * @throws Exception + */ + private function routedTotalBatch(string $tenant, array $metrics, array $queries): array + { + $plan = $this->extractRoutingPlan(array_merge($queries, [Query::equal('metric', $metrics)])); + $route = $this->selectAggregateSource($plan); + $this->recordRoute('getTotalBatch', $plan, $route); + + if ($route === 'daily') { + $totalsByMetric = $this->sumDailyBatch($tenant, $metrics, $this->translateInclusiveMidnightForDaily($queries)); + $this->maybeDualReadBatch($tenant, $metrics, $queries, $route, $plan, $totalsByMetric); + return $totalsByMetric; + } + if ($route === 'hybrid') { + $totalsByMetric = $this->totalBatchHybrid($tenant, $metrics, $queries); + $this->maybeDualReadBatch($tenant, $metrics, $queries, $route, $plan, $totalsByMetric); + return $totalsByMetric; + } + if ($route === 'split') { + $totalsByMetric = $this->totalBatchSplit($tenant, $metrics, $queries, $plan); + $this->maybeDualReadBatch($tenant, $metrics, $queries, $route, $plan, $totalsByMetric); + return $totalsByMetric; + } + + return $this->totalBatchFromTable($tenant, $metrics, $queries, Usage::TYPE_EVENT); + } + + /** + * Hybrid batched event totals: closed days per metric from the daily + * rollup, today's partial from the raw events table, folded by an outer + * per-metric SUM over UNION ALL — the batched shape of + * sumHybridDailyAndRaw(). + * + * @param array $metrics + * @param array $queries + * @return array + * @throws Exception + */ + private function totalBatchHybrid(string $tenant, array $metrics, array $queries): array + { + $startOfToday = (new DateTime('today', new DateTimeZone('UTC')))->format('Y-m-d H:i:s.v'); + + $dailyTableName = $this->getEventsDailyTableName(); + $eventsTableName = $this->getEventsTableName(); + + $split = $this->splitTimeQueries($queries); + + $rawQueries = array_merge($queries, [Query::greaterThanEqual('time', $startOfToday)]); + $dailyQueries = array_merge( + $split['nonTime'], + $this->buildDailyTimeQueries($split['time']), + [Query::lessThan('time', $startOfToday)], + ); + + $rawParsed = $this->parseQueries($tenant, $rawQueries, Usage::TYPE_EVENT); + $dailyParsed = $this->parseQueries($tenant, $dailyQueries, Usage::TYPE_EVENT); + + $rawBuilder = $this->newBuilder(Usage::TYPE_EVENT) + ->from($eventsTableName) + ->select(['metric']) + ->selectRaw('SUM(`value`) AS `total`') + ->filter([Query::equal('metric', $metrics)]) + ->groupByRaw('`metric`'); + $this->applyFilters($rawBuilder, $tenant, $rawParsed); + $rawStatement = $rawBuilder->build(); + $rawSql = $this->qualifyDdl($rawStatement->query, $eventsTableName); + + $dailyBuilder = $this->newBuilder(Usage::TYPE_EVENT) + ->from($dailyTableName) + ->select(['metric']) + ->selectRaw('SUM(`value`) AS `total`') + ->filter([Query::equal('metric', $metrics)]) + ->groupByRaw('`metric`'); + $this->applyFilters($dailyBuilder, $tenant, $dailyParsed); + $dailyStatement = $dailyBuilder->build(); + [$dailySql, $dailyBindings] = $this->prefixNamedBindings( + $this->qualifyDdl($dailyStatement->query, $dailyTableName), + $dailyStatement->namedBindings ?? [], + 'd_', + ); + + $sql = " + SELECT `metric`, sum(`total`) AS `agg_val` FROM ( + {$dailySql} + UNION ALL + {$rawSql} + ) + GROUP BY `metric` + FORMAT JSON + "; + + $result = $this->query($sql, array_merge($rawStatement->namedBindings ?? [], $dailyBindings)); + + $totalsByMetric = []; + foreach ($this->decodeRows($result) as $row) { + $totalsByMetric[self::toStr($row['metric'] ?? null)] = self::toInt($row['agg_val'] ?? null); + } + + return $totalsByMetric; + } + + /** + * Batched twin of maybeDualRead(): with the same sampling, re-run the + * batch against the raw events table and log a warning per metric whose + * routed total diverges. + * + * @param array $metrics + * @param array $queries + * @param array{metric: ?string, start: ?string, end: ?string, filterColumns: array, dimensions: array, interval: ?string, orderColumns?: array, hasCursor?: bool, hasIrregularTimeFilter?: bool, endInclusive?: bool} $plan + * @param array $rolledTotals + */ + private function maybeDualReadBatch(string $tenant, array $metrics, array $queries, string $route, array $plan, array $rolledTotals): void + { + if ($this->dualReadSampleRate <= 0.0) { + return; + } + if (mt_rand() / mt_getrandmax() > $this->dualReadSampleRate) { + return; + } + + try { + $rawTotals = $this->totalBatchFromTable($tenant, $metrics, $queries, Usage::TYPE_EVENT); + } catch (Throwable $e) { + return; + } + + foreach ($metrics as $metric) { + $rawTotal = $rawTotals[$metric] ?? 0; + $rolledTotal = $rolledTotals[$metric] ?? 0; + + if ($rawTotal === 0 && $rolledTotal === 0) { + continue; + } + + $denominator = $rawTotal === 0 ? max(abs($rolledTotal), 1) : abs($rawTotal); + $delta = abs($rolledTotal - $rawTotal) / $denominator; + if ($delta > 0.01) { + $this->appendRouteLogEntry([ + 'operation' => 'dual_read_warning', + 'metric' => $metric, + 'route' => $route . ':delta=' . round($delta, 4), + 'start' => $plan['start'], + 'end' => $plan['end'], + 'dimensions' => $plan['dimensions'], + 'interval' => $plan['interval'], + ]); + } + } + } + /** * Resolve the ClickHouse parameter type for a column. * diff --git a/src/Usage/Usage.php b/src/Usage/Usage.php index 9572925..49d3b43 100644 --- a/src/Usage/Usage.php +++ b/src/Usage/Usage.php @@ -287,4 +287,17 @@ public function sumDailyBatch(string $tenant, array $metrics, array $queries = [ { return $this->adapter->sumDailyBatch($tenant, $metrics, $queries); } + + /** + * Backfill the pre-aggregated daily events table from the raw events + * table for the half-open, UTC-midnight-aligned window [$from, $to). + * Refuses a window that already holds rollup rows unless $force is + * passed after the caller cleared the range. + * + * @throws \Exception + */ + public function backfillDaily(string $from, string $to, bool $force = false): void + { + $this->adapter->backfillDaily($from, $to, $force); + } } diff --git a/tests/Usage/Adapter/ClickHouseBackfillTest.php b/tests/Usage/Adapter/ClickHouseBackfillTest.php new file mode 100644 index 0000000..a777746 --- /dev/null +++ b/tests/Usage/Adapter/ClickHouseBackfillTest.php @@ -0,0 +1,173 @@ +adapter = new ClickHouseAdapter( + getenv('CLICKHOUSE_HOST') ?: 'clickhouse', + getenv('CLICKHOUSE_USER') ?: 'default', + getenv('CLICKHOUSE_PASSWORD') ?: 'clickhouse', + (int) (getenv('CLICKHOUSE_PORT') ?: 8123), + (bool) (getenv('CLICKHOUSE_SECURE') ?: false), + namespace: 'utopia_usage_backfill', + database: getenv('CLICKHOUSE_DATABASE') ?: 'default', + sharedTables: true, + ); + $this->usage = new Usage($this->adapter); + $this->usage->setup(); + $this->usage->purge('1'); + + $this->seedHistoricalRow('backfill.metric', 100, '-5 days'); + $this->seedHistoricalRow('backfill.metric', 200, '-3 days'); + $this->seedHistoricalRow('backfill.other', 40, '-4 days'); + } + + protected function tearDown(): void + { + $this->usage->purge('1'); + } + + private function seedHistoricalRow(string $metric, int $value, string $modifier): void + { + $eventsTable = $this->resolveTableName($this->adapter, 'getEventsTableName'); + $database = $this->databaseName($this->adapter); + + $time = (new DateTime($modifier, new DateTimeZone('UTC')))->format('Y-m-d H:i:s.v'); + $id = bin2hex(random_bytes(16)); + + $this->queryRaw($this->adapter, sprintf( + "INSERT INTO `%s`.`%s` (id, metric, value, time, tenant) VALUES ('%s', '%s', %d, '%s', '1')", + $database, + $eventsTable, + $id, + addslashes($metric), + $value, + $time, + )); + } + + private function truncateDaily(): void + { + $dailyTable = $this->resolveTableName($this->adapter, 'getEventsDailyTableName'); + $database = $this->databaseName($this->adapter); + $this->queryRaw($this->adapter, sprintf('TRUNCATE TABLE `%s`.`%s`', $database, $dailyTable)); + } + + /** + * @return array{string, string} day-aligned [from, to) covering the seeds + */ + private function window(): array + { + $from = (new DateTime('-7 days', new DateTimeZone('UTC')))->setTime(0, 0, 0)->format('Y-m-d H:i:s'); + $to = (new DateTime('today', new DateTimeZone('UTC')))->format('Y-m-d H:i:s'); + + return [$from, $to]; + } + + /** + * @return array + */ + private function dailyTotals(string $from, string $to): array + { + return $this->usage->sumDailyBatch('1', ['backfill.metric', 'backfill.other'], [ + Query::greaterThanEqual('time', $from), + Query::lessThan('time', $to), + ]); + } + + public function testBackfillRestoresTheRollupToRawTotals(): void + { + [$from, $to] = $this->window(); + + // The MV rolled the seeds up on insert; wipe the rollup so the window + // is genuinely missing, as it is for any table that predates the MV. + $this->truncateDaily(); + $this->assertSame( + ['backfill.metric' => 0, 'backfill.other' => 0], + $this->dailyTotals($from, $to), + 'the rollup must be empty before the backfill for this test to prove anything', + ); + + $this->usage->backfillDaily($from, $to); + + $this->assertSame( + ['backfill.metric' => 300, 'backfill.other' => 40], + $this->dailyTotals($from, $to), + 'a backfilled window must re-aggregate to the exact totals the raw table holds', + ); + } + + public function testBackfillRefusesAWindowTheRollupAlreadyCovers(): void + { + [$from, $to] = $this->window(); + + // The MV already rolled the seeds up on insert. + $this->expectException(Exception::class); + $this->expectExceptionMessage('double-count'); + + $this->usage->backfillDaily($from, $to); + } + + public function testForceBackfillsOverAnOverlappingWindow(): void + { + [$from, $to] = $this->window(); + + // With force, the caller owns overlap safety; here the rollup was + // cleared first, so force writes the same rows the guard path would. + $this->truncateDaily(); + $this->usage->backfillDaily($from, $to, force: true); + + $this->assertSame( + ['backfill.metric' => 300, 'backfill.other' => 40], + $this->dailyTotals($from, $to), + ); + } + + public function testBackfillRejectsPartialDayBounds(): void + { + $from = (new DateTime('-7 days', new DateTimeZone('UTC')))->setTime(12, 0, 0)->format('Y-m-d H:i:s'); + $to = (new DateTime('today', new DateTimeZone('UTC')))->format('Y-m-d H:i:s'); + + $this->expectException(Exception::class); + $this->expectExceptionMessage('UTC midnights'); + + $this->usage->backfillDaily($from, $to); + } + + public function testBackfillRejectsNonUtcMidnightOffsets(): void + { + // Midnight in +05:00 is 19:00 UTC — accepting it would backfill a + // different range than the caller intended. + $to = (new DateTime('today', new DateTimeZone('UTC')))->format('Y-m-d H:i:s'); + + $this->expectException(Exception::class); + $this->expectExceptionMessage('UTC midnights'); + + $this->usage->backfillDaily('2026-07-01 00:00:00+05:00', $to); + } + + public function testBackfillRejectsAnInvertedWindow(): void + { + [$from, $to] = $this->window(); + + $this->expectException(Exception::class); + $this->expectExceptionMessage('ascending'); + + $this->usage->backfillDaily($to, $from); + } +} diff --git a/tests/Usage/Adapter/ClickHouseGaugeProjectionTest.php b/tests/Usage/Adapter/ClickHouseGaugeProjectionTest.php new file mode 100644 index 0000000..f9a6c7e --- /dev/null +++ b/tests/Usage/Adapter/ClickHouseGaugeProjectionTest.php @@ -0,0 +1,135 @@ +adapter = new ClickHouseAdapter( + getenv('CLICKHOUSE_HOST') ?: 'clickhouse', + getenv('CLICKHOUSE_USER') ?: 'default', + getenv('CLICKHOUSE_PASSWORD') ?: 'clickhouse', + (int) (getenv('CLICKHOUSE_PORT') ?: 8123), + (bool) (getenv('CLICKHOUSE_SECURE') ?: false), + namespace: 'utopia_usage_gauge_proj', + database: getenv('CLICKHOUSE_DATABASE') ?: 'default', + sharedTables: true, + ); + $this->usage = new Usage($this->adapter); + + // A leftover table from an earlier run may carry the old projection + // shape under the same name (setup() adds IF NOT EXISTS); drop it so + // the schema under test is the one this code creates. + $gauges = $this->resolveTableName($this->adapter, 'getGaugesTableName'); + $database = $this->databaseName($this->adapter); + $this->queryRaw($this->adapter, "DROP TABLE IF EXISTS `{$database}`.`{$gauges}`"); + + $this->usage->setup(); + $this->usage->purge('1'); + + // Two samples per series so argMax has something to resolve, plus a + // second resourceType under the same metric. + $this->seedGaugeRow('proj.users', 5, 'user', '-3 days'); + $this->seedGaugeRow('proj.users', 9, 'user', '-1 hour'); + $this->seedGaugeRow('proj.users', 2, 'team', '-2 days'); + } + + protected function tearDown(): void + { + $this->usage->purge('1'); + } + + private function seedGaugeRow(string $metric, int $value, string $resourceType, string $modifier): void + { + $gauges = $this->resolveTableName($this->adapter, 'getGaugesTableName'); + $database = $this->databaseName($this->adapter); + $time = (new DateTime($modifier, new DateTimeZone('UTC')))->format('Y-m-d H:i:s.v'); + $id = bin2hex(random_bytes(16)); + + $this->queryRaw($this->adapter, sprintf( + "INSERT INTO `%s`.`%s` (id, metric, value, time, resourceType, tenant) VALUES ('%s', '%s', %d, '%s', '%s', '1')", + $database, + $gauges, + $id, + addslashes($metric), + $value, + $time, + addslashes($resourceType), + )); + } + + public function testGroupedLatestValueReadIsServedByAProjection(): void + { + $gauges = $this->resolveTableName($this->adapter, 'getGaugesTableName'); + $database = $this->databaseName($this->adapter); + + // force_optimize_projection makes ClickHouse ERROR when no projection + // matches — success here IS the assertion that the shape matches. + $forced = $this->queryRaw($this->adapter, sprintf( + 'SELECT `metric`, argMax(`value`, `time`) AS `value`, `resourceType`' + . " FROM `%s`.`%s` WHERE `tenant` IN ('1') AND `metric` IN ('proj.users')" + . ' GROUP BY `metric`, `resourceType`' + . ' SETTINGS optimize_use_projections = 1, force_optimize_projection = 1' + . ' FORMAT JSON', + $database, + $gauges, + )); + + $decoded = json_decode($forced, true); + $rows = is_array($decoded) && is_array($decoded['data'] ?? null) ? $decoded['data'] : []; + $byType = []; + foreach ($rows as $row) { + if (is_array($row) && is_string($row['resourceType'] ?? null) && is_numeric($row['value'] ?? null)) { + $byType[$row['resourceType']] = (int) $row['value']; + } + } + + $this->assertSame( + ['user' => 9, 'team' => 2], + ['user' => $byType['user'] ?? null, 'team' => $byType['team'] ?? null], + 'the projection-served read must return the latest sample per series, same as raw argMax', + ); + } + + public function testAdapterGroupedReadMatchesProjectionServedValues(): void + { + // The adapter's own grouped read (the billing prefetch shape) must + // agree with the projection-served result. + $rows = $this->usage->find('1', [ + Query::equal('metric', ['proj.users']), + UsageQuery::groupBy('resourceType'), + Query::limit(100), + ], Usage::TYPE_GAUGE); + + $byType = []; + foreach ($rows as $row) { + $resourceType = $row->getAttribute('resourceType'); + if (is_string($resourceType)) { + $byType[$resourceType] = (int) $row->getValue(); + } + } + + $this->assertSame(9, $byType['user'] ?? null); + $this->assertSame(2, $byType['team'] ?? null); + } +} diff --git a/tests/Usage/Adapter/ClickHouseRoutingTest.php b/tests/Usage/Adapter/ClickHouseRoutingTest.php index 862eee4..8ce7764 100644 --- a/tests/Usage/Adapter/ClickHouseRoutingTest.php +++ b/tests/Usage/Adapter/ClickHouseRoutingTest.php @@ -91,21 +91,29 @@ public function testClosedDayWindowRoutesToDaily(): void $log = $this->adapter->getRouteLog(); $this->assertCount(1, $log); - $this->assertSame('daily', $log[0]['route']); + $this->assertSame('split', $log[0]['route'], 'an inclusive midnight bound needs the boundary instant read raw; the interior still comes from the rollup'); $this->assertSame($rawSum, $sum, 'daily MV must re-aggregate to the same total as raw'); } - public function testInclusiveMidnightUpperBoundExcludesEndDayOnDailyRoute(): void + public function testInclusiveMidnightUpperBoundExcludesTheEndDayButKeepsTheMidnightInstant(): void { - $this->adapter->clearRouteLog(); - + // Two rows on the end day: one at 14:00 (must be excluded — the bound + // is midnight, not end-of-day) and one at exactly midnight (must be + // included — the bound is inclusive). A day-granularity rollup row + // cannot express that difference: routing this window to 'daily' + // translated `<= midnight` into `< midnight` and silently dropped the + // midnight row, under-billing by its value. It now routes 'split': + // interior days from the rollup, the boundary instant from raw. $this->seedHistoricalRow('routed.metric', 9999, '-2 days +14 hours', ['path' => '/v1/late']); $start = (new DateTime('-7 days', new DateTimeZone('UTC')))->setTime(0, 0, 0)->format('Y-m-d H:i:s'); $end = (new DateTime('-2 days', new DateTimeZone('UTC')))->setTime(0, 0, 0)->format('Y-m-d H:i:s'); + $this->seedHistoricalRowAt('routed.metric', 42, $end); + $rawSum = $this->sumRaw('routed.metric', $start, $end); + $this->adapter->clearRouteLog(); $sum = $this->usage->sum('1', [ Query::equal('metric', ['routed.metric']), Query::greaterThanEqual('time', $start), @@ -114,23 +122,24 @@ public function testInclusiveMidnightUpperBoundExcludesEndDayOnDailyRoute(): voi $log = $this->adapter->getRouteLog(); $this->assertCount(1, $log); - $this->assertSame('daily', $log[0]['route']); - $this->assertSame($rawSum, $sum, 'daily MV must not include the end-day full-day row for inclusive-midnight upper bounds'); - $this->assertSame(300, $sum); + $this->assertSame('split', $log[0]['route']); + $this->assertSame($rawSum, $sum, 'the routed read must match raw exactly at an inclusive midnight bound'); + $this->assertSame(342, $sum, 'the 300 interior + the 42 row at exactly midnight, never the 9999 mid-end-day row'); } - public function testMidDayClosedWindowFallsBackToRaw(): void + public function testMidDayClosedWindowRoutesSplit(): void { - // Daily rows are stored at midnight; a mid-day caller bound - // would exclude the partial first day and over-include the - // last day if forwarded to the daily MV. Routing must reject - // non-day-aligned bounds and fall through to the raw scan. - $this->adapter->clearRouteLog(); - + // Daily rows are stored at midnight; a mid-day caller bound cannot be + // forwarded to the daily MV wholesale. It routes 'split' instead: the + // interior whole days from the rollup, the partial edge days from raw + // — never over- or under-including the edges. $start = (new DateTime('-7 days 12:30:00', new DateTimeZone('UTC')))->format('Y-m-d H:i:s'); $end = (new DateTime('-2 days 12:30:00', new DateTimeZone('UTC')))->format('Y-m-d H:i:s'); - $this->usage->sum('1', [ + $rawSum = $this->sumRaw('routed.metric', $start, $end); + + $this->adapter->clearRouteLog(); + $sum = $this->usage->sum('1', [ Query::equal('metric', ['routed.metric']), Query::greaterThanEqual('time', $start), Query::lessThanEqual('time', $end), @@ -138,7 +147,8 @@ public function testMidDayClosedWindowFallsBackToRaw(): void $log = $this->adapter->getRouteLog(); $this->assertCount(1, $log); - $this->assertSame('raw', $log[0]['route']); + $this->assertSame('split', $log[0]['route']); + $this->assertSame($rawSum, $sum, 'split must equal the raw scan over the same mid-day bounds'); } public function testWindowStraddlesTodayRoutesHybrid(): void @@ -239,7 +249,7 @@ public function testMidDayStartWithHybridWindowFallsBackToRaw(): void $log = $this->adapter->getRouteLog(); $this->assertCount(1, $log); - $this->assertSame('raw', $log[0]['route'], 'a mid-day start with a hybrid window must fall back to raw'); + $this->assertSame('split', $log[0]['route'], 'a mid-day start with a straddling window routes split: rollup interior, raw edges'); $this->assertSame(0, $sum, 'pre-start events on the same day must not be included in the result'); } @@ -281,7 +291,7 @@ public function testDuplicateTimeFiltersTakeTightestBound(): void $log = $this->adapter->getRouteLog(); $this->assertCount(1, $log); - $this->assertSame('daily', $log[0]['route']); + $this->assertSame('split', $log[0]['route'], 'tightest bound is an inclusive midnight, which routes split'); $this->assertSame($startTighter, $log[0]['start']); $this->assertSame($endTighter, $log[0]['end']); } @@ -394,6 +404,221 @@ public function testValueFilterPurgeDoesNotMatchAggregateDailyRows(): void $this->assertSame(10, $dailySum, 'daily MV row must survive a value-only purge'); } + public function testClosedDayWindowRoutesTotalBatchToDaily(): void + { + $this->adapter->clearRouteLog(); + + $start = (new DateTime('-7 days', new DateTimeZone('UTC')))->setTime(0, 0, 0)->format('Y-m-d H:i:s'); + $end = (new DateTime('-2 days', new DateTimeZone('UTC')))->setTime(0, 0, 0)->format('Y-m-d H:i:s'); + + $rawSum = $this->sumRaw('routed.metric', $start, $end); + + $totals = $this->usage->getTotalBatch('1', ['routed.metric', 'routed.absent'], [ + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + ], Usage::TYPE_EVENT); + + $log = $this->adapter->getRouteLog(); + $this->assertCount(1, $log); + $this->assertSame('getTotalBatch', $log[0]['operation']); + $this->assertSame('split', $log[0]['route'], 'an inclusive midnight bound needs the boundary instant read raw; the interior still comes from the rollup'); + $this->assertSame($rawSum, $totals['routed.metric'], 'the daily rollup must re-aggregate to the same batch total as raw'); + $this->assertSame(0, $totals['routed.absent'], 'an absent metric still comes back as zero'); + } + + public function testOpenWindowRoutesTotalBatchToHybrid(): void + { + $this->adapter->clearRouteLog(); + + $start = (new DateTime('-7 days', new DateTimeZone('UTC')))->setTime(0, 0, 0)->format('Y-m-d H:i:s'); + $end = (new DateTime('+1 day', new DateTimeZone('UTC')))->setTime(0, 0, 0)->format('Y-m-d H:i:s'); + + $rawSum = $this->sumRaw('routed.metric', $start, $end); + + $totals = $this->usage->getTotalBatch('1', ['routed.metric'], [ + Query::greaterThanEqual('time', $start), + Query::lessThan('time', $end), + ], Usage::TYPE_EVENT); + + $log = $this->adapter->getRouteLog(); + $this->assertCount(1, $log); + $this->assertSame('getTotalBatch', $log[0]['operation']); + $this->assertSame('hybrid', $log[0]['route']); + $this->assertSame($rawSum, $totals['routed.metric'], 'closed days from the rollup plus today from raw must equal the raw batch total'); + } + + public function testNonRollupFilterKeepsTotalBatchOnRaw(): void + { + $this->adapter->clearRouteLog(); + + $start = (new DateTime('-7 days', new DateTimeZone('UTC')))->setTime(0, 0, 0)->format('Y-m-d H:i:s'); + $end = (new DateTime('-2 days', new DateTimeZone('UTC')))->setTime(0, 0, 0)->format('Y-m-d H:i:s'); + + $totals = $this->usage->getTotalBatch('1', ['routed.metric'], [ + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + Query::equal('path', ['/v1/a']), + ], Usage::TYPE_EVENT); + + $log = $this->adapter->getRouteLog(); + $this->assertCount(1, $log); + $this->assertSame('getTotalBatch', $log[0]['operation']); + $this->assertSame('raw', $log[0]['route'], 'a filter column the rollup does not carry must fall back to the raw table'); + $this->assertSame(100, $totals['routed.metric']); + } + + public function testNonAlignedWindowRoutesSumToSplit(): void + { + // Billing-shaped window: mid-day start, mid-day end — 0 of 451k + // production teams have midnight-aligned invoice dates, so this is + // the shape routing must serve or it serves nothing. + $day = fn (int $back, string $time): string => + (new DateTime("-{$back} days", new DateTimeZone('UTC')))->format('Y-m-d') . ' ' . $time; + + $start = $day(7, '14:30:00'); + $end = $day(1, '10:00:00'); + + // Boundary-precise seeds: excluded-before, head edge, interior, tail + // edge, excluded-after. + $this->seedHistoricalRowAt('routed.metric', 1000, $day(7, '10:00:00')); + $this->seedHistoricalRowAt('routed.metric', 3, $day(7, '18:00:00')); + $this->seedHistoricalRowAt('routed.metric', 7, $day(4, '12:00:00')); + $this->seedHistoricalRowAt('routed.metric', 13, $day(1, '08:00:00')); + $this->seedHistoricalRowAt('routed.metric', 5000, $day(1, '11:30:00')); + + $queries = [ + Query::equal('metric', ['routed.metric']), + Query::greaterThanEqual('time', $start), + Query::lessThan('time', $end), + ]; + + $rawSum = $this->sumRawHalfOpen('routed.metric', $start, $end); + + $this->adapter->clearRouteLog(); + $sum = $this->usage->sum('1', $queries, 'value', Usage::TYPE_EVENT); + + $log = $this->adapter->getRouteLog(); + $this->assertCount(1, $log); + $this->assertSame('split', $log[0]['route']); + $this->assertSame($rawSum, $sum, 'daily interior + raw edges must partition the window exactly'); + } + + public function testNonAlignedWindowRoutesTotalBatchToSplit(): void + { + $day = fn (int $back, string $time): string => + (new DateTime("-{$back} days", new DateTimeZone('UTC')))->format('Y-m-d') . ' ' . $time; + + $start = $day(7, '14:30:00'); + $end = $day(1, '10:00:00'); + + $this->seedHistoricalRowAt('routed.metric', 3, $day(7, '18:00:00')); + $this->seedHistoricalRowAt('routed.other', 21, $day(4, '12:00:00')); + $this->seedHistoricalRowAt('routed.other', 9, $day(1, '08:00:00')); + + $queries = [ + Query::greaterThanEqual('time', $start), + Query::lessThan('time', $end), + ]; + + $expected = [ + 'routed.metric' => $this->sumRawHalfOpen('routed.metric', $start, $end), + 'routed.other' => $this->sumRawHalfOpen('routed.other', $start, $end), + ]; + + $this->adapter->clearRouteLog(); + $totals = $this->usage->getTotalBatch('1', ['routed.metric', 'routed.other'], $queries, Usage::TYPE_EVENT); + + $log = $this->adapter->getRouteLog(); + $this->assertCount(1, $log); + $this->assertSame('getTotalBatch', $log[0]['operation']); + $this->assertSame('split', $log[0]['route']); + $this->assertSame($expected, ['routed.metric' => $totals['routed.metric'], 'routed.other' => $totals['routed.other']]); + } + + public function testIrregularTimeFilterStaysRaw(): void + { + // A notBetween carves a mid-day hole inside the window. Day-granularity + // rollup rows cannot honor it, and the split interior would drop it — + // both would over-count, so it must stay raw. Value-checked: the raw + // path applies the hole, and routed reads must match it. + $day = fn (int $back, string $time): string => + (new DateTime("-{$back} days", new DateTimeZone('UTC')))->format('Y-m-d') . ' ' . $time; + + $start = $day(7, '00:00:00'); + $end = $day(2, '00:00:00'); + + $this->seedHistoricalRowAt('routed.metric', 40, $day(4, '11:00:00')); + + $this->adapter->clearRouteLog(); + $sum = $this->usage->sum('1', [ + Query::equal('metric', ['routed.metric']), + Query::greaterThanEqual('time', $start), + Query::lessThan('time', $end), + Query::notBetween('time', $day(4, '10:00:00'), $day(4, '12:00:00')), + ], 'value', Usage::TYPE_EVENT); + + $log = $this->adapter->getRouteLog(); + $this->assertCount(1, $log); + $this->assertSame('raw', $log[0]['route'], 'a non-window time filter cannot be expressed on day rows'); + + $rawWithoutHole = $this->sumRawHalfOpen('routed.metric', $start, $end); + $this->assertSame($rawWithoutHole - 40, $sum, 'the mid-day hole must exclude the seeded row'); + } + + public function testSubDayWindowStaysRaw(): void + { + $day = fn (int $back, string $time): string => + (new DateTime("-{$back} days", new DateTimeZone('UTC')))->format('Y-m-d') . ' ' . $time; + + $this->adapter->clearRouteLog(); + + $this->usage->sum('1', [ + Query::equal('metric', ['routed.metric']), + Query::greaterThanEqual('time', $day(1, '10:00:00')), + Query::lessThan('time', $day(1, '14:00:00')), + ], 'value', Usage::TYPE_EVENT); + + $log = $this->adapter->getRouteLog(); + $this->assertCount(1, $log); + $this->assertSame('raw', $log[0]['route'], 'a window with no whole interior day has nothing the rollup can answer'); + } + + /** + * Raw ground truth over a half-open window, matching the routed calls. + */ + private function sumRawHalfOpen(string $metric, string $start, string $end): int + { + $reflection = new ReflectionClass($this->adapter); + $sumFromTable = $reflection->getMethod('sumFromTable'); + $sumFromTable->setAccessible(true); + $result = $sumFromTable->invoke($this->adapter, '1', [ + Query::equal('metric', [$metric]), + Query::greaterThanEqual('time', $start), + Query::lessThan('time', $end), + ], 'value', Usage::TYPE_EVENT); + $this->adapter->clearRouteLog(); + return is_int($result) ? $result : 0; + } + + /** + * Seed one raw event row at an absolute UTC timestamp. + */ + private function seedHistoricalRowAt(string $metric, int $value, string $timestamp): void + { + $eventsTable = $this->resolveTableName($this->adapter, 'getEventsTableName'); + $database = $this->databaseName($this->adapter); + $id = bin2hex(random_bytes(16)); + $this->queryRaw($this->adapter, sprintf( + "INSERT INTO `%s`.`%s` (id, metric, value, time, tenant) VALUES ('%s', '%s', %d, '%s.000', '1')", + $database, + $eventsTable, + $id, + addslashes($metric), + $value, + $timestamp, + )); + } + private function sumRaw(string $metric, string $start, string $end): int { $reflection = new ReflectionClass($this->adapter); diff --git a/tests/Usage/Adapter/ClickHouseSchemaTest.php b/tests/Usage/Adapter/ClickHouseSchemaTest.php index 1742f6a..58c8510 100644 --- a/tests/Usage/Adapter/ClickHouseSchemaTest.php +++ b/tests/Usage/Adapter/ClickHouseSchemaTest.php @@ -70,19 +70,27 @@ public function testEventProjectionsLeadWithTenantAndKeyOnTheHourlyBucket(): voi $this->assertStringContainsString("toStartOfHour(time, 'UTC') AS timeBucket", $ddl); } - public function testGaugeProjectionsAreLeftOnTheirOriginalShape(): void + public function testGaugeProjectionsCarryBothTheWindowedAndLatestSlates(): void { $ddl = $this->showCreate($this->resolveTableName($this->adapter, 'getGaugesTableName')); - // Gauges are deliberately excluded from the events reshape: they show - // no measured read problem, and bucketing `time` away would only cost - // a migration, since argMax orders on the raw column. Pinned here so - // the exclusion is not "finished" without fresh measurements. + // The windowed slate keeps its original time-keyed shape — windowed + // grouped reads need the time predicate expressible on the projection. $this->assertStringContainsString( "GROUP BY\n metric,\n time,\n tenant,\n service", $ddl ); $this->assertStringNotContainsString('timeBucket', $ddl); + + // The fresh measurements the old pin asked for arrived: unwindowed + // grouped latest-value reads (the billing prefetch) read ~300k rows + // per query against the time-keyed slate and force_optimize_projection + // refuses it. The latest slate answers them at one state per series: + // keyed (tenant, metric, dims), no time key. + $this->assertStringContainsString( + "PROJECTION p_latest_by_service\n (\n SELECT\n tenant,\n metric,\n service,\n argMax(value, time) AS value\n GROUP BY\n tenant,\n metric,\n service", + $ddl + ); } public function testEventsTableSwapsBloomForSetOnLowCardinality(): void diff --git a/tests/Usage/Adapter/ClickHouseSplitParityTest.php b/tests/Usage/Adapter/ClickHouseSplitParityTest.php new file mode 100644 index 0000000..b5302cd --- /dev/null +++ b/tests/Usage/Adapter/ClickHouseSplitParityTest.php @@ -0,0 +1,188 @@ +adapter = new ClickHouseAdapter( + getenv('CLICKHOUSE_HOST') ?: 'clickhouse', + getenv('CLICKHOUSE_USER') ?: 'default', + getenv('CLICKHOUSE_PASSWORD') ?: 'clickhouse', + (int) (getenv('CLICKHOUSE_PORT') ?: 8123), + (bool) (getenv('CLICKHOUSE_SECURE') ?: false), + namespace: 'utopia_usage_split_parity', + database: getenv('CLICKHOUSE_DATABASE') ?: 'default', + sharedTables: true, + ); + $this->usage = new Usage($this->adapter); + $this->usage->setup(); + $this->usage->purge('1'); + + // Boundary-dense seeding: every day from 40 days ago through today + // gets rows at midnight exactly, mid-morning, and one millisecond + // before midnight — so any off-by-one at a day edge changes a total. + for ($back = 40; $back >= 0; $back--) { + $day = (new DateTime("-{$back} days", new DateTimeZone('UTC')))->format('Y-m-d'); + $this->seedAt($this->metricA, 100 + $back, $day . ' 00:00:00.000'); + $this->seedAt($this->metricA, 200 + $back, $day . ' 09:15:00.000'); + $this->seedAt($this->metricA, 300 + $back, $day . ' 23:59:59.999'); + $this->seedAt($this->metricB, 7, $day . ' 12:00:00.000'); + } + } + + protected function tearDown(): void + { + $this->usage->purge('1'); + } + + private function seedAt(string $metric, int $value, string $timestamp): void + { + $eventsTable = $this->resolveTableName($this->adapter, 'getEventsTableName'); + $database = $this->databaseName($this->adapter); + $id = bin2hex(random_bytes(16)); + + $this->queryRaw($this->adapter, sprintf( + "INSERT INTO `%s`.`%s` (id, metric, value, time, tenant) VALUES ('%s', '%s', %d, '%s', '1')", + $database, + $eventsTable, + $id, + addslashes($metric), + $value, + $timestamp, + )); + } + + /** + * @return array + * label => [start, end, inclusiveEnd, expectedRoute] + */ + public static function windowShapes(): array + { + $day = static fn (int $back): string => + (new DateTime("-{$back} days", new DateTimeZone('UTC')))->format('Y-m-d'); + + return [ + // Billing-shaped: anchored at an arbitrary time of day, closed. + 'mid-day both ends' => [$day(30) . ' 14:37:11', $day(3) . ' 14:37:11', false, 'split'], + 'mid-day start, midnight end' => [$day(30) . ' 06:00:00', $day(3) . ' 00:00:00', false, 'split'], + 'midnight start, mid-day end' => [$day(30) . ' 00:00:00', $day(3) . ' 18:20:00', false, 'split'], + // Day-aligned closed: the pre-existing daily route. + 'both midnight' => [$day(30) . ' 00:00:00', $day(3) . ' 00:00:00', false, 'daily'], + // Reaching into the running day. + 'mid-day start, straddles today' => [$day(30) . ' 14:00:00', $day(-1) . ' 00:00:00', false, 'split'], + 'midnight start, straddles today' => [$day(30) . ' 00:00:00', $day(-1) . ' 00:00:00', false, 'hybrid'], + // Inclusive upper bound (lessThanEqual) at both alignments. + 'inclusive mid-day end' => [$day(30) . ' 08:00:00', $day(3) . ' 08:00:00', true, 'split'], + // Inclusive at midnight covers the midnight instant, which a day + // row cannot represent — split reads that instant from raw. The + // pre-existing 'daily' route silently dropped it. + 'inclusive midnight end' => [$day(30) . ' 00:00:00', $day(3) . ' 00:00:00', true, 'split'], + // Degenerate/small: no whole interior day to route. + 'sub-day window' => [$day(5) . ' 06:00:00', $day(5) . ' 18:00:00', false, 'raw'], + 'exactly one interior day' => [$day(6) . ' 13:00:00', $day(4) . ' 11:00:00', false, 'split'], + // Long window, the shape a monthly invoice cycle actually uses. + 'month-long mid-day' => [$day(35) . ' 03:33:33', $day(4) . ' 21:11:11', false, 'split'], + ]; + } + + /** + * @dataProvider windowShapes + */ + public function testRoutedSumEqualsRawForEveryWindowShape(string $start, string $end, bool $inclusiveEnd, string $expectedRoute): void + { + $bound = $inclusiveEnd + ? Query::lessThanEqual('time', $end) + : Query::lessThan('time', $end); + + $queries = [ + Query::equal('metric', [$this->metricA]), + Query::greaterThanEqual('time', $start), + $bound, + ]; + + $raw = $this->rawSum($queries); + + $this->adapter->clearRouteLog(); + $routed = $this->usage->sum('1', $queries, 'value', Usage::TYPE_EVENT); + $log = $this->adapter->getRouteLog(); + + $this->assertCount(1, $log); + $this->assertSame($expectedRoute, $log[0]['route'], "unexpected route for [{$start}, {$end}" . ($inclusiveEnd ? ']' : ')') . ']'); + $this->assertSame($raw, $routed, "routed sum diverged from raw for [{$start}, {$end}" . ($inclusiveEnd ? ']' : ')') . ']'); + $this->assertGreaterThan(0, $raw, 'the shape must actually cover seeded rows or it proves nothing'); + } + + /** + * @dataProvider windowShapes + */ + public function testRoutedBatchEqualsRawForEveryWindowShape(string $start, string $end, bool $inclusiveEnd, string $expectedRoute): void + { + $bound = $inclusiveEnd + ? Query::lessThanEqual('time', $end) + : Query::lessThan('time', $end); + + $queries = [ + Query::greaterThanEqual('time', $start), + $bound, + ]; + + $expected = [ + $this->metricA => $this->rawSum(array_merge($queries, [Query::equal('metric', [$this->metricA])])), + $this->metricB => $this->rawSum(array_merge($queries, [Query::equal('metric', [$this->metricB])])), + ]; + + $this->adapter->clearRouteLog(); + $totals = $this->usage->getTotalBatch('1', [$this->metricA, $this->metricB], $queries, Usage::TYPE_EVENT); + $log = $this->adapter->getRouteLog(); + + $this->assertCount(1, $log); + $this->assertSame($expectedRoute, $log[0]['route']); + $this->assertSame( + $expected, + [$this->metricA => $totals[$this->metricA], $this->metricB => $totals[$this->metricB]], + "routed batch diverged from raw for [{$start}, {$end}" . ($inclusiveEnd ? ']' : ')') . ']', + ); + } + + /** + * Ground truth straight from the raw events table. + * + * @param array $queries + */ + private function rawSum(array $queries): int + { + $reflection = new \ReflectionClass($this->adapter); + $method = $reflection->getMethod('sumFromTable'); + $method->setAccessible(true); + $result = $method->invoke($this->adapter, '1', $queries, 'value', Usage::TYPE_EVENT); + $this->adapter->clearRouteLog(); + + return is_int($result) ? $result : 0; + } +}