SQL优化器

SQL优化器技能

概述

慢SQL查询会复合影响。一个慢查询会影响100个事务。在扩展之前先优化。需要建立完善的SQL查询分析和优化机制。

核心原则: 慢SQL查询会复合影响。一个慢查询会影响100个事务。在扩展之前先优化。

何时使用

始终:

• 优化慢查询
• 数据库性能问题
• 扩展数据库之前
• 性能分析后的查询调优
• 索引设计优化
• 查询性能监控

触发短语:

• "优化这个SQL查询"
• "查询太慢了"
• "如何优化数据库性能？"
• "检查SQL执行计划"
• "索引怎么设计？"
• "SQL性能分析"

SQL优化功能

查询分析

• 执行计划分析
• 查询复杂度评估
• 索引使用检查
• 统计信息分析
• 性能瓶颈识别

索引优化

• 索引设计建议
• 索引使用效率分析
• 冗余索引检测
• 索引优化方案
• 覆盖索引设计

性能调优

• 查询重写建议
• 表结构优化
• 分区策略分析
• 缓存优化建议
• 参数调优

常见SQL性能问题

缺少索引

Copy code
问题:
查询条件没有合适的索引支持

后果:
- 全表扫描
- 查询速度慢
- CPU占用高
- 锁竞争严重

解决方案:
- 创建合适的索引
- 复合索引优化
- 覆盖索引设计
- 索引选择性分析

查询设计不当

Copy code
问题:
SQL查询语句设计不合理

后果:
- 执行计划不优
- 资源浪费
- 性能下降
- 扩展性差

解决方案:
- 重写查询语句
- 避免子查询
- 优化JOIN操作
- 减少数据传输

统计信息过时

Copy code
问题:
数据库统计信息不准确

后果:
- 执行计划错误
- 索引选择不当
- 查询性能下降
- 资源分配不合理

解决方案:
- 更新统计信息
- 定期维护
- 自动统计收集
- 手动统计更新

SQL优化策略

查询重写技巧

Copy code
避免子查询:
- 使用JOIN替代子查询
- 使用EXISTS替代IN
- 优化相关子查询
- 减少嵌套层次

优化JOIN:
- 选择合适的JOIN类型
- 确保连接字段有索引
- 减少JOIN的数据量
- 优化JOIN顺序

索引设计原则

Copy code
单列索引:
- 高选择性字段
- 频繁查询条件
- 排序字段
- 外键字段

复合索引:
- 多条件查询
- 字段顺序重要
- 覆盖查询需求
- 减少回表查询

代码实现示例

SQL查询优化器

