814620a...main · reshape

git.iko.soy/reshape

on main help sign in

.github/workflows/lint.yaml+3−3

since splitsnapshots

unifiedsplit

@@ -15,7 +15,7 @@ jobs:

15 runs-on: ubuntu-latest

16 steps:

17 - name: Checkout

18− uses: actions/checkout@v2

18+ uses: actions/checkout@v4

19 - name: Select Rust toolchain with Clippy

20 uses: actions-rs/toolchain@v1

21 with:

@@ -23,6 +23,6 @@ jobs:

23 components: clippy

24 override: true

25 - name: Use cache for Rust dependencies

26− uses: Swatinem/rust-cache@v1

26+ uses: Swatinem/rust-cache@v2

27 - name: Lint using Clippy

28− run: cargo clippy

28+ run: cargo clippy

.github/workflows/publish-crate.yamlnew file+20

@@ -0,0 +1,20 @@

1+name: Publish crate

3+on: workflow_dispatch

5+jobs:

6+ publish-crate:

7+ runs-on: ubuntu-latest

8+ steps:

9+ - name: Checkout

10+ uses: actions/checkout@v2

11+ - name: Select Rust toolchain

12+ uses: actions-rs/toolchain@v1

13+ with:

14+ toolchain: stable

15+ - name: Publish to crates.io

16+ uses: actions-rs/cargo@v1

17+ with:

18+ command: publish

19+ env:

20+ CARGO_REGISTRY_TOKEN: ${{ secrets.CRATES_IO_TOKEN }}

.github/workflows/publish-docker.yamlnew file+25

@@ -0,0 +1,25 @@

1+name: Publish Docker image

3+on:

4+ workflow_dispatch:

5+ inputs:

6+ version:

7+ description: "Version (without 'v' prefix)"

8+ required: true

9+ type: string

10+

11+jobs:

12+ publish-docker:

13+ runs-on: ubuntu-latest

14+ steps:

15+ - name: Checkout

16+ uses: actions/checkout@v2

17+ - name: Log in to Docker

18+ env:

19+ DOCKER_PASSWORD: ${{ secrets.DOCKER_ACCESS_TOKEN }}

20+ run: |

21+ docker login -u ${{ secrets.DOCKER_USER }} -p $DOCKER_PASSWORD

22+ - name: Build Docker image

23+ run: docker build . --tag ${{ secrets.DOCKER_USER }}/reshape:${{ inputs.version }} --tag ${{ secrets.DOCKER_USER }}/reshape:latest

24+ - name: Push Docker image

25+ run: docker push --all-tags ${{ secrets.DOCKER_USER }}/reshape

Cargo.toml+7−3

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

1[package]

2name = "reshape"

3−version = "0.3.1"

3+version = "0.7.0"

4description = "An easy-to-use, zero-downtime schema migration tool for Postgres"

5homepage = "https://github.com/fabianlindfors/reshape"

6documentation = "https://github.com/fabianlindfors/reshape"

@@ -9,6 +9,7 @@ license = "MIT"

9keywords = ["postgres", "migrations"]

10edition = "2021"

11authors = ["Fabian Lindfors"]

12+rust-version = "1.70"

13[dependencies]

14[dependencies]

14postgres = { version = "0.19.2", features = ["with-serde_json-1"] }

15postgres = { version = "0.19.2", features = ["with-serde_json-1"] }

@@ -16,7 +17,10 @@ serde = { version = "1.0", features = ["derive"] }

16serde_json = "1.0"

17serde_json = "1.0"

17typetag = "0.1.7"

18typetag = "0.1.7"

18anyhow = { version = "1.0.44", features = ["backtrace"] }

19anyhow = { version = "1.0.44", features = ["backtrace"] }

19−clap = { version = "3.0.0", features = ["derive"] }

20+clap = { version = "3.1.9", features = ["derive"] }

20toml = "0.5"

21toml = "0.5"

21version = "3.0.0"

22version = "3.0.0"

22−colored = "2"

23+colored = "2"

24+rand = "0.8"

25+dotenv = "0.15.0"

26+lexical-sort = "0.3.1"

Dockerfile+1−1

@@ -1,4 +1,4 @@

1−FROM rust:1.58.0 AS builder

1+FROM rust:1.70.0 AS builder

2WORKDIR /usr/src/reshape

3COPY . .

4RUN cargo build --release

README.md+312−87

@@ -2,54 +2,56 @@

