AIOps Query Cost Analyzer
When to Use
- Identifying which queries consume the most resources on Trino/ClickHouse/Spark
- Automatically recommending query optimizations based on execution patterns
- Enforcing per-user or per-team query cost budgets
- Generating cost chargeback reports broken down by team/project/DAG
- Detecting query cost regressions after code deployments
Trino Query Cost Analysis
-- Top 20 most expensive queries (CPU + memory) in last 24h
SELECT
query_id,
user,
source, -- dbt, airflow, jupyter, etc.
state,
CAST(error_code AS VARCHAR) AS error_code,
DATE_TRUNC('hour', created) AS query_hour,
ROUND(cpu_time_millis / 1000.0 / 60, 1) AS cpu_minutes,
ROUND(peak_memory_bytes / 1e9, 2) AS peak_memory_gb,
ROUND(cumulative_memory / 1e12, 3) AS cumulative_memory_tb_ms,
ROUND(input_bytes / 1e9, 2) AS input_gb,
ROUND(input_rows / 1e6, 1) AS input_rows_M,
wall_time_millis / 1000 AS wall_seconds,
-- Cost proxy: CPU minutes × $0.10 + GB scanned × $0.001
ROUND(cpu_time_millis / 1000.0 / 60 * 0.10 + input_bytes / 1e9 * 0.001, 4) AS estimated_cost_usd,
SUBSTR(query, 1, 200) AS query_preview
FROM system.runtime.queries
WHERE created >= NOW() - INTERVAL '24' HOUR
AND state = 'FINISHED'
ORDER BY estimated_cost_usd DESC
LIMIT 20;
-- Cost by user/team over past 7 days
SELECT
user,
source,
COUNT(*) AS query_count,
ROUND(SUM(cpu_time_millis) / 1000.0 / 3600, 1) AS cpu_hours,
ROUND(SUM(input_bytes) / 1e12, 2) AS input_TB,
ROUND(SUM(peak_memory_bytes) / 1e9, 1) AS peak_mem_gb,
ROUND(
SUM(cpu_time_millis / 1000.0 / 60 * 0.10 + input_bytes / 1e9 * 0.001), 2
) AS estimated_cost_usd,
APPROX_PERCENTILE(wall_time_millis / 1000.0, 0.95) AS p95_wall_seconds
FROM system.runtime.queries
WHERE created >= NOW() - INTERVAL '7' DAY
AND state = 'FINISHED'
GROUP BY user, source
ORDER BY estimated_cost_usd DESC;
-- Queries without partition filters (full-table scans)
SELECT
query_id,
user,
input_rows,
input_bytes,
query
FROM system.runtime.queries
WHERE created >= NOW() - INTERVAL '24' HOUR
AND state = 'FINISHED'
AND input_rows > 100000000 -- scanned > 100M rows
AND query NOT LIKE '%partition_date%'
AND query NOT LIKE '%WHERE%'
ORDER BY input_bytes DESC
LIMIT 10;
ClickHouse Query Cost Analysis
-- Top expensive queries in last 24h
SELECT
user,
query_kind,
round(query_duration_ms / 1000, 2) AS duration_sec,
formatReadableSize(memory_usage) AS memory,
formatReadableSize(read_bytes) AS read_bytes,
formatReadableQuantity(read_rows) AS read_rows,
round(memory_usage / 1e9, 2) AS memory_gb,
arrayStringConcat(tables, ', ') AS tables_accessed,
-- Cost proxy
round(query_duration_ms / 1000 * 0.001 + read_bytes / 1e9 * 0.001, 5) AS cost_usd,
LEFT(query, 300) AS query_preview
FROM system.query_log
WHERE type = 'QueryFinish'
AND event_time >= now() - INTERVAL 24 HOUR
ORDER BY memory_usage DESC
LIMIT 20;
-- Users exceeding daily memory budget (> 100 GB total)
SELECT
user,
COUNT() AS query_count,
formatReadableSize(SUM(memory_usage)) AS total_memory,
formatReadableSize(SUM(read_bytes)) AS total_read_bytes,
round(SUM(query_duration_ms) / 3600000, 2) AS total_hours
FROM system.query_log
WHERE type = 'QueryFinish'
AND event_time >= toStartOfDay(now())
GROUP BY user
HAVING SUM(memory_usage) > 100 * 1e9 -- 100 GB
ORDER BY SUM(memory_usage) DESC;
-- Slow queries by table (join with table stats for context)
SELECT
arrayJoin(tables) AS table_name,
COUNT() AS query_count,
median(query_duration_ms) AS median_ms,
quantile(0.95)(query_duration_ms) AS p95_ms,
formatReadableSize(median(read_bytes)) AS median_scan
FROM system.query_log
WHERE type = 'QueryFinish'
AND event_time >= now() - INTERVAL 7 DAY
GROUP BY table_name
ORDER BY p95_ms DESC
LIMIT 20;
LLM-Driven Query Rewrite Agent
import anthropic
import re
client = anthropic.Anthropic()
QUERY_OPTIMIZER_SYSTEM = """You are a Trino/ClickHouse SQL query optimizer.
Analyze the provided query and its execution stats, then suggest specific rewrites to reduce cost.
Focus on:
1. Missing partition filters (add WHERE partition_date = ...)
2. Full table scans without predicate pushdown
3. Missing broadcast hints for small tables (< 100 MB)
4. GROUP BY before JOIN (reduce rows before joining)
5. APPROX_DISTINCT instead of COUNT(DISTINCT) where exactness not required
6. Unnecessary columns in SELECT *
7. Missing LIMIT on exploratory queries
Respond in JSON:
{
"issues": [{"type": "string", "description": "string", "severity": "high|medium|low"}],
"rewritten_query": "string",
"estimated_cost_reduction_pct": number,
"explanation": "string"
}"""
def analyze_and_rewrite_query(
query: str,
execution_stats: dict,
engine: str = "trino",
) -> dict:
"""Use LLM to analyze a slow/expensive query and suggest rewrites."""
user_message = f"""
Engine: {engine}
Execution stats:
- CPU time: {execution_stats.get('cpu_minutes', 'N/A')} minutes
- Data scanned: {execution_stats.get('input_gb', 'N/A')} GB
- Peak memory: {execution_stats.get('peak_memory_gb', 'N/A')} GB
- Wall time: {execution_stats.get('wall_seconds', 'N/A')} seconds
- Rows scanned: {execution_stats.get('input_rows_M', 'N/A')}M
Query:
```sql
{query}
Analyze this query and suggest cost-reducing rewrites.
"""
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=2048,
system=QUERY_OPTIMIZER_SYSTEM,
messages=[{"role": "user", "content": user_message}],
)
text = response.content[0].text
# Extract JSON from response
json_match = re.search(r'\{.*\}', text, re.DOTALL)
if json_match:
import json
return json.loads(json_match.group())
return {"raw_response": text}
def batch_analyze_expensive_queries(
queries: list[dict], # [{query_id, query, cpu_minutes, input_gb, peak_memory_gb, user}]
top_n: int = 10,
) -> list[dict]:
"""Analyze top N most expensive queries and generate optimization report."""
# Sort by cost
sorted_queries = sorted(queries, key=lambda q: q.get("input_gb", 0), reverse=True)[:top_n]
results = []
for q in sorted_queries:
analysis = analyze_and_rewrite_query(
query=q["query"],
execution_stats=q,
)
results.append({
"query_id": q.get("query_id"),
"user": q.get("user"),
"original_cost_gb": q.get("input_gb"),
"analysis": analysis,
})
return results
---
## Cost Anomaly Detection
```python
import numpy as np
from datetime import datetime, timedelta
def detect_query_cost_anomalies(
daily_costs: dict[str, list[float]], # {user: [daily_cost_usd_30d]}
alert_threshold_sigma: float = 3.0,
) -> list[dict]:
"""
Detect users/teams with anomalous query cost today vs their 30-day baseline.
"""
alerts = []
for user, costs in daily_costs.items():
if len(costs) < 7:
continue
baseline = np.array(costs[:-1]) # all but today
today = costs[-1]
mean = np.mean(baseline)
std = np.std(baseline)
if std == 0:
continue
z = (today - mean) / std
if z > alert_threshold_sigma:
alerts.append({
"user": user,
"today_cost_usd": round(today, 2),
"baseline_mean_usd": round(mean, 2),
"z_score": round(z, 2),
"severity": "critical" if z > 5.0 else "warning",
"message": f"Query cost anomaly: {user} spent ${today:.2f} today vs ${mean:.2f} baseline (z={z:.1f}σ)",
})
return sorted(alerts, key=lambda a: a["z_score"], reverse=True)
Cost Chargeback Report
def generate_chargeback_report(
trino_query_stats: list[dict],
team_mapping: dict[str, str], # {user: team}
month: str, # "2024-01"
) -> str:
"""Generate monthly cost chargeback report by team."""
from collections import defaultdict
team_costs = defaultdict(lambda: {"queries": 0, "cpu_hours": 0, "input_TB": 0, "cost_usd": 0.0})
for q in trino_query_stats:
team = team_mapping.get(q["user"], "unknown")
team_costs[team]["queries"] += 1
team_costs[team]["cpu_hours"] += q.get("cpu_minutes", 0) / 60
team_costs[team]["input_TB"] += q.get("input_gb", 0) / 1000
team_costs[team]["cost_usd"] += q.get("estimated_cost_usd", 0)
lines = [
f"# Query Cost Chargeback Report — {month}",
"",
"| Team | Queries | CPU Hours | Data Scanned (TB) | Estimated Cost |",
"|------|---------|-----------|-------------------|----------------|",
]
total_cost = 0
for team, stats in sorted(team_costs.items(), key=lambda x: x[1]["cost_usd"], reverse=True):
lines.append(
f"| {team} | {stats['queries']:,} | {stats['cpu_hours']:.1f} | "
f"{stats['input_TB']:.2f} | ${stats['cost_usd']:.2f} |"
)
total_cost += stats["cost_usd"]
lines += ["", f"**Total**: ${total_cost:.2f}"]
return "\n".join(lines)
Anti-Patterns
- Cost analysis only on failed queries — expensive queries often succeed but consume disproportionate resources; analyze all finished queries, not just errors.
- No query tagging —
SELECT * from user "john" is impossible to attribute; enforce SET SESSION query_tag = 'dag=etl_orders,task=load_orders' in all pipeline queries.
- Optimizing the wrong queries — a 10-second query run 10,000× per day costs more than a 10-minute query run once; optimize by total cost, not worst single query.
- LLM rewrites without testing — auto-applying LLM-suggested rewrites without running EXPLAIN first can change semantics; require human review for rewrite approval.
- Chargeback without optimization help — showing teams their costs without providing optimization guidance creates finger-pointing; pair chargeback with self-service optimization tips.
References
- Trino system tables:
trino.io/docs/current/connector/system.html
- ClickHouse query_log:
clickhouse.com/docs/en/operations/system-tables/query_log
- BigQuery INFORMATION_SCHEMA.JOBS:
cloud.google.com/bigquery/docs/information-schema-jobs
- Related skills:
[[de-cost-optimization]], [[trino-iceberg]], [[clickhouse-olap]], [[aiops-infrastructure-anomaly-detection]]