第164集MySQL千万级表变更字段不锁表Java实战

/**
 * MySQL表变更配置类
 * @author 运维实战
 */
@Configuration
@EnableConfigurationProperties(MySQLTableAlterProperties.class)
public class MySQLTableAlterConfig {
    
    @Autowired
    private MySQLTableAlterProperties properties;
    
    /**
     * MySQL表变更服务
     * @return MySQL表变更服务
     */
    @Bean
    public MySQLTableAlterService mysqlTableAlterService() {
        return new MySQLTableAlterService();
    }
    
    /**
     * MySQL在线DDL服务
     * @return MySQL在线DDL服务
     */
    @Bean
    public MySQLOnlineDDLService mysqlOnlineDDLService() {
        return new MySQLOnlineDDLService();
    }
    
    /**
     * MySQL表结构变更服务
     * @return MySQL表结构变更服务
     */
    @Bean
    public MySQLTableStructureService mysqlTableStructureService() {
        return new MySQLTableStructureService();
    }
    
    /**
     * MySQL数据迁移服务
     * @return MySQL数据迁移服务
     */
    @Bean
    public MySQLDataMigrationService mysqlDataMigrationService() {
        return new MySQLDataMigrationService();
    }
    
    /**
     * MySQL表变更监控服务
     * @return MySQL表变更监控服务
     */
    @Bean
    public MySQLTableAlterMonitorService mysqlTableAlterMonitorService() {
        return new MySQLTableAlterMonitorService();
    }
    
    private static final Logger logger = LoggerFactory.getLogger(MySQLTableAlterConfig.class);
}

/**
 * MySQL表变更属性配置
 * @author 运维实战
 */
@Data
@ConfigurationProperties(prefix = "mysql.table-alter")
public class MySQLTableAlterProperties {
    
    /**
     * 是否启用MySQL表变更
     */
    private boolean enableTableAlter = true;
    
    /**
     * MySQL连接配置
     */
    private DatabaseConfig database = new DatabaseConfig();
    
    /**
     * 表变更策略
     */
    private String alterStrategy = "ONLINE_DDL"; // ONLINE_DDL, PT_ONLINE_SCHEMA_CHANGE, GH_OST
    
    /**
     * 是否启用在线DDL
     */
    private boolean enableOnlineDDL = true;
    
    /**
     * 在线DDL超时时间（秒）
     */
    private long onlineDDLTimeout = 3600;
    
    /**
     * 是否启用pt-online-schema-change
     */
    private boolean enablePtOnlineSchemaChange = false;
    
    /**
     * pt-online-schema-change路径
     */
    private String ptOnlineSchemaChangePath = "/usr/bin/pt-online-schema-change";
    
    /**
     * pt-online-schema-change超时时间（秒）
     */
    private long ptOnlineSchemaChangeTimeout = 7200;
    
    /**
     * 是否启用gh-ost
     */
    private boolean enableGhOst = false;
    
    /**
     * gh-ost路径
     */
    private String ghOstPath = "/usr/bin/gh-ost";
    
    /**
     * gh-ost超时时间（秒）
     */
    private long ghOstTimeout = 7200;
    
    /**
     * 是否启用数据迁移
     */
    private boolean enableDataMigration = true;
    
    /**
     * 数据迁移批次大小
     */
    private int dataMigrationBatchSize = 1000;
    
    /**
     * 数据迁移间隔（毫秒）
     */
    private long dataMigrationInterval = 100;
    
    /**
     * 是否启用变更回滚
     */
    private boolean enableRollback = true;
    
    /**
     * 回滚超时时间（秒）
     */
    private long rollbackTimeout = 1800;
    
    /**
     * 是否启用性能监控
     */
    private boolean enablePerformanceMonitoring = true;
    
    /**
     * 性能监控间隔（毫秒）
     */
    private long performanceMonitoringInterval = 30000;
    
    /**
     * 是否启用监控
     */
    private boolean enableMonitoring = true;
    
    /**
     * 监控间隔（毫秒）
     */
    private long monitorInterval = 30000;
    
    /**
     * 是否启用告警
     */
    private boolean enableAlert = true;
    
    /**
     * 表变更超时告警阈值（秒）
     */
    private long tableAlterTimeoutAlertThreshold = 3600;
    
    /**
     * 数据迁移失败告警阈值
     */
    private int dataMigrationFailureAlertThreshold = 5;
    
    /**
     * 表变更失败告警阈值
     */
    private int tableAlterFailureAlertThreshold = 3;
    
    /**
     * 数据库配置类
     */
    @Data
    public static class DatabaseConfig {
        /**
         * 数据库URL
         */
        private String url = "jdbc:mysql://localhost:3306/test";
        
        /**
         * 数据库用户名
         */
        private String username = "root";
        
        /**
         * 数据库密码
         */
        private String password = "password";
        
        /**
         * 数据库驱动
         */
        private String driverClassName = "com.mysql.cj.jdbc.Driver";
        
        /**
         * 连接池配置
         */
        private ConnectionPoolConfig connectionPool = new ConnectionPoolConfig();
    }
    
    /**
     * 连接池配置类
     */
    @Data
    public static class ConnectionPoolConfig {
        /**
         * 最小连接数
         */
        private int minIdle = 5;
        
        /**
         * 最大连接数
         */
        private int maxActive = 20;
        
        /**
         * 连接超时时间（毫秒）
         */
        private long connectionTimeout = 30000;
        
        /**
         * 空闲连接超时时间（毫秒）
         */
        private long idleTimeout = 600000;
        
        /**
         * 最大生命周期（毫秒）
         */
        private long maxLifetime = 1800000;
    }
}

/**
 * 表变更数据模型类
 * @author 运维实战
 */
@Data
public class TableAlterRequest {
    
    /**
     * 变更ID
     */
    private String alterId;
    
    /**
     * 数据库名称
     */
    private String databaseName;
    
    /**
     * 表名称
     */
    private String tableName;
    
    /**
     * 变更类型
     */
    private String alterType; // ADD_COLUMN, DROP_COLUMN, MODIFY_COLUMN, ADD_INDEX, DROP_INDEX
    
