跳到主要内容

宽表 → 窄表数据迁移指南（PostgreSQL）

目标：把某个表单（FormEntity）的数据从宽表模式（form_entity_data 的 valueN / labelN 列）平滑迁移到窄表模式（每字段一列），并在测试/生产环境安全切换。

本指南同时可作为通用数据迁移的模板：结构化的「准备 → 建表 → 迁移 → 校验 → 切换 → 回滚」流程。当前迁移脚本只支持 PostgreSQL。

配套 SQL 脚本位于后端仓库 ceta-central/doc/wide-to-normalized-migration/：

脚本作用
migrate_wide_to_normalized_pg.sql主迁移函数（搬 valueN 列数据）
migrate_longtext_to_normalized_pg.sql长文本迁移函数（从子表回填内联列）
verify_migration_pg.sql校验函数（行数 / 主键集合 / 抽样值比对）

占位符说明（执行前先替换）​

下文 SQL 里的尖括号占位符都需要替换成你的实际值：

占位符含义怎么获取
<feid>要迁移的表单 ID（form_entity.id，即 FormEntity 的数字主键，不是 token）见下方查询
<yyyymmdd>备份表名用的日期后缀（如 20260710）填当天日期
<standalone_table_name>目标窄表名取 form_entity.standalone_table_name
<old_split_table>旧分表宽表名（仅 B→C 场景）分表宽表迁移时的旧表名
<new_normalized_table>新窄表名（仅 B→C 场景，须不同于旧分表名）自己规划

查 <feid>（按 PBC token + 表单 token 定位）：

SELECT id, token, standalone, standalone_table_name
  FROM form_entity
 WHERE pbc_token = '<你的PBC token>' AND token = '<你的表单 token>';

得到的 id 就是要代入的 <feid>。例如查出 id = 13257，则 §5.1 写成 SELECT migrate_form_entity_wide_to_normalized(13257, true, true, NULL);。

0. TL;DR（操作顺序）​

  1. 停写该表单数据（维护窗口，或确认无写入）。
  2. 备份：先备份源数据 + 相关元数据（§4）。备份是纯读、零风险，且必须在任何「改动性操作」（建表会改 schema、回填 normalized_storage_field）之前完成，才能整体回滚。
  3. 建窄表结构：调用平台已有的 createOrUpdateTable（见开启分表（窄表）），它会建自然表 + 回填 form_entity_field.normalized_storage_field。
  4. 安装迁移函数（首次必做）：在目标库执行三个 .sql 脚本各一次，把函数建进库，否则调用会报 function ... does not exist（§5.0）。
  5. 迁移数据：执行 migrate_wide_to_normalized_pg.sql 里的函数，先 dry_run=true 预演，再正式跑（§5）。
    • 全局宽表（A）→ 窄表：p_source_table 传 NULL。
    • 分表宽表（B）→ 窄表：p_source_table 传旧分表名（见 §5.4）。
  6. 迁移长文本（若表单含 TEXTAREA / RICH_TEXT / LONG_TEXT 字段）：执行 migrate_longtext_to_normalized_pg.sql，把子表 form_entity_data_text 的长文本回填到窄表内联列（§5.5）。主迁移只搬 valueN，长文本必须单独跑这一步，否则长文本丢失。
  7. 校验：执行 verify_migration_pg.sql，比对行数与抽样值（§6）。
  8. 切换开关：确认 standalone=1 / standalone_table_normalized=1 / standalone_table_created=1 + 清 FormEntity 缓存（§7）。
  9. 灰度验证：读写走窄表，观察。
  10. 稳定后再清理源表旧数据（§8，延迟执行、可回滚）。

顺序说明（为什么备份在建表之前）：createOrUpdateTable 会新建表并 UPDATE form_entity_field.normalized_storage_field——这是第一个改动性操作。备份必须在它之前，且备份范围要包含被改动的元数据（见 §4），否则无法完整回滚。

1. 关键不变量（迁移必须保持）​

  • id 不变（最重要）：窄表行必须沿用宽表行的 id。CETA 里大量引用关系（外部引用、reciprocal link、附件 / 长文本 / ACL 子表）都按 form_entity_data_id 指向数据行的 id，一旦 id 变化，这些引用会全部错乱。因此迁移绝不能让窄表用自增序列重新生成 id。
  • 系统列原样搬：deleted / object_version / created_time / user_id 等全部保留。
  • 软删数据也要迁（deleted=1 的行）：否则历史/审计对不上，且 id 复用会冲突。

脚本如何保证 id 不变（migrate_wide_to_normalized_pg.sql）：

  1. id 作为系统列被显式映射（INSERT ... SELECT src.id ...），不走窄表自增；
  2. 用 ON CONFLICT (id) DO NOTHING，既沿用原 id 又保证幂等（可重跑不重复插入）；
  3. 迁移后 setval 把窄表自增序列推进到 max(id)（§5.3），避免后续新增撞已用 id。

校验脚本 verify_migration_pg.sql 会额外比对 id_only_in_source / id_only_in_target（都应为 0），并按同一 id 抽样比对字段值，确保 id 集合完全一致。此外长文本回填依赖 id 不变（靠 窄表.id = form_entity_data_text.form_entity_data_id 关联），所以必须先跑主迁移再跑长文本迁移。

三种存储模式、字段列映射、长文本存储差异等背景，见上一节 开启分表（窄表）。

2. 迁移前置检查清单​

在跑迁移前，逐项确认：

  • 目标环境是 PostgreSQL。
  • 该表单 form_entity.standalone_table_name 已设置且合法（非保留表名）。
  • 已通过平台「开启窄表 + 建表」，standalone=1、standalone_table_normalized=1、standalone_table_created=1 均已由后端自动置 1（无需手动改 SQL；见开启分表（窄表））。
  • 已备份源数据 + 元数据（§4）——先于建表。
  • 目标窄表已在库中存在（建表成功的标志）。
  • form_entity_field 里每个业务字段的 normalized_storage_field 非空（开启窄表时后端会回填；若为空说明窄表没开或字段是后加的）。
  • 维护窗口内，该表单无并发写入（或已停写）。
  • 若是分表宽表（B→C）：目标窄表名与旧分表名不同（见 §5.4），且旧分表名已记录用于迁移源。

