risingwave_meta_model_migration/
m20260705_000000_subscription_dependent_object_fk.rs

1use sea_orm_migration::prelude::*;
2
3use crate::m20230908_072257_init::Object;
4use crate::sea_orm::{ConnectionTrait, DatabaseBackend, Statement, TransactionTrait};
5use crate::utils::ColumnDefExt;
6
7#[derive(DeriveMigrationName)]
8pub struct Migration;
9
10const FK_NAME: &str = "FK_subscription_dependent_table_id";
11const NEW_TABLE: &str = "subscription_new";
12
13#[async_trait::async_trait]
14impl MigrationTrait for Migration {
15    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
16        // `dependent_table_id` stores the object oid of the upstream table/MV. Add the missing
17        // backend FK so deleting that object cascades stale subscription rows in the meta store.
18        let backend = manager.get_database_backend();
19        let cleanup_sql = match backend {
20            DatabaseBackend::MySql => {
21                "DELETE o FROM object AS o \
22                 JOIN subscription AS s ON s.subscription_id = o.oid \
23                 LEFT JOIN object AS dependent ON dependent.oid = s.dependent_table_id \
24                 WHERE dependent.oid IS NULL"
25            }
26            DatabaseBackend::Postgres => {
27                "DELETE FROM object WHERE oid IN (\
28                 SELECT subscription_id FROM subscription \
29                 WHERE dependent_table_id NOT IN (SELECT oid FROM object))"
30            }
31            DatabaseBackend::Sqlite => return recreate_table(manager, true).await,
32        };
33        manager
34            .get_connection()
35            .execute(Statement::from_string(backend, cleanup_sql))
36            .await?;
37        manager
38            .alter_table(
39                Table::alter()
40                    .table(Subscription::Table)
41                    .add_foreign_key(&dependent_object_foreign_key())
42                    .to_owned(),
43            )
44            .await?;
45        Ok(())
46    }
47
48    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
49        match manager.get_database_backend() {
50            DatabaseBackend::MySql | DatabaseBackend::Postgres => {
51                manager
52                    .alter_table(
53                        Table::alter()
54                            .table(Subscription::Table)
55                            .drop_foreign_key(Alias::new(FK_NAME))
56                            .to_owned(),
57                    )
58                    .await?;
59            }
60            DatabaseBackend::Sqlite => {
61                recreate_table(manager, false).await?;
62            }
63        }
64        Ok(())
65    }
66}
67
68fn dependent_object_foreign_key() -> TableForeignKey {
69    TableForeignKey::new()
70        .name(FK_NAME)
71        .from_tbl(Subscription::Table)
72        .from_col(Subscription::DependentTableId)
73        .to_tbl(Object::Table)
74        .to_col(Object::Oid)
75        .on_delete(ForeignKeyAction::Cascade)
76        .to_owned()
77}
78
79async fn recreate_table(manager: &SchemaManager<'_>, with_dependent_fk: bool) -> Result<(), DbErr> {
80    let backend = manager.get_database_backend();
81    let txn = manager.get_connection().begin().await?;
82    {
83        let txn_manager = SchemaManager::new(&txn);
84
85        if with_dependent_fk {
86            txn.execute(Statement::from_string(
87                backend,
88                "DELETE FROM object WHERE oid IN (\
89                 SELECT subscription_id FROM subscription \
90                 WHERE dependent_table_id NOT IN (SELECT oid FROM object))",
91            ))
92            .await?;
93        }
94
95        let mut create = Table::create();
96        create
97            .table(Alias::new(NEW_TABLE))
98            .col(
99                ColumnDef::new(Subscription::SubscriptionId)
100                    .integer()
101                    .primary_key(),
102            )
103            .col(ColumnDef::new(Subscription::Name).string().not_null())
104            .col(
105                ColumnDef::new(Subscription::Definition)
106                    .rw_long_text(&txn_manager)
107                    .not_null(),
108            )
109            .col(ColumnDef::new(Subscription::RetentionSeconds).big_integer())
110            .col(ColumnDef::new(Subscription::SubscriptionState).integer())
111            .col(
112                ColumnDef::new(Subscription::DependentTableId)
113                    .integer()
114                    .not_null(),
115            )
116            .foreign_key(
117                &mut ForeignKey::create()
118                    .name("FK_subscription_object_id")
119                    .from(Alias::new(NEW_TABLE), Subscription::SubscriptionId)
120                    .to(Object::Table, Object::Oid)
121                    .on_delete(ForeignKeyAction::Cascade)
122                    .to_owned(),
123            );
124        if with_dependent_fk {
125            create.foreign_key(
126                &mut ForeignKey::create()
127                    .name(FK_NAME)
128                    .from(Alias::new(NEW_TABLE), Subscription::DependentTableId)
129                    .to(Object::Table, Object::Oid)
130                    .on_delete(ForeignKeyAction::Cascade)
131                    .to_owned(),
132            );
133        }
134        txn_manager.create_table(create).await?;
135
136        txn.execute(Statement::from_string(
137            backend,
138            "INSERT INTO subscription_new \
139             (subscription_id, name, retention_seconds, definition, subscription_state, dependent_table_id) \
140             SELECT subscription_id, name, retention_seconds, definition, subscription_state, dependent_table_id \
141             FROM subscription WHERE dependent_table_id IN (SELECT oid FROM object)",
142        ))
143        .await?;
144
145        txn_manager
146            .drop_table(Table::drop().table(Subscription::Table).to_owned())
147            .await?;
148        txn_manager
149            .rename_table(
150                Table::rename()
151                    .table(Alias::new(NEW_TABLE), Subscription::Table)
152                    .to_owned(),
153            )
154            .await?;
155    }
156    txn.commit().await?;
157    Ok(())
158}
159
160#[derive(DeriveIden)]
161enum Subscription {
162    Table,
163    SubscriptionId,
164    Name,
165    Definition,
166    RetentionSeconds,
167    SubscriptionState,
168    DependentTableId,
169}
risingwave_meta_model_migration/m20260705_000000_subscription_dependent_object_fk.rs

risingwave_meta_model_migration/
m20260705_000000_subscription_dependent_object_fk.rs
