-- 創(chuàng)建測試表
CREATE TABLE user_orders (
    id INT PRIMARY KEY AUTO_INCREMENT,
    user_id VARCHAR(20) NOT NULL,
    order_amount DECIMAL(10,2),
    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_user_id (user_id)
);

-- 插入測試數(shù)據(jù)
INSERT INTO user_orders (user_id, order_amount) VALUES 
('U001', 299.99), ('U002', 199.50), ('U003', 399.00);

-- 問題查詢：使用數(shù)字查詢字符串字段
SELECT * FROM user_orders WHERE user_id = 1001;

-- 正確查詢：使用字符串查詢字符串字段
SELECT * FROM user_orders WHERE user_id = 'U001';

-- 分析問題查詢
EXPLAIN SELECT * FROM user_orders WHERE user_id = 1001;
-- 結(jié)果：type=ALL, key=NULL (全表掃描)

-- 分析正確查詢
EXPLAIN SELECT * FROM user_orders WHERE user_id = 'U001';
-- 結(jié)果：type=ref, key=idx_user_id (使用索引)

-- 創(chuàng)建類型轉(zhuǎn)換檢測函數(shù)
DELIMITER //
CREATE FUNCTION detect_type_conversion(
    table_name VARCHAR(64),
    column_name VARCHAR(64)
) RETURNS TEXT
READS SQL DATA
BEGIN
    DECLARE result TEXT DEFAULT '';
    DECLARE done INT DEFAULT FALSE;
    DECLARE query_text TEXT;
    
    -- 檢測常見的類型轉(zhuǎn)換問題
    SET result = CONCAT(
        'Column: ', column_name, 
        ' - Check for implicit conversions in WHERE clauses'
    );
    
    RETURN result;
END //
DELIMITER ;

-- 使用性能模式檢測類型轉(zhuǎn)換
SELECT 
    DIGEST_TEXT,
    COUNT_STAR,
    AVG_TIMER_WAIT/1000000000 as avg_time_sec
FROM performance_schema.events_statements_summary_by_digest 
WHERE DIGEST_TEXT LIKE '%WHERE%'
AND DIGEST_TEXT REGEXP '=[[:space:]]*[0-9]+[[:space:]]*'
ORDER BY AVG_TIMER_WAIT DESC
LIMIT 10;

-- 創(chuàng)建訂單表
CREATE TABLE orders (
    id INT PRIMARY KEY AUTO_INCREMENT,
    order_no VARCHAR(32) NOT NULL,
    create_time DATETIME NOT NULL,
    status TINYINT DEFAULT 1,
    INDEX idx_create_time (create_time),
    INDEX idx_order_no (order_no)
);

-- 插入測試數(shù)據(jù)
INSERT INTO orders (order_no, create_time, status) VALUES
('ORD20240101001', '2024-01-01 10:30:00', 1),
('ORD20240101002', '2024-01-01 14:20:00', 2),
('ORD20240102001', '2024-01-02 09:15:00', 1);

-- 錯誤用法：在索引字段上使用函數(shù)
SELECT * FROM orders WHERE DATE(create_time) = '2024-01-01';
SELECT * FROM orders WHERE UPPER(order_no) = 'ORD20240101001';
SELECT * FROM orders WHERE SUBSTRING(order_no, 1, 8) = 'ORD20240';

-- 正確用法：避免在索引字段上使用函數(shù)
SELECT * FROM orders 
WHERE create_time >= '2024-01-01 00:00:00' 
AND create_time < '2024-01-02 00:00:00';

SELECT * FROM orders WHERE order_no = 'ORD20240101001';
SELECT * FROM orders WHERE order_no LIKE 'ORD20240%';

-- MySQL 8.0+ 支持函數(shù)索引
ALTER TABLE orders ADD INDEX idx_date_create ((DATE(create_time)));
ALTER TABLE orders ADD INDEX idx_upper_order_no ((UPPER(order_no)));

-- 現(xiàn)在這些查詢可以使用索引了
SELECT * FROM orders WHERE DATE(create_time) = '2024-01-01';
SELECT * FROM orders WHERE UPPER(order_no) = 'ORD20240101001';

-- 對于MySQL 5.7及以下版本，使用虛擬列
ALTER TABLE orders ADD COLUMN create_date DATE 
GENERATED ALWAYS AS (DATE(create_time)) VIRTUAL;

ALTER TABLE orders ADD INDEX idx_create_date (create_date);

-- 查詢虛擬列
SELECT * FROM orders WHERE create_date = '2024-01-01';