3. 建窄表结构​

窄表结构必须由平台代码 FormEntityDataStandaloneTableNormalizedService.createOrUpdateTable(formEntity) 生成，不要手写 DDL——因为列名、label 列、类型映射都由代码统一决定，且该方法会回填 form_entity_field.normalized_storage_field，这是迁移 SQL 依赖的映射来源。

详细步骤见 开启分表（窄表）。建表后确认：

  • 库中存在表 <standalone_table_name>，含 id + 全部系统列 + 每字段业务列。
  • SELECT token, storage_field, normalized_storage_field, type FROM form_entity_field WHERE form_entity_id = <feid> 中 normalized_storage_field 均非空。

4. 备份（在建表之前）​

备份是纯读操作、零风险，必须在建表（§3，第一个改动性操作）之前完成。范围要覆盖两类：

4.1 源数据

-- A. 全局宽表源
CREATE TABLE bak_fed_<feid>_<yyyymmdd> AS
SELECT * FROM form_entity_data WHERE form_entity_id = <feid>;

-- B. 分表宽表源（旧分表名）
CREATE TABLE bak_split_<feid>_<yyyymmdd> AS
SELECT * FROM <old_split_table>;

4.2 元数据（会被建表/切换步骤 UPDATE，必须备份才能回滚）

-- form_entity 的开关 + standalone_table_name（切换步骤会改）
CREATE TABLE bak_fe_<feid>_<yyyymmdd> AS
SELECT * FROM form_entity WHERE id = <feid>;

-- form_entity_field 的 normalized_storage_field（建表步骤会回填/改）
CREATE TABLE bak_fef_<feid>_<yyyymmdd> AS
SELECT * FROM form_entity_field WHERE form_entity_id = <feid>;

为什么元数据也要备份：createOrUpdateTable 会 UPDATE form_entity_field.normalized_storage_field，切换步骤会 UPDATE form_entity 的标志位。只备源数据、不备元数据的话，回滚时无法还原字段映射与开关。

5. 迁移数据​

核心脚本：migrate_wide_to_normalized_pg.sql。它提供一个 PL/pgSQL 函数 migrate_form_entity_wide_to_normalized(p_form_entity_id, p_dry_run, p_include_deleted, p_source_table)：

  • 源表可选：p_source_table=NULL 用全局宽表 form_entity_data（A→C）；传旧分表名则从分表迁（B→C）。
  • 动态列映射：读 form_entity_field，把每个字段的源宽表列（storage_field 及其 label）映射到窄表列（normalized_storage_field 及其 _label），并用 information_schema 双向校验列存在性，不存在的列自动跳过。
  • 系统列：id / form_entity_id / deleted / created_time / updated_time / user_id / updated_user_id / flow_instance_id / thread_id / node_id / object_version 原样搬。
  • id 保留 + 幂等：INSERT ... SELECT ... ON CONFLICT (id) DO NOTHING，可安全重跑；已迁移的行不会重复。
  • 源≠目标保护：若源表名等于目标窄表名，函数直接报错（防止 B→C 用同名表自己插自己，见 §5.4）。
  • dry_run：p_dry_run=true 只打印将要执行的 SQL 和预计行数，不写数据。
  • 软删：p_include_deleted=true（默认）连 deleted=1 的行一起迁。

5.0 安装迁移函数（首次必做）​

⚠️ 这几个函数不是数据库自带的，必须先在目标库执行一次建函数脚本，否则直接调用会报 function ... does not exist。

在存放 form_entity_data 的库（form 引擎的库，且能访问 form_entity / form_entity_field 元数据表）上，依次执行三个脚本各一次，把函数建进库：

# 用 psql（把连接串换成你的库）
psql "<connection-string>" -f migrate_wide_to_normalized_pg.sql
psql "<connection-string>" -f migrate_longtext_to_normalized_pg.sql   # 含长文本字段时需要
psql "<connection-string>" -f verify_migration_pg.sql

或在任意 SQL 客户端里，把每个 .sql 文件（见附录 C）整段内容粘贴执行一次。每个文件就是一条 CREATE OR REPLACE FUNCTION，执行成功后函数即常驻该库，后续 §5.1 起的 SELECT 才能调用。

前提：目标库是 PostgreSQL（这些函数只支持 PG）。建函数本身不写业务数据，可安全执行；重复执行会覆盖同名函数（CREATE OR REPLACE），无副作用。

5.1 预演（务必先做）​

-- A. 全局宽表源
SELECT migrate_form_entity_wide_to_normalized(<feid>, true, true, NULL);
-- B. 分表宽表源（旧分表名）
SELECT migrate_form_entity_wide_to_normalized(<feid>, true, true, 'old_split_table_xxx');

检查输出的 INSERT 语句列映射是否符合预期（列数、label 列配对、系统列齐全）。

5.2 正式迁移​

BEGIN;
-- A→C
SELECT migrate_form_entity_wide_to_normalized(<feid>, false, true, NULL);
-- 或 B→C
-- SELECT migrate_form_entity_wide_to_normalized(<feid>, false, true, 'old_split_table_xxx');
-- 观察返回的迁移行数，与 §6 校验通过后再 COMMIT
COMMIT;

5.3 迁移后修复自增序列​

窄表 id 是自增列，但迁移时显式插入了宽表 id。必须把序列推进到当前最大 id，避免后续新增撞主键：

SELECT setval(
  pg_get_serial_sequence('<standalone_table_name>', 'id'),
  (SELECT COALESCE(MAX(id), 1) FROM <standalone_table_name>)
);

（脚本中的函数在非 dry_run 且迁移行数 > 0 时会自动执行这一步。）

5.4 分表宽表（B→C）的表名处理（重要）​

分表宽表（B）的数据在旧的独立表里，列结构仍是 valueN / labelN。迁到窄表（C）的难点是表名：

  • createOrUpdateTable 依据 form_entity.standalone_table_name 建/改表。若该值仍指向旧分表，它会往旧分表里追加 normalized 列（而不是新建），结果同一张表里既有 valueN 又有 applicant_name，且运行时 normalized 分支只认新列——旧数据读不到、结构混乱。
  • 因此 B→C 必须让新窄表用一个不同于旧分表的新表名。