3[![Test status badge](https://github.com/fabianlindfors/Reshape/actions/workflows/test.yaml/badge.svg)](https://github.com/fabianlindfors/reshape/actions/workflows/test.yaml) [![Latest release](https://shields.io/github/v/release/fabianlindfors/reshape?display_name=tag&sort=semver&color=blue)](https://github.com/fabianlindfors/reshape/releases)

5+> Also check out [ReshapeDB](https://reshapedb.com), a new database built from the ground up to make zero-downtime schema and data migrations as simple and safe as possible. If you'd like to chat about it, please [reach out](contact@reshapedb.com)!

5Reshape is an easy-to-use, zero-downtime schema migration tool for Postgres. It automatically handles complex migrations that would normally require downtime or manual multi-step changes. During a migration, Reshape ensures both the old and new schema are available at the same time, allowing you to gradually roll out your application. It will also perform all changes without excessive locking, avoiding downtime caused by blocking other queries. For a more thorough introduction to Reshape, check out the [introductory blog post](https://fabianlindfors.se/blog/schema-migrations-in-postgres-using-reshape/).

7Reshape is an easy-to-use, zero-downtime schema migration tool for Postgres. It automatically handles complex migrations that would normally require downtime or manual multi-step changes. During a migration, Reshape ensures both the old and new schema are available at the same time, allowing you to gradually roll out your application. It will also perform all changes without excessive locking, avoiding downtime caused by blocking other queries. For a more thorough introduction to Reshape, check out the [introductory blog post](https://fabianlindfors.se/blog/schema-migrations-in-postgres-using-reshape/).

7Designed for Postgres 12 and later.

9Designed for Postgres 12 and later.

9−*Note: Reshape is **experimental** and should not be used in production. It can (and probably will) break your application.*

10−

11- [How it works](#how-it-works)

12- [Getting started](#getting-started)

13− - [Installation](#installation)

13+ - [Installation](#installation)

14− - [Creating your first migration](#creating-your-first-migration)

14+ - [Creating your first migration](#creating-your-first-migration)

15− - [Preparing your application](#preparing-your-application)

15+ - [Preparing your application](#preparing-your-application)

16− - [Running your migration](#running-your-migration)

16+ - [Running your migration](#running-your-migration)

17− - [Using during development](#using-during-development)

17+ - [Using during development](#using-during-development)

18- [Writing migrations](#writing-migrations)

19− - [Basics](#basics)

19+ - [Basics](#basics)

20− - [Tables](#tables)

20+ - [Tables](#tables)

21− - [Create table](#create-table)

21+ - [Create table](#create-table)

22− - [Rename table](#rename-table)

22+ - [Rename table](#rename-table)

23− - [Remove table](#remove-table)

23+ - [Remove table](#remove-table)

24− - [Add foreign key](#add-foreign-key)

24+ - [Add foreign key](#add-foreign-key)

25+ - [Remove foreign key](#remove-foreign-key)

25− - [Columns](#columns)

26+ - [Columns](#columns)

26− - [Add column](#add-column)

27+ - [Add column](#add-column)

27− - [Alter column](#alter-column)

28+ - [Alter column](#alter-column)

28− - [Remove column](#remove-column)

29+ - [Remove column](#remove-column)

29− - [Indices](#indices)

30+ - [Indices](#indices)

30− - [Add index](#add-index)

31+ - [Add index](#add-index)

31− - [Remove index](#remove-index)

32+ - [Remove index](#remove-index)

32− - [Enums](#enums)

33+ - [Enums](#enums)

33− - [Create enum](#create-enum)

34+ - [Create enum](#create-enum)

34− - [Remove enum](#remove-enum)

35+ - [Remove enum](#remove-enum)

36+ - [Custom](#custom)

37+ - [Complex changes across tables](#complex-changes-across-tables)

35- [Commands and options](#commands-and-options)

38- [Commands and options](#commands-and-options)

36− - [`reshape migrate`](#reshape-migrate)

39+ - [`reshape migration start`](#reshape-migration-start)

37− - [`reshape complete`](#reshape-complete)

40+ - [`reshape migration complete`](#reshape-migration-complete)

38− - [`reshape abort`](#reshape-abort)

41+ - [`reshape migration abort`](#reshape-migration-abort)

39− - [`reshape generate-schema-query`](#reshape-generate-schema-query)

42+ - [`reshape schema-query`](#reshape-schema-query)

40− - [Connection options](#connection-options)

43+ - [Connection options](#connection-options)

41- [License](#license)

44- [License](#license)

43−

44## How it works

46## How it works

46Reshape works by creating views that encapsulate the underlying tables, which your application will interact with. During a migration, Reshape will automatically create a new set of views and set up triggers to translate inserts and updates between the old and new schema. This means that every deployment is a three-phase process:

48Reshape works by creating views that encapsulate the underlying tables, which your application will interact with. During a migration, Reshape will automatically create a new set of views and set up triggers to translate inserts and updates between the old and new schema. This means that every deployment is a three-phase process:

48−1. **Start migration** (`reshape migrate`): Sets up views and triggers to ensure both the new and old schema are usable at the same time.

50+1. **Start migration** (`reshape migration start`): Sets up views and triggers to ensure both the new and old schema are usable at the same time.

492. **Roll out application**: Your application can be gradually rolled out without downtime. The existing deployment will continue using the old schema whilst the new deployment uses the new schema.

512. **Roll out application**: Your application can be gradually rolled out without downtime. The existing deployment will continue using the old schema whilst the new deployment uses the new schema.

50−3. **Complete migration** (`reshape complete`): Removes the old schema and any intermediate data and triggers.

52+3. **Complete migration** (`reshape migration complete`): Removes the old schema and any intermediate data and triggers.

52−If the application deployment fails, you should run `reshape abort` which will roll back any changes made by `reshape migrate` without losing data.

54+If the application deployment fails, you should run `reshape migration abort` which will roll back any changes made by `reshape migration start` without losing data.

54## Getting started

56## Getting started

@@ -61,7 +63,7 @@ Binaries are available for macOS and Linux under [Releases](https://github.com/f

62#### Cargo

64#### Cargo

64−Reshape can be installed using [Cargo](https://doc.rust-lang.org/cargo/):

66+Reshape can be installed using [Cargo](https://doc.rust-lang.org/cargo/) (requires Rust 1.58 or later):

66```shell

68```shell

67cargo install reshape

69cargo install reshape

@@ -69,10 +71,10 @@ cargo install reshape

70#### Docker

72#### Docker

72−Reshape is available as a Docker image on [Docker Hub](https://hub.docker.com/repository/docker/fabianlindfors/reshape).

74+Reshape is available as a Docker image on [Docker Hub](https://hub.docker.com/repository/docker/fabianlindfors/reshape).

74```shell

76```shell

75−docker run -v $(pwd):/usr/share/app fabianlindfors/reshape reshape migrate

77+docker run -v $(pwd):/usr/share/app fabianlindfors/reshape reshape migration start

76```

78```

78### Creating your first migration

80### Creating your first migration

@@ -101,9 +103,14 @@ This is the equivalent of running `CREATE TABLE users (id INTEGER GENERATED ALWA

101

103

102### Preparing your application

104### Preparing your application

103

105

104−Reshape relies on your application using a specific schema. When establishing the connection to Postgres in your application, you need to run a query to select the most recent schema. This query can be generated using: `reshape generate-schema-query`.

106+Reshape relies on your application using a specific schema. When establishing the connection to Postgres in your application, you need to run a query to select the most recent schema. The simplest way to do this is to use one of the helper libraries:

105

107

108+- [Rust](https://github.com/fabianlindfors/reshape-helper)

109+- [Ruby (and Rails)](https://github.com/fabianlindfors/reshape-ruby)

110+- [Python (and Django)](https://github.com/fabianlindfors/reshape-python)

111+- [Go](https://github.com/leourbina/reshape-helper)

112+

106−To pass it along to your application, you can for example use an environment variable in your run script: `RESHAPE_SCHEMA_QUERY=$(reshape generate-schema-query)`. Then in your application:

113+If your application is not using one of the languages with an available helper library, you can instead generate the query with the command: `reshape schema-query`. To pass it along to your application, you can for example use an environment variable in your run script: `RESHAPE_SCHEMA_QUERY=$(reshape schema-query)`. Then in your application:

107

114

108```python

115```python

109# Example for Python

116# Example for Python

@@ -116,16 +123,16 @@ db.execute(reshape_schema_query)

116To create your new `users` table, run:

123To create your new `users` table, run:

117

124

118```bash

125```bash

119−reshape migrate

126+reshape migration start --complete

120```

127```

121

128

122−As this is the first migration, Reshape will automatically complete it. For subsequent migrations, you will need to first run `reshape migrate`, roll out your application and then complete the migration using `reshape complete`.

129+We use the `--complete` flag to automatically complete the migration. During a production deployment, you should first run `reshape migration start` followed by `reshape migration complete` once your application has been fully rolled out.

123

130

124If nothing else is specified, Reshape will try to connect to a Postgres database running on `localhost` using `postgres` as both username and password. See [Connection options](#connection-options) for details on how to change the connection settings.

131If nothing else is specified, Reshape will try to connect to a Postgres database running on `localhost` using `postgres` as both username and password. See [Connection options](#connection-options) for details on how to change the connection settings.

125

132

126### Using during development

133### Using during development

127

134

128−When adding new migrations during development, we recommend running `reshape migrate` but skipping `reshape complete`. This way, the new migrations can be iterated on by updating the migration file and running `reshape abort` followed by `reshape migrate`.

135+When adding new migrations during development, we recommend running `reshape migration start` but skipping `reshape migration complete`. This way, the new migrations can be iterated on by updating the migration file and running `reshape migration abort` followed by `reshape migration start`.

129

136

130## Writing migrations

137## Writing migrations

131

138

@@ -160,9 +167,9 @@ Every action has a `type`. The supported types are detailed below.

160

167

161#### Create table

168#### Create table

162

169

163−The `create_table` action will create a new table with the specified columns, indices and constraints.

170+The `create_table` action will create a new table with the specified columns, indices and constraints. You can optionally provide an `up` option to backfill values from an existing table.

164

171

165−*Example: create a `customers` table with a few columns and a primary key*

172+_Example: create a `customers` table with a few columns and a primary key_

166

173

167```toml

174```toml

168[[actions]]

175[[actions]]

@@ -180,13 +187,13 @@ primary_key = ["id"]

180 type = "TEXT"

187 type = "TEXT"

181

188

182 # Columns default to nullable

189 # Columns default to nullable

183− nullable = false

190+ nullable = false

184

191

185 # default can be any valid SQL value, in this case a string literal

192 # default can be any valid SQL value, in this case a string literal

186− default = "'PLACEHOLDER'"

193+ default = "'PLACEHOLDER'"

187```

194```

188

195

189−*Example: create `users` and `items` tables with a foreign key between them*

196+_Example: create `users` and `items` tables with a foreign key between them_

190

197

191```toml

198```toml

192[[actions]]

199[[actions]]

@@ -219,11 +226,34 @@ primary_key = ["id"]

219 referenced_columns = ["id"]

226 referenced_columns = ["id"]

220```

227```

221

228

229+_Example: create `profiles` table based on existing `users` table_

230+

231+```toml

232+[[actions]]

233+type = "create_table"

234+name = "profiles"

235+primary_key = ["user_id"]

236+

237+ [[actions.columns]]

238+ name = "user_id"

239+ type = "INTEGER"

240+

241+ [[actions.columns]]

242+ name = "user_email"

243+ type = "TEXT"

244+

245+ # Backfill from `users` table and copy `users.email` to `user_email` column

246+ # This will perform an upsert based on the primary key to avoid duplicate rows

247+ [actions.up]

248+ table = "users"

249+ values = { user_id = "id", user_email = "email" }

250+```

251+

222#### Rename table

252#### Rename table

223

253

224The `rename_table` action will change the name of an existing table.

254The `rename_table` action will change the name of an existing table.

225

255

226−*Example: change name of `users` table to `customers`*

256+_Example: change name of `users` table to `customers`_

227

257

228```toml

258```toml

229[[actions]]

259[[actions]]

@@ -236,7 +266,7 @@ new_name = "customers"

236

266

237The `remove_table` action will remove an existing table.

267The `remove_table` action will remove an existing table.

238

268

239−*Example: remove `users` table*

269+_Example: remove `users` table_

240

270

241```toml

271```toml

242[[actions]]

272[[actions]]

@@ -248,7 +278,7 @@ table = "users"

248

278

249The `add_foreign_key` action will add a foreign key between two existing tables. The migration will fail if the existing column values aren't valid references.

279The `add_foreign_key` action will add a foreign key between two existing tables. The migration will fail if the existing column values aren't valid references.

250

280

251−*Example: create foreign key from `items` to `users` table*

281+_Example: create foreign key from `items` to `users` table_

252

282

253```toml

283```toml

254[[actions]]

284[[actions]]

@@ -261,13 +291,26 @@ table = "items"

261 referenced_columns = ["id"]

291 referenced_columns = ["id"]

262```

292```

263

293

294+#### Remove foreign key

295+

296+The `remove_foreign_key` action will remove an existing foreign key. The foreign key will only be removed once the migration is completed, which means that your new application must continue to adhere to the foreign key constraint.

297+

298+_Example: remove foreign key `items_user_id_fkey` from `users` table_

299+

300+```toml

301+[[actions]]

302+type = "remove_foreign_key"

303+table = "items"

304+foreign_key = "items_user_id_fkey"

305+```

306+

264### Columns

307### Columns

265

308

266#### Add column

309#### Add column

267

310

268−The `add_column` action will add a new column to an existing table. You can optionally provide an `up` setting. This should be an SQL expression which will be run for all existing rows to backfill the new column.

311+The `add_column` action will add a new column to an existing table. You can optionally provide an `up` setting. This should be an SQL expression which will be run for all existing rows to backfill the new column. `up` may also reference another table to perform cross-table migrations (see ["Complex changes across tables"](#complex-changes-across-tables)).

269

312

270−*Example: add a new column `reference` to table `products`*

313+_Example: add a new column `reference` to table `products`_

271

314

272```toml

315```toml

273[[actions]]

316[[actions]]

@@ -281,7 +324,7 @@ table = "products"

281 default = "10"

324 default = "10"

282```

325```

283

326

284−*Example: replace an existing `name` column with two new columns, `first_name` and `last_name`*

327+_Example: replace an existing `name` column with two new columns, `first_name` and `last_name`_

285

328

286```toml

329```toml

287[[actions]]

330[[actions]]

@@ -317,14 +360,48 @@ column = "name"

317down = "first_name || ' ' || last_name"

360down = "first_name || ' ' || last_name"

318```

361```

319

362

363+_Example: extract nested value from unstructured JSON `data` column to new `name` column_

364+

365+```toml

366+[[actions]]

367+type = "add_column"

368+table = "users"

369+

370+# #>> '{}' converts the JSON string value to TEXT

371+up = "data['path']['to']['value'] #>> '{}'"

372+

373+ [actions.column]

374+ name = "name"

375+ type = "TEXT"

376+```

377+

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

379+

380+```toml

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

382+[[actions]]

383+type = "add_column"

384+table = "profiles"

385+

386+ [actions.column]

387+ name = "email"

388+ type = "TEXT"

389+ nullable = false

390+

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

392+ [actions.up]

393+ table = "users"

394+ value = "email"

395+ where = "user_id = id"

396+```

320

397

321#### Alter column

398#### Alter column

322

399

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

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

324

401

325−When performing more complex changes than a rename, `up` and `down` must be provided. These should be SQL expressions which determine how to transform between the new and old version of the column. Inside those expressions, you can reference the current column value by the column name.

402+When performing more complex changes than a rename, `up` and `down` should be provided. These should be SQL expressions which determine how to transform between the new and old version of the column. Inside those expressions, you can reference the current column value by the column name.

326

403

327−*Example: rename `last_name` column on `users` table to `family_name`*

404+_Example: rename `last_name` column on `users` table to `family_name`_

328

405

329```toml

406```toml

330[[actions]]

407[[actions]]

@@ -336,7 +413,7 @@ column = "last_name"

336 name = "family_name"

413 name = "family_name"

337```

414```

338

415

339−*Example: change the type of `reference` column from `INTEGER` to `TEXT`*

416+_Example: change the type of `reference` column from `INTEGER` to `TEXT`_

340

417

341```toml

418```toml

342[[actions]]

419[[actions]]

@@ -351,7 +428,7 @@ down = "CAST(reference AS INTEGER)" # Converts from text value to integer

351 type = "TEXT" # Previous type was 'INTEGER'

428 type = "TEXT" # Previous type was 'INTEGER'

352```

429```

353

430

354−*Example: increment all values of an `index` column by one*

431+_Example: increment all values of an `index` column by one_

355

432

356```toml

433```toml

357[[actions]]

434[[actions]]

@@ -366,7 +443,22 @@ down = "index - 1" # Decrement to revert for old schema

366 name = "index"

443 name = "index"

367```

444```

368

445

446+_Example: make `name` column not nullable_

447+

448+```toml

449+[[actions]]

450+type = "alter_column"

451+table = "users"

452+column = "name"

453+

454+# Use "N/A" for any rows that currently have a NULL name

455+up = "COALESCE(name, 'N/A')"

456+

457+ [actions.changes]

458+ nullable = false

459+```

460+

369−*Example: change default value of `created_at` column to current time*

461+_Example: change default value of `created_at` column to current time_

370

462

371```toml

463```toml

372[[actions]]

464[[actions]]

@@ -380,11 +472,11 @@ column = "created_at"

380

472

381#### Remove column

473#### Remove column

382

474

383−The `remove_column` action will remove an existing column from a table. You can optionally provide a `down` setting. This should be an SQL expression which will be used to determine values for the old schema when inserting or updating rows using the new schema. The `down` setting must be provided when the removed column is `NOT NULL` or doesn't have a default value.

475+The `remove_column` action will remove an existing column from a table. You can optionally provide a `down` setting. This should be an SQL expression which will be used to determine values for the old schema when inserting or updating rows using the new schema. `down` may also reference another table to perform cross-table migrations (see ["Complex changes across tables"](#complex-changes-across-tables)) . The `down` setting must be provided when the removed column is `NOT NULL` or doesn't have a default value.

384

476

385Any indices that cover the column will be removed.

477Any indices that cover the column will be removed.

386

478

387−*Example: remove column `name` from table `users`*

479+_Example: remove column `name` from table `users`_

388

480

389```toml

481```toml

390[[actions]]

482[[actions]]

@@ -396,18 +488,34 @@ column = "name"

396down = "'N/A'"

488down = "'N/A'"

397```

489```

398

490

491+_Example: remove `email` column from `users` table and use column from `profiles` table instead_

492+

493+```toml

494+[[actions]]

495+type = "remove_column"

496+table = "users"

497+column = "email"

498+

499+ # Our application will use the `profiles.email` column instead

500+ # For backwards compatibility, we will write back to the removed `email` column whenever `profiles` is changed

501+ [actions.down]

502+ table = "profiles"

503+ value = "profiles.email"

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

505+```

506+

399### Indices

507### Indices

400

508

401#### Add index

509#### Add index

402

510

403The `add_index` action will add a new index to an existing table.

511The `add_index` action will add a new index to an existing table.

404

512

405−*Example: create a `users` table with a unique index on the `name` column*

513+_Example: create a `users` table with a unique index on the `name` column_

406

514

407```toml

515```toml

408[[actions]]

516[[actions]]

409type = "create_table"

517type = "create_table"

410−table = "users"

518+name = "users"

411primary_key = "id"

519primary_key = "id"

412

520

413 [[actions.columns]]

521 [[actions.columns]]

@@ -431,7 +539,7 @@ table = "users"

431 unique = true

539 unique = true

432```

540```

433

541

434−*Example: add GIN index to `data` column on `products` table*

542+_Example: add GIN index to `data` column on `products` table_

435

543

436```toml

544```toml

437[[actions]]

545[[actions]]

@@ -450,7 +558,7 @@ table = "products"

450

558

451The `remove_index` action will remove an existing index. The index won't actually be removed until the migration is completed.

559The `remove_index` action will remove an existing index. The index won't actually be removed until the migration is completed.

452

560

453−*Example: remove the `name_idx` index*

561+_Example: remove the `name_idx` index_

454

562

455```toml

563```toml

456[[actions]]

564[[actions]]

@@ -464,7 +572,7 @@ index = "name_idx"

464

572

465The `create_enum` action will create a new [enum type](https://www.postgresql.org/docs/current/datatype-enum.html) with the specified values.

573The `create_enum` action will create a new [enum type](https://www.postgresql.org/docs/current/datatype-enum.html) with the specified values.

466

574

467−*Example: add a new `mood` enum type with three possible values*

575+_Example: add a new `mood` enum type with three possible values_

468

576

469```toml

577```toml

470[[actions]]

578[[actions]]

@@ -477,7 +585,7 @@ values = ["happy", "ok", "sad"]

477

585

478The `remove_enum` action will remove an existing [enum type](https://www.postgresql.org/docs/current/datatype-enum.html). Make sure all usages of the enum has been removed before running the migration. The enum will only be removed once the migration is completed.

586The `remove_enum` action will remove an existing [enum type](https://www.postgresql.org/docs/current/datatype-enum.html). Make sure all usages of the enum has been removed before running the migration. The enum will only be removed once the migration is completed.

479

587

480−*Example: remove the `mood` enum type*

588+_Example: remove the `mood` enum type_

481

589

482```toml

590```toml

483[[actions]]

591[[actions]]

@@ -485,61 +593,178 @@ type = "remove_enum"

485enum = "mood"

593enum = "mood"

486```

594```

487

595

596+### Custom

597+

598+The `custom` action lets you create a migration which runs custom SQL. It should be used with great care as it provides no guarantees of zero-downtime and will simply run whatever SQL is provided. Use other actions whenever possible as they are explicitly designed for zero downtime.

599+

600+There are three optional settings available which all accept SQL queries. All queries need to be idempotent, for example by using `IF NOT EXISTS` wherever available.

601+

602+- `start`: run when a migration is started using `reshape migration start`

603+- `complete`: run when a migration is completed using `reshape migration complete`

604+- `abort`: run when a migration is aborted using `reshape migration abort`

605+

606+_Example: enable PostGIS and pg_stat_statements extensions_

607+

608+```toml

609+[[actions]]

610+type = "custom"

611+

612+start = """

613+ CREATE EXTENSION IF NOT EXISTS postgis;

614+ CREATE EXTENSION IF NOT EXISTS pg_stat_statements;

615+"""

616+

617+abort = """

618+ DROP EXTENSION IF EXISTS postgis;

619+ DROP EXTENSION IF EXISTS pg_stat_statements;

620+"""

621+```

622+

623+### Complex changes across tables

624+

625+The `up` and `down` options available when creating tables, adding columns and removing columns can also perform more complex changes that span tables.

626+

627+_Example: move `email` column from `users` to `profiles` table_

628+

629+```toml

630+[[actions]]

631+type = "add_column"

632+table = "profiles"

633+

634+ [actions.column]

635+ name = "email"

636+ type = "TEXT"

637+ nullable = false

638+

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

640+ [actions.up]

641+ table = "users"

642+ value = "users.email"

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

644+

645+[[actions]]

646+type = "remove_column"

647+table = "users"

648+column = "email"

649+

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

651+ [actions.down]

652+ table = "profiles"

653+ value = "profiles.email"

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

655+```

656+

657+_Example: turn a 1:N relationship between `users` and `accounts` into N:M and change the format of the associated `role`_

658+

659+```toml

660+# Add `user_account_connections` as a junction table

661+[[actions]]

662+type = "create_table"

663+name = "user_account_connections"

664+primary_key = ["account_id", "user_id"]

665+

666+ [[actions.columns]]

667+ name = "account_id"

668+ type = "INTEGER"

669+

670+ [[actions.columns]]

671+ name = "user_id"

672+ type = "INTEGER"

673+

674+ # `role` is currently stored directly on the `users` table but is part of the relationship

675+ [[actions.columns]]

676+ name = "role"

677+ type = "TEXT"

678+ nullable = false

679+

680+ # Backfill the new table from `users` and uppercase the `role`

681+ [actions.up]

682+ table = "users"

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

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

685+

686+[[actions]]

687+type = "remove_column"

688+table = "users"

689+column = "account_id"

690+

691+ # When `user_account_connections` is updated, we write the `account_id` back to `users`

692+ [actions.down]

693+ table = "user_account_connections"

694+ value = "user_account_connections.account_id"

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

696+

697+[[actions]]

698+type = "remove_column"

699+table = "users"

700+column = "account_role"

701+

702+ # When `user_account_connections` is updated, we write the lowercase role back to `users`

703+ [actions.down]

704+ table = "user_account_connections"

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

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

707+```

708+

488## Commands and options

709## Commands and options

489

710

490−### `reshape migrate`

711+### `reshape migration start`

491

712

492−Starts a new migration, applying all migrations under `migrations/` that haven't yet been applied. After the command has completed, both the old and new schema will be usable at the same time. When you have rolled out the new version of your application which uses the new schema, you should run `reshape complete`.

713+Starts a new migration, applying all migrations under `migrations/` that haven't yet been applied. After the command has completed, both the old and new schema will be usable at the same time. When you have rolled out the new version of your application which uses the new schema, you should run `reshape migration complete`.

493

714

494#### Options

715#### Options

495

716

496−*See also [Connection options](#connection-options)*

717+_See also [Connection options](#connection-options)_

497

718

498−| Option | Default | Description |

719+| Option | Default | Description |

499−| ------ | ------- | ----------- |

720+| ------------------ | ------------- | --------------------------------------------------------------------------------------------------------------- |

500−| `--complete`, `-c` | `false` | Automatically complete migration after applying it. |

721+| `--complete`, `-c` | `false` | Automatically complete migration after applying it. |

501−| `--dirs` | `migrations/` | Directories to search for migration files. Multiple directories can be specified using `--dirs dir1 dir2 dir3`. |

722+| `--dirs` | `migrations/` | Directories to search for migration files. Multiple directories can be specified using `--dirs dir1 dir2 dir3`. |

502

723

503−### `reshape complete`

724+### `reshape migration complete`

504

725

505−Completes migrations previously started with `reshape complete`.

726+Completes migrations previously started with `reshape migration complete`.

506

727

507#### Options

728#### Options

508

729

509See [Connection options](#connection-options)

730See [Connection options](#connection-options)

510

731

511−### `reshape abort`

732+### `reshape migration abort`

512

733

513−Aborts any migrations which haven't yet been completed.

734+Aborts any migrations which haven't yet been completed.

514

735

515#### Options

736#### Options

516

737

517See [Connection options](#connection-options)

738See [Connection options](#connection-options)

518

739

519−### `reshape generate-schema-query`

740+### `reshape schema-query`

520

741

521Generates the SQL query you need to run in your application before using the database. This command does not require a database connection. Instead it will generate the query based on the latest migration in the `migrations/` directory (or the directories specified by `--dirs`).

742Generates the SQL query you need to run in your application before using the database. This command does not require a database connection. Instead it will generate the query based on the latest migration in the `migrations/` directory (or the directories specified by `--dirs`).

522

743

744+If your application is written in Rust, we recommend using the [Rust helper library](https://github.com/fabianlindfors/reshape-helper/) instead.

745+

523The query should look something like `SET search_path TO migration_1_initial_migration`.

746The query should look something like `SET search_path TO migration_1_initial_migration`.

524

747

525#### Options

748#### Options

526

749

527−| Option | Default | Description |

750+| Option | Default | Description |

528−| ------ | ------- | ----------- |

751+| -------- | ------------- | --------------------------------------------------------------------------------------------------------------- |

529| `--dirs` | `migrations/` | Directories to search for migration files. Multiple directories can be specified using `--dirs dir1 dir2 dir3`. |

752| `--dirs` | `migrations/` | Directories to search for migration files. Multiple directories can be specified using `--dirs dir1 dir2 dir3`. |

530

753

531### Connection options

754### Connection options

532

755

533−The options below can be used with all commands that communicate with Postgres. Use either a connection URL or specify each connection option individually.

756+The options below can be used with all commands that communicate with Postgres. Use either a [connection URL](https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING) or specify each connection option individually.

757+

758+All options can also be set using environment variables instead of flags. If a `.env` file exists, then variables will be automatically loaded from there.

534

759

535−| Option | Default | Description |

536−| ------ | ------- | ----------- |

761+| ------------ | ----------- | -------------------- | ------------------------------------------- |

537−| `--url` | | URI to connect to your Postgres database.<br>Can also be provided with the environment variable `DATABASE_URL`. |

762+| `--url` | | `DB_URL` | URL to your Postgres database |

538−| `--host` | `localhost` | Hostname to use when connecting to Postgres |

539−| `--port` | `5432` | Port which Postgres is listening on |

540−| `--database` | `postgres` | Database name |

541−| `--username` | `postgres` | Postgres username |

542−| `--password` | `postgres` | Postgres password |

543

768

544## License

769## License

545

770

src/db.rs+76−5

@@ -1,5 +1,8 @@

1+use std::{cmp::min, time::Duration};

1−use anyhow::anyhow;

3+use anyhow::{anyhow, Context};

2use postgres::{types::ToSql, NoTls, Row};

4use postgres::{types::ToSql, NoTls, Row};

5+use rand::prelude::*;

4// DbLocker wraps a regular DbConn, only allowing access using the

7// DbLocker wraps a regular DbConn, only allowing access using the

5// `lock` method. This method will acquire the advisory lock before

8// `lock` method. This method will acquire the advisory lock before

@@ -22,7 +25,24 @@ impl DbLocker {

22 const LOCK_KEY: i64 = 4036779288569897133;

25 const LOCK_KEY: i64 = 4036779288569897133;

24 pub fn connect(config: &postgres::Config) -> anyhow::Result<Self> {

27 pub fn connect(config: &postgres::Config) -> anyhow::Result<Self> {

25− let pg = config.connect(NoTls)?;

28+ let mut pg = config.connect(NoTls)?;

29+

30+ // When running DDL queries that acquire locks, we risk causing a "lock queue".

31+ // When attempting to acquire a lock, Postgres will wait for any long running queries to complete.

32+ // At the same time, it will block other queries until the lock has been acquired and released.

33+ // This has the bad effect of the long-running query blocking other queries because of us, forming

34+ // a queue of other queries until we release our lock.

35+ //

36+ // We set the lock_timeout setting to avoid this. This puts an upper bound for how long Postgres will

37+ // wait to acquire locks and also the maximum amount of time a long-running query can block other queries.

38+ // We should also add automatic retries to handle these timeouts gracefully.

39+ //

40+ // Reference: https://medium.com/paypal-tech/postgresql-at-scale-database-schema-changes-without-downtime-20d3749ed680

41+ //

42+ // TODO: Make lock_timeout configurable

43+ pg.simple_query("SET lock_timeout = '1s'")

44+ .context("failed to set lock_timeout")?;

45+

26 Ok(Self {

46 Ok(Self {

27 client: DbConn::new(pg),

47 client: DbConn::new(pg),

28 })

48 })

@@ -86,12 +106,12 @@ impl DbConn {

106

87impl Conn for DbConn {

107impl Conn for DbConn {

88 fn run(&mut self, query: &str) -> anyhow::Result<()> {

108 fn run(&mut self, query: &str) -> anyhow::Result<()> {

89− self.client.batch_execute(query)?;

109+ retry_automatically(|| self.client.batch_execute(query))?;

90 Ok(())

110 Ok(())

91 }

111 }

112

93 fn query(&mut self, query: &str) -> anyhow::Result<Vec<Row>> {

113 fn query(&mut self, query: &str) -> anyhow::Result<Vec<Row>> {

94− let rows = self.client.query(query, &[])?;

114+ let rows = retry_automatically(|| self.client.query(query, &[]))?;

95 Ok(rows)

115 Ok(rows)

96 }

116 }

117

@@ -100,7 +120,7 @@ impl Conn for DbConn {

100 query: &str,

120 query: &str,

101 params: &[&(dyn ToSql + Sync)],

121 params: &[&(dyn ToSql + Sync)],

102 ) -> anyhow::Result<Vec<Row>> {

122 ) -> anyhow::Result<Vec<Row>> {

103− let rows = self.client.query(query, params)?;

123+ let rows = retry_automatically(|| self.client.query(query, params))?;

104 Ok(rows)

124 Ok(rows)

105 }

125 }

106

126

@@ -151,3 +171,54 @@ impl Conn for Transaction<'_> {

151 Ok(Transaction { transaction })

171 Ok(Transaction { transaction })

152 }

172 }

153}

173}

174+

175+// Retry a database operation with exponential backoff and jitter

176+fn retry_automatically<T>(

177+ mut f: impl FnMut() -> Result<T, postgres::Error>,

178+) -> Result<T, postgres::Error> {

179+ const STARTING_WAIT_TIME: u64 = 100;

180+ const MAX_WAIT_TIME: u64 = 3_200;

181+ const MAX_ATTEMPTS: u32 = 10;

182+

183+ let mut rng = rand::thread_rng();

184+ let mut attempts = 0;

185+ loop {

186+ let result = f();

187+

188+ let error = match result {

189+ Ok(_) => return result,

190+ Err(err) => err,

191+ };

192+

193+ // If we got a database error, we check if it's retryable.

194+ // If we didn't get a database error, then it's most likely some kind of connection

195+ // error which should also be retried.

196+ if let Some(db_error) = error.as_db_error() {

197+ if !error_retryable(db_error) {

198+ return Err(error);

199+ }

200+ }

201+

202+ attempts += 1;

203+ if attempts >= MAX_ATTEMPTS {

204+ return Err(error);

205+ }

206+

207+ // The wait time increases exponentially, starting at 100ms and doubling up to a max of 3.2s.

208+ let wait_time = min(

209+ MAX_WAIT_TIME,

210+ STARTING_WAIT_TIME * u64::pow(2, attempts - 1),

211+ );

212+

213+ // The jitter is up to half the wait time

214+ let jitter: u64 = rng.gen_range(0..wait_time / 2);

215+

216+ std::thread::sleep(Duration::from_millis(wait_time + jitter));

217+ }

218+}

219+

220+// Check if a database error can be retried

221+fn error_retryable(error: &postgres::error::DbError) -> bool {

222+ // LOCK_NOT_AVAILABLE is caused by lock_timeout being exceeded

223+ matches!(error.code(), &postgres::error::SqlState::LOCK_NOT_AVAILABLE)

224+}

src/helpers.rs+7−16

@@ -2,37 +2,28 @@ use anyhow::Context;

3use crate::db::Conn;

5−pub fn set_up_helpers(db: &mut dyn Conn, current_migration: &Option<String>) -> anyhow::Result<()> {

5+pub fn set_up_helpers(db: &mut dyn Conn, target_migration: &str) -> anyhow::Result<()> {

6− let predicate = if let Some(current_migration) = current_migration {

7− format!(

8− "current_setting('search_path') = 'migration_{}' OR setting_bool",

9− current_migration

10− )

11− } else {

12− "setting_bool".to_string()

13− };

14−

15 let query = format!(

6 let query = format!(

16 "

7 "

17− CREATE OR REPLACE FUNCTION reshape.is_old_schema()

8+ CREATE OR REPLACE FUNCTION reshape.is_new_schema()

18 RETURNS BOOLEAN AS $$

9 RETURNS BOOLEAN AS $$

19 DECLARE

10 DECLARE

20− setting TEXT := current_setting('reshape.is_old_schema', TRUE);

11+ setting TEXT := current_setting('reshape.is_new_schema', TRUE);

21 setting_bool BOOLEAN := setting IS NOT NULL AND setting = 'YES';

12 setting_bool BOOLEAN := setting IS NOT NULL AND setting = 'YES';

22 BEGIN

13 BEGIN

23− RETURN {};

14+ RETURN current_setting('search_path') = 'migration_{}' OR setting_bool;

24 END

15 END

25 $$ language 'plpgsql';

16 $$ language 'plpgsql';

26 ",

17 ",

27− predicate

18+ target_migration,

28 );

19 );

29 db.query(&query)

20 db.query(&query)

30− .context("failed creating helper function reshape.is_old_schema()")?;

21+ .context("failed creating helper function reshape.is_new_schema()")?;

32 Ok(())

23 Ok(())

33}

24}

35pub fn tear_down_helpers(db: &mut dyn Conn) -> anyhow::Result<()> {

26pub fn tear_down_helpers(db: &mut dyn Conn) -> anyhow::Result<()> {

36− db.query("DROP FUNCTION IF EXISTS reshape.is_old_schema;")?;

27+ db.query("DROP FUNCTION IF EXISTS reshape.is_new_schema;")?;

37 Ok(())

28 Ok(())

38}

29}

src/lib.rs+29−36

@@ -30,6 +30,7 @@ impl Reshape {

30 pub fn new_with_options(

31 host: &str,

32 port: u16,

33+ database: &str,

33 username: &str,

34 username: &str,

34 password: &str,

35 password: &str,

35 ) -> anyhow::Result<Reshape> {

36 ) -> anyhow::Result<Reshape> {

@@ -38,6 +39,7 @@ impl Reshape {

38 .host(host)

39 .host(host)

39 .port(port)

40 .port(port)

40 .user(username)

41 .user(username)

42+ .dbname(database)

41 .password(password);

43 .password(password);

43 Self::new_with_config(&config)

45 Self::new_with_config(&config)

@@ -145,13 +147,15 @@ fn migrate(

145) -> anyhow::Result<()> {

147) -> anyhow::Result<()> {

146 // Make sure no migration is in progress

148 // Make sure no migration is in progress

147 if let State::InProgress { .. } = &state {

149 if let State::InProgress { .. } = &state {

150+ println!(

148− println!("Migration already in progress, please complete using 'reshape complete'");

151+ "Migration already in progress, please complete using 'reshape migration complete'"

152+ );

149 return Ok(());

153 return Ok(());

150 }

154 }

151

155

152 if let State::Completing { .. } = &state {

156 if let State::Completing { .. } = &state {

153 println!(

157 println!(

154− "Migration already in progress and has started completion, please finish using 'reshape complete'"

158+ "Migration already in progress and has started completion, please finish using 'reshape migration complete'"

155 );

159 );

156 return Ok(());

160 return Ok(());

157 }

161 }

@@ -160,7 +164,6 @@ fn migrate(

160 // with the already applied ones stored in the state. This will throw an error if the

164 // with the already applied ones stored in the state. This will throw an error if the

161 // two sets of migrations don't agree, for example if a new migration has been added

165 // two sets of migrations don't agree, for example if a new migration has been added

162 // in between two existing ones.

166 // in between two existing ones.

163− let current_migration = state::current_migration(db)?;

164 let remaining_migrations = state::remaining_migrations(db, migrations)?;

167 let remaining_migrations = state::remaining_migrations(db, migrations)?;

165 if remaining_migrations.is_empty() {

168 if remaining_migrations.is_empty() {

166 println!("No migrations left to apply");

169 println!("No migrations left to apply");

@@ -175,8 +178,8 @@ fn migrate(

175 {

178 {

176 if existing_migrations != &remaining_migrations {

179 if existing_migrations != &remaining_migrations {

177 return Err(anyhow!(

180 return Err(anyhow!(

178− "a previous migration seems to have failed without cleaning up. Please run `reshape abort` and then run migrate again."

181+ "a previous migration seems to have failed without cleaning up. Please run `reshape migration abort` and then run migrate again."

179− ));

182+ ));

180 }

183 }

181 }

184 }

182

185

@@ -189,14 +192,15 @@ fn migrate(

189

192

190 println!("Applying {} migrations\n", remaining_migrations.len());

193 println!("Applying {} migrations\n", remaining_migrations.len());

191

194

195+ let target_migration = remaining_migrations.last().unwrap().name.to_string();

192− helpers::set_up_helpers(db, &current_migration).context("failed to set up helpers")?;

196+ helpers::set_up_helpers(db, &target_migration).context("failed to set up helpers")?;

193

197

194 let mut new_schema = Schema::new();

198 let mut new_schema = Schema::new();

195 let mut last_migration_index = usize::MAX;

199 let mut last_migration_index = usize::MAX;

196 let mut last_action_index = usize::MAX;

200 let mut last_action_index = usize::MAX;

197 let mut result: anyhow::Result<()> = Ok(());

201 let mut result: anyhow::Result<()> = Ok(());

198

202

199− for (migration_index, migration) in remaining_migrations.iter().enumerate() {

203+ 'outer: for (migration_index, migration) in remaining_migrations.iter().enumerate() {

200 println!("Migrating '{}':", migration.name);

204 println!("Migrating '{}':", migration.name);

201 last_migration_index = migration_index;

205 last_migration_index = migration_index;

202

206

@@ -206,7 +210,8 @@ fn migrate(

206 let description = action.describe();

210 let description = action.describe();

207 print!(" + {} ", description);

211 print!(" + {} ", description);

208

212

213+ let ctx =

209− let ctx = MigrationContext::new(migration_index, action_index);

214+ MigrationContext::new(migration_index, action_index, state::current_migration(db)?);

210 result = action

215 result = action

211 .run(&ctx, db, &new_schema)

216 .run(&ctx, db, &new_schema)

212 .with_context(|| format!("failed to {}", description));

217 .with_context(|| format!("failed to {}", description));

@@ -216,7 +221,7 @@ fn migrate(

216 println!("{}", "done".green());

221 println!("{}", "done".green());

217 } else {

222 } else {

218 println!("{}", "failed".red());

223 println!("{}", "failed".red());

219− break;

224+ break 'outer;

220 }

225 }

221 }

226 }

222

227

@@ -230,7 +235,7 @@ fn migrate(

230 // Set to the Aborting state. This is to ensure that the failed

235 // Set to the Aborting state. This is to ensure that the failed

231 // migration is fully aborted and nothing is left dangling.

236 // migration is fully aborted and nothing is left dangling.

232 // If the abort is interrupted for any reason, the user can try again

237 // If the abort is interrupted for any reason, the user can try again

233− // by running `reshape abort`.

238+ // by running `reshape migration abort`.

234 state.aborting(

239 state.aborting(

235 remaining_migrations.clone(),

240 remaining_migrations.clone(),

236 last_migration_index + 1,

241 last_migration_index + 1,

@@ -244,7 +249,6 @@ fn migrate(

244 }

249 }

245

250

246 // Create schema and views for migration

251 // Create schema and views for migration

247− let target_migration = remaining_migrations.last().unwrap().name.to_string();

248 create_schema_for_migration(db, &target_migration, &new_schema)

252 create_schema_for_migration(db, &target_migration, &new_schema)

249 .with_context(|| format!("failed to create schema for migration {}", target_migration))?;

253 .with_context(|| format!("failed to create schema for migration {}", target_migration))?;

250

254

@@ -252,27 +256,14 @@ fn migrate(

252 state.in_progress(remaining_migrations);

256 state.in_progress(remaining_migrations);

253 state.save(db).context("failed to save in-progress state")?;

257 state.save(db).context("failed to save in-progress state")?;

254

258

255− // If we started from a blank slate, we can finish the migration immediately

256− if current_migration.is_none() {

257− println!("Automatically completing migrations\n");

258− complete(db, state).context("failed to automatically complete migrations")?;

259−

260− println!("Migrations complete:");

261− println!(

262− " - Run '{}' from your application to use the latest schema",

263− schema_query_for_migration(&target_migration)

264− );

265− } else {

266− println!("Migrations have been applied and the new schema is ready for use:");

259+ println!("Migrations have been applied and the new schema is ready for use:");

267− println!(

260+ println!(

268− " - Run '{}' from your application to use the latest schema",

261+ " - Run '{}' from your application to use the latest schema",

269− schema_query_for_migration(&target_migration)

262+ schema_query_for_migration(&target_migration)

270− );

263+ );

271− println!(

264+ println!(

272− " - Run 'reshape complete' once your application has been updated and the previous schema is no longer in use"

265+ " - Run 'reshape migration complete' once your application has been updated and the previous schema is no longer in use"

273− );

266+ );

274− }

275−

276 Ok(())

267 Ok(())

277}

268}

278

269

@@ -293,7 +284,7 @@ fn complete(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {

293 current_action_index

284 current_action_index

294 } => (migrations, current_migration_index, current_action_index),

285 } => (migrations, current_migration_index, current_action_index),

295 State::Aborting { .. } => {

286 State::Aborting { .. } => {

296− return Err(anyhow!("migration been aborted and can't be completed. Please finish using `reshape abort`."))

287+ return Err(anyhow!("migration been aborted and can't be completed. Please finish using `reshape migration abort`."))

297 }

288 }

298 State::Applying { .. } => {

289 State::Applying { .. } => {

299 return Err(anyhow!("a previous migration unexpectedly failed. Please run `reshape migrate` to try applying the migration again."))

290 return Err(anyhow!("a previous migration unexpectedly failed. Please run `reshape migrate` to try applying the migration again."))

@@ -330,7 +321,8 @@ fn complete(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {

330 let description = action.describe();

321 let description = action.describe();

331 print!(" + {} ", description);

322 print!(" + {} ", description);

332

323

324+ let ctx =

333− let ctx = MigrationContext::new(migration_index, action_index);

325+ MigrationContext::new(migration_index, action_index, state::current_migration(db)?);

334

326

335 // Update state to indicate that this action has been completed.

327 // Update state to indicate that this action has been completed.

336 // We won't save this new state until after the action has completed.

328 // We won't save this new state until after the action has completed.

@@ -421,7 +413,7 @@ fn abort(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {

421 last_action_index,

413 last_action_index,

422 } => (migrations, last_migration_index, last_action_index),

414 } => (migrations, last_migration_index, last_action_index),

423 State::Completing { .. } => {

415 State::Completing { .. } => {

424− return Err(anyhow!("Migration completion has already been started. Please run `reshape complete` again to finish it."));

416+ return Err(anyhow!("migration completion has already been started. Please run `reshape migration complete` again to finish it."));

425 }

417 }

426 State::Idle => {

418 State::Idle => {

427 println!("No migration is in progress");

419 println!("No migration is in progress");

@@ -455,7 +447,8 @@ fn abort(db: &mut DbConn, state: &mut State) -> anyhow::Result<()> {

455 continue;

447 continue;

456 }

448 }

457

449

450+ let ctx =

458− let ctx = MigrationContext::new(migration_index, action_index);

451+ MigrationContext::new(migration_index, action_index, state::current_migration(db)?);

459 action

452 action

460 .abort(&ctx, db)

453 .abort(&ctx, db)

461 .with_context(|| format!("failed to abort migration {}", migration.name))

454 .with_context(|| format!("failed to abort migration {}", migration.name))

src/main.rs+93−18

@@ -4,6 +4,7 @@ use std::{

4 path::Path,

5};

7+use anyhow::Context;

7use clap::{Args, Parser};

8use clap::{Args, Parser};

8use reshape::{

9use reshape::{

9 migrations::{Action, Migration},

10 migrations::{Action, Migration},

@@ -12,22 +13,69 @@ use reshape::{

12use serde::{Deserialize, Serialize};

13use serde::{Deserialize, Serialize};

14#[derive(Parser)]

15#[derive(Parser)]

16+#[clap(name = "Reshape", version, about)]

15struct Opts {

17struct Opts {

16 #[clap(subcommand)]

18 #[clap(subcommand)]

17 cmd: Command,

19 cmd: Command,

18}

20}

20#[derive(Parser)]

22#[derive(Parser)]

23+#[clap(about)]

21enum Command {

24enum Command {

25+ #[clap(subcommand)]

26+ Migration(MigrationCommand),

27+

28+ #[clap(

29+ about = "Output the query your application should use to select the right schema",

30+ display_order = 2

31+ )]

32+ SchemaQuery(FindMigrationsOptions),

33+

34+ #[clap(

35+ about = "Deprecated. Use `reshape schema-query` instead",

36+ display_order = 3

37+ )]

38+ GenerateSchemaQuery(FindMigrationsOptions),

39+

40+ #[clap(

41+ about = "Deprecated. Use `reshape migration start` instead",

42+ display_order = 4

43+ )]

22 Migrate(MigrateOptions),

44 Migrate(MigrateOptions),

45+ #[clap(

46+ about = "Deprecated. Use `reshape migration complete` instead",

47+ display_order = 5

48+ )]

23 Complete(ConnectionOptions),

49 Complete(ConnectionOptions),

50+ #[clap(

51+ about = "Deprecated. Use `reshape migration abort` instead",

52+ display_order = 6

53+ )]

24− Remove(ConnectionOptions),

54+ Abort(ConnectionOptions),

55+}

56+

57+#[derive(Parser)]

58+#[clap(about = "Commands for managing migrations", display_order = 1)]

59+enum MigrationCommand {

60+ #[clap(

61+ about = "Starts a new migration, applying any migrations which haven't yet been applied",

62+ display_order = 1

63+ )]

64+ Start(MigrateOptions),

65+

66+ #[clap(about = "Completes an in-progress migration", display_order = 2)]

67+ Complete(ConnectionOptions),

68+

69+ #[clap(

70+ about = "Aborts an in-progress migration without losing any data",

71+ display_order = 3

72+ )]

25 Abort(ConnectionOptions),

73 Abort(ConnectionOptions),

26− GenerateSchemaQuery(FindMigrationsOptions),

27}

74}

29#[derive(Args)]

76#[derive(Args)]

30struct MigrateOptions {

77struct MigrateOptions {

78+ // Some comment

31 #[clap(long, short)]

79 #[clap(long, short)]

32 complete: bool,

80 complete: bool,

33 #[clap(flatten)]

81 #[clap(flatten)]

@@ -65,7 +113,7 @@ fn main() -> anyhow::Result<()> {

113

66fn run(opts: Opts) -> anyhow::Result<()> {

114fn run(opts: Opts) -> anyhow::Result<()> {

67 match opts.cmd {

115 match opts.cmd {

68− Command::Migrate(opts) => {

116+ Command::Migration(MigrationCommand::Start(opts)) | Command::Migrate(opts) => {

69 let mut reshape = reshape_from_connection_options(&opts.connection_options)?;

117 let mut reshape = reshape_from_connection_options(&opts.connection_options)?;

70 let migrations = find_migrations(&opts.find_migrations_options)?;

118 let migrations = find_migrations(&opts.find_migrations_options)?;

71 reshape.migrate(migrations)?;

119 reshape.migrate(migrations)?;

@@ -77,20 +125,16 @@ fn run(opts: Opts) -> anyhow::Result<()> {

125

78 Ok(())

126 Ok(())

79 }

127 }

80− Command::Complete(opts) => {

128+ Command::Migration(MigrationCommand::Complete(opts)) | Command::Complete(opts) => {

81 let mut reshape = reshape_from_connection_options(&opts)?;

129 let mut reshape = reshape_from_connection_options(&opts)?;

82 reshape.complete()

130 reshape.complete()

83 }

131 }

84− Command::Remove(opts) => {

85− let mut reshape = reshape_from_connection_options(&opts)?;

86− reshape.remove()

87− }

88− Command::Abort(opts) => {

132+ Command::Migration(MigrationCommand::Abort(opts)) | Command::Abort(opts) => {

89 let mut reshape = reshape_from_connection_options(&opts)?;

133 let mut reshape = reshape_from_connection_options(&opts)?;

90 reshape.abort()

134 reshape.abort()

91 }

135 }

92− Command::GenerateSchemaQuery(find_migrations_options) => {

136+ Command::SchemaQuery(opts) | Command::GenerateSchemaQuery(opts) => {

93− let migrations = find_migrations(&find_migrations_options)?;

137+ let migrations = find_migrations(&opts)?;

94 let query = migrations

138 let query = migrations

95 .last()

139 .last()

96 .map(|migration| reshape::schema_query_for_migration(&migration.name));

140 .map(|migration| reshape::schema_query_for_migration(&migration.name));

@@ -102,13 +146,35 @@ fn run(opts: Opts) -> anyhow::Result<()> {

102}

146}

103

147

104fn reshape_from_connection_options(opts: &ConnectionOptions) -> anyhow::Result<Reshape> {

148fn reshape_from_connection_options(opts: &ConnectionOptions) -> anyhow::Result<Reshape> {

105− let env_url = std::env::var("POSTGRES_URL").ok();

149+ // Load environment variables from .env file if it exists

106− let url = env_url.as_ref().or_else(|| opts.url.as_ref());

150+ dotenv::dotenv().ok();

107

151

108− match url {

152+ let url_env = std::env::var("DB_URL").ok();

153+ let url = url_env.as_ref().or_else(|| opts.url.as_ref());

154+

155+ // Use the connection URL if it has been set

156+ if let Some(url) = url {

109− Some(url) => Reshape::new(url),

157+ return Reshape::new(url);

110− None => Reshape::new_with_options(&opts.host, opts.port, &opts.username, &opts.password),

111 }

158 }

159+

160+ let host_env = std::env::var("DB_HOST").ok();

161+ let host = host_env.as_ref().unwrap_or_else(|| &opts.host);

162+

163+ let port = std::env::var("DB_PORT")

164+ .ok()

165+ .and_then(|port| port.parse::<u16>().ok())

166+ .unwrap_or(opts.port);

167+

168+ let username_env = std::env::var("DB_USERNAME").ok();

169+ let username = username_env.as_ref().unwrap_or_else(|| &opts.username);

170+

171+ let password_env = std::env::var("DB_PASSWORD").ok();

172+ let password = password_env.as_ref().unwrap_or_else(|| &opts.password);

173+

174+ let database_env = std::env::var("DB_NAME").ok();

175+ let database = database_env.as_ref().unwrap_or_else(|| &opts.database);

176+

177+ Reshape::new_with_options(host, port, database, username, password)

112}

178}

113

179

114fn find_migrations(opts: &FindMigrationsOptions) -> anyhow::Result<Vec<Migration>> {

180fn find_migrations(opts: &FindMigrationsOptions) -> anyhow::Result<Vec<Migration>> {

@@ -130,7 +196,13 @@ fn find_migrations(opts: &FindMigrationsOptions) -> anyhow::Result<Vec<Migration

130 }

196 }

131

197

132 // Sort all files by their file names (without extension)

198 // Sort all files by their file names (without extension)

133− file_paths.sort_unstable_by_key(|path| path.as_path().file_stem().unwrap().to_os_string());

199+ // The files are sorted naturally, e.g. "1_test_migration" < "10_test_migration"

200+ file_paths.sort_unstable_by(|path1, path2| {

201+ let file1 = path1.as_path().file_stem().unwrap().to_str().unwrap();

202+ let file2 = path2.as_path().file_stem().unwrap().to_str().unwrap();

203+

204+ lexical_sort::natural_cmp(file1, file2)

205+ });

134

206

135 file_paths

207 file_paths

136 .iter()

208 .iter()

@@ -146,7 +218,10 @@ fn find_migrations(opts: &FindMigrationsOptions) -> anyhow::Result<Vec<Migration

146 .map(|result| {

218 .map(|result| {

147 result.and_then(|(path, data)| {

219 result.and_then(|(path, data)| {

148 let extension = path.extension().and_then(|ext| ext.to_str()).unwrap();

220 let extension = path.extension().and_then(|ext| ext.to_str()).unwrap();

221+ let file_migration =

149− let file_migration = decode_migration_file(&data, extension)?;

222+ decode_migration_file(&data, extension).with_context(|| {

223+ format!("failed to parse migration file {}", path.display())

224+ })?;

150

225

151 let file_name = path.file_stem().and_then(|name| name.to_str()).unwrap();

226 let file_name = path.file_stem().and_then(|name| name.to_str()).unwrap();

152 Ok(Migration {

227 Ok(Migration {

src/migrations/add_column.rs+214−58

@@ -3,14 +3,25 @@ use crate::{

3 db::{Conn, Transaction},

4 schema::Schema,

5};

6−use anyhow::Context;

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

7use 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 {

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

32 )

43 )

33 }

44 }

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+

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

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

36 format!(

56 format!(

37 "{}_add_column_not_null_{}_{}",

57 "{}_add_column_not_null_{}_{}",

@@ -61,7 +81,7 @@ impl Action for AddColumn {

61 let temp_column_name = self.temp_column_name(ctx);

81 let temp_column_name = self.temp_column_name(ctx);

63 let mut definition_parts = vec![

83 let mut definition_parts = vec![

64− temp_column_name.to_string(),

84+ format!("\"{}\"", temp_column_name.to_string()),

65 self.column.data_type.to_string(),

85 self.column.data_type.to_string(),

66 ];

86 ];

@@ -86,53 +106,189 @@ impl Action for AddColumn {

86 );

106 );

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

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

108

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+

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

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

90− let declarations: Vec<String> = table

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

91− .columns

92− .iter()

93− .map(|column| {

94− format!(

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

96− table = table.real_name,

97− alias = column.name,

98− real_name = column.real_name,

99− )

100− })

101− .collect();

102−

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

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

104− let query = format!(

125+ let query = format!(

105− r#"

126+ r#"

106− CREATE OR REPLACE FUNCTION {trigger_name}()

127+ CREATE OR REPLACE FUNCTION {trigger_name}()

107− RETURNS TRIGGER AS $$

128+ RETURNS TRIGGER AS $$

108− BEGIN

129+ BEGIN

109− IF reshape.is_old_schema() THEN

130+ IF NOT reshape.is_new_schema() THEN

110− DECLARE

131+ DECLARE

111− {declarations}

132+ {declarations}

112− BEGIN

133+ BEGIN

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

134+ NEW."{temp_column_name}" = {up};

114− END;

135+ END;

115− END IF;

136+ END IF;

116− RETURN NEW;

137+ RETURN NEW;

117− END

138+ END

118− $$ language 'plpgsql';

139+ $$ language 'plpgsql';

119−

140+

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

141+ 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}();

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

122− "#,

143+ "#,

123− temp_column_name = temp_column_name,

144+ temp_column_name = temp_column_name,

124− trigger_name = self.trigger_name(ctx),

145+ trigger_name = self.trigger_name(ctx),

125− up = up,

146+ up = up,

126− table = self.table,

147+ table = self.table,

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

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

128− );

149+ );

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

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

130− }

131

151

132− // Backfill values in batches

152+ // Backfill values in batches

133− if self.up.is_some() {

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

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

155+ }

156+

157+ if let Transformation::Update {

158+ table: from_table,

159+ value,

160+ r#where,

161+ } = up

162+ {

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+

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

169+

170+ let from_table_assignments: Vec<String> = from_table

171+ .columns

172+ .iter()

173+ .map(|column| {

174+ format!(

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

176+ table = from_table.name,

177+ alias = column.name,

178+ real_name = column.real_name,

179+ )

180+ })

181+ .collect();

182+

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

184+ let query = format!(

185+ r#"

186+ CREATE OR REPLACE FUNCTION {trigger_name}()

187+ RETURNS TRIGGER AS $$

188+ #variable_conflict use_variable

189+ BEGIN

190+ IF NOT reshape.is_new_schema() THEN

191+ DECLARE

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

193+ BEGIN

194+ {assignments}

195+

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

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

198+

199+ UPDATE public."{changed_table_real}"

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

201+ WHERE {where};

202+

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

204+ END;

205+ END IF;

206+ RETURN NEW;

207+ END

208+ $$ language 'plpgsql';

209+

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

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

212+ "#,

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

214+ changed_table_real = table.real_name,

215+ from_table = from_table.name,

216+ from_table_real = from_table.real_name,

217+ trigger_name = self.trigger_name(ctx),

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

219+ temp_column_name = temp_column_name,

220+ );

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

222+

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+

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

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

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

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

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

291+ }

136 }

292 }

137

293

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

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

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

142 if !self.column.nullable {

298 if !self.column.nullable {

143 let query = format!(

299 let query = format!(

144 r#"

300 r#"

145− ALTER TABLE "{table}"

301+ ALTER TABLE "{table}"

146− ADD CONSTRAINT "{constraint_name}"

302+ ADD CONSTRAINT "{constraint_name}"

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

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

148− "#,

304+ "#,

149 table = self.table,

305 table = self.table,

150 constraint_name = self.not_null_constraint_name(ctx),

306 constraint_name = self.not_null_constraint_name(ctx),

151 column = temp_column_name,

307 column = temp_column_name,

@@ -167,11 +323,11 @@ impl Action for AddColumn {

167 // Remove triggers and procedures

323 // Remove triggers and procedures

168 let query = format!(

324 let query = format!(

169 r#"

325 r#"

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

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

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

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

172 "#,

328 "#,

173− table = self.table,

174 trigger_name = self.trigger_name(ctx),

329 trigger_name = self.trigger_name(ctx),

330+ reverse_trigger_name = self.reverse_trigger_name(ctx),

175 );

331 );

176 transaction

332 transaction

177 .run(&query)

333 .run(&query)

@@ -262,11 +418,11 @@ impl Action for AddColumn {

262 // Remove triggers and procedures

418 // Remove triggers and procedures

263 let query = format!(

419 let query = format!(

264 r#"

420 r#"

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

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

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

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

267 "#,

423 "#,

268− table = self.table,

269 trigger_name = self.trigger_name(ctx),

424 trigger_name = self.trigger_name(ctx),

425+ reverse_trigger_name = self.reverse_trigger_name(ctx),

270 );

426 );

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

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

272

428

src/migrations/add_foreign_key.rs+1−1

@@ -48,7 +48,7 @@ impl Action for AddForeignKey {

48 ALTER TABLE "{table}"

49 ADD CONSTRAINT {constraint_name}

50 FOREIGN KEY ({columns})

51− REFERENCES {referenced_table} ({referenced_columns})

51+ REFERENCES "{referenced_table}" ({referenced_columns})

52 NOT VALID

53 "#,

54 table = table.real_name,

src/migrations/alter_column.rs+4−4

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

103 CREATE OR REPLACE FUNCTION {up_trigger}()

104 RETURNS TRIGGER AS $$

105 BEGIN

106− IF reshape.is_old_schema() THEN

106+ IF NOT reshape.is_new_schema() THEN

107 DECLARE

108 {declarations}

109 {existing_column} public.{table}.{existing_column_real}%TYPE := NEW.{existing_column_real};

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

121 CREATE OR REPLACE FUNCTION {down_trigger}()

122 RETURNS TRIGGER AS $$

123 BEGIN

124− IF NOT reshape.is_old_schema() THEN

124+ IF reshape.is_new_schema() THEN

125 DECLARE

126 {declarations}

127 {existing_column} public.{table}.{temp_column}%TYPE := NEW.{temp_column};

@@ -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

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

187 // This constraint is set as NOT VALID so it doesn't apply to existing rows and

188 // the existing rows don't need to be scanned under an exclusive lock.

189 // Thanks to this, we can set the full column as NOT NULL later with minimal locking.

190− if !column.nullable {

190+ if !self.changes.nullable.unwrap_or(column.nullable) {

191 let query = format!(

192 r#"

193 ALTER TABLE "{table}"

src/migrations/common.rs+14−13

@@ -69,17 +69,26 @@ 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− db.query("SET reshape.is_old_schema = 'YES'")?;

76−

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

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

79 loop {

81 loop {

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

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

82 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+

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

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

85 let primary_key_where = primary_key

94 let primary_key_where = primary_key

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

122 ORDER BY {primary_key_columns}

131 ORDER BY {primary_key_columns}

123 LIMIT {batch_size}

132 LIMIT {batch_size}

124 ), update AS (

133 ), update AS (

125− UPDATE public."{table}"

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

126− SET "{column}" = "{column}"

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

127 FROM rows

136 FROM rows

128 WHERE {primary_key_where}

137 WHERE {primary_key_where}

129 RETURNING {returning_columns}

138 RETURNING {returning_columns}

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

132 FROM update

141 FROM update

133 LIMIT 1

142 LIMIT 1

134 "#,

143 "#,

135− table = table,

136− primary_key_columns = primary_key_columns,

137− cursor_where = cursor_where,

138 batch_size = BATCH_SIZE,

144 batch_size = BATCH_SIZE,

139− column = column,

140− primary_key_where = primary_key_where,

141− returning_columns = returning_columns,

142 );

145 );

143 let last_value = db

146 let last_value = db

144 .query_with_params(&query, &params)?

147 .query_with_params(&query, &params)?

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

152 cursor = last_value

155 cursor = last_value

153 }

156 }

154

157

155− db.query("SET reshape.is_old_schema = ''")?;

156−

157 Ok(())

158 Ok(())

158}

159}

159

160

src/migrations/create_table.rs+122−9

@@ -1,6 +1,9 @@

1+use std::collections::HashMap;

1use super::{common::ForeignKey, Action, Column, MigrationContext};

3use super::{common::ForeignKey, Action, Column, MigrationContext};

2use crate::{

4use crate::{

3 db::{Conn, Transaction},

5 db::{Conn, Transaction},

6+ migrations::common,

4 schema::Schema,

7 schema::Schema,

5};

8};

6use anyhow::Context;

9use anyhow::Context;

@@ -14,6 +17,21 @@ pub struct CreateTable {

15 #[serde(default)]

18 #[serde(default)]

16 pub foreign_keys: Vec<ForeignKey>,

19 pub foreign_keys: Vec<ForeignKey>,

20+

21+ pub up: Option<Transformation>,

22+}

23+

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

25+pub struct Transformation {

26+ table: String,

27+ values: HashMap<String, String>,

28+ upsert_constraint: Option<String>,

29+}

30+

31+impl CreateTable {

32+ fn trigger_name(&self, ctx: &MigrationContext) -> String {

33+ format!("{}_create_table_{}", ctx.prefix(), self.name)

34+ }

17}

35}

19#[typetag::serde(name = "create_table")]

37#[typetag::serde(name = "create_table")]

@@ -24,7 +42,7 @@ impl Action for CreateTable {

25 fn run(

43 fn run(

26 &self,

44 &self,

27− _ctx: &MigrationContext,

45+ ctx: &MigrationContext,

28 db: &mut dyn Conn,

46 db: &mut dyn Conn,

29 schema: &Schema,

47 schema: &Schema,

30 ) -> anyhow::Result<()> {

48 ) -> anyhow::Result<()> {

@@ -85,7 +103,7 @@ impl Action for CreateTable {

85 ));

103 ));

86 }

104 }

105

88− db.run(&format!(

106+ let query = &format!(

89 r#"

107 r#"

90 CREATE TABLE "{name}" (

108 CREATE TABLE "{name}" (

91 {definition}

109 {definition}

@@ -93,26 +111,121 @@ impl Action for CreateTable {

93 "#,

111 "#,

94 name = self.name,

112 name = self.name,

95 definition = definition_rows.join(",\n"),

113 definition = definition_rows.join(",\n"),

96− ))

114+ );

115+ db.run(query).context("failed to create table")?;

116+

117+ if let Some(Transformation {

118+ table: from_table,

119+ values,

120+ upsert_constraint,

121+ }) = &self.up

122+ {

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

124+

125+ let declarations: Vec<String> = from_table

126+ .columns

127+ .iter()

128+ .map(|column| {

129+ format!(

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

131+ table = from_table.real_name,

132+ alias = column.name,

133+ real_name = column.real_name,

134+ )

135+ })

136+ .collect();

137+

138+ let (insert_columns, insert_values): (Vec<&str>, Vec<&str>) = values

139+ .iter()

140+ .map(|(k, v)| -> (&str, &str) { (k, v) }) // Force &String to &str

141+ .unzip();

142+

143+ let update_set: Vec<String> = values

144+ .iter()

145+ .map(|(field, value)| format!("\"{field}\" = {value}"))

146+ .collect();

147+

148+ // Constraint to check for conflicts. Defaults to the primary key constraint.

149+ let conflict_constraint_name = match upsert_constraint {

150+ Some(custom_constraint) => custom_constraint.clone(),

151+ _ => format!("{table}_pkey", table = self.name),

152+ };

153+

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

155+ let query = format!(

156+ r#"

157+ CREATE OR REPLACE FUNCTION {trigger_name}()

158+ RETURNS TRIGGER AS $$

159+ #variable_conflict use_variable

160+ BEGIN

161+ IF NOT reshape.is_new_schema() THEN

162+ DECLARE

163+ {declarations}

164+ BEGIN

165+ INSERT INTO public."{changed_table_real}" ({columns})

166+ VALUES ({values})

167+ ON CONFLICT ON CONSTRAINT "{conflict_constraint_name}"

168+ DO UPDATE SET

169+ {updates};

170+ END;

171+ END IF;

172+ RETURN NEW;

173+ END

174+ $$ language 'plpgsql';

175+

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

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

178+ "#,

179+ changed_table_real = self.name,

180+ from_table_real = from_table.real_name,

181+ trigger_name = self.trigger_name(ctx),

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

183+ columns = insert_columns.join(", "),

184+ values = insert_values.join(", "),

185+ updates = update_set.join(",\n"),

186+ );

97− .context("failed to create table")?;

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

188+

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

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

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

192+ }

193+

98 Ok(())

194 Ok(())

99 }

195 }

100

196

101 fn complete<'a>(

197 fn complete<'a>(

102 &self,

198 &self,

103− _ctx: &MigrationContext,

199+ ctx: &MigrationContext,

104− _db: &'a mut dyn Conn,

200+ db: &'a mut dyn Conn,

105 ) -> anyhow::Result<Option<Transaction<'a>>> {

201 ) -> anyhow::Result<Option<Transaction<'a>>> {

106− // Do nothing

202+ // Remove triggers and procedures

203+ let query = format!(

204+ r#"

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

206+ "#,

207+ trigger_name = self.trigger_name(ctx),

208+ );

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

210+

107 Ok(None)

211 Ok(None)

108 }

212 }

109

213

110 fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

214 fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

111

215

112− fn abort(&self, _ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {

216+ fn abort(&self, ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {

217+ // Remove triggers and procedures

218+ let query = format!(

219+ r#"

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

221+ "#,

222+ trigger_name = self.trigger_name(ctx),

223+ );

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

225+

113 db.run(&format!(

226 db.run(&format!(

114 r#"

227 r#"

115− DROP TABLE IF EXISTS {name}

228+ DROP TABLE IF EXISTS "{name}"

116 "#,

229 "#,

117 name = self.name,

230 name = self.name,

118 ))

231 ))

src/migrations/custom.rsnew file+61

@@ -0,0 +1,61 @@

1+use super::{Action, MigrationContext};

2+use crate::{

3+ db::{Conn, Transaction},

4+ schema::Schema,

5+};

6+use serde::{Deserialize, Serialize};

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

9+pub struct Custom {

10+ #[serde(default)]

11+ pub start: Option<String>,

12+

13+ #[serde(default)]

14+ pub complete: Option<String>,

15+

16+ #[serde(default)]

17+ pub abort: Option<String>,

18+}

19+

20+#[typetag::serde(name = "custom")]

21+impl Action for Custom {

22+ fn describe(&self) -> String {

23+ "Running custom migration".to_string()

24+ }

25+

26+ fn run(

27+ &self,

28+ _ctx: &MigrationContext,

29+ db: &mut dyn Conn,

30+ _schema: &Schema,

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

32+ if let Some(start_query) = &self.start {

33+ println!("Running query: {}", start_query);

34+ db.run(start_query)?;

35+ }

36+

37+ Ok(())

38+ }

39+

40+ fn complete<'a>(

41+ &self,

42+ _ctx: &MigrationContext,

43+ db: &'a mut dyn Conn,

44+ ) -> anyhow::Result<Option<Transaction<'a>>> {

45+ if let Some(complete_query) = &self.complete {

46+ db.run(complete_query)?;

47+ }

48+

49+ Ok(None)

50+ }

51+

52+ fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

53+

54+ fn abort(&self, _ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {

55+ if let Some(abort_query) = &self.abort {

56+ db.run(abort_query)?;

57+ }

58+

59+ Ok(())

60+ }

61+}

src/migrations/mod.rs+13−1

@@ -39,9 +39,15 @@ pub use create_enum::CreateEnum;

39mod remove_enum;

40pub use remove_enum::RemoveEnum;

42+mod custom;

43+pub use custom::Custom;

44+

42mod add_foreign_key;

45mod add_foreign_key;

43pub use add_foreign_key::AddForeignKey;

46pub use add_foreign_key::AddForeignKey;

48+mod remove_foreign_key;

49+pub use remove_foreign_key::RemoveForeignKey;

50+

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

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

46pub struct Migration {

52pub struct Migration {

47 pub name: String,

53 pub name: String,

@@ -82,13 +88,19 @@ impl Clone for Migration {

82pub struct MigrationContext {

88pub struct MigrationContext {

83 migration_index: usize,

89 migration_index: usize,

84 action_index: usize,

90 action_index: usize,

91+ existing_schema_name: Option<String>,

85}

92}

87impl MigrationContext {

94impl MigrationContext {

95+ pub fn new(

88− pub fn new(migration_index: usize, action_index: usize) -> Self {

96+ migration_index: usize,

97+ action_index: usize,

98+ existing_schema_name: Option<String>,

99+ ) -> Self {

89 MigrationContext {

100 MigrationContext {

90 migration_index,

101 migration_index,

91 action_index,

102 action_index,

103+ existing_schema_name,

92 }

104 }

93 }

105 }

106

src/migrations/remove_column.rs+339−34

@@ -3,14 +3,25 @@ use crate::{

3 db::{Conn, Transaction},

4 schema::Schema,

5};

6−use anyhow::Context;

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

7use serde::{Deserialize, Serialize};

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

10pub struct RemoveColumn {

11 pub table: String,

12 pub column: String,

13− pub down: Option<String>,

13+ pub down: 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 RemoveColumn {

27impl RemoveColumn {

@@ -22,6 +33,33 @@ impl RemoveColumn {

22 self.column

33 self.column

23 )

34 )

24 }

35 }

36+

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+

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

47+ format!(

48+ "{}_remove_column_{}_{}_nn",

49+ ctx.prefix(),

50+ self.table,

51+ self.column

52+ )

53+ }

54+

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

56+ format!(

57+ "{}_add_column_not_null_{}_{}",

58+ ctx.prefix(),

59+ self.table,

60+ self.column

61+ )

62+ }

25}

63}

27#[typetag::serde(name = "remove_column")]

65#[typetag::serde(name = "remove_column")]

@@ -39,10 +77,13 @@ impl Action for RemoveColumn {

39 db: &mut dyn Conn,

77 db: &mut dyn Conn,

40 schema: &Schema,

78 schema: &Schema,

41 ) -> anyhow::Result<()> {

79 ) -> anyhow::Result<()> {

80+ let table = schema.get_table(db, &self.table)?;

81+ let column = table

82+ .get_column(&self.column)

83+ .ok_or_else(|| anyhow!("no such column {} exists", self.column))?;

84+

42 // Add down trigger

85 // Add down trigger

43 if let Some(down) = &self.down {

86 if let Some(down) = &self.down {

44− let table = schema.get_table(db, &self.table)?;

45−

46 let declarations: Vec<String> = table

87 let declarations: Vec<String> = table

47 .columns

88 .columns

48 .iter()

89 .iter()

@@ -56,32 +97,226 @@ impl Action for RemoveColumn {

56 })

97 })

57 .collect();

98 .collect();

100+ if let Transformation::Simple(down) = down {

59− let query = format!(

101+ let query = format!(

60− r#"

102+ r#"

61− CREATE OR REPLACE FUNCTION {trigger_name}()

103+ CREATE OR REPLACE FUNCTION {trigger_name}()

62− RETURNS TRIGGER AS $$

104+ RETURNS TRIGGER AS $$

63− BEGIN

105+ BEGIN

64− IF NOT reshape.is_old_schema() IS NULL THEN

106+ IF reshape.is_new_schema() THEN

65− DECLARE

107+ DECLARE

108+ {declarations}

109+ BEGIN

110+ NEW.{column_name} = {down};

111+ END;

112+ END IF;

113+ RETURN NEW;

114+ END

115+ $$ language 'plpgsql';

116+

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

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

119+ "#,

120+ column_name = self.column,

121+ trigger_name = self.trigger_name(ctx),

122+ down = down,

123+ table = self.table,

66− {declarations}

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

125+ );

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

127+ }

128+

129+ if let Transformation::Update {

130+ table: from_table,

131+ value,

132+ r#where,

133+ } = down

134+ {

135+ let existing_schema_name = match &ctx.existing_schema_name {

136+ Some(name) => name,

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

138+ };

139+

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

141+

142+ let maybe_null_check = if !column.nullable {

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

144+ // We will add a null check to the down function on the new schema below as well to cover both cases.

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

146+ // NOT NULL is not checked at the end of a transaction, but immediately upon update.

147+ let query = format!(

148+ r#"

149+ CREATE OR REPLACE FUNCTION {trigger_name}()

150+ RETURNS TRIGGER AS $$

67 BEGIN

151 BEGIN

68− NEW.{column_name} = {down};

152+ IF NOT reshape.is_new_schema() THEN

153+ IF NEW.{column} IS NULL THEN

154+ RAISE EXCEPTION '{column} can not be null';

69− END;

155+ END IF;

70− END IF;

156+ END IF;

71− RETURN NEW;

157+ RETURN NEW;

72− END

158+ END

73− $$ language 'plpgsql';

159+ $$ language 'plpgsql';

74−

160+

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

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

162+

163+ CREATE CONSTRAINT TRIGGER "{trigger_name}"

164+ AFTER INSERT OR UPDATE

165+ ON "{table}"

166+ FOR EACH ROW

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

167+ EXECUTE PROCEDURE {trigger_name}();

77− "#,

168+ "#,

78− column_name = self.column,

169+ table = self.table,

79− trigger_name = self.trigger_name(ctx),

170+ trigger_name = self.not_null_constraint_trigger_name(ctx),

80− down = down,

171+ column = self.column,

172+ );

173+ db.run(&query)

174+ .context("failed to create null constraint trigger")?;

175+

176+ db.run(&format!(

177+ r#"

178+ ALTER TABLE {table}

179+ ALTER COLUMN {column}

180+ DROP NOT NULL

181+ "#,

81− table = self.table,

182+ table = self.table,

183+ column = self.column

184+ ))

185+ .context("failed to remove column not null constraint")?;

186+

187+ format!(

188+ r#"

189+ IF {value} IS NULL THEN

190+ RAISE EXCEPTION '{column_name} can not be null';

191+ END IF;

192+ "#,

193+ column_name = self.column,

194+ )

195+ } else {

196+ "".to_string()

197+ };

198+

199+ let into_variables = from_table

200+ .columns

201+ .iter()

202+ .map(|column| {

203+ format!(

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

205+ alias = column.name,

206+ real_name = column.real_name,

207+ )

208+ })

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

210+ .join(", ");

211+

212+ let query = format!(

213+ r#"

214+ CREATE OR REPLACE FUNCTION {trigger_name}()

215+ RETURNS TRIGGER AS $$

216+ #variable_conflict use_variable

217+ BEGIN

218+ IF reshape.is_new_schema() THEN

219+ DECLARE

220+ {from_table} record;

221+ BEGIN

222+ SELECT {into_variables}

223+ INTO {from_table};

224+

225+ {maybe_null_check}

226+

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

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

229+

230+ UPDATE "migration_{existing_schema_name}"."{changed_table}" "{changed_table}"

231+ SET "{column_name}" = {value}

232+ WHERE {where};

233+

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

235+ END;

236+ END IF;

237+ RETURN NEW;

238+ END

239+ $$ language 'plpgsql';

240+

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

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

243+ "#,

244+ changed_table = self.table,

245+ from_table = from_table.name,

246+ from_table_real = from_table.real_name,

247+ column_name = self.column,

248+ trigger_name = self.trigger_name(ctx),

249+ );

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

251+

252+ let changed_into_variables = table

253+ .columns

254+ .iter()

255+ .map(|column| {

256+ format!(

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

258+ alias = column.name,

259+ real_name = column.real_name,

260+ )

261+ })

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

263+ .join(", ");

264+

265+ let from_table_columns = from_table

266+ .columns

267+ .iter()

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

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

270+ .join(", ");

271+

272+ let query = format!(

273+ r#"

274+ CREATE OR REPLACE FUNCTION {trigger_name}()

275+ RETURNS TRIGGER AS $$

276+ #variable_conflict use_variable

277+ BEGIN

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

279+ DECLARE

280+ {changed_table} record;

281+ __temp_row record;

282+ BEGIN

283+ SELECT {changed_into_variables}

284+ INTO {changed_table};

285+

286+ SELECT *

287+ INTO __temp_row

288+ FROM (

289+ SELECT {from_table_columns}

290+ FROM public.{from_table_real}

291+ ) {from_table}

292+ WHERE {where};

293+

294+ DECLARE

295+ {from_table} record;

296+ BEGIN

297+ {from_table} := __temp_row;

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

299+ END;

300+ END;

301+ END IF;

302+ RETURN NEW;

303+ END

304+ $$ language 'plpgsql';

305+

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

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

308+ "#,

309+ changed_table = table.name,

310+ changed_table_real = table.real_name,

311+ from_table = from_table.name,

312+ from_table_real = from_table.real_name,

313+ column_name_real = column.real_name,

314+ trigger_name = self.reverse_trigger_name(ctx),

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

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

83− );

316+ );

317+ db.run(&query)

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

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

319+ }

85 }

320 }

321

87 Ok(())

322 Ok(())

@@ -111,12 +346,15 @@ impl Action for RemoveColumn {

111 ALTER TABLE "{table}"

346 ALTER TABLE "{table}"

112 DROP COLUMN IF EXISTS "{column}";

347 DROP COLUMN IF EXISTS "{column}";

113

348

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

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

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

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

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

116 "#,

352 "#,

117 table = self.table,

353 table = self.table,

118 column = self.column,

354 column = self.column,

119 trigger_name = self.trigger_name(ctx),

355 trigger_name = self.trigger_name(ctx),

356+ reverse_trigger_name = self.reverse_trigger_name(ctx),

357+ null_trigger_name = self.not_null_constraint_trigger_name(ctx),

120 );

358 );

121 db.run(&query)

359 db.run(&query)

122 .context("failed to drop column and down trigger")?;

360 .context("failed to drop column and down trigger")?;

@@ -133,14 +371,81 @@ impl Action for RemoveColumn {

133 }

371 }

134

372

135 fn abort(&self, ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {

373 fn abort(&self, ctx: &MigrationContext, db: &mut dyn Conn) -> anyhow::Result<()> {

374+ // We might have temporaily removed the NOT NULL check and have to reinstate it

375+ let has_not_null_function = !db

376+ .query_with_params(

377+ "

378+ SELECT routine_name

379+ FROM information_schema.routines

380+ WHERE routine_schema = 'public'

381+ AND routine_name = $1

382+ ",

383+ &[&self.not_null_constraint_trigger_name(ctx)],

384+ )

385+ .context("failed to get any NOT NULL function")?

386+ .is_empty();

387+

388+ if has_not_null_function {

389+ // Make column NOT NULL again without taking any long lived locks with a temporary constraint

390+ let query = format!(

391+ r#"

392+ ALTER TABLE "{table}"

393+ ADD CONSTRAINT "{constraint_name}"

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

395+ "#,

396+ table = self.table,

397+ constraint_name = self.not_null_constraint_name(ctx),

398+ column = self.column,

399+ );

400+ db.run(&query)

401+ .context("failed to add NOT NULL constraint")?;

402+

403+ let query = format!(

404+ r#"

405+ ALTER TABLE "{table}"

406+ VALIDATE CONSTRAINT "{constraint_name}"

407+ "#,

408+ table = self.table,

409+ constraint_name = self.not_null_constraint_name(ctx),

410+ );

411+ db.run(&query)

412+ .context("failed to validate NOT NULL constraint")?;

413+

414+ // This ALTER TABLE call will not require any exclusive locks as it can use the validated constraint from above

415+ db.run(&format!(

416+ r#"

417+ ALTER TABLE {table}

418+ ALTER COLUMN {column}

419+ SET NOT NULL

420+ "#,

421+ table = self.table,

422+ column = self.column

423+ ))

424+ .context("failed to reinstate column NOT NULL")?;

425+

426+ // Drop the temporary constraint

427+ let query = format!(

428+ r#"

429+ ALTER TABLE "{table}"

430+ DROP CONSTRAINT "{constraint_name}"

431+ "#,

432+ table = self.table,

433+ constraint_name = self.not_null_constraint_name(ctx),

434+ );

435+ db.run(&query)

436+ .context("failed to drop NOT NULL constraint")?;

437+ }

438+

136 // Remove function and trigger

439 // Remove function and trigger

137 db.run(&format!(

440 db.run(&format!(

138 r#"

441 r#"

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

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

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

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

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

141 "#,

445 "#,

142− table = self.table,

143 trigger_name = self.trigger_name(ctx),

446 trigger_name = self.trigger_name(ctx),

447+ reverse_trigger_name = self.reverse_trigger_name(ctx),

448+ null_trigger_name = self.not_null_constraint_trigger_name(ctx),

144 ))

449 ))

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

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

146

451

src/migrations/remove_foreign_key.rsnew file+91

@@ -0,0 +1,91 @@

1+use super::{Action, MigrationContext};

2+use crate::{

3+ db::{Conn, Transaction},

4+ schema::Schema,

5+};

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

7+use serde::{Deserialize, Serialize};

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

10+pub struct RemoveForeignKey {

11+ table: String,

12+ foreign_key: String,

13+}

14+

15+#[typetag::serde(name = "remove_foreign_key")]

16+impl Action for RemoveForeignKey {

17+ fn describe(&self) -> String {

18+ format!(

19+ "Removing foreign key \"{}\" from table \"{}\"",

20+ self.foreign_key, self.table

21+ )

22+ }

23+

24+ fn run(

25+ &self,

26+ _ctx: &MigrationContext,

27+ db: &mut dyn Conn,

28+ schema: &Schema,

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

30+ // The foreign key is only removed once the migration is completed.

31+ // Removing it earlier would be hard/undesirable for several reasons:

32+ // - Postgres doesn't have an easy way to temporarily disable a foreign key check.

33+ // If it did, we could disable the FK for the new schema.

34+ // - Even if we could, it probably wouldn't be a good idea as it would cause temporary

35+ // inconsistencies for the old schema which still expects the FK to hold.

36+ // - For the same reason, we can't remove the FK when the migration is first applied.

37+ // If the migration was to be aborted, then the FK would have to be recreated with

38+ // the risk that it would no longer be valid.

39+

40+ // Ensure foreign key exists

41+ let table = schema.get_table(db, &self.table)?;

42+ let fk_exists = !db

43+ .query(&format!(

44+ r#"

45+ SELECT constraint_name

46+ FROM information_schema.table_constraints

47+ WHERE

48+ constraint_type = 'FOREIGN KEY' AND

49+ table_name = '{table_name}' AND

50+ constraint_name = '{foreign_key}'

51+ "#,

52+ table_name = table.real_name,

53+ foreign_key = self.foreign_key,

54+ ))

55+ .context("failed to check for foreign key")?

56+ .is_empty();

57+

58+ if !fk_exists {

59+ return Err(anyhow!(

60+ "no foreign key \"{}\" exists on table \"{}\"",

61+ self.foreign_key,

62+ self.table

63+ ));

64+ }

65+

66+ Ok(())

67+ }

68+

69+ fn complete<'a>(

70+ &self,

71+ _ctx: &MigrationContext,

72+ db: &'a mut dyn Conn,

73+ ) -> anyhow::Result<Option<Transaction<'a>>> {

74+ db.run(&format!(

75+ r#"

76+ ALTER TABLE {table}

77+ DROP CONSTRAINT IF EXISTS {foreign_key}

78+ "#,

79+ table = self.table,

80+ foreign_key = self.foreign_key,

81+ ))

82+ .context("failed to remove foreign key")?;

83+ Ok(None)

84+ }

85+

86+ fn update_schema(&self, _ctx: &MigrationContext, _schema: &mut Schema) {}

87+

88+ fn abort(&self, _ctx: &MigrationContext, _db: &mut dyn Conn) -> anyhow::Result<()> {

89+ Ok(())

90+ }

91+}

src/schema.rs+1−1

@@ -209,7 +209,7 @@ impl Schema {

209 let real_columns: Vec<(String, String, bool, Option<String>)> = db

210 .query(&format!(

211 "

212− SELECT column_name, data_type, is_nullable, column_default

212+ SELECT column_name, CASE WHEN data_type = 'USER-DEFINED' THEN udt_name ELSE data_type END, is_nullable, column_default

213 FROM information_schema.columns

214 WHERE table_name = '{table}' AND table_schema = 'public'

215 ORDER BY ordinal_position

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 = "users.email"

312+ where = "profiles.user_id = users.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+}

tests/alter_column.rs+100−1

@@ -113,7 +113,6 @@ fn alter_column_set_not_null() {

113 table = "users"

114 column = "name"

115 up = "COALESCE(name, 'TEST_DEFAULT_VALUE')"

116− down = "name"

117

116

118 [actions.changes]

117 [actions.changes]

119 nullable = false

118 nullable = false

@@ -159,6 +158,106 @@ fn alter_column_set_not_null() {

159 .query_one("SELECT name from users WHERE id = 4", &[])

158 .query_one("SELECT name from users WHERE id = 4", &[])

160 .unwrap();

159 .unwrap();

161 assert_eq!("Jane Doe", result.get::<_, &str>("name"));

160 assert_eq!("Jane Doe", result.get::<_, &str>("name"));

161+

162+ // Ensure NULL can't be inserted using the new schema

163+ let result = new_db.simple_query("INSERT INTO users (id, name) VALUES (5, NULL)");

164+ assert!(result.is_err(), "expected insert to fail");

165+ });

166+

167+ test.after_completion(|db| {

168+ // Ensure NULL can't be inserted

169+ let result = db.simple_query("INSERT INTO users (id, name) VALUES (5, NULL)");

170+ assert!(result.is_err(), "expected insert to fail");

171+ });

172+

173+ test.after_abort(|db| {

174+ // Ensure NULL can be inserted

175+ let result = db.simple_query("INSERT INTO users (id, name) VALUES (5, NULL)");

176+ assert!(result.is_ok(), "expected insert to succeed");

177+ });

178+

179+ test.run();

180+}

181+

182+#[test]

183+fn alter_column_set_nullable() {

184+ let mut test = Test::new("Set column nullable");

185+

186+ test.first_migration(

187+ r#"

188+ name = "create_user_table"

189+

190+ [[actions]]

191+ type = "create_table"

192+ name = "users"

193+ primary_key = ["id"]

194+

195+ [[actions.columns]]

196+ name = "id"

197+ type = "INTEGER"

198+

199+ [[actions.columns]]

200+ name = "name"

201+ type = "TEXT"

202+ nullable = false

203+ "#,

204+ );

205+

206+ test.second_migration(

207+ r#"

208+ name = "set_name_nullable"

209+

210+ [[actions]]

211+ type = "alter_column"

212+ table = "users"

213+ column = "name"

214+ down = "COALESCE(name, 'TEST_DEFAULT_VALUE')"

215+

216+ [actions.changes]

217+ nullable = true

218+ "#,

219+ );

220+

221+ test.after_first(|db| {

222+ // Insert a test user

223+ db.simple_query(

224+ "

225+ INSERT INTO users (id, name) VALUES

226+ (1, 'John Doe')

227+ ",

228+ )

229+ .unwrap();

230+ });

231+

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

233+ // Insert data using new schema and make sure the old schema gets correct values

234+ new_db

235+ .simple_query("INSERT INTO users (id, name) VALUES (2, NULL)")

236+ .unwrap();

237+ let result = old_db

238+ .query_one("SELECT name from users WHERE id = 2", &[])

239+ .unwrap();

240+ assert_eq!("TEST_DEFAULT_VALUE", result.get::<_, &str>("name"));

241+

242+ // Ensure NULL can't be inserted using the old schema

243+ let result = old_db.simple_query("INSERT INTO users (id, name) VALUES (3, NULL)");

244+ assert!(result.is_err(), "expected insert to fail");

245+

246+ // Ensure NULL can be inserted using the new schema

247+ let result = new_db.simple_query("INSERT INTO users (id, name) VALUES (4, NULL)");

248+ assert!(result.is_ok(), "expected insert to succeed");

249+ });

250+

251+ test.after_completion(|db| {

252+ // Ensure NULL can be inserted

253+ let result = db.simple_query("INSERT INTO users (id, name) VALUES (5, NULL)");

254+ assert!(result.is_ok(), "expected insert to succeed");

255+ });

256+

257+ test.after_abort(|db| {

258+ // Ensure NULL can't be inserted

259+ let result = db.simple_query("INSERT INTO users (id, name) VALUES (5, NULL)");

260+ assert!(result.is_err(), "expected insert to fail");

162 });

261 });

163

262

164 test.run();

263 test.run();

tests/common.rs+16−1

@@ -12,6 +12,7 @@ pub struct Test<'a> {

12 second_migration: Option<Migration>,

13 expect_failure: bool,

15+ clear_fn: Option<fn(&mut Client) -> ()>,

15 after_first_fn: Option<fn(&mut Client) -> ()>,

16 after_first_fn: Option<fn(&mut Client) -> ()>,

16 intermediate_fn: Option<fn(&mut Client, &mut Client) -> ()>,

17 intermediate_fn: Option<fn(&mut Client, &mut Client) -> ()>,

17 after_completion_fn: Option<fn(&mut Client) -> ()>,

18 after_completion_fn: Option<fn(&mut Client) -> ()>,

@@ -36,6 +37,7 @@ impl Test<'_> {

36 first_migration: None,

37 first_migration: None,

37 second_migration: None,

38 second_migration: None,

38 expect_failure: false,

39 expect_failure: false,

40+ clear_fn: None,

39 after_first_fn: None,

41 after_first_fn: None,

40 intermediate_fn: None,

42 intermediate_fn: None,

41 after_completion_fn: None,

43 after_completion_fn: None,

@@ -54,6 +56,12 @@ impl Test<'_> {

54 self

56 self

55 }

57 }

59+ #[allow(dead_code)]

60+ pub fn clear(&mut self, f: fn(&mut Client) -> ()) -> &mut Self {

61+ self.clear_fn = Some(f);

62+ self

63+ }

64+

57 #[allow(dead_code)]

65 #[allow(dead_code)]

58 pub fn after_first(&mut self, f: fn(&mut Client) -> ()) -> &mut Self {

66 pub fn after_first(&mut self, f: fn(&mut Client) -> ()) -> &mut Self {

59 self.after_first_fn = Some(f);

67 self.after_first_fn = Some(f);

@@ -115,6 +123,10 @@ impl Test<'_> {

115 print_subheading("Clearing database");

123 print_subheading("Clearing database");

116 self.reshape.remove().unwrap();

124 self.reshape.remove().unwrap();

117

125

126+ if let Some(clear_fn) = self.clear_fn {

127+ clear_fn(&mut self.old_db);

128+ }

129+

118 // Apply first migration, will automatically complete

130 // Apply first migration, will automatically complete

119 print_subheading("Applying first migration");

131 print_subheading("Applying first migration");

120 let first_migration = self

132 let first_migration = self

@@ -128,6 +140,9 @@ impl Test<'_> {

128 .simple_query(&reshape::schema_query_for_migration(&first_migration.name))

140 .simple_query(&reshape::schema_query_for_migration(&first_migration.name))

129 .unwrap();

141 .unwrap();

130

142

143+ // Automatically complete first migration

144+ self.reshape.complete().unwrap();

145+

131 // Run setup function

146 // Run setup function

132 if let Some(after_first_fn) = self.after_first_fn {

147 if let Some(after_first_fn) = self.after_first_fn {

133 print_subheading("Running setup and first checks");

148 print_subheading("Running setup and first checks");

@@ -196,7 +211,7 @@ impl Test<'_> {

196}

211}

197

212

198fn print_heading(text: &str) {

213fn print_heading(text: &str) {

199− let delimiter = std::iter::repeat("=").take(50).collect::<String>();

214+ let delimiter = std::iter::repeat("=").take(80).collect::<String>();

200

215

201 println!();

216 println!();

202 println!();

217 println!();

tests/complex.rsnew file+379

@@ -0,0 +1,379 @@

1+mod common;

2+use 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, the equivalent `users.email` will also be updated

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+

163+#[test]

164+fn extract_relation_into_new_table() {

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

166+

167+ test.first_migration(

168+ r#"

169+ name = "create_tables"

170+

171+ [[actions]]

172+ type = "create_table"

173+ name = "accounts"

174+ primary_key = ["id"]

175+

176+ [[actions.columns]]

177+ name = "id"

178+ type = "INTEGER"

179+

180+ [[actions]]

181+ type = "create_table"

182+ name = "users"

183+ primary_key = ["id"]

184+

185+ [[actions.columns]]

186+ name = "id"

187+ type = "INTEGER"

188+

189+ [[actions.columns]]

190+ name = "account_id"

191+ type = "INTEGER"

192+ nullable = false

193+

194+ [[actions.columns]]

195+ name = "account_role"

196+ type = "TEXT"

197+ nullable = false

198+ "#,

199+ );

200+

201+ test.second_migration(

202+ r#"

203+ name = "add_account_user_connection"

204+

205+ [[actions]]

206+ type = "create_table"

207+ name = "user_account_connections"

208+ primary_key = ["account_id", "user_id"]

209+

210+ [[actions.columns]]

211+ name = "account_id"

212+ type = "INTEGER"

213+

214+ [[actions.columns]]

215+ name = "user_id"

216+ type = "INTEGER"

217+

218+ [[actions.columns]]

219+ name = "role"

220+ type = "TEXT"

221+ nullable = false

222+

223+ [actions.up]

224+ table = "users"

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

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

227+

228+ [[actions]]

229+ type = "remove_column"

230+ table = "users"

231+ column = "account_id"

232+

233+ [actions.down]

234+ table = "user_account_connections"

235+ value = "user_account_connections.account_id"

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

237+

238+ [[actions]]

239+ type = "remove_column"

240+ table = "users"

241+ column = "account_role"

242+

243+ [actions.down]

244+ table = "user_account_connections"

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

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

247+ "#,

248+ );

249+

250+ test.after_first(|db| {

251+ db.simple_query("INSERT INTO accounts (id) VALUES (1)")

252+ .unwrap();

253+ db.simple_query("INSERT INTO users (id, account_id, account_role) VALUES (1, 1, 'admin')")

254+ .unwrap();

255+ });

256+

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

258+ // Ensure connections was backfilled

259+ let rows: Vec<(i32, i32, String)> = new_db

260+ .query(

261+ "

262+ SELECT account_id, user_id, role

263+ FROM user_account_connections

264+ ",

265+ &[],

266+ )

267+ .unwrap()

268+ .iter()

269+ .map(|row| (row.get("account_id"), row.get("user_id"), row.get("role")))

270+ .collect();

271+ assert_eq!(1, rows.len());

272+

273+ let row = rows.first().unwrap();

274+ assert_eq!(1, row.0);

275+ assert_eq!(1, row.1);

276+ assert_eq!("ADMIN", row.2);

277+

278+ // Ensure inserted user in old schema creates a new connection

279+ old_db

280+ .simple_query(

281+ "INSERT INTO users (id, account_id, account_role) VALUES (2, 1, 'developer')",

282+ )

283+ .unwrap();

284+ assert!(

285+ new_db

286+ .query(

287+ "

288+ SELECT account_id, user_id, role

289+ FROM user_account_connections

290+ WHERE account_id = 1 AND user_id = 2 AND role = 'DEVELOPER'

291+ ",

292+ &[],

293+ )

294+ .unwrap()

295+ .len()

296+ == 1

297+ );

298+

299+ // Ensure NOT NULL constraint still applies to old schema

300+ let result = old_db

301+ .simple_query(

302+ "INSERT INTO users (id, account_id, account_role) VALUES (2, NULL, 'developer')",

303+ );

304+ assert!(result.is_err());

305+

306+ // Ensure updated user role in old schema updates connection in new schema

307+ old_db

308+ .simple_query("UPDATE users SET account_role = 'admin' WHERE id = 2")

309+ .unwrap();

310+ assert!(

311+ new_db

312+ .query(

313+ "

314+ SELECT account_id, user_id, role

315+ FROM user_account_connections

316+ WHERE account_id = 1 AND user_id = 2 AND role = 'ADMIN'

317+ ",

318+ &[],

319+ )

320+ .unwrap()

321+ .len()

322+ == 1

323+ );

324+

325+ // Ensure updated connection in new schema updates old schema user

326+ new_db

327+ .simple_query(

328+ "UPDATE user_account_connections SET role = 'DEVELOPER' WHERE account_id = 1 AND user_id = 2",

329+ )

330+ .unwrap();

331+ assert!(

332+ old_db

333+ .query(

334+ "

335+ SELECT id

336+ FROM users

337+ WHERE id = 2 AND account_id = 1 AND account_role = 'developer'

338+ ",

339+ &[],

340+ )

341+ .unwrap()

342+ .len()

343+ == 1

344+ );

345+

346+ // Ensure insert of user with connection through new schema updates user in old schema

347+ new_db

348+ .simple_query(

349+ r#"

350+ BEGIN;

351+ INSERT INTO users (id) VALUES (3);

352+ INSERT INTO user_account_connections (user_id, account_id, role) VALUES (3, 1, 'DEVELOPER');

353+ COMMIT;

354+ "#,

355+ )

356+ .unwrap();

357+ new_db

358+ .simple_query(

359+ "",

360+ )

361+ .unwrap();

362+ assert!(

363+ old_db

364+ .query(

365+ "

366+ SELECT id

367+ FROM users

368+ WHERE id = 3 AND account_id = 1 AND account_role = 'developer'

369+ ",

370+ &[],

371+ )

372+ .unwrap()

373+ .len()

374+ == 1

375+ );

376+ });

377+

378+ test.run();

379+}

tests/custom.rsnew file+128

@@ -0,0 +1,128 @@

1+mod common;

2+use common::Test;

4+#[test]

5+fn custom_enable_extension() {

6+ let mut test = Test::new("Custom migration");

8+ test.clear(|db| {

9+ db.simple_query(

10+ "

11+ DROP EXTENSION IF EXISTS bloom;

12+ DROP EXTENSION IF EXISTS btree_gin;

13+ DROP EXTENSION IF EXISTS btree_gist;

14+ ",

15+ )

16+ .unwrap();

17+ });

18+

19+ test.first_migration(

20+ r#"

21+ name = "empty_migration"

22+

23+ [[actions]]

24+ type = "custom"

25+ "#,

26+ );

27+

28+ test.second_migration(

29+ r#"

30+ name = "enable_extensions"

31+

32+ [[actions]]

33+ type = "custom"

34+

35+ start = """

36+ CREATE EXTENSION IF NOT EXISTS bloom;

37+ CREATE EXTENSION IF NOT EXISTS btree_gin;

38+ """

39+

40+ complete = "CREATE EXTENSION IF NOT EXISTS btree_gist"

41+

42+ abort = """

43+ DROP EXTENSION IF EXISTS bloom;

44+ DROP EXTENSION IF EXISTS btree_gin;

45+ """

46+ "#,

47+ );

48+

49+ test.intermediate(|db, _| {

50+ let bloom_activated = !db

51+ .query("SELECT * FROM pg_extension WHERE extname = 'bloom'", &[])

52+ .unwrap()

53+ .is_empty();

54+ assert!(bloom_activated);

55+

56+ let btree_gin_activated = !db

57+ .query(

58+ "SELECT * FROM pg_extension WHERE extname = 'btree_gin'",

59+ &[],

60+ )

61+ .unwrap()

62+ .is_empty();

63+ assert!(btree_gin_activated);

64+

65+ let btree_gist_activated = !db

66+ .query(

67+ "SELECT * FROM pg_extension WHERE extname = 'btree_gist'",

68+ &[],

69+ )

70+ .unwrap()

71+ .is_empty();

72+ assert!(!btree_gist_activated);

73+ });

74+

75+ test.after_completion(|db| {

76+ let bloom_activated = !db

77+ .query("SELECT * FROM pg_extension WHERE extname = 'bloom'", &[])

78+ .unwrap()

79+ .is_empty();

80+ assert!(bloom_activated);

81+

82+ let btree_gin_activated = !db

83+ .query(

84+ "SELECT * FROM pg_extension WHERE extname = 'btree_gin'",

85+ &[],

86+ )

87+ .unwrap()

88+ .is_empty();

89+ assert!(btree_gin_activated);

90+

91+ let btree_gist_activated = !db

92+ .query(

93+ "SELECT * FROM pg_extension WHERE extname = 'btree_gist'",

94+ &[],

95+ )

96+ .unwrap()

97+ .is_empty();

98+ assert!(btree_gist_activated);

99+ });

100+

101+ test.after_abort(|db| {

102+ let bloom_activated = !db

103+ .query("SELECT * FROM pg_extension WHERE extname = 'bloom'", &[])

104+ .unwrap()

105+ .is_empty();

106+ assert!(!bloom_activated);

107+

108+ let btree_gin_activated = !db

109+ .query(

110+ "SELECT * FROM pg_extension WHERE extname = 'btree_gin'",

111+ &[],

112+ )

113+ .unwrap()

114+ .is_empty();

115+ assert!(!btree_gin_activated);

116+

117+ let btree_gist_activated = !db

118+ .query(

119+ "SELECT * FROM pg_extension WHERE extname = 'btree_gist'",

120+ &[],

121+ )

122+ .unwrap()

123+ .is_empty();

124+ assert!(!btree_gist_activated);

125+ });

126+

127+ test.run();

128+}

tests/remove_column.rs+86

@@ -128,4 +128,90 @@ fn remove_column_with_index() {

128

129 assert_eq!(0, count, "expected index to not exist");

130 });

131+

132+ test.run();

133+}

134+

135+#[test]

136+fn remove_column_with_complex_down() {

137+ let mut test = Test::new("Remove column complex");

138+

139+ test.first_migration(

140+ r#"

141+ name = "create_tables"

142+

143+ [[actions]]

144+ type = "create_table"

145+ name = "users"

146+ primary_key = ["id"]

147+

148+ [[actions.columns]]

149+ name = "id"

150+ type = "INTEGER"

151+

152+ [[actions.columns]]

153+ name = "email"

154+ type = "TEXT"

155+

156+ [[actions]]

157+ type = "create_table"

158+ name = "profiles"

159+ primary_key = ["user_id"]

160+

161+ [[actions.columns]]

162+ name = "user_id"

163+ type = "INTEGER"

164+

165+ [[actions.columns]]

166+ name = "email"

167+ type = "TEXT"

168+ "#,

169+ );

170+

171+ test.second_migration(

172+ r#"

173+ name = "remove_users_email_column"

174+

175+ [[actions]]

176+ type = "remove_column"

177+ table = "users"

178+ column = "email"

179+

180+ [actions.down]

181+ table = "profiles"

182+ value = "profiles.email"

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

184+ "#,

185+ );

186+

187+ test.after_first(|db| {

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

189+ .unwrap();

190+ db.simple_query("INSERT INTO profiles (user_id, email) VALUES (1, 'test@example.com')")

191+ .unwrap();

192+ });

193+

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

195+ new_db

196+ .simple_query("UPDATE profiles SET email = 'test2@example.com' WHERE user_id = 1")

197+ .unwrap();

198+

199+ // Ensure new email was propagated to users table in old schema

200+ let email: String = old_db

201+ .query(

202+ "

203+ SELECT email

204+ FROM users

205+ WHERE id = 1

206+ ",

207+ &[],

208+ )

209+ .unwrap()

210+ .first()

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

212+ .unwrap();

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

214+ });

215+

216+ test.run();

131}

217}

tests/remove_foreign_key.rsnew file+124

@@ -0,0 +1,124 @@

1+mod common;

2+use common::Test;

4+#[test]

5+fn remove_foreign_key() {

6+ let mut test = Test::new("Remove foreign key");

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]]

22+ type = "create_table"

23+ name = "items"

24+ primary_key = ["id"]

25+

26+ [[actions.columns]]

27+ name = "id"

28+ type = "INTEGER"

29+

30+ [[actions.columns]]

31+ name = "user_id"

32+ type = "INTEGER"

33+

34+ [[actions.foreign_keys]]

35+ columns = ["user_id"]

36+ referenced_table = "users"

37+ referenced_columns = ["id"]

38+ "#,

39+ );

40+

41+ test.second_migration(

42+ r#"

43+ name = "remove_foreign_key"

44+

45+ [[actions]]

46+ type = "remove_foreign_key"

47+ table = "items"

48+ foreign_key = "items_user_id_fkey"

49+ "#,

50+ );

51+

52+ test.after_first(|db| {

53+ // Insert some test users

54+ db.simple_query("INSERT INTO users (id) VALUES (1), (2)")

55+ .unwrap();

56+ });

57+

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

59+ // Ensure items can't be inserted if they don't reference valid users

60+ // The foreign key is only removed when the migration is completed so

61+ // it should still be enforced for the new and old schema.

62+ let result = old_db.simple_query("INSERT INTO items (id, user_id) VALUES (3, 3)");

63+ assert!(

64+ result.is_err(),

65+ "expected insert against old schema to fail"

66+ );

67+

68+ let result = new_db.simple_query("INSERT INTO items (id, user_id) VALUES (3, 3)");

69+ assert!(

70+ result.is_err(),

71+ "expected insert against new schema to fail"

72+ );

73+ });

74+

75+ test.after_completion(|db| {

76+ // Ensure items can be inserted even if they don't reference valid users

77+ let result = db

78+ .simple_query("INSERT INTO items (id, user_id) VALUES (5, 3)")

79+ .map(|_| ());

80+ assert!(

81+ result.is_ok(),

82+ "expected insert to not fail, got {:?}",

83+ result

84+ );

85+

86+ // Ensure foreign key doesn't exist

87+ let foreign_keys = db

88+ .query(

89+ "

90+ SELECT tc.constraint_name

91+ FROM information_schema.table_constraints AS tc

92+ WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name='items';

93+ ",

94+ &[],

95+ )

96+ .unwrap();

97+ assert!(

98+ foreign_keys.is_empty(),

99+ "expected no foreign keys to exist on items table"

100+ );

101+ });

102+

103+ test.after_abort(|db| {

104+ // Ensure items can't be inserted if they don't reference valid users

105+ let result = db.simple_query("INSERT INTO items (id, user_id) VALUES (3, 3)");

106+ assert!(result.is_err(), "expected insert to fail");

107+

108+ // Ensure foreign key still exists

109+ let fk_exists = !db

110+ .query(

111+ "

112+ SELECT tc.constraint_name

113+ FROM information_schema.table_constraints AS tc

114+ WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name='items';

115+ ",

116+ &[],

117+ )

118+ .unwrap()

119+ .is_empty();

120+ assert!(fk_exists);

121+ });

122+

123+ test.run()

124+}

tests/remove_table.rs+2

@@ -39,4 +39,6 @@ fn remove_table() {

39 // Ensure the table is not accessible through the new schema

40 assert!(new_db.query("SELECT id FROM users", &[]).is_err());

41 });

42+

43+ test.run();

42}

44}