MySQL数据库实现批量表分区完整示例

CREATE TABLE sales (
    id INT NOT NULL,
    sale_date DATE NOT NULL,
    amount DECIMAL(10,2)
)
PARTITION BY RANGE (YEAR(sale_date)) (
    PARTITION p0 VALUES LESS THAN (2020),
    PARTITION p1 VALUES LESS THAN (2021),
    PARTITION p2 VALUES LESS THAN (2022),
    PARTITION p3 VALUES LESS THAN MAXVALUE
);

CREATE DEFINER=`root`@`%` PROCEDURE `convert_table_to_partition`(IN tbl_name VARCHAR(200),OUT out_status INT)
BEGIN
    DECLARE done INT DEFAULT FALSE;
    -- 输出状态，开始执行状态 100，执行成功状态 200，执行失败状态 50
    SET out_status = 100;
    
    -- 创建一个新的空表，不包含表分区（要转换为分区表，必须是空表）
    SET @create_empty_tbl_sql = CONCAT(
        'CREATE TABLE ', tbl_name, '_partitioned LIKE ', tbl_name, ';'
    );
    PREPARE stmt FROM @create_empty_tbl_sql;
    EXECUTE stmt;
    DEALLOCATE PREPARE stmt;

    -- 获取所有唯一的 year_no 和 month_no 组合作为构建分区定义分区键
    SET @partition_def = '';
    SET @query = CONCAT(
        'SELECT GROUP_CONCAT(
            CONCAT("PARTITION p_", year_no, "_", LPAD(month_no, 2, "0"), 
                   " VALUES LESS THAN (", 
                   CASE WHEN month_no = 12 THEN year_no + 1 ELSE year_no END, ", ", 
                   CASE WHEN month_no = 12 THEN 1 ELSE month_no + 1 END, ")")
            ORDER BY year_no, month_no SEPARATOR ",\n"
        ) INTO @partition_def
        FROM (
            SELECT DISTINCT year_no, month_no
            FROM ', tbl_name, '
            ORDER BY year_no, month_no
        ) AS unique_years_months;'
    );
    PREPARE stmt FROM @query;
    EXECUTE stmt;
    DEALLOCATE PREPARE stmt;

    -- 调试信息：输出分区定义字符串
    --SELECT tbl_name,@partition_def;

    -- 检查是否有有效的分区定义
    IF @partition_def IS NULL OR @partition_def = '' THEN
        SELECT tbl_name,'No data found for partitioning. Skipping partition creation and data migration.';
        -- 空表则直接添加 p_max 分区用于捕获未来数据
        SET @partition_def = '\nPARTITION p_max VALUES LESS THAN (MAXVALUE, MAXVALUE)';
    ELSE
        -- 添加 p_max 分区用于捕获未来数据
        SET @partition_def = CONCAT(@partition_def, ',\nPARTITION p_max VALUES LESS THAN (MAXVALUE, MAXVALUE)');
    END IF;

    -- 使用 ALTER TABLE 添加分区定义
    SET @add_partitions_sql = CONCAT(
        'ALTER TABLE ', tbl_name, '_partitioned 
        PARTITION BY RANGE COLUMNS(year_no, month_no) (', @partition_def, ');'
    );

    -- 调试信息：输出添加分区的 SQL 语句
    --SELECT tbl_name,@add_partitions_sql;

    PREPARE stmt FROM @add_partitions_sql;
    EXECUTE stmt;
    DEALLOCATE PREPARE stmt;

    -- 迁移数据到新的分区表
    SET @insert_into_partitioned_sql = CONCAT(
        'INSERT INTO ', tbl_name, '_partitioned SELECT * FROM ', tbl_name, ';'
    );

    -- 调试信息：输出插入数据的 SQL 语句
    -- SELECT tbl_name,@insert_into_partitioned_sql;

    PREPARE stmt FROM @insert_into_partitioned_sql;
    EXECUTE stmt;
    DEALLOCATE PREPARE stmt;

    -- 验证数据迁移是否成功
    SET @count_original = CONCAT('SELECT COUNT(*) INTO @count_original FROM ', tbl_name);
    PREPARE stmt FROM @count_original;
    EXECUTE stmt;
    DEALLOCATE PREPARE stmt;

    SET @count_partitioned = CONCAT('SELECT COUNT(*) INTO @count_partitioned FROM ', tbl_name, '_partitioned');
    PREPARE stmt FROM @count_partitioned;
    EXECUTE stmt;
    DEALLOCATE PREPARE stmt;

    -- 比较原表和新分区表的数据行数
    -- SELECT tbl_name,@count_original, @count_partitioned;

    
    -- 如果数据迁移成功，删除旧表并重命名新表（无论是否有数据，均删除缓存表）
    IF @count_original = @count_partitioned THEN
        -- 删除旧表
        SET @drop_old_table_sql = CONCAT('DROP TABLE IF EXISTS ', tbl_name);
        PREPARE stmt FROM @drop_old_table_sql;
        EXECUTE stmt;
        DEALLOCATE PREPARE stmt;

        -- 重命名新表为旧表名
        SET @rename_tables_sql = CONCAT('RENAME TABLE ', tbl_name, '_partitioned TO ', tbl_name);
        PREPARE stmt FROM @rename_tables_sql;
        EXECUTE stmt;
        DEALLOCATE PREPARE stmt;

        -- SELECT tbl_name,'Table conversion and data migration completed successfully.';
        SET out_status = 200;
    ELSE
        -- SELECT tbl_name,'Data migration failed, check the logs for more information.';
        SET out_status = 50;
    END IF;
    
