From c11bdbc89b51e7f8c1750ba21048f1f3cf62cc86 Mon Sep 17 00:00:00 2001 From: Damodar Lohani Date: Mon, 31 Aug 2026 08:48:42 +0545 Subject: [PATCH] perf(clickhouse): split mid-hour windows so grouped event reads route MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The event projections added in #34 key on toStartOfHour(time), and the rewrite onto that key only applies when both window bounds land exactly on an hour. Console windows never do, so the rewrite always declined and the projections were never used. Split a mid-hour window into an hour-aligned interior, which is expressible on the projection key, plus the one or two partial hours either side, which stay on raw `time`. The branches partition the window, so an outer SUM over their UNION ALL is exact; ORDER BY / LIMIT move to that outer query. A window with no whole hour in it, a sub-hour interval, `aggregate('max')`, gauges, and any shape no projection covers all keep their single query. Expand the slate to the dimensions the console breaks down on: method, status, clientType, clientName, deviceName, osName, sdk. Each is its own single-dim projection — a shared (method, status) one keys the cross product, 19.2M rows on a 100M-row fixture against 1.15M + 2.55M apart. --- src/Usage/Adapter/ClickHouse.php | 415 +++++++++-- .../Adapter/ClickHouseDimRoutingTest.php | 25 +- .../Adapter/ClickHouseWindowSplitTest.php | 702 ++++++++++++++++++ 3 files changed, 1065 insertions(+), 77 deletions(-) create mode 100644 tests/Usage/Adapter/ClickHouseWindowSplitTest.php diff --git a/src/Usage/Adapter/ClickHouse.php b/src/Usage/Adapter/ClickHouse.php index 43558e3..4500ac9 100644 --- a/src/Usage/Adapter/ClickHouse.php +++ b/src/Usage/Adapter/ClickHouse.php @@ -828,12 +828,152 @@ private function applyFilters(ClickHouseBuilder $builder, ?string $tenant, array /** * Re-express `time` filters on the hourly bucket so a grouped read can route. * Only hour-aligned bounds are exactly expressible; anything else returns null - * and keeps the raw predicate, so no caller's window shifts. + * and leaves the caller to split the window or keep the raw predicate, so no + * caller's window shifts. * * @param array $filters * @return array{filters: array, conditions: array, bindings: array}|null */ private function bucketAlignedFilters(array $filters): ?array + { + $window = $this->timeWindowBounds($filters); + if ($window === null) { + return null; + } + + $conditions = []; + $bindings = []; + foreach ([['bucketFrom', '>=', $window['lower']], ['bucketTo', '<', $window['upper']]] as [$name, $operator, $bound]) { + if ($bound === null) { + continue; + } + if ($bound % self::HOUR_MILLIS !== 0) { + return null; + } + $conditions[] = self::EVENT_TIME_BUCKET . " {$operator} {{$name}:" . $this->getParamType('time') . '}'; + $bindings[$name] = $this->epochMillisToWire($bound); + } + + return ['filters' => $window['filters'], 'conditions' => $conditions, 'bindings' => $bindings]; + } + + /** + * Whether some projection can serve this shape: it has to hold the grouped + * dims and every filtered column, or the optimizer falls back to the base + * table. Splitting such a read would buy three base-table branches instead + * of one, so the caller keeps its single query. + * + * @param array{filters: array, groupBy?: array} $parsed + */ + private function eventProjectionCovers(array $parsed): bool + { + $dims = array_values($parsed['groupBy'] ?? []); + + $filtered = []; + foreach ($parsed['filters'] as $filter) { + $filtered[] = $filter->getAttribute(); + } + + foreach (self::EVENT_PROJECTIONS as $projection) { + if ($dims !== [] && $dims !== $projection['dims']) { + continue; + } + if (array_diff($filtered, array_merge(['tenant', 'metric', 'time'], $projection['dims'])) === []) { + return true; + } + } + + return false; + } + + /** + * Split a window whose edges fall mid-hour into an hour-aligned interior — + * the only part expressible on the projection's key — plus the one or two + * partial hours either side, which stay on raw `time`. Null when there is no + * whole hour to gain, leaving the caller on its single unsplit query. + * + * @param array $filters + * @return array{filters: array, branches: array, bindings: array}>}|null + */ + private function splitEventWindow(array $filters): ?array + { + $window = $this->timeWindowBounds($filters); + if ($window === null) { + return null; + } + + $lower = $window['lower']; + $upper = $window['upper']; + + // intdiv truncates toward zero, so a pre-epoch bound would round the wrong way. + if (($lower ?? 0) < 0 || ($upper ?? 0) < 0) { + return null; + } + + $interiorFrom = $lower === null ? null : \intdiv($lower + self::HOUR_MILLIS - 1, self::HOUR_MILLIS) * self::HOUR_MILLIS; + $interiorTo = $upper === null ? null : \intdiv($upper, self::HOUR_MILLIS) * self::HOUR_MILLIS; + + // No whole hour between the edges — a single partial hour is cheaper unsplit. + if ($interiorFrom !== null && $interiorTo !== null && $interiorFrom >= $interiorTo) { + return null; + } + + $edges = []; + if ($lower !== null && $lower !== $interiorFrom) { + $edges[] = ['head', $lower, $interiorFrom]; + } + if ($upper !== null && $upper !== $interiorTo) { + $edges[] = ['tail', $interiorTo, $upper]; + } + if ($edges === []) { + return null; + } + + $branches = [$this->windowBranch(self::EVENT_TIME_BUCKET, 'bucket', $interiorFrom, $interiorTo)]; + foreach ($edges as [$name, $from, $to]) { + $branches[] = $this->windowBranch('`time`', $name, $from, $to); + } + + return ['filters' => $window['filters'], 'branches' => $branches]; + } + + /** + * Half-open bound predicates for one branch of a split window. Placeholders + * are named per branch so the merged statement has no collisions. + * + * @return array{timeExpr: string, conditions: array, bindings: array} + */ + private function windowBranch(string $timeExpr, string $name, ?int $from, ?int $to): array + { + $conditions = []; + $bindings = []; + foreach ([[$name . 'From', '>=', $from], [$name . 'To', '<', $to]] as [$param, $operator, $bound]) { + if ($bound === null) { + continue; + } + $conditions[] = "{$timeExpr} {$operator} {{$param}:" . $this->getParamType('time') . '}'; + $bindings[$param] = $this->epochMillisToWire($bound); + } + + return ['timeExpr' => $timeExpr, 'conditions' => $conditions, 'bindings' => $bindings]; + } + + /** Wire format for an epoch-ms instant, keeping the millisecond part an edge bound carries. */ + private function epochMillisToWire(int $millis): string + { + return (new DateTimeImmutable('@' . \intdiv($millis, 1000)))->format('Y-m-d H:i:s') + . '.' . str_pad((string) ($millis % 1000), 3, '0', STR_PAD_LEFT); + } + + /** + * Reduce the `time` filters to one half-open [lower, upper) window in epoch + * ms and hand back the remaining filters. Null for anything unreadable — + * dropping a bound would silently widen the window. + * + * @param array $filters + * @return array{filters: array, lower: ?int, upper: ?int}|null + */ + private function timeWindowBounds(array $filters): ?array { $kept = []; $lower = null; @@ -886,20 +1026,7 @@ private function bucketAlignedFilters(array $filters): ?array $upper = $to === null ? $upper : ($upper === null ? $to : min($upper, $to)); } - $conditions = []; - $bindings = []; - foreach ([['bucketFrom', '>=', $lower], ['bucketTo', '<', $upper]] as [$name, $operator, $bound]) { - if ($bound === null) { - continue; - } - if ($bound % 3_600_000 !== 0) { - return null; - } - $conditions[] = self::EVENT_TIME_BUCKET . " {$operator} {{$name}:" . $this->getParamType('time') . '}'; - $bindings[$name] = (new DateTimeImmutable('@' . \intdiv($bound, 1000)))->format('Y-m-d H:i:s.v'); - } - - return ['filters' => $kept, 'conditions' => $conditions, 'bindings' => $bindings]; + return ['filters' => $kept, 'lower' => $lower, 'upper' => $upper]; } /** @@ -980,6 +1107,9 @@ private function decodeTotal(string $result): int /** Reads must emit this verbatim to route; the optimizer matches on the exact expression. */ private const EVENT_TIME_BUCKET = "toStartOfHour(`time`, 'UTC')"; + /** Width of that bucket, in the epoch milliseconds window bounds are reduced to. */ + private const HOUR_MILLIS = 3_600_000; + /** Sub-hour intervals need detail the bucket summed away. @var list */ private const BUCKET_ROUTABLE_INTERVALS = ['1h', '1d', '1w', '1M']; @@ -988,8 +1118,16 @@ private function decodeTotal(string $result): int * the base events table. The ClickHouse optimizer transparently routes * grouped reads whose GROUP BY shape matches. * - * The (method, status) cross-product is bounded (~180 keys) so the two - * dims share a single projection; everything else is single-dim. + * Single-dim throughout: the console asks for these as separate breakdowns, + * and a shared (method, status) projection keys the cross-product — 19.2M + * rows on a 100M-row fixture against 1.15M + 2.55M for the two alone. + * ip, hostname, city and resourceId are left off as too high-cardinality. + * + * `path` is the outlier already here: it holds resource ids rather than + * route templates, so its key count tracks a customer's resource count and + * can approach the base table (48% of the rows at 30k paths per tenant), + * which buys the split little. Measure before assuming it still earns its + * ingest cost. * * @var array}> */ @@ -997,6 +1135,13 @@ private function decodeTotal(string $result): int ['name' => 'p_by_path', 'dims' => ['path']], ['name' => 'p_by_country', 'dims' => ['country']], ['name' => 'p_by_service', 'dims' => ['service']], + ['name' => 'p_by_method', 'dims' => ['method']], + ['name' => 'p_by_status', 'dims' => ['status']], + ['name' => 'p_by_clientType', 'dims' => ['clientType']], + ['name' => 'p_by_clientName', 'dims' => ['clientName']], + ['name' => 'p_by_deviceName', 'dims' => ['deviceName']], + ['name' => 'p_by_osName', 'dims' => ['osName']], + ['name' => 'p_by_sdk', 'dims' => ['sdk']], ]; /** @@ -2496,89 +2641,221 @@ private function findAggregatedFromTable(?string $tenant, array $parsed, string $valueExpr = 'max(`value`) AS `value`'; } - $builder = $this->newBuilder($type) - ->from($tableName) - ->select(['metric']) - ->selectRaw($valueExpr); - - $groupParts = ['`metric`']; - // Only events carry an hourly key, and only hour-or-coarser derives from it. $bucketed = null; + $split = null; if ( $type === Usage::TYPE_EVENT && (!$hasInterval || in_array($parsed['groupByInterval'], self::BUCKET_ROUTABLE_INTERVALS, true)) ) { $bucketed = $this->bucketAlignedFilters($parsed['filters']); + // Splitting is only worth three branches when one of them can route: + // the projections store sum(value), so `max` has nothing to reach. + if ($bucketed === null && !isset($parsed['aggregate']) && $this->eventProjectionCovers($parsed)) { + $split = $this->splitEventWindow($parsed['filters']); + } + } + + if ($split !== null) { + $result = $this->querySplitAggregate($tenant, $parsed, $split, $tableName, $type, $valueExpr); + + return $this->parseAggregatedResults($result, $type); + } + + $builder = $this->aggregateBranchBuilder( + $tenant, + $parsed, + $bucketed === null ? $parsed['filters'] : $bucketed['filters'], + $tableName, + $type, + $valueExpr, + $bucketed === null ? '`time`' : self::EVENT_TIME_BUCKET, + $bucketed === null ? [] : $bucketed['conditions'], + ); + + foreach ($this->aggregateOrderClauses($parsed, $hasInterval) as $clause) { + $builder->orderByRaw($clause); + } + + if (isset($parsed['limit'])) { + $builder->limit($parsed['limit']); + } + if (isset($parsed['offset'])) { + $builder->offset($parsed['offset']); } - $timeExpr = $bucketed === null ? '`time`' : self::EVENT_TIME_BUCKET; + + $statement = $builder->build(); + $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; + + $result = $this->query($sql, array_merge($statement->namedBindings ?? [], $bucketed['bindings'] ?? [])); + + return $this->parseAggregatedResults($result, $type); + } + + /** + * One grouped branch of an aggregated read. Branches differ only in the time + * expression they bucket on and the bounds they carry, so the SELECT and + * GROUP BY shape — which decides whether a projection can serve them — is + * built in one place. + * + * @param array{groupByInterval?: string, groupBy?: array} $parsed + * @param array $filters + * @param array $conditions + * @throws Exception + */ + private function aggregateBranchBuilder( + ?string $tenant, + array $parsed, + array $filters, + string $tableName, + string $type, + string $valueExpr, + string $timeExpr, + array $conditions, + ): ClickHouseBuilder { + $builder = $this->newBuilder($type) + ->from($tableName) + ->select(['metric']) + ->selectRaw($valueExpr); // Bucket column is only emitted when time bucketing is requested. // Without it the result is a flat aggregate per (metric, …dims). - if ($hasInterval) { + if (isset($parsed['groupByInterval'])) { $intervalSql = UsageQuery::VALID_INTERVALS[$parsed['groupByInterval']]; $builder->selectRaw("toStartOfInterval({$timeExpr}, {$intervalSql}) AS `bucket`"); - $groupParts[] = '`bucket`'; } foreach ($parsed['groupBy'] ?? [] as $dim) { - $escapedDim = $this->escapeIdentifier($dim); - $builder->selectRaw($escapedDim); - $groupParts[] = $escapedDim; + $builder->selectRaw($this->escapeIdentifier($dim)); } - $bucketBindings = []; - if ($bucketed !== null) { - $parsed['filters'] = $bucketed['filters']; - $bucketBindings = $bucketed['bindings']; - foreach ($bucketed['conditions'] as $condition) { - $builder->whereRaw($condition); - } + foreach ($conditions as $condition) { + $builder->whereRaw($condition); } - $this->applyFilters($builder, $tenant, $parsed); + $this->applyFilters($builder, $tenant, ['filters' => $filters]); - $builder->groupByRaw(implode(', ', $groupParts)); + $builder->groupByRaw(implode(', ', $this->aggregateGroupParts($parsed))); - // Default ORDER BY: - // - With time bucketing: bucket ASC (chronological time series). - // - Dim-only: value DESC (top-N table semantics). - // For caller-supplied ORDER BY, `time` is rewritten to `bucket` - // only when bucket is present; otherwise sorting by time is - // invalid (the column is no longer in the SELECT after GROUP BY). - $orderAttributes = $parsed['orderAttributes']; - if (!empty($orderAttributes)) { - foreach ($orderAttributes as $entry) { - $attribute = $entry['attribute']; - if ($attribute === 'time') { - if (!$hasInterval) { - throw new Exception( - 'orderBy("time") requires groupByInterval — without time bucketing the result has no time column' - ); - } - $attribute = 'bucket'; + return $builder; + } + + /** + * @param array{groupByInterval?: string, groupBy?: array} $parsed + * @return array + */ + private function aggregateGroupParts(array $parsed): array + { + $parts = ['`metric`']; + + if (isset($parsed['groupByInterval'])) { + $parts[] = '`bucket`'; + } + + foreach ($parsed['groupBy'] ?? [] as $dim) { + $parts[] = $this->escapeIdentifier($dim); + } + + return $parts; + } + + /** + * Default ORDER BY: + * - With time bucketing: bucket ASC (chronological time series). + * - Dim-only: value DESC (top-N table semantics). + * For caller-supplied ORDER BY, `time` is rewritten to `bucket` + * only when bucket is present; otherwise sorting by time is + * invalid (the column is no longer in the SELECT after GROUP BY). + * + * @param array{orderAttributes: array} $parsed + * @return array + * @throws Exception + */ + private function aggregateOrderClauses(array $parsed, bool $hasInterval): array + { + if (empty($parsed['orderAttributes'])) { + return [$hasInterval ? '`bucket` ASC' : '`value` DESC']; + } + + $clauses = []; + foreach ($parsed['orderAttributes'] as $entry) { + $attribute = $entry['attribute']; + if ($attribute === 'time') { + if (!$hasInterval) { + throw new Exception( + 'orderBy("time") requires groupByInterval — without time bucketing the result has no time column' + ); } - $builder->orderByRaw($this->escapeIdentifier($attribute) . ' ' . $entry['direction']); + $attribute = 'bucket'; } - } elseif ($hasInterval) { - $builder->orderByRaw('`bucket` ASC'); - } else { - $builder->orderByRaw('`value` DESC'); + $clauses[] = $this->escapeIdentifier($attribute) . ' ' . $entry['direction']; } + return $clauses; + } + + /** + * Read a mid-hour window as an hour-aligned interior — which routes to a + * projection — UNION ALL its partial edge hours off the base table, summed + * back together outside. The branches partition the window, so re-summing + * outside is exact; ORDER BY / LIMIT move out with it because no single + * branch holds a whole group. + * + * @param array{groupByInterval?: string, groupBy?: array, orderAttributes: array, limit?: int, offset?: int} $parsed + * @param array{filters: array, branches: array, bindings: array}>} $split + * @throws Exception + */ + private function querySplitAggregate( + ?string $tenant, + array $parsed, + array $split, + string $tableName, + string $type, + string $valueExpr, + ): string { + $branches = []; + $bindings = []; + + foreach ($split['branches'] as $index => $branch) { + $statement = $this->aggregateBranchBuilder( + $tenant, + $parsed, + $split['filters'], + $tableName, + $type, + $valueExpr, + $branch['timeExpr'], + $branch['conditions'], + )->build(); + + // Each branch numbers its own bindings from param0, so they are + // renamed apart before the branches become one statement. + [$branchSql, $renamed] = $this->prefixNamedBindings( + $this->qualifyDdl($statement->query, $tableName), + $statement->namedBindings ?? [], + 'b' . $index . '_', + ); + + $branches[] = $branchSql; + $bindings = array_merge($bindings, $renamed, $branch['bindings']); + } + + $groupParts = $this->aggregateGroupParts($parsed); + $selectParts = array_merge([$groupParts[0], $valueExpr], array_slice($groupParts, 1)); + + $sql = 'SELECT ' . implode(', ', $selectParts) + . ' FROM (' . implode(' UNION ALL ', $branches) . ')' + . ' GROUP BY ' . implode(', ', $groupParts) + . ' ORDER BY ' . implode(', ', $this->aggregateOrderClauses($parsed, isset($parsed['groupByInterval']))); + if (isset($parsed['limit'])) { - $builder->limit($parsed['limit']); + $sql .= ' LIMIT ' . (int) $parsed['limit']; } if (isset($parsed['offset'])) { - $builder->offset($parsed['offset']); + $sql .= ' OFFSET ' . (int) $parsed['offset']; } - $statement = $builder->build(); - $sql = $this->qualifyDdl($statement->query, $tableName) . ' FORMAT JSON'; - - $result = $this->query($sql, array_merge($statement->namedBindings ?? [], $bucketBindings)); - - return $this->parseAggregatedResults($result, $type); + return $this->query($sql . ' FORMAT JSON', $bindings); } /** diff --git a/tests/Usage/Adapter/ClickHouseDimRoutingTest.php b/tests/Usage/Adapter/ClickHouseDimRoutingTest.php index efff584..48296a7 100644 --- a/tests/Usage/Adapter/ClickHouseDimRoutingTest.php +++ b/tests/Usage/Adapter/ClickHouseDimRoutingTest.php @@ -12,8 +12,8 @@ use Utopia\Usage\UsageQuery; /** - * Routing tests for the per-dim projection slate: p_by_path, p_by_country, - * p_by_service. Each grouped scenario asserts: + * Routing tests for the per-dim projection slate. Each grouped scenario + * asserts: * (a) the totals match the raw scan, and * (b) the ClickHouse optimizer picked the matching projection per * `system.query_log.projections`. @@ -119,6 +119,8 @@ public static function topNProjectionProvider(): array 'by_path' => [['path'], 'p_by_path'], 'by_country' => [['country'], 'p_by_country'], 'by_service' => [['service'], 'p_by_service'], + 'by_method' => [['method'], 'p_by_method'], + 'by_status' => [['status'], 'p_by_status'], ]; } @@ -282,10 +284,11 @@ public function testSubHourIntervalReadsTheBaseTable(): void $this->assertNoProjectionUsed($queryId); } - public function testMidHourWindowKeepsRawPredicateAndTotals(): void + public function testMidHourWindowSplitsSoTheInteriorRoutes(): void { - // The window edge cannot be expressed on the bucket, so the rewrite is - // declined: slower, but the caller's boundary is not moved. + // Neither edge is expressible on the bucket, so the whole hours between + // them are read from the projection and the two partial hours from the + // base table — without moving the caller's boundary. $start = (new DateTime('-7 days', new DateTimeZone('UTC')))->setTime(13, 37, 21)->format('Y-m-d H:i:s'); $end = (new DateTime('-2 days', new DateTimeZone('UTC')))->setTime(9, 14, 3)->format('Y-m-d H:i:s'); @@ -300,7 +303,7 @@ public function testMidHourWindowKeepsRawPredicateAndTotals(): void ], Usage::TYPE_EVENT); $this->assertSame($this->rawTotal($start, $end), $this->totalOf($rolled)); - $this->assertNoProjectionUsed($queryId); + $this->assertProjectionUsed($queryId, 'p_by_path'); } public function testDayIntervalBucketsMatchAnUnbucketedDayRollup(): void @@ -325,7 +328,7 @@ public function testDayIntervalBucketsMatchAnUnbucketedDayRollup(): void Query::lessThanEqual('time', $end), ], Usage::TYPE_EVENT); - $this->assertProjectionUsed($queryId, 'p_by_path'); + $this->assertAnyProjectionUsed($queryId); $this->assertNotEmpty($rolled); $this->assertSame($this->rawDayBuckets($start, $end), $this->bucketsOf($rolled)); } @@ -343,7 +346,7 @@ public function testTimeSeriesDayIntervalRoutesAndMatchesRawScan(): void $this->adapter->setNextQueryId($queryId); $series = $this->usage->getTimeSeries('1', [$this->metric], '1d', $start, $end, [], false, Usage::TYPE_EVENT); - $this->assertProjectionUsed($queryId, 'p_by_path'); + $this->assertAnyProjectionUsed($queryId); $points = []; foreach ($series[$this->metric]['data'] as $point) { @@ -513,6 +516,12 @@ private function assertProjectionUsed(string $queryId, string $projectionName): ); } + /** A read with no dimension can be served by any projection in the slate. */ + private function assertAnyProjectionUsed(string $queryId): void + { + $this->assertNotEmpty($this->projectionsForQueryId($queryId), "expected a projection to fire for query_id {$queryId}"); + } + private function assertNoProjectionUsed(string $queryId): void { $projections = $this->projectionsForQueryId($queryId); diff --git a/tests/Usage/Adapter/ClickHouseWindowSplitTest.php b/tests/Usage/Adapter/ClickHouseWindowSplitTest.php new file mode 100644 index 0000000..fafaacd --- /dev/null +++ b/tests/Usage/Adapter/ClickHouseWindowSplitTest.php @@ -0,0 +1,702 @@ +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_window_split', + database: getenv('CLICKHOUSE_DATABASE') ?: 'default', + sharedTables: true, + ); + $this->usage = new Usage($this->adapter); + $this->usage->setup(); + $this->usage->purge('1'); + + $origin = new DateTime('-9 days', new DateTimeZone('UTC')); + $this->origin = $origin->setTime((int) $origin->format('H'), 0, 0); + + // Four readings an hour across three days, each on a different second and + // millisecond so an edge bound that lands mid-hour has rows either side. + $rows = []; + for ($hour = -2; $hour < 74; $hour++) { + for ($slot = 0; $slot < 4; $slot++) { + $rows[] = $this->seedRow($hour, $slot); + } + } + $this->insertRows($rows); + } + + protected function tearDown(): void + { + $this->usage->purge('1'); + } + + private function instant(int $hour, int $minute = 0, int $second = 0, int $milli = 0): string + { + $time = (clone $this->origin) + ->modify("{$hour} hours") + ->modify("{$minute} minutes") + ->modify("{$second} seconds"); + + return $time->format('Y-m-d H:i:s.') . str_pad((string) $milli, 3, '0', STR_PAD_LEFT); + } + + /** + * @return array + */ + private function seedRow(int $hour, int $slot): array + { + // The first seeded hour is negative; shift it non-negative before any + // modulo, which in PHP keeps the sign of the dividend. + $n = $hour + 2; + + return [ + 'time' => $this->instant($hour, $slot * 17 + 3, $slot * 13 + 1, $slot * 111 + 7), + 'value' => 1 + ($n * 7 + $slot * 3) % 23, + 'path' => '/v1/p' . ($n % 3), + 'country' => 'C' . ($slot % 2), + 'service' => ['storage', 'databases', 'functions'][($n + $slot) % 3], + 'method' => ['GET', 'POST', 'DELETE'][$slot % 3], + 'status' => ['200', '201', '404', '500'][($n + $slot) % 4], + 'clientType' => ['browser', 'server'][$n % 2], + 'clientName' => 'client' . (($n + $slot) % 4), + 'deviceName' => ['desktop', 'smartphone'][$slot % 2], + 'osName' => ['Linux', 'Mac', 'Windows'][($n + 1) % 3], + 'sdk' => ['web', 'flutter', 'node'][($slot + 1) % 3], + ]; + } + + /** + * @param array> $rows + */ + private function insertRows(array $rows): void + { + $table = $this->resolveTableName($this->adapter, 'getEventsTableName'); + $database = $this->databaseName($this->adapter); + + $columns = ['path', 'country', 'service', 'method', 'status', 'clientType', 'clientName', 'deviceName', 'osName', 'sdk']; + + $values = []; + foreach ($rows as $row) { + $cells = [ + "'" . bin2hex(random_bytes(16)) . "'", + "'" . addslashes($this->metric) . "'", + (string) $row['value'], + "'" . $row['time'] . "'", + "'1'", + ]; + foreach ($columns as $column) { + $cells[] = "'" . addslashes((string) $row[$column]) . "'"; + } + $values[] = '(' . implode(', ', $cells) . ')'; + } + + $sql = "INSERT INTO `{$database}`.`{$table}` (id, metric, value, time, tenant, " . implode(', ', $columns) . ') VALUES ' + . implode(', ', $values); + + $this->queryRaw($this->adapter, $sql); + } + + /** + * The same grouped read taken straight off the base table with projections + * disabled — the reference a split window has to reproduce exactly. + * + * @param array $dims + * @param array $extraWhere + * @return array + */ + private function rawGrouped( + string $start, + string $end, + array $dims = [], + ?string $interval = null, + string $aggregate = 'sum', + array $extraWhere = [], + ): array { + $database = $this->databaseName($this->adapter); + $table = $this->resolveTableName($this->adapter, 'getEventsTableName'); + + $select = []; + if ($interval !== null) { + $select[] = 'toStartOfInterval(`time`, ' . UsageQuery::VALID_INTERVALS[$interval] . ') AS `bucket`'; + } + foreach ($dims as $dim) { + $select[] = "`{$dim}`"; + } + + $keys = $select === [] ? [] : $select; + $where = array_merge([ + "`tenant` = '1'", + "`metric` = '" . addslashes($this->metric) . "'", + "`time` >= '{$start}'", + "`time` <= '{$end}'", + ], $extraWhere); + + $sql = 'SELECT ' . implode(', ', array_merge($keys, ["{$aggregate}(`value`) AS `value`"])) + . " FROM `{$database}`.`{$table}`" + . ' WHERE ' . implode(' AND ', $where) + . ($keys === [] ? '' : ' GROUP BY ' . implode(', ', $keys)) + . ' SETTINGS optimize_use_projections = 0 FORMAT JSONEachRow'; + + $out = []; + foreach (explode("\n", trim($this->queryRaw($this->adapter, $sql))) as $line) { + if ($line === '') { + continue; + } + $row = json_decode($line, true); + if (!is_array($row) || !is_numeric($row['value'] ?? null)) { + continue; + } + $out[$this->keyOf($row, $dims, $interval !== null)] = (int) $row['value']; + } + ksort($out); + + return $out; + } + + /** + * @param array<\Utopia\Usage\Metric> $metrics + * @param array $dims + * @return array + */ + private function normalize(array $metrics, array $dims = [], bool $hasBucket = false): array + { + $out = []; + foreach ($metrics as $metric) { + $row = $metric->getArrayCopy(); + $out[$this->keyOf($row, $dims, $hasBucket)] = (int) $metric->getValue(0); + } + ksort($out); + + return $out; + } + + /** + * @param array $row + * @param array $dims + */ + private function keyOf(array $row, array $dims, bool $hasBucket): string + { + $parts = []; + if ($hasBucket) { + // find() renames the bucket to `time` and stamps the zone, while the raw + // scan returns it bare — and coarser than a day it is a Date, not a + // DateTime. Strip both back to one spelling so the keys line up. + $bucket = $row['bucket'] ?? $row['time'] ?? ''; + $parts[] = is_string($bucket) ? str_replace(['T', '+00:00'], [' ', ''], $bucket) : ''; + } + foreach ($dims as $dim) { + $value = $row[$dim] ?? null; + $parts[] = is_scalar($value) ? (string) $value : ''; + } + + return implode('|', $parts); + } + + /** + * @return array, 1: ?string, 2: ?string}> + */ + public static function splitShapeProvider(): array + { + return [ + 'breakdown_path' => [['path'], null, 'p_by_path'], + 'breakdown_country' => [['country'], null, 'p_by_country'], + 'breakdown_method' => [['method'], null, 'p_by_method'], + 'breakdown_status' => [['status'], null, 'p_by_status'], + 'breakdown_clientType' => [['clientType'], null, 'p_by_clientType'], + 'breakdown_clientName' => [['clientName'], null, 'p_by_clientName'], + 'breakdown_deviceName' => [['deviceName'], null, 'p_by_deviceName'], + 'breakdown_osName' => [['osName'], null, 'p_by_osName'], + 'breakdown_sdk' => [['sdk'], null, 'p_by_sdk'], + 'breakdown_service' => [['service'], null, 'p_by_service'], + // A dimensionless chart can be served by any projection in the slate, + // so it only asserts that one of them fired. + 'chart_1h' => [[], '1h', null], + 'chart_1d' => [[], '1d', null], + 'chart_1w' => [[], '1w', null], + 'chart_1M' => [[], '1M', null], + 'chart_1d_by_path' => [['path'], '1d', 'p_by_path'], + 'chart_1h_by_country' => [['country'], '1h', 'p_by_country'], + ]; + } + + /** + * @dataProvider splitShapeProvider + * @param array $dims + */ + public function testSplitWindowMatchesRawScanAndRoutes(array $dims, ?string $interval, ?string $projection): void + { + $start = $this->instant(0, 37, 21, 456); + $end = $this->instant(71, 14, 3, 789); + + $queries = []; + foreach ($dims as $dim) { + $queries[] = UsageQuery::groupBy($dim); + } + if ($interval !== null) { + $queries[] = UsageQuery::groupByInterval('time', $interval); + } + $queries[] = Query::equal('metric', [$this->metric]); + $queries[] = Query::greaterThanEqual('time', $start); + $queries[] = Query::lessThanEqual('time', $end); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', $queries, Usage::TYPE_EVENT); + + $this->assertNotEmpty($rolled); + $this->assertSame( + $this->rawGrouped($start, $end, $dims, $interval), + $this->normalize($rolled, $dims, $interval !== null), + ); + if ($projection === null) { + $this->assertNotEmpty($this->projectionsForQueryId($queryId), 'expected the interior to route'); + } else { + $this->assertProjectionUsed($queryId, $projection); + } + } + + public function testTopNLimitAppliesAfterTheBranchesAreRecombined(): void + { + // A LIMIT inside a branch would truncate before the edge hours were added + // back, so it belongs to the outer query. + $start = $this->instant(0, 37, 21, 456); + $end = $this->instant(71, 14, 3, 789); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupBy('clientName'), + Query::equal('metric', [$this->metric]), + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + Query::limit(2), + ], Usage::TYPE_EVENT); + + $expected = $this->rawGrouped($start, $end, ['clientName']); + arsort($expected); + $expected = array_slice($expected, 0, 2, true); + ksort($expected); + + $this->assertCount(2, $rolled); + $this->assertSame($expected, $this->normalize($rolled, ['clientName'])); + $this->assertProjectionUsed($queryId, 'p_by_clientName'); + } + + /** + * @return array + */ + public static function oneSidedEdgeProvider(): array + { + return ['upper_edge_only' => ['aligned', 'mid'], 'lower_edge_only' => ['mid', 'aligned']]; + } + + /** + * @dataProvider oneSidedEdgeProvider + */ + public function testOneSidedEdgeStillRoutes(string $lower, string $upper): void + { + $start = $lower === 'aligned' ? $this->instant(1) : $this->instant(0, 37, 21, 456); + // An inclusive bound one tick below the boundary is the aligned spelling. + $end = $upper === 'aligned' ? $this->instant(70, 59, 59, 999) : $this->instant(71, 14, 3, 789); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupBy('path'), + Query::equal('metric', [$this->metric]), + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + ], Usage::TYPE_EVENT); + + $this->assertSame($this->rawGrouped($start, $end, ['path']), $this->normalize($rolled, ['path'])); + $this->assertProjectionUsed($queryId, 'p_by_path'); + } + + public function testSplitCarriesEveryBranchesOwnBindings(): void + { + // Three branches each number their placeholders from param0. If the + // renaming mismatched a value, a branch would filter on the wrong metric + // or path and the totals would silently drift. + $this->insertRows([['time' => $this->instant(3, 20), 'value' => 1000, 'path' => '/v1/other'] + $this->seedRow(3, 0)]); + + $start = $this->instant(0, 37, 21, 456); + $end = $this->instant(71, 14, 3, 789); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupBy('path'), + Query::equal('metric', [$this->metric]), + Query::equal('path', ['/v1/p0', '/v1/p2']), + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + ], Usage::TYPE_EVENT); + + $this->assertSame( + $this->rawGrouped($start, $end, ['path'], null, 'sum', ["`path` IN ('/v1/p0', '/v1/p2')"]), + $this->normalize($rolled, ['path']), + ); + $this->assertProjectionUsed($queryId, 'p_by_path'); + } + + public function testWindowInsideOneHourFallsBackToTheUnsplitQuery(): void + { + // Peeling edges off would leave no whole hour to route, so the read keeps + // its single raw-time query. + $start = $this->instant(5, 12, 30); + $end = $this->instant(5, 47, 10, 500); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupBy('path'), + Query::equal('metric', [$this->metric]), + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + ], Usage::TYPE_EVENT); + + $this->assertNotEmpty($rolled); + $this->assertSame($this->rawGrouped($start, $end, ['path']), $this->normalize($rolled, ['path'])); + $this->assertNoProjectionUsed($queryId); + $this->assertStringNotContainsString('UNION ALL', $this->queryTextFor($queryId)); + } + + public function testWindowSpanningOneBoundaryWithoutAWholeHourFallsBack(): void + { + $start = $this->instant(5, 31); + $end = $this->instant(6, 29); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupBy('path'), + Query::equal('metric', [$this->metric]), + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + ], Usage::TYPE_EVENT); + + $this->assertSame($this->rawGrouped($start, $end, ['path']), $this->normalize($rolled, ['path'])); + $this->assertStringNotContainsString('UNION ALL', $this->queryTextFor($queryId)); + } + + /** + * @return array + */ + public static function deltaWindowProvider(): array + { + // The concurrency delta folder reads events on a 5-minute-aligned window + // and writes the result back as a gauge, so its totals must not move + // whether or not the window is wide enough to peel an hour out of. + return [ + 'five_minute_slice' => ['00:00', '04:59.999', null, false], + 'five_minute_bucket' => ['00:00', '04:59.999', '5m', false], + 'multi_hour_slice' => ['00:00', '3:04:59.999', null, true], + ]; + } + + /** + * @dataProvider deltaWindowProvider + */ + public function testFiveMinuteAlignedDeltaWindowKeepsItsResults(string $from, string $to, ?string $interval, bool $splits): void + { + [$fromMinute, $fromSecond] = array_map('intval', explode(':', $from)); + $start = $this->instant(2, $fromMinute, $fromSecond); + $end = $this->offsetInstant(2, $to); + + $queries = [ + UsageQuery::groupBy('service'), + Query::equal('metric', [$this->metric]), + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + ]; + if ($interval !== null) { + $queries[] = UsageQuery::groupByInterval('time', $interval); + } + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', $queries, Usage::TYPE_EVENT); + + $this->assertNotEmpty($rolled); + $this->assertSame( + $this->rawGrouped($start, $end, ['service'], $interval), + $this->normalize($rolled, ['service'], $interval !== null), + ); + $this->assertSame($splits, str_contains($this->queryTextFor($queryId), 'UNION ALL')); + } + + /** Accepts `mm:ss[.mmm]` or `h:mm:ss[.mmm]` relative to the given hour. */ + private function offsetInstant(int $hour, string $offset): string + { + $parts = explode(':', $offset); + $hour += count($parts) === 3 ? (int) array_shift($parts) : 0; + [$second, $milli] = array_pad(explode('.', $parts[1]), 2, '0'); + + return $this->instant($hour, (int) $parts[0], (int) $second, (int) $milli); + } + + public function testSubHourIntervalKeepsReadingTheBaseTable(): void + { + // A 15m bucket needs detail the hourly interior has already summed away. + $start = $this->instant(0, 37, 21, 456); + $end = $this->instant(71, 14, 3, 789); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupByInterval('time', '15m'), + UsageQuery::groupBy('path'), + Query::equal('metric', [$this->metric]), + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + ], Usage::TYPE_EVENT); + + $this->assertSame( + $this->rawGrouped($start, $end, ['path'], '15m'), + $this->normalize($rolled, ['path'], true), + ); + $this->assertNoProjectionUsed($queryId); + $this->assertStringNotContainsString('UNION ALL', $this->queryTextFor($queryId)); + } + + public function testDimsNoProjectionCoversAreNotSplit(): void + { + $start = $this->instant(0, 37, 21, 456); + $end = $this->instant(71, 14, 3, 789); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupBy('path'), + UsageQuery::groupBy('country'), + Query::equal('metric', [$this->metric]), + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + ], Usage::TYPE_EVENT); + + $this->assertSame( + $this->rawGrouped($start, $end, ['path', 'country']), + $this->normalize($rolled, ['path', 'country']), + ); + $this->assertStringNotContainsString('UNION ALL', $this->queryTextFor($queryId)); + } + + public function testFilterOutsideTheProjectionIsNotSplit(): void + { + $start = $this->instant(0, 37, 21, 456); + $end = $this->instant(71, 14, 3, 789); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupBy('path'), + Query::equal('metric', [$this->metric]), + Query::equal('country', ['C0']), + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + ], Usage::TYPE_EVENT); + + $this->assertSame( + $this->rawGrouped($start, $end, ['path'], null, 'sum', ["`country` = 'C0'"]), + $this->normalize($rolled, ['path']), + ); + $this->assertStringNotContainsString('UNION ALL', $this->queryTextFor($queryId)); + } + + public function testAggregateMaxIsNotSplit(): void + { + // The projections store sum(value), so a max() read has nothing to route + // to and would pay for the extra branches. + $start = $this->instant(0, 37, 21, 456); + $end = $this->instant(71, 14, 3, 789); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupBy('path'), + UsageQuery::aggregate('max'), + Query::equal('metric', [$this->metric]), + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + ], Usage::TYPE_EVENT); + + $this->assertSame( + $this->rawGrouped($start, $end, ['path'], null, 'max'), + $this->normalize($rolled, ['path']), + ); + $this->assertStringNotContainsString('UNION ALL', $this->queryTextFor($queryId)); + } + + public function testGaugeReadIsNeverSplit(): void + { + // Billing rolls gauges up through the same grouped path; it has to keep + // reading raw `time`. + $start = $this->instant(0, 37, 21, 456); + $end = $this->instant(71, 14, 3, 789); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $this->usage->find('1', [ + UsageQuery::groupBy('resourceId'), + Query::equal('metric', [$this->metric]), + Query::greaterThanEqual('time', $start), + Query::lessThanEqual('time', $end), + ], Usage::TYPE_GAUGE); + + $text = $this->queryTextFor($queryId); + $this->assertStringNotContainsString('UNION ALL', $text); + $this->assertStringNotContainsString('toStartOfHour', $text); + } + + public function testExclusiveBoundsSplitOnTheSameWindow(): void + { + $start = $this->instant(0, 37, 21, 456); + $end = $this->instant(71, 14, 3, 790); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupBy('path'), + Query::equal('metric', [$this->metric]), + Query::greaterThan('time', $this->instant(0, 37, 21, 455)), + Query::lessThan('time', $end), + ], Usage::TYPE_EVENT); + + $this->assertSame( + $this->rawGrouped($start, $this->instant(71, 14, 3, 789), ['path']), + $this->normalize($rolled, ['path']), + ); + $this->assertProjectionUsed($queryId, 'p_by_path'); + } + + public function testBetweenBoundsSplitOnTheSameWindow(): void + { + $start = $this->instant(0, 37, 21, 456); + $end = $this->instant(71, 14, 3, 789); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupBy('path'), + Query::equal('metric', [$this->metric]), + Query::between('time', $start, $end), + ], Usage::TYPE_EVENT); + + $this->assertSame($this->rawGrouped($start, $end, ['path']), $this->normalize($rolled, ['path'])); + $this->assertProjectionUsed($queryId, 'p_by_path'); + } + + public function testOpenEndedWindowSplitsOnTheBoundedSideOnly(): void + { + $start = $this->instant(0, 37, 21, 456); + $end = $this->instant(80); + + $queryId = bin2hex(random_bytes(8)); + $this->adapter->setNextQueryId($queryId); + $rolled = $this->usage->find('1', [ + UsageQuery::groupBy('path'), + Query::equal('metric', [$this->metric]), + Query::greaterThanEqual('time', $start), + ], Usage::TYPE_EVENT); + + $this->assertSame($this->rawGrouped($start, $end, ['path']), $this->normalize($rolled, ['path'])); + $this->assertProjectionUsed($queryId, 'p_by_path'); + } + + private function queryTextFor(string $queryId): string + { + $this->queryRaw($this->adapter, 'SYSTEM FLUSH LOGS'); + + $escaped = addslashes($queryId); + $raw = $this->queryRaw($this->adapter, 'SELECT query FROM system.query_log ' + . "WHERE query_id = '{$escaped}' AND type = 'QueryFinish' " + . 'ORDER BY event_time DESC LIMIT 1 FORMAT JSON'); + + $json = json_decode($raw, true); + $data = (is_array($json) && is_array($json['data'] ?? null)) ? $json['data'] : []; + $row = $data[0] ?? null; + + return (is_array($row) && is_string($row['query'] ?? null)) ? $row['query'] : ''; + } + + private function assertProjectionUsed(string $queryId, string $projectionName): void + { + $projections = $this->projectionsForQueryId($queryId); + $matches = array_filter($projections, fn (string $p): bool => str_ends_with($p, '.' . $projectionName) || $p === $projectionName); + $this->assertNotEmpty( + $matches, + "expected projection {$projectionName} to fire for query_id {$queryId}; saw: " . implode(', ', $projections) + ); + } + + private function assertNoProjectionUsed(string $queryId): void + { + $projections = $this->projectionsForQueryId($queryId); + $this->assertEmpty( + $projections, + "expected no projection to fire for query_id {$queryId}; saw: " . implode(', ', $projections) + ); + } + + /** + * @return array + */ + private function projectionsForQueryId(string $queryId): array + { + $this->queryRaw($this->adapter, 'SYSTEM FLUSH LOGS'); + + $escaped = addslashes($queryId); + $raw = $this->queryRaw($this->adapter, 'SELECT projections FROM system.query_log ' + . "WHERE query_id = '{$escaped}' AND type = 'QueryFinish' " + . 'ORDER BY event_time DESC LIMIT 1 FORMAT JSON'); + + $json = json_decode($raw, true); + if (!is_array($json) || !is_array($json['data'] ?? null) || $json['data'] === []) { + return []; + } + $row = $json['data'][0]; + $projections = is_array($row) ? ($row['projections'] ?? []) : []; + + $out = []; + foreach (is_array($projections) ? $projections : [] as $projection) { + if (is_string($projection)) { + $out[] = $projection; + } + } + + return $out; + } +}