    /**
     * 变更SQL
     */
    private String alterSql;
    
    /**
     * 变更描述
     */
    private String description;
    
    /**
     * 变更策略
     */
    private String strategy; // ONLINE_DDL, PT_ONLINE_SCHEMA_CHANGE, GH_OST
    
    /**
     * 变更状态
     */
    private String status; // PENDING, RUNNING, COMPLETED, FAILED, ROLLBACK
    
    /**
     * 开始时间
     */
    private Long startTime;
    
    /**
     * 结束时间
     */
    private Long endTime;
    
    /**
     * 错误信息
     */
    private String errorMessage;
    
    /**
     * 变更参数
     */
    private Map<String, Object> parameters;
    
    /**
     * 扩展属性
     */
    private Map<String, Object> extendedProperties;
    
    public TableAlterRequest() {
        this.alterId = UUID.randomUUID().toString();
        this.status = "PENDING";
        this.parameters = new HashMap<>();
        this.extendedProperties = new HashMap<>();
    }
    
    public TableAlterRequest(String databaseName, String tableName, String alterType, String alterSql) {
        this();
        this.databaseName = databaseName;
        this.tableName = tableName;
        this.alterType = alterType;
        this.alterSql = alterSql;
    }
    
    /**
     * 验证表变更请求
     * @return 是否有效
     */
    public boolean validate() {
        if (alterId == null || alterId.isEmpty()) {
            return false;
        }
        
        if (databaseName == null || databaseName.isEmpty()) {
            return false;
        }
        
        if (tableName == null || tableName.isEmpty()) {
            return false;
        }
        
        if (alterType == null || alterType.isEmpty()) {
            return false;
        }
        
        if (alterSql == null || alterSql.isEmpty()) {
            return false;
        }
        
        return true;
    }
    
    /**
     * 计算变更耗时
     * @return 变更耗时（毫秒）
     */
    public long getDuration() {
        if (startTime != null && endTime != null) {
            return endTime - startTime;
        }
        return 0;
    }
    
    /**
     * 是否完成
     * @return 是否完成
     */
    public boolean isCompleted() {
        return "COMPLETED".equals(status);
    }
    
    /**
     * 是否失败
     * @return 是否失败
     */
    public boolean isFailed() {
        return "FAILED".equals(status);
    }
    
    /**
     * 添加变更参数
     * @param key 键
     * @param value 值
     */
    public void addParameter(String key, Object value) {
        if (parameters == null) {
            parameters = new HashMap<>();
        }
        parameters.put(key, value);
    }
    
    /**
     * 添加扩展属性
     * @param key 键
     * @param value 值
     */
    public void addExtendedProperty(String key, Object value) {
        if (extendedProperties == null) {
            extendedProperties = new HashMap<>();
        }
        extendedProperties.put(key, value);
    }
    
    /**
     * 获取扩展属性
     * @param key 键
     * @return 值
     */
    public Object getExtendedProperty(String key) {
        return extendedProperties != null ? extendedProperties.get(key) : null;
    }
}

/**
 * 表变更结果类
 * @author 运维实战
 */
@Data
public class TableAlterResult {
    
    private boolean success;
    private String alterId;
    private String databaseName;
    private String tableName;
    private String alterType;
    private String status;
    private String error;
    private long startTime;
    private long endTime;
    
    public TableAlterResult() {
        this.success = false;
    }
    
    public TableAlterResult(String alterId, String databaseName, String tableName, String alterType) {
        this();
        this.alterId = alterId;
        this.databaseName = databaseName;
        this.tableName = tableName;
        this.alterType = alterType;
    }
    
    /**
     * 获取变更耗时
     * @return 变更耗时（毫秒）
     */
    public long getDuration() {
        return endTime - startTime;
    }
    
    /**
     * 是否成功
     * @return 是否成功
     */
    public boolean isSuccess() {
        return success;
    }
}

/**
 * 表结构信息类
 * @author 运维实战
 */
@Data
public class TableStructureInfo {
    
    /**
     * 数据库名称
     */
    private String databaseName;
    
    /**
     * 表名称
     */
    private String tableName;
    
    /**
     * 表引擎
     */
    private String engine;
    
    /**
     * 表字符集
     */
    private String charset;
    
    /**
     * 表排序规则
     */
    private String collation;
    
    /**
     * 表行数
     */
    private Long rowCount;
    
    /**
     * 表大小（字节）
     */
    private Long tableSize;
    
    /**
     * 索引大小（字节）
     */
    private Long indexSize;
    
    /**
     * 字段信息
     */
    private List<ColumnInfo> columns;
    
    /**
     * 索引信息
     */
    private List<IndexInfo> indexes;
    
    /**
     * 创建时间
     */
    private Long createTime;
    
    /**
     * 更新时间
     */
    private Long updateTime;
    
    public TableStructureInfo() {
        this.columns = new ArrayList<>();
        this.indexes = new ArrayList<>();
    }
    
    public TableStructureInfo(String databaseName, String tableName) {
        this();
        this.databaseName = databaseName;
        this.tableName = tableName;
    }
    
    /**
     * 验证表结构信息
     * @return 是否有效
     */
    public boolean validate() {
        if (databaseName == null || databaseName.isEmpty()) {
            return false;
        }
        
        if (tableName == null || tableName.isEmpty()) {
            return false;
        }
        
        return true;
    }
    
    /**
     * 添加字段信息
     * @param columnInfo 字段信息
     */
    public void addColumn(ColumnInfo columnInfo) {
        if (columns == null) {
            columns = new ArrayList<>();
        }
        columns.add(columnInfo);
    }
    
    /**
     * 添加索引信息
     * @param indexInfo 索引信息
     */
    public void addIndex(IndexInfo indexInfo) {
        if (indexes == null) {
            indexes = new ArrayList<>();
        }
        indexes.add(indexInfo);
    }
}

/**
 * 字段信息类
 * @author 运维实战
 */
@Data
public class ColumnInfo {
    
    /**
     * 字段名称
     */
    private String columnName;
    
    /**
     * 字段类型
     */
    private String columnType;
    