-- 創(chuàng)建用戶行為表
CREATE TABLE user_behavior (
    id INT PRIMARY KEY AUTO_INCREMENT,
    user_id INT NOT NULL,
    action_type VARCHAR(20) NOT NULL,
    target_id INT NOT NULL,
    create_time DATETIME NOT NULL,
    -- 創(chuàng)建復合索引
    INDEX idx_user_action_time (user_id, action_type, create_time),
    INDEX idx_target_time (target_id, create_time)
);

-- 插入測試數(shù)據(jù)
INSERT INTO user_behavior (user_id, action_type, target_id, create_time) VALUES
(1001, 'view', 2001, '2024-01-01 10:00:00'),
(1001, 'click', 2002, '2024-01-01 11:00:00'),
(1002, 'view', 2001, '2024-01-01 12:00:00');

-- 能使用索引的查詢（遵循最左前綴）
SELECT * FROM user_behavior WHERE user_id = 1001;
SELECT * FROM user_behavior WHERE user_id = 1001 AND action_type = 'view';
SELECT * FROM user_behavior WHERE user_id = 1001 AND action_type = 'view' 
    AND create_time > '2024-01-01';

-- 不能使用索引的查詢（違背最左前綴）
SELECT * FROM user_behavior WHERE action_type = 'view';
SELECT * FROM user_behavior WHERE create_time > '2024-01-01';
SELECT * FROM user_behavior WHERE action_type = 'view' AND create_time > '2024-01-01';

-- 分析索引使用情況
EXPLAIN SELECT * FROM user_behavior WHERE user_id = 1001;
-- key: idx_user_action_time, key_len: 4 (只使用了user_id部分)

EXPLAIN SELECT * FROM user_behavior WHERE user_id = 1001 AND action_type = 'view';
-- key: idx_user_action_time, key_len: 86 (使用了user_id + action_type)

EXPLAIN SELECT * FROM user_behavior WHERE action_type = 'view';
-- key: NULL (索引失效，全表掃描)

-- 查看索引統(tǒng)計信息
SELECT 
    TABLE_NAME,
    INDEX_NAME,
    COLUMN_NAME,
    SEQ_IN_INDEX,
    CARDINALITY
FROM information_schema.STATISTICS 
WHERE TABLE_SCHEMA = DATABASE() 
AND TABLE_NAME = 'user_behavior'
ORDER BY INDEX_NAME, SEQ_IN_INDEX;

import pymysql
import json
from datetime import datetime

class IndexEffectivenessMonitor:
    def __init__(self, host, user, password, database):
        self.connection = pymysql.connect(
            host=host, user=user, password=password, 
            database=database, charset='utf8mb4'
        )
    
    def check_unused_indexes(self):
        """檢測未使用的索引"""
        query = """
        SELECT 
            t.TABLE_SCHEMA,
            t.TABLE_NAME,
            t.INDEX_NAME,
            t.COLUMN_NAME
        FROM information_schema.STATISTICS t
        LEFT JOIN performance_schema.table_io_waits_summary_by_index_usage p
            ON t.TABLE_SCHEMA = p.OBJECT_SCHEMA
            AND t.TABLE_NAME = p.OBJECT_NAME
            AND t.INDEX_NAME = p.INDEX_NAME
        WHERE t.TABLE_SCHEMA NOT IN ('mysql', 'information_schema', 'performance_schema')
            AND p.INDEX_NAME IS NULL
            AND t.INDEX_NAME != 'PRIMARY'
        ORDER BY t.TABLE_SCHEMA, t.TABLE_NAME, t.INDEX_NAME;
        """
        
        with self.connection.cursor() as cursor:
            cursor.execute(query)
            return cursor.fetchall()
    
    def analyze_slow_queries(self):
        """分析慢查詢中的索引使用情況"""
        query = """
        SELECT 
            DIGEST_TEXT,
            COUNT_STAR as execution_count,
            AVG_TIMER_WAIT/1000000000 as avg_time_seconds,
            SUM_ROWS_EXAMINED/COUNT_STAR as avg_rows_examined,
            SUM_ROWS_SENT/COUNT_STAR as avg_rows_sent
        FROM performance_schema.events_statements_summary_by_digest
        WHERE AVG_TIMER_WAIT > 1000000000  -- 超過1秒的查詢
        ORDER BY AVG_TIMER_WAIT DESC
        LIMIT 20;
        """
        
        with self.connection.cursor() as cursor:
            cursor.execute(query)
            return cursor.fetchall()
    
    def generate_optimization_report(self):
        """生成優(yōu)化報告"""
        unused_indexes = self.check_unused_indexes()
        slow_queries = self.analyze_slow_queries()
        
        report = {
            'timestamp': datetime.now().isoformat(),
            'unused_indexes': unused_indexes,
            'slow_queries': slow_queries,
            'recommendations': self._generate_recommendations(unused_indexes, slow_queries)
        }
        
        return json.dumps(report, indent=2, ensure_ascii=False)
    
    def _generate_recommendations(self, unused_indexes, slow_queries):
        """生成優(yōu)化建議"""
        recommendations = []
        
        if unused_indexes:
            recommendations.append({
                'type': 'unused_indexes',
                'message': f'發(fā)現(xiàn) {len(unused_indexes)} 個未使用的索引，建議刪除以節(jié)省存儲空間',
                'action': 'DROP INDEX'
            })
        
        for query in slow_queries:
            if query[3] > 1000:  # 平均掃描行數(shù)超過1000
                recommendations.append({
                    'type': 'missing_index',
                    'message': f'查詢掃描行數(shù)過多: {query[3]:.0f}',
                    'query': query[0][:100] + '...',
                    'action': 'ADD INDEX'
                })
        
        return recommendations

