From 4c5a403401283f41c867bb41c471fd7924f0a462 Mon Sep 17 00:00:00 2001 From: Levi van Noort <73097785+levivannoort@users.noreply.github.com> Date: Fri, 4 Sep 2026 10:53:30 +0200 Subject: [PATCH 1/3] feat(clickhouse): route getTotalBatch through the daily rollup; add backfillDaily MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit getTotalBatch() always scanned the raw events table, even for windows the daily rollup answers — for Appwrite billing that is a whole-month raw scan per metric batch per project per hour, and it is the shape that saturates ClickHouse once the raw table grows. Batched event totals now route through the same source selection sum()/getTotal() use: closed-day windows read the daily rollup, open windows read rollup + today's raw tail, and anything the rollup cannot answer (non-rollup filter columns, cursors, sub-day bounds) stays raw. Decisions land in the route log under `getTotalBatch`, and the dualReadSampleRate canary gets a batched twin that logs per-metric drift. Gauges are untouched — they have no rollup. Routing is only correct once the rollup covers the window, and the daily MV was created without POPULATE, so backfillDaily($from, $to) runs the exact MV aggregation over a half-open, UTC-midnight-aligned window. SummingMergeTree folds duplicates by adding them, so the call refuses misaligned bounds and refuses a window that already holds rollup rows unless force is passed after the caller cleared the range. --- src/Usage/Adapter.php | 15 + src/Usage/Adapter/ClickHouse.php | 309 ++++++++++++++++-- src/Usage/Usage.php | 13 + .../Usage/Adapter/ClickHouseBackfillTest.php | 161 +++++++++ tests/Usage/Adapter/ClickHouseRoutingTest.php | 63 ++++ 5 files changed, 534 insertions(+), 27 deletions(-) create mode 100644 tests/Usage/Adapter/ClickHouseBackfillTest.php 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..7b08e32 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -1643,6 +1643,88 @@ 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. + * + * @throws Exception + */ + public function backfillDaily(string $from, string $to, bool $force = false): void + { + try { + $fromDt = new DateTime($from, new DateTimeZone('UTC')); + $toDt = new DateTime($to, 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. * @@ -4244,37 +4326,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 +4359,198 @@ 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; + } + + 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} $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..1a8f6d7 --- /dev/null +++ b/tests/Usage/Adapter/ClickHouseBackfillTest.php @@ -0,0 +1,161 @@ +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 testBackfillRejectsAnInvertedWindow(): void + { + [$from, $to] = $this->window(); + + $this->expectException(Exception::class); + $this->expectExceptionMessage('ascending'); + + $this->usage->backfillDaily($to, $from); + } +} diff --git a/tests/Usage/Adapter/ClickHouseRoutingTest.php b/tests/Usage/Adapter/ClickHouseRoutingTest.php index 862eee4..e3c3f4e 100644 --- a/tests/Usage/Adapter/ClickHouseRoutingTest.php +++ b/tests/Usage/Adapter/ClickHouseRoutingTest.php @@ -394,6 +394,69 @@ 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('daily', $log[0]['route']); + $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']); + } + private function sumRaw(string $metric, string $start, string $end): int { $reflection = new ReflectionClass($this->adapter); From 1f96a94de578f6b819d58570a72e335980f6818c Mon Sep 17 00:00:00 2001 From: Levi van Noort <73097785+levivannoort@users.noreply.github.com> Date: Fri, 4 Sep 2026 12:45:13 +0200 Subject: [PATCH 2/3] fix(clickhouse): normalize backfill bounds to UTC; document the concurrency contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A bound carrying its own offset ('2026-01-01 00:00:00+05:00') passed the midnight check in its own zone but was bound offset-less, so ClickHouse read it as UTC midnight — a different instant than the caller intended. Bounds now normalize to UTC before validation, so offset midnights that are not UTC midnights are refused. backfillDaily() is also documented as a single-operator migration primitive: the overlap check and the INSERT are separate statements, so concurrent calls for the same window can double the rollup — serialize, and parity-check before routing traffic. --- src/Usage/Adapter/ClickHouse.php | 15 +++++++++++++-- tests/Usage/Adapter/ClickHouseBackfillTest.php | 12 ++++++++++++ 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index 7b08e32..091e9e2 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -1659,13 +1659,24 @@ private function createDailyMaterializedView(): void * 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 { - $fromDt = new DateTime($from, new DateTimeZone('UTC')); - $toDt = new DateTime($to, new DateTimeZone('UTC')); + // 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); } diff --git a/tests/Usage/Adapter/ClickHouseBackfillTest.php b/tests/Usage/Adapter/ClickHouseBackfillTest.php index 1a8f6d7..a777746 100644 --- a/tests/Usage/Adapter/ClickHouseBackfillTest.php +++ b/tests/Usage/Adapter/ClickHouseBackfillTest.php @@ -149,6 +149,18 @@ public function testBackfillRejectsPartialDayBounds(): void $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(); From bc4c2b3aac32ebd26f343d78e2f88a98287ee6e4 Mon Sep 17 00:00:00 2001 From: Levi van Noort <73097785+levivannoort@users.noreply.github.com> Date: Fri, 4 Sep 2026 14:17:57 +0200 Subject: [PATCH 3/3] feat(clickhouse): split route for non-aligned windows; latest-value gauge projections MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two measured gaps, both verified against the largest production install. 1. Routing never fired for the windows production uses. daily/hybrid require a midnight-aligned start, but billing cycles are anchored at the moment a team upgraded: 0 of 451,293 production teams have a midnight-aligned invoice date. Every billing read therefore fell to raw. The new 'split' route reads the interior whole days from the rollup and only the partial head/tail days from raw, so a window of any shape uses the rollup for everything it can answer. Measured on 12 real high-volume tenants x 5 billing-shaped windows: 60/60 totals identical to raw, 3.71B rows read down to 221M (16.8x). Split also fixes a pre-existing correctness bug it inherited from the daily route: an inclusive midnight upper bound covers the midnight instant, which a day-granularity row cannot express, so translating `<= midnight` to `< midnight` silently dropped rows timestamped exactly at the bound. Those windows now route split, which reads that instant from raw. A non-window time filter (notBetween and friends) carves a shape no day row can honor and now pins the read to raw instead of over-counting. 2. Gauge projections could not collapse a series. Their GROUP BY carried raw `time`, making them 1:1 with the base table — measured in production: 964M projection rows against 966M base rows, i.e. column pruning only. The added latest-value slate is keyed (tenant, metric, dims) with no time key, one argMax state per series: 966M samples collapse to 3.28M series (294 samples per series), and force_optimize_projection now accepts the grouped prefetch shape that it refused before. The windowed slate is unchanged — windowed grouped reads still need the time predicate expressible on the projection — and the new slate's merge cost tracks its own row count, ~0.3% of the base table's. Existing installs must MATERIALIZE the new projections once; setup() only attaches them to parts written afterwards. Production already shows why that step matters: the current slate covers 39-43 of 45 active parts, and a partially covered projection is unusable for the query. --- src/Usage/Adapter/ClickHouse.php | 283 +++++++++++++++++- .../Adapter/ClickHouseGaugeProjectionTest.php | 135 +++++++++ tests/Usage/Adapter/ClickHouseRoutingTest.php | 200 +++++++++++-- tests/Usage/Adapter/ClickHouseSchemaTest.php | 18 +- .../Adapter/ClickHouseSplitParityTest.php | 188 ++++++++++++ 5 files changed, 787 insertions(+), 37 deletions(-) create mode 100644 tests/Usage/Adapter/ClickHouseGaugeProjectionTest.php create mode 100644 tests/Usage/Adapter/ClickHouseSplitParityTest.php diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index 091e9e2..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); @@ -3140,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); } @@ -3148,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 { @@ -3160,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(); @@ -3221,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; } } } @@ -3238,6 +3299,8 @@ private function extractRoutingPlan(array $queries): array 'interval' => $interval, 'orderColumns' => $orderColumns, 'hasCursor' => $hasCursor, + 'hasIrregularTimeFilter' => $hasIrregularTimeFilter, + 'endInclusive' => $endInclusive, ]; } @@ -3254,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 { @@ -3270,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'; } @@ -3297,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. */ @@ -3621,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 { @@ -3660,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 { @@ -3757,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. * @@ -4435,6 +4687,11 @@ private function routedTotalBatch(string $tenant, array $metrics, array $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); } @@ -4520,7 +4777,7 @@ private function totalBatchHybrid(string $tenant, array $metrics, array $queries * * @param array $metrics * @param array $queries - * @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 * @param array $rolledTotals */ private function maybeDualReadBatch(string $tenant, array $metrics, array $queries, string $route, array $plan, array $rolledTotals): void 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 e3c3f4e..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']); } @@ -411,7 +421,7 @@ public function testClosedDayWindowRoutesTotalBatchToDaily(): void $log = $this->adapter->getRouteLog(); $this->assertCount(1, $log); $this->assertSame('getTotalBatch', $log[0]['operation']); - $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, $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'); } @@ -457,6 +467,158 @@ public function testNonRollupFilterKeepsTotalBatchOnRaw(): void $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; + } +}