git.iko.soy/reshape
on main help sign in
Initial commit
7d8302afabianlindfors
.github/workflows/lint.yamlnew file+25
compare with main
unifiedsplit
@@ -0,0 +1,25 @@
1+name: Lint
2+
3+on:
4+ push:
5+ branches: [ main ]
6+
7+env:
8+ CARGO_TERM_COLOR: always
9+
10+jobs:
11+ lint:
12+ runs-on: ubuntu-latest
13+ steps:
14+ - name: Checkout
15+ uses: actions/checkout@v2
16+ - name: Select Rust toolchain with Clippy
17+ uses: actions-rs/toolchain@v1
18+ with:
19+ toolchain: stable
20+ components: clippy
21+ override: true
22+ - name: Use cache for Rust dependencies
23+ uses: Swatinem/rust-cache@v1
24+ - name: Lint using Clippy
25+ run: cargo clippy
.github/workflows/release.yamlnew file+29
@@ -0,0 +1,29 @@
1+name: Release
2+
3+on:
4+ release:
5+ types: [ created ]
6+
7+env:
8+ CARGO_TERM_COLOR: always
9+
10+jobs:
11+ build:
12+ runs-on: ubuntu-latest
13+ steps:
14+ - name: Checkout
15+ uses: actions/checkout@v2
16+ - name: Select Rust toolchain
17+ uses: actions-rs/toolchain@v1
18+ with:
19+ toolchain: stable
20+ - name: Use cache for Rust dependencies
21+ uses: Swatinem/rust-cache@v1
22+ - name: Build
23+ run: cargo build --release && mv target/release/reshape ./reshape-linux_amd64
24+ - name: Save binary to release
25+ uses: skx/github-action-publish-binaries@master
26+ env:
27+ GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
28+ with:
29+ args: reshape-linux_amd64
.github/workflows/test.yamlnew file+42
@@ -0,0 +1,42 @@
1+name: Tests
2+
3+on:
4+ push:
5+ branches: [ main ]
6+
7+env:
8+ CARGO_TERM_COLOR: always
9+
10+jobs:
11+ integration-tests:
12+ runs-on: ubuntu-latest
13+
14+ services:
15+ postgres:
16+ image: postgres
17+ ports:
18+ - 5432:5432
19+ env:
20+ POSTGRES_DB: migra_test
21+ POSTGRES_USER: postgres
22+ POSTGRES_PASSWORD: postgres
23+ # Set health checks to wait until postgres has started
24+ options: >-
25+ --health-cmd pg_isready
26+ --health-interval 10s
27+ --health-timeout 5s
28+ --health-retries 5
29+
30+ steps:
31+ - name: Checkout
32+ uses: actions/checkout@v2
33+ - name: Select Rust toolchain
34+ uses: actions-rs/toolchain@v1
35+ with:
36+ toolchain: stable
37+ - name: Use cache for Rust dependencies
38+ uses: Swatinem/rust-cache@v1
39+ - name: Run integration tests
40+ run: cargo test -- --test-threads=1
41+ env:
42+ POSTGRES_CONNECTION_STRING: "postgres://postgres:postgres@127.0.0.1/migra_test"
.gitignorenew file+2
@@ -0,0 +1,2 @@
1+/target
2+Cargo.lock
Cargo.tomlnew file+15
@@ -0,0 +1,15 @@
1+[package]
2+name = "reshape"
3+version = "0.1.0"
4+edition = "2021"
5+
6+[dependencies]
7+postgres = { version = "0.19.2", features = ["with-serde_json-1"] }
8+serde = { version = "1.0", features = ["derive"] }
9+serde_json = "1.0"
10+typetag = "0.1.7"
11+anyhow = "1.0.44"
12+clap = "3.0.0-beta.5"
13+toml = "0.5"
14+version = "3.0.0"
15+colored = "2"
README.mdnew file+245
@@ -0,0 +1,245 @@
1+# Reshape
2+
3+[![Test status badge](https://github.com/fabianlindfors/Reshape/actions/workflows/test.yaml/badge.svg)](https://github.com/fabianlindfors/Reshape/actions/workflows/test.yaml)
4+
5+Reshape 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.
6+
7+*Note: Reshape is **experimental** and should not be used in production. It can (and probably will) destroy your data and break your application.*
8+
9+- [Getting started]()
10+ - [Installation]()
11+ - [Creating your first migration]()
12+ - [Preparing your application]()
13+ - [Running your migration]()
14+- [Writing migrations]()
15+ - [Basics]()
16+ - [Tables]()
17+ - [Create table]()
18+ - [Rename table]()
19+ - [Drop table]()
20+ - [Columns]()
21+ - [Create column]()
22+ - [Alter column]()
23+ - [Remove column]()
24+- [How it works]()
25+
26+## Getting started
27+
28+### Installation
29+
30+On macOS:
31+
32+```brew install reshape```
33+
34+On Debian:
35+
36+```apt-get install reshape```
37+
38+### Creating your first migration
39+
40+Each migration should be stored as a separate file under `migrations/`. The files can be in either JSON or TOML format. The name of the file will become the name of your migration and they will be sorted by file name. We recommend prefixing every migration with an incrementing number.
41+
42+Let's create a simple migration to set up a new table `users` with two fields, `id` and `name`. We'll create a file called `migration/1_create_users_table.toml`:
43+
44+```toml
45+[[actions]]
46+type = "create_table"
47+table = "users"
48+
49+ [[actions.columns]]
50+ name = "id"
51+ type = "SERIAL"
52+
53+ [[actions.columns]]
54+ name = "name"
55+ type = "TEXT"
56+```
57+
58+This is the equivalent of running `CREATE TABLE users (id SERIAL, name TEXT)`.
59+
60+### Preparing your application
61+
62+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`.
63+
64+To pass it along to your application, you could use an environment variable in your build script: `RESHAPE_SCHEMA_QUERY=$(reshape generate-schema-query)`. Then in your application:
65+
66+```python
67+# Example for Python
68+reshape_schema_query = os.getenv("RESHAPE_SCHEMA_QUERY")
69+db.execute(reshape_schema_query)
70+```
71+
72+### Running your migration
73+
74+To create your new `users` table, run:
75+
76+```bash
77+reshape migrate
78+```
79+
80+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`.
81+
82+## Writing migrations
83+
84+### Basics
85+
86+Every migration consists of one or more actions. The actions will be run sequentially. Here's an example of a migration with two actions to create two tables, `customers` and `products`:
87+
88+```toml
89+[[actions]]
90+type = "create_table"
91+table = "customers"
92+
93+ [[actions.columns]]
94+ name = "id"
95+ type = "SERIAL"
96+
97+[[actions]]
98+type = "create_table"
99+table = "products"
100+
101+ [[actions.columns]]
102+ name = "sku"
103+ type = "TEXT"
104+```
105+
106+Every action has a `type`. The supported types are detailed below.
107+
108+### Create table
109+
110+The `create_table` action will create a new table with the specified columns and indices.
111+
112+*Example: creating a `customers` table with a few columns and a primary key*
113+
114+```toml
115+[[actions]]
116+type = "create_table"
117+table = "customers"
118+primary_key = "id"
119+
120+ [[actions.columns]]
121+ name = "id"
122+ type = "SERIAL"
123+
124+ [[actions.columns]]
125+ name = "name"
126+ type = "SERIAL"
127+
128+ # Columns default to nullable
129+ nullable = false
130+
131+ # default can be any valid SQL value, in this case a string literal
132+ default = "'PLACEHOLDER'"
133+```
134+
135+### Add column
136+
137+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.
138+
139+*Example: add a new column `reference` to table `products`*
140+
141+```toml
142+[[actions]]
143+type = "add_column"
144+table = "products"
145+
146+ [actions.column]
147+ name = "reference"
148+ type = "INTEGER"
149+ nullable = false
150+ default = "10"
151+```
152+
153+*Example: replace an existing `name` column with two new columns, `first_name` and `last_name`*
154+
155+```toml
156+[[actions]]
157+type = "add_column"
158+table = "users"
159+
160+# Extract the first name from the existing name column
161+up = "(STRING_TO_ARRAY(name, ' '))[1]"
162+
163+ [actions.column]
164+ name = "first_name"
165+ type = "TEXT"
166+
167+
168+[[actions]]
169+type = "add_column"
170+table = "users"
171+
172+# Extract the last name from the existing name column
173+up = "(STRING_TO_ARRAY(name, ' '))[2]"
174+
175+ [actions.column]
176+ name = "last_name"
177+ type = "TEXT"
178+
179+
180+[[actions]]
181+type = "remove_column"
182+table = "users"
183+column = "name"
184+
185+# Reconstruct name column by concatenating first and last name
186+down = "first_name || ' ' || last_name"
187+```
188+
189+
190+### Alter column
191+
192+The `alter_column` action enables many different changes to an existing column, for example renaming, changing type and changing existing values.
193+
194+When performing more complex changes than a rename, `up` and `down` must be provided. These should be set to 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.
195+
196+*Example: rename `last_name` column on `users` table to `family_name`*
197+
198+```toml
199+[[actions]]
200+type = "alter_column"
201+table = "users"
202+column = "last_name"
203+
204+ [actions.changes]
205+ name = "family_name"
206+```
207+
208+*Example: change the type of `reference` column from `INTEGER` to `TEXT`*
209+
210+```toml
211+[[actions]]
212+type = "alter_column"
213+table = "users"
214+column = "reference"
215+
216+up = "CAST(reference AS TEXT)" # Converts from integer value to text
217+down = "CAST(reference AS INTEGER)" # Converts from text value to integer
218+
219+ [actions.changes]
220+ type = "TEXT" # Previous type was 'INTEGER'
221+```
222+
223+*Example: increment all values of a `index` column by one*
224+
225+```toml
226+[[actions]]
227+type = "alter_column"
228+table = "users"
229+
230+up = "index + 1" # Increment for new schema
231+down = "index - 1" # Decrement to revert for old schema
232+
233+ [actions.changes]
234+ name = "index"
235+
236+```
237+
238+
239+## How it works
240+
241+Reshape 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 migration is a two phase process:
242+
243+1. **Start migration** (`reshape migrate`): Create new views to ensure both the new and old schema are usable at the same time.
244+ - After phase one is complete, you can start the roll out of your application. Once the roll out is complete, the second phase can be run.
245+2. **Complete migration** (`reshape complete`): Removes the old schema and any intermediate data.
src/db.rsnew file+80
@@ -0,0 +1,80 @@
1+use postgres::{types::ToSql, NoTls, Row};
2+
3+pub trait Conn {
4+ fn run(&mut self, query: &str) -> anyhow::Result<()>;
5+ fn query(&mut self, query: &str) -> anyhow::Result<Vec<Row>>;
6+ fn query_with_params(
7+ &mut self,
8+ query: &str,
9+ params: &[&(dyn ToSql + Sync)],
10+ ) -> anyhow::Result<Vec<Row>>;
11+}
12+
13+pub struct DbConn {
14+ client: postgres::Client,
15+}
16+
17+impl DbConn {
18+ pub fn connect(connection_string: &str) -> anyhow::Result<DbConn> {
19+ let client = postgres::Client::connect(connection_string, NoTls)?;
20+ Ok(DbConn { client })
21+ }
22+
23+ pub fn transaction(&mut self) -> anyhow::Result<Transaction> {
24+ let transaction = self.client.transaction()?;
25+ Ok(Transaction { transaction })
26+ }
27+}
28+
29+impl Conn for DbConn {
30+ fn run(&mut self, query: &str) -> anyhow::Result<()> {
31+ self.client.batch_execute(query)?;
32+ Ok(())
33+ }
34+
35+ fn query(&mut self, query: &str) -> anyhow::Result<Vec<Row>> {
36+ let rows = self.client.query(query, &[])?;
37+ Ok(rows)
38+ }
39+
40+ fn query_with_params(
41+ &mut self,
42+ query: &str,
43+ params: &[&(dyn ToSql + Sync)],
44+ ) -> anyhow::Result<Vec<Row>> {
45+ let rows = self.client.query(query, params)?;
46+ Ok(rows)
47+ }
48+}
49+
50+pub struct Transaction<'a> {
51+ transaction: postgres::Transaction<'a>,
52+}
53+
54+impl Transaction<'_> {
55+ pub fn commit(self) -> anyhow::Result<()> {
56+ self.transaction.commit()?;
57+ Ok(())
58+ }
59+}
60+
61+impl Conn for Transaction<'_> {
62+ fn run(&mut self, query: &str) -> anyhow::Result<()> {
63+ self.transaction.batch_execute(query)?;
64+ Ok(())
65+ }
66+
67+ fn query(&mut self, query: &str) -> anyhow::Result<Vec<Row>> {
68+ let rows = self.transaction.query(query, &[])?;
69+ Ok(rows)
70+ }
71+
72+ fn query_with_params(
73+ &mut self,
74+ query: &str,
75+ params: &[&(dyn ToSql + Sync)],
76+ ) -> anyhow::Result<Vec<Row>> {
77+ let rows = self.transaction.query(query, params)?;
78+ Ok(rows)
79+ }
80+}
src/lib.rsnew file+297
@@ -0,0 +1,297 @@
1+use crate::migrations::Migration;
2+use colored::*;
3+use db::{Conn, DbConn};
4+use schema::Table;
5+
6+mod db;
7+pub mod migrations;
8+mod schema;
9+mod state;
10+
11+pub use crate::state::{State, Status};
12+
13+pub struct Reshape {
14+ pub state: State,
15+ db: DbConn,
16+}
17+
18+impl Reshape {
19+ pub fn new(connection_string: &str) -> anyhow::Result<Reshape> {
20+ let mut db = DbConn::connect(connection_string)?;
21+ let state = State::load(&mut db);
22+
23+ Ok(Reshape { db, state })
24+ }
25+
26+ pub fn migrate<T>(&mut self, migrations: T) -> anyhow::Result<()>
27+ where
28+ T: IntoIterator<Item = Migration>,
29+ {
30+ self.state.set_migrations(migrations)?;
31+ self.state.save(&mut self.db)?;
32+
33+ // Make sure no migration is in progress
34+ if let state::Status::InProgress {
35+ target_migration: _,
36+ target_schema: _,
37+ } = &self.state.status
38+ {
39+ println!("Migration already in progress, please complete using 'reshape complete'");
40+ return Ok(());
41+ }
42+
43+ let current_migration = &self.state.current_migration.clone();
44+ let remaining_migrations = Self::get_remaining_migrations(&self.state);
45+ if remaining_migrations.is_empty() {
46+ println!("No migrations left to apply");
47+ return Ok(());
48+ }
49+
50+ println!(" Applying {} migrations\n", remaining_migrations.len());
51+
52+ let target_migration = remaining_migrations.last().unwrap().name.to_string();
53+
54+ let mut new_schema = self.state.current_schema.clone();
55+
56+ for migration in remaining_migrations {
57+ println!("Migrating '{}':", migration.name);
58+
59+ for step in &migration.actions {
60+ print!(" + {} ", step.describe());
61+ step.run(&mut self.db, &new_schema)?;
62+ step.update_schema(&mut new_schema)?;
63+ println!("{}", "done".green());
64+ }
65+
66+ println!("");
67+ }
68+
69+ // Create schema and views for migration
70+ self.create_schema_for_migration(&target_migration, &new_schema)?;
71+
72+ // Update state once migrations have been performed
73+ self.state.status = state::Status::InProgress {
74+ target_migration: target_migration.to_string(),
75+ target_schema: new_schema,
76+ };
77+ self.state.save(&mut self.db)?;
78+
79+ // If we started from a blank slate, we can finish the migration immediately
80+ if current_migration.is_none() {
81+ println!("Automatically completing migrations\n");
82+ self.complete_migration()?;
83+
84+ println!("Migrations complete:");
85+ println!(
86+ " - Run '{}' from your application to use the latest schema",
87+ generate_schema_query(&target_migration)
88+ );
89+ } else {
90+ println!("Migrations have been applied and the new schema is ready for use:");
91+ println!(
92+ " - Run '{}' from your application to use the latest schema",
93+ generate_schema_query(&target_migration)
94+ );
95+ println!(
96+ " - Run 'reshape complete' once your application has been updated and the previous schema is no longer in use"
97+ );
98+ }
99+
100+ Ok(())
101+ }
102+
103+ pub fn complete_migration(&mut self) -> anyhow::Result<()> {
104+ // Make sure a migration is in progress
105+ let (target_migration, target_schema) = match &self.state.status {
106+ state::Status::InProgress {
107+ target_migration,
108+ target_schema,
109+ } => (target_migration, target_schema),
110+ _ => {
111+ println!("No migration in progress");
112+ return Ok(());
113+ }
114+ };
115+
116+ let remaining_migrations = Self::get_remaining_migrations(&self.state);
117+ let mut temp_schema = self.state.current_schema.clone();
118+
119+ // Run all the completion changes as a transaction to avoid incomplete updates
120+ let mut transaction = self.db.transaction()?;
121+
122+ // Remove previous migration's schema
123+ if let Some(current_migration) = &self.state.current_migration {
124+ let query = format!(
125+ "DROP SCHEMA IF EXISTS {} CASCADE",
126+ schema_name_for_migration(current_migration)
127+ );
128+ transaction.run(&query)?;
129+ }
130+
131+ for migration in remaining_migrations {
132+ println!("Completing '{}':", migration.name);
133+
134+ for step in &migration.actions {
135+ print!(" + {} ", step.describe());
136+ step.complete(&mut transaction, &temp_schema)?;
137+ step.update_schema(&mut temp_schema)?;
138+ println!("{}", "done".green());
139+ }
140+
141+ println!("");
142+ }
143+
144+ // Remove any temporary is new columns from tables
145+ for table in temp_schema.tables.values_mut() {
146+ if table.has_is_new {
147+ table.has_is_new = false;
148+
149+ let query = format!(
150+ "ALTER TABLE {table} DROP COLUMN __reshape_is_new CASCADE",
151+ table = table.name,
152+ );
153+ transaction.run(&query)?;
154+
155+ // The view will automatically be dropped by CASCADE so let's recreate it
156+ let schema = schema_name_for_migration(target_migration);
157+ Self::create_view_for_table(&mut transaction, table, &schema, false)?;
158+ }
159+ }
160+
161+ self.state.current_migration = Some(target_migration.to_string());
162+ self.state.current_schema = target_schema.clone();
163+ self.state.status = state::Status::Idle;
164+ self.state.save(&mut transaction)?;
165+
166+ transaction.commit()?;
167+
168+ Ok(())
169+ }
170+
171+ fn create_schema_for_migration(
172+ &mut self,
173+ migration_name: &str,
174+ schema: &schema::Schema,
175+ ) -> anyhow::Result<()> {
176+ // Create schema for migration
177+ let schema_name = schema_name_for_migration(migration_name);
178+ let create_schema_query = format!("CREATE SCHEMA IF NOT EXISTS {}", schema_name);
179+ self.db.run(&create_schema_query)?;
180+
181+ // Create views inside schema
182+ for table in schema.tables.values() {
183+ Self::create_view_for_table(&mut self.db, table, &schema_name, true)?;
184+ }
185+
186+ Ok(())
187+ }
188+
189+ fn create_view_for_table(
190+ db: &mut impl Conn,
191+ table: &Table,
192+ schema: &str,
193+ use_alias: bool,
194+ ) -> anyhow::Result<()> {
195+ let mut select_columns: Vec<String> = table
196+ .columns
197+ .iter()
198+ .map(|column| {
199+ if use_alias {
200+ format!("{} AS {}", column.real_name(), column.name)
201+ } else {
202+ column.name.to_string()
203+ }
204+ })
205+ .collect();
206+
207+ if table.has_is_new {
208+ select_columns.push("__reshape_is_new".to_string());
209+ }
210+
211+ let query = format!(
212+ "CREATE OR REPLACE VIEW {schema}.{table} AS
213+ SELECT {columns}
214+ FROM {table}",
215+ schema = schema,
216+ table = table.name,
217+ columns = select_columns.join(","),
218+ );
219+ db.run(&query)?;
220+
221+ if table.has_is_new {
222+ let query = format!(
223+ "ALTER VIEW {schema}.{view} ALTER __reshape_is_new SET DEFAULT TRUE",
224+ schema = schema,
225+ view = table.name,
226+ );
227+ db.run(&query)?;
228+ }
229+
230+ Ok(())
231+ }
232+
233+ fn get_remaining_migrations(state: &State) -> Vec<&Migration> {
234+ match &state.current_migration {
235+ Some(current_migration) => state
236+ .migrations
237+ .iter()
238+ .skip_while(|migration| &migration.name != current_migration)
239+ .skip(1)
240+ .collect(),
241+ None => state.migrations.iter().collect(),
242+ }
243+ }
244+
245+ pub fn remove(&mut self) -> anyhow::Result<()> {
246+ // Remove migration schemas and views
247+ if let Some(current_migration) = &self.state.current_migration {
248+ let query = format!(
249+ "DROP SCHEMA IF EXISTS {} CASCADE",
250+ schema_name_for_migration(current_migration)
251+ );
252+ self.db.run(&query)?;
253+ }
254+
255+ if let Status::InProgress {
256+ target_migration,
257+ target_schema: _,
258+ } = &self.state.status
259+ {
260+ let query = format!(
261+ "DROP SCHEMA IF EXISTS {} CASCADE",
262+ schema_name_for_migration(target_migration)
263+ );
264+ self.db.run(&query)?;
265+ }
266+
267+ // Remove all tables
268+ let schema = &self.state.current_schema;
269+ for table in schema.tables.values() {
270+ let query = format!("DROP TABLE IF EXISTS {} CASCADE", table.name);
271+ self.db.run(&query)?;
272+ }
273+
274+ // Reset state
275+ self.state.clear(&mut self.db)?;
276+
277+ println!("Reshape and all data has been removed");
278+
279+ Ok(())
280+ }
281+
282+ pub fn latest_schema(&self) -> Option<String> {
283+ self.state
284+ .migrations
285+ .last()
286+ .map(|migration| schema_name_for_migration(&migration.name))
287+ }
288+}
289+
290+pub fn generate_schema_query(migration_name: &str) -> String {
291+ let schema_name = schema_name_for_migration(migration_name);
292+ format!("SET search_path TO {}", schema_name)
293+}
294+
295+fn schema_name_for_migration(migration_name: &str) -> String {
296+ format!("migration_{}", migration_name)
297+}
src/main.rsnew file+125
@@ -0,0 +1,125 @@
1+use std::{
2+ fs::{self, DirEntry, File},
3+ io::Read,
4+ path::{Path, PathBuf},
5+};
6+
7+use clap::Parser;
8+use reshape::{
9+ migrations::{Action, Migration},
10+ Reshape,
11+};
12+use serde::{Deserialize, Serialize};
13+
14+#[derive(Parser)]
15+struct Opts {
16+ #[clap(subcommand)]
17+ cmd: Command,
18+ #[clap(default_value = "postgres://postgres:postgres@localhost:5432/postgres")]
19+ url: String,
20+}
21+
22+#[derive(Parser)]
23+enum Command {
24+ Migrate,
25+ Finish,
26+ Remove,
27+ LatestSchema,
28+}
29+
30+fn main() {
31+ let opts: Opts = Opts::parse();
32+
33+ let result = run(opts);
34+ if let Err(e) = result {
35+ println!("Error: {}", e);
36+ }
37+}
38+
39+fn run(opts: Opts) -> anyhow::Result<()> {
40+ let mut reshape = Reshape::new(&opts.url)?;
41+
42+ match opts.cmd {
43+ Command::Migrate => migrate(&mut reshape),
44+ Command::Finish => reshape.complete_migration(),
45+ Command::Remove => reshape.remove(),
46+ Command::LatestSchema => {
47+ println!(
48+ "{}",
49+ reshape.latest_schema().unwrap_or_else(|| "".to_string())
50+ );
51+ Ok(())
52+ }
53+ }
54+}
55+
56+fn migrate(reshape: &mut Reshape) -> anyhow::Result<()> {
57+ let migrations = find_migrations()?;
58+ reshape.migrate(migrations)
59+}
60+
61+fn find_migrations() -> anyhow::Result<Vec<Migration>> {
62+ let path = Path::new("migrations");
63+
64+ // Return early if path doesn't exist
65+ if !path.exists() {
66+ return Ok(Vec::new());
67+ }
68+
69+ let mut paths: Vec<PathBuf> = fs::read_dir(path)?
70+ .collect::<std::io::Result<Vec<DirEntry>>>()?
71+ .iter()
72+ .map(|entry| entry.path())
73+ .collect();
74+
75+ // Sort all files by their file names (without extension)
76+ paths.sort_unstable_by_key(|path| path.as_path().file_stem().unwrap().to_os_string());
77+
78+ paths
79+ .iter()
80+ .map(|path| {
81+ let mut file = File::open(path)?;
82+
83+ // Read file data
84+ let mut data = String::new();
85+ file.read_to_string(&mut data)?;
86+
87+ Ok((path, data))
88+ })
89+ .map(|result| {
90+ result.and_then(|(path, data)| {
91+ let extension = path.extension().and_then(|ext| ext.to_str()).unwrap();
92+ let file_migration = decode_migration_file(&data, extension)?;
93+
94+ let file_name = path.file_stem().and_then(|name| name.to_str()).unwrap();
95+ Ok(Migration {
96+ name: file_migration.name.unwrap_or_else(|| file_name.to_string()),
97+ description: file_migration.description,
98+ actions: file_migration.actions,
99+ })
100+ })
101+ })
102+ .collect()
103+}
104+
105+fn decode_migration_file(data: &str, extension: &str) -> anyhow::Result<FileMigration> {
106+ let migration: FileMigration = match extension {
107+ "json" => serde_json::from_str(data)?,
108+ "toml" => toml::from_str(data)?,
109+ extension => {
110+ return Err(anyhow::anyhow!(
111+ "unrecognized file extension '{}'",
112+ extension
113+ ))
114+ }
115+ };
116+
117+ Ok(migration)
118+}
119+
120+#[derive(Serialize, Deserialize)]
121+struct FileMigration {
122+ name: Option<String>,
123+ description: Option<String>,
124+ actions: Vec<Box<dyn Action>>,
125+}
src/migrations/add_column.rsnew file+213
@@ -0,0 +1,213 @@
1+use super::{Action, Column};
2+use crate::{
3+ db::Conn,
4+ schema::{self, Schema},
5+};
6+use serde::{Deserialize, Serialize};
7+
8+#[derive(Serialize, Deserialize, Debug)]
9+pub struct AddColumn {
10+ pub table: String,
11+ pub up: Option<String>,
12+ pub column: Column,
13+}
14+
15+impl AddColumn {
16+ fn trigger_name(&self) -> String {
17+ format!("add_column_{}_{}", self.table, self.column.name)
18+ }
19+
20+ fn not_null_constraint_name(&self) -> String {
21+ format!("add_column_not_null_{}_{}", self.table, self.column.name)
22+ }
23+}
24+
25+#[typetag::serde(name = "add_column")]
26+impl Action for AddColumn {
27+ fn describe(&self) -> String {
28+ format!(
29+ "Adding column \"{}\" to \"{}\"",
30+ self.column.name, self.table
31+ )
32+ }
33+
34+ fn run(&self, db: &mut dyn Conn, schema: &Schema) -> anyhow::Result<()> {
35+ let table = schema.find_table(&self.table)?;
36+
37+ let mut definition_parts = vec![
38+ self.column.name.to_string(),
39+ self.column.data_type.to_string(),
40+ ];
41+ if let Some(default) = &self.column.default {
42+ definition_parts.push("DEFAULT".to_string());
43+ definition_parts.push(default.to_string());
44+ }
45+
46+ // Add temporary column
47+ let query = format!(
48+ "
49+ ALTER TABLE {table}
50+ ADD COLUMN IF NOT EXISTS {definition};
51+ ",
52+ table = self.table,
53+ definition = definition_parts.join(" "),
54+ );
55+ db.run(&query)?;
56+
57+ if let Some(up) = &self.up {
58+ let query = format!(
59+ "
60+ ALTER TABLE {table}
61+ ADD COLUMN IF NOT EXISTS __reshape_is_new BOOLEAN DEFAULT FALSE NOT NULL;
62+ ",
63+ table = self.table,
64+ );
65+ db.run(&query)?;
66+
67+ let declarations: Vec<String> = table
68+ .columns
69+ .iter()
70+ .map(|column| {
71+ format!(
72+ "{name} public.{table}.{name}%TYPE := NEW.{name};",
73+ table = table.name,
74+ name = column.name,
75+ )
76+ })
77+ .collect();
78+
79+ // Add triggers to fill in values as they are inserted/updated
80+ let query = format!(
81+ "
82+ CREATE OR REPLACE FUNCTION {trigger_name}()
83+ RETURNS TRIGGER AS $$
84+ BEGIN
85+ IF NOT NEW.__reshape_is_new THEN
86+ DECLARE
87+ {declarations}
88+ BEGIN
89+ NEW.{column_name} = {up};
90+ END;
91+ END IF;
92+ RETURN NEW;
93+ END
94+ $$ language 'plpgsql';
95+
96+ DROP TRIGGER IF EXISTS {trigger_name} ON {table};
97+ CREATE TRIGGER {trigger_name} BEFORE UPDATE OR INSERT ON {table} FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();
98+ ",
99+ column_name = self.column.name,
100+ trigger_name = self.trigger_name(),
101+ up = up,
102+ table = self.table,
103+ declarations = declarations.join("\n"),
104+ );
105+ db.run(&query)?;
106+ }
107+
108+ // Add a temporary NOT NULL constraint if the column shouldn't be nullable.
109+ // This constraint is set as NOT VALID so it doesn't apply to existing rows and
110+ // the existing rows don't need to be scanned under an exclusive lock.
111+ // Thanks to this, we can set the full column as NOT NULL later with minimal locking.
112+ if !self.column.nullable {
113+ let query = format!(
114+ "
115+ ALTER TABLE {table}
116+ ADD CONSTRAINT {constraint_name}
117+ CHECK ({column} IS NOT NULL) NOT VALID
118+ ",
119+ table = self.table,
120+ constraint_name = self.not_null_constraint_name(),
121+ column = self.column.name,
122+ );
123+ db.run(&query)?;
124+ }
125+
126+ // Fill in values
127+ if let Some(up) = &self.up {
128+ let query = format!(
129+ "
130+ UPDATE {table} SET {column_name} = {up}
131+ ",
132+ table = self.table,
133+ column_name = self.column.name,
134+ up = up,
135+ );
136+ db.run(&query)?;
137+ }
138+
139+ Ok(())
140+ }
141+
142+ fn complete(&self, db: &mut dyn Conn, _schema: &Schema) -> anyhow::Result<()> {
143+ // Remove triggers, procedures and temporary column
144+ let query = format!(
145+ "
146+ DROP TRIGGER IF EXISTS {trigger_name} ON {table};
147+ DROP FUNCTION IF EXISTS {trigger_name};
148+ ",
149+ table = self.table,
150+ trigger_name = self.trigger_name(),
151+ );
152+ db.run(&query)?;
153+
154+ // Update column to be NOT NULL if necessary
155+ if !self.column.nullable {
156+ // Validate the temporary constraint (should always be valid).
157+ // This performs a sequential scan but does not take an exclusive lock.
158+ let query = format!(
159+ "
160+ ALTER TABLE {table}
161+ VALIDATE CONSTRAINT {constraint_name}
162+ ",
163+ table = self.table,
164+ constraint_name = self.not_null_constraint_name(),
165+ );
166+ db.run(&query)?;
167+
168+ // Update the column to be NOT NULL.
169+ // This requires an exclusive lock but since PG 12 it can check
170+ // the existing constraint for correctness which makes the lock short-lived.
171+ // Source: https://dba.stackexchange.com/a/268128
172+ let query = format!(
173+ "
174+ ALTER TABLE {table}
175+ ALTER COLUMN {column} SET NOT NULL
176+ ",
177+ table = self.table,
178+ column = self.column.name,
179+ );
180+ db.run(&query)?;
181+
182+ // Drop the temporary constraint
183+ let query = format!(
184+ "
185+ ALTER TABLE {table}
186+ DROP CONSTRAINT {constraint_name}
187+ ",
188+ table = self.table,
189+ constraint_name = self.not_null_constraint_name(),
190+ );
191+ db.run(&query)?;
192+ }
193+
194+ Ok(())
195+ }
196+
197+ fn update_schema(&self, schema: &mut Schema) -> anyhow::Result<()> {
198+ let table = schema.find_table_mut(&self.table)?;
199+
200+ if self.up.is_some() {
201+ table.has_is_new = true;
202+ }
203+
204+ table.add_column(schema::Column {
205+ name: self.column.name.to_string(),
206+ real_name: None,
207+ data_type: self.column.data_type.to_string(),
208+ nullable: self.column.nullable,
209+ });
210+
211+ Ok(())
212+ }
213+}
src/migrations/alter_column.rsnew file+323
@@ -0,0 +1,323 @@
1+use super::Action;
2+use crate::{db::Conn, schema::Schema};
3+use anyhow::anyhow;
4+use serde::{Deserialize, Serialize};
5+
6+#[derive(Serialize, Deserialize, Debug)]
7+pub struct AlterColumn {
8+ pub table: String,
9+ pub column: String,
10+ pub up: Option<String>,
11+ pub down: Option<String>,
12+ pub changes: ColumnChanges,
13+}
14+
15+#[derive(Serialize, Deserialize, Debug)]
16+pub struct ColumnChanges {
17+ pub name: Option<String>,
18+ #[serde(rename = "type")]
19+ pub data_type: Option<String>,
20+ pub nullable: Option<bool>,
21+}
22+
23+impl AlterColumn {
24+ fn insert_trigger_name(&self) -> String {
25+ format!("alter_column_insert_trigger_{}_{}", self.table, self.column)
26+ }
27+
28+ fn update_old_trigger_name(&self) -> String {
29+ format!(
30+ "alter_column_update_old_trigger_{}_{}",
31+ self.table, self.column
32+ )
33+ }
34+
35+ fn update_new_trigger_name(&self) -> String {
36+ format!(
37+ "alter_column_update_new_trigger_{}_{}",
38+ self.table, self.column
39+ )
40+ }
41+
42+ fn not_null_constraint_name(&self) -> String {
43+ format!(
44+ "alter_column_temporary_not_null_{}_{}",
45+ self.table, self.column
46+ )
47+ }
48+
49+ fn can_short_circuit(&self) -> bool {
50+ self.changes.name.is_some()
51+ && self.changes.data_type.is_none()
52+ && self.changes.nullable.is_none()
53+ }
54+}
55+
56+#[typetag::serde(name = "uppercase_column")]
57+impl Action for AlterColumn {
58+ fn describe(&self) -> String {
59+ format!("Altering column \"{}\" on \"{}\"", self.column, self.table)
60+ }
61+
62+ fn run(&self, db: &mut dyn Conn, schema: &Schema) -> anyhow::Result<()> {
63+ let column = schema
64+ .find_table(&self.table)
65+ .and_then(|table| table.find_column(&self.column))?;
66+
67+ // If we are only changing the name of a column, we can do that without creating a temporary column
68+ if self.can_short_circuit() {
69+ if let Some(new_name) = &self.changes.name {
70+ let query = format!(
71+ "
72+ ALTER TABLE {table}
73+ RENAME COLUMN {existing_name} TO {new_name}
74+ ",
75+ table = self.table,
76+ existing_name = column.real_name(),
77+ new_name = new_name,
78+ );
79+ db.run(&query)?;
80+ }
81+
82+ return Ok(());
83+ }
84+
85+ // If we couldn't short circuit, then we need up and down functions
86+ let (up, down) = match (&self.up, &self.down) {
87+ (Some(up), Some(down)) => (up, down),
88+ _ => return Err(anyhow!("missing up or down values")),
89+ };
90+
91+ let temporary_column = format!("__new__{}", column.real_name());
92+ let temporary_column_type = self.changes.data_type.as_ref().unwrap_or(&column.data_type);
93+
94+ // Add temporary, nullable column
95+ let query = format!(
96+ "
97+ ALTER TABLE {table}
98+ ADD COLUMN IF NOT EXISTS {temp_column} {temp_column_type};
99+
100+ ALTER TABLE {table}
101+ ADD COLUMN IF NOT EXISTS __reshape_is_new BOOLEAN DEFAULT FALSE NOT NULL;
102+ ",
103+ table = self.table,
104+ temp_column = temporary_column,
105+ temp_column_type = temporary_column_type,
106+ );
107+ db.run(&query)?;
108+
109+ // Add triggers to fill in values as they are inserted/updated
110+ let query = format!(
111+ "
112+ CREATE OR REPLACE FUNCTION {insert_trigger}()
113+ RETURNS TRIGGER AS $$
114+ BEGIN
115+ IF NEW.__reshape_is_new THEN
116+ DECLARE
117+ {existing_column} public.{table}.{temp_column}%TYPE := NEW.{temp_column};
118+ BEGIN
119+ {existing_column} = NEW.{temp_column};
120+ NEW.{existing_column} = {down};
121+ END;
122+ ELSIF NOT NEW.__reshape_is_new THEN
123+ DECLARE
124+ {existing_column} public.{table}.{existing_column}%TYPE := NEW.{existing_column};
125+ BEGIN
126+ {existing_column} = NEW.{existing_column};
127+ NEW.{temp_column} = {up};
128+ END;
129+ END IF;
130+ RETURN NEW;
131+ END
132+ $$ language 'plpgsql';
133+
134+ DROP TRIGGER IF EXISTS {insert_trigger} ON {table};
135+ CREATE TRIGGER {insert_trigger} BEFORE INSERT ON {table} FOR EACH ROW EXECUTE PROCEDURE {insert_trigger}();
136+
137+
138+ CREATE OR REPLACE FUNCTION {update_old_trigger}()
139+ RETURNS TRIGGER AS $$
140+ BEGIN
141+ NEW.{temp_column} = UPPER(NEW.{existing_column});
142+ RETURN NEW;
143+ END
144+ $$ language 'plpgsql';
145+
146+ DROP TRIGGER IF EXISTS {update_old_trigger} ON {table};
147+ CREATE TRIGGER {update_old_trigger} BEFORE UPDATE OF {existing_column} ON {table} FOR EACH ROW EXECUTE PROCEDURE {update_old_trigger}();
148+
149+
150+ CREATE OR REPLACE FUNCTION {update_new_trigger}()
151+ RETURNS TRIGGER AS $$
152+ BEGIN
153+ NEW.{existing_column} = LOWER(NEW.{temp_column});
154+ RETURN NEW;
155+ END
156+ $$ language 'plpgsql';
157+
158+ DROP TRIGGER IF EXISTS {update_new_trigger} ON {table};
159+ CREATE TRIGGER {update_new_trigger} BEFORE UPDATE OF {temp_column} ON {table} FOR EACH ROW EXECUTE PROCEDURE {update_new_trigger}();
160+ ",
161+ existing_column = column.real_name(),
162+ temp_column = temporary_column,
163+ up = up,
164+ down = down,
165+ table = self.table,
166+ insert_trigger = self.insert_trigger_name(),
167+ update_old_trigger = self.update_old_trigger_name(),
168+ update_new_trigger = self.update_new_trigger_name(),
169+ );
170+ db.run(&query)?;
171+
172+ // Add a temporary NOT NULL constraint if the column shouldn't be nullable.
173+ // This constraint is set as NOT VALID so it doesn't apply to existing rows and
174+ // the existing rows don't need to be scanned under an exclusive lock.
175+ // Thanks to this, we can set the full column as NOT NULL later with minimal locking.
176+ if !column.nullable {
177+ let query = format!(
178+ "
179+ ALTER TABLE {table}
180+ ADD CONSTRAINT {constraint_name}
181+ CHECK ({column} IS NOT NULL) NOT VALID
182+ ",
183+ table = self.table,
184+ constraint_name = self.not_null_constraint_name(),
185+ column = temporary_column,
186+ );
187+ db.run(&query)?;
188+ }
189+
190+ // Fill in values
191+ let query = format!(
192+ "
193+ UPDATE {table} SET {temp_column} = {up}
194+ ",
195+ table = self.table,
196+ temp_column = temporary_column,
197+ up = up,
198+ );
199+ db.run(&query)?;
200+
201+ Ok(())
202+ }
203+
204+ fn complete(&self, db: &mut dyn Conn, schema: &Schema) -> anyhow::Result<()> {
205+ if self.can_short_circuit() {
206+ return Ok(());
207+ }
208+
209+ let column = schema
210+ .find_table(&self.table)
211+ .and_then(|table| table.find_column(&self.column))?;
212+
213+ let column_name = self.changes.name.as_deref().unwrap_or(column.real_name());
214+
215+ // Remove old column
216+ let query = format!(
217+ "
218+ ALTER TABLE {} DROP COLUMN {} CASCADE
219+ ",
220+ self.table,
221+ column.real_name()
222+ );
223+ db.run(&query)?;
224+
225+ // Rename temporary column
226+ let query = format!(
227+ "
228+ ALTER TABLE {table} RENAME COLUMN __new__{real_name} TO {name}
229+ ",
230+ table = self.table,
231+ real_name = column.real_name(),
232+ name = column_name,
233+ );
234+ db.run(&query)?;
235+
236+ // Remove triggers and procedures
237+ let query = format!(
238+ "
239+ DROP TRIGGER IF EXISTS {insert_trigger} ON {table};
240+ DROP FUNCTION IF EXISTS {insert_trigger};
241+
242+ DROP TRIGGER IF EXISTS {update_old_trigger} ON {table};
243+ DROP FUNCTION IF EXISTS {update_old_trigger};
244+
245+ DROP TRIGGER IF EXISTS {update_new_trigger} ON {table};
246+ DROP FUNCTION IF EXISTS {update_new_trigger};
247+ ",
248+ table = self.table,
249+ insert_trigger = self.insert_trigger_name(),
250+ update_old_trigger = self.update_old_trigger_name(),
251+ update_new_trigger = self.update_new_trigger_name(),
252+ );
253+ db.run(&query)?;
254+
255+ // Update column to be NOT NULL if necessary
256+ if !column.nullable {
257+ // Validate the temporary constraint (should always be valid).
258+ // This performs a sequential scan but does not take an exclusive lock.
259+ let query = format!(
260+ "
261+ ALTER TABLE {table}
262+ VALIDATE CONSTRAINT {constraint_name}
263+ ",
264+ table = self.table,
265+ constraint_name = self.not_null_constraint_name(),
266+ );
267+ db.run(&query)?;
268+
269+ // Update the column to be NOT NULL.
270+ // This requires an exclusive lock but since PG 12 it can check
271+ // the existing constraint for correctness which makes the lock short-lived.
272+ // Source: https://dba.stackexchange.com/a/268128
273+ let query = format!(
274+ "
275+ ALTER TABLE {table}
276+ ALTER COLUMN {column} SET NOT NULL
277+ ",
278+ table = self.table,
279+ column = column_name,
280+ );
281+ db.run(&query)?;
282+
283+ // Drop the temporary constraint
284+ let query = format!(
285+ "
286+ ALTER TABLE {table}
287+ DROP CONSTRAINT {constraint_name}
288+ ",
289+ table = self.table,
290+ constraint_name = self.not_null_constraint_name(),
291+ );
292+ db.run(&query)?;
293+ }
294+
295+ Ok(())
296+ }
297+
298+ fn update_schema(&self, schema: &mut Schema) -> anyhow::Result<()> {
299+ let table = schema.find_table_mut(&self.table)?;
300+ let column = table.find_column_mut(&self.column)?;
301+
302+ // If we are only changing the name of a column, we haven't created a temporary column
303+ if self.can_short_circuit() {
304+ if let Some(new_name) = &self.changes.name {
305+ column.real_name = None;
306+ column.name = new_name.to_string();
307+ }
308+
309+ return Ok(());
310+ }
311+
312+ column.name = self
313+ .changes
314+ .name
315+ .as_ref()
316+ .map(|n| n.to_string())
317+ .unwrap_or(self.column.to_string());
318+ column.real_name = Some(format!("__new__{}", self.column));
319+ table.has_is_new = true;
320+
321+ Ok(())
322+ }
323+}
src/migrations/common.rsnew file+18
@@ -0,0 +1,18 @@
1+use serde::{Deserialize, Serialize};
2+
3+#[derive(Serialize, Deserialize, Debug)]
4+pub struct Column {
5+ pub name: String,
6+
7+ #[serde(rename = "type")]
8+ pub data_type: String,
9+
10+ #[serde(default = "nullable_default")]
11+ pub nullable: bool,
12+
13+ pub default: Option<String>,
14+}
15+
16+fn nullable_default() -> bool {
17+ true
18+}
src/migrations/create_table.rsnew file+70
@@ -0,0 +1,70 @@
1+use super::{Action, Column};
2+use crate::{
3+ db::Conn,
4+ schema::{Schema, Table},
5+};
6+use serde::{Deserialize, Serialize};
7+
8+#[derive(Serialize, Deserialize, Debug)]
9+pub struct CreateTable {
10+ pub name: String,
11+ pub columns: Vec<Column>,
12+}
13+
14+#[typetag::serde(name = "create_table")]
15+impl Action for CreateTable {
16+ fn describe(&self) -> String {
17+ format!("Creating table \"{}\"", self.name)
18+ }
19+
20+ fn run(&self, db: &mut dyn Conn, _schema: &Schema) -> anyhow::Result<()> {
21+ let column_definitions: Vec<String> = self
22+ .columns
23+ .iter()
24+ .map(|column| {
25+ let mut parts = vec![column.name.to_string(), column.data_type.to_string()];
26+
27+ if let Some(default) = &column.default {
28+ parts.push("DEFAULT".to_string());
29+ parts.push(default.to_string());
30+ }
31+
32+ if !column.nullable {
33+ parts.push("NOT NULL".to_string());
34+ }
35+
36+ parts.join(" ")
37+ })
38+ .collect();
39+
40+ let query = format!(
41+ "CREATE TABLE {} (
42+ {}
43+ )",
44+ self.name,
45+ column_definitions.join(",\n"),
46+ );
47+ db.run(&query)?;
48+ Ok(())
49+ }
50+
51+ fn complete(&self, _db: &mut dyn Conn, _schema: &Schema) -> anyhow::Result<()> {
52+ // Do nothing
53+ Ok(())
54+ }
55+
56+ fn update_schema(&self, schema: &mut Schema) -> anyhow::Result<()> {
57+ let mut table = Table::new(self.name.to_string());
58+ for column in &self.columns {
59+ table.add_column(crate::schema::Column {
60+ name: column.name.to_string(),
61+ real_name: None,
62+ data_type: column.data_type.to_string(),
63+ nullable: column.nullable,
64+ });
65+ }
66+ schema.add_table(table);
67+
68+ Ok(())
69+ }
70+}
src/migrations/mod.rsnew file+64
@@ -0,0 +1,64 @@
1+use crate::{db::Conn, schema::Schema};
2+use core::fmt::Debug;
3+use serde::{Deserialize, Serialize};
4+
5+// Re-export migration types
6+mod common;
7+pub use common::Column;
8+
9+mod create_table;
10+pub use create_table::CreateTable;
11+
12+mod alter_column;
13+pub use alter_column::{AlterColumn, ColumnChanges};
14+
15+mod add_column;
16+pub use add_column::AddColumn;
17+
18+mod remove_column;
19+pub use remove_column::RemoveColumn;
20+
21+#[derive(Serialize, Deserialize, Debug)]
22+pub struct Migration {
23+ pub name: String,
24+ pub description: Option<String>,
25+ pub actions: Vec<Box<dyn Action>>,
26+}
27+
28+impl Migration {
29+ pub fn new(name: impl Into<String>, description: Option<String>) -> Migration {
30+ Migration {
31+ name: name.into(),
32+ description,
33+ actions: vec![],
34+ }
35+ }
36+
37+ pub fn with_action(mut self, action: impl Action + 'static) -> Self {
38+ self.actions.push(Box::new(action));
39+ self
40+ }
41+}
42+
43+impl PartialEq for Migration {
44+ fn eq(&self, other: &Self) -> bool {
45+ self.name == other.name
46+ }
47+}
48+
49+impl Eq for Migration {}
50+
51+impl Clone for Migration {
52+ fn clone(&self) -> Self {
53+ let serialized = serde_json::to_string(self).unwrap();
54+ serde_json::from_str(&serialized).unwrap()
55+ }
56+}
57+
58+#[typetag::serde(tag = "type")]
59+pub trait Action: Debug {
60+ fn describe(&self) -> String;
61+ fn run(&self, db: &mut dyn Conn, schema: &Schema) -> anyhow::Result<()>;
62+ fn complete(&self, db: &mut dyn Conn, schema: &Schema) -> anyhow::Result<()>;
63+ fn update_schema(&self, schema: &mut Schema) -> anyhow::Result<()>;
64+}
src/migrations/remove_column.rsnew file+100
@@ -0,0 +1,100 @@
1+use super::Action;
2+use crate::{db::Conn, schema::Schema};
3+use serde::{Deserialize, Serialize};
4+
5+#[derive(Serialize, Deserialize, Debug)]
6+pub struct RemoveColumn {
7+ pub table: String,
8+ pub column: String,
9+ pub down: Option<String>,
10+}
11+
12+impl RemoveColumn {
13+ fn trigger_name(&self) -> String {
14+ format!("remove_column_{}_{}", self.table, self.column)
15+ }
16+}
17+
18+#[typetag::serde(name = "remove_column")]
19+impl Action for RemoveColumn {
20+ fn describe(&self) -> String {
21+ format!(
22+ "Removing column \"{}\" from \"{}\"",
23+ self.column, self.table
24+ )
25+ }
26+
27+ fn run(&self, db: &mut dyn Conn, schema: &Schema) -> anyhow::Result<()> {
28+ // Add down trigger
29+ if let Some(down) = &self.down {
30+ let table = schema.find_table(&self.table)?;
31+
32+ let declarations: Vec<String> = table
33+ .columns
34+ .iter()
35+ .map(|column| {
36+ format!(
37+ "{name} public.{table}.{name}%TYPE := NEW.{name};",
38+ table = table.name,
39+ name = column.name,
40+ )
41+ })
42+ .collect();
43+
44+ let query = format!(
45+ "
46+ CREATE OR REPLACE FUNCTION {trigger_name}()
47+ RETURNS TRIGGER AS $$
48+ BEGIN
49+ IF NEW.{column_name} IS NULL THEN
50+ DECLARE
51+ {declarations}
52+ BEGIN
53+ NEW.{column_name} = {down};
54+ END;
55+ END IF;
56+ RETURN NEW;
57+ END
58+ $$ language 'plpgsql';
59+
60+ DROP TRIGGER IF EXISTS {trigger_name} ON {table};
61+ CREATE TRIGGER {trigger_name} BEFORE UPDATE OR INSERT ON {table} FOR EACH ROW EXECUTE PROCEDURE {trigger_name}();
62+ ",
63+ column_name = self.column,
64+ trigger_name = self.trigger_name(),
65+ down = down,
66+ table = self.table,
67+ declarations = declarations.join("\n"),
68+ );
69+ db.run(&query)?;
70+ }
71+
72+ Ok(())
73+ }
74+
75+ fn complete(&self, db: &mut dyn Conn, _schema: &Schema) -> anyhow::Result<()> {
76+ // Remove column, function and trigger
77+ let query = format!(
78+ "
79+ ALTER TABLE {table}
80+ DROP COLUMN {column};
81+
82+ DROP TRIGGER IF EXISTS {trigger_name} ON {table};
83+ DROP FUNCTION IF EXISTS {trigger_name};
84+ ",
85+ table = self.table,
86+ column = self.column,
87+ trigger_name = self.trigger_name(),
88+ );
89+ db.run(&query)?;
90+
91+ Ok(())
92+ }
93+
94+ fn update_schema(&self, schema: &mut Schema) -> anyhow::Result<()> {
95+ let table = schema.find_table_mut(&self.table)?;
96+ table.remove_column(&self.column);
97+
98+ Ok(())
99+ }
100+}
src/schema.rsnew file+112
@@ -0,0 +1,112 @@
1+use anyhow::anyhow;
2+use serde::{Deserialize, Serialize};
3+use std::collections::HashMap;
4+
5+#[derive(Serialize, Deserialize, Clone, Debug)]
6+pub struct Schema {
7+ pub tables: HashMap<String, Table>,
8+}
9+
10+impl Schema {
11+ pub fn new() -> Schema {
12+ Schema {
13+ tables: HashMap::new(),
14+ }
15+ }
16+
17+ pub fn add_table(&mut self, table: Table) -> &mut Self {
18+ self.tables.insert(table.name.to_string(), table);
19+ self
20+ }
21+
22+ pub fn find_table(&self, name: &str) -> anyhow::Result<&Table> {
23+ self.tables
24+ .get(name)
25+ .ok_or_else(|| anyhow!("no table {}", name))
26+ }
27+
28+ pub fn find_table_mut(&mut self, name: &str) -> anyhow::Result<&mut Table> {
29+ self.tables
30+ .get_mut(name)
31+ .ok_or_else(|| anyhow!("no table {}", name))
32+ }
33+}
34+
35+impl Default for Schema {
36+ fn default() -> Self {
37+ Self::new()
38+ }
39+}
40+
41+#[derive(Serialize, Deserialize, Clone, Debug)]
42+pub struct Table {
43+ pub name: String,
44+ pub columns: Vec<Column>,
45+ #[serde(skip)]
46+ pub has_is_new: bool,
47+}
48+
49+impl Table {
50+ pub fn new(name: impl Into<String>) -> Table {
51+ Table {
52+ name: name.into(),
53+ columns: vec![],
54+ has_is_new: false,
55+ }
56+ }
57+
58+ pub fn add_column(&mut self, column: Column) -> &mut Self {
59+ self.columns.push(column);
60+ self
61+ }
62+
63+ pub fn remove_column(&mut self, column_name: &str) -> &mut Self {
64+ if let Some(index) = self.columns.iter().position(|c| c.name == column_name) {
65+ self.columns.remove(index);
66+ }
67+ self
68+ }
69+
70+ pub fn find_column(&self, name: &str) -> anyhow::Result<&Column> {
71+ self.columns
72+ .iter()
73+ .find(|column| column.name == name)
74+ .ok_or_else(|| anyhow!("no column {} on table {}", name, self.name))
75+ }
76+
77+ pub fn find_column_mut(&mut self, name: &str) -> anyhow::Result<&mut Column> {
78+ self.columns
79+ .iter_mut()
80+ .find(|column| column.name == name)
81+ .ok_or(anyhow!("no column {} on table {}", name, self.name))
82+ }
83+}
84+
85+#[derive(Serialize, Deserialize, Clone, Debug)]
86+pub struct Column {
87+ pub name: String,
88+ #[serde(skip)]
89+ pub real_name: Option<String>,
90+ pub data_type: String,
91+ pub nullable: bool,
92+}
93+
94+impl Column {
95+ pub fn real_name(&self) -> &str {
96+ self.real_name.as_ref().unwrap_or(&self.name)
97+ }
98+
99+ //pub fn new(
100+ // name: impl Into<String>,
101+ // real_name: impl Into<String>,
102+ // datatype: impl Into<String>,
103+ // nullable: bool,
104+ //) -> Column {
105+ // Column {
106+ // name: name.into(),
107+ // real_name: real_name.into(),
108+ // datatype: datatype.into(),
109+ // nullable,
110+ // }
111+ //}
112+}
src/state.rsnew file+111
@@ -0,0 +1,111 @@
1+use crate::schema::Schema;
2+use crate::{db::Conn, migrations::Migration};
3+use anyhow::anyhow;
4+
5+use serde::{Deserialize, Serialize};
6+use version::version;
7+
8+#[derive(Serialize, Deserialize, Debug)]
9+pub struct State {
10+ pub version: String,
11+ pub status: Status,
12+ pub current_schema: Schema,
13+ pub current_migration: Option<String>,
14+ pub migrations: Vec<Migration>,
15+}
16+
17+#[derive(Serialize, Deserialize, Debug)]
18+#[serde(tag = "status")]
19+pub enum Status {
20+ #[serde(rename = "idle")]
21+ Idle,
22+
23+ #[serde(rename = "in_progress")]
24+ InProgress {
25+ target_migration: String,
26+ target_schema: Schema,
27+ },
28+}
29+
30+impl State {
31+ pub fn load(db: &mut impl Conn) -> State {
32+ Self::ensure_schema_and_table(db);
33+
34+ let results = db
35+ .query("SELECT value FROM reshape.data WHERE key = 'state'")
36+ .unwrap();
37+
38+ match results.first() {
39+ Some(row) => {
40+ let json: serde_json::Value = row.get(0);
41+ serde_json::from_value(json).unwrap()
42+ }
43+ None => Default::default(),
44+ }
45+ }
46+
47+ pub fn save(&self, db: &mut impl Conn) -> anyhow::Result<()> {
48+ Self::ensure_schema_and_table(db);
49+
50+ let json = serde_json::to_value(self)?;
51+ db.query_with_params(
52+ "INSERT INTO reshape.data (key, value) VALUES ('state', $1) ON CONFLICT (key) DO UPDATE SET value = $1",
53+ &[&json]
54+ )?;
55+ Ok(())
56+ }
57+
58+ pub fn clear(&mut self, db: &mut impl Conn) -> anyhow::Result<()> {
59+ db.run("DROP SCHEMA reshape CASCADE")?;
60+
61+ let default = Self::default();
62+ self.status = default.status;
63+ self.current_migration = default.current_migration;
64+ self.current_schema = default.current_schema;
65+ self.migrations = default.migrations;
66+
67+ Ok(())
68+ }
69+
70+ pub fn set_migrations<T>(&mut self, migrations: T) -> Result<(), anyhow::Error>
71+ where
72+ T: IntoIterator<Item = Migration>,
73+ {
74+ let mut new_iter = migrations.into_iter();
75+
76+ // Ensure the new migration match up with the existing ones
77+ for pair in self.migrations.iter().zip(new_iter.by_ref()) {
78+ let (existing, ref new) = pair;
79+ if existing != new {
80+ return Err(anyhow!(
81+ "existing migration {} does not match new migration {}",
82+ existing.name,
83+ new.name
84+ ));
85+ }
86+ }
87+
88+ // Add any new migrations
89+ self.migrations.extend(new_iter);
90+ Ok(())
91+ }
92+
93+ fn ensure_schema_and_table(db: &mut impl Conn) {
94+ db.run("CREATE SCHEMA IF NOT EXISTS reshape").unwrap();
95+
96+ db.run("CREATE TABLE IF NOT EXISTS reshape.data (key TEXT PRIMARY KEY, value JSONB)")
97+ .unwrap();
98+ }
99+}
100+
101+impl Default for State {
102+ fn default() -> Self {
103+ State {
104+ version: version!().to_string(),
105+ status: Status::Idle,
106+ current_migration: None,
107+ current_schema: Schema::new(),
108+ migrations: vec![],
109+ }
110+ }
111+}
tests/add_column.rsnew file+293
@@ -0,0 +1,293 @@
1+use reshape::migrations::{AddColumn, Column, CreateTable, Migration};
2+use reshape::Status;
3+
4+mod common;
5+
6+#[test]
7+fn add_column() {
8+ let (mut reshape, mut old_db, mut new_db) = common::setup();
9+
10+ let create_users_table = Migration::new("create_users_table", None).with_action(CreateTable {
11+ name: "users".to_string(),
12+ columns: vec![
13+ Column {
14+ name: "id".to_string(),
15+ data_type: "SERIAL".to_string(),
16+ nullable: true,
17+ default: None,
18+ },
19+ Column {
20+ name: "name".to_string(),
21+ data_type: "TEXT".to_string(),
22+ nullable: false,
23+ default: None,
24+ },
25+ ],
26+ });
27+ let add_first_last_name_columns = Migration::new("add_first_and_last_name_columns", None)
28+ .with_action(AddColumn {
29+ table: "users".to_string(),
30+ column: Column {
31+ name: "first".to_string(),
32+ data_type: "TEXT".to_string(),
33+ nullable: false,
34+ default: None,
35+ },
36+ up: Some("(STRING_TO_ARRAY(name, ' '))[1]".to_string()),
37+ })
38+ .with_action(AddColumn {
39+ table: "users".to_string(),
40+ column: Column {
41+ name: "last".to_string(),
42+ data_type: "TEXT".to_string(),
43+ nullable: false,
44+ default: None,
45+ },
46+ up: Some("(STRING_TO_ARRAY(name, ' '))[2]".to_string()),
47+ });
48+
49+ let first_migrations = vec![create_users_table.clone()];
50+ let second_migrations = vec![
51+ create_users_table.clone(),
52+ add_first_last_name_columns.clone(),
53+ ];
54+
55+ // Run first migration, should automatically finish
56+ reshape.migrate(first_migrations.clone()).unwrap();
57+ assert!(matches!(reshape.state.status, Status::Idle));
58+ assert_eq!(
59+ Some(&create_users_table.name),
60+ reshape.state.current_migration.as_ref()
61+ );
62+
63+ // Update search paths
64+ old_db
65+ .simple_query(&reshape::generate_schema_query(
66+ &first_migrations.last().unwrap().name,
67+ ))
68+ .unwrap();
69+ new_db
70+ .simple_query(&reshape::generate_schema_query(
71+ &first_migrations.last().unwrap().name,
72+ ))
73+ .unwrap();
74+
75+ // Insert some test users
76+ new_db
77+ .simple_query(
78+ "
79+ INSERT INTO users (id, name) VALUES
80+ (1, 'John Doe'),
81+ (2, 'Jane Doe');
82+ ",
83+ )
84+ .unwrap();
85+
86+ // Run second migration
87+ reshape.migrate(second_migrations.clone()).unwrap();
88+ new_db
89+ .simple_query(&reshape::generate_schema_query(
90+ &second_migrations.last().unwrap().name,
91+ ))
92+ .unwrap();
93+
94+ // Check that the existing users have the new columns populated
95+ let expected = vec![("John", "Doe"), ("Jane", "Doe")];
96+ assert!(new_db
97+ .query("SELECT first, last FROM users ORDER BY id", &[],)
98+ .unwrap()
99+ .iter()
100+ .map(|row| (row.get("first"), row.get("last")))
101+ .eq(expected));
102+
103+ // Insert data using old schema and make sure the new columns are populated
104+ old_db
105+ .simple_query("INSERT INTO users (id, name) VALUES (3, 'Test Testsson')")
106+ .unwrap();
107+ let (first_name, last_name): (String, String) = new_db
108+ .query_one("SELECT first, last from users WHERE id = 3", &[])
109+ .map(|row| (row.get("first"), row.get("last")))
110+ .unwrap();
111+ assert_eq!(
112+ ("Test", "Testsson"),
113+ (first_name.as_ref(), last_name.as_ref())
114+ );
115+
116+ reshape.complete_migration().unwrap();
117+}
118+
119+#[test]
120+fn add_column_nullable() {
121+ let (mut reshape, mut old_db, mut new_db) = common::setup();
122+
123+ let create_users_table = Migration::new("create_users_table", None).with_action(CreateTable {
124+ name: "users".to_string(),
125+ columns: vec![Column {
126+ name: "id".to_string(),
127+ data_type: "SERIAL".to_string(),
128+ nullable: true,
129+ default: None,
130+ }],
131+ });
132+ let add_name_column = Migration::new("add_nullable_name_column", None).with_action(AddColumn {
133+ table: "users".to_string(),
134+ column: Column {
135+ name: "name".to_string(),
136+ data_type: "TEXT".to_string(),
137+ nullable: true,
138+ default: None,
139+ },
140+ up: None,
141+ });
142+
143+ let first_migrations = vec![create_users_table.clone()];
144+ let second_migrations = vec![create_users_table.clone(), add_name_column.clone()];
145+
146+ // Run first migration, should automatically finish
147+ reshape.migrate(first_migrations.clone()).unwrap();
148+ assert!(matches!(reshape.state.status, Status::Idle));
149+ assert_eq!(
150+ Some(&create_users_table.name),
151+ reshape.state.current_migration.as_ref()
152+ );
153+
154+ // Update search paths
155+ old_db
156+ .simple_query(&reshape::generate_schema_query(
157+ &first_migrations.last().unwrap().name,
158+ ))
159+ .unwrap();
160+ new_db
161+ .simple_query(&reshape::generate_schema_query(
162+ &first_migrations.last().unwrap().name,
163+ ))
164+ .unwrap();
165+
166+ // Insert some test values
167+ new_db
168+ .simple_query(
169+ "
170+ INSERT INTO users (id) VALUES
171+ (1),
172+ (2);
173+ ",
174+ )
175+ .unwrap();
176+
177+ // Run second migration
178+ reshape.migrate(second_migrations.clone()).unwrap();
179+ new_db
180+ .simple_query(&reshape::generate_schema_query(
181+ &second_migrations.last().unwrap().name,
182+ ))
183+ .unwrap();
184+
185+ // Ensure existing data got updated
186+ let expected: Vec<Option<String>> = vec![None, None];
187+ assert!(new_db
188+ .query("SELECT name FROM users ORDER BY id", &[],)
189+ .unwrap()
190+ .iter()
191+ .map(|row| row.get::<_, Option<String>>("name"))
192+ .eq(expected));
193+
194+ // Insert data using old schema and ensure new column is NULL
195+ old_db
196+ .simple_query("INSERT INTO users (id) VALUES (3)")
197+ .unwrap();
198+ let name: Option<String> = new_db
199+ .query_one("SELECT name from users WHERE id = 3", &[])
200+ .map(|row| (row.get("name")))
201+ .unwrap();
202+ assert_eq!(None, name);
203+
204+ // Ensure data can be inserted against new schema
205+ new_db
206+ .simple_query("INSERT INTO users (id, name) VALUES (3, 'Test Testsson'), (4, NULL)")
207+ .unwrap();
208+
209+ reshape.complete_migration().unwrap();
210+}
211+
212+#[test]
213+fn add_column_with_default() {
214+ let (mut reshape, mut old_db, mut new_db) = common::setup();
215+
216+ let create_users_table = Migration::new("create_users_table", None).with_action(CreateTable {
217+ name: "users".to_string(),
218+ columns: vec![Column {
219+ name: "id".to_string(),
220+ data_type: "SERIAL".to_string(),
221+ nullable: true,
222+ default: None,
223+ }],
224+ });
225+ let add_name_column =
226+ Migration::new("add_name_column_with_default", None).with_action(AddColumn {
227+ table: "users".to_string(),
228+ column: Column {
229+ name: "name".to_string(),
230+ data_type: "TEXT".to_string(),
231+ nullable: false,
232+ default: Some("'DEFAULT'".to_string()),
233+ },
234+ up: None,
235+ });
236+
237+ let first_migrations = vec![create_users_table.clone()];
238+ let second_migrations = vec![create_users_table.clone(), add_name_column.clone()];
239+
240+ // Run first migration, should automatically finish
241+ reshape.migrate(first_migrations.clone()).unwrap();
242+ assert!(matches!(reshape.state.status, Status::Idle));
243+ assert_eq!(
244+ Some(&create_users_table.name),
245+ reshape.state.current_migration.as_ref()
246+ );
247+
248+ // Update search paths
249+ old_db
250+ .simple_query(&reshape::generate_schema_query(
251+ &first_migrations.last().unwrap().name,
252+ ))
253+ .unwrap();
254+ new_db
255+ .simple_query(&reshape::generate_schema_query(
256+ &first_migrations.last().unwrap().name,
257+ ))
258+ .unwrap();
259+
260+ // Insert some test values
261+ new_db
262+ .simple_query("INSERT INTO users (id) VALUES (1), (2)")
263+ .unwrap();
264+
265+ // Run second migration
266+ reshape.migrate(second_migrations.clone()).unwrap();
267+ new_db
268+ .simple_query(&reshape::generate_schema_query(
269+ &second_migrations.last().unwrap().name,
270+ ))
271+ .unwrap();
272+
273+ // Ensure existing data got updated with defaults
274+ let expected = vec!["DEFAULT".to_string(), "DEFAULT".to_string()];
275+ assert!(new_db
276+ .query("SELECT name FROM users ORDER BY id", &[],)
277+ .unwrap()
278+ .iter()
279+ .map(|row| row.get::<_, String>("name"))
280+ .eq(expected));
281+
282+ // Insert data using old schema and ensure new column gets the default value
283+ old_db
284+ .simple_query("INSERT INTO users (id) VALUES (3)")
285+ .unwrap();
286+ let name: String = new_db
287+ .query_one("SELECT name from users WHERE id = 3", &[])
288+ .map(|row| row.get("name"))
289+ .unwrap();
290+ assert_eq!("DEFAULT", name);
291+
292+ reshape.complete_migration().unwrap();
293+}
tests/alter_column.rsnew file+287
@@ -0,0 +1,287 @@
1+use reshape::migrations::{AlterColumn, Column, ColumnChanges, CreateTable, Migration};
2+use reshape::Status;
3+
4+mod common;
5+
6+#[test]
7+fn alter_column_data() {
8+ let (mut reshape, mut old_db, mut new_db) = common::setup();
9+
10+ let create_users_table = Migration::new("create_users_table", None).with_action(CreateTable {
11+ name: "users".to_string(),
12+ columns: vec![
13+ Column {
14+ name: "id".to_string(),
15+ data_type: "SERIAL".to_string(),
16+ nullable: true,
17+ default: None,
18+ },
19+ Column {
20+ name: "name".to_string(),
21+ data_type: "TEXT".to_string(),
22+ nullable: false,
23+ default: None,
24+ },
25+ ],
26+ });
27+ let uppercase_name = Migration::new("uppercase_name", None).with_action(AlterColumn {
28+ table: "users".to_string(),
29+ column: "name".to_string(),
30+ up: Some("UPPER(name)".to_string()),
31+ down: Some("LOWER(name)".to_string()),
32+ changes: ColumnChanges {
33+ data_type: None,
34+ nullable: None,
35+ name: None,
36+ },
37+ });
38+
39+ let first_migrations = vec![create_users_table.clone()];
40+ let second_migrations = vec![create_users_table.clone(), uppercase_name.clone()];
41+
42+ // Run first migration, should automatically finish
43+ reshape.migrate(first_migrations.clone()).unwrap();
44+ assert!(matches!(reshape.state.status, Status::Idle));
45+ assert_eq!(
46+ Some(&create_users_table.name),
47+ reshape.state.current_migration.as_ref()
48+ );
49+
50+ // Update search paths
51+ old_db
52+ .simple_query(&reshape::generate_schema_query(
53+ &first_migrations.last().unwrap().name,
54+ ))
55+ .unwrap();
56+
57+ // Insert some test users
58+ old_db
59+ .simple_query(
60+ "
61+ INSERT INTO users (id, name) VALUES
62+ (1, 'john Doe'),
63+ (2, 'jane Doe');
64+ ",
65+ )
66+ .unwrap();
67+
68+ // Run second migration
69+ reshape.migrate(second_migrations.clone()).unwrap();
70+ new_db
71+ .simple_query(&reshape::generate_schema_query(
72+ &second_migrations.last().unwrap().name,
73+ ))
74+ .unwrap();
75+
76+ // Check that the existing users has the altered data
77+ let expected = vec!["JOHN DOE", "JANE DOE"];
78+ assert!(new_db
79+ .query("SELECT name FROM users ORDER BY id", &[],)
80+ .unwrap()
81+ .iter()
82+ .map(|row| row.get::<_, String>("name"))
83+ .eq(expected));
84+
85+ // Insert data using old schema and make sure the new schema gets correct values
86+ old_db
87+ .simple_query("INSERT INTO users (id, name) VALUES (3, 'test testsson')")
88+ .unwrap();
89+ let result = new_db
90+ .query_one("SELECT name from users WHERE id = 3", &[])
91+ .unwrap();
92+ assert_eq!("TEST TESTSSON", result.get::<_, &str>("name"));
93+
94+ // Insert data using new schema and make sure the old schema gets correct values
95+ new_db
96+ .simple_query("INSERT INTO users (id, name) VALUES (4, 'TEST TESTSSON')")
97+ .unwrap();
98+ let result = old_db
99+ .query_one("SELECT name from users WHERE id = 4", &[])
100+ .unwrap();
101+ assert_eq!("test testsson", result.get::<_, &str>("name"));
102+
103+ reshape.complete_migration().unwrap();
104+}
105+
106+#[test]
107+fn alter_column_set_not_null() {
108+ let (mut reshape, mut old_db, mut new_db) = common::setup();
109+
110+ let create_users_table = Migration::new("create_users_table", None).with_action(CreateTable {
111+ name: "users".to_string(),
112+ columns: vec![
113+ Column {
114+ name: "id".to_string(),
115+ data_type: "SERIAL".to_string(),
116+ nullable: true,
117+ default: None,
118+ },
119+ Column {
120+ name: "name".to_string(),
121+ data_type: "TEXT".to_string(),
122+ nullable: true,
123+ default: None,
124+ },
125+ ],
126+ });
127+ let set_name_not_null = Migration::new("set_name_not_null", None).with_action(AlterColumn {
128+ table: "users".to_string(),
129+ column: "name".to_string(),
130+ up: Some("COALESCE(name, 'TEST_DEFAULT_VALUE')".to_string()),
131+ down: Some("name".to_string()),
132+ changes: ColumnChanges {
133+ data_type: None,
134+ nullable: Some(false),
135+ name: None,
136+ },
137+ });
138+
139+ let first_migrations = vec![create_users_table.clone()];
140+ let second_migrations = vec![create_users_table.clone(), set_name_not_null.clone()];
141+
142+ // Run first migration, should automatically finish
143+ reshape.migrate(first_migrations.clone()).unwrap();
144+ assert!(matches!(reshape.state.status, Status::Idle));
145+ assert_eq!(
146+ Some(&create_users_table.name),
147+ reshape.state.current_migration.as_ref()
148+ );
149+
150+ // Update search paths
151+ old_db
152+ .simple_query(&reshape::generate_schema_query(
153+ &first_migrations.last().unwrap().name,
154+ ))
155+ .unwrap();
156+
157+ // Insert some test users
158+ old_db
159+ .simple_query(
160+ "
161+ INSERT INTO users (id, name) VALUES
162+ (1, 'John Doe'),
163+ (2, NULL);
164+ ",
165+ )
166+ .unwrap();
167+
168+ // Run second migration
169+ reshape.migrate(second_migrations.clone()).unwrap();
170+ new_db
171+ .simple_query(&reshape::generate_schema_query(
172+ &second_migrations.last().unwrap().name,
173+ ))
174+ .unwrap();
175+
176+ // Check that existing users got the correct values
177+ let expected = vec!["John Doe", "TEST_DEFAULT_VALUE"];
178+ assert!(new_db
179+ .query("SELECT name FROM users ORDER BY id", &[],)
180+ .unwrap()
181+ .iter()
182+ .map(|row| row.get::<_, String>("name"))
183+ .eq(expected));
184+
185+ // Insert data using old schema and make sure the new schema gets correct values
186+ old_db
187+ .simple_query("INSERT INTO users (id, name) VALUES (3, NULL)")
188+ .unwrap();
189+ let result = new_db
190+ .query_one("SELECT name from users WHERE id = 3", &[])
191+ .unwrap();
192+ assert_eq!("TEST_DEFAULT_VALUE", result.get::<_, &str>("name"));
193+
194+ // Insert data using new schema and make sure the old schema gets correct values
195+ new_db
196+ .simple_query("INSERT INTO users (id, name) VALUES (4, 'Jane Doe')")
197+ .unwrap();
198+ let result = old_db
199+ .query_one("SELECT name from users WHERE id = 4", &[])
200+ .unwrap();
201+ assert_eq!("Jane Doe", result.get::<_, &str>("name"));
202+
203+ reshape.complete_migration().unwrap();
204+}
205+
206+#[test]
207+fn alter_column_rename() {
208+ let (mut reshape, mut old_db, mut new_db) = common::setup();
209+
210+ let create_users_table = Migration::new("create_users_table", None).with_action(CreateTable {
211+ name: "users".to_string(),
212+ columns: vec![
213+ Column {
214+ name: "id".to_string(),
215+ data_type: "SERIAL".to_string(),
216+ nullable: true,
217+ default: None,
218+ },
219+ Column {
220+ name: "name".to_string(),
221+ data_type: "TEXT".to_string(),
222+ nullable: true,
223+ default: None,
224+ },
225+ ],
226+ });
227+ let rename_to_full_name =
228+ Migration::new("rename_to_full_name", None).with_action(AlterColumn {
229+ table: "users".to_string(),
230+ column: "name".to_string(),
231+ up: None, // up and down are not required when only renaming a column
232+ down: None,
233+ changes: ColumnChanges {
234+ data_type: None,
235+ nullable: None,
236+ name: Some("full_name".to_string()),
237+ },
238+ });
239+
240+ let first_migrations = vec![create_users_table.clone()];
241+ let second_migrations = vec![create_users_table.clone(), rename_to_full_name.clone()];
242+
243+ // Run first migration, should automatically finish
244+ reshape.migrate(first_migrations.clone()).unwrap();
245+ assert!(matches!(reshape.state.status, Status::Idle));
246+ assert_eq!(
247+ Some(&create_users_table.name),
248+ reshape.state.current_migration.as_ref()
249+ );
250+
251+ // Update search paths
252+ old_db
253+ .simple_query(&reshape::generate_schema_query(
254+ &first_migrations.last().unwrap().name,
255+ ))
256+ .unwrap();
257+
258+ // Insert some test data
259+ old_db
260+ .simple_query(
261+ "
262+ INSERT INTO users (id, name) VALUES
263+ (1, 'John Doe'),
264+ (2, 'Jane Doe');
265+ ",
266+ )
267+ .unwrap();
268+
269+ // Run second migration
270+ reshape.migrate(second_migrations.clone()).unwrap();
271+ new_db
272+ .simple_query(&reshape::generate_schema_query(
273+ &second_migrations.last().unwrap().name,
274+ ))
275+ .unwrap();
276+
277+ // Check that existing values can be queried using new column name
278+ let expected = vec!["John Doe", "Jane Doe"];
279+ assert!(new_db
280+ .query("SELECT full_name FROM users ORDER BY id", &[],)
281+ .unwrap()
282+ .iter()
283+ .map(|row| row.get::<_, String>("full_name"))
284+ .eq(expected));
285+
286+ reshape.complete_migration().unwrap();
287+}
tests/common.rsnew file+15
@@ -0,0 +1,15 @@
1+use postgres::{Client, NoTls};
2+use reshape::Reshape;
3+
4+pub fn setup() -> (Reshape, Client, Client) {
5+ let connection_string = std::env::var("POSTGRES_CONNECTION_STRING")
6+ .unwrap_or("postgres://postgres:postgres@localhost/reshape_test".to_string());
7+
8+ let old_db = Client::connect(&connection_string, NoTls).unwrap();
9+ let new_db = Client::connect(&connection_string, NoTls).unwrap();
10+
11+ let mut reshape = Reshape::new(&connection_string).unwrap();
12+ reshape.remove().unwrap();
13+
14+ (reshape, old_db, new_db)
15+}
tests/create_table.rsnew file+100
@@ -0,0 +1,100 @@
1+use reshape::{
2+ migrations::{Column, CreateTable, Migration},
3+ Status,
4+};
5+
6+mod common;
7+
8+#[test]
9+fn create_table() {
10+ let (mut reshape, mut db, _) = common::setup();
11+
12+ let create_table_migration =
13+ Migration::new("create_users_table", None).with_action(CreateTable {
14+ name: "users".to_string(),
15+ columns: vec![
16+ Column {
17+ name: "id".to_string(),
18+ data_type: "SERIAL".to_string(),
19+ nullable: true, // Will be ignored by Postgres as the column is a SERIAL
20+ default: None,
21+ },
22+ Column {
23+ name: "name".to_string(),
24+ data_type: "TEXT".to_string(),
25+ nullable: true,
26+ default: None,
27+ },
28+ Column {
29+ name: "created_at".to_string(),
30+ data_type: "TIMESTAMP".to_string(),
31+ nullable: false,
32+ default: Some("NOW()".to_string()),
33+ },
34+ ],
35+ });
36+
37+ reshape
38+ .migrate(vec![create_table_migration.clone()])
39+ .unwrap();
40+ assert!(matches!(reshape.state.status, Status::Idle));
41+ assert_eq!(
42+ Some(&create_table_migration.name),
43+ reshape.state.current_migration.as_ref()
44+ );
45+
46+ // Ensure table was created
47+ let result = db
48+ .query_opt(
49+ "
50+ SELECT table_name
51+ FROM information_schema.tables
52+ WHERE table_name = 'users' AND table_schema = 'public'",
53+ &[],
54+ )
55+ .unwrap();
56+ assert!(result.is_some());
57+
58+ // Ensure table has the right columns
59+ let result = db
60+ .query(
61+ "
62+ SELECT column_name, column_default, is_nullable, data_type
63+ FROM information_schema.columns
64+ WHERE table_name = 'users' AND table_schema = 'public'
65+ ORDER BY ordinal_position",
66+ &[],
67+ )
68+ .unwrap();
69+
70+ // id column
71+ let id_row = &result[0];
72+ assert_eq!("id", id_row.get::<_, String>("column_name"));
73+ assert!(id_row.get::<_, Option<String>>("column_default").is_some());
74+ assert_eq!("NO", id_row.get::<_, String>("is_nullable"));
75+ assert_eq!("integer", id_row.get::<_, String>("data_type"));
76+
77+ // name column
78+ let name_row = &result[1];
79+ assert_eq!("name", name_row.get::<_, String>("column_name"));
80+ assert!(name_row
81+ .get::<_, Option<String>>("column_default")
82+ .is_none());
83+ assert_eq!("YES", name_row.get::<_, String>("is_nullable"));
84+ assert_eq!("text", name_row.get::<_, String>("data_type"));
85+
86+ // created_at column
87+ let created_at_column = &result[2];
88+ assert_eq!(
89+ "created_at",
90+ created_at_column.get::<_, String>("column_name")
91+ );
92+ assert!(created_at_column
93+ .get::<_, Option<String>>("column_default")
94+ .is_some());
95+ assert_eq!("NO", created_at_column.get::<_, String>("is_nullable"));
96+ assert_eq!(
97+ "timestamp without time zone",
98+ created_at_column.get::<_, String>("data_type")
99+ );
100+}
tests/remove_column.rsnew file+76
@@ -0,0 +1,76 @@
1+use reshape::migrations::{Column, CreateTable, Migration, RemoveColumn};
2+
3+mod common;
4+
5+#[test]
6+fn remove_column() {
7+ let (mut reshape, mut old_db, mut new_db) = common::setup();
8+
9+ let create_table_migration =
10+ Migration::new("create_users_table", None).with_action(CreateTable {
11+ name: "users".to_string(),
12+ columns: vec![
13+ Column {
14+ name: "id".to_string(),
15+ data_type: "SERIAL".to_string(),
16+ nullable: true, // Will be ignored by Postgres as the column is a SERIAL
17+ default: None,
18+ },
19+ Column {
20+ name: "name".to_string(),
21+ data_type: "TEXT".to_string(),
22+ nullable: false,
23+ default: None,
24+ },
25+ ],
26+ });
27+ let remove_column_migration =
28+ Migration::new("remove_name_column", None).with_action(RemoveColumn {
29+ table: "users".to_string(),
30+ column: "name".to_string(),
31+ down: Some("'TEST_DOWN_VALUE'".to_string()),
32+ });
33+
34+ let first_migrations = vec![create_table_migration.clone()];
35+ let second_migrations = vec![
36+ create_table_migration.clone(),
37+ remove_column_migration.clone(),
38+ ];
39+
40+ // Run migrations
41+ reshape.migrate(first_migrations.clone()).unwrap();
42+ reshape.migrate(second_migrations.clone()).unwrap();
43+
44+ // Update schemas of Postgres connections
45+ let old_schema_query = reshape::generate_schema_query(&first_migrations.last().unwrap().name);
46+ let new_schema_query = reshape::generate_schema_query(&second_migrations.last().unwrap().name);
47+ old_db.simple_query(&old_schema_query).unwrap();
48+ new_db.simple_query(&new_schema_query).unwrap();
49+
50+ // Insert using old schema and ensure it can be retrieved through new schema
51+ old_db
52+ .simple_query("INSERT INTO users(name) VALUES ('John Doe')")
53+ .unwrap();
54+ let results = new_db
55+ .query("SELECT id FROM users WHERE id = 1", &[])
56+ .unwrap();
57+ assert_eq!(1, results.len());
58+ assert_eq!(1, results[0].get::<_, i32>("id"));
59+
60+ // Ensure the name column is not accesible through the new schema
61+ assert!(new_db.query("SELECT id, name FROM users", &[]).is_err());
62+
63+ // Insert using new schema and ensure the down function is correctly applied
64+ new_db
65+ .simple_query("INSERT INTO users DEFAULT VALUES")
66+ .unwrap();
67+ let result = old_db
68+ .query_opt("SELECT name FROM users WHERE id = 2", &[])
69+ .unwrap();
70+ assert_eq!(
71+ Some("TEST_DOWN_VALUE"),
72+ result.as_ref().map(|row| row.get("name"))
73+ );
74+
75+ reshape.complete_migration().unwrap();
76+}