pythonCopy code
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 SQLOptimizer:
    """SQL查询优化器"""
    
    def __init__(self):
        self.optimization_rules = self._setup_optimization_rules()
        self.index_suggestions = []
        self.performance_metrics = {}
    
    def analyze_query(self, query: str) -> QueryAnalysis:
        """分析SQL查询"""
        # 格式化SQL
        formatted_query = sqlparse.format(query, reindent=True, keyword_case='upper')
        
        # 解析查询类型
        query_type = self._detect_query_type(formatted_query)
        
        # 提取查询组件
        tables = self._extract_tables(formatted_query)
        columns = self._extract_columns(formatted_query)
        joins = self._extract_joins(formatted_query)
        where_conditions = self._extract_where_conditions(formatted_query)
        group_by = self._extract_group_by(formatted_query)
        order_by = self._extract_order_by(formatted_query)
        
        # 计算复杂度
        complexity_score = self._calculate_complexity(
            tables, columns, joins, where_conditions, group_by, order_by
        )
        
        # 分析问题和建议
        issues = []
        recommendations = []
        
        self._analyze_performance_issues(
            query_type, tables, columns, joins, where_conditions, 
            group_by, order_by, issues, recommendations
        )
        
        return 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
        )
    
    def optimize_query(self, query: str) -> Dict[str, Any]:
        """优化SQL查询"""
        analysis = self.analyze_query(query)
        
        optimized_query = query
        optimizations_applied = []
        
        # 应用优化规则
        for rule in self.optimization_rules:
            result = rule.apply(optimized_query, analysis)
            if result.modified:
                optimized_query = result.query
                optimizations_applied.append(rule.name)
        
        # 生成索引建议
        index_suggestions = self._generate_index_suggestions(analysis)
        
        # 生成执行计划建议
        execution_plan_hints = self._generate_execution_plan_hints(analysis)
        
        return {
            'original_query': query,
            'optimized_query': optimized_query,
            'analysis': analysis,
            'optimizations_applied': optimizations_applied,
            'index_suggestions': index_suggestions,
            'execution_plan_hints': execution_plan_hints,
            'performance_improvement_estimate': self._estimate_improvement(analysis)
        }
    
    def _detect_query_type(self, query: str) -> QueryType:
        """检测查询类型"""
        query_upper = query.strip().upper()
        
        if query_upper.startswith('SELECT'):
            return QueryType.SELECT
        elif query_upper.startswith('INSERT'):
            return QueryType.INSERT
        elif query_upper.startswith('UPDATE'):
            return QueryType.UPDATE
        elif query_upper.startswith('DELETE'):
            return QueryType.DELETE
        elif query_upper.startswith('CREATE'):
            return QueryType.CREATE
        elif query_upper.startswith('DROP'):
            return QueryType.DROP
        elif query_upper.startswith('ALTER'):
            return QueryType.ALTER
        else:
            return QueryType.SELECT  # 默认
    
    def _extract_tables(self, query: str) -> List[str]:
        """提取表名"""
        tables = []
        
        # 使用正则表达式提取表名
        # FROM子句
        from_pattern = r'FROM\s+([a-zA-Z_][a-zA-Z0-9_]*)'
        from_matches = re.findall(from_pattern, query, re.IGNORECASE)
        tables.extend(from_matches)
        
        # JOIN子句
        join_pattern = r'JOIN\s+([a-zA-Z_][a-zA-Z0-9_]*)'
        join_matches = re.findall(join_pattern, query, re.IGNORECASE)
        tables.extend(join_matches)
        
        # INSERT INTO子句
        insert_pattern = r'INSERT\s+INTO\s+([a-zA-Z_][a-zA-Z0-9_]*)'
        insert_matches = re.findall(insert_pattern, query, re.IGNORECASE)
        tables.extend(insert_matches)
        
        # UPDATE子句
        update_pattern = r'UPDATE\s+([a-zA-Z_][a-zA-Z0-9_]*)'
        update_matches = re.findall(update_pattern, query, re.IGNORECASE)
        tables.extend(update_matches)
        
        return list(set(tables))  # 去重
    
    def _extract_columns(self, query: str) -> List[str]:
        """提取列名"""
        columns = []
        
        # SELECT子句中的列
        select_pattern = r'SELECT\s+(.*?)\s+FROM'
        select_match = re.search(select_pattern, query, re.IGNORECASE | re.DOTALL)
        if select_match:
            select_clause = select_match.group(1)
            
            # 排除函数和聚合
            if select_clause != '*':
                # 简单的列名提取
                column_pattern = r'([a-zA-Z_][a-zA-Z0-9_]*)\s*(?:,|$|\s+FROM)'
                column_matches = re.findall(column_pattern, select_clause)
                columns.extend(column_matches)
        
        return list(set(columns))
    
    def _extract_joins(self, query: str) -> List[Dict[str, str]]:
        """提取JOIN信息"""
        joins = []
        
        # JOIN模式
        join_pattern = r'(INNER|LEFT|RIGHT|FULL|CROSS)\s+JOIN\s+([a-zA-Z_][a-zA-Z0-9_]*)\s+ON\s+([^JOIN]+)'
        join_matches = re.findall(join_pattern, query, re.IGNORECASE)
        
        for join_type, table, condition in join_matches:
            joins.append({
                'type': join_type.upper(),
                'table': table,
                'condition': condition.strip()
            })
        
        return joins
    
    def _extract_where_conditions(self, query: str) -> List[str]:
        """提取WHERE条件"""
        conditions = []
        
        where_pattern = r'WHERE\s+(.*?)(?:\s+GROUP\s+BY|\s+ORDER\s+BY|\s+LIMIT|$)'
        where_match = re.search(where_pattern, query, re.IGNORECASE | re.DOTALL)
        
        if where_match:
            where_clause = where_match.group(1)
            # 分割AND和OR条件
            and_conditions = re.split(r'\s+AND\s+', where_clause, flags=re.IGNORECASE)
            for condition in and_conditions:
                or_conditions = re.split(r'\s+OR\s+', condition, flags=re.IGNORECASE)
                conditions.extend([c.strip() for c in or_conditions])
        
        return conditions
    
    def _extract_group_by(self, query: str) -> List[str]:
        """提取GROUP BY字段"""
        group_by = []
        
        group_pattern = r'GROUP\s+BY\s+(.*?)(?:\s+ORDER\s+BY|\s+LIMIT|$)'
        group_match = re.search(group_pattern, query, re.IGNORECASE)
        
        if group_match:
            group_clause = group_match.group(1)
            group_by = [col.strip() for col in group_clause.split(',')]
        
        return group_by
    
    def _extract_order_by(self, query: str) -> List[str]:
        """提取ORDER BY字段"""
        order_by = []
        
        order_pattern = r'ORDER\s+BY\s+(.*?)(?:\s+LIMIT|$)'
        order_match = re.search(order_pattern, query, re.IGNORECASE)
        
        if order_match:
            order_clause = order_match.group(1)
            order_by = [col.strip() for col in order_clause.split(',')]
        
        return order_by
    
    def _calculate_complexity(self, tables: List[str], columns: List[str], 
                            joins: List[Dict], where_conditions: List[str],
                            group_by: List[str], order_by: List[str]) -> int:
        """计算查询复杂度"""
        score = 0
        
        # 表数量影响
        score += len(tables) * 10
        
        # 列数量影响
        score += len(columns) * 2
        
        # JOIN影响
        score += len(joins) * 15
        
        # WHERE条件影响
        score += len(where_conditions) * 5
        
        # GROUP BY影响
        score += len(group_by) * 8
        
        # ORDER BY影响
        score += len(order_by) * 3
        
        # 子查询影响
        # 这里简化处理，实际应该解析嵌套层次
        
        return score
    
    def _analyze_performance_issues(self, query_type: QueryType, tables: List[str],
                                  columns: List[str], joins: List[Dict],
                                  where_conditions: List[str], group_by: List[str],
                                  order_by: List[str], issues: List[str],
                                  recommendations: List[str]):
        """分析性能问题"""
        
        # 检查SELECT *
        if query_type == QueryType.SELECT and '*' in columns:
            issues.append("查询使用了SELECT *，可能返回不必要的数据")
            recommendations.append("只查询需要的列，避免使用SELECT *")
        
        # 检查JOIN数量
        if len(joins) > 5:
            issues.append(f"查询包含过多JOIN操作({len(joins)}个)")
            recommendations.append("考虑分解复杂查询或优化表结构")
        
        # 检查WHERE条件
        if not where_conditions and query_type == QueryType.SELECT:
            issues.append("查询缺少WHERE条件，可能扫描全表")
            recommendations.append("添加适当的WHERE条件限制数据范围")
        
        # 检查GROUP BY和ORDER BY
        if group_by and order_by:
            if set(group_by) & set(order_by):
                recommendations.append("GROUP BY和ORDER BY使用相同字段，可以利用索引")
        
        # 检查表数量
        if len(tables) > 10:
            issues.append(f"查询涉及过多表({len(tables)}个)")
            recommendations.append("考虑使用视图或分解查询")
    
    def _generate_index_suggestions(self, analysis: QueryAnalysis) -> List[Dict[str, Any]]:
        """生成索引建议"""
        suggestions = []
        
        # 基于WHERE条件建议索引
        for condition in analysis.where_conditions:
            # 简单的字段提取
            field_pattern = r'([a-zA-Z_][a-zA-Z0-9_]*)\s*[=<>!]'
            field_matches = re.findall(field_pattern, condition)
            
            for field in field_matches:
                suggestions.append({
                    'type': 'single_column',
                    'table': analysis.tables[0] if analysis.tables else 'unknown',
                    'column': field,
                    'reason': f'WHERE条件中使用: {condition}',
                    'priority': 'high'
                })
        
        # 基于JOIN条件建议索引
        for join in analysis.joins:
            # 提取JOIN字段
            condition = join['condition']
            field_pattern = r'([a-zA-Z_][a-zA-Z0-9_]*)\s*=\s*([a-zA-Z_][a-zA-Z0-9_]*)\.([a-zA-Z_][a-zA-Z0-9_]*)'
            field_matches = re.findall(field_pattern, condition)
            
            for table1, table2, field in field_matches:
                suggestions.append({
                    'type': 'foreign_key',
                    'table': table1,
                    'column': field,
                    'reason': f'JOIN条件: {condition}',
                    'priority': 'high'
                })
        
        # 基于ORDER BY建议索引
        if analysis.order_by:
            for field in analysis.order_by:
                clean_field = field.split()[0]  # 移除ASC/DESC
                suggestions.append({
                    'type': 'order_by',
                    'table': analysis.tables[0] if analysis.tables else 'unknown',
                    'column': clean_field,
                    'reason': f'ORDER BY使用: {field}',
                    'priority': 'medium'
                })
        
        return suggestions
    
    def _generate_execution_plan_hints(self, analysis: QueryAnalysis) -> List[str]:
        """生成执行计划提示"""
        hints = []
        
        if analysis.complexity_score > 100:
            hints.append("查询复杂度较高，建议分解为多个简单查询")
        
        if len(analysis.joins) > 3:
            hints.append("考虑使用EXISTS替代部分JOIN操作")
        
        if analysis.group_by and len(analysis.group_by) > 3:
            hints.append("GROUP BY字段较多，考虑使用覆盖索引")
        
        return hints
    
    def _estimate_improvement(self, analysis: QueryAnalysis) -> Dict[str, float]:
        """估算性能提升"""
        improvement = {
            'query_speed': 0.0,
            'resource_usage': 0.0,
            'scalability': 0.0
        }
        
        # 基于问题数量估算
        issue_count = len(analysis.issues)
        if issue_count > 0:
            improvement['query_speed'] = min(0.8, issue_count * 0.15)
            improvement['resource_usage'] = min(0.6, issue_count * 0.12)
            improvement['scalability'] = min(0.4, issue_count * 0.08)
        
        return improvement
    
    def _setup_optimization_rules(self) -> List['OptimizationRule']:
        """设置优化规则"""
        return [
            SelectStarOptimization(),
            SubqueryOptimization(),
            JoinOptimization(),
            WhereOptimization(),
            IndexHintOptimization()
        ]