    /**
     * 是否允许NULL
     */
    private Boolean nullable;
    
    /**
     * 默认值
     */
    private String defaultValue;
    
    /**
     * 字段注释
     */
    private String comment;
    
    /**
     * 字段位置
     */
    private Integer position;
    
    public ColumnInfo() {}
    
    public ColumnInfo(String columnName, String columnType) {
        this.columnName = columnName;
        this.columnType = columnType;
    }
}

/**
 * 索引信息类
 * @author 运维实战
 */
@Data
public class IndexInfo {
    
    /**
     * 索引名称
     */
    private String indexName;
    
    /**
     * 索引类型
     */
    private String indexType;
    
    /**
     * 索引字段
     */
    private List<String> columns;
    
    /**
     * 是否唯一索引
     */
    private Boolean unique;
    
    /**
     * 索引注释
     */
    private String comment;
    
    public IndexInfo() {
        this.columns = new ArrayList<>();
    }
    
    public IndexInfo(String indexName, String indexType) {
        this();
        this.indexName = indexName;
        this.indexType = indexType;
    }
}

/**
 * 基础MySQL表变更服务
 * @author 运维实战
 */
@Service
public class MySQLTableAlterService {
    
    @Autowired
    private MySQLTableAlterProperties properties;
    
    @Autowired
    private DataSource dataSource;
    
    @Autowired
    private MySQLOnlineDDLService mysqlOnlineDDLService;
    
    @Autowired
    private MySQLTableStructureService mysqlTableStructureService;
    
    @Autowired
    private MySQLDataMigrationService mysqlDataMigrationService;
    
    @Autowired
    private MySQLTableAlterMonitorService mysqlTableAlterMonitorService;
    
    private static final Logger logger = LoggerFactory.getLogger(MySQLTableAlterService.class);
    
    /**
     * 执行表变更
     * @param alterRequest 表变更请求
     * @return 表变更结果
     */
    public TableAlterResult executeTableAlter(TableAlterRequest alterRequest) {
        logger.info("执行表变更，变更ID: {}, 表: {}.{}", 
            alterRequest.getAlterId(), alterRequest.getDatabaseName(), alterRequest.getTableName());
        
        TableAlterResult result = new TableAlterResult(
            alterRequest.getAlterId(), alterRequest.getDatabaseName(), 
            alterRequest.getTableName(), alterRequest.getAlterType());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 验证表变更请求
            if (!alterRequest.validate()) {
                result.setSuccess(false);
                result.setError("表变更请求验证失败");
                result.setEndTime(System.currentTimeMillis());
                return result;
            }
            
            // 更新状态为运行中
            alterRequest.setStatus("RUNNING");
            alterRequest.setStartTime(System.currentTimeMillis());
            
            // 根据变更策略执行表变更
            TableAlterResult alterResult = null;
            switch (properties.getAlterStrategy().toUpperCase()) {
                case "ONLINE_DDL":
                    alterResult = mysqlOnlineDDLService.executeOnlineDDL(alterRequest);
                    break;
                case "PT_ONLINE_SCHEMA_CHANGE":
                    alterResult = executePtOnlineSchemaChange(alterRequest);
                    break;
                case "GH_OST":
                    alterResult = executeGhOst(alterRequest);
                    break;
                default:
                    alterResult = mysqlOnlineDDLService.executeOnlineDDL(alterRequest);
                    break;
            }
            
            if (alterResult != null && alterResult.isSuccess()) {
                result.setSuccess(true);
                result.setStatus("COMPLETED");
                result.setEndTime(System.currentTimeMillis());
                
                // 更新变更请求状态
                alterRequest.setStatus("COMPLETED");
                alterRequest.setEndTime(System.currentTimeMillis());
                
                // 记录变更成功指标
                mysqlTableAlterMonitorService.recordTableAlter(alterRequest.getTableName(), true);
                
                logger.info("表变更成功，变更ID: {}, 表: {}.{}, 耗时: {}ms", 
                    alterRequest.getAlterId(), alterRequest.getDatabaseName(), 
                    alterRequest.getTableName(), result.getDuration());
            } else {
                result.setSuccess(false);
                result.setStatus("FAILED");
                result.setError(alterResult != null ? alterResult.getError() : "表变更失败");
                result.setEndTime(System.currentTimeMillis());
                
                // 更新变更请求状态
                alterRequest.setStatus("FAILED");
                alterRequest.setEndTime(System.currentTimeMillis());
                alterRequest.setErrorMessage(result.getError());
                
                // 记录变更失败指标
                mysqlTableAlterMonitorService.recordTableAlter(alterRequest.getTableName(), false);
            }
            
            return result;
            
        } catch (Exception e) {
            logger.error("表变更异常，变更ID: {}", alterRequest.getAlterId(), e);
            result.setSuccess(false);
            result.setStatus("FAILED");
            result.setError("表变更异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            
            // 更新变更请求状态
            alterRequest.setStatus("FAILED");
            alterRequest.setEndTime(System.currentTimeMillis());
            alterRequest.setErrorMessage(result.getError());
            
            // 记录变更异常指标
            mysqlTableAlterMonitorService.recordTableAlter(alterRequest.getTableName(), false);
            
            return result;
        }
    }
    