推荐流程：

  1. 记下旧分表名 old_split_table（迁移源）。

  2. 把表单的 standalone_table_name 改成新名字（如 old_split_table_n），然后通过平台「开启窄表」触发后端流程——后端会自动把 standalone_table_normalized 置 1 并回填字段映射。

    表名本身通常在平台/表单设置里改；standalone_table_normalized 由后端在开启窄表时置 1，不需要手动 UPDATE。若你确实要在数据库层临时校正表名，注意这只是辅助操作，标志位仍以后端流程为准。

  3. 走 §3 通过平台「开启窄表 + 建表」建新窄表（新表名，全套 normalized 列），建表成功后后端自动把 standalone_table_created 置 1。

  4. 迁移时 p_source_table 传旧分表名：

    SELECT migrate_form_entity_wide_to_normalized(<feid>, false, true, 'old_split_table');

    函数内有「源表≠目标表」硬校验，若你忘了改名（新旧同名），会直接报错阻止误操作。

  5. 校验、切换、稳定后清理旧分表（§8）。

若产品上要求「表名不变」，可在稳定后把新窄表 RENAME 回原名（需同步更新 standalone_table_name），但这会增加复杂度和停机，建议优先接受新表名。

5.5 迁移长文本（TEXTAREA / RICH_TEXT / LONG_TEXT，必做）​

若表单含长文本类字段，主迁移之后必须再跑长文本迁移，否则这些字段在窄表里是空的。脚本：migrate_longtext_to_normalized_pg.sql，函数 migrate_form_entity_longtext_to_normalized(feid, dry_run)。

它遍历表单的长文本字段，把子表 form_entity_data_text 的主值（locale IS NULL）UPDATE 回填到窄表的 normalized_storage_field 内联列（按 窄表.id = form_entity_data_text.form_entity_data_id 关联）。

-- 预演
SELECT migrate_form_entity_longtext_to_normalized(<feid>, true);
-- 正式（放同一迁移事务里，紧接主迁移之后）
BEGIN;
SELECT migrate_form_entity_wide_to_normalized(<feid>, false, true, NULL);  -- 主迁移
SELECT migrate_form_entity_longtext_to_normalized(<feid>, false);          -- 长文本回填
COMMIT;

要点：

  • 依赖主迁移已完成（窄表已有对应 id 的行），本步是 UPDATE 回填，不是 INSERT。
  • 幂等：纯 UPDATE，可重跑。
  • i18n 译文不迁：form_entity_data_text 里 locale 非空的行两种模式都用子表，保持不动。
  • A→C 和 B→C 都要跑这一步（长文本子表 form_entity_data_text 是全局的，与源是宽表还是分表无关）。

6. 校验​

脚本：verify_migration_pg.sql。同样支持 p_source_table 参数（B→C 传旧分表名）：

-- A→C
SELECT * FROM verify_form_entity_migration(<feid>, 20, NULL);
-- B→C
SELECT * FROM verify_form_entity_migration(<feid>, 20, 'old_split_table');

校验点：

  1. row_count：目标窄表行数 == 源表该 form_entity_id 行数（含软删）。
  2. id_only_in_source / id_only_in_target：源/目标 id 集合完全相同（无缺、无多）。
  3. value_mismatch：随机抽 N 行，逐字段比对源列值与窄表列值（含 label），不一致会输出明细。
  4. longtext_mismatch：逐长文本字段比对子表 form_entity_data_text 主值与窄表内联列（0 为通过）。若 FAIL，通常是 §5.5 长文本迁移漏跑。

任一 status != OK，不要切换开关，回滚（§9）后排查。

7. 切换到窄表模式​

运行时是否读写窄表，由 FormEntity.isNormalizedTableCreated() 判定（即 standalone_table_normalized=1 && standalone_table_created=1）。

正常情况下这三个标志位在「开启窄表 + 建表」时就已由后端自动置好（见开启分表（窄表）），所以数据校验通过后通常不需要再改标志位，直接进入清缓存 + 冒烟即可。

只有当你确认标志位因异常没置到位时，才手动校正（排查用）：

-- 仅排查用：确认/补齐三个标志位（正常流程由后端自动完成）
SELECT id, standalone, standalone_table_normalized, standalone_table_created
  FROM form_entity WHERE id = <feid>;

-- 如确有缺失再手动补（谨慎）
-- UPDATE form_entity
--    SET standalone = 1, standalone_table_normalized = 1, standalone_table_created = 1
--  WHERE id = <feid>;

关键：确认标志位后必须让应用清除 FormEntity 缓存（平台对 FormEntity 有缓存），否则运行时仍按旧标志走宽表。清缓存方式取决于平台（重启该服务 / 调用缓存失效接口 / 等缓存 TTL 到期）。切换应在维护窗口做。

切换后做读写冒烟：列表查询、详情、新增、编辑、删除各一次，确认都落到窄表。

8. 清理宽表旧数据（延迟、可回滚）​

不要在切换后立即删源数据。 建议观察数天，确认窄表稳定后再清理：

-- A→C：物理删除该表单在全局宽表里的行（已备份 + 窄表已验证稳定后）
DELETE FROM form_entity_data WHERE form_entity_id = <feid>;

-- B→C：旧分表整表可直接 DROP（已备份 + 稳定后）
-- DROP TABLE <old_split_table>;

-- 长文本子表：长文本主值已内联到窄表，子表里该表单的【非 i18n 主值行 (locale IS NULL)】可清理。
-- ⚠️ 保留 locale 非空的 i18n 译文行（窄表模式仍用子表存 i18n）！
-- DELETE FROM form_entity_data_text WHERE form_entity_id = <feid> AND locale IS NULL;

清理前保留 §4 的备份表至少一个发布周期。长文本子表清理尤其谨慎：只删 locale IS NULL 的主值行，绝不能删 locale 非空的 i18n 译文行。

9. 回滚方案​

迁移/切换出问题时：