class OptimizationRule:
    """优化规则基类"""
    
    def __init__(self, name: str):
        self.name = name
    
    def apply(self, query: str, analysis: QueryAnalysis) -> 'OptimizationResult':
        """应用优化规则"""
        raise NotImplementedError

@dataclass
class OptimizationResult:
    """优化结果"""
    query: str
    modified: bool
    reason: str

class SelectStarOptimization(OptimizationRule):
    """SELECT *优化"""
    
    def __init__(self):
        super().__init__("SELECT *优化")
    
    def apply(self, query: str, analysis: QueryAnalysis) -> OptimizationResult:
        if '*' in analysis.columns:
            # 这里应该根据实际表结构替换*
            # 简化处理，提示用户手动优化
            return OptimizationResult(
                query=query,
                modified=False,
                reason="需要手动指定具体列名替代SELECT *"
            )
        
        return OptimizationResult(query=query, modified=False, reason="无需优化")

class SubqueryOptimization(OptimizationRule):
    """子查询优化"""
    
    def __init__(self):
        super().__init__("子查询优化")
    
    def apply(self, query: str, analysis: QueryAnalysis) -> OptimizationResult:
        # 检查是否可以转换为JOIN
        # 简化处理
        optimized_query = query
        
        # 检查IN子查询
        in_pattern = r'IN\s*\(\s*SELECT\s+.*?\)'
        if re.search(in_pattern, optimized_query, re.IGNORECASE | re.DOTALL):
            # 可以建议使用EXISTS或JOIN替代
            return OptimizationResult(
                query=optimized_query,
                modified=False,
                reason="建议使用EXISTS或JOIN替代IN子查询"
            )
        
        return OptimizationResult(query=optimized_query, modified=False, reason="无需优化")

