# 库级冷热分离
#### 操作场景
在企业数据管理中，随着业务的快速发展，数据库中的数据量急剧增加，导致存储成本上升和查询性能下降。为解决这一问题，通常需要对不再活跃的数据进行归档处理，以降低存储成本并提高查询效率。然而，手动进行冷热数据分离操作复杂且容易出错。通过存储过程自动化冷热数据分离，可以显著减少手动操作，提高数据管理的效率和准确性。对整个库进行冷热分离场景时，支持通过存储过程进行冷热分离。根据是否有增量数据，分为以下两种场景：
- [通过存储过程对各个表依次进行冷数据归档]
  针对整个库中不再有增量数据的场景进行归档，将以库中所有表为对象（分区表即为分区），指导您通过DAS连接TaurusDB实例，利用存储过程对各表依次进行冷数据归档。为避免归档时间过长，建议整库不要超过50张表。
  
- [保留指定库下分区表最新的N个分区，通过存储过程对库中其余表进行冷数据归档]
  针对整个库中分区表存在增量数据的场景进行归档，将保留分区表中最新的N个分区不进行归档，以库中其余表为对象（分区表即为分区），指导您通过DAS连接TaurusDB实例，利用存储过程对各表依次进行冷数据归档。建议使用[INTERVAL RANGE](https://support.huaweicloud.com/kerneldesc-taurusdb/taurusdb_20_0051.html)分区功能自动拓展分区，结合自动设置冷表，将低频使用的分区的数据归档到OBS上。
  
 
#### 约束限制
- 使用存储过程对整个库进行冷热分离，要求内核版本号不低于2.0.72.251200。内核版本的查询方法请参见[如何查看云数据库 TaurusDB实例的版本号](https://support.huaweicloud.com/taurusdb_faq/taurusdb_faq_0141.html)。
- 归档的表要满足约束限制，具体请参考[归档冷表](https://support.huaweicloud.com/usermanual-taurusdb/taurusdb_03_0028.html)。
- 以下代码示例仅供参考，请根据业务场景进行测试和验证。
 
 #### 通过存储过程对各个表依次进行冷数据归档
1. [登录TaurusDB管理控制台](https://console.huaweicloud.com/gaussdbformysql/#/management/list)。
2. 单击管理控制台左上角的![](https://support.huaweicloud.com/bestpractice-taurusdb/figure/zh-cn_image_0000002671347251.png)，选择区域。
3. 在"实例管理"页面，选择目标实例，单击操作列的"登录"，进入数据管理服务实例登录界面。 
   您也可以在"实例管理"页面，单击目标实例名称，进入"实例概览"页面。在页面右上角，单击"登录"，进入数据管理服务实例登录界面。
   
   
4. 正确输入数据库用户名和密码，单击"测试连接"。
5. 测试连接通过后，单击"登录"，即可进入您的数据库并进行管理。
6. 单击"SQL操作 \> SQL窗口"，执行如下命令，创建存储过程。 
   确认存储过程中指定的数据库存在，如不存在，请首先[创建数据库](https://support.huaweicloud.com/usermanual-taurusdb/taurusdb_03_0121.html)并导入数据。
   ```
   DELIMITER //
   CREATE PROCEDURE archive_table_non_first_partition(
       IN dbName VARCHAR(64)  -- 输入参数:指定数据库名称
   )
   BEGIN
       -- ===================== 所有DECLARE必须放在最开头 =====================
       -- 1. 声明变量
       DECLARE tableName VARCHAR(64);       -- 存储表名
       DECLARE partitionName VARCHAR(64);   -- 存储分区名
       DECLARE done INT DEFAULT 0;           -- 游标循环结束标记
       DECLARE partitionCount INT DEFAULT 0;-- 单个表的分区总数
       DECLARE partitionIndex INT DEFAULT 0;-- 分区遍历索引(用于跳过首个分区)
       -- 2. 声明所有游标(提前集中声明,避免语法错误)
       -- 游标1:获取指定数据库下的所有分区表
       DECLARE cur_partition_tables CURSOR FOR
           SELECT DISTINCT t.TABLE_NAME
           FROM INFORMATION_SCHEMA.TABLES t
           INNER JOIN INFORMATION_SCHEMA.PARTITIONS p
               ON t.TABLE_SCHEMA = p.TABLE_SCHEMA
               AND t.TABLE_NAME = p.TABLE_NAME
           WHERE t.TABLE_SCHEMA = dbName
             AND t.TABLE_TYPE = 'BASE TABLE'
             AND p.PARTITION_NAME IS NOT NULL  -- 筛选分区表
           ORDER BY t.TABLE_NAME;
       -- 游标2:获取单个分区表的所有分区(按创建顺序排序)
       DECLARE cur_partitions CURSOR FOR
           SELECT PARTITION_NAME
           FROM INFORMATION_SCHEMA.PARTITIONS
           WHERE TABLE_SCHEMA = dbName
             AND TABLE_NAME = tableName
             AND PARTITION_NAME IS NOT NULL
           ORDER BY PARTITION_ORDINAL_POSITION;
       -- 游标3:获取指定数据库下的所有无分区表
       DECLARE cur_non_partition_tables CURSOR FOR
           SELECT t.TABLE_NAME
           FROM information_schema.TABLES t
           WHERE t.TABLE_SCHEMA = dbName
             AND t.TABLE_TYPE = 'BASE TABLE'
             AND NOT EXISTS (  -- 排除分区表
                 SELECT 1
                 FROM information_schema.PARTITIONS p
                 WHERE p.TABLE_SCHEMA = t.TABLE_SCHEMA
                   AND p.TABLE_NAME = t.TABLE_NAME
                   AND p.PARTITION_NAME IS NOT NULL
             );
       -- 3. 声明游标结束处理程序(仅需声明一次)
       DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = 1;
       -- ===================== 第一部分：处理分区表(非首个分区) =====================
       -- 步骤1:遍历所有分区表
       OPEN cur_partition_tables;
       partition_table_loop: LOOP
           FETCH cur_partition_tables INTO tableName;
           IF done = 1 THEN
               LEAVE partition_table_loop;
           END IF;
           -- 重置变量,统计当前表的分区总数
           SET done = 0;
           SET partitionIndex = 0;
           SELECT COUNT(DISTINCT PARTITION_NAME) INTO partitionCount
           FROM INFORMATION_SCHEMA.PARTITIONS
           WHERE TABLE_SCHEMA = dbName
             AND TABLE_NAME = tableName
             AND PARTITION_NAME IS NOT NULL;
           -- 跳过只有1个分区的表(无非首个分区可处理)
           IF partitionCount <= 1 THEN
               ITERATE partition_table_loop;
           END IF;
           -- 步骤2:遍历当前分区表的所有分区
           OPEN cur_partitions;
           partition_loop: LOOP
               FETCH cur_partitions INTO partitionName;
               IF done = 1 THEN
                   LEAVE partition_loop;
               END IF;
               -- 跳过首个分区
               SET partitionIndex = partitionIndex + 1;
               IF partitionIndex = 1 THEN
                   ITERATE partition_loop;
               END IF;
               -- 拼接并执行分区级归档SQL
               SET @sql = CONCAT(
                   'call dbms_schs.make_io_transfer(\'start\', \'',
                   dbName, '\', \'', tableName, '\', \'', partitionName, '\', \'\', \'obs\')'
               );
               PREPARE stmt FROM @sql;
               EXECUTE stmt;
               DEALLOCATE PREPARE stmt;
           END LOOP partition_loop;
           CLOSE cur_partitions;
       END LOOP partition_table_loop;
       CLOSE cur_partition_tables;
       -- ===================== 第二部分：处理无分区表 =====================
       -- 重置结束标记(避免继承上一轮游标状态)
       SET done = 0;
       -- 步骤1:遍历所有无分区表
       OPEN cur_non_partition_tables;
       non_partition_table_loop: LOOP
           FETCH cur_non_partition_tables INTO tableName;
           IF done = 1 THEN
               LEAVE non_partition_table_loop;
           END IF;
           -- 拼接并执行无分区表归档SQL(修复参数顺序,和分区表保持一致)
           SET @sql = CONCAT(
               'call dbms_schs.make_io_transfer(\'start\', \'',
               dbName, '\', \'', tableName, '\', \'\', \'\', \'obs\')'
           );
           PREPARE stmt FROM @sql;
           EXECUTE stmt;
           DEALLOCATE PREPARE stmt;
       END LOOP non_partition_table_loop;
       CLOSE cur_non_partition_tables;
   END //
   DELIMITER ;
   ```
   
   
7. （可选）执行SQL语句，确认当前库下的表类型。 
   以test库为例，test库下有分区表t1_p，t2_p, order_info，此外还有三个非分区表t1，t2，t3。
   ```
   SELECT TABLE_SCHEMA,TABLE_NAME,PARTITION_NAME from information_schema.PARTITIONS where TABLE_SCHEMA = 'test';
   ```
   ![](https://support.huaweicloud.com/bestpractice-taurusdb/figure/zh-cn_image_0000002661357565.png "点击放大")
   
   
8. 执行存储过程，归档指定库下的所有表。直到所有表归档成功后才会返回结果，整个过程可能耗时较长。 
   ```
   call archive_table_non_first_partition('test');
   ```
   ![](https://support.huaweicloud.com/bestpractice-taurusdb/figure/zh-cn_image_0000002631118264.png "点击放大")
   
   
9. 执行存储过程，查看冷数据归档信息。 
   sys.schs_show_all(*database* , *table* , *partition*)是TaurusDB内置的存储过程，三个参数分别表示库名、表名和分区名。
   - 如果不指定表名和分区名，则表示查询该库下的所有归档表。
   
   - 如果不指定分区名，则表示查询特定表的所有归档分区。
   
   
   例如：
   - 查询某个实例上所有冷表
     ```
     CALL sys.schs_show_all( "", "", "");
     ```
     
   
   - 查询库名为test的所有冷分区或者冷表
     ```
     CALL sys.schs_show_all( "test", "", "");
     ```
     
   
   - 查询库名为test，表名为table1的冷分区或者冷表情况
     ```
     CALL sys.schs_show_all( "test", "table1", "");
     ```
     
   
   
   因为TaurusDB冷热分离特性不支持归档分区表的首个分区，因此，表t1_p的首个分区p0，表t2_p的首个（即唯一）分区p0和表order_info的首个分区p_Beijing未进行归档。
   下面以test库为示例：
   ```
   call sys.schs_show_all('test', '', '');
   ```
   当status列显示为FINISH时，表示分区表归档成功。
   ![](https://support.huaweicloud.com/bestpractice-taurusdb/figure/zh-cn_image_0000002630958356.png "点击放大")
   
   
10. 确认磁盘使用量的指标值降低，表示冷表的存储空间已释放（监控采集可能存在延迟）。 
    ![](https://support.huaweicloud.com/bestpractice-taurusdb/figure/zh-cn_image_0000002661477505.png "点击放大")
    
    
    
 
 #### 保留指定库下分区表最新的N个分区，通过存储过程对库中其余表进行冷数据归档
1. [登录TaurusDB管理控制台](https://console.huaweicloud.com/gaussdbformysql/#/management/list)。
2. 单击管理控制台左上角的![](https://support.huaweicloud.com/bestpractice-taurusdb/figure/zh-cn_image_0000002671347251.png)，选择区域。
3. 在"实例管理"页面，选择目标实例，单击操作列的"登录"，进入数据管理服务实例登录界面。 
   您也可以在"实例管理"页面，单击目标实例名称，进入"实例概览"页面。在页面右上角，单击"登录"，进入数据管理服务实例登录界面。
   
   
4. 正确输入数据库用户名和密码，单击"测试连接"。
5. 测试连接通过后，单击"登录"，即可进入您的数据库并进行管理。
6. 单击"SQL操作 \> SQL窗口"，执行如下命令，创建存储过程。 
   确认存储过程中指定的数据库存在，如不存在，请首先[创建数据库](https://support.huaweicloud.com/usermanual-taurusdb/taurusdb_03_0121.html)并导入数据。
   ```
   DELIMITER //
   -- 定义单表存储过程，如已创建则跳过
   CREATE PROCEDURE archive_table_skip_first_and_retain_n(
       IN dbName VARCHAR(64),
       IN tableName VARCHAR(64),
       IN retainPartitionNum INT
   )
   BEGIN
       DECLARE partitionName VARCHAR(64);
       DECLARE done INT DEFAULT 0;
       DECLARE totalPartitionCount INT DEFAULT 0;
       DECLARE currentIndex INT DEFAULT 0;
       -- 游标声明
       DECLARE cur_partitions CURSOR FOR
           SELECT PARTITION_NAME
           FROM INFORMATION_SCHEMA.PARTITIONS
           WHERE TABLE_SCHEMA = dbName
             AND TABLE_NAME = tableName
             AND PARTITION_NAME IS NOT NULL
           ORDER BY PARTITION_ORDINAL_POSITION;
       -- 游标结束处理程序
       DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = 1;
       -- ===================== 全局统一处理：库名、表名大小写条件 =====================
       SET @schema_cond = '';
       SET @table_cond = '';
       -- 库名大小写处理
       IF dbName IS NOT NULL AND dbName <> '' THEN
           IF @@global.lower_case_table_names = 1 THEN
               SET @schema_cond = CONCAT(' AND LOWER(a.schema_name) = \'', LOWER(dbName), '\'');
           ELSE
               SET @schema_cond = CONCAT(' AND a.schema_name = \'', dbName, '\'');
           END IF;
       END IF;
       -- 表名大小写处理
       IF tableName IS NOT NULL AND tableName <> '' THEN
           IF @@global.lower_case_table_names = 1 THEN
               SET @table_cond = CONCAT(' AND LOWER(a.table_name) = \'', LOWER(tableName), '\'');
           ELSE
               SET @table_cond = CONCAT(' AND a.table_name = \'', tableName, '\'');
           END IF;
       END IF;
       -- ===================== 获取总分区数 =====================
       SELECT COUNT(DISTINCT PARTITION_NAME) INTO totalPartitionCount
       FROM INFORMATION_SCHEMA.PARTITIONS
       WHERE TABLE_SCHEMA = dbName
         AND TABLE_NAME = tableName
         AND PARTITION_NAME IS NOT NULL;
       -- ===================== 无分区表：先判断是否已归档，已归档则跳过 =====================
       IF totalPartitionCount = 0 THEN
           -- 检查无分区表是否已经归档（partition_name = ''）
           SET @check_sql = CONCAT(
               'SELECT COUNT(*) INTO @is_archived FROM mysql.schs_io_transfer a ',
               'INNER JOIN INFORMATION_SCHEMA.INNODB_TABLESPACES c ON a.space_id = c.SPACE AND c.NAME NOT LIKE \'__recyclebin__%\' ',
               'WHERE a.space_id != 4294967295 ',
               'AND a.target_type = \'OBS\' ',
               'AND a.transfer_status = \'FINISH\' ',
               'AND a.partition_name = \'\' ',
               @schema_cond,
               @table_cond
           );
           PREPARE check_stmt FROM @check_sql;
           EXECUTE check_stmt;
           DEALLOCATE PREPARE check_stmt;
           -- 已归档：直接跳过，不执行
           IF @is_archived > 0 THEN
               SET done = 1;
           ELSE
               -- 未归档：执行归档
               SET @sql = CONCAT('call dbms_schs.make_io_transfer(\'start\',\'',
                   dbName,'\',\'',tableName,'\',\'\',\'\',\'obs\')');
               PREPARE stmt FROM @sql;
               EXECUTE stmt;
               DEALLOCATE PREPARE stmt;
               SET done = 1;
           END IF;
       END IF;
       -- ===================== 分区表处理 =====================
       IF done != 1 THEN
           OPEN cur_partitions;
           partition_loop: LOOP
               FETCH cur_partitions INTO partitionName;
               IF done = 1 THEN LEAVE partition_loop; END IF;
               SET currentIndex = currentIndex + 1;
               -- 1. 跳过第一个分区
               IF currentIndex = 1 THEN
                   ITERATE partition_loop;
               END IF;
               -- 2. 保留最新N个分区
               IF currentIndex > (totalPartitionCount - retainPartitionNum) THEN
                   ITERATE partition_loop;
               END IF;
               -- 3. 判断是否已归档
               SET @check_sql = CONCAT(
                   'SELECT COUNT(*) INTO @is_archived FROM mysql.schs_io_transfer a ',
                   'INNER JOIN INFORMATION_SCHEMA.PARTITIONS b ON a.schema_name = b.TABLE_SCHEMA AND a.table_name = b.TABLE_NAME ',
                   'AND (a.partition_name = b.PARTITION_NAME OR (a.partition_name = \'\' AND b.PARTITION_NAME IS NULL)) ',
                   'INNER JOIN INFORMATION_SCHEMA.INNODB_TABLESPACES c ON a.space_id = c.SPACE AND c.NAME NOT LIKE \'__recyclebin__%\' ',
                   'WHERE a.space_id != 4294967295 ',
                   'AND a.target_type = \'OBS\' ',
                   'AND a.transfer_status = \'FINISH\' ',
                   'AND a.partition_name = \'', partitionName, '\' ',
                   @schema_cond,
                   @table_cond
               );
               PREPARE check_stmt FROM @check_sql;
               EXECUTE check_stmt;
               DEALLOCATE PREPARE check_stmt;
               -- 已归档则跳过
               IF @is_archived > 0 THEN
                   ITERATE partition_loop;
               END IF;
               -- 执行归档
               SET @sql = CONCAT('call dbms_schs.make_io_transfer(\'start\',\'',
                   dbName,'\',\'',tableName,'\',\'',partitionName,'\',\'\',\'obs\')');
               PREPARE stmt FROM @sql;
               EXECUTE stmt;
               DEALLOCATE PREPARE stmt;
           END LOOP partition_loop;
           CLOSE cur_partitions;
       END IF;
       -- 清理会话变量
       SET @schema_cond = NULL;
       SET @table_cond = NULL;
       SET @check_sql = NULL;
       SET @is_archived = NULL;
   END //
   CREATE PROCEDURE archive_database_skip_first_and_retain_n(
       IN dbName VARCHAR(64),        -- 要归档的整个数据库名
       IN retainPartitionNum INT     -- 每张表保留最新N个分区
   )
   BEGIN
       DECLARE tableName VARCHAR(64);
       DECLARE done INT DEFAULT 0;
       -- 游标：遍历当前库下【所有业务表】（分区表 + 普通表）
       DECLARE cur_all_tables CURSOR FOR
           SELECT TABLE_NAME
           FROM INFORMATION_SCHEMA.TABLES
           WHERE TABLE_SCHEMA = dbName
             AND TABLE_TYPE = 'BASE TABLE'  -- 只查业务表
           ORDER BY TABLE_NAME;
       -- 游标结束处理
       DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = 1;
       -- 打开游标，遍历所有表
       OPEN cur_all_tables;
       table_loop: LOOP
           FETCH cur_all_tables INTO tableName;
           IF done = 1 THEN
               LEAVE table_loop;
           END IF;
           -- ===================== 核心：直接复用单表存储过程 =====================
           -- 对每张表 自动执行：跳过首分区 + 保留N个 + 不重复归档
           CALL archive_table_skip_first_and_retain_n(
               dbName,
               tableName,
               retainPartitionNum
           );
       END LOOP table_loop;
       CLOSE cur_all_tables;
       -- 清理变量
       SET done = 0;
   END //
   DELIMITER ;
   ```
   
   
7. （可选）查看当前库中表归档情况 
   下面以test库为例，库中存在分区表sales，t1_p，t2_p：
   ![](https://support.huaweicloud.com/bestpractice-taurusdb/figure/zh-cn_image_0000002661467901.png "点击放大")
   当前已归档情况：
   ![](https://support.huaweicloud.com/bestpractice-taurusdb/figure/zh-cn_image_0000002661347967.png "点击放大")
   
   
8. 创建EVENT定时调用存储过程。 
   在DAS上创建如下脚本，设定从4月1日开始，每月1号01:00对test库中的表进行归档，归档操作将针对除首个和最新的2个分区外的所有分区进行，并且每间隔一个月会检查表是否创建了新的分区，对最新的2个分区外的旧分区进行归档。
   ```
   CREATE EVENT IF NOT EXISTS event_archive_test_monthly
   ON SCHEDULE
   EVERY 1 MONTH
   STARTS '2026-04-01 01:00:00'
   DO
   CALL archive_database_skip_first_and_retain_n('test', 2);
   ```
   
   
9. （可选）查看对应表的归档状态。 
   插入数据，结合[INTERVAL RANGE](https://support.huaweicloud.com/kerneldesc-taurusdb/taurusdb_20_0051.html)自动创建分区。
   ![](https://support.huaweicloud.com/bestpractice-taurusdb/figure/zh-cn_image_0000002630948768.png "点击放大")
   新分区_p20220601000000,_p20220701000000创建后，_p20220401000000,_p20220501000000被自动归档。
   ![](https://support.huaweicloud.com/bestpractice-taurusdb/figure/zh-cn_image_0000002631108690.png "点击放大")