Skip to content

Commit ce30423

Browse files
lohanidamodarlohanidamodar
committed
feat(adapter): add per-dim daily rollup MVs (by_path, by_country, by_service, by_method_status)
1 parent 246c79a commit ce30423

3 files changed

Lines changed: 467 additions & 0 deletions

File tree

‎src/Usage/Adapter/ClickHouse.php‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1030,6 +1030,22 @@ private function formatParamValue(mixed $value): string
10301030
return '';
10311031
}
10321032

1033+
/**
1034+
* Per-dim rollup MV slate. Each entry creates a SummingMergeTree table
1035+
* + materialized view that fans out off the raw events INSERT.
1036+
*
1037+
* The (method, status) cross-product is bounded (~180 keys) so the two
1038+
* dims share a single MV; everything else is single-dim.
1039+
*
1040+
* @var array<array{name: string, dims: array<int, string>}>
1041+
*/
1042+
private const DIM_ROLLUPS = [
1043+
['name' => 'by_path', 'dims' => ['path']],
1044+
['name' => 'by_country', 'dims' => ['country']],
1045+
['name' => 'by_service', 'dims' => ['service']],
1046+
['name' => 'by_method_status', 'dims' => ['method', 'status']],
1047+
];
1048+
10331049
/**
10341050
* Setup ClickHouse table structure.
10351051
*
@@ -1038,6 +1054,7 @@ private function formatParamValue(mixed $value): string
10381054
* 2. Events daily table (SummingMergeTree) for pre-aggregation
10391055
* 3. Events daily materialized view
10401056
* 4. Gauges table (MergeTree) with simple schema
1057+
* 5. Per-dim rollup tables + MVs (by_path, by_country, by_service, by_method_status)
10411058
*
10421059
* @throws Exception
10431060
*/
@@ -1063,6 +1080,12 @@ public function setup(): void
10631080
// --- Events daily materialized view ---
10641081
$this->createDailyMaterializedView();
10651082

1083+
// --- Per-dim rollup tables + MVs ---
1084+
foreach (self::DIM_ROLLUPS as $rollup) {
1085+
$this->createDimRollupTable($rollup['name'], $rollup['dims']);
1086+
$this->createDimRollupMaterializedView($rollup['name'], $rollup['dims']);
1087+
}
1088+
10661089
// --- Gauges table ---
10671090
$this->createTable(
10681091
$this->getGaugesTableName(),
@@ -1071,6 +1094,107 @@ public function setup(): void
10711094
);
10721095
}
10731096

