Menu
The review-first prompt is selected by default. You can switch to a direct command or download a local copy.
Read SKILL.md and any companion files shown by SkillsMP before deciding whether to install.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
npx skills add https://github.com/microwind/ai-skills --skill sqlThe command stays on one line. Scroll horizontally to inspect it before copying.
Prefer a local copy? Download the files currently available to SkillsMP.
Showing SKILL.md
Based on SOC occupation classification
| name | SQL优化器 |
| description | 当优化SQL查询时，分析执行计划，检查索引使用，优化查询性能。验证SQL语法，分析查询复杂度，和最佳实践。 |
| license | MIT |
慢SQL查询会复合影响。一个慢查询会影响100个事务。在扩展之前先优化。需要建立完善的SQL查询分析和优化机制。
核心原则: 慢SQL查询会复合影响。一个慢查询会影响100个事务。在扩展之前先优化。
始终:
触发短语:
问题:
查询条件没有合适的索引支持
后果:
- 全表扫描
- 查询速度慢
- CPU占用高
- 锁竞争严重
解决方案:
- 创建合适的索引
- 复合索引优化
- 覆盖索引设计
- 索引选择性分析
问题:
SQL查询语句设计不合理
后果:
- 执行计划不优
- 资源浪费
- 性能下降
- 扩展性差
解决方案:
- 重写查询语句
- 避免子查询
- 优化JOIN操作
- 减少数据传输
问题:
数据库统计信息不准确
后果:
- 执行计划错误
- 索引选择不当
- 查询性能下降
- 资源分配不合理
解决方案:
- 更新统计信息
- 定期维护
- 自动统计收集
- 手动统计更新
避免子查询:
- 使用JOIN替代子查询
- 使用EXISTS替代IN
- 优化相关子查询
- 减少嵌套层次
优化JOIN:
- 选择合适的JOIN类型
- 确保连接字段有索引
- 减少JOIN的数据量
- 优化JOIN顺序
单列索引:
- 高选择性字段
- 频繁查询条件
- 排序字段
- 外键字段
复合索引:
- 多条件查询
- 字段顺序重要
- 覆盖查询需求
- 减少回表查询
import re
import sqlparse
from typing import Dict, List, Any, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
class QueryType(Enum):
SELECT = "SELECT"
INSERT = "INSERT"
UPDATE = "UPDATE"
DELETE = "DELETE"
CREATE = "CREATE"
DROP = "DROP"
ALTER = "ALTER"
@dataclass
class QueryAnalysis:
"""查询分析结果"""
query_type: QueryType
complexity_score: int
tables: List[str]
columns: List[str]
joins: List[Dict[str, str]]
where_conditions: List[str]
group_by: List[str]
order_by: List[str]
issues: List[str]
recommendations: List[str]
execution_plan: Optional[Dict[str, Any]] = None
class :
():
.optimization_rules = ._setup_optimization_rules()
.index_suggestions = []
.performance_metrics = {}
() -> QueryAnalysis:
formatted_query = sqlparse.(query, reindent=, keyword_case=)
query_type = ._detect_query_type(formatted_query)
tables = ._extract_tables(formatted_query)
columns = ._extract_columns(formatted_query)
joins = ._extract_joins(formatted_query)
where_conditions = ._extract_where_conditions(formatted_query)
group_by = ._extract_group_by(formatted_query)
order_by = ._extract_order_by(formatted_query)
complexity_score = ._calculate_complexity(
tables, columns, joins, where_conditions, group_by, order_by
)
issues = []
recommendations = []
._analyze_performance_issues(
query_type, tables, columns, joins, where_conditions,
group_by, order_by, issues, recommendations
)
QueryAnalysis(
query_type=query_type,
complexity_score=complexity_score,
tables=tables,
columns=columns,
joins=joins,
where_conditions=where_conditions,
group_by=group_by,
order_by=order_by,
issues=issues,
recommendations=recommendations
)
() -> [, ]:
analysis = .analyze_query(query)
optimized_query = query
optimizations_applied = []
rule .optimization_rules:
result = rule.apply(optimized_query, analysis)
result.modified:
optimized_query = result.query
optimizations_applied.append(rule.name)
index_suggestions = ._generate_index_suggestions(analysis)
execution_plan_hints = ._generate_execution_plan_hints(analysis)
{
: query,
: optimized_query,
: analysis,
: optimizations_applied,
: index_suggestions,
: execution_plan_hints,
: ._estimate_improvement(analysis)
}
() -> QueryType:
query_upper = query.strip().upper()
query_upper.startswith():
QueryType.SELECT
query_upper.startswith():
QueryType.INSERT
query_upper.startswith():
QueryType.UPDATE
query_upper.startswith():
QueryType.DELETE
query_upper.startswith():
QueryType.CREATE
query_upper.startswith():
QueryType.DROP
query_upper.startswith():
QueryType.ALTER
:
QueryType.SELECT
() -> []:
tables = []
from_pattern =
from_matches = re.findall(from_pattern, query, re.IGNORECASE)
tables.extend(from_matches)
join_pattern =
join_matches = re.findall(join_pattern, query, re.IGNORECASE)
tables.extend(join_matches)
insert_pattern =
insert_matches = re.findall(insert_pattern, query, re.IGNORECASE)
tables.extend(insert_matches)
update_pattern =
update_matches = re.findall(update_pattern, query, re.IGNORECASE)
tables.extend(update_matches)
((tables))
() -> []:
columns = []
select_pattern =
select_match = re.search(select_pattern, query, re.IGNORECASE | re.DOTALL)
select_match:
select_clause = select_match.group()
select_clause != :
column_pattern =
column_matches = re.findall(column_pattern, select_clause)
columns.extend(column_matches)
((columns))
() -> [[, ]]:
joins = []
join_pattern =
join_matches = re.findall(join_pattern, query, re.IGNORECASE)
join_type, table, condition join_matches:
joins.append({
: join_type.upper(),
: table,
: condition.strip()
})
joins
() -> []:
conditions = []
where_pattern =
where_match = re.search(where_pattern, query, re.IGNORECASE | re.DOTALL)
where_match:
where_clause = where_match.group()
and_conditions = re.split(, where_clause, flags=re.IGNORECASE)
condition and_conditions:
or_conditions = re.split(, condition, flags=re.IGNORECASE)
conditions.extend([c.strip() c or_conditions])
conditions
() -> []:
group_by = []
group_pattern =
group_match = re.search(group_pattern, query, re.IGNORECASE)
group_match:
group_clause = group_match.group()
group_by = [col.strip() col group_clause.split()]
group_by
() -> []:
order_by = []
order_pattern =
order_match = re.search(order_pattern, query, re.IGNORECASE)
order_match:
order_clause = order_match.group()
order_by = [col.strip() col order_clause.split()]
order_by
() -> :
score =
score += (tables) *
score += (columns) *
score += (joins) *
score += (where_conditions) *
score += (group_by) *
score += (order_by) *
score
():
query_type == QueryType.SELECT columns:
issues.append()
recommendations.append()
(joins) > :
issues.append()
recommendations.append()
where_conditions query_type == QueryType.SELECT:
issues.append()
recommendations.append()
group_by order_by:
(group_by) & (order_by):
recommendations.append()
(tables) > :
issues.append()
recommendations.append()
() -> [[, ]]:
suggestions = []
condition analysis.where_conditions:
field_pattern =
field_matches = re.findall(field_pattern, condition)
field field_matches:
suggestions.append({
: ,
: analysis.tables[] analysis.tables ,
: field,
: ,
:
})
join analysis.joins:
condition = join[]
field_pattern =
field_matches = re.findall(field_pattern, condition)
table1, table2, field field_matches:
suggestions.append({
: ,
: table1,
: field,
: ,
:
})
analysis.order_by:
field analysis.order_by:
clean_field = field.split()[]
suggestions.append({
: ,
: analysis.tables[] analysis.tables ,
: clean_field,
: ,
:
})
suggestions
() -> []:
hints = []
analysis.complexity_score > :
hints.append()
(analysis.joins) > :
hints.append()
analysis.group_by (analysis.group_by) > :
hints.append()
hints
() -> [, ]:
improvement = {
: ,
: ,
:
}
issue_count = (analysis.issues)
issue_count > :
improvement[] = (, issue_count * )
improvement[] = (, issue_count * )
improvement[] = (, issue_count * )
improvement
() -> []:
[
SelectStarOptimization(),
SubqueryOptimization(),
JoinOptimization(),
WhereOptimization(),
IndexHintOptimization()
]
:
():
.name = name
() -> :
NotImplementedError
:
query:
modified:
reason:
():
():
().__init__()
() -> OptimizationResult:
analysis.columns:
OptimizationResult(
query=query,
modified=,
reason=
)
OptimizationResult(query=query, modified=, reason=)
():
():
().__init__()
() -> OptimizationResult:
optimized_query = query
in_pattern =
re.search(in_pattern, optimized_query, re.IGNORECASE | re.DOTALL):
OptimizationResult(
query=optimized_query,
modified=,
reason=
)
OptimizationResult(query=optimized_query, modified=, reason=)
():
():
().__init__()
() -> OptimizationResult:
optimized_query = query
(analysis.joins) > :
OptimizationResult(
query=optimized_query,
modified=,
reason=
)
OptimizationResult(query=optimized_query, modified=, reason=)
():
():
().__init__()
() -> OptimizationResult:
optimized_query = query
condition analysis.where_conditions:
re.search(, condition, re.IGNORECASE):
OptimizationResult(
query=optimized_query,
modified=,
reason=
)
OptimizationResult(query=optimized_query, modified=, reason=)
():
():
().__init__()
() -> OptimizationResult:
OptimizationResult(
query=query,
modified=,
reason=
)
():
optimizer = SQLOptimizer()
sample_queries = [
,
,
]
i, query (sample_queries, ):
()
()
analysis = optimizer.analyze_query(query)
()
()
()
()
analysis.issues:
()
issue analysis.issues:
()
analysis.recommendations:
()
rec analysis.recommendations:
()
optimization = optimizer.optimize_query(query)
()
optimization[]:
()
suggestion optimization[]:
()
improvement = optimization[]
()
__name__ == :
main()
import time
import psycopg2
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import statistics
@dataclass
class PerformanceMetrics:
"""性能指标"""
query_time: float
rows_examined: int
rows_returned: int
index_usage: Dict[str, int]
buffer_usage: Dict[str, int]
cpu_usage: float
memory_usage: float
class DatabasePerformanceMonitor:
"""数据库性能监控"""
def __init__(self, connection_string: str):
self.connection_string = connection_string
self.metrics_history = []
self.slow_queries = []
def monitor_query_performance(self, query: str, params: tuple = None) -> PerformanceMetrics:
"""监控查询性能"""
start_time = time.time()
with psycopg2.connect(self.connection_string) as conn:
conn.cursor() cursor:
explain_query =
cursor.execute(explain_query, params)
explain_result = cursor.fetchone()[]
cursor.execute(query, params)
rows = cursor.fetchall()
end_time = time.time()
plan = explain_result[]
metrics = PerformanceMetrics(
query_time=end_time - start_time,
rows_examined=plan.get(, ),
rows_returned=(rows),
index_usage=._extract_index_usage(plan),
buffer_usage=._extract_buffer_usage(plan),
cpu_usage=._get_cpu_usage(),
memory_usage=._get_memory_usage()
)
.metrics_history.append(metrics)
metrics.query_time > :
.slow_queries.append({
: query,
: params,
: metrics,
: datetime.now()
})
metrics
() -> [, ]:
index_usage = {}
():
node:
node_type = node[]
node:
index_name = node[]
index_usage[index_name] = index_usage.get(index_name, ) +
node:
sub_plan node[]:
traverse_plan(sub_plan)
traverse_plan(plan)
index_usage
() -> [, ]:
buffers = plan.get(, {})
{
: buffers.get(, ),
: buffers.get(, ),
: buffers.get(, ),
: buffers.get(, ),
: buffers.get(, ),
: buffers.get(, ),
: buffers.get(, ),
: buffers.get(, ),
: buffers.get(, ),
: buffers.get(, )
}
() -> :
psutil
psutil.cpu_percent()
() -> :
psutil
psutil.virtual_memory().percent
() -> [, ]:
cutoff_time = datetime.now() - timedelta(hours=hours)
recent_metrics = [
m m .metrics_history
datetime.now() - timedelta(hours=hours) <= datetime.now()
]
recent_metrics:
{: }
query_times = [m.query_time m recent_metrics]
cpu_usage = [m.cpu_usage m recent_metrics]
memory_usage = [m.memory_usage m recent_metrics]
analysis = {
: hours,
: (recent_metrics),
: statistics.mean(query_times),
: (query_times),
: (query_times),
: statistics.mean(cpu_usage),
: statistics.mean(memory_usage),
: ([m m recent_metrics m.query_time > ]),
: ._calculate_trend(query_times)
}
analysis
() -> :
(values) < :
x = (((values)))
n = (values)
sum_x = (x)
sum_y = (values)
sum_xy = (x[i] * values[i] i (n))
sum_x2 = (x[i] ** i (n))
slope = (n * sum_xy - sum_x * sum_y) / (n * sum_x2 - sum_x ** )
slope > :
slope < -:
:
() -> [, ]:
.metrics_history:
{: }
slow_query_analysis = ._analyze_slow_queries()
index_analysis = ._analyze_index_usage()
buffer_analysis = ._analyze_buffer_usage()
recommendations = ._generate_recommendations(
slow_query_analysis, index_analysis, buffer_analysis
)
{
: {
: (.metrics_history),
: (.slow_queries),
: statistics.mean([m.query_time m .metrics_history])
},
: slow_query_analysis,
: index_analysis,
: buffer_analysis,
: recommendations
}
() -> [, ]:
.slow_queries:
{: }
slow_times = [sq[].query_time sq .slow_queries]
{
: (.slow_queries),
: statistics.mean(slow_times),
: (slow_times),
: (slow_times),
: [
{
: sq[][:] + ,
: sq[].query_time,
: sq[].rows_examined,
: sq[].isoformat()
}
sq .slow_queries[-:]
]
}
() -> [, ]:
all_index_usage = {}
metrics .metrics_history:
index, count metrics.index_usage.items():
all_index_usage[index] = all_index_usage.get(index, ) + count
sorted_indexes = (all_index_usage.items(), key= x: x[], reverse=)
{
: (all_index_usage.values()),
: (all_index_usage),
: sorted_indexes[:],
: []
}
() -> [, ]:
total_shared_hit = (m.buffer_usage.get(, ) m .metrics_history)
total_shared_read = (m.buffer_usage.get(, ) m .metrics_history)
hit_ratio = total_shared_hit / (total_shared_hit + total_shared_read) (total_shared_hit + total_shared_read) >
{
: hit_ratio,
: total_shared_hit,
: total_shared_read,
: hit_ratio >
}
() -> []:
recommendations = []
slow_query_analysis.get(, ) > :
recommendations.append(
)
recommendations.append()
hit_ratio = buffer_analysis.get(, )
hit_ratio < :
recommendations.append(
)
index_analysis.get(, ) > :
recommendations.append()
recommendations
():
monitor = DatabasePerformanceMonitor()
queries = [
,
,
]
query queries:
()
metrics = monitor.monitor_query_performance(query)
()
()
()
()
report = monitor.generate_optimization_report()
()
()
()
()
report[]:
()
rec report[]:
()
__name__ == :
main()