END

CREATE DEFINER=`root`@`%` PROCEDURE `tables_convert_to_partition`()
BEGIN
    DECLARE done INT DEFAULT FALSE;
    DECLARE tbl_name VARCHAR(64);
    DECLARE convert_status INT;
    DECLARE cur CURSOR FOR
        SELECT TABLE_NAME
        FROM information_schema.TABLES
        WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME LIKE 'ai_result_%';
    DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = TRUE;

    OPEN cur;

    read_loop: LOOP
        FETCH cur INTO tbl_name;
        IF done THEN
            LEAVE read_loop;
        END IF;
      
        -- 调试信息：输出正在转换的表
        SELECT tbl_name,'covering...';

        CALL convert_table_to_partition(tbl_name,@status);
        
        SET convert_status = @status;
        
        -- 根据返回的状态进行相应的处理
        CASE convert_status
            WHEN 100 THEN
                -- 开始状态，可以忽略，因为这是预期的初始状态
                SELECT tbl_name, 'Started conversion.';
            WHEN 200 THEN
                -- 成功完成
                SELECT tbl_name, 'Conversion and data migration completed successfully.';
            WHEN 50 THEN
                -- 失败
                SELECT tbl_name, 'Data migration failed. Check the logs for more information.';
            ELSE
                -- 未知状态
                SELECT tbl_name, CONCAT('Unknown status: ', status);
        END CASE;
    END LOOP;

    CLOSE cur;
END

