bf1fe13 · reshape · git.iko.soy

git.iko.soy/reshape

Add reverse trigger for complex up and down tranformations

commitbf1fe13d5f812296bf884635195555251dc35843 authorfabianlindfors <fabian@flapplabs.se> dateSat 20 Jan 2024, 01:58 (3 years ago) parentde44fdd

3 files changed+389−70

compare with main

unifiedsplit

src/migrations/add_column.rs+115−23 src/migrations/remove_column.rs+110−42 tests/complex.rs+164−5

src/migrations/add_column.rs+115−23

@@ -3,7 +3,7 @@ use crate::{

33 db::{Conn, Transaction},

44 schema::Schema,

55};

6−use anyhow::Context;

6+use anyhow::{bail, Context};

77use serde::{Deserialize, Serialize};

99#[derive(Serialize, Deserialize, Debug)]

@@ -43,6 +43,15 @@ impl AddColumn {

4343 )

4444 }

4545

46+ fn reverse_trigger_name(&self, ctx: &MigrationContext) -> String {

47+ format!(

48+ "{}_add_column_{}_{}_rev",

49+ ctx.prefix(),

50+ self.table,

51+ self.column.name

52+ )

53+ }

54+

4655 fn not_null_constraint_name(&self, ctx: &MigrationContext) -> String {

4756 format!(

4857 "{}_add_column_not_null_{}_{}",

@@ -97,21 +106,21 @@ impl Action for AddColumn {

97106 );

98107 db.run(&query).context("failed to add column")?;

99108

109+ let declarations: Vec<String> = table

110+ .columns

111+ .iter()

112+ .map(|column| {

113+ format!(

114+ "{alias} public.{table}.{real_name}%TYPE := NEW.{real_name};",

115+ table = table.real_name,

116+ alias = column.name,

117+ real_name = column.real_name,

118+ )

119+ })

120+ .collect();

121+

100122 if let Some(up) = &self.up {

101123 if let Transformation::Simple(up) = up {

102− let declarations: Vec<String> = table

103− .columns

104− .iter()

105− .map(|column| {

106− format!(

107− "{alias} public.{table}.{real_name}%TYPE := NEW.{real_name};",

108− table = table.real_name,

109− alias = column.name,

110− real_name = column.real_name,

111− )

112− })

113− .collect();

114−

115124 // Add triggers to fill in values as they are inserted/updated

116125 let query = format!(

117126 r#"

@@ -151,15 +160,20 @@ impl Action for AddColumn {

151160 r#where,

152161 } = up

153162 {

163+ let existing_schema_name = match &ctx.existing_schema_name {

164+ Some(name) => name,

165+ None => bail!("can't use update without previous migration"),

166+ };

167+

154168 let from_table = schema.get_table(db, &from_table)?;

155169

156− let declarations: Vec<String> = from_table

170+ let from_table_assignments: Vec<String> = from_table

157171 .columns

158172 .iter()

159173 .map(|column| {

160174 format!(

161− "{alias} public.{table}.{real_name}%TYPE := NEW.{real_name};",

162− table = from_table.real_name,

175+ "{table}.{alias} = NEW.{real_name};",

176+ table = from_table.name,

163177 alias = column.name,

164178 real_name = column.real_name,

165179 )

@@ -175,11 +189,18 @@ impl Action for AddColumn {

175189 BEGIN

176190 IF NOT reshape.is_new_schema() THEN

177191 DECLARE

178− {declarations}

192+ {from_table} migration_{existing_schema_name}.{from_table}%ROWTYPE;

179193 BEGIN

194+ {assignments}

195+

196+ -- Don't trigger reverse trigger when making this update

197+ perform set_config('reshape.disable_triggers', 'TRUE', TRUE);

198+

180199 UPDATE public."{changed_table_real}"

181200 SET "{temp_column_name}" = {value}

182201 WHERE {where};

202+

203+ perform set_config('reshape.disable_triggers', '', TRUE);

183204 END;

184205 END IF;

185206 RETURN NEW;

@@ -189,14 +210,81 @@ impl Action for AddColumn {

189210 DROP TRIGGER IF EXISTS "{trigger_name}" ON "{from_table_real}";

190211 CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{from_table_real}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();

191212 "#,

213+ assignments = from_table_assignments.join("\n"),

192214 changed_table_real = table.real_name,

215+ from_table = from_table.name,

193216 from_table_real = from_table.real_name,

194217 trigger_name = self.trigger_name(ctx),

195− declarations = declarations.join("\n"),

218+ // declarations = from_table_declarations.join("\n"),

196219 temp_column_name = temp_column_name,

197220 );

198221 db.run(&query).context("failed to create up trigger")?;

199222

223+ let from_table_columns = from_table

224+ .columns

225+ .iter()

226+ .map(|column| format!("{} as {}", column.real_name, column.name))

227+ .collect::<Vec<String>>()

228+ .join(", ");

229+

230+ let changed_table_assignments: Vec<String> = table

231+ .columns

232+ .iter()

233+ .map(|column| {

234+ format!(

235+ "{table}.{alias} := NEW.{real_name};",

236+ table = table.name,

237+ alias = column.name,

238+ real_name = column.real_name,

239+ )

240+ })

241+ .collect();

242+

243+ // Add triggers to fill in values as they are inserted/updated

244+ let query = format!(

245+ r#"

246+ CREATE OR REPLACE FUNCTION {trigger_name}()

247+ RETURNS TRIGGER AS $$

248+ #variable_conflict use_variable

249+ BEGIN

250+ IF NOT reshape.is_new_schema() AND NOT current_setting('reshape.disable_triggers', TRUE) = 'TRUE' THEN

251+ DECLARE

252+ {changed_table} migration_{existing_schema_name}.{changed_table}%ROWTYPE;

253+ __temp_row migration_{existing_schema_name}.{from_table}%ROWTYPE;

254+ BEGIN

255+ {changed_table_assignments}

256+

257+ SELECT {from_table_columns}

258+ INTO __temp_row

259+ FROM migration_{existing_schema_name}.{from_table} {from_table}

260+ WHERE {where};

261+

262+ DECLARE

263+ {from_table} migration_{existing_schema_name}.{from_table}%ROWTYPE;

264+ BEGIN

265+ {from_table} = __temp_row;

266+ NEW.{temp_column_name} = {value};

267+ END;

268+ END;

269+ END IF;

270+ RETURN NEW;

271+ END

272+ $$ language 'plpgsql';

273+

274+ DROP TRIGGER IF EXISTS "{trigger_name}" ON "{changed_table_real}";

275+ CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{changed_table_real}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();

276+ "#,

277+ changed_table_assignments = changed_table_assignments.join("\n"),

278+ changed_table_real = table.real_name,

279+ changed_table = table.name,

280+ from_table = from_table.name,

281+ trigger_name = self.reverse_trigger_name(ctx),

282+ temp_column_name = temp_column_name,

283+ // declarations = declarations.join("\n"),

284+ );

285+ db.run(&query)

286+ .context("failed to create reverse up trigger")?;

287+

200288 // Backfill values in batches by touching the from table

201289 common::batch_touch_rows(db, &from_table.real_name, None)

202290 .context("failed to batch update existing rows")?;

@@ -210,10 +298,10 @@ impl Action for AddColumn {

210298 if !self.column.nullable {

211299 let query = format!(

212300 r#"

213− ALTER TABLE "{table}"

214− ADD CONSTRAINT "{constraint_name}"

215− CHECK ("{column}" IS NOT NULL) NOT VALID

216− "#,

301+ ALTER TABLE "{table}"

302+ ADD CONSTRAINT "{constraint_name}"

303+ CHECK ("{column}" IS NOT NULL) NOT VALID

304+ "#,

217305 table = self.table,

218306 constraint_name = self.not_null_constraint_name(ctx),

219307 column = temp_column_name,

@@ -236,8 +324,10 @@ impl Action for AddColumn {

236324 let query = format!(

237325 r#"

238326 DROP FUNCTION IF EXISTS "{trigger_name}" CASCADE;

327+ DROP FUNCTION IF EXISTS "{reverse_trigger_name}" CASCADE;

239328 "#,

240329 trigger_name = self.trigger_name(ctx),

330+ reverse_trigger_name = self.reverse_trigger_name(ctx),

241331 );

242332 transaction

243333 .run(&query)

@@ -329,8 +419,10 @@ impl Action for AddColumn {

329419 let query = format!(

330420 r#"

331421 DROP FUNCTION IF EXISTS "{trigger_name}" CASCADE;

422+ DROP FUNCTION IF EXISTS "{reverse_trigger_name}" CASCADE;

332423 "#,

333424 trigger_name = self.trigger_name(ctx),

425+ reverse_trigger_name = self.reverse_trigger_name(ctx),

334426 );

335427 db.run(&query).context("failed to drop up trigger")?;

336428

src/migrations/remove_column.rs+110−42

@@ -34,6 +34,15 @@ impl RemoveColumn {

3434 )

3535 }

3636

37+ fn reverse_trigger_name(&self, ctx: &MigrationContext) -> String {

38+ format!(

39+ "{}_remove_column_{}_{}_rev",

40+ ctx.prefix(),

41+ self.table,

42+ self.column

43+ )

44+ }

45+

3746 fn null_constraint_trigger_name(&self, ctx: &MigrationContext) -> String {

3847 format!(

3948 "{}_remove_column_{}_{}_nn",

@@ -121,19 +130,6 @@ impl Action for RemoveColumn {

121130

122131 let from_table = schema.get_table(db, &from_table)?;

123132

124− let declarations: Vec<String> = from_table

125− .columns

126− .iter()

127− .map(|column| {

128− format!(

129− "{alias} public.{table}.{real_name}%TYPE := NEW.{real_name};",

130− table = from_table.real_name,

131− alias = column.name,

132− real_name = column.real_name,

133− )

134− })

135− .collect();

136−

137133 let maybe_null_check = if !column.nullable {

138134 // Replace NOT NULL constraint with a constraint trigger that only triggers on the old schema.

139135 // As we are using a complex down function, we must remove the NOT NULL check for the new schema.

@@ -190,6 +186,19 @@ impl Action for RemoveColumn {

190186 "".to_string()

191187 };

192188

189+ let into_variables = from_table

190+ .columns

191+ .iter()

192+ .map(|column| {

193+ format!(

194+ "NEW.{real_name} AS {alias}",

195+ alias = column.name,

196+ real_name = column.real_name,

197+ )

198+ })

199+ .collect::<Vec<String>>()

200+ .join(", ");

201+

193202 let query = format!(

194203 r#"

195204 CREATE OR REPLACE FUNCTION {trigger_name}()

@@ -198,12 +207,21 @@ impl Action for RemoveColumn {

198207 BEGIN

199208 IF reshape.is_new_schema() THEN

200209 DECLARE

201− {declarations}

210+ {from_table} record;

202211 BEGIN

212+ SELECT {into_variables}

213+ INTO {from_table};

214+

203215 {maybe_null_check}

216+

217+ -- Don't trigger reverse trigger when making this update

218+ perform set_config('reshape.disable_triggers', 'TRUE', TRUE);

219+

204220 UPDATE "migration_{existing_schema_name}"."{changed_table}" "{changed_table}"

205221 SET "{column_name}" = {value}

206222 WHERE {where};

223+

224+ perform set_config('reshape.disable_triggers', '', TRUE);

207225 END;

208226 END IF;

209227 RETURN NEW;

@@ -214,12 +232,81 @@ impl Action for RemoveColumn {

214232 CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{from_table_real}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();

215233 "#,

216234 changed_table = self.table,

235+ from_table = from_table.name,

217236 from_table_real = from_table.real_name,

218237 column_name = self.column,

219238 trigger_name = self.trigger_name(ctx),

220− declarations = declarations.join("\n"),

239+ // declarations = declarations.join("\n"),

221240 );

222241 db.run(&query).context("failed to create down trigger")?;

242+

243+ let changed_into_variables = table

244+ .columns

245+ .iter()

246+ .map(|column| {

247+ format!(

248+ "NEW.{real_name} AS {alias}",

249+ alias = column.name,

250+ real_name = column.real_name,

251+ )

252+ })

253+ .collect::<Vec<String>>()

254+ .join(", ");

255+

256+ let from_table_columns = from_table

257+ .columns

258+ .iter()

259+ .map(|column| format!("{} as {}", column.real_name, column.name))

260+ .collect::<Vec<String>>()

261+ .join(", ");

262+

263+ let query = format!(

264+ r#"

265+ CREATE OR REPLACE FUNCTION {trigger_name}()

266+ RETURNS TRIGGER AS $$

267+ #variable_conflict use_variable

268+ BEGIN

269+ IF reshape.is_new_schema() AND NOT current_setting('reshape.disable_triggers', TRUE) = 'TRUE' THEN

270+ DECLARE

271+ {changed_table} record;

272+ __temp_row record;

273+ BEGIN

274+ SELECT {changed_into_variables}

275+ INTO {changed_table};

276+

277+ SELECT *

278+ INTO __temp_row

279+ FROM (

280+ SELECT {from_table_columns}

281+ FROM public.{from_table_real}

282+ ) {from_table}

283+ WHERE {where};

284+

285+ DECLARE

286+ {from_table} record;

287+ BEGIN

288+ {from_table} := __temp_row;

289+ NEW.{column_name_real} = {value};

290+ END;

291+ END;

292+ END IF;

293+ RETURN NEW;

294+ END

295+ $$ language 'plpgsql';

296+

297+ DROP TRIGGER IF EXISTS "{trigger_name}" ON "{changed_table_real}";

298+ CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{changed_table_real}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();

299+ "#,

300+ changed_table = table.name,

301+ changed_table_real = table.real_name,

302+ from_table = from_table.name,

303+ from_table_real = from_table.real_name,

304+ column_name_real = column.real_name,

305+ trigger_name = self.reverse_trigger_name(ctx),

306+ // declarations = declarations.join("\n"),

307+ );

308+ db.run(&query)

309+ .context("failed to create reverse down trigger")?;

223310 }

224311 }

225312

@@ -247,28 +334,19 @@ impl Action for RemoveColumn {

247334 }

248335

249336 // Remove column, function and trigger

250− let trigger_table = match &self.down {

251− Some(Transformation::Update {

252− table,

253− value: _,

254− r#where: _,

255− }) => table,

256− _ => &self.table,

257− };

258337 let query = format!(

259338 r#"

260339 ALTER TABLE "{table}"

261340 DROP COLUMN IF EXISTS "{column}";

262341

263− DROP TRIGGER IF EXISTS "{trigger_name}" ON "{trigger_table}";

264− DROP FUNCTION IF EXISTS "{trigger_name}";

265−

266− DROP TRIGGER IF EXISTS "{null_trigger_name}" ON "{table}";

267− DROP FUNCTION IF EXISTS "{null_trigger_name}";

342+ DROP FUNCTION IF EXISTS "{trigger_name}" CASCADE;

343+ DROP FUNCTION IF EXISTS "{reverse_trigger_name}" CASCADE;

344+ DROP FUNCTION IF EXISTS "{null_trigger_name}" CASCADE;

268345 "#,

269346 table = self.table,

270347 column = self.column,

271348 trigger_name = self.trigger_name(ctx),

349+ reverse_trigger_name = self.reverse_trigger_name(ctx),

272350 null_trigger_name = self.null_constraint_trigger_name(ctx),

273351 );

274352 db.run(&query)

@@ -299,24 +377,14 @@ impl Action for RemoveColumn {

299377 .context("failed to reinstate column not null constraint")?;

300378

301379 // Remove function and trigger

302− let trigger_table = match &self.down {

303− Some(Transformation::Update {

304− table,

305− value: _,

306− r#where: _,

307− }) => table,

308− _ => &self.table,

309− };

310380 db.run(&format!(

311381 r#"

312− DROP TRIGGER IF EXISTS "{trigger_name}" ON "{trigger_table}";

313− DROP FUNCTION IF EXISTS "{trigger_name}";

314−

315− DROP TRIGGER IF EXISTS "{null_trigger_name}" ON "{table}";

316− DROP FUNCTION IF EXISTS "{null_trigger_name}";

382+ DROP FUNCTION IF EXISTS "{trigger_name}" CASCADE;

383+ DROP FUNCTION IF EXISTS "{reverse_trigger_name}" CASCADE;

384+ DROP FUNCTION IF EXISTS "{null_trigger_name}" CASCADE;

317385 "#,

318− table = self.table,

319386 trigger_name = self.trigger_name(ctx),

387+ reverse_trigger_name = self.reverse_trigger_name(ctx),

320388 null_trigger_name = self.null_constraint_trigger_name(ctx),

321389 ))

322390 .context("failed to drop down trigger")?;

tests/complex.rs+164−5

@@ -1,6 +1,165 @@

11mod common;

22use common::Test;

4+#[test]

5+fn move_column_between_tables() {

6+ let mut test = Test::new("Move column between tables");

8+ test.first_migration(

9+ r#"

10+ name = "create_tables"

11+

12+ [[actions]]

13+ type = "create_table"

14+ name = "users"

15+ primary_key = ["id"]

16+

17+ [[actions.columns]]

18+ name = "id"

19+ type = "INTEGER"

20+

21+ [[actions.columns]]

22+ name = "email"

23+ type = "TEXT"

24+

25+ [[actions]]

26+ type = "create_table"

27+ name = "profiles"

28+ primary_key = ["id"]

29+

30+ [[actions.columns]]

31+ name = "id"

32+ type = "INTEGER"

33+

34+ [[actions.columns]]

35+ name = "user_id"

36+ type = "INTEGER"

37+ nullable = false

38+ "#,

39+ );

40+

41+ test.second_migration(

42+ r#"

43+ name = "move_email_column"

44+

45+ [[actions]]

46+ type = "add_column"

47+ table = "profiles"

48+

49+ [actions.column]

50+ name = "email"

51+ type = "TEXT"

52+ nullable = false

53+

54+ # When `users` is updated in the old schema, we write the email value to `profiles`

55+ # When `profiles` is updated in the old schema, we touch the equivalent `users` row to update it

56+ [actions.up]

57+ table = "users"

58+ value = "users.email"

59+ where = "profiles.user_id = users.id"

60+

61+ [[actions]]

62+ type = "remove_column"

63+ table = "users"

64+ column = "email"

65+

66+ # When `profiles` is changed in the new schema, we write the email address back to the removed column

67+ [actions.down]

68+ table = "profiles"

69+ value = "profiles.email"

70+ where = "users.id = profiles.user_id"

71+ "#,

72+ );

73+

74+ test.after_first(|db| {

75+ db.simple_query("INSERT INTO users (id, email) VALUES (1, 'test1@test.com')")

76+ .unwrap();

77+ db.simple_query("INSERT INTO users (id, email) VALUES (2, 'test2@test.com')")

78+ .unwrap();

79+

80+ db.simple_query("INSERT INTO profiles (id, user_id) VALUES (1, 1)")

81+ .unwrap();

82+ db.simple_query("INSERT INTO profiles (id, user_id) VALUES (2, 2)")

83+ .unwrap();

84+ });

85+

86+ test.intermediate(|old_db, new_db| {

87+ // Ensure emails were backfilled into profiles

88+ let profiles_emails: Vec<String> = new_db

89+ .query(

90+ r#"

91+ SELECT email

92+ FROM profiles

93+ ORDER BY id

94+ "#,

95+ &[],

96+ )

97+ .unwrap()

98+ .iter()

99+ .map(|row| row.get("email"))

100+ .collect();

101+ assert_eq!(vec!("test1@test.com", "test2@test.com"), profiles_emails);

102+

103+ // Ensure insert in old schema updates new

104+ old_db

105+ .simple_query("INSERT INTO users (id, email) VALUES (3, 'test3@test.com')")

106+ .unwrap();

107+ old_db

108+ .simple_query("INSERT INTO profiles (id, user_id) VALUES (3, 3)")

109+ .unwrap();

110+ let new_email: String = new_db

111+ .query("SELECT email FROM profiles WHERE id = 3", &[])

112+ .unwrap()

113+ .first()

114+ .map(|row| row.get("email"))

115+ .unwrap();

116+ assert_eq!("test3@test.com", new_email);

117+

118+ // Ensure updates in old schema updates new

119+ old_db

120+ .simple_query("UPDATE users SET email = 'test3+updated@test.com' WHERE id = 3")

121+ .unwrap();

122+ let new_email: String = new_db

123+ .query("SELECT email FROM profiles WHERE id = 3", &[])

124+ .unwrap()

125+ .first()

126+ .map(|row| row.get("email"))

127+ .unwrap();

128+ assert_eq!("test3+updated@test.com", new_email);

129+

130+ // Ensure insert in new schema updates old

131+ new_db

132+ .simple_query(

133+ "INSERT INTO profiles (id, user_id, email) VALUES (4, 4, 'test4@test.com')",

134+ )

135+ .unwrap();

136+ new_db

137+ .simple_query("INSERT INTO users (id) VALUES (4)")

138+ .unwrap();

139+ let old_email: String = old_db

140+ .query("SELECT email FROM users WHERE id = 4", &[])

141+ .unwrap()

142+ .first()

143+ .map(|row| row.get("email"))

144+ .unwrap();

145+ assert_eq!("test4@test.com", old_email);

146+

147+ // Ensure update in new schema updates old

148+ new_db

149+ .simple_query("UPDATE profiles SET email = 'test4+updated@test.com' WHERE id = 4")

150+ .unwrap();

151+ let old_email: String = old_db

152+ .query("SELECT email FROM users WHERE id = 4", &[])

153+ .unwrap()

154+ .first()

155+ .map(|row| row.get("email"))

156+ .unwrap();

157+ assert_eq!("test4+updated@test.com", old_email);

158+ });

159+

160+ test.run();

161+}

162+

4163#[test]

5164fn extract_relation_into_new_table() {

6165 let mut test = Test::new("Extract relation into new table");

@@ -64,7 +223,7 @@ fn extract_relation_into_new_table() {

64223 [actions.up]

65224 table = "users"

66225 values = { user_id = "id", account_id = "account_id", role = "UPPER(account_role)" }

67− where = "user_id = id"

226+ where = "user_account_connections.user_id = users.id"

68227

69228 [[actions]]

70229 type = "remove_column"

@@ -73,8 +232,8 @@ fn extract_relation_into_new_table() {

73232

74233 [actions.down]

75234 table = "user_account_connections"

76− value = "account_id"

77− where = "id = user_id"

235+ value = "user_account_connections.account_id"

236+ where = "users.id = user_account_connections.user_id"

78237

79238 [[actions]]

80239 type = "remove_column"

@@ -83,8 +242,8 @@ fn extract_relation_into_new_table() {

83242

84243 [actions.down]

85244 table = "user_account_connections"

86− value = "LOWER(role)"

87− where = "id = user_id"

245+ value = "LOWER(user_account_connections.role)"

246+ where = "users.id = user_account_connections.user_id"

88247 "#,

89248 );

90249