RisingWave 元数据模型演进实战：基于 SeaORM 的迁移文件与模型文件生成指南

• 数据库
• 流处理
• 后端
• 数据工程

导读

RisingWave 的 Meta 服务使用关系型数据库（PostgreSQL / MySQL / SQLite）持久化系统目录（Catalog）与集群元数据，这些持久化层的 Rust 实体定义集中在src/meta/model下，并由src/meta/model/migration这个独立的 SeaORM 迁移工程负责版本化演进。本文以 src/meta/model/src/README.md 为核心骨架，完整讲解"定义版本间变更 → 生成迁移文件 → 应用到数据库 → 反向生成模型文件"的标准化工作流，并深入剖析仓库中真实迁移脚本、派生宏与后端差异处理，帮助你掌握在 RisingWave 元数据模型上安全增删表、变更列、维护枚举与数组类型的全套实战方法。

一、认识 RisingWave 的 Meta 数据模型与迁移工程

在 RisingWave 中，Meta 节点负责维护集群的 Catalog（库、表、物化视图、源、Sink、索引、函数、用户与权限等）以及 Hummock 存储层、流作业调度等运行态元数据。这些数据需要落库持久化，其 ORM 层采用 SeaORM 框架，实体（Entity）与迁移（Migration）分别组织在两个 crate 中：

从源码结构看，model/src目前包含 47 个实体模块（以 lib.rs 中的pub mod声明为准），覆盖database、schema、table、source、sink、index、function、user、worker、streaming_job、fragment、hummock_version_delta、object_dependency等核心领域对象；而migration/src中从m20230908_072257_init到m20260805_000000_object_belong_to_oid已累计注册 74 个迁移文件（见 lib.rs）。

src/meta/model/src/README.md给出的核心思路是：用迁移脚本驱动数据库 Schema 演进，再从数据库反向生成模型文件，避免手工维护实体代码，同时把 PostgreSQL 特有的数组、枚举等类型单独手工定义在模型文件中。

二、版本变更的标准工作流（三步骤总览）

原文档将一次元数据模型变更归纳为如下流程：

1. 生成新迁移文件并应用到数据库：使用 Migrator CLI 的generate与up子命令，得到一个可执行的迁移脚本并真正改变数据库结构；
2. 从数据库反向生成模型文件：使用sea-orm-cli generate entity读取最新表结构，自动生成实体代码，拷贝到模型 crate 中；
3. 手工补充 PG 专属类型与辅助函数：数组、枚举等类型需要手工定义在模型文件中，必要时再补充辅助函数。

下面逐节展开每个步骤的命令、代码写法与仓库内的真实范例。

三、第一步：生成新迁移文件

在 src/meta/model/src/README.md 中，生成迁移文件以本地 PostgreSQL 为例：

export DATABASE_URL=postgres://postgres:@localhost:5432/postgres cargo run -- generate MIGRATION_NAME cargo run -- up

需要说明的是：

• MIGRATION_NAME是一个描述性标识符，Migrator 会根据它生成一个形如m20240101_000000_migration_name.rs的独立文件；
• 必须在src/meta/model/migration目录下运行，而不是项目根目录（migration/README.md 对此有明确提示）；
• DATABASE_URL是必填的连接串，但generate子命令"需要数据库端点但实际上并不会使用它"——也就是说生成阶段即使数据库不可用也可以完成；要真正应用迁移时才需要可用的数据库。

main.rs 展示了 CLI 的入口实现：sea_orm_migration::prelude::cli::run_cli(risingwave_meta_model_migration::Migrator)。这里传入的Migrator类型定义在 lib.rs 中，它的migrations()方法按时间顺序返回一个Vec<Box<dyn MigrationTrait>>，即全部已注册迁移的调度表——新增迁移文件后，记得在这里追加一行Box::new(module::Migration)，否则 Migrator 不会执行它。

迁移文件命名与编号

从仓库中已有的迁移文件可以看出命名规范为m<YYYYMMDD>_<HHMMSS>_<snake_case_name>.rs，例如：

• m20230908_072257_init.rs—— 初始 Schema
• m20240617_070131_index_column_properties.rs—— 给index表加列（README 中推荐的参考样例）
• m20250106_072104_fragment_relation.rs—— 新增 fragment 关联表
• m20250810_000000_add_user_admin_field.rs—— 给用户表加管理员字段
• m20260805_000000_object_belong_to_oid.rs—— 最近期的幂等迁移