    /**
     * 批量执行表变更
     * @param alterRequests 表变更请求列表
     * @return 批量表变更结果
     */
    public TableAlterBatchResult batchExecuteTableAlter(List<TableAlterRequest> alterRequests) {
        logger.info("批量执行表变更，数量: {}", alterRequests.size());
        
        TableAlterBatchResult result = new TableAlterBatchResult();
        result.setTotalCount(alterRequests.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 验证和预处理变更请求
            List<TableAlterRequest> validRequests = new ArrayList<>();
            List<TableAlterRequest> invalidRequests = new ArrayList<>();
            
            for (TableAlterRequest alterRequest : alterRequests) {
                if (alterRequest.validate()) {
                    validRequests.add(alterRequest);
                } else {
                    invalidRequests.add(alterRequest);
                    logger.warn("表变更请求验证失败，变更ID: {}", alterRequest.getAlterId());
                }
            }
            
            // 批量执行有效变更请求
            if (!validRequests.isEmpty()) {
                TableAlterBatchResult batchResult = executeBatchTableAlter(validRequests);
                
                result.setSuccessCount(batchResult.getSuccessCount());
                result.setFailureCount(batchResult.getFailureCount() + invalidRequests.size());
                result.setResults(batchResult.getResults());
            } else {
                result.setFailureCount(alterRequests.size());
            }
            
            result.setEndTime(System.currentTimeMillis());
            
            // 记录批量变更指标
            mysqlTableAlterMonitorService.recordTableAlterBatch(alterRequests.size(), result.getSuccessCount());
            
            logger.info("批量表变更完成，总数: {}, 成功: {}, 失败: {}, 耗时: {}ms", 
                result.getTotalCount(), result.getSuccessCount(), result.getFailureCount(), result.getDuration());
            
            return result;
            
        } catch (Exception e) {
            logger.error("批量表变更异常", e);
            result.setSuccess(false);
            result.setError("批量表变更异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 获取表结构信息
     * @param databaseName 数据库名称
     * @param tableName 表名称
     * @return 表结构信息
     */
    public TableStructureInfo getTableStructure(String databaseName, String tableName) {
        logger.info("获取表结构信息，表: {}.{}", databaseName, tableName);
        
        try {
            TableStructureInfo tableStructure = mysqlTableStructureService.getTableStructure(databaseName, tableName);
            
            if (tableStructure != null) {
                logger.info("获取表结构信息成功，表: {}.{}, 字段数: {}, 索引数: {}", 
                    databaseName, tableName, tableStructure.getColumns().size(), tableStructure.getIndexes().size());
            } else {
                logger.warn("表结构信息不存在，表: {}.{}", databaseName, tableName);
            }
            
            return tableStructure;
            
        } catch (Exception e) {
            logger.error("获取表结构信息异常，表: {}.{}", databaseName, tableName, e);
            return null;
        }
    }
    
    /**
     * 回滚表变更
     * @param alterRequest 表变更请求
     * @return 回滚结果
     */
    public TableAlterResult rollbackTableAlter(TableAlterRequest alterRequest) {
        logger.info("回滚表变更，变更ID: {}, 表: {}.{}", 
            alterRequest.getAlterId(), alterRequest.getDatabaseName(), alterRequest.getTableName());
        
        TableAlterResult result = new TableAlterResult(
            alterRequest.getAlterId(), alterRequest.getDatabaseName(), 
            alterRequest.getTableName(), alterRequest.getAlterType());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 检查是否支持回滚
            if (!properties.isEnableRollback()) {
                result.setSuccess(false);
                result.setError("回滚功能未启用");
                result.setEndTime(System.currentTimeMillis());
                return result;
            }
            
            // 执行回滚操作
            boolean rollbackSuccess = executeRollback(alterRequest);
            
            if (rollbackSuccess) {
                result.setSuccess(true);
                result.setStatus("ROLLBACK_COMPLETED");
                result.setEndTime(System.currentTimeMillis());
                
                // 更新变更请求状态
                alterRequest.setStatus("ROLLBACK");
                
                logger.info("表变更回滚成功，变更ID: {}, 表: {}.{}, 耗时: {}ms", 
                    alterRequest.getAlterId(), alterRequest.getDatabaseName(), 
                    alterRequest.getTableName(), result.getDuration());
            } else {
                result.setSuccess(false);
                result.setStatus("ROLLBACK_FAILED");
                result.setError("回滚操作失败");
                result.setEndTime(System.currentTimeMillis());
            }
            
            return result;
            
        } catch (Exception e) {
            logger.error("表变更回滚异常，变更ID: {}", alterRequest.getAlterId(), e);
            result.setSuccess(false);
            result.setStatus("ROLLBACK_FAILED");
            result.setError("表变更回滚异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 执行pt-online-schema-change
     * @param alterRequest 表变更请求
     * @return 表变更结果
     */
    private TableAlterResult executePtOnlineSchemaChange(TableAlterRequest alterRequest) {
        logger.info("执行pt-online-schema-change，变更ID: {}", alterRequest.getAlterId());
        
        TableAlterResult result = new TableAlterResult(
            alterRequest.getAlterId(), alterRequest.getDatabaseName(), 
            alterRequest.getTableName(), alterRequest.getAlterType());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 构建pt-online-schema-change命令
            String command = buildPtOnlineSchemaChangeCommand(alterRequest);
            
            // 执行命令
            Process process = Runtime.getRuntime().exec(command);
            
            // 等待执行完成
            boolean finished = process.waitFor(properties.getPtOnlineSchemaChangeTimeout(), TimeUnit.SECONDS);
            
            if (finished && process.exitValue() == 0) {
                result.setSuccess(true);
                result.setStatus("COMPLETED");
                result.setEndTime(System.currentTimeMillis());
                
                logger.info("pt-online-schema-change执行成功，变更ID: {}, 耗时: {}ms", 
                    alterRequest.getAlterId(), result.getDuration());
            } else {
                result.setSuccess(false);
                result.setStatus("FAILED");
                result.setError("pt-online-schema-change执行失败");
                result.setEndTime(System.currentTimeMillis());
            }
            
            return result;
            
        } catch (Exception e) {
            logger.error("pt-online-schema-change执行异常，变更ID: {}", alterRequest.getAlterId(), e);
            result.setSuccess(false);
            result.setStatus("FAILED");
            result.setError("pt-online-schema-change执行异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 执行gh-ost
     * @param alterRequest 表变更请求
     * @return 表变更结果
     */
    private TableAlterResult executeGhOst(TableAlterRequest alterRequest) {
        logger.info("执行gh-ost，变更ID: {}", alterRequest.getAlterId());
        
        TableAlterResult result = new TableAlterResult(
            alterRequest.getAlterId(), alterRequest.getDatabaseName(), 
            alterRequest.getTableName(), alterRequest.getAlterType());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 构建gh-ost命令
            String command = buildGhOstCommand(alterRequest);
            
            // 执行命令
            Process process = Runtime.getRuntime().exec(command);
            
            // 等待执行完成
            boolean finished = process.waitFor(properties.getGhOstTimeout(), TimeUnit.SECONDS);
            
            if (finished && process.exitValue() == 0) {
                result.setSuccess(true);
                result.setStatus("COMPLETED");
                result.setEndTime(System.currentTimeMillis());
                
                logger.info("gh-ost执行成功，变更ID: {}, 耗时: {}ms", 
                    alterRequest.getAlterId(), result.getDuration());
            } else {
                result.setSuccess(false);
                result.setStatus("FAILED");
                result.setError("gh-ost执行失败");
                result.setEndTime(System.currentTimeMillis());
            }
            
            return result;
            
        } catch (Exception e) {
            logger.error("gh-ost执行异常，变更ID: {}", alterRequest.getAlterId(), e);
            result.setSuccess(false);
            result.setStatus("FAILED");
            result.setError("gh-ost执行异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 执行批量表变更
     * @param alterRequests 表变更请求列表
     * @return 批量表变更结果
     */
    private TableAlterBatchResult executeBatchTableAlter(List<TableAlterRequest> alterRequests) {
        TableAlterBatchResult result = new TableAlterBatchResult();
        result.setTotalCount(alterRequests.size());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            List<TableAlterResult> results = new ArrayList<>();
            int successCount = 0;
            int failureCount = 0;
            
            for (TableAlterRequest alterRequest : alterRequests) {
                try {
                    TableAlterResult alterResult = executeTableAlter(alterRequest);
                    results.add(alterResult);
                    
                    if (alterResult.isSuccess()) {
                        successCount++;
                    } else {
                        failureCount++;
                    }
                } catch (Exception e) {
                    logger.error("批量表变更单条执行异常，变更ID: {}", alterRequest.getAlterId(), e);
                    failureCount++;
                }
            }
            
            result.setSuccessCount(successCount);
            result.setFailureCount(failureCount);
            result.setResults(results);
            result.setEndTime(System.currentTimeMillis());
            
            return result;
            
        } catch (Exception e) {
            logger.error("执行批量表变更异常", e);
            result.setSuccess(false);
            result.setError("执行批量表变更异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 执行回滚操作
     * @param alterRequest 表变更请求
     * @return 是否成功
     */
    private boolean executeRollback(TableAlterRequest alterRequest) {
        try {
            // 实现回滚逻辑
            // 这里可以根据具体的变更类型实现相应的回滚操作
            
            return true;
            
        } catch (Exception e) {
            logger.error("执行回滚操作异常，变更ID: {}", alterRequest.getAlterId(), e);
            return false;
        }
    }
    
    /**
     * 构建pt-online-schema-change命令
     * @param alterRequest 表变更请求
     * @return 命令字符串
     */
    private String buildPtOnlineSchemaChangeCommand(TableAlterRequest alterRequest) {
        StringBuilder command = new StringBuilder();
        command.append(properties.getPtOnlineSchemaChangePath());
        command.append(" --alter='").append(alterRequest.getAlterSql()).append("'");
        command.append(" --host=").append(extractHostFromUrl(properties.getDatabase().getUrl()));
        command.append(" --port=").append(extractPortFromUrl(properties.getDatabase().getUrl()));
        command.append(" --user=").append(properties.getDatabase().getUsername());
        command.append(" --password=").append(properties.getDatabase().getPassword());
        command.append(" --database=").append(alterRequest.getDatabaseName());
        command.append(" --table=").append(alterRequest.getTableName());
        command.append(" --execute");
        
        return command.toString();
    }
    
    /**
     * 构建gh-ost命令
     * @param alterRequest 表变更请求
     * @return 命令字符串
     */
    private String buildGhOstCommand(TableAlterRequest alterRequest) {
        StringBuilder command = new StringBuilder();
        command.append(properties.getGhOstPath());
        command.append(" --alter='").append(alterRequest.getAlterSql()).append("'");
        command.append(" --host=").append(extractHostFromUrl(properties.getDatabase().getUrl()));
        command.append(" --port=").append(extractPortFromUrl(properties.getDatabase().getUrl()));
        command.append(" --user=").append(properties.getDatabase().getUsername());
        command.append(" --password=").append(properties.getDatabase().getPassword());
        command.append(" --database=").append(alterRequest.getDatabaseName());
        command.append(" --table=").append(alterRequest.getTableName());
        command.append(" --execute");
        
        return command.toString();
    }
    
    /**
     * 从URL中提取主机名
     * @param url 数据库URL
     * @return 主机名
     */
    private String extractHostFromUrl(String url) {
        try {
            URI uri = new URI(url.replace("jdbc:", ""));
            return uri.getHost();
        } catch (Exception e) {
            return "localhost";
        }
    }
    
    /**
     * 从URL中提取端口
     * @param url 数据库URL
     * @return 端口
     */
    private String extractPortFromUrl(String url) {
        try {
            URI uri = new URI(url.replace("jdbc:", ""));
            return String.valueOf(uri.getPort());
        } catch (Exception e) {
            return "3306";
        }
    }
}

/**
 * MySQL表变更批量结果类
 * @author 运维实战
 */
@Data
public class TableAlterBatchResult {
    
    private boolean success;
    private int totalCount;
    private int successCount;
    private int failureCount;
    private List<TableAlterResult> results;
    private String error;
    private long startTime;
    private long endTime;
    
    public TableAlterBatchResult() {
        this.success = false;
        this.successCount = 0;
        this.failureCount = 0;
        this.results = new ArrayList<>();
    }
    
    /**
     * 获取变更耗时
     * @return 变更耗时（毫秒）
     */
    public long getDuration() {
        return endTime - startTime;
    }
    
    /**
     * 是否成功
     * @return 是否成功
     */
    public boolean isSuccess() {
        return success;
    }
}

/**
 * MySQL在线DDL服务
 * @author 运维实战
 */
@Service
public class MySQLOnlineDDLService {
    
    @Autowired
    private MySQLTableAlterProperties properties;
    
    @Autowired
    private DataSource dataSource;
    
    @Autowired
    private MySQLTableAlterMonitorService mysqlTableAlterMonitorService;
    
    private static final Logger logger = LoggerFactory.getLogger(MySQLOnlineDDLService.class);
    
    /**
     * 执行在线DDL
     * @param alterRequest 表变更请求
     * @return 表变更结果
     */
    public TableAlterResult executeOnlineDDL(TableAlterRequest alterRequest) {
        logger.info("执行在线DDL，变更ID: {}, 表: {}.{}", 
            alterRequest.getAlterId(), alterRequest.getDatabaseName(), alterRequest.getTableName());
        
        TableAlterResult result = new TableAlterResult(
            alterRequest.getAlterId(), alterRequest.getDatabaseName(), 
            alterRequest.getTableName(), alterRequest.getAlterType());
        result.setStartTime(System.currentTimeMillis());
        
        try {
            // 验证在线DDL支持
            if (!isOnlineDDLSupported(alterRequest)) {
                result.setSuccess(false);
                result.setError("不支持在线DDL操作");
                result.setEndTime(System.currentTimeMillis());
                return result;
            }
            
            // 构建在线DDL SQL
            String onlineDDLSql = buildOnlineDDLSql(alterRequest);
            
            // 执行在线DDL
            boolean success = executeOnlineDDLSql(onlineDDLSql, alterRequest);
            
            if (success) {
                result.setSuccess(true);
                result.setStatus("COMPLETED");
                result.setEndTime(System.currentTimeMillis());
                
                logger.info("在线DDL执行成功，变更ID: {}, 表: {}.{}, 耗时: {}ms", 
                    alterRequest.getAlterId(), alterRequest.getDatabaseName(), 
                    alterRequest.getTableName(), result.getDuration());
            } else {
                result.setSuccess(false);
                result.setStatus("FAILED");
                result.setError("在线DDL执行失败");
                result.setEndTime(System.currentTimeMillis());
            }
            
            return result;
            
        } catch (Exception e) {
            logger.error("在线DDL执行异常，变更ID: {}", alterRequest.getAlterId(), e);
            result.setSuccess(false);
            result.setStatus("FAILED");
            result.setError("在线DDL执行异常: " + e.getMessage());
            result.setEndTime(System.currentTimeMillis());
            return result;
        }
    }
    
    /**
     * 检查在线DDL支持
     * @param alterRequest 表变更请求
     * @return 是否支持
     */
    private boolean isOnlineDDLSupported(TableAlterRequest alterRequest) {
        try {
            // 检查MySQL版本
            String version = getMySQLVersion();
            if (version == null || !version.startsWith("5.6")) {
                return false;
            }
            
            // 检查变更类型支持
            String alterType = alterRequest.getAlterType();
            switch (alterType.toUpperCase()) {
                case "ADD_COLUMN":
                case "DROP_COLUMN":
                case "MODIFY_COLUMN":
                case "ADD_INDEX":
                case "DROP_INDEX":
                    return true;
                default:
                    return false;
            }
            
        } catch (Exception e) {
            logger.error("检查在线DDL支持异常", e);
            return false;
        }
    }
    
    /**
     * 构建在线DDL SQL
     * @param alterRequest 表变更请求
     * @return 在线DDL SQL
     */
    private String buildOnlineDDLSql(TableAlterRequest alterRequest) {
        StringBuilder sql = new StringBuilder();
        sql.append("ALTER TABLE ").append(alterRequest.getDatabaseName())
           .append(".").append(alterRequest.getTableName());
        sql.append(" ").append(alterRequest.getAlterSql());
        sql.append(" ALGORITHM=INPLACE, LOCK=NONE");
        
        return sql.toString();
    }
    
    /**
     * 执行在线DDL SQL
     * @param sql 在线DDL SQL
     * @param alterRequest 表变更请求
     * @return 是否成功
     */
    private boolean executeOnlineDDLSql(String sql, TableAlterRequest alterRequest) {
        try (Connection connection = dataSource.getConnection()) {
            // 设置超时时间
            connection.setNetworkTimeout(null, (int) properties.getOnlineDDLTimeout() * 1000);
            
            // 执行SQL
            try (Statement statement = connection.createStatement()) {
                statement.setQueryTimeout((int) properties.getOnlineDDLTimeout());
                statement.execute(sql);
                
                logger.info("在线DDL SQL执行成功，变更ID: {}, SQL: {}", alterRequest.getAlterId(), sql);
                return true;
            }
            
        } catch (Exception e) {
            logger.error("在线DDL SQL执行异常，变更ID: {}, SQL: {}", alterRequest.getAlterId(), sql, e);
            return false;
        }
    }
    
    /**
     * 获取MySQL版本
     * @return MySQL版本
     */
    private String getMySQLVersion() {
        try (Connection connection = dataSource.getConnection()) {
            try (Statement statement = connection.createStatement()) {
                try (ResultSet resultSet = statement.executeQuery("SELECT VERSION()")) {
                    if (resultSet.next()) {
                        return resultSet.getString(1);
                    }
                }
            }
        } catch (Exception e) {
            logger.error("获取MySQL版本异常", e);
        }
        return null;
    }
}

/**
 * MySQL表结构服务
 * @author 运维实战
 */
@Service
public class MySQLTableStructureService {
    
    @Autowired
    private DataSource dataSource;
    
    @Autowired
    private MySQLTableAlterMonitorService mysqlTableAlterMonitorService;
    
    private static final Logger logger = LoggerFactory.getLogger(MySQLTableStructureService.class);
    
    /**
     * 获取表结构信息
     * @param databaseName 数据库名称
     * @param tableName 表名称
     * @return 表结构信息
     */
    public TableStructureInfo getTableStructure(String databaseName, String tableName) {
        logger.info("获取表结构信息，表: {}.{}", databaseName, tableName);
        
        try {
            TableStructureInfo tableStructure = new TableStructureInfo(databaseName, tableName);
            
            // 获取表基本信息
            getTableBasicInfo(tableStructure);
            
            // 获取字段信息
            getTableColumns(tableStructure);
            
            // 获取索引信息
            getTableIndexes(tableStructure);
            
            logger.info("获取表结构信息成功，表: {}.{}, 字段数: {}, 索引数: {}", 
                databaseName, tableName, tableStructure.getColumns().size(), tableStructure.getIndexes().size());
            
            return tableStructure;
            
        } catch (Exception e) {
            logger.error("获取表结构信息异常，表: {}.{}", databaseName, tableName, e);
            return null;
        }
    }
    
    /**
     * 获取表基本信息
     * @param tableStructure 表结构信息
     */
    private void getTableBasicInfo(TableStructureInfo tableStructure) {
        try (Connection connection = dataSource.getConnection()) {
            String sql = "SELECT ENGINE, TABLE_COLLATION, TABLE_ROWS, " +
                        "DATA_LENGTH, INDEX_LENGTH, CREATE_TIME, UPDATE_TIME " +
                        "FROM INFORMATION_SCHEMA.TABLES " +
                        "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?";
            
            try (PreparedStatement statement = connection.prepareStatement(sql)) {
                statement.setString(1, tableStructure.getDatabaseName());
                statement.setString(2, tableStructure.getTableName());
                
                try (ResultSet resultSet = statement.executeQuery()) {
                    if (resultSet.next()) {
                        tableStructure.setEngine(resultSet.getString("ENGINE"));
                        tableStructure.setCollation(resultSet.getString("TABLE_COLLATION"));
                        tableStructure.setRowCount(resultSet.getLong("TABLE_ROWS"));
                        tableStructure.setTableSize(resultSet.getLong("DATA_LENGTH"));
                        tableStructure.setIndexSize(resultSet.getLong("INDEX_LENGTH"));
                        tableStructure.setCreateTime(resultSet.getTimestamp("CREATE_TIME").getTime());
                        tableStructure.setUpdateTime(resultSet.getTimestamp("UPDATE_TIME").getTime());
                    }
                }
            }
            
        } catch (Exception e) {
            logger.error("获取表基本信息异常，表: {}.{}", 
                tableStructure.getDatabaseName(), tableStructure.getTableName(), e);
        }
    }
    
    /**
     * 获取表字段信息
     * @param tableStructure 表结构信息
     */
    private void getTableColumns(TableStructureInfo tableStructure) {
        try (Connection connection = dataSource.getConnection()) {
            String sql = "SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, " +
                        "COLUMN_DEFAULT, COLUMN_COMMENT, ORDINAL_POSITION " +
                        "FROM INFORMATION_SCHEMA.COLUMNS " +
                        "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? " +
                        "ORDER BY ORDINAL_POSITION";
            
            try (PreparedStatement statement = connection.prepareStatement(sql)) {
                statement.setString(1, tableStructure.getDatabaseName());
                statement.setString(2, tableStructure.getTableName());
                
                try (ResultSet resultSet = statement.executeQuery()) {
                    while (resultSet.next()) {
                        ColumnInfo columnInfo = new ColumnInfo();
                        columnInfo.setColumnName(resultSet.getString("COLUMN_NAME"));
                        columnInfo.setColumnType(resultSet.getString("COLUMN_TYPE"));
                        columnInfo.setNullable("YES".equals(resultSet.getString("IS_NULLABLE")));
                        columnInfo.setDefaultValue(resultSet.getString("COLUMN_DEFAULT"));
                        columnInfo.setComment(resultSet.getString("COLUMN_COMMENT"));
                        columnInfo.setPosition(resultSet.getInt("ORDINAL_POSITION"));
                        
                        tableStructure.addColumn(columnInfo);
                    }
                }
            }
            
        } catch (Exception e) {
            logger.error("获取表字段信息异常，表: {}.{}", 
                tableStructure.getDatabaseName(), tableStructure.getTableName(), e);
        }
    }
    
    /**
     * 获取表索引信息
     * @param tableStructure 表结构信息
     */
    private void getTableIndexes(TableStructureInfo tableStructure) {
        try (Connection connection = dataSource.getConnection()) {
            String sql = "SELECT INDEX_NAME, INDEX_TYPE, COLUMN_NAME, NON_UNIQUE " +
                        "FROM INFORMATION_SCHEMA.STATISTICS " +
                        "WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? " +
                        "ORDER BY INDEX_NAME, SEQ_IN_INDEX";
            
            try (PreparedStatement statement = connection.prepareStatement(sql)) {
                statement.setString(1, tableStructure.getDatabaseName());
                statement.setString(2, tableStructure.getTableName());
                
                try (ResultSet resultSet = statement.executeQuery()) {
                    Map<String, IndexInfo> indexMap = new HashMap<>();
                    
                    while (resultSet.next()) {
                        String indexName = resultSet.getString("INDEX_NAME");
                        IndexInfo indexInfo = indexMap.get(indexName);
                        
                        if (indexInfo == null) {
                            indexInfo = new IndexInfo();
                            indexInfo.setIndexName(indexName);
                            indexInfo.setIndexType(resultSet.getString("INDEX_TYPE"));
                            indexInfo.setUnique(resultSet.getInt("NON_UNIQUE") == 0);
                            indexInfo.setColumns(new ArrayList<>());
                            indexMap.put(indexName, indexInfo);
                        }
                        
                        indexInfo.getColumns().add(resultSet.getString("COLUMN_NAME"));
                    }
                    
                    // 添加索引信息到表结构
                    for (IndexInfo indexInfo : indexMap.values()) {
                        tableStructure.addIndex(indexInfo);
                    }
                }
            }
            
        } catch (Exception e) {
            logger.error("获取表索引信息异常，表: {}.{}", 
                tableStructure.getDatabaseName(), tableStructure.getTableName(), e);
        }
    }
    
    /**
     * 验证表结构变更
     * @param tableStructure 表结构信息
     * @param alterRequest 表变更请求
     * @return 是否有效
     */
    public boolean validateTableStructureAlter(TableStructureInfo tableStructure, TableAlterRequest alterRequest) {
        try {
            // 验证表是否存在
            if (tableStructure == null) {
                return false;
            }
            
            // 验证变更类型
            String alterType = alterRequest.getAlterType();
            switch (alterType.toUpperCase()) {
                case "ADD_COLUMN":
                    return validateAddColumn(tableStructure, alterRequest);
                case "DROP_COLUMN":
                    return validateDropColumn(tableStructure, alterRequest);
                case "MODIFY_COLUMN":
                    return validateModifyColumn(tableStructure, alterRequest);
                case "ADD_INDEX":
                    return validateAddIndex(tableStructure, alterRequest);
                case "DROP_INDEX":
                    return validateDropIndex(tableStructure, alterRequest);
                default:
                    return false;
            }
            
        } catch (Exception e) {
            logger.error("验证表结构变更异常", e);
            return false;
        }
    }
    
    /**
     * 验证添加字段
     * @param tableStructure 表结构信息
     * @param alterRequest 表变更请求
     * @return 是否有效
     */
    private boolean validateAddColumn(TableStructureInfo tableStructure, TableAlterRequest alterRequest) {
        // 检查字段是否已存在
        String columnName = extractColumnNameFromAlterSql(alterRequest.getAlterSql());
        if (columnName != null) {
            for (ColumnInfo column : tableStructure.getColumns()) {
                if (columnName.equals(column.getColumnName())) {
                    return false;
                }
            }
        }
        return true;
    }
    
    /**
     * 验证删除字段
     * @param tableStructure 表结构信息
     * @param alterRequest 表变更请求
     * @return 是否有效
     */
    private boolean validateDropColumn(TableStructureInfo tableStructure, TableAlterRequest alterRequest) {
        // 检查字段是否存在
        String columnName = extractColumnNameFromAlterSql(alterRequest.getAlterSql());
        if (columnName != null) {
            for (ColumnInfo column : tableStructure.getColumns()) {
                if (columnName.equals(column.getColumnName())) {
                    return true;
                }
            }
        }
        return false;
    }
    
    /**
     * 验证修改字段
     * @param tableStructure 表结构信息
     * @param alterRequest 表变更请求
     * @return 是否有效
     */
    private boolean validateModifyColumn(TableStructureInfo tableStructure, TableAlterRequest alterRequest) {
        // 检查字段是否存在
        String columnName = extractColumnNameFromAlterSql(alterRequest.getAlterSql());
        if (columnName != null) {
            for (ColumnInfo column : tableStructure.getColumns()) {
                if (columnName.equals(column.getColumnName())) {
                    return true;
                }
            }
        }
        return false;
    }
    
    /**
     * 验证添加索引
     * @param tableStructure 表结构信息
     * @param alterRequest 表变更请求
     * @return 是否有效
     */
    private boolean validateAddIndex(TableStructureInfo tableStructure, TableAlterRequest alterRequest) {
        // 检查索引是否已存在
        String indexName = extractIndexNameFromAlterSql(alterRequest.getAlterSql());
        if (indexName != null) {
            for (IndexInfo index : tableStructure.getIndexes()) {
                if (indexName.equals(index.getIndexName())) {
                    return false;
                }
            }
        }
        return true;
    }
    
    /**
     * 验证删除索引
     * @param tableStructure 表结构信息
     * @param alterRequest 表变更请求
     * @return 是否有效
     */
    private boolean validateDropIndex(TableStructureInfo tableStructure, TableAlterRequest alterRequest) {
        // 检查索引是否存在
        String indexName = extractIndexNameFromAlterSql(alterRequest.getAlterSql());
        if (indexName != null) {
            for (IndexInfo index : tableStructure.getIndexes()) {
                if (indexName.equals(index.getIndexName())) {
                    return true;
                }
            }
        }
        return false;
    }
    
    /**
     * 从ALTER SQL中提取字段名
     * @param alterSql ALTER SQL
     * @return 字段名
     */
    private String extractColumnNameFromAlterSql(String alterSql) {
        try {
            // 简单的字段名提取逻辑
            // 这里可以根据具体的SQL格式实现更复杂的解析
            if (alterSql.toUpperCase().contains("ADD COLUMN")) {
                String[] parts = alterSql.split("ADD COLUMN");
                if (parts.length > 1) {
                    String columnPart = parts[1].trim();
                    String[] columnParts = columnPart.split("\\s+");
                    if (columnParts.length > 0) {
                        return columnParts[0];
                    }
                }
            } else if (alterSql.toUpperCase().contains("DROP COLUMN")) {
                String[] parts = alterSql.split("DROP COLUMN");
                if (parts.length > 1) {
                    String columnPart = parts[1].trim();
                    String[] columnParts = columnPart.split("\\s+");
                    if (columnParts.length > 0) {
                        return columnParts[0];
                    }
                }
            } else if (alterSql.toUpperCase().contains("MODIFY COLUMN")) {
                String[] parts = alterSql.split("MODIFY COLUMN");
                if (parts.length > 1) {
                    String columnPart = parts[1].trim();
                    String[] columnParts = columnPart.split("\\s+");
                    if (columnParts.length > 0) {
                        return columnParts[0];
                    }
                }
            }
        } catch (Exception e) {
            logger.error("提取字段名异常，SQL: {}", alterSql, e);
        }
        return null;
    }
    
    /**
     * 从ALTER SQL中提取索引名
     * @param alterSql ALTER SQL
     * @return 索引名
     */
    private String extractIndexNameFromAlterSql(String alterSql) {
        try {
            // 简单的索引名提取逻辑
            if (alterSql.toUpperCase().contains("ADD INDEX")) {
                String[] parts = alterSql.split("ADD INDEX");
                if (parts.length > 1) {
                    String indexPart = parts[1].trim();
                    String[] indexParts = indexPart.split("\\s+");
                    if (indexParts.length > 0) {
                        return indexParts[0];
                    }
                }
            } else if (alterSql.toUpperCase().contains("DROP INDEX")) {
                String[] parts = alterSql.split("DROP INDEX");
                if (parts.length > 1) {
                    String indexPart = parts[1].trim();
                    String[] indexParts = indexPart.split("\\s+");
                    if (indexParts.length > 0) {
                        return indexParts[0];
                    }
                }
            }
        } catch (Exception e) {
            logger.error("提取索引名异常，SQL: {}", alterSql, e);
        }
        return null;
    }
}