说明：正常「开启窄表」时这些标志位是后端自动置 1 的。回滚是唯一需要手动把标志位改回去的场景——因为你要主动把表单退回宽表/分表模式，所以下面的 UPDATE form_entity SET ... 是有意为之的操作。

  • 切换前（还没改开关）：直接清掉新窄表数据重来即可，源表（全局宽表 / 旧分表）未动，零影响。

    DELETE FROM <new_normalized_table> WHERE form_entity_id = <feid>;

    从元数据备份还原字段映射（建表回填过 normalized_storage_field）：

    UPDATE form_entity_field f
       SET normalized_storage_field = b.normalized_storage_field
      FROM bak_fef_<feid>_<yyyymmdd> b
     WHERE f.id = b.id;

  • 切换后发现问题：把开关切回源模式 + 清缓存，源数据仍在（只要还没执行 §8 清理）。

    -- A→C 回滚：切回全局宽表
    UPDATE form_entity SET standalone = 0, standalone_table_normalized = 0, standalone_table_created = 0 WHERE id = <feid>;

    -- B→C 回滚：切回旧分表（standalone 保持 1，normalized 关掉，表名指回旧分表）
    UPDATE form_entity
       SET standalone_table_normalized = 0,
           standalone_table_created = 1,
           standalone_table_name = '<old_split_table>'
     WHERE id = <feid>;
    -- 两种情况都要接着清 FormEntity 缓存

    注意：切回源模式后，窄表期间产生的增量写入不会自动回灌源表。所以切换后的验证要快，一旦回滚，需人工评估窄表期间的增量数据（这也是为什么切换要放在低峰/维护窗口）。元数据可用 §4.2 备份还原。

附录 A：涉及的表与关键列​

表关键列说明
form_entityid, standalone, standalone_table_name, standalone_table_normalized, standalone_table_created表单定义 + 窄表开关
form_entity_fieldform_entity_id, token, type, storage_field, normalized_storage_field字段↔宽表列/窄表列映射来源
form_entity_dataid, form_entity_id, deleted, value1..N, label1..N, 系统列全局宽表（A 源）
<old_split_table>同宽表列结构（value1..N / label1..N + 系统列）分表宽表（B 源）
<standalone_table_name>id, form_entity_id, deleted, <normalized cols>, <..._label>, 系统列窄表（C 目标）
form_entity_data_textid, form_entity_data_id, form_entity_field_id, long_text, locale长文本子表：宽表模式存长文本主值（locale IS NULL）+ 两种模式的 i18n 译文（locale 非空）

附录 B：label/value 字段类型​

有 _label 列（value + label 两列）的字段类型：SELECT、ACL、CASCADER、TREE_SELECT，以及 DATE。其余字段类型只有单个 value 列。LOOKUP 字段无存储列（跳过）。

附录 C：迁移脚本全文​

以下为配套的三个 PostgreSQL 脚本全文，可直接复制到目标库执行。原始文件位于后端仓库 ceta-central/doc/wide-to-normalized-migration/，以仓库版本为准。

migrate_wide_to_normalized_pg.sql — 主迁移函数（搬 valueN 列数据）
-- =============================================================================
-- 宽表 -> 窄表 数据迁移函数 (PostgreSQL only)
--
-- 覆盖两种源:
--   A. 全局宽表 form_entity_data                       -> 窄表 (p_source_table 传 NULL, 默认)
--   B. 分表宽表 (旧的 standalone_table_name, 列仍是 valueN) -> 窄表 (p_source_table 传旧分表名)
-- 两种源的列结构都是 valueN/labelN, 字段映射一致 (storage_field -> normalized_storage_field),
-- 唯一差别是"源表名"和"是否带 form_entity_id 过滤"(见下)。
--
-- 用法:
--   1. 先在目标库执行本文件, 创建函数。
--   2. 预演(不写数据):
--        -- A. 全局宽表源
--        SELECT migrate_form_entity_wide_to_normalized(<feid>, true, true, NULL);
--        -- B. 分表宽表源 (旧分表名)
--        SELECT migrate_form_entity_wide_to_normalized(<feid>, true, true, 'old_split_table_xxx');
--   3. 正式迁移(放事务里, 校验通过再 COMMIT):
--        BEGIN;
--        SELECT migrate_form_entity_wide_to_normalized(<feid>, false, true, NULL);
--        COMMIT;
--
-- 参数:
--   p_form_entity_id  要迁移的表单 id
--   p_dry_run         true=只打印将执行的 SQL 与源行数, 不写入; false=真正迁移
--   p_include_deleted true=连 deleted=1 的历史行一起迁(推荐), false=只迁 deleted=0
--   p_source_table    源表名; NULL=全局宽表 form_entity_data; 否则=旧分表宽表名(B->C)
--
-- 重要:
--   * 目标窄表名从 form_entity.standalone_table_name 读取。
--   * B->C 场景: 新窄表必须用【不同于旧分表】的表名 (先给表单改 standalone_table_name 再建表),
--     否则本函数会因"源表==目标表"而直接报错 (防止自己插自己)。
--   * 列映射从 form_entity_field 动态推导, 用 information_schema 双向校验列存在, 不存在则跳过。
--   * 【id 不变·最重要】窄表行必须沿用宽表行的 id: CETA 大量引用(外部引用/reciprocal link/
--     附件/长文本/ACL 子表)都按 form_entity_data_id 指向数据行 id, id 一旦变化引用全部错乱。
--     本函数把 id 作为系统列【显式】映射(SELECT src.id), 绝不走窄表自增; 幂等
--     (ON CONFLICT (id) DO NOTHING); 迁移后 setval 把 id 序列推进到 max(id), 防后续新增撞主键。
--   * 保留全部系统列(deleted/object_version/created_time/user_id 等)。
--
-- 返回: 文本报告。
-- =============================================================================

CREATE OR REPLACE FUNCTION migrate_form_entity_wide_to_normalized(
    p_form_entity_id  BIGINT,
    p_dry_run         BOOLEAN DEFAULT true,
    p_include_deleted BOOLEAN DEFAULT true,
    p_source_table    TEXT    DEFAULT NULL
) RETURNS TEXT
LANGUAGE plpgsql
AS $func$
DECLARE
    v_target_table   TEXT;
    v_source_table   TEXT;
    v_normalized     INT;
    v_src_count      BIGINT;
    v_migrated       BIGINT := 0;
    v_select_cols    TEXT := '';
    v_insert_cols    TEXT := '';
    v_sql            TEXT;
    v_where          TEXT;
    v_seq            TEXT;
    v_src_has_feid   BOOLEAN;
    r                RECORD;
    c_system_cols    TEXT[] := ARRAY[
        'id','form_entity_id','deleted','created_time','updated_time',
        'user_id','updated_user_id','flow_instance_id','thread_id','node_id','object_version'
    ];
    v_col            TEXT;
    v_has_src        BOOLEAN;
    v_has_dst        BOOLEAN;
