| name | SQL优化与索引 |
| description | 当优化数据库性能时,分析查询执行计划,设计高效索引,优化SQL语句。调整数据库配置,监控性能指标,和实施最佳实践。 |
| license | MIT |
SQL优化与索引技能
概述
SQL优化是提升数据库性能的核心技术。不当的查询设计和索引策略会导致系统响应缓慢、资源浪费、用户体验差。需要系统性的优化方法。
核心原则: 好的SQL优化应该查询高效、索引合理、资源利用充分、响应快速。坏的SQL优化会查询缓慢、资源浪费、系统不稳定。
何时使用
始终:
- 查询响应时间过长时
- 系统负载过高时
- 数据库性能下降时
- 用户体验不佳时
- 资源使用率异常时
- 并发访问困难时
触发短语:
- "SQL查询优化"
- "索引设计策略"
- "查询性能分析"
- "执行计划解读"
- "数据库调优"
- "慢查询优化"
SQL优化功能
查询分析
- 执行计划分析
- 查询成本评估
- 性能瓶颈识别
- 资源使用分析
- 慢查询检测
索引优化
- 索引设计策略
- 复合索引优化
- 索引使用分析
- 索引维护管理
- 索引效果评估
语句优化
- 查询重写
- 连接优化
- 子查询优化
- 聚合查询优化
- 分页查询优化
配置调优
- 内存配置优化
- 连接池配置
- 缓存策略调整
- 并发参数优化
- 存储引擎优化
常见SQL优化问题
全表扫描
问题:
查询导致全表扫描,性能极差
错误示例:
- 缺少合适的索引
- 索引选择性差
- 查询条件不当
- 统计信息过时
解决方案:
1. 创建合适的索引
2. 优化查询条件
3. 更新统计信息
4. 重写查询语句
索引失效
问题:
索引存在但不被使用,查询仍然缓慢
错误示例:
- 索引列使用函数
- 类型不匹配
- 隐式转换
- 前导通配符
解决方案:
1. 避免索引列计算
2. 保证类型一致
3. 使用函数索引
4. 优化查询条件
连接查询性能差
问题:
多表连接查询性能差,响应时间长
错误示例:
- 缺少连接索引
- 连接顺序不当
- 笛卡尔积
- 不必要的连接
解决方案:
1. 优化连接索引
2. 调整连接顺序
3. 避免笛卡尔积
4. 减少连接表数
子查询性能问题
问题:
子查询执行效率低,影响整体性能
错误示例:
- 相关子查询
- 多层嵌套子查询
- 非优化子查询
- 重复计算
解决方案:
1. 使用JOIN代替子查询
2. 优化子查询结构
3. 使用EXISTS代替IN
4. 减少嵌套层次
代码实现示例
SQL查询分析器
import re
import json
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import time
class QueryType(Enum):
"""查询类型"""
SELECT = "select"
INSERT = "insert"
UPDATE = "update"
DELETE = "delete"
CREATE = "create"
DROP = "drop"
ALTER = "alter"
class OptimizationLevel(Enum):
"""优化级别"""
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class QueryAnalysis:
"""查询分析结果"""
query: str
query_type: QueryType
complexity_score: int
tables: List[str]
columns: List[str]
joins: List[str]
where_conditions: List[str]
group_by: List[str]
order_by: List[str]
having_conditions: []
subqueries: []
:
:
description:
impact:
difficulty:
example:
:
table_name:
column_names: []
index_type:
estimated_improvement:
reason:
:
():
.optimization_patterns = {
: ._detect_full_table_scan,
: ._detect_missing_index,
: ._detect_inefficient_join,
: ._detect_subquery_optimization,
: ._detect_function_on_index,
: ._detect_select_star,
: ._detect_implicit_conversion,
}
.index_analyzer = IndexAnalyzer()
() -> QueryAnalysis:
normalized_query = ._normalize_query(query)
query_type = ._identify_query_type(normalized_query)
tables = ._extract_tables(normalized_query)
columns = ._extract_columns(normalized_query)
joins = ._extract_joins(normalized_query)
where_conditions = ._extract_where_conditions(normalized_query)
group_by = ._extract_group_by(normalized_query)
order_by = ._extract_order_by(normalized_query)
having_conditions = ._extract_having_conditions(normalized_query)
subqueries = ._extract_subqueries(normalized_query)
complexity_score = ._calculate_complexity_score(
tables, joins, where_conditions, group_by, having_conditions, subqueries
)
QueryAnalysis(
query=query,
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,
having_conditions=having_conditions,
subqueries=subqueries
)
() -> [OptimizationSuggestion]:
suggestions = []
pattern_name, pattern_func .optimization_patterns.items():
pattern_suggestions = pattern_func(analysis)
suggestions.extend(pattern_suggestions)
index_suggestions = ._generate_index_suggestions(analysis)
suggestions.extend(index_suggestions)
suggestions.sort(key= x: ._get_impact_priority(x.impact), reverse=)
suggestions
() -> :
query = re.sub(, , query)
query = re.sub(, , query, flags=re.DOTALL)
query = re.sub(, , query)
query = query.strip()
query
() -> QueryType:
query_upper = query.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_match = re.search(, query, re.IGNORECASE)
from_match:
from_clause = from_match.group()
from_clause = re.sub(, , from_clause)
table_names = re.findall(, from_clause)
tables.extend(table_names)
join_matches = re.findall(, query, re.IGNORECASE)
tables.extend(join_matches)
((tables))
() -> []:
columns = []
select_match = re.search(, query, re.IGNORECASE | re.DOTALL)
select_match:
select_clause = select_match.group()
select_clause:
columns.append()
:
column_names = re.findall(, select_clause)
columns.extend([col col column_names col.upper() [, , , , , ]])
((columns))
() -> []:
joins = []
join_patterns = [
,
,
,
,
]
pattern join_patterns:
matches = re.findall(pattern, query, re.IGNORECASE)
matches:
join_info =
joins.append(join_info)
joins
() -> []:
conditions = []
where_match = re.search(, query, re.IGNORECASE | re.DOTALL)
where_match:
where_clause = where_match.group().strip()
condition_parts = re.split(, where_clause, flags=re.IGNORECASE)
conditions.extend([part.strip() part condition_parts part.strip()])
conditions
() -> []:
group_by_match = re.search(, query, re.IGNORECASE)
group_by_match:
group_by_clause = group_by_match.group()
[col.strip() col group_by_clause.split()]
[]
() -> []:
order_by_match = re.search(, query, re.IGNORECASE)
order_by_match:
order_by_clause = order_by_match.group()
[col.strip() col order_by_clause.split()]
[]
() -> []:
conditions = []
having_match = re.search(, query, re.IGNORECASE | re.DOTALL)
having_match:
having_clause = having_match.group().strip()
condition_parts = re.split(, having_clause, flags=re.IGNORECASE)
conditions.extend([part.strip() part condition_parts part.strip()])
conditions
() -> []:
subqueries = []
subquery_patterns = [
,
,
,
]
pattern subquery_patterns:
matches = re.findall(pattern, query, re.IGNORECASE | re.DOTALL)
subqueries.extend(matches)
subqueries
() -> :
score =
score += (tables) *
score += (joins) *
score += (where_conditions) *
score += (group_by) *
score += (having_conditions) *
score += (subqueries) *
score
() -> [OptimizationSuggestion]:
suggestions = []
analysis.where_conditions ( cond cond analysis.where_conditions):
suggestions.append(OptimizationSuggestion(
=,
description=,
impact=,
difficulty=,
example=
))
suggestions
() -> [OptimizationSuggestion]:
suggestions = []
condition analysis.where_conditions:
columns = re.findall(, condition)
column columns:
column analysis.columns:
suggestions.append(OptimizationSuggestion(
=,
description=,
impact=,
difficulty=,
example=
))
suggestions
() -> [OptimizationSuggestion]:
suggestions = []
join analysis.joins:
join:
on_condition = join.split()[].strip()
on_condition:
suggestions.append(OptimizationSuggestion(
=,
description=,
impact=,
difficulty=,
example=
))
suggestions
() -> [OptimizationSuggestion]:
suggestions = []
subquery analysis.subqueries:
subquery subquery:
suggestions.append(OptimizationSuggestion(
=,
description=,
impact=,
difficulty=,
example=
))
suggestions
() -> [OptimizationSuggestion]:
suggestions = []
condition analysis.where_conditions:
re.search(, condition):
suggestions.append(OptimizationSuggestion(
=,
description=,
impact=,
difficulty=,
example=
))
suggestions
() -> [OptimizationSuggestion]:
suggestions = []
analysis.columns:
suggestions.append(OptimizationSuggestion(
=,
description=,
impact=,
difficulty=,
example=
))
suggestions
() -> [OptimizationSuggestion]:
suggestions = []
condition analysis.where_conditions:
condition re.search(, condition):
suggestions.append(OptimizationSuggestion(
=,
description=,
impact=,
difficulty=,
example=
))
suggestions
() -> [OptimizationSuggestion]:
suggestions = []
analysis.where_conditions:
columns = []
condition analysis.where_conditions:
condition_columns = re.findall(, condition)
columns.extend(condition_columns)
columns:
unique_columns = ((columns))
(unique_columns) > :
suggestions.append(OptimizationSuggestion(
=,
description=,
impact=,
difficulty=,
example=
))
:
suggestions.append(OptimizationSuggestion(
=,
description=,
impact=,
difficulty=,
example=
))
analysis.order_by analysis.where_conditions:
order_columns = [col.split()[] col analysis.order_by]
suggestions.append(OptimizationSuggestion(
=,
description=,
impact=,
difficulty=,
example=
))
suggestions
() -> :
impact_priorities = {
: ,
: ,
:
}
impact_priorities.get(impact, )
:
():
.index_types = {
: ,
: ,
: ,
: ,
: ,
:
}
() -> [, ]:
analysis = {
: [],
: [],
: [],
: []
}
index existing_indexes:
index[].lower() query.lower():
index_columns = index[]
query_lower = query.lower()
used =
column index_columns:
column.lower() query_lower:
used =
used:
analysis[].append(index)
:
analysis[].append(index)
analysis
():
()
analyzer = SQLQueryAnalyzer()
queries = [
,
,
]
i, query (queries, ):
()
()
analysis = analyzer.analyze_query(query)
()
()
()
()
()
()
()
suggestions = analyzer.generate_optimization_suggestions(analysis)
suggestions:
()
j, suggestion (suggestions, ):
()
()
:
()
__name__ == :
main()
索引设计器
from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from enum import Enum
import json
class IndexType(Enum):
"""索引类型"""
BTREE = "btree"
HASH = "hash"
GIST = "gist"
GIN = "gin"
SPATIAL = "spatial"
FULLTEXT = "fulltext"
class IndexUsage(Enum):
"""索引用途"""
EQUALITY = "equality"
RANGE = "range"
ORDER = "order"
GROUP = "group"
JOIN = "join"
SEARCH = "search"
@dataclass
class ColumnInfo:
"""列信息"""
name: str
data_type: str
is_nullable: bool
is_unique: bool
cardinality: int
selectivity: float
avg_length: int
@dataclass
class TableInfo:
"""表信息"""
name: str
row_count: int
columns: List[ColumnInfo]
primary_key: []
foreign_keys: [, ]
:
name:
table_name:
columns: []
index_type: IndexType
is_unique:
is_partial:
where_condition: [] =
usage_types: [IndexUsage] = field(default_factory=)
:
definition: IndexDefinition
reason:
estimated_improvement:
creation_cost:
priority:
:
():
.design_rules = {
: ._design_primary_key_index,
: ._design_foreign_key_index,
: ._design_equality_search_index,
: ._design_range_search_index,
: ._design_order_by_index,
: ._design_group_by_index,
: ._design_join_optimization_index,
: ._design_full_text_search_index,
}
() -> [IndexRecommendation]:
recommendations = []
structure_recommendations = ._analyze_table_structure(table_info)
recommendations.extend(structure_recommendations)
query_recommendations = ._analyze_query_patterns(table_info, query_patterns)
recommendations.extend(query_recommendations)
existing_recommendations = ._analyze_existing_indexes(table_info)
recommendations.extend(existing_recommendations)
recommendations.sort(key= x: x.estimated_improvement / x.creation_cost, reverse=)
recommendations
() -> [IndexRecommendation]:
recommendations = []
table_info.primary_key:
primary_key_rec = ._design_primary_key_index(table_info, table_info.primary_key)
recommendations.append(primary_key_rec)
fk_column, ref_table table_info.foreign_keys.items():
fk_rec = ._design_foreign_key_index(table_info, [fk_column], ref_table)
recommendations.append(fk_rec)
column table_info.columns:
column.is_unique column.name table_info.primary_key:
unique_rec = ._design_unique_column_index(table_info, column)
recommendations.append(unique_rec)
recommendations
() -> [IndexRecommendation]:
recommendations = []
query query_patterns:
query_lower = query.lower()
query_lower query_lower:
equality_recommendations = ._analyze_equality_queries(table_info, query)
recommendations.extend(equality_recommendations)
(op query_lower op [, , , , , ]):
range_recommendations = ._analyze_range_queries(table_info, query)
recommendations.extend(range_recommendations)
query_lower:
order_recommendations = ._analyze_order_queries(table_info, query)
recommendations.extend(order_recommendations)
query_lower:
group_recommendations = ._analyze_group_queries(table_info, query)
recommendations.extend(group_recommendations)
query_lower:
join_recommendations = ._analyze_join_queries(table_info, query)
recommendations.extend(join_recommendations)
query_lower query_lower:
search_recommendations = ._analyze_search_queries(table_info, query)
recommendations.extend(search_recommendations)
recommendations
() -> [IndexRecommendation]:
recommendations = []
recommendations
() -> IndexRecommendation:
index_def = IndexDefinition(
name=,
table_name=table_info.name,
columns=columns,
index_type=IndexType.BTREE,
is_unique=,
is_partial=,
usage_types=[IndexUsage.EQUALITY, IndexUsage.JOIN]
)
IndexRecommendation(
definition=index_def,
reason=,
estimated_improvement=,
creation_cost=,
priority=
)
() -> IndexRecommendation:
index_def = IndexDefinition(
name=,
table_name=table_info.name,
columns=columns,
index_type=IndexType.BTREE,
is_unique=,
is_partial=,
usage_types=[IndexUsage.JOIN, IndexUsage.EQUALITY]
)
IndexRecommendation(
definition=index_def,
reason=,
estimated_improvement=,
creation_cost=,
priority=
)
() -> IndexRecommendation:
index_def = IndexDefinition(
name=,
table_name=table_info.name,
columns=[column.name],
index_type=IndexType.BTREE,
is_unique=,
is_partial=,
usage_types=[IndexUsage.EQUALITY]
)
IndexRecommendation(
definition=index_def,
reason=,
estimated_improvement=,
creation_cost=,
priority=
)
() -> [IndexRecommendation]:
recommendations = []
where_match = re.search(, query, re.IGNORECASE | re.DOTALL)
where_match:
where_clause = where_match.group()
equality_conditions = re.findall(, where_clause, re.IGNORECASE)
equality_conditions:
columns = [cond[] cond equality_conditions]
valid_columns = [col col columns col [c.name c table_info.columns]]
valid_columns:
index_def = IndexDefinition(
name=,
table_name=table_info.name,
columns=valid_columns,
index_type=IndexType.BTREE,
is_unique=,
is_partial=,
usage_types=[IndexUsage.EQUALITY]
)
recommendations.append(IndexRecommendation(
definition=index_def,
reason=,
estimated_improvement=,
creation_cost=,
priority=
))
recommendations
() -> [IndexRecommendation]:
recommendations = []
range_patterns = [
,
,
,
,
]
pattern range_patterns:
matches = re.findall(pattern, query, re.IGNORECASE)
matches:
column = [] (, )
column [c.name c table_info.columns]:
index_def = IndexDefinition(
name=,
table_name=table_info.name,
columns=[column],
index_type=IndexType.BTREE,
is_unique=,
is_partial=,
usage_types=[IndexUsage.RANGE]
)
recommendations.append(IndexRecommendation(
definition=index_def,
reason=,
estimated_improvement=,
creation_cost=,
priority=
))
recommendations
() -> [IndexRecommendation]:
recommendations = []
order_match = re.search(, query, re.IGNORECASE)
order_match:
order_clause = order_match.group()
order_columns = [col.strip().split()[] col order_clause.split()]
valid_columns = [col col order_columns col [c.name c table_info.columns]]
valid_columns:
index_def = IndexDefinition(
name=,
table_name=table_info.name,
columns=valid_columns,
index_type=IndexType.BTREE,
is_unique=,
is_partial=,
usage_types=[IndexUsage.ORDER]
)
recommendations.append(IndexRecommendation(
definition=index_def,
reason=,
estimated_improvement=,
creation_cost=,
priority=
))
recommendations
() -> [IndexRecommendation]:
recommendations = []
group_match = re.search(, query, re.IGNORECASE)
group_match:
group_clause = group_match.group()
group_columns = [col.strip() col group_clause.split()]
valid_columns = [col col group_columns col [c.name c table_info.columns]]
valid_columns:
index_def = IndexDefinition(
name=,
table_name=table_info.name,
columns=valid_columns,
index_type=IndexType.BTREE,
is_unique=,
is_partial=,
usage_types=[IndexUsage.GROUP]
)
recommendations.append(IndexRecommendation(
definition=index_def,
reason=,
estimated_improvement=,
creation_cost=,
priority=
))
recommendations
() -> [IndexRecommendation]:
recommendations = []
join_patterns = [
,
,
]
pattern join_patterns:
matches = re.findall(pattern, query, re.IGNORECASE)
matches:
() == :
table1, col1, table2, col2 =
table1.lower() == table_info.name.lower():
column = col1
table2.lower() == table_info.name.lower():
column = col2
:
column [c.name c table_info.columns]:
index_def = IndexDefinition(
name=,
table_name=table_info.name,
columns=[column],
index_type=IndexType.BTREE,
is_unique=,
is_partial=,
usage_types=[IndexUsage.JOIN]
)
recommendations.append(IndexRecommendation(
definition=index_def,
reason=,
estimated_improvement=,
creation_cost=,
priority=
))
recommendations
() -> [IndexRecommendation]:
recommendations = []
like_matches = re.findall(, query, re.IGNORECASE)
column like_matches:
column [c.name c table_info.columns]:
index_def = IndexDefinition(
name=,
table_name=table_info.name,
columns=[column],
index_type=IndexType.FULLTEXT,
is_unique=,
is_partial=,
usage_types=[IndexUsage.SEARCH]
)
recommendations.append(IndexRecommendation(
definition=index_def,
reason=,
estimated_improvement=,
creation_cost=,
priority=
))
recommendations
() -> :
script_lines = [, , ]
i, rec (recommendations, ):
script_lines.append()
script_lines.append()
script_lines.append()
script_lines.append()
index_def = rec.definition
columns_str = .join(index_def.columns)
index_def.is_unique:
create_stmt =
:
create_stmt =
create_stmt +=
index_def.is_partial index_def.where_condition:
create_stmt +=
create_stmt +=
script_lines.append(create_stmt)
script_lines.append()
.join(script_lines)
():
()
table_info = TableInfo(
name=,
row_count=,
columns=[
ColumnInfo(, , , , , , ),
ColumnInfo(, , , , , , ),
ColumnInfo(, , , , , , ),
ColumnInfo(, , , , , , ),
ColumnInfo(, , , , , , ),
ColumnInfo(, , , , , , ),
],
primary_key=[],
foreign_keys={: }
)
query_patterns = [
,
,
,
,
,
]
designer = IndexDesigner()
recommendations = designer.analyze_table_and_recommend_indexes(table_info, query_patterns)
()
()
()
()
()
i, rec (recommendations, ):
()
()
()
()
()
()
()
()
()
script = designer.generate_index_creation_script(recommendations)
(script)
__name__ == :
main()
SQL优化最佳实践
查询设计原则
- **避免SELECT ***: 只查询需要的列,减少数据传输
- 合理使用WHERE: 有效的过滤条件减少结果集
- 优化JOIN: 使用合适的连接类型和顺序
- 控制子查询: 避免过度嵌套,考虑使用JOIN
- 使用参数化: 防止SQL注入,提高执行效率
索引设计策略
- 选择合适的列: 高选择性、频繁查询的列
- 复合索引设计: 考虑查询顺序和列基数
- 避免过度索引: 平衡查询性能和写入成本
- 定期维护: 重建碎片化索引,更新统计信息
- 监控使用情况: 识别未使用和冗余索引
性能监控方法
- 执行计划分析: 理解查询执行路径
- 慢查询日志: 识别和优化性能问题
- 性能计数器: 监控数据库资源使用
- 查询分析工具: 使用专业工具分析性能
- 基准测试: 建立性能基准和回归测试
数据库配置优化
- 内存配置: 合理分配缓冲池和缓存
- 连接池设置: 优化连接数和超时设置
- 存储引擎选择: 根据场景选择合适引擎
- 日志配置: 平衡安全性和性能
- 参数调优: 根据工作负载调整参数
相关技能
- nosql-databases - NoSQL数据库应用
- backup-recovery - 备份与恢复
- migration-validator - 迁移验证
- transaction-management - 事务管理