feat: add groupByInterval query support for time-bucketed aggregation

Add UsageQuery class extending Query with a custom groupByInterval method
that enables time-bucketed aggregated queries. When present in the queries
array, the ClickHouse adapter switches from raw row returns to aggregated
results grouped by time bucket (SUM for events, argMax for gauges).

Supported intervals: 1m, 5m, 15m, 1h, 1d, 1w, 1M.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: lohanidamodar

src/Usage/Adapter/ClickHouse.php

@@ -7,6 +7,7 @@
 use Utopia\Fetch\Client;
 use Utopia\Usage\Metric;
 use Utopia\Usage\Usage;
+use Utopia\Usage\UsageQuery;
 use Utopia\Validator\Hostname;
 
 /**
@@ -1434,6 +1435,11 @@ public function find(array $queries = [], ?string $type = null): array
     /**
      * Find metrics from a specific table.
      *
+     * When a `groupByInterval` query is present, switches to aggregated mode:
+     * - Events: SELECT metric, SUM(value) as value, toStartOfInterval(time, INTERVAL ...) as time
+     * - Gauges: SELECT metric, argMax(value, time) as value, toStartOfInterval(time, INTERVAL ...) as time
+     * Results are grouped by metric and time bucket, ordered by time ASC.
+     *
      * @param array<Query> $queries
      * @param string $type 'event' or 'gauge'
      * @return array<Metric>
@@ -1446,6 +1452,11 @@ private function findFromTable(array $queries, string $type): array
 
         $parsed = $this->parseQueries($queries, $type);
 
+        // Check if groupByInterval is requested
+        if (isset($parsed['groupByInterval'])) {
+            return $this->findAggregatedFromTable($parsed, $fromTable, $type);
+        }
+
         $selectColumns = $this->getSelectColumns($type);
 
         $whereData = $this->buildWhereClause($parsed['filters'], $parsed['params']);
@@ -1471,6 +1482,115 @@ private function findFromTable(array $queries, string $type): array
         return $this->parseResults($result, $type);
     }
 
+    /**
+     * Find aggregated metrics from a table using time-bucketed grouping.
+     *
+     * Produces SQL like:
+     *   SELECT metric, SUM(value) as value,
+     *          toStartOfInterval(time, INTERVAL 1 HOUR) as time
+     *   FROM table WHERE ... GROUP BY metric, time ORDER BY time ASC
+     *
+     * @param array<string, mixed> $parsed Parsed query data from parseQueries()
+     * @param string $fromTable Fully qualified table reference
+     * @param string $type 'event' or 'gauge'
+     * @return array<Metric>
+     * @throws Exception
+     */
+    private function findAggregatedFromTable(array $parsed, string $fromTable, string $type): array
+    {
+        /** @var string $interval */
+        $interval = $parsed['groupByInterval'];
+        $intervalSql = UsageQuery::VALID_INTERVALS[$interval];
+
+        // Choose aggregation function based on metric type
+        $valueExpr = $type === Usage::TYPE_GAUGE
+            ? 'argMax(value, time) as value'
+            : 'SUM(value) as value';
+
+        // Use 'bucket' alias to avoid collision with the raw 'time' column,
+        // then alias back to 'time' in outer context for consistent Metric parsing.
+        $timeBucketExpr = "toStartOfInterval(time, {$intervalSql})";
+
+        $whereData = $this->buildWhereClause($parsed['filters'], $parsed['params']);
+        $whereClause = $whereData['clause'];
+        $params = $whereData['params'];
+
+        // Use custom ORDER BY if specified, otherwise default to bucket ASC
+        $orderClause = ' ORDER BY bucket ASC';
+        if (!empty($parsed['orderBy'])) {
+            $orderClause = ' ORDER BY ' . implode(', ', $parsed['orderBy']);
+        }
+
+        $limitClause = isset($parsed['limit']) ? ' LIMIT {limit:UInt64}' : '';
+        $offsetClause = isset($parsed['offset']) ? ' OFFSET {offset:UInt64}' : '';
+
+        $sql = "
+            SELECT metric, {$valueExpr}, {$timeBucketExpr} as bucket
+            FROM {$fromTable}{$whereClause}
+            GROUP BY metric, bucket{$orderClause}{$limitClause}{$offsetClause}
+            FORMAT JSON
+        ";
+
+        $result = $this->query($sql, $params);
+
+        return $this->parseAggregatedResults($result, $type);
+    }
+
+    /**
+     * Parse ClickHouse JSON results from an aggregated (groupByInterval) query into Metric array.
+     *
+     * Maps the 'bucket' column back to 'time' for consistent Metric objects.
+     *
+     * @param string $result Raw JSON response from ClickHouse
+     * @param string $type 'event' or 'gauge'
+     * @return array<Metric>
+     */
+    private function parseAggregatedResults(string $result, string $type = 'event'): array
+    {
+        if (empty(trim($result))) {
+            return [];
+        }
+
+        $json = json_decode($result, true);
+
+        if (!is_array($json) || !isset($json['data']) || !is_array($json['data'])) {
+            return [];
+        }
+
+        $rows = $json['data'];
+        $metrics = [];
+
+        foreach ($rows as $row) {
+            if (!is_array($row)) {
+                continue;
+            }
+
+            $document = [];
+
+            foreach ($row as $key => $value) {
+                if ($key === 'bucket') {
+                    // Map 'bucket' back to 'time' for consistent Metric objects
+                    $parsedTime = (string) $value;
+                    if (strpos($parsedTime, 'T') === false) {
+                        $parsedTime = str_replace(' ', 'T', $parsedTime) . '+00:00';
+                    }
+                    $document['time'] = $parsedTime;
+                } elseif ($key === 'value') {
+                    $document[$key] = $value !== null ? (int) $value : null;
+                } else {
+                    $document[$key] = $value;
+                }
+            }
+
+            // Set the type based on which table we queried
+            $document['type'] = $type;
+
+            $metrics[] = new Metric($document);
+        }
+
+        return $metrics;
+    }
+
     /**
      * Count metrics using Query objects.
      *
@@ -2204,7 +2324,7 @@ private function getParamType(string $attribute): string
      *
      * @param array<Query> $queries
      * @param string $type 'event' or 'gauge' — used for attribute validation
-     * @return array{filters: array<string>, params: array<string, mixed>, orderBy?: array<string>, limit?: int, offset?: int}
+     * @return array{filters: array<string>, params: array<string, mixed>, orderBy?: array<string>, limit?: int, offset?: int, groupByInterval?: string}
      * @throws Exception
      */
     private function parseQueries(array $queries, string $type = 'event'): array