BEGIN
    -- 1. 读取表单窄表配置 (目标表)
    SELECT standalone_table_name, standalone_table_normalized
      INTO v_target_table, v_normalized
      FROM form_entity
     WHERE id = p_form_entity_id;

    IF v_target_table IS NULL OR length(trim(v_target_table)) = 0 THEN
        RAISE EXCEPTION 'form_entity % has no standalone_table_name; build the normalized table first', p_form_entity_id;
    END IF;
    IF v_normalized IS DISTINCT FROM 1 THEN
        RAISE EXCEPTION 'form_entity % is not marked standalone_table_normalized=1', p_form_entity_id;
    END IF;

    -- 2. 解析源表: NULL=全局宽表; 否则=分表宽表
    v_source_table := lower(COALESCE(NULLIF(trim(p_source_table), ''), 'form_entity_data'));
    v_target_table := lower(v_target_table);

    -- 2a. 源表 != 目标表 (防止 B->C 用同名表自己插自己)
    IF v_source_table = v_target_table THEN
        RAISE EXCEPTION 'source table (%) equals target table (%); for split(B)->normalized(C), give the form a NEW standalone_table_name and build it before migrating',
            v_source_table, v_target_table;
    END IF;

    -- 2b. 源/目标表都必须存在
    IF NOT EXISTS (SELECT 1 FROM information_schema.tables
                    WHERE table_schema=current_schema() AND table_name=v_source_table) THEN
        RAISE EXCEPTION 'source table "%" does not exist', v_source_table;
    END IF;
    IF NOT EXISTS (SELECT 1 FROM information_schema.tables
                    WHERE table_schema=current_schema() AND table_name=v_target_table) THEN
        RAISE EXCEPTION 'target normalized table "%" does not exist; run createOrUpdateTable first', v_target_table;
    END IF;

    -- 源表是否有 form_entity_id 列 (全局宽表和分表宽表都有; 保险起见判定一下)
    SELECT EXISTS(SELECT 1 FROM information_schema.columns
                   WHERE table_schema=current_schema() AND table_name=v_source_table AND column_name='form_entity_id')
      INTO v_src_has_feid;

    -- 3. 系统列映射 (两边同名, 双向校验)
    FOREACH v_col IN ARRAY c_system_cols LOOP
        SELECT EXISTS(SELECT 1 FROM information_schema.columns
                       WHERE table_schema=current_schema() AND table_name=v_source_table AND column_name=v_col) INTO v_has_src;
        SELECT EXISTS(SELECT 1 FROM information_schema.columns
                       WHERE table_schema=current_schema() AND table_name=v_target_table AND column_name=v_col) INTO v_has_dst;
        IF v_has_src AND v_has_dst THEN
            v_select_cols := v_select_cols || format('src.%I, ', v_col);
            v_insert_cols := v_insert_cols || format('%I, ', v_col);
        END IF;
    END LOOP;

    -- 4. 业务字段列映射: 源 storage_field(+label) -> 目标 normalized_storage_field(+_label)
    FOR r IN
        SELECT token, type, storage_field, normalized_storage_field
          FROM form_entity_field
         WHERE form_entity_id = p_form_entity_id
           AND type <> 'LOOKUP'
           AND storage_field IS NOT NULL
           AND normalized_storage_field IS NOT NULL
    LOOP
        SELECT EXISTS(SELECT 1 FROM information_schema.columns
                       WHERE table_schema=current_schema() AND table_name=v_source_table AND column_name=lower(r.storage_field)) INTO v_has_src;
        SELECT EXISTS(SELECT 1 FROM information_schema.columns
                       WHERE table_schema=current_schema() AND table_name=v_target_table AND column_name=lower(r.normalized_storage_field)) INTO v_has_dst;
        IF v_has_src AND v_has_dst THEN
            v_select_cols := v_select_cols || format('src.%I, ', lower(r.storage_field));
            v_insert_cols := v_insert_cols || format('%I, ', lower(r.normalized_storage_field));
        ELSE
            RAISE NOTICE 'skip value column for token % (src % exists=%, dst % exists=%)',
                r.token, r.storage_field, v_has_src, r.normalized_storage_field, v_has_dst;
        END IF;

        -- label 列: 仅 SELECT/ACL/CASCADER/TREE_SELECT/DATE
        IF r.type IN ('SELECT','ACL','CASCADER','TREE_SELECT','DATE') THEN
            DECLARE
                v_src_label TEXT := regexp_replace(lower(r.storage_field), 'value', 'label');
                v_dst_label TEXT := lower(r.normalized_storage_field) || '_label';
            BEGIN
                SELECT EXISTS(SELECT 1 FROM information_schema.columns
                               WHERE table_schema=current_schema() AND table_name=v_source_table AND column_name=v_src_label) INTO v_has_src;
                SELECT EXISTS(SELECT 1 FROM information_schema.columns
                               WHERE table_schema=current_schema() AND table_name=v_target_table AND column_name=v_dst_label) INTO v_has_dst;
                IF v_has_src AND v_has_dst THEN
                    v_select_cols := v_select_cols || format('src.%I, ', v_src_label);
                    v_insert_cols := v_insert_cols || format('%I, ', v_dst_label);
                ELSE
                    RAISE NOTICE 'skip label column for token % (src % exists=%, dst % exists=%)',
                        r.token, v_src_label, v_has_src, v_dst_label, v_has_dst;
                END IF;
            END;
        END IF;
    END LOOP;

    v_select_cols := rtrim(v_select_cols, ', ');
    v_insert_cols := rtrim(v_insert_cols, ', ');

    IF length(v_insert_cols) = 0 THEN
        RAISE EXCEPTION 'no mappable columns resolved for form_entity %; check normalized_storage_field', p_form_entity_id;
    END IF;

    -- 5. WHERE 条件
    --    源表有 form_entity_id 列时按其过滤; 分表理论上只含该表单数据, 但带上更安全。
    IF v_src_has_feid THEN
        v_where := format('src.form_entity_id = %L', p_form_entity_id);
    ELSE
        v_where := 'TRUE';
    END IF;
    IF NOT p_include_deleted THEN
        v_where := v_where || ' AND src.deleted = 0';
    END IF;

    EXECUTE format('SELECT count(*) FROM %I src WHERE %s', v_source_table, v_where) INTO v_src_count;

    -- 6. 组装 INSERT
    v_sql := format(
        'INSERT INTO %I (%s) SELECT %s FROM %I src WHERE %s ON CONFLICT (id) DO NOTHING',
        v_target_table, v_insert_cols, v_select_cols, v_source_table, v_where
    );

    IF p_dry_run THEN
        RETURN format(
            E'[DRY-RUN] form_entity_id=%s\n source_table=%s\n target_table=%s\n source_rows=%s\n SQL:\n%s',
            p_form_entity_id, v_source_table, v_target_table, v_src_count, v_sql
        );
    END IF;

    -- 7. 执行迁移
    EXECUTE v_sql;
    GET DIAGNOSTICS v_migrated = ROW_COUNT;

    -- 8. 推进自增序列到 max(id)
    IF v_migrated > 0 THEN
        v_seq := pg_get_serial_sequence(v_target_table, 'id');
        IF v_seq IS NOT NULL THEN
            EXECUTE format('SELECT setval(%L, (SELECT COALESCE(MAX(id),1) FROM %I))', v_seq, v_target_table);
        END IF;
    END IF;

    RETURN format(
        E'[MIGRATED] form_entity_id=%s\n source_table=%s\n target_table=%s\n source_rows=%s\n inserted_rows=%s (ON CONFLICT skipped=%s)\n sequence_advanced=%s',
        p_form_entity_id, v_source_table, v_target_table, v_src_count, v_migrated, (v_src_count - v_migrated),
        CASE WHEN v_migrated > 0 THEN 'yes' ELSE 'no' END
    );