文件名中的时间戳决定了迁移的执行顺序，SeaORM 会为每个已执行的迁移在数据库中登记版本记录，保证每个迁移文件只被应用一次。

四、第二步：编写迁移脚本（up / down）

generate生成的文件会包含一份模板迁移脚本，需要你替换为自己的实现。原文档给出了MigrationTrait的标准骨架：

#[async_trait::async_trait] impl MigrationTrait for Migration { async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { // Replace the sample below with your own migration scripts todo!(); } async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { // Replace the sample below with your own migration scripts todo!(); } }

up定义升级操作，down定义对应的回滚操作。原文档指出，你可以在迁移文件中定义表、索引、外键，也可以执行任意 DML 操作来保证数据正确性（例如数据回填、格式转换）。

4.1 一个最小可用的增列迁移

原文档推荐的参考样例 m20240617_070131_index_column_properties.rs 非常典型：

#[async_trait::async_trait] impl MigrationTrait for Migration { async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { manager .alter_table( Table::alter() .table(Index::Table) .add_column(ColumnDef::new(Index::IndexColumnProperties).rw_binary(manager)) .to_owned(), ) .await } async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { manager .alter_table( Table::alter() .table(Index::Table) .drop_column(Index::IndexColumnProperties) .to_owned(), ) .await } } #[derive(DeriveIden)] enum Index { Table, IndexColumnProperties, }

这里有两个值得学习的点：

1. 每个被引用的表名、列名都要用#[derive(DeriveIden)]枚举声明，SeaORM 会将其转换为合法标识符，避免手写字符串带来的大小写与保留字问题；
2. 大字段列使用rw_binary(manager)而不是binary()，这是仓库自带的跨后端扩展（见下文第六节）。

4.2 建表、索引与外键：以 init 迁移为例

初始化迁移 m20230908_072257_init.rs 是仓库中规模最大的迁移，它完整展示了 SeaORM 建表 API 的用法，也定义了 RisingWave 元数据模型的基本盘：

• Cluster 表：以 UUID 作为集群 ID；
• Worker / WorkerProperty 表：Worker 表记录节点host、port、status、worker_type；WorkerProperty 通过外键FK_worker_property_worker_id关联 Worker，并带ON DELETE CASCADE；
• User / Object / ObjectDependency / UserPrivilege 表：构成完整的用户-对象-权限模型，Object表自引用（schema_id、database_id都指向Object.oid）；
• Database / Schema / StreamingJob / Fragment / Actor / ActorDispatcher 表：覆盖 Catalog 与流图执行元数据，Fragment.StreamNode、Actor.Splits、Actor.ExprContext等大字段均用rw_binary存储 protobuf 序列化数据；
• Table / Source / Sink / Index / View / Function 表：完整刻画表、源、Sink、索引、视图、函数对象，其中Definition（SQL 定义文本）用rw_long_text；
• SystemParameter / CatalogVersion 表：系统参数与目录版本号。

索引与外键的创建方式：