# 使用示例
monitor = IndexEffectivenessMonitor('localhost', 'root', 'password', 'testdb')
report = monitor.generate_optimization_report()
print(report)

-- 創(chuàng)建性能測試存儲過程
DELIMITER //
CREATE PROCEDURE benchmark_index_performance()
BEGIN
    DECLARE i INT DEFAULT 1;
    DECLARE start_time TIMESTAMP;
    DECLARE end_time TIMESTAMP;
    
    -- 清空查詢緩存
    RESET QUERY CACHE;
    
    -- 測試有索引的查詢
    SET start_time = NOW(6);
    WHILE i <= 1000 DO
        SELECT COUNT(*) FROM user_behavior WHERE user_id = i;
        SET i = i + 1;
    END WHILE;
    SET end_time = NOW(6);
    
    SELECT 'With Index' as test_type, 
           TIMESTAMPDIFF(MICROSECOND, start_time, end_time) as execution_time_microseconds;
    
    -- 測試無索引的查詢（臨時刪除索引）
    DROP INDEX idx_user_action_time ON user_behavior;
    
    SET i = 1;
    SET start_time = NOW(6);
    WHILE i <= 1000 DO
        SELECT COUNT(*) FROM user_behavior WHERE user_id = i;
        SET i = i + 1;
    END WHILE;
    SET end_time = NOW(6);
    
    SELECT 'Without Index' as test_type,
           TIMESTAMPDIFF(MICROSECOND, start_time, end_time) as execution_time_microseconds;
    
    -- 重新創(chuàng)建索引
    CREATE INDEX idx_user_action_time ON user_behavior (user_id, action_type, create_time);
END //
DELIMITER ;

-- 執(zhí)行性能測試
CALL benchmark_index_performance();

-- 1. 選擇性原則：優(yōu)先為高選擇性字段創(chuàng)建索引
SELECT 
    COLUMN_NAME,
    COUNT(DISTINCT COLUMN_NAME) / COUNT(*) as selectivity
FROM information_schema.COLUMNS c
JOIN your_table t ON 1=1
WHERE c.TABLE_NAME = 'your_table'
GROUP BY COLUMN_NAME
ORDER BY selectivity DESC;

-- 2. 覆蓋索引原則：讓索引包含查詢所需的所有字段
CREATE INDEX idx_covering ON orders (user_id, status, create_time, order_amount);

-- 3. 前綴索引原則：對于長字符串字段使用前綴索引
CREATE INDEX idx_order_no_prefix ON orders (order_no(10));

-- 4. 索引合并原則：避免創(chuàng)建過多單列索引
-- 錯誤做法
CREATE INDEX idx_user_id ON orders (user_id);
CREATE INDEX idx_status ON orders (status);
CREATE INDEX idx_create_time ON orders (create_time);

-- 正確做法
CREATE INDEX idx_user_status_time ON orders (user_id, status, create_time);

