4e5cc5c · reshape · git.iko.soy

git.iko.soy/reshape

on main help sign in

Add support for complex up function in

4e5cc5cfabianlindfors3 years ago

README.md+20

compare with main

unifiedsplit

@@ -351,6 +351,26 @@ up = "data['path']['to']['value'] #>> '{}'"

351 type = "TEXT"

352```

353

354+_Example: duplicate `email` column from `users` to `profiles` table_

355+

356+```toml

357+# `profiles` has `user_id` column which maps to `users.id`

358+[[actions]]

359+type = "add_column"

360+table = "profiles"

361+

362+ [actions.column]

363+ name = "email"

364+ type = "TEXT"

365+ nullable = false

366+

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

368+ [actions.up]

369+ table = "users"

370+ value = "email"

371+ where = "user_id = id"

372+```

373+

354#### Alter column

374#### Alter column

355

375

356The `alter_column` action enables many different changes to an existing column, for example renaming, changing type and changing existing values.

376The `alter_column` action enables many different changes to an existing column, for example renaming, changing type and changing existing values.

src/migrations/add_column.rs+116−52

@@ -9,8 +9,19 @@ use serde::{Deserialize, Serialize};

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

10pub struct AddColumn {

11 pub table: String,

12− pub up: Option<String>,

13 pub column: Column,

12 pub column: Column,

13+ pub up: Option<Transformation>,

14+}

15+

16+#[derive(Serialize, Deserialize, Debug)]

17+#[serde(untagged)]

18+pub enum Transformation {

19+ Simple(String),

20+ Update {

21+ table: String,

22+ value: String,

23+ r#where: String,

24+ },

14}

25}

16impl AddColumn {

27impl AddColumn {

@@ -87,52 +98,109 @@ impl Action for AddColumn {

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

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

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

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

101+ if let Transformation::Simple(up) = up {

90− let declarations: Vec<String> = table

102+ let declarations: Vec<String> = table

91− .columns

103+ .columns

92− .iter()

104+ .iter()

93− .map(|column| {

105+ .map(|column| {

94− format!(

106+ format!(

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

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

96− table = table.real_name,

108+ table = table.real_name,

97− alias = column.name,

109+ alias = column.name,

98− real_name = column.real_name,

110+ real_name = column.real_name,

99− )

111+ )

100− })

112+ })

101− .collect();

113+ .collect();

102−

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

104− let query = format!(

105− r#"

106− CREATE OR REPLACE FUNCTION {trigger_name}()

107− RETURNS TRIGGER AS $$

108− BEGIN

109− IF NOT reshape.is_new_schema() THEN

110− DECLARE

111− {declarations}

112− BEGIN

113− NEW.{temp_column_name} = {up};

114− END;

115− END IF;

116− RETURN NEW;

117− END

118− $$ language 'plpgsql';

119−

120− DROP TRIGGER IF EXISTS "{trigger_name}" ON "{table}";

121− CREATE TRIGGER "{trigger_name}" BEFORE UPDATE OR INSERT ON "{table}" FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();

122− "#,

123− temp_column_name = temp_column_name,

124− trigger_name = self.trigger_name(ctx),

125− up = up,

126− table = self.table,

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

128− );

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

130− }

131

114

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

116+ let query = format!(

117+ r#"

118+ CREATE OR REPLACE FUNCTION {trigger_name}()

119+ RETURNS TRIGGER AS $$

120+ BEGIN

121+ IF NOT reshape.is_new_schema() THEN

122+ DECLARE

123+ {declarations}

124+ BEGIN

125+ NEW.{temp_column_name} = {up};

126+ END;

127+ END IF;

128+ RETURN NEW;

129+ END

130+ $$ language 'plpgsql';

131+

132+ DROP TRIGGER IF EXISTS "{trigger_name}" ON "{table}";

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

134+ "#,

135+ temp_column_name = temp_column_name,

136+ trigger_name = self.trigger_name(ctx),

137+ up = up,

138+ table = self.table,

139+ declarations = declarations.join("\n"),

140+ );

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

142+

132− // Backfill values in batches

143+ // Backfill values in batches

133− if self.up.is_some() {

144+ common::batch_touch_rows(db, &table.real_name, Some(&temp_column_name))

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

146+ }

147+

148+ if let Transformation::Update {

149+ table: from_table,

150+ value,

151+ r#where,

152+ } = up

153+ {

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

155+

156+ let declarations: Vec<String> = from_table

157+ .columns

158+ .iter()

159+ .map(|column| {

160+ format!(

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

162+ table = from_table.real_name,

163+ alias = column.name,

164+ real_name = column.real_name,

165+ )

166+ })

167+ .collect();

168+

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

170+ let query = format!(

171+ r#"

172+ CREATE OR REPLACE FUNCTION {trigger_name}()

173+ RETURNS TRIGGER AS $$

174+ #variable_conflict use_variable

175+ BEGIN

176+ IF NOT reshape.is_new_schema() THEN

177+ DECLARE

178+ {declarations}

179+ BEGIN

180+ UPDATE public."{changed_table_real}"

181+ SET "{temp_column_name}" = {value}

182+ WHERE {where};

183+ END;

184+ END IF;

185+ RETURN NEW;

186+ END

187+ $$ language 'plpgsql';

188+

189+ DROP TRIGGER IF EXISTS "{trigger_name}" ON "{from_table_real}";

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

191+ "#,

192+ changed_table_real = table.real_name,

193+ from_table_real = from_table.real_name,

194+ trigger_name = self.trigger_name(ctx),

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

196+ temp_column_name = temp_column_name,

197+ );

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

199+

200+ // Backfill values in batches by touching the from table

134− common::batch_touch_rows(db, &table.real_name, &temp_column_name)

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

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

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

203+ }

136 }

204 }

137

205

138 // Add a temporary NOT NULL constraint if the column shouldn't be nullable.

206 // Add a temporary NOT NULL constraint if the column shouldn't be nullable.

@@ -167,10 +235,8 @@ impl Action for AddColumn {

167 // Remove triggers and procedures

235 // Remove triggers and procedures

168 let query = format!(

236 let query = format!(

169 r#"

237 r#"

170− DROP TRIGGER IF EXISTS "{trigger_name}" ON "{table}";

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

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

172 "#,

239 "#,

173− table = self.table,

174 trigger_name = self.trigger_name(ctx),

240 trigger_name = self.trigger_name(ctx),

175 );

241 );

176 transaction

242 transaction

@@ -262,10 +328,8 @@ impl Action for AddColumn {

262 // Remove triggers and procedures

328 // Remove triggers and procedures

263 let query = format!(

329 let query = format!(

264 r#"

330 r#"

265− DROP TRIGGER IF EXISTS "{trigger_name}" ON "{table}";

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

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

267 "#,

332 "#,

268− table = self.table,

269 trigger_name = self.trigger_name(ctx),

333 trigger_name = self.trigger_name(ctx),

270 );

334 );

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

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

src/migrations/alter_column.rs+1−1

@@ -150,7 +150,7 @@ impl Action for AlterColumn {

150 .context("failed to create up and down triggers")?;

151

152 // Backfill values in batches by touching the previous column

153− common::batch_touch_rows(db, &table.real_name, &column.real_name)

153+ common::batch_touch_rows(db, &table.real_name, Some(&column.real_name))

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

155

156 // Duplicate any indices to the temporary column

src/migrations/common.rs+14−9

@@ -69,7 +69,11 @@ impl ToSql for PostgresRawValue {

69 postgres::types::to_sql_checked!();

70}

72−pub fn batch_touch_rows(db: &mut dyn Conn, table: &str, column: &str) -> anyhow::Result<()> {

72+pub fn batch_touch_rows(

73+ db: &mut dyn Conn,

74+ table: &str,

75+ column: Option<&str>,

76+) -> anyhow::Result<()> {

73 const BATCH_SIZE: u16 = 1000;

77 const BATCH_SIZE: u16 = 1000;

75 let mut cursor: Option<PostgresRawValue> = None;

79 let mut cursor: Option<PostgresRawValue> = None;

@@ -78,6 +82,13 @@ pub fn batch_touch_rows(db: &mut dyn Conn, table: &str, column: &str) -> anyhow:

78 let mut params: Vec<&(dyn ToSql + Sync)> = Vec::new();

82 let mut params: Vec<&(dyn ToSql + Sync)> = Vec::new();

80 let primary_key = get_primary_key_columns_for_table(db, table)?;

84 let primary_key = get_primary_key_columns_for_table(db, table)?;

85+

86+ // If no column to touch is passed, we default to the first primary key column (just to make some "update")

87+ let touched_column = match column {

88+ Some(column) => column,

89+ None => primary_key.first().unwrap(),

90+ };

91+

81 let primary_key_columns = primary_key.join(", ");

92 let primary_key_columns = primary_key.join(", ");

83 let primary_key_where = primary_key

94 let primary_key_where = primary_key

@@ -120,8 +131,8 @@ pub fn batch_touch_rows(db: &mut dyn Conn, table: &str, column: &str) -> anyhow:

120 ORDER BY {primary_key_columns}

131 ORDER BY {primary_key_columns}

121 LIMIT {batch_size}

132 LIMIT {batch_size}

122 ), update AS (

133 ), update AS (

123− UPDATE public."{table}"

134+ UPDATE public."{table}" "{table}"

124− SET "{column}" = "{column}"

135+ SET "{touched_column}" = "{table}"."{touched_column}"

125 FROM rows

136 FROM rows

126 WHERE {primary_key_where}

137 WHERE {primary_key_where}

127 RETURNING {returning_columns}

138 RETURNING {returning_columns}

@@ -130,13 +141,7 @@ pub fn batch_touch_rows(db: &mut dyn Conn, table: &str, column: &str) -> anyhow:

130 FROM update

141 FROM update

131 LIMIT 1

142 LIMIT 1

132 "#,

143 "#,

133− table = table,

134− primary_key_columns = primary_key_columns,

135− cursor_where = cursor_where,

136 batch_size = BATCH_SIZE,

144 batch_size = BATCH_SIZE,

137− column = column,

138− primary_key_where = primary_key_where,

139− returning_columns = returning_columns,

140 );

145 );

141 let last_value = db

146 let last_value = db

142 .query_with_params(&query, &params)?

147 .query_with_params(&query, &params)?

tests/add_column.rs+99

@@ -260,3 +260,102 @@ fn add_column_with_default() {

260

261 test.run();

262}

263+

264+#[test]

265+fn add_column_with_complex_up() {

266+ let mut test = Test::new("Add column complex");

267+

268+ test.first_migration(

269+ r#"

270+ name = "create_tables"

271+

272+ [[actions]]

273+ type = "create_table"

274+ name = "users"

275+ primary_key = ["id"]

276+

277+ [[actions.columns]]

278+ name = "id"

279+ type = "INTEGER"

280+

281+ [[actions.columns]]

282+ name = "email"

283+ type = "TEXT"

284+

285+ [[actions]]

286+ type = "create_table"

287+ name = "profiles"

288+ primary_key = ["user_id"]

289+

290+ [[actions.columns]]

291+ name = "user_id"

292+ type = "INTEGER"

293+ "#,

294+ );

295+

296+ test.second_migration(

297+ r#"

298+ name = "add_profiles_email_column"

299+

300+ [[actions]]

301+ type = "add_column"

302+ table = "profiles"

303+

304+ [actions.column]

305+ name = "email"

306+ type = "TEXT"

307+ nullable = false

308+

309+ [actions.up]

310+ table = "users"

311+ value = "email"

312+ where = "user_id = id"

313+ "#,

314+ );

315+

316+ test.after_first(|db| {

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

318+ .unwrap();

319+ db.simple_query("INSERT INTO profiles (user_id) VALUES (1)")

320+ .unwrap();

321+ });

322+

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

324+ // Ensure email was backfilled on profiles

325+ let email: String = new_db

326+ .query(

327+ "

328+ SELECT email

329+ FROM profiles

330+ WHERE user_id = 1

331+ ",

332+ &[],

333+ )

334+ .unwrap()

335+ .first()

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

337+ .unwrap();

338+ assert_eq!("test@example.com", email);

339+

340+ // Ensure email change in old schema is propagated to profiles table in new schema

341+ old_db

342+ .simple_query("UPDATE users SET email = 'test2@example.com' WHERE id = 1")

343+ .unwrap();

344+ let email: String = new_db

345+ .query(

346+ "

347+ SELECT email

348+ FROM profiles

349+ WHERE user_id = 1

350+ ",

351+ &[],

352+ )

353+ .unwrap()

354+ .first()

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

356+ .unwrap();

357+ assert_eq!("test2@example.com", email);

358+ });

359+

360+ test.run();

361+}