manager .create_index( MigrationIndex::create() .table(Worker::Table) .name("idx_worker_host_port") .unique() .col(Worker::Host) .col(Worker::Port) .to_owned(), ) .await?; manager .create_table( MigrationTable::create() .table(WorkerProperty::Table) .col(ColumnDef::new(WorkerProperty::WorkerId).integer().primary_key()) // ... .foreign_key( &mut ForeignKey::create() .name("FK_worker_property_worker_id") .from(WorkerProperty::Table, WorkerProperty::WorkerId) .to(Worker::Table, Worker::WorkerId) .on_delete(ForeignKeyAction::Cascade) .to_owned(), ) .to_owned(), ) .await?;

init 迁移还演示了初始化数据的写法：通过Query::insert()创建集群 ID（随机 UUID）、内置用户root/postgres/rwadmin、内置数据库dev以及public/pg_catalog/information_schema/rw_catalog四个内置 Schema，并在 MySQL 与 PostgreSQL 下用ALTER TABLE ... AUTO_INCREMENT/SELECT setval(...)重置自增序列起点，保证后续对象的 OID 从固定值开始。

4.3 不要修改已发布的历史迁移

migration/README.md 的警告框强调了一个重要约束：

每个迁移文件只能被应用一次，并被记录在系统表中。对于新的 Schema 变更，必须生成新的迁移文件。除非你确信对迁移文件的修改还没有包含在任何已发布版本中，否则不要修改已经发布的迁移文件。

这正是"版本演进"的意义：已发布的迁移已进入线上数据库的版本记录，改动它会导致不同环境间 Schema 不一致。遇到需要调整的 Schema，一律通过新的迁移文件来追加变更。

五、第三步：应用迁移并查看状态

migration/README.md 完整列出了 Migrator CLI 支持的子命令，整理如下：

在开发调试阶段，还可以用 SQLite 内存数据库快速验证迁移逻辑，而无需启动真实的 PostgreSQL：

DATABASE_URL=sqlite::memory: cargo run -- generate MIGRATION_NAME

需要特别留意的是 MySQL 后端下的约束：Migrator 只有在up返回后才记录迁移版本，而MySQL 的 DDL 会隐式提交事务。这意味着执行中途崩溃会导致迁移半途执行但版本未记录。仓库中较新的迁移（如 m20260805_000000_object_belong_to_oid.rs）会刻意把每个 DDL 步骤包在has_column/has_index/has_foreign_key等存在性检查中，使新的 Meta 主节点可以安全地重试一个只执行了一部分的迁移——这是编写高可用环境迁移脚本的重要模式。

六、第四步：从数据库生成模型文件

迁移应用完毕、表结构确定之后，就可以用sea-orm-cli反向生成实体模型了。原文档给出的命令如下：

cargo run -- up sea-orm-cli generate entity -u postgres://postgres:@localhost:5432/postgres -s public -o {target_dir} cp {target_dir}/xxx.rs src/meta/src/model/

对这条命令的解读与补充：

• -u指定数据库连接串，-s指定 schema（PostgreSQL 场景下通常为public），-o指定输出目录；
• 生成出的实体代码无需手工编写；拷贝到模型 crate 后，还要完成三件收尾工作：
  1. 在 src/meta/model/src/lib.rs 中pub mod声明该模块；
  2. 如果它是新的实体表，把它加入for_all_meta_model_entities!宏的实体清单；
  3. 按需在 prelude.rs 中重新导出Entity，方便业务代码统一use。

注：原文档中拷贝目标写为src/meta/src/model/，在本仓库的实际布局中，模型文件的目标目录应为src/meta/model/src/（即 src/meta/model/src），请以实际目录为准。

实体清单的一致性保障

lib.rs中的for_all_meta_model_entities!宏用macro_rules维护了一份全量实体清单（lib.rs），而 tests/meta_model_entities.rs 中的测试会：

1. 用syn解析src/下所有.rs文件，提取每个#[sea_orm(table_name = "...")]声明的表名；
2. 与for_all_meta_model_entities!枚举出的模块逐一比对；
3. 断言两者完全一致，否则报出"Missing in for_all_meta_model_entities / Unexpected in for_all_meta_model_entities"。

这意味着新增实体表后必须同步更新宏清单，否则 CI 中的该测试会失败。

七、手工定义 PG 专属类型：数组与枚举

原文档强调：数组（Array）与枚举（Enum）类型基本只有 PostgreSQL 原生支持，因此需要在模型文件中手工定义。仓库中的做法不是直接暴露原生类型，而是借助 SeaORM 的FromJsonQueryResult/DeriveActiveEnum等派生宏包装成可落库、可序列化的 Rust 类型。

7.1 数组字段：包装为 JSON 类型

原文档给出的I32Array示例：

// We define integer array typed fields as json and derive it using the follow one. #[derive(Clone, Debug, PartialEq, FromJsonQueryResult, Eq, Serialize, Deserialize, Default)] pub struct I32Array(pub Vec<i32>);

这样整数数组在数据库中以 JSON 形式存储，SeaORM 的FromJsonQueryResult派生负责 JSON 与 Rust 结构之间的互相转换。该类型在 lib.rs 中通过仓库自带的derive_from_json_struct!宏实例化：

derive_from_json_struct!(TableIdArray, Vec<TableId>); derive_from_json_struct!(EpochArray, Vec<Epoch>); derive_from_json_struct!(I32Array, Vec<i32>); derive_from_json_struct!(Property, BTreeMap<String, String>);

其中Property被大量用于存储表 / Source / Sink 的 with 属性键值对。

7.2 枚举字段：映射为字符串

原文档给出的枚举示例：

// We define enum typed fields as string and derive it using the follow one. #[derive(Clone, Debug, PartialEq, Eq, EnumIter, DeriveActiveEnum)] #[sea_orm(rs_type = "String", db_type = "String(None)")] pub enum WorkerStatus { #[sea_orm(string_value = "STARTING")] Starting, #[sea_orm(string_value = "RUNNING")] Running, }

关键点在于：

• rs_type = "String"表示 Rust 侧类型为字符串；
• db_type = "String(None)"表示数据库中存储为字符串列；
• #[sea_orm(string_value = "...")]显式指定每个枚举变体对应的数据库字符串值。

仓库中实际使用这一模式的枚举包括 lib.rs 中的JobStatus（INITIAL/CREATING/CREATED）、CreateType（BACKGROUND/FOREGROUND）与DispatcherType（HASH/BROADCAST/SIMPLE/NO_SHUFFLE），它们都额外实现了与 protobuf 枚举（PbStreamJobStatus、PbCreateType、PbDispatcherType）的双向转换，保证数据库字符串值与 gRPC 协议枚举一一对应。例如CreateType的转换：

impl From<CreateType> for PbCreateType { fn from(create_type: CreateType) -> Self { match create_type { CreateType::Background => Self::Background, CreateType::Foreground => Self::Foreground, } } }

StreamingParallelism（Adaptive/Fixed(usize)/Custom）则作为一个 JSON 枚举直接存储在streaming_job表的parallelismJSON 列中，展示了"枚举+复杂载荷"的另一种表达方式。

7.3 复杂载荷：protobuf 二进制包装宏

除了数组与枚举，RisingWave 的元数据表中还大量存储 protobuf 序列化后的结构（流节点、列目录、SST 信息等）。为此 lib.rs 提供了三个底层宏：

例如derive_from_blob!展开出的结构以Vec<u8>存列，并提供to_protobuf()/from_protobuf()在 protobuf 消息与存储字节之间互转：

macro_rules! derive_from_blob { ($struct_name:ident, $field_type:ty) => { #[derive(Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, DeriveValueType)] pub struct $struct_name(#[sea_orm] Vec<u8>); impl $struct_name { pub fn to_protobuf(&self) -> $field_type { prost::Message::decode(self.0.as_slice()).unwrap() } fn from_protobuf(val: &$field_type) -> Self { Self(prost::Message::encode_to_vec(val)) } } // ... }; }

理解这几个宏，就能看懂table.rs、fragment.rs等实体文件中那些"长得不像普通列"的字段类型，也方便为新的复杂字段选择正确的包装方式。

八、跨后端差异：MySQL 的长度限制与ColumnDefExt

原文档（及其引用的 migration/README.md）特别强调了一个 MySQL 后端陷阱：

MySQL 的VARCHAR、TEXT、BLOB、BINARY等类型相对 PostgreSQL / SQLite 有更严格的最大长度限制（最大 65,535 字节）。当需要存储更大的数据（如 SQL 定义、UDF body、protobuf 编码的内部数据等）时，避免使用ColumnDef::text或ColumnDef::blob这类内置构造器，改用 ./src/utils.rs 中定义的扩展方法。

对应的实现位于 migration/src/utils.rs，它通过easy_ext给ColumnDef增加了两个方法：

/// Set column type as `longblob` for MySQL, `bytea` for Postgres, and `blob` for Sqlite. pub fn rw_binary(&mut self, manager: &SchemaManager) -> &mut Self { match manager.get_database_backend() { DatabaseBackend::MySql => self.custom(extension::mysql::MySqlType::LongBlob), DatabaseBackend::Postgres | DatabaseBackend::Sqlite => self.blob(), } } /// Set column type as `longtext` for MySQL, and `text` for Postgres and Sqlite. pub fn rw_long_text(&mut self, manager: &SchemaManager) -> &mut Self { match manager.get_database_backend() { DatabaseBackend::MySql => self.custom(Alias::new("longtext")), DatabaseBackend::Postgres | DatabaseBackend::Sqlite => self.text(), } }

规则总结：

在 init 迁移中还能看到另一处 MySQL 特殊处理：MySQL 默认以utf8_general_ci编码字符串（大小写不敏感），而 RisingWave 需要大小写敏感的比较，因此在建表前会执行ALTER DATABASE CHARACTER SET utf8mb4 COLLATE utf8mb4_bin调整数据库排序规则（m20230908_072257_init.rs）。

九、实战：一个完整的幂等迁移 + 数据回填范例

如果要为线上集群新增一列并回填历史数据，仓库中最新、也最值得借鉴的样例是 m20260805_000000_object_belong_to_oid.rs。它给object表添加belong_to_oid列（记录对象归属的父对象），并展示了三个高级模式：

1. 逐步骤幂等（MySQL DDL 隐式提交场景）

if !manager.has_column("object", "belong_to_oid").await? { // ... add column } if !manager.has_index("object", INDEX_NAME).await? { // ... create index } if matches!(backend, DatabaseBackend::MySql | DatabaseBackend::Postgres) && !has_foreign_key(manager).await? { // ... add foreign key }

每个 DDL 步骤前都做存在性检查，保证迁移在部分执行后可以被安全重试。

2. 按后端分写的 SQL 回填

由于 SQLite 无法单独为既有表追加外键，它在ALTER TABLE时直接以内联REFERENCES的方式定义外键；而后端差异更大的回填 UPDATE 语句则分别为 MySQL / PostgreSQL / SQLite 各写一份（如belongs_to_job_id回填、__iceberg_sink_前缀隐式对象的归属推导等），保持三端行为一致。

3. 内嵌单元测试

该文件末尾自带两个#[tokio::test]：test_sqlite_backfill_and_cascade在sqlite::memory:上构造迷你表结构，执行迁移后断言belong_to_oid回填结果与级联删除行为；test_sqlite_partial_run_retry则模拟"Meta 在第一个 DDL 语句后崩溃但未记录版本"与"全部语句执行完但未记录版本"两种场景，验证迁移可重复执行且结果幂等。这为"如何给迁移写测试"提供了直接范本。

十、模型文件中的辅助函数与转换逻辑

原文档最后一条是"如有必要，在模型文件中定义其他辅助函数"。仓库中的典型用法包括：

• 枚举与 protobuf 的双向转换：如前文CreateType、JobStatus、DispatcherType与Pb*类型的From实现（lib.rs、lib.rs），供控制器（controller）层在数据库枚举与 gRPC 协议之间互转；
• 包装类型的便捷方法：如I32Array的into_u32_array()（把Vec<i32>转成Vec<u32>，见 lib.rs），以及derive_from_blob!生成的to_protobuf()/inner_ref()等；
• 预置类型别名：TransactionId = i32、Epoch = i64、CompactionTaskId = i64等语义化别名（lib.rs），让实体字段的意图更清晰。

这些辅助函数通常服务于src/meta下的 controller 与 service 层，是"模型文件 — 业务逻辑"之间的衔接层。

十一、小结：一次完整变更的检查清单

综合原文档与仓库源码，一次 RisingWave 元数据模型变更应完成以下动作：

1. 在 src/meta/model/migration 目录下执行DATABASE_URL=... cargo run -- generate <MIGRATION_NAME>生成迁移文件；
2. 在迁移文件中实现up/down：定义或修改表、索引、外键，必要时执行数据 DML；大字段用rw_binary/rw_long_text；
3. 在 migration/src/lib.rs 的migrations()中注册新迁移；
4. 执行cargo run -- up（开发期可用fresh/refresh/status调试），确认 Schema 与数据正确；
5. 用sea-orm-cli generate entity从数据库生成实体，拷贝到 src/meta/model/src；
6. 在 lib.rs 中声明模块、加入for_all_meta_model_entities!清单，必要时在 prelude.rs 中导出 Entity；
7. 数组、枚举等 PG 专属类型手工定义为FromJsonQueryResult/DeriveActiveEnum包装类型，protobuf 复杂载荷用derive_from_blob!系列宏；
8. 为复杂迁移编写幂等保护与单元测试（参考 m20260805_000000_object_belong_to_oid.rs 的测试写法），并保证 meta_model_entities.rs 的一致性测试通过；
9. 记住红线：已发布的迁移文件不可修改，新变更一律追加新迁移。

按照这套流程，你可以在不破坏线上数据的前提下，安全、可回滚、可测试地推动 RisingWave 元数据模型的持续演进。

• 数据库
• 流处理
• 后端
• 数据工程