class JoinOptimization(OptimizationRule):
    """JOIN优化"""
    
    def __init__(self):
        super().__init__("JOIN优化")
    
    def apply(self, query: str, analysis: QueryAnalysis) -> OptimizationResult:
        optimized_query = query
        
        # 检查JOIN顺序
        if len(analysis.joins) > 2:
            return OptimizationResult(
                query=optimized_query,
                modified=False,
                reason="建议优化JOIN顺序，小表在前"
            )
        
        return OptimizationResult(query=optimized_query, modified=False, reason="无需优化")

class WhereOptimization(OptimizationRule):
    """WHERE条件优化"""
    
    def __init__(self):
        super().__init__("WHERE条件优化")
    
    def apply(self, query: str, analysis: QueryAnalysis) -> OptimizationResult:
        optimized_query = query
        
        # 检查函数使用
        for condition in analysis.where_conditions:
            if re.search(r'[A-Z_]+\(', condition, re.IGNORECASE):
                return OptimizationResult(
                    query=optimized_query,
                    modified=False,
                    reason="WHERE条件中使用了函数，可能影响索引使用"
                )
        
        return OptimizationResult(query=optimized_query, modified=False, reason="无需优化")

class IndexHintOptimization(OptimizationRule):
    """索引提示优化"""
    
    def __init__(self):
        super().__init__("索引提示优化")
    
    def apply(self, query: str, analysis: QueryAnalysis) -> OptimizationResult:
        # 可以添加索引提示
        # 简化处理
        return OptimizationResult(
            query=query,
            modified=False,
            reason="可以考虑添加索引提示"
        )

