StarRocks AI Query Autotuner
When to Use
- Automatically diagnose slow StarRocks queries without manual EXPLAIN reading
- Recommend materialized views based on slow query patterns
- Detect missing statistics and trigger ANALYZE automatically
- Generate query hints (LEADING, JOIN_METHOD) for bad join plans
- Classify query anti-patterns (full table scan, Cartesian join, no partition filter)
Autotuner Agent Workflow
Input: slow query SQL or query_id
│
├── 1. Fetch EXPLAIN COSTS plan
├── 2. Classify anti-patterns (full scan / bad join / no stats)
├── 3. Analyze statistics freshness (SHOW ANALYZE STATUS)
├── 4. Check colocate group membership
├── 5. Generate recommendations
│ ├── MV suggestion
│ ├── Index suggestion (bitmap/bloom)
│ ├── Query hint (LEADING/JOIN_METHOD)
│ └── Rewrite with partition filter
└── 6. Output: prioritized action list
Step 1: Fetch and Parse EXPLAIN COSTS
import pymysql
import re
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class PlanNode:
node_type: str
output_rows: float
cost: float
table: Optional[str] = None
partition_info: Optional[str] = None
is_full_scan: bool = False
def get_explain_costs(host: str, sql: str) -> str:
conn = pymysql.connect(host=host, port=9030, user="root", db="sales")
cursor = conn.cursor()
cursor.execute(f"EXPLAIN COSTS {sql}")
rows = cursor.fetchall()
conn.close()
return "\n".join(r[0] for r in rows)
def parse_explain_plan(plan_text: str) -> List[PlanNode]:
"""Extract key nodes from EXPLAIN COSTS output."""
nodes = []
lines = plan_text.split("\n")
for line in lines:
line = line.strip()
# Detect OlapScanNode (table scan)
if "OlapScanNode" in line or "SCAN" in line.upper():
table_match = re.search(r'TABLE:\s*(\w+)', line, re.IGNORECASE)
partition_match = re.search(r'partitions=(\d+)/(\d+)', line)
rows_match = re.search(r'cardinality=(\d+)', line)
is_full_scan = False
if partition_match:
scanned = int(partition_match.group(1))
total = int(partition_match.group(2))
is_full_scan = (scanned == total and total > 1)
nodes.append(PlanNode(
node_type="SCAN",
output_rows=float(rows_match.group(1)) if rows_match else 0,
cost=0.0,
table=table_match.group(1) if table_match else None,
partition_info=f"{partition_match.group(1)}/{partition_match.group(2)}" if partition_match else None,
is_full_scan=is_full_scan,
))
# Detect join types
elif "CrossJoin" in line or "CROSS JOIN" in line.upper():
nodes.append(PlanNode(node_type="CROSS_JOIN", output_rows=0, cost=0))
elif "HashJoin" in line or "HASH_JOIN" in line.upper():
rows_match = re.search(r'cardinality=(\d+)', line)
nodes.append(PlanNode(
node_type="HASH_JOIN",
output_rows=float(rows_match.group(1)) if rows_match else 0,
cost=0,
))
return nodes
Step 2: Anti-Pattern Classifier
@dataclass
class QueryIssue:
severity: str # "critical", "warning", "info"
category: str # "full_scan", "missing_stats", "bad_join", "no_partition_filter", "cartesian"
description: str
recommendation: str
def classify_issues(plan_nodes: List[PlanNode], plan_text: str) -> List[QueryIssue]:
issues = []
# Full table scans
for node in plan_nodes:
if node.node_type == "SCAN" and node.is_full_scan:
issues.append(QueryIssue(
severity="critical",
category="full_scan",
description=f"Full partition scan on table {node.table}: {node.partition_info} partitions scanned",
recommendation=f"Add WHERE clause on partition column of {node.table} (e.g., WHERE dt = '...')"
))
# Cartesian join
if any(n.node_type == "CROSS_JOIN" for n in plan_nodes):
issues.append(QueryIssue(
severity="critical",
category="cartesian",
description="Cartesian (CROSS) join detected — exponential row explosion",
recommendation="Add explicit JOIN ON condition or WHERE clause to eliminate cross join"
))
# Very large intermediate results
for node in plan_nodes:
if node.node_type == "HASH_JOIN" and node.output_rows > 1_000_000_000:
issues.append(QueryIssue(
severity="warning",
category="bad_join",
description=f"Hash join produces {node.output_rows/1e9:.1f}B rows — possible fanout",
recommendation="Verify join cardinality; add filters before join; consider AGGREGATE or LIMIT upstream"
))
# No partition filter (heuristic: check if plan mentions all partitions)
if re.search(r'partitions=(\d+)/\1', plan_text): # scanned == total
issues.append(QueryIssue(
severity="warning",
category="no_partition_filter",
description="Table scan reads all partitions",
recommendation="Add equality or range filter on the partition column (e.g., WHERE dt >= '2024-01-01')"
))
return issues
Step 3: Statistics Staleness Check
from datetime import datetime, timedelta
def check_stale_statistics(
host: str,
db: str,
tables: List[str],
max_age_hours: int = 24,
) -> List[dict]:
conn = pymysql.connect(host=host, port=9030, user="root", db=db)
cursor = conn.cursor()
stale = []
for table in tables:
cursor.execute(f"""
SELECT TableName, Status, StartTime, EndTime
FROM _statistics_.analyze_jobs
WHERE DbName = '{db}' AND TableName = '{table}'
ORDER BY StartTime DESC
LIMIT 1
""")
row = cursor.fetchone()
if not row:
stale.append({"table": table, "reason": "no_statistics"})
else:
status = row[1]
end_time = row[3]
if isinstance(end_time, str):
end_time = datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S")
age_hours = (datetime.utcnow() - end_time).total_seconds() / 3600
if status != "SUCCESS" or age_hours > max_age_hours:
stale.append({
"table": table,
"reason": f"stale ({age_hours:.1f}h old, status={status})",
})
conn.close()
return stale
Step 4: Materialized View Recommendation
def recommend_materialized_view(sql: str, slow_tables: List[str]) -> str:
"""Generate a MV DDL recommendation based on query pattern."""
# Detect GROUP BY + aggregation pattern
group_by_match = re.search(r'GROUP BY\s+(.+?)(?:ORDER BY|LIMIT|$)', sql, re.IGNORECASE | re.DOTALL)
select_match = re.search(r'SELECT\s+(.+?)\s+FROM', sql, re.IGNORECASE | re.DOTALL)
if not group_by_match:
return "No GROUP BY detected — MV not recommended for this query pattern"
group_cols = group_by_match.group(1).strip()
table_name = slow_tables[0] if slow_tables else "your_table"
mv_suggestion = f"""
-- Recommended Materialized View:
-- This MV pre-aggregates the GROUP BY pattern from the slow query.
-- Refresh every hour for near-real-time analytics.
CREATE MATERIALIZED VIEW {table_name}_mv_auto
DISTRIBUTED BY HASH({group_cols.split(',')[0].strip()}) BUCKETS 8
REFRESH ASYNC EVERY (INTERVAL 1 HOUR)
AS
{sql.rstrip(';')};
-- After creation, verify the optimizer uses the MV:
-- EXPLAIN {sql}
-- Look for: MaterializedViewScanNode instead of OlapScanNode
"""
return mv_suggestion
Step 5: Query Rewrite with Hints
def add_join_hints(sql: str, join_order: List[str], join_method: str = "BROADCAST") -> str:
"""Inject LEADING and JOIN_METHOD hints into a SELECT statement."""
leading = ", ".join(join_order)
hint = f"/*+ LEADING({leading}) JOIN_METHOD({join_order[0]}, {join_order[1]}, {join_method}) */"
# Insert hint after SELECT
rewritten = re.sub(
r'^SELECT\s+',
f'SELECT {hint} ',
sql.strip(),
count=1,
flags=re.IGNORECASE,
)
return rewritten
def add_partition_filter(sql: str, table: str, partition_col: str, ds: str) -> str:
"""Add a partition filter if missing."""
if partition_col.lower() not in sql.lower():
filter_clause = f"{table}.{partition_col} = '{ds}'"
if "WHERE" in sql.upper():
sql = re.sub(r'\bWHERE\b', f"WHERE {filter_clause} AND", sql, count=1, flags=re.IGNORECASE)
else:
sql = re.sub(r'\bGROUP BY\b', f"WHERE {filter_clause} GROUP BY", sql, count=1, flags=re.IGNORECASE)
return sql
Step 6: Full Autotuner Report
def autotune_query(host: str, db: str, sql: str) -> str:
"""Full autotuner: analyze query, return prioritized recommendations."""
report_lines = ["## StarRocks Query Autotuner Report", ""]
# Get plan
plan_text = get_explain_costs(host, sql)
plan_nodes = parse_explain_plan(plan_text)
# Classify issues
issues = classify_issues(plan_nodes, plan_text)
if issues:
report_lines.append("### Issues Found")
for issue in sorted(issues, key=lambda i: {"critical": 0, "warning": 1, "info": 2}[i.severity]):
report_lines.append(f"- [{issue.severity.upper()}] {issue.description}")
report_lines.append(f" → {issue.recommendation}")
report_lines.append("")
# Check statistics
tables = [n.table for n in plan_nodes if n.table]
if tables:
stale = check_stale_statistics(host, db, tables)
if stale:
report_lines.append("### Stale Statistics (auto-ANALYZE recommended)")
for s in stale:
report_lines.append(f"- {s['table']}: {s['reason']}")
report_lines.append(f" → Run: ANALYZE TABLE {db}.{s['table']} WITH ASYNC MODE")
report_lines.append("")
# MV recommendation
slow_tables = [n.table for n in plan_nodes if n.is_full_scan and n.table]
if slow_tables and "GROUP BY" in sql.upper():
mv = recommend_materialized_view(sql, slow_tables)
report_lines.append("### Materialized View Recommendation")
report_lines.append(mv)
if not issues and not stale:
report_lines.append("### Result: No issues detected")
report_lines.append("Query plan looks healthy.")
return "\n".join(report_lines)
Slow Query Log Analysis
-- Find top 10 slowest queries from StarRocks query log
SELECT
QueryId,
StartTime,
EndTime,
QueryTime,
State,
LEFT(Statement, 200) AS sql_preview
FROM information_schema.be_cloud_native_compactions
-- For query history, use:
SHOW PROCESSLIST;
-- Or query FE audit log directly:
-- /data/starrocks/fe/log/fe.audit.log
-- Lines with "Query"
Python: parse FE audit log for slow queries:
import re
from datetime import datetime
def parse_fe_audit_log(log_path: str, min_query_time_ms: int = 5000):
"""Extract slow queries from StarRocks FE audit log."""
slow = []
pattern = re.compile(
r'QueryDetail\|(?P<ts>[^|]+)\|(?P<user>[^|]+)\|(?P<db>[^|]+)\|'
r'(?P<time>\d+)\|(?P<rows>\d+)\|(?P<bytes>\d+)\|(?P<state>\w+)\|'
r'(?P<sql>.+)'
)
with open(log_path) as f:
for line in f:
m = pattern.search(line)
if m and int(m.group("time")) >= min_query_time_ms:
slow.append({
"ts": m.group("ts"),
"user": m.group("user"),
"db": m.group("db"),
"time_ms": int(m.group("time")),
"state": m.group("state"),
"sql": m.group("sql")[:500],
})
return sorted(slow, key=lambda x: x["time_ms"], reverse=True)
Anti-Patterns
- Running EXPLAIN COSTS on production during peak hours — EXPLAIN compiles the query plan; light overhead but avoids running for complex sub-queries during peak load.
- Auto-applying MV recommendations without testing — MVs consume storage and refresh resources; always test query improvement before creating in production.
- Adding LEADING hints without verifying cardinality — wrong join order hint makes the plan worse; validate with EXPLAIN COSTS before deploying hint.
- Triggering ANALYZE synchronously in autotuner — blocks; always use
WITH ASYNC MODE and check status separately.
- Classifying issues from plan text only — partition_info in EXPLAIN shows runtime skipping, not always visible in text; always verify with actual query timing.
References
- EXPLAIN COSTS:
docs.starrocks.io/docs/administration/management/Query_management/Cost_based_optimizer/
- Query hints:
docs.starrocks.io/docs/sql-reference/sql-statements/hint/
- Materialized Views:
docs.starrocks.io/docs/using_starrocks/Materialized_view/
- Related skills:
[[starrocks-explain-plan]], [[starrocks-cbo]], [[starrocks-query-optimizer]], [[starrocks-materialized-views]]