CREATE DEFINER=`root`@`%` PROCEDURE `add_monthly_partitions`(IN tbl_name VARCHAR(64), IN year_no INT, IN month_no INT,OUT out_status INT)
BEGIN
    DECLARE done INT DEFAULT FALSE;
    
    -- 输出状态，开始执行状态 100，执行成功状态 200，执行失败状态 50
    SET out_status = 100;
    
    -- 检查待添加的分区是否已经存在
    SET @partition_exists = EXISTS (SELECT 1 FROM information_schema.PARTITIONS 
                                    WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = tbl_name 
                                      AND PARTITION_NAME = CONCAT('p_', year_no, '_', LPAD(month_no, 2, '0')));
                                      
    IF @partition_exists THEN
        -- 如果分区已存在，直接返回消息
        -- SELECT CONCAT('Partition p_', year_no, '_', LPAD(month_no, 2, '0'), ' already exists. No action taken.') AS message;
        SET out_status = 200;
    ELSE
      -- 检查表中是否已经存在 p_max 分区
      SET @has_p_max = EXISTS (SELECT 1 FROM information_schema.PARTITIONS 
                               WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = tbl_name 
                                 AND PARTITION_NAME = 'p_max');

      -- 构建添加分区的 SQL 语句
      IF @has_p_max THEN
          -- 如果存在 p_max 分区，则重新组织分区，将 p_max 分割成新分区和更新后的 p_max
          SET @reorganize_partition_sql = CONCAT(
              'ALTER TABLE ', tbl_name, ' REORGANIZE PARTITION p_max INTO (
                  PARTITION p_', year_no, '_', LPAD(month_no, 2, '0'), 
                  ' VALUES LESS THAN (', 
                  CASE WHEN month_no = 12 THEN year_no + 1 ELSE year_no END, ', ', 
                  CASE WHEN month_no = 12 THEN 1 ELSE month_no + 1 END, '),
                  PARTITION p_max VALUES LESS THAN (MAXVALUE, MAXVALUE)
              )'
          );
          PREPARE stmt FROM @reorganize_partition_sql;
          EXECUTE stmt;
          DEALLOCATE PREPARE stmt;

          -- SELECT CONCAT('Partition p_', year_no, '_', LPAD(month_no, 2, '0'), ' and updated p_max added successfully.') AS message;
          SET out_status = 200;
      ELSE
          -- 如果不存在 p_max 分区，则直接添加新分区
          SET @add_partition_sql = CONCAT(
              'ALTER TABLE ', tbl_name, ' ADD PARTITION (
                  PARTITION p_', year_no, '_', LPAD(month_no, 2, '0'), 
                  ' VALUES LESS THAN (', 
                  CASE WHEN month_no = 12 THEN year_no + 1 ELSE year_no END, ', ', 
                  CASE WHEN month_no = 12 THEN 1 ELSE month_no + 1 END, ')
              )'
          );
          PREPARE stmt FROM @add_partition_sql;
          EXECUTE stmt;
          DEALLOCATE PREPARE stmt;
          
          -- SELECT CONCAT('Partition p_', year_no, '_', LPAD(month_no, 2, '0'), ' added successfully.') AS message;
          SET out_status = 200;
      END IF;
    END IF;
END

CREATE DEFINER=`root`@`%` PROCEDURE `tables_add_monthly_partition`(IN tbl_prefix VARCHAR(64), IN year_no INT, IN month_no INT,OUT out_status INT)
BEGIN
    DECLARE done INT DEFAULT FALSE;
    DECLARE tbl_name VARCHAR(64);
    DECLARE cur CURSOR FOR
        SELECT TABLE_NAME
        FROM information_schema.TABLES
        WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME LIKE CONCAT(tbl_prefix, '%');
    DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = TRUE;

    -- 输出状态，开始执行状态 100，执行成功状态 200，执行失败状态 50
    SET out_status = 100;

    -- 打开游标
    OPEN cur;

    read_loop: LOOP
        FETCH cur INTO tbl_name;
        IF done THEN
            LEAVE read_loop;
        END IF;

        -- 检查表是否已经是分区表
        SET @is_partitioned = EXISTS (SELECT 1 FROM information_schema.PARTITIONS 
                                      WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = tbl_name);

        IF NOT @is_partitioned THEN
            -- 如果表不是分区表，先调用 convert_table_to_partition 进行转换
            CALL convert_table_to_partition(tbl_name,@status);
            
            -- 转换后再次检查是否成功转换为分区表
            SET @is_partitioned = EXISTS (SELECT 1 FROM information_schema.PARTITIONS 
                                          WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = tbl_name);
            
            IF NOT @is_partitioned THEN
                -- 如果转换失败，跳过后续操作并输出错误信息
                SELECT CONCAT('Failed to convert table ', tbl_name, ' to partitioned. Skipping.') AS message;
                SET out_status = 50;
                ITERATE read_loop;
            END IF;
        END IF;

        -- 调用 add_monthly_partitions 为当前表添加分区
        CALL add_monthly_partitions(tbl_name, year_no, month_no,@status);

        -- 可选：输出操作结果（用于调试）
        -- SELECT CONCAT('Processed table: ', tbl_name) AS status;
    END LOOP;

    -- 关闭游标
    CLOSE cur;

    -- 输出完成信息
    -- SELECT CONCAT('Batch partition addition completed for tables with prefix "', tbl_prefix, '".') AS message;
    SET out_status = 200;
END