@@ -2214,6 +2334,7 @@ private function parseQueries(array $queries, string $type = 'event'): array
         $orderBy = [];
         $limit = null;
         $offset = null;
+        $groupByInterval = null;
         $paramCounter = 0;
 
         foreach ($queries as $query) {
@@ -2417,6 +2538,18 @@ private function parseQueries(array $queries, string $type = 'event'): array
                     $offset = $offsetVal;
                     $params['offset'] = $offset;
                     break;
+
+                case UsageQuery::TYPE_GROUP_BY_INTERVAL:
+                    $this->validateAttributeName($attribute, $type);
+                    $interval = $values[0] ?? '1h';
+                    if (!is_string($interval) || !isset(UsageQuery::VALID_INTERVALS[$interval])) {
+                        throw new \Exception(
+                            "Invalid groupByInterval interval '{$interval}'. Allowed: "
+                            . implode(', ', array_keys(UsageQuery::VALID_INTERVALS))
+                        );
+                    }
+                    $groupByInterval = $interval;
+                    break;
             }
         }
 
@@ -2437,6 +2570,10 @@ private function parseQueries(array $queries, string $type = 'event'): array
             $result['offset'] = $offset;
         }
 
+        if ($groupByInterval !== null) {
+            $result['groupByInterval'] = $groupByInterval;
+        }
+
         return $result;
     }
 

src/Usage/Adapter/Database.php

@@ -8,6 +8,7 @@
 use Utopia\Database\Query as DatabaseQuery;
 use Utopia\Usage\Metric;
 use Utopia\Usage\Usage;
+use Utopia\Usage\UsageQuery;
 use Utopia\Query\Query;
 
 class Database extends SQL
