Menü
Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
npx skills add https://github.com/microwind/ai-skills --skill sqlDer Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.
Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.
Basierend auf der SOC-Berufsklassifikation
SKILL.md wird angezeigt
| 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()