END;
$func$;

-- 可选: 迁移后删除函数
-- DROP FUNCTION migrate_form_entity_wide_to_normalized(BIGINT, BOOLEAN, BOOLEAN, TEXT);
migrate_longtext_to_normalized_pg.sql — 长文本迁移函数（子表回填内联列）
-- =============================================================================
-- 长文本(TEXTAREA / RICH_TEXT / LONG_TEXT)迁移: form_entity_data_text -> 窄表内联列 (PostgreSQL)
--
-- 背景:
--   * 宽表模式(A/B): 长文本值不在 valueN 列, 而在子表 form_entity_data_text
--     (long_text 列, 按 form_entity_data_id + form_entity_field_id 关联, locale IS NULL 为主值)。
--   * 窄表模式(C): 长文本值【内联】在窄表的 normalized_storage_field 列 (TEXT)。
--   因此主迁移脚本(migrate_wide_to_normalized_pg.sql)只搬 valueN, 长文本必须由本函数额外迁移。
--
-- 注意:
--   * i18n 译文(form_entity_data_text 里 locale 非空的行)两种模式都存子表, 无需迁移, 本函数不动它们。
--   * 依赖主迁移已完成(窄表里已有对应 id 的行), 本函数用 UPDATE 回填长文本列。
--   * 幂等: 纯 UPDATE, 可重跑。
--
-- 用法:
--   预演:  SELECT migrate_form_entity_longtext_to_normalized(<feid>, true);
--   正式:  BEGIN; SELECT migrate_form_entity_longtext_to_normalized(<feid>, false); COMMIT;
--
-- 参数:
--   p_form_entity_id 表单 id
--   p_dry_run        true=只打印将执行的 UPDATE 与预计行数; false=真正执行
-- =============================================================================

CREATE OR REPLACE FUNCTION migrate_form_entity_longtext_to_normalized(
    p_form_entity_id BIGINT,
    p_dry_run        BOOLEAN DEFAULT true
) RETURNS TEXT
LANGUAGE plpgsql
AS $func$
DECLARE
    v_target_table TEXT;
    v_normalized   INT;
    v_field_cnt    INT := 0;
    v_updated_tot  BIGINT := 0;
    v_updated      BIGINT;
    v_report       TEXT := '';
    v_sql          TEXT;
    v_has_dst      BOOLEAN;
    r              RECORD;
BEGIN
    SELECT standalone_table_name, standalone_table_normalized
      INTO v_target_table, v_normalized
      FROM form_entity WHERE id = p_form_entity_id;

    IF v_target_table IS NULL THEN
        RAISE EXCEPTION 'form_entity % has no standalone_table_name', p_form_entity_id;
    END IF;
    IF v_normalized IS DISTINCT FROM 1 THEN
        RAISE EXCEPTION 'form_entity % is not standalone_table_normalized=1', p_form_entity_id;
    END IF;
    v_target_table := lower(v_target_table);

    -- 遍历长文本类字段
    FOR r IN
        SELECT id AS field_id, token, type, normalized_storage_field
          FROM form_entity_field
         WHERE form_entity_id = p_form_entity_id
           AND type IN ('LONG_TEXT','TEXTAREA','RICH_TEXT')
           AND normalized_storage_field IS NOT NULL
    LOOP
        -- 目标列必须存在
        SELECT EXISTS(SELECT 1 FROM information_schema.columns
                       WHERE table_schema=current_schema()
                         AND table_name=v_target_table
                         AND column_name=lower(r.normalized_storage_field)) INTO v_has_dst;
        IF NOT v_has_dst THEN
            v_report := v_report || format(E'  skip token=%s (target col %s missing)\n', r.token, r.normalized_storage_field);
            CONTINUE;
        END IF;

        v_field_cnt := v_field_cnt + 1;

        -- 用子表主值(locale IS NULL)回填窄表内联列; 按 id 关联(窄表 id == form_entity_data_id)
        v_sql := format(
            'UPDATE %I t SET %I = x.long_text '
            || 'FROM (SELECT form_entity_data_id, long_text FROM form_entity_data_text '
            || '      WHERE form_entity_field_id = %L AND locale IS NULL) x '
            || 'WHERE t.id = x.form_entity_data_id',
            v_target_table, lower(r.normalized_storage_field), r.field_id
        );

        IF p_dry_run THEN
            EXECUTE format(
                'SELECT count(*) FROM form_entity_data_text x JOIN %I t ON t.id = x.form_entity_data_id '
                || 'WHERE x.form_entity_field_id = %L AND x.locale IS NULL',
                v_target_table, r.field_id) INTO v_updated;
            v_report := v_report || format(E'  token=%s field_id=%s -> col=%s : %s rows\n    %s\n',
                r.token, r.field_id, r.normalized_storage_field, v_updated, v_sql);
        ELSE
            EXECUTE v_sql;
            GET DIAGNOSTICS v_updated = ROW_COUNT;
            v_updated_tot := v_updated_tot + v_updated;
            v_report := v_report || format(E'  token=%s -> col=%s : updated %s rows\n',
                r.token, r.normalized_storage_field, v_updated);
        END IF;
    END LOOP;

    IF v_field_cnt = 0 THEN
        RETURN format('[LONGTEXT] form_entity_id=%s: no long-text fields to migrate', p_form_entity_id);
    END IF;

    IF p_dry_run THEN
        RETURN format(E'[LONGTEXT DRY-RUN] form_entity_id=%s target=%s fields=%s\n%s',
            p_form_entity_id, v_target_table, v_field_cnt, v_report);
    ELSE
        RETURN format(E'[LONGTEXT MIGRATED] form_entity_id=%s target=%s fields=%s total_updated=%s\n%s',
            p_form_entity_id, v_target_table, v_field_cnt, v_updated_tot, v_report);
    END IF;