# 使用示例
def main():
    optimizer = SQLOptimizer()
    
    # 示例查询
    sample_queries = [
        """
        SELECT u.name, o.order_date, p.product_name
        FROM users u
        INNER JOIN orders o ON u.id = o.user_id
        INNER JOIN order_items oi ON o.id = oi.order_id
        INNER JOIN products p ON oi.product_id = p.id
        WHERE u.status = 'active' AND o.order_date > '2023-01-01'
        ORDER BY o.order_date DESC
        """,
        
        """
        SELECT * FROM products WHERE price > 100
        """,
        
        """
        SELECT COUNT(*) FROM orders WHERE user_id IN (SELECT id FROM users WHERE status = 'active')
        """
    ]
    
    for i, query in enumerate(sample_queries, 1):
        print(f"\n=== 查询 {i} ===")
        print(f"原始查询: {query.strip()}")
        
        # 分析查询
        analysis = optimizer.analyze_query(query)
        print(f"查询类型: {analysis.query_type.value}")
        print(f"复杂度分数: {analysis.complexity_score}")
        print(f"涉及表: {analysis.tables}")
        print(f"涉及列: {analysis.columns}")
        
        if analysis.issues:
            print("发现的问题:")
            for issue in analysis.issues:
                print(f"  - {issue}")
        
        if analysis.recommendations:
            print("优化建议:")
            for rec in analysis.recommendations:
                print(f"  - {rec}")
        
        # 优化查询
        optimization = optimizer.optimize_query(query)
        print(f"优化后查询: {optimization['optimized_query'].strip()}")
        
        if optimization['index_suggestions']:
            print("索引建议:")
            for suggestion in optimization['index_suggestions']:
                print(f"  - {suggestion['table']}.{suggestion['column']} ({suggestion['type']}) - {suggestion['reason']}")
        
        improvement = optimization['performance_improvement_estimate']
        print(f"预计性能提升: 查询速度{improvement['query_speed']:.1%}, 资源使用{improvement['resource_usage']:.1%}")