1097+
/**
1098+
* @return string e.g. utopia_usage_events_daily_by_path
1099+
*/
1100+
private function getDimRollupTableName(string $name): string
1101+
{
1102+
return $this->getTableName() . '_events_daily_' . $name;
1103+
}
1104+
1105+
private function getDimRollupMvName(string $name): string
1106+
{
1107+
return $this->getDimRollupTableName($name) . '_mv';
1108+
}
1109+
1110+
/**
1111+
* @param array<int, string> $dims
1112+
*/
1113+
private function createDimRollupTable(string $name, array $dims): void
1114+
{
1115+
$tableName = $this->getDimRollupTableName($name);
1116+
$escapedTable = $this->escapeIdentifier($this->database) . '.' . $this->escapeIdentifier($tableName);
1117+
1118+
$columns = [
1119+
'metric String',
1120+
'value Int64',
1121+
'time DateTime64(3) CODEC(Delta(4), LZ4)',
1122+
];
1123+
1124+
foreach ($dims as $dim) {
1125+
$columns[] = $this->escapeIdentifier($dim) . ' LowCardinality(Nullable(String))';
1126+
}
1127+
1128+
if ($this->sharedTables) {
1129+
$columns[] = 'tenant Nullable(String)';
1130+
}
1131+
1132+
$orderByParts = $this->sharedTables ? ['tenant', 'metric', 'time'] : ['metric', 'time'];
1133+
foreach ($dims as $dim) {
1134+
$orderByParts[] = $dim;
1135+
}
1136+
$orderBy = '(' . implode(', ', array_map(fn ($c) => $this->escapeIdentifier($c), $orderByParts)) . ')';
1137+
1138+
$columnDefs = implode(",\n ", $columns);
1139+
1140+
$sql = "
1141+
CREATE TABLE IF NOT EXISTS {$escapedTable} (
1142+
{$columnDefs}
1143+
)
1144+
ENGINE = SummingMergeTree(value)
1145+
ORDER BY {$orderBy}
1146+
PARTITION BY toYYYYMM(time)
1147+
SETTINGS index_granularity = 8192, allow_nullable_key = 1
1148+
";
1149+
1150+
$this->query($sql);
1151+
}
1152+
1153+
/**
1154+
* @param array<int, string> $dims
1155+
*/
1156+
private function createDimRollupMaterializedView(string $name, array $dims): void
1157+
{
1158+
$tableName = $this->getDimRollupTableName($name);
1159+
$mvName = $this->getDimRollupMvName($name);
1160+
1161+
$escapedEvents = $this->escapeIdentifier($this->database) . '.' . $this->escapeIdentifier($this->getEventsTableName());
1162+
$escapedTarget = $this->escapeIdentifier($this->database) . '.' . $this->escapeIdentifier($tableName);
1163+
$escapedMv = $this->escapeIdentifier($this->database) . '.' . $this->escapeIdentifier($mvName);
1164+
1165+
$dimList = implode(', ', array_map(fn ($d) => $this->escapeIdentifier($d), $dims));
1166+
1167+
$selectColumns = ['metric', 'toStartOfDay(time) AS time'];
1168+
foreach ($dims as $dim) {
1169+
$selectColumns[] = $this->escapeIdentifier($dim);
1170+
}
1171+
if ($this->sharedTables) {
1172+
$selectColumns[] = 'tenant';
1173+
}
1174+
$selectColumns[] = 'sum(value) AS value';
1175+
1176+
$groupByColumns = ['metric', 'time'];
1177+
foreach ($dims as $dim) {
1178+
$groupByColumns[] = $this->escapeIdentifier($dim);
1179+
}
1180+
if ($this->sharedTables) {
1181+
$groupByColumns[] = 'tenant';
1182+
}
1183+
1184+
$selectSql = implode(', ', $selectColumns);
1185+
$groupSql = implode(', ', $groupByColumns);
1186+
1187+
$sql = "
1188+
CREATE MATERIALIZED VIEW IF NOT EXISTS {$escapedMv}
1189+
TO {$escapedTarget}
1190+
AS SELECT {$selectSql}
1191+
FROM {$escapedEvents}
1192+
GROUP BY {$groupSql}
1193+
";
1194+
1195+
$this->query($sql);
1196+
}
1197+
10741198
/**
10751199
* Create a MergeTree table for the given type.
10761200
*

‎tests/Benchmark/EventsBench.php‎

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,160 @@ public function testBenchmarks(): void
118118
], Usage::TYPE_EVENT);
119119
});
120120

121+
// Multi-dim MV scenarios — bench_*_mv variants exercise the same
122+
// request shape with setUseDailyRollups on. Commit 5 wires the
123+
// routing through; until then both variants scan raw and produce
124+
// the same numbers.
125+
$this->runBench('bench_events_topN_path_30d_mv', function (string $queryId) use ($start, $end): void {
126+
$this->adapter->setUseDailyRollups(true);
127+
$this->adapter->setNextQueryId($queryId);
128+
$this->usage->find([
129+
UsageQuery::groupBy('path'),
130+
Query::equal('metric', [$this->metric]),
131+
Query::greaterThanEqual('time', $start),
132+
Query::lessThanEqual('time', $end),
133+
Query::limit(500),
134+
], Usage::TYPE_EVENT);
135+
$this->adapter->setUseDailyRollups(false);
136+
});
137+
138+
$this->runBench('bench_events_topN_country_30d_mv', function (string $queryId) use ($start, $end): void {
139+
$this->adapter->setUseDailyRollups(true);
140+
$this->adapter->setNextQueryId($queryId);
141+
$this->usage->find([
142+
UsageQuery::groupBy('country'),
143+
Query::equal('metric', [$this->metric]),
144+
Query::greaterThanEqual('time', $start),
145+
Query::lessThanEqual('time', $end),
146+
Query::limit(200),
147+
], Usage::TYPE_EVENT);
148+
$this->adapter->setUseDailyRollups(false);
149+
});
150+
151+
$this->runBench('bench_events_topN_service_30d_mv', function (string $queryId) use ($start, $end): void {
152+
$this->adapter->setUseDailyRollups(true);
153+
$this->adapter->setNextQueryId($queryId);
154+
$this->usage->find([
155+
UsageQuery::groupBy('service'),
156+
Query::equal('metric', [$this->metric]),
157+
Query::greaterThanEqual('time', $start),
158+
Query::lessThanEqual('time', $end),
159+
Query::limit(200),
160+
], Usage::TYPE_EVENT);
161+
$this->adapter->setUseDailyRollups(false);
162+
});
163+
164+
$this->runBench('bench_events_topN_method_status_30d_mv', function (string $queryId) use ($start, $end): void {
165+
$this->adapter->setUseDailyRollups(true);
166+
$this->adapter->setNextQueryId($queryId);
167+
$this->usage->find([
168+
UsageQuery::groupBy('method'),
169+
UsageQuery::groupBy('status'),
170+
Query::equal('metric', [$this->metric]),
171+
Query::greaterThanEqual('time', $start),
172+
Query::lessThanEqual('time', $end),
173+
Query::limit(200),
174+
], Usage::TYPE_EVENT);
175+
$this->adapter->setUseDailyRollups(false);
176+
});
177+
178+
$todayStart = (new \DateTime('today', new \DateTimeZone('UTC')))->format('Y-m-d H:i:s');
179+
$todayEnd = (new \DateTime('+2 hour'))->format('Y-m-d H:i:s');
180+
$this->runBench('bench_events_topN_path_today_partial', function (string $queryId) use ($todayStart, $todayEnd): void {
181+
$this->adapter->setUseDailyRollups(true);
182+
$this->adapter->setNextQueryId($queryId);
183+
$this->usage->find([
184+
UsageQuery::groupBy('path'),
185+
Query::equal('metric', [$this->metric]),
186+
Query::greaterThanEqual('time', $todayStart),
187+
Query::lessThanEqual('time', $todayEnd),
188+
Query::limit(500),
189+
], Usage::TYPE_EVENT);
190+
$this->adapter->setUseDailyRollups(false);
191+
});
192+
193+
// Must route to raw — filter on a column the path MV doesn't index.
194+
$this->runBench('bench_events_topN_path_30d_filtered_resource', function (string $queryId) use ($start, $end): void {
195+
$this->adapter->setUseDailyRollups(true);
196+
$this->adapter->setNextQueryId($queryId);
197+
$this->usage->find([
198+
UsageQuery::groupBy('path'),
199+
Query::equal('metric', [$this->metric]),
200+
Query::equal('resource', ['function']),
201+
Query::greaterThanEqual('time', $start),
202+
Query::lessThanEqual('time', $end),
203+
Query::limit(500),
204+
], Usage::TYPE_EVENT);
205+
$this->adapter->setUseDailyRollups(false);
206+
});
207+
208+
// Multi-dim breakdown — no single MV covers both path AND country.
209+
$this->runBench('bench_events_topN_path_country', function (string $queryId) use ($start, $end): void {
210+
$this->adapter->setUseDailyRollups(true);
211+
$this->adapter->setNextQueryId($queryId);
212+
$this->usage->find([
213+
UsageQuery::groupBy('path'),
214+
UsageQuery::groupBy('country'),
215+
Query::equal('metric', [$this->metric]),
216+
Query::greaterThanEqual('time', $start),
217+
Query::lessThanEqual('time', $end),
218+
Query::limit(500),
219+
], Usage::TYPE_EVENT);
220+
$this->adapter->setUseDailyRollups(false);
221+
});
222+
223+
// Write fan-out cost with all MVs attached. Compare against
224+
// bench_insert_10k for the multiplier (must stay ≤ 1.3×).
225+
$this->runBench('bench_insert_with_mvs', function (string $queryId): void {
226+
$batch = [];
227+
for ($i = 0; $i < 10000; $i++) {
228+
$batch[] = [
229+
'metric' => 'bench.insert.with_mvs',
230+
'value' => $i,
231+
'tags' => [
232+
'path' => '/v1/mv/' . ($i % 100),
233+
'method' => 'POST',
234+
'status' => '201',
235+
'service' => 'storage',
236+
'country' => 'us',
237+
],
238+
];
239+
}
240+
$this->adapter->setNextQueryId($queryId);
241+
$this->usage->addBatch($batch, Usage::TYPE_EVENT);
242+
}, 3);
243+
244+
// Catches buffered/async insert lag: write then read the same key.
245+
$this->runBench('bench_mv_lag', function (string $queryId): void {
246+
$this->usage->addBatch([
247+
['metric' => 'bench.mv.lag', 'value' => 1, 'tags' => ['path' => '/v1/mv-lag']],
248+
], Usage::TYPE_EVENT);
249+
$this->adapter->setNextQueryId($queryId);
250+
$this->usage->find([
251+
UsageQuery::groupBy('path'),
252+
Query::equal('metric', ['bench.mv.lag']),
253+
Query::limit(10),
254+
], Usage::TYPE_EVENT);
255+
}, 3);
256+
257+
// Storage footprint per busy project — reads system.parts.
258+
$this->runBench('bench_mv_storage_per_busy_project', function (string $queryId): void {
259+
$reflection = new \ReflectionClass($this->adapter);
260+
$getDatabase = $reflection->getProperty('database');
261+
$getDatabase->setAccessible(true);
262+
$databaseValue = $getDatabase->getValue($this->adapter);
263+
$database = is_string($databaseValue) ? $databaseValue : '';
264+
$namespace = $this->namespace;
265+
$this->adapter->setNextQueryId($queryId);
266+
$this->runRawSql(
267+
"SELECT sum(bytes_on_disk) AS total_bytes "
268+
. "FROM system.parts "
269+
. "WHERE database = '" . addslashes($database) . "' "
270+
. "AND table LIKE '" . addslashes($namespace) . "_usage_events_daily_by_%' "
271+
. "AND active = 1 FORMAT JSON"
272+
);
273+
}, 3);
274+
121275
$this->assertNotEmpty($this->results, 'Benchmark scenarios must record results');
122276
}
123277
}

0 commit comments

Comments
 (0)