sql

name	SQL优化器
description	当优化SQL查询时，分析执行计划，检查索引使用，优化查询性能。验证SQL语法，分析查询复杂度，和最佳实践。
license	MIT

SQL优化器技能

概述

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

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

何时使用

始终:

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

触发短语:

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

SQL优化功能

查询分析

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

索引优化

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

性能调优

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

常见SQL性能问题

缺少索引

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

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

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

查询设计不当

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

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

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

统计信息过时

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

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

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

SQL优化策略

查询重写技巧

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

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

索引设计原则

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

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

代码实现示例

SQL查询优化器

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()

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 - 数据库设计优化