if __name__ == "__main__":
    main()

数据库性能监控

pythonCopy code
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:
            with conn.cursor() as cursor:
                # 执行EXPLAIN ANALYZE
                explain_query = f"EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) {query}"
                cursor.execute(explain_query, params)
                explain_result = cursor.fetchone()[0]
                
                # 执行实际查询
                cursor.execute(query, params)
                rows = cursor.fetchall()
        
        end_time = time.time()
        
        # 解析执行计划
        plan = explain_result[0]
        
        metrics = PerformanceMetrics(
            query_time=end_time - start_time,
            rows_examined=plan.get('Actual Rows', 0),
            rows_returned=len(rows),
            index_usage=self._extract_index_usage(plan),
            buffer_usage=self._extract_buffer_usage(plan),
            cpu_usage=self._get_cpu_usage(),
            memory_usage=self._get_memory_usage()
        )
        
        self.metrics_history.append(metrics)
        
        # 检查是否为慢查询
        if metrics.query_time > 1.0:  # 超过1秒
            self.slow_queries.append({
                'query': query,
                'params': params,
                'metrics': metrics,
                'timestamp': datetime.now()
            })
        
        return metrics
    
    def _extract_index_usage(self, plan: Dict[str, Any]) -> Dict[str, int]:
        """提取索引使用情况"""
        index_usage = {}
        
        def traverse_plan(node):
            if 'Node Type' in node:
                node_type = node['Node Type']
                
                if 'Index Name' in node:
                    index_name = node['Index Name']
                    index_usage[index_name] = index_usage.get(index_name, 0) + 1
                
                if 'Plans' in node:
                    for sub_plan in node['Plans']:
                        traverse_plan(sub_plan)
        
        traverse_plan(plan)
        return index_usage
    
    def _extract_buffer_usage(self, plan: Dict[str, Any]) -> Dict[str, int]:
        """提取缓冲区使用情况"""
        buffers = plan.get('Buffers', {})
        
        return {
            'shared_hit': buffers.get('shared hit', 0),
            'shared_read': buffers.get('shared read', 0),
            'shared_dirtied': buffers.get('shared dirtied', 0),
            'shared_written': buffers.get('shared written', 0),
            'local_hit': buffers.get('local hit', 0),
            'local_read': buffers.get('local read', 0),
            'local_dirtied': buffers.get('local dirtied', 0),
            'local_written': buffers.get('local written', 0),
            'temp_read': buffers.get('temp read', 0),
            'temp_written': buffers.get('temp written', 0)
        }
    
    def _get_cpu_usage(self) -> float:
        """获取CPU使用率"""
        import psutil
        return psutil.cpu_percent()
    
    def _get_memory_usage(self) -> float:
        """获取内存使用率"""
        import psutil
        return psutil.virtual_memory().percent
    
    def analyze_performance_trends(self, hours: int = 24) -> Dict[str, Any]:
        """分析性能趋势"""
        cutoff_time = datetime.now() - timedelta(hours=hours)
        recent_metrics = [
            m for m in self.metrics_history 
            if datetime.now() - timedelta(hours=hours) <= datetime.now()
        ]
        
        if not recent_metrics:
            return {'message': '没有足够的数据进行分析'}
        
        query_times = [m.query_time for m in recent_metrics]
        cpu_usage = [m.cpu_usage for m in recent_metrics]
        memory_usage = [m.memory_usage for m in recent_metrics]
        
        analysis = {
            'period_hours': hours,
            'total_queries': len(recent_metrics),
            'avg_query_time': statistics.mean(query_times),
            'max_query_time': max(query_times),
            'min_query_time': min(query_times),
            'avg_cpu_usage': statistics.mean(cpu_usage),
            'avg_memory_usage': statistics.mean(memory_usage),
            'slow_query_count': len([m for m in recent_metrics if m.query_time > 1.0]),
            'trend': self._calculate_trend(query_times)
        }
        
        return analysis
    
    def _calculate_trend(self, values: List[float]) -> str:
        """计算趋势"""
        if len(values) < 2:
            return 'insufficient_data'
        
        # 简单的线性趋势计算
        x = list(range(len(values)))
        n = len(values)
        
        sum_x = sum(x)
        sum_y = sum(values)
        sum_xy = sum(x[i] * values[i] for i in range(n))
        sum_x2 = sum(x[i] ** 2 for i in range(n))
        
        # 计算斜率
        slope = (n * sum_xy - sum_x * sum_y) / (n * sum_x2 - sum_x ** 2)
        
        if slope > 0.01:
            return 'increasing'
        elif slope < -0.01:
            return 'decreasing'
        else:
            return 'stable'
    
    def generate_optimization_report(self) -> Dict[str, Any]:
        """生成优化报告"""
        if not self.metrics_history:
            return {'message': '没有性能数据'}
        
        # 分析慢查询
        slow_query_analysis = self._analyze_slow_queries()
        
        # 分析索引使用
        index_analysis = self._analyze_index_usage()
        
        # 分析缓冲区使用
        buffer_analysis = self._analyze_buffer_usage()
        
        # 生成建议
        recommendations = self._generate_recommendations(
            slow_query_analysis, index_analysis, buffer_analysis
        )
        
        return {
            'summary': {
                'total_queries': len(self.metrics_history),
                'slow_queries': len(self.slow_queries),
                'avg_query_time': statistics.mean([m.query_time for m in self.metrics_history])
            },
            'slow_queries': slow_query_analysis,
            'index_usage': index_analysis,
            'buffer_usage': buffer_analysis,
            'recommendations': recommendations
        }
    
    def _analyze_slow_queries(self) -> Dict[str, Any]:
        """分析慢查询"""
        if not self.slow_queries:
            return {'message': '没有慢查询'}
        
        slow_times = [sq['metrics'].query_time for sq in self.slow_queries]
        
        return {
            'count': len(self.slow_queries),
            'avg_time': statistics.mean(slow_times),
            'max_time': max(slow_times),
            'min_time': min(slow_times),
            'queries': [
                {
                    'query': sq['query'][:100] + '...',
                    'time': sq['metrics'].query_time,
                    'rows_examined': sq['metrics'].rows_examined,
                    'timestamp': sq['timestamp'].isoformat()
                }
                for sq in self.slow_queries[-5:]  # 最近5个慢查询
            ]
        }
    
    def _analyze_index_usage(self) -> Dict[str, Any]:
        """分析索引使用"""
        all_index_usage = {}
        
        for metrics in self.metrics_history:
            for index, count in metrics.index_usage.items():
                all_index_usage[index] = all_index_usage.get(index, 0) + count
        
        # 排序
        sorted_indexes = sorted(all_index_usage.items(), key=lambda x: x[1], reverse=True)
        
        return {
            'total_index_usage': sum(all_index_usage.values()),
            'unique_indexes': len(all_index_usage),
            'top_indexes': sorted_indexes[:10],
            'unused_indexes': []  # 需要与数据库中的索引列表比较
        }
    
    def _analyze_buffer_usage(self) -> Dict[str, Any]:
        """分析缓冲区使用"""
        total_shared_hit = sum(m.buffer_usage.get('shared_hit', 0) for m in self.metrics_history)
        total_shared_read = sum(m.buffer_usage.get('shared_read', 0) for m in self.metrics_history)
        
        hit_ratio = total_shared_hit / (total_shared_hit + total_shared_read) if (total_shared_hit + total_shared_read) > 0 else 0
        
        return {
            'shared_buffer_hit_ratio': hit_ratio,
            'total_shared_hit': total_shared_hit,
            'total_shared_read': total_shared_read,
            'cache_efficiency': 'good' if hit_ratio > 0.95 else 'poor'
        }
    
    def _generate_recommendations(self, slow_query_analysis: Dict, 
                                index_analysis: Dict, buffer_analysis: Dict) -> List[str]:
        """生成优化建议"""
        recommendations = []
        
        # 慢查询建议
        if slow_query_analysis.get('count', 0) > 0:
            recommendations.append(
                f"发现{slow_query_analysis['count']}个慢查询，平均执行时间{slow_query_analysis['avg_time']:.3f}秒"
            )
            recommendations.append("建议分析慢查询的执行计划并优化索引")
        
        # 缓冲区建议
        hit_ratio = buffer_analysis.get('shared_buffer_hit_ratio', 0)
        if hit_ratio < 0.95:
            recommendations.append(
                f"共享缓冲区命中率{hit_ratio:.1%}较低，建议增加shared_buffers配置"
            )
        
        # 索引建议
        if index_analysis.get('unique_indexes', 0) > 20:
            recommendations.append("索引数量较多，建议清理未使用的索引")
        
        return recommendations