END;
$func$;

-- DROP FUNCTION migrate_form_entity_longtext_to_normalized(BIGINT, BOOLEAN);
verify_migration_pg.sql — 校验函数（行数 / 主键集合 / 抽样值 / 长文本比对）
-- =============================================================================
-- 宽表/分表 -> 窄表 迁移校验函数 (PostgreSQL only)
--
-- 用法:
--   -- A. 全局宽表源 (form_entity_data)
--   SELECT * FROM verify_form_entity_migration(<feid>, 20, NULL);
--   -- B. 分表宽表源 (旧分表名)
--   SELECT * FROM verify_form_entity_migration(<feid>, 20, 'old_split_table_xxx');
--
-- 参数:
--   p_form_entity_id 表单 id
--   p_sample_rows    抽样比对的行数 (默认 20)
--   p_source_table   源表名; NULL=全局宽表 form_entity_data; 否则=旧分表宽表名
--
-- 校验项 (每行返回一个检查结果):
--   1. row_count          目标窄表行数 vs 源表行数 (含软删)
--   2. id_only_in_source  只在源表存在的 id 数 (应为 0)
--   3. id_only_in_target  只在目标窄表存在的 id 数 (应为 0)
--   4. value_mismatch     抽样比对字段值不一致的单元格数 (应为 0)
--
-- 任一 status != 'OK' 时不要切换开关, 回滚后排查。
-- =============================================================================

CREATE OR REPLACE FUNCTION verify_form_entity_migration(
    p_form_entity_id BIGINT,
    p_sample_rows    INT  DEFAULT 20,
    p_source_table   TEXT DEFAULT NULL
) RETURNS TABLE(check_name TEXT, status TEXT, detail TEXT)
LANGUAGE plpgsql
AS $func$
DECLARE
    v_target_table TEXT;
    v_source_table TEXT;
    v_src_cnt      BIGINT;
    v_tgt_cnt      BIGINT;
    v_only_src     BIGINT;
    v_only_tgt     BIGINT;
    v_mismatch     BIGINT := 0;
    v_src_has_feid BOOLEAN;
    v_src_filter   TEXT;
    r              RECORD;
    v_id           BIGINT;
    v_src_val      TEXT;
    v_dst_val      TEXT;
    v_src_label    TEXT;
    v_dst_label    TEXT;
    v_has_src      BOOLEAN;
    v_has_dst      BOOLEAN;