@@ -392,6 +393,11 @@ private function convertQueriesToDatabase(array $queries): array
                         $dbQueries[] = DatabaseQuery::offset((int) $val);
                     }
                     break;
+
+                case UsageQuery::TYPE_GROUP_BY_INTERVAL:
+                    // groupByInterval is not supported by the Database adapter.
+                    // Silently skip — callers get raw (non-aggregated) results.
+                    break;
             }
         }
 

src/Usage/UsageQuery.php

@@ -0,0 +1,110 @@
+<?php
+
+namespace Utopia\Usage;
+
+use Utopia\Query\Query;
+
+/**
+ * Usage Query
+ *
+ * Extends the base Query class with usage-specific query types.
+ * Currently adds support for `groupByInterval` which enables time-bucketed
+ * aggregated queries in the ClickHouse adapter.
+ *
+ * Example usage:
+ * ```php
+ * $queries = [
+ *     UsageQuery::groupByInterval('time', '1h'),
+ *     Query::equal('metric', ['bandwidth']),
+ *     Query::greaterThanEqual('time', '2026-03-01'),
+ *     Query::lessThanEqual('time', '2026-04-01'),
+ * ];
+ * $results = $usage->find($queries, 'event');
+ * ```
+ *
+ * When `groupByInterval` is present in the queries array, the ClickHouse adapter
+ * switches from raw row returns to aggregated results grouped by time bucket:
+ * - Events: SUM(value) per bucket
+ * - Gauges: argMax(value, time) per bucket
+ */
+class UsageQuery extends Query
+{
+    public const TYPE_GROUP_BY_INTERVAL = 'groupByInterval';
+
+    /**
+     * Valid interval values and their ClickHouse INTERVAL equivalents.
+     */
+    public const VALID_INTERVALS = [
+        '1m' => 'INTERVAL 1 MINUTE',
+        '5m' => 'INTERVAL 5 MINUTE',
+        '15m' => 'INTERVAL 15 MINUTE',
+        '1h' => 'INTERVAL 1 HOUR',
+        '1d' => 'INTERVAL 1 DAY',
+        '1w' => 'INTERVAL 1 WEEK',
+        '1M' => 'INTERVAL 1 MONTH',
+    ];
+
+    /**
+     * Create a groupByInterval query.
+     *
+     * When passed to `find()`, this switches the adapter to return time-bucketed
+     * aggregated results instead of raw rows.
+     *
+     * @param string $attribute The time attribute to bucket (usually 'time')
+     * @param string $interval The bucket size: '1m', '5m', '15m', '1h', '1d', '1w', '1M'
+     * @return self
+     */
+    public static function groupByInterval(string $attribute, string $interval): self
+    {
+        if (!isset(self::VALID_INTERVALS[$interval])) {
+            throw new \InvalidArgumentException(
+                "Invalid interval '{$interval}'. Allowed: " . implode(', ', array_keys(self::VALID_INTERVALS))
+            );
+        }
+
+        return new self(self::TYPE_GROUP_BY_INTERVAL, $attribute, [$interval]);
+    }
+
+    /**
+     * Check if a query is a groupByInterval query.
+     *
+     * @param Query $query
+     * @return bool
+     */
+    public static function isGroupByInterval(Query $query): bool
+    {
+        return $query->getMethod() === self::TYPE_GROUP_BY_INTERVAL;
+    }
+
+    /**
+     * Extract the groupByInterval query from an array of queries, if present.
+     *
+     * @param array<Query> $queries
+     * @return self|null The groupByInterval query, or null if not present
+     */
+    public static function extractGroupByInterval(array $queries): ?self
+    {
+        foreach ($queries as $query) {
+            if ($query instanceof self && $query->getMethod() === self::TYPE_GROUP_BY_INTERVAL) {
+                return $query;
+            }
+        }
+
+        return null;
+    }
+
+    /**
+     * Remove groupByInterval queries from an array of queries.
+     *
+     * Returns the remaining queries that should be processed normally.
+     *
+     * @param array<Query> $queries
+     * @return array<Query>
+     */
+    public static function removeGroupByInterval(array $queries): array
+    {
+        return array_values(array_filter($queries, function (Query $query) {
+            return !self::isGroupByInterval($query);
+        }));
+    }
+}