# 使用示例
def main():
    monitor = DatabasePerformanceMonitor("postgresql://user:password@localhost/dbname")
    
    # 监控一些查询
    queries = [
        "SELECT * FROM users WHERE status = 'active'",
        "SELECT COUNT(*) FROM orders WHERE created_at > '2023-01-01'",
        "SELECT u.name, COUNT(o.id) FROM users u LEFT JOIN orders o ON u.id = o.user_id GROUP BY u.name"
    ]
    
    for query in queries:
        print(f"执行查询: {query}")
        metrics = monitor.monitor_query_performance(query)
        print(f"执行时间: {metrics.query_time:.3f}秒")
        print(f"检查行数: {metrics.rows_examined}")
        print(f"返回行数: {metrics.rows_returned}")
        print()
    
    # 生成报告
    report = monitor.generate_optimization_report()
    print("=== 性能优化报告 ===")
    print(f"总查询数: {report['summary']['total_queries']}")
    print(f"慢查询数: {report['summary']['slow_queries']}")
    print(f"平均执行时间: {report['summary']['avg_query_time']:.3f}秒")
    
    if report['recommendations']:
        print("\n优化建议:")
        for rec in report['recommendations']:
            print(f"- {rec}")

if __name__ == "__main__":
    main()

SQL优化最佳实践

查询设计原则

1. **避免SELECT ***: 只查询需要的列
2. 合理使用WHERE: 限制数据范围
3. 优化JOIN: 选择合适的JOIN类型和顺序
4. 使用索引: 为查询条件创建合适的索引

索引设计原则

1. 高选择性: 选择性高的字段适合建索引
2. 复合索引: 多条件查询使用复合索引
3. 覆盖索引: 包含查询所有字段的索引
4. 定期维护: 定期分析和重建索引

性能监控

1. 慢查询日志: 记录和分析慢查询
2. 执行计划: 分析查询执行计划
3. 性能指标: 监控关键性能指标
4. 趋势分析: 分析性能变化趋势

相关技能

• database-query-analyzer - 数据库查询分析
• api-validator - API接口验证和设计
• performance-profiler - 性能分析和监控
• database-design - 数据库设计优化