# MySQL性能監(jiān)控配置示例
mysql_monitoring:
  slow_query_threshold: 2.0  # 慢查詢閾值（秒）
  
  alerts:
    - name: "慢查詢數(shù)量告警"
      condition: "slow_queries_per_minute > 10"
      severity: "warning"
      
    - name: "索引失效告警"
      condition: "full_table_scans_per_minute > 5"
      severity: "critical"
      
    - name: "連接數(shù)告警"
      condition: "active_connections > 80% of max_connections"
      severity: "warning"
  
  metrics:
    - query_response_time
    - index_usage_ratio
    - table_scan_ratio
    - connection_utilization
    - buffer_pool_hit_ratio

-- 1. 分析慢查詢?nèi)罩?
SELECT 
    sql_text,
    exec_count,
    avg_timer_wait/1000000000 as avg_time_sec,
    sum_rows_examined/exec_count as avg_rows_examined
FROM performance_schema.events_statements_summary_by_digest
WHERE avg_timer_wait > 1000000000
ORDER BY avg_timer_wait DESC;

-- 2. 發(fā)現(xiàn)問題SQL
SELECT o.*, u.username, p.product_name 
FROM orders o
JOIN users u ON o.user_id = u.id
JOIN order_items oi ON o.id = oi.order_id
JOIN products p ON oi.product_id = p.id
WHERE DATE(o.create_time) = '2024-01-15'
AND o.status IN (1, 2, 3)
AND u.user_type = 'VIP';

-- 3. 分析執(zhí)行計劃
EXPLAIN SELECT o.*, u.username, p.product_name 
FROM orders o
JOIN users u ON o.user_id = u.id
JOIN order_items oi ON o.id = oi.order_id
JOIN products p ON oi.product_id = p.id
WHERE DATE(o.create_time) = '2024-01-15'
AND o.status IN (1, 2, 3)
AND u.user_type = 'VIP';

-- 優(yōu)化前的索引結(jié)構(gòu)
SHOW INDEX FROM orders;
SHOW INDEX FROM users;
SHOW INDEX FROM order_items;
SHOW INDEX FROM products;

-- 優(yōu)化方案1：修復函數(shù)索引問題
-- 原問題：WHERE DATE(o.create_time) = '2024-01-15'
-- 解決方案：改為范圍查詢
SELECT o.*, u.username, p.product_name 
FROM orders o
JOIN users u ON o.user_id = u.id
JOIN order_items oi ON o.id = oi.order_id
JOIN products p ON oi.product_id = p.id
WHERE o.create_time >= '2024-01-15 00:00:00'
AND o.create_time < '2024-01-16 00:00:00'
AND o.status IN (1, 2, 3)
AND u.user_type = 'VIP';

-- 優(yōu)化方案2：創(chuàng)建復合索引
CREATE INDEX idx_orders_time_status ON orders (create_time, status);
CREATE INDEX idx_users_type ON users (user_type);
CREATE INDEX idx_order_items_order_product ON order_items (order_id, product_id);

-- 優(yōu)化方案3：查詢重寫，減少JOIN
-- 分步查詢，先篩選再關(guān)聯(lián)
SELECT o.id, o.user_id, o.create_time, o.status
FROM orders o
WHERE o.create_time >= '2024-01-15 00:00:00'
AND o.create_time < '2024-01-16 00:00:00'
AND o.status IN (1, 2, 3);

-- 然后基于結(jié)果進行關(guān)聯(lián)查詢
SELECT o.*, u.username, p.product_name 
FROM (
    SELECT * FROM orders 
    WHERE create_time >= '2024-01-15 00:00:00'
    AND create_time < '2024-01-16 00:00:00'
    AND status IN (1, 2, 3)
    LIMIT 1000
) o
JOIN users u ON o.user_id = u.id AND u.user_type = 'VIP'
JOIN order_items oi ON o.id = oi.order_id
JOIN products p ON oi.product_id = p.id;

-- 創(chuàng)建分區(qū)表
CREATE TABLE orders_partitioned (
    id INT PRIMARY KEY AUTO_INCREMENT,
    user_id INT NOT NULL,
    order_amount DECIMAL(10,2),
    create_time DATETIME NOT NULL,
    status TINYINT DEFAULT 1,
    INDEX idx_user_status (user_id, status),
    INDEX idx_create_time (create_time)
) PARTITION BY RANGE (YEAR(create_time)) (
    PARTITION p2022 VALUES LESS THAN (2023),
    PARTITION p2023 VALUES LESS THAN (2024),
    PARTITION p2024 VALUES LESS THAN (2025),
    PARTITION p2025 VALUES LESS THAN (2026)
);

-- 分區(qū)裁剪查詢
SELECT * FROM orders_partitioned 
WHERE create_time >= '2024-01-01' 
AND create_time < '2024-02-01'
AND user_id = 1001;

