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; + } +}