BEGIN
    SELECT standalone_table_name INTO v_target_table
      FROM form_entity WHERE id = p_form_entity_id;
    IF v_target_table IS NULL THEN
        RETURN QUERY SELECT 'config'::TEXT, 'FAIL'::TEXT, 'no standalone_table_name'::TEXT;
        RETURN;
    END IF;
    v_target_table := lower(v_target_table);
    v_source_table := lower(COALESCE(NULLIF(trim(p_source_table), ''), 'form_entity_data'));

    -- 源表 form_entity_id 过滤 (分表若无该列则不过滤)
    SELECT EXISTS(SELECT 1 FROM information_schema.columns
                   WHERE table_schema=current_schema() AND table_name=v_source_table AND column_name='form_entity_id')
      INTO v_src_has_feid;
    IF v_src_has_feid THEN
        v_src_filter := format('form_entity_id = %L', p_form_entity_id);
    ELSE
        v_src_filter := 'TRUE';
    END IF;

    -- 1. 行数
    EXECUTE format('SELECT count(*) FROM %I WHERE %s', v_source_table, v_src_filter) INTO v_src_cnt;
    EXECUTE format('SELECT count(*) FROM %I WHERE form_entity_id = %L', v_target_table, p_form_entity_id) INTO v_tgt_cnt;
    RETURN QUERY SELECT 'row_count'::TEXT,
        CASE WHEN v_src_cnt = v_tgt_cnt THEN 'OK' ELSE 'FAIL' END,
        format('source=%s target=%s', v_src_cnt, v_tgt_cnt);

    -- 2/3. id 集合差异
    EXECUTE format(
        'SELECT count(*) FROM %I s WHERE %s AND NOT EXISTS (SELECT 1 FROM %I n WHERE n.id=s.id)',
        v_source_table, v_src_filter, v_target_table) INTO v_only_src;
    RETURN QUERY SELECT 'id_only_in_source'::TEXT,
        CASE WHEN v_only_src = 0 THEN 'OK' ELSE 'FAIL' END,
        format('%s ids missing in target', v_only_src);

    EXECUTE format(
        'SELECT count(*) FROM %I n WHERE n.form_entity_id=%L AND NOT EXISTS (SELECT 1 FROM %I s WHERE s.id=n.id)',
        v_target_table, p_form_entity_id, v_source_table) INTO v_only_tgt;
    RETURN QUERY SELECT 'id_only_in_target'::TEXT,
        CASE WHEN v_only_tgt = 0 THEN 'OK' ELSE 'FAIL' END,
        format('%s ids extra in target', v_only_tgt);

    -- 4. 抽样值比对
    FOR v_id IN
        EXECUTE format('SELECT id FROM %I WHERE %s ORDER BY random() LIMIT %s',
                       v_source_table, v_src_filter, p_sample_rows)
    LOOP
        FOR r IN
            SELECT token, type, storage_field, normalized_storage_field
              FROM form_entity_field
             WHERE form_entity_id = p_form_entity_id
               AND type <> 'LOOKUP'
               AND storage_field IS NOT NULL
               AND normalized_storage_field IS NOT NULL
        LOOP
            SELECT EXISTS(SELECT 1 FROM information_schema.columns
                           WHERE table_schema=current_schema() AND table_name=v_source_table AND column_name=lower(r.storage_field)) INTO v_has_src;
            SELECT EXISTS(SELECT 1 FROM information_schema.columns
                           WHERE table_schema=current_schema() AND table_name=v_target_table AND column_name=lower(r.normalized_storage_field)) INTO v_has_dst;
            IF v_has_src AND v_has_dst THEN
                EXECUTE format('SELECT %I::text FROM %I WHERE id=%L', lower(r.storage_field), v_source_table, v_id) INTO v_src_val;
                EXECUTE format('SELECT %I::text FROM %I WHERE id=%L', lower(r.normalized_storage_field), v_target_table, v_id) INTO v_dst_val;
                IF v_src_val IS DISTINCT FROM v_dst_val THEN
                    v_mismatch := v_mismatch + 1;
                    RAISE NOTICE 'mismatch id=% token=% source(%)=% target(%)=%',
                        v_id, r.token, r.storage_field, v_src_val, r.normalized_storage_field, v_dst_val;
                END IF;
            END IF;

            IF r.type IN ('SELECT','ACL','CASCADER','TREE_SELECT','DATE') THEN
                DECLARE
                    c_src_label TEXT := regexp_replace(lower(r.storage_field), 'value', 'label');
                    c_dst_label TEXT := lower(r.normalized_storage_field) || '_label';
                BEGIN
                    SELECT EXISTS(SELECT 1 FROM information_schema.columns
                                   WHERE table_schema=current_schema() AND table_name=v_source_table AND column_name=c_src_label) INTO v_has_src;
                    SELECT EXISTS(SELECT 1 FROM information_schema.columns
                                   WHERE table_schema=current_schema() AND table_name=v_target_table AND column_name=c_dst_label) INTO v_has_dst;
                    IF v_has_src AND v_has_dst THEN
                        EXECUTE format('SELECT %I::text FROM %I WHERE id=%L', c_src_label, v_source_table, v_id) INTO v_src_label;
                        EXECUTE format('SELECT %I::text FROM %I WHERE id=%L', c_dst_label, v_target_table, v_id) INTO v_dst_label;
                        IF v_src_label IS DISTINCT FROM v_dst_label THEN
                            v_mismatch := v_mismatch + 1;
                            RAISE NOTICE 'label mismatch id=% token=% source(%)=% target(%)=%',
                                v_id, r.token, c_src_label, v_src_label, c_dst_label, v_dst_label;
                        END IF;
                    END IF;
                END;
            END IF;
        END LOOP;
    END LOOP;

    RETURN QUERY SELECT 'value_mismatch'::TEXT,
        CASE WHEN v_mismatch = 0 THEN 'OK' ELSE 'FAIL' END,
        format('%s mismatched cells over sampled %s rows (see NOTICE for detail)', v_mismatch, p_sample_rows);

    -- 5. 长文本校验: 逐个长文本字段, 比对 子表主值(locale IS NULL) 与 窄表内联列。
    --    统计"子表有值但窄表内联列为空/不等"的行数 (应为 0, 否则说明 §5.5 长文本迁移漏跑或不一致)。
    DECLARE
        v_lt_mismatch BIGINT := 0;
        v_lt_fields   INT := 0;
        v_cnt         BIGINT;
        v_has_dst     BOOLEAN;
        rr            RECORD;
    BEGIN
        FOR rr IN
            SELECT id AS field_id, token, normalized_storage_field
              FROM form_entity_field
             WHERE form_entity_id = p_form_entity_id
               AND type IN ('LONG_TEXT','TEXTAREA','RICH_TEXT')
               AND normalized_storage_field IS NOT NULL
        LOOP
            SELECT EXISTS(SELECT 1 FROM information_schema.columns
                           WHERE table_schema=current_schema() AND table_name=v_target_table
                             AND column_name=lower(rr.normalized_storage_field)) INTO v_has_dst;
            IF NOT v_has_dst THEN CONTINUE; END IF;
            v_lt_fields := v_lt_fields + 1;
            EXECUTE format(
                'SELECT count(*) FROM form_entity_data_text x JOIN %I t ON t.id = x.form_entity_data_id '
                || 'WHERE x.form_entity_field_id = %L AND x.locale IS NULL '
                || 'AND x.long_text IS DISTINCT FROM t.%I',
                v_target_table, rr.field_id, lower(rr.normalized_storage_field)) INTO v_cnt;
            IF v_cnt > 0 THEN
                v_lt_mismatch := v_lt_mismatch + v_cnt;
                RAISE NOTICE 'long-text mismatch token=% field_id=% col=% : % rows differ',
                    rr.token, rr.field_id, rr.normalized_storage_field, v_cnt;
            END IF;
        END LOOP;

        IF v_lt_fields = 0 THEN
            RETURN QUERY SELECT 'longtext_mismatch'::TEXT, 'OK'::TEXT, 'no long-text fields'::TEXT;
        ELSE
            RETURN QUERY SELECT 'longtext_mismatch'::TEXT,
                CASE WHEN v_lt_mismatch = 0 THEN 'OK' ELSE 'FAIL' END,
                format('%s long-text rows differ over %s fields (0 expected; run §5.5 if FAIL)', v_lt_mismatch, v_lt_fields);
        END IF;
    END;
END;
$func$;

-- DROP FUNCTION verify_form_entity_migration(BIGINT, INT, TEXT);