class DatabaseRouter:
    def __init__(self):
        self.master_db = self._connect_master()
        self.slave_dbs = [self._connect_slave(i) for i in range(3)]
        self.current_slave = 0
    
    def execute_read_query(self, query, use_index_hint=True):
        """執(zhí)行讀查詢，自動選擇從庫"""
        if use_index_hint:
            query = self._add_index_hints(query)
        
        slave_db = self._get_next_slave()
        return slave_db.execute(query)
    
    def execute_write_query(self, query):
        """執(zhí)行寫查詢，使用主庫"""
        return self.master_db.execute(query)
    
    def _add_index_hints(self, query):
        """自動添加索引提示"""
        # 分析查詢并添加適當?shù)乃饕崾?
        if 'WHERE user_id' in query:
            query = query.replace('FROM orders', 'FROM orders USE INDEX (idx_user_id)')
        return query
    
    def _get_next_slave(self):
        """輪詢選擇從庫"""
        slave = self.slave_dbs[self.current_slave]
        self.current_slave = (self.current_slave + 1) % len(self.slave_dbs)
        return slave

-- 創(chuàng)建索引推薦分析視圖
CREATE VIEW index_recommendation AS
SELECT 
    t.TABLE_SCHEMA,
    t.TABLE_NAME,
    GROUP_CONCAT(DISTINCT 
        CASE WHEN s.COLUMN_NAME IS NOT NULL 
        THEN CONCAT('WHERE ', s.COLUMN_NAME) END
    ) as missing_indexes,
    COUNT(DISTINCT s.DIGEST) as query_count,
    AVG(s.AVG_TIMER_WAIT)/1000000000 as avg_time_seconds
FROM information_schema.TABLES t
JOIN performance_schema.events_statements_summary_by_digest s
    ON s.DIGEST_TEXT LIKE CONCAT('%', t.TABLE_NAME, '%')
WHERE t.TABLE_SCHEMA NOT IN ('mysql', 'information_schema', 'performance_schema')
    AND s.AVG_TIMER_WAIT > 1000000000
    AND s.DIGEST_TEXT REGEXP 'WHERE.*='
GROUP BY t.TABLE_SCHEMA, t.TABLE_NAME
HAVING query_count > 10
ORDER BY avg_time_seconds DESC;

-- 查看推薦結(jié)果
SELECT * FROM index_recommendation LIMIT 10;

class MySQLOptimizationAdvisor:
    def __init__(self, db_connection):
        self.db = db_connection
        self.rules = self._load_optimization_rules()
    
    def analyze_and_recommend(self):
        """分析數(shù)據(jù)庫并提供優(yōu)化建議"""
        recommendations = []
        
        # 檢查慢查詢
        slow_queries = self._get_slow_queries()
        for query in slow_queries:
            if self._has_function_in_where(query['sql']):
                recommendations.append({
                    'type': 'function_optimization',
                    'priority': 'high',
                    'description': '檢測到WHERE子句中使用函數(shù)，建議重寫查詢',
                    'sql': query['sql'],
                    'suggestion': self._suggest_function_fix(query['sql'])
                })
        
        # 檢查未使用的索引
        unused_indexes = self._get_unused_indexes()
        if unused_indexes:
            recommendations.append({
                'type': 'unused_index_cleanup',
                'priority': 'medium',
                'description': f'發(fā)現(xiàn){len(unused_indexes)}個未使用的索引',
                'indexes': unused_indexes,
                'suggestion': 'DROP INDEX statements'
            })
        
        # 檢查缺失的索引
        missing_indexes = self._suggest_missing_indexes()
        for suggestion in missing_indexes:
            recommendations.append({
                'type': 'missing_index',
                'priority': 'high',
                'description': '建議創(chuàng)建索引以優(yōu)化查詢性能',
                'table': suggestion['table'],
                'columns': suggestion['columns'],
                'suggestion': suggestion['create_sql']
            })
        
        return self._prioritize_recommendations(recommendations)
    
    def _suggest_function_fix(self, sql):
        """建議函數(shù)查詢的修復方案"""
        fixes = {
            'DATE(': '使用范圍查詢替代DATE()函數(shù)',
            'UPPER(': '考慮創(chuàng)建函數(shù)索引或使用COLLATE',
            'SUBSTRING(': '使用LIKE操作符或前綴索引'
        }
        
        for func, suggestion in fixes.items():
            if func in sql:
                return suggestion
        
        return '重寫查詢以避免在索引字段上使用函數(shù)'