diff --git a/benches/apply_benchmark.rs b/benches/apply_benchmark.rs index 73d380cb2..d08fc58bd 100644 --- a/benches/apply_benchmark.rs +++ b/benches/apply_benchmark.rs @@ -495,6 +495,26 @@ fn pick_random(ids: &[Id], rng: &mut StdRng) -> Option { } } +/// Build the SQL for a fresh `comments` row and record its id in `all_comments`. +fn new_comment_sql( + rng: &mut StdRng, + ctr_comment: &mut IdCounter, + all_posts: &[Id], + all_users: &[Id], + all_comments: &mut Vec, +) -> String { + let id = ctr_comment.next(rng); + let label = id.short_label(); + let pid = pick_random(all_posts, rng).unwrap(); + let uid = pick_random(all_users, rng).unwrap(); + let ts: i64 = rng.random_range(1000000..2000000); + all_comments.push(id); + format!( + "INSERT INTO comments (id, post_id, user_id, content, created_at, is_deleted) \ + VALUES ({id}, {pid}, {uid}, 'Comment {label}', {ts}, 0)" + ) +} + /// Generate a mixed batch of `op_count` operations (INSERT/UPDATE/DELETE) /// spread across all five tables. Returns the same logical changes in three /// representations: SQL text, changeset bytes, and patchset bytes. @@ -600,18 +620,13 @@ fn generate_scenario( VALUES ({id}, {uid}, 'New post {label}', 'Content {label}', {ts}, 0, 1)" ) } - 2 => { - let id = ctr_comment.next(rng); - let label = id.short_label(); - let pid = pick_random(&all_posts, rng).unwrap(); - let uid = pick_random(&all_users, rng).unwrap(); - let ts: i64 = rng.random_range(1000000..2000000); - all_comments.push(id); - format!( - "INSERT INTO comments (id, post_id, user_id, content, created_at, is_deleted) \ - VALUES ({id}, {pid}, {uid}, 'Comment {label}', {ts}, 0)" - ) - } + 2 => new_comment_sql( + rng, + &mut ctr_comment, + &all_posts, + &all_users, + &mut all_comments, + ), 3 => { let id = ctr_tag.next(rng); let label = id.short_label(); @@ -647,15 +662,12 @@ fn generate_scenario( format!("UPDATE comments SET is_deleted = 1 WHERE id = {id}") } else { // No comments yet, fall back to an INSERT. - let id = ctr_comment.next(rng); - let label = id.short_label(); - let pid = pick_random(&all_posts, rng).unwrap(); - let uid = pick_random(&all_users, rng).unwrap(); - let ts: i64 = rng.random_range(1000000..2000000); - all_comments.push(id); - format!( - "INSERT INTO comments (id, post_id, user_id, content, created_at, is_deleted) \ - VALUES ({id}, {pid}, {uid}, 'Comment {label}', {ts}, 0)" + new_comment_sql( + rng, + &mut ctr_comment, + &all_posts, + &all_users, + &mut all_comments, ) } } diff --git a/benches/builder_vs_rusqlite.rs b/benches/builder_vs_rusqlite.rs index d9d31d41f..f22728230 100644 --- a/benches/builder_vs_rusqlite.rs +++ b/benches/builder_vs_rusqlite.rs @@ -10,8 +10,8 @@ use criterion::{Criterion, criterion_group, criterion_main}; use rusqlite::Connection; use sqlite_diff_rs::{ - ChangeDelete, ChangeSet, ChangeUpdate, DiffOps, Insert, PatchDelete, PatchSet, PatchUpdate, - SimpleTable, TableSchema, Value, testing::byte_diff_report, + ChangeDelete, ChangeSet, ChangeUpdate, DiffFormat, DiffOps, DiffSetBuilder, Insert, + PatchDelete, PatchSet, PatchUpdate, SimpleTable, TableSchema, Value, testing::byte_diff_report, }; use std::hint::black_box; use std::string::String; @@ -503,100 +503,86 @@ static PATCH_COMMENT_DELETE_PK: &[Val] = &[Val::Integer(5)]; static PATCH_POST_TAG_DELETE_PK: &[Val] = &[Val::Integer(3), Val::Integer(5)]; static PATCH_USER_DELETE_PK: &[Val] = &[Val::Integer(4)]; -/// Generate changeset using rusqlite's session API -fn rusqlite_changeset_with(schema: &str, operations: &[&str]) -> Vec { +/// Run `operations` on `schema` under a rusqlite session and stream the result with `emit`. +fn rusqlite_session_with( + schema: &str, + operations: &[&str], + emit: impl FnOnce(&mut rusqlite::session::Session<'_>, &mut Vec) -> rusqlite::Result<()>, +) -> Vec { let conn = Connection::open_in_memory().unwrap(); - - // Set up session conn.execute_batch(schema).unwrap(); let mut session = rusqlite::session::Session::new(&conn).unwrap(); session.attach::<&str>(None).unwrap(); - - // Execute operations for op in operations { conn.execute(op, []).unwrap(); } - - // Generate changeset let mut bytes = Vec::new(); - session.changeset_strm(&mut bytes).unwrap(); + emit(&mut session, &mut bytes).unwrap(); bytes } +/// Generate changeset using rusqlite's session API +fn rusqlite_changeset_with(schema: &str, operations: &[&str]) -> Vec { + rusqlite_session_with(schema, operations, |session, bytes| { + session.changeset_strm(bytes) + }) +} + /// Generate patchset using rusqlite's session API fn rusqlite_patchset_with(schema: &str, operations: &[&str]) -> Vec { - let conn = Connection::open_in_memory().unwrap(); - - // Set up session - conn.execute_batch(schema).unwrap(); - let mut session = rusqlite::session::Session::new(&conn).unwrap(); - session.attach::<&str>(None).unwrap(); - - // Execute operations - for op in operations { - conn.execute(op, []).unwrap(); - } - - // Generate patchset - let mut bytes = Vec::new(); - session.patchset_strm(&mut bytes).unwrap(); - bytes + rusqlite_session_with(schema, operations, |session, bytes| { + session.patchset_strm(bytes) + }) } -/// Add insert operations for all tables to the changeset builder -fn add_inserts_to_changeset<'a>( - mut builder: ChangeSet, &'a str, &'a [u8]>, +/// Builder over the borrowed bench tables in format `F`. +type BenchBuilder<'a, F> = DiffSetBuilder, &'a str, &'a [u8]>; + +/// Add insert operations for all tables to a changeset or patchset builder +fn add_inserts<'a, F>( + mut builder: BenchBuilder<'a, F>, users: &TableSchema<&'a str>, posts: &TableSchema<&'a str>, comments: &TableSchema<&'a str>, tags: &TableSchema<&'a str>, post_tags: &TableSchema<&'a str>, -) -> ChangeSet, &'a str, &'a [u8]> { - // User inserts - for row in USER_ROWS { - let mut insert: Insert<_, &str, &[u8]> = Insert::from(users.clone()); - for (i, val) in row.iter().enumerate() { - insert = insert.set(i, *val).unwrap(); - } - builder = builder.insert(insert); - } - - // Post inserts - for row in POST_ROWS { - let mut insert: Insert<_, &str, &[u8]> = Insert::from(posts.clone()); - for (i, val) in row.iter().enumerate() { - insert = insert.set(i, *val).unwrap(); - } - builder = builder.insert(insert); - } - - // Tag inserts - for row in TAG_ROWS { - let mut insert: Insert<_, &str, &[u8]> = Insert::from(tags.clone()); - for (i, val) in row.iter().enumerate() { - insert = insert.set(i, *val).unwrap(); - } - builder = builder.insert(insert); - } - - // Post-tag inserts - for row in POST_TAG_ROWS { - let mut insert: Insert<_, &str, &[u8]> = Insert::from(post_tags.clone()); - for (i, val) in row.iter().enumerate() { - insert = insert.set(i, *val).unwrap(); +) -> BenchBuilder<'a, F> +where + F: DiffFormat<&'a str, &'a [u8]>, + BenchBuilder<'a, F>: DiffOps, &'a str, &'a [u8], Format = F>, +{ + for (table, rows) in [ + (users, USER_ROWS), + (posts, POST_ROWS), + (tags, TAG_ROWS), + (post_tags, POST_TAG_ROWS), + (comments, COMMENT_ROWS), + ] { + for row in rows { + let mut insert: Insert<_, &str, &[u8]> = Insert::from(table.clone()); + for (i, val) in row.iter().enumerate() { + insert = insert.set(i, *val).unwrap(); + } + builder = builder.insert(insert); } - builder = builder.insert(insert); } + builder +} - // Comment inserts - for row in COMMENT_ROWS { - let mut insert: Insert<_, &str, &[u8]> = Insert::from(comments.clone()); - for (i, val) in row.iter().enumerate() { - insert = insert.set(i, *val).unwrap(); +/// Apply a batch of `(old_row, new_row)` updates against one table. +fn apply_changeset_updates<'a>( + mut builder: ChangeSet, &'a str, &'a [u8]>, + table: &TableSchema<&'a str>, + updates: &[(&[Val], &[Val])], +) -> ChangeSet, &'a str, &'a [u8]> { + for (old, new) in updates { + let mut update: ChangeUpdate, &'a str, &'a [u8]> = + ChangeUpdate::from(table.clone()); + for (i, (o, n)) in old.iter().zip(new.iter()).enumerate() { + update = update.set(i, *o, *n).unwrap(); } - builder = builder.insert(insert); + builder = builder.update(update); } - builder } @@ -607,24 +593,10 @@ fn add_updates_to_changeset<'a>( posts: &TableSchema<&'a str>, ) -> ChangeSet, &'a str, &'a [u8]> { // User updates - for (old, new) in USER_UPDATES { - let mut update: ChangeUpdate, &'a str, &'a [u8]> = - ChangeUpdate::from(users.clone()); - for (i, (o, n)) in old.iter().zip(new.iter()).enumerate() { - update = update.set(i, *o, *n).unwrap(); - } - builder = builder.update(update); - } + builder = apply_changeset_updates(builder, users, USER_UPDATES); // Post updates - for (old, new) in POST_UPDATES { - let mut update: ChangeUpdate, &'a str, &'a [u8]> = - ChangeUpdate::from(posts.clone()); - for (i, (o, n)) in old.iter().zip(new.iter()).enumerate() { - update = update.set(i, *o, *n).unwrap(); - } - builder = builder.update(update); - } + builder = apply_changeset_updates(builder, posts, POST_UPDATES); builder } @@ -667,67 +639,26 @@ fn builder_changeset() -> Vec { let mut builder: ChangeSet, &str, &[u8]> = ChangeSet::new(); // Add all operations - builder = add_inserts_to_changeset(builder, &users, &posts, &comments, &tags, &post_tags); + builder = add_inserts(builder, &users, &posts, &comments, &tags, &post_tags); builder = add_updates_to_changeset(builder, &users, &posts); builder = add_deletes_to_changeset(builder, &users, &comments, &post_tags); builder.build() } -/// Add insert operations for all tables to the patchset builder -fn add_inserts_to_patchset<'a>( +/// Apply a batch of `(column, new_value)` updates against one table. +fn apply_patchset_updates<'a>( mut builder: PatchSet, &'a str, &'a [u8]>, - users: &TableSchema<&'a str>, - posts: &TableSchema<&'a str>, - comments: &TableSchema<&'a str>, - tags: &TableSchema<&'a str>, - post_tags: &TableSchema<&'a str>, + table: &TableSchema<&'a str>, + updates: &[&[(usize, Val)]], ) -> PatchSet, &'a str, &'a [u8]> { - // User inserts - for row in USER_ROWS { - let mut insert: Insert<_, &str, &[u8]> = Insert::from(users.clone()); - for (i, val) in row.iter().enumerate() { - insert = insert.set(i, *val).unwrap(); - } - builder = builder.insert(insert); - } - - // Post inserts - for row in POST_ROWS { - let mut insert: Insert<_, &str, &[u8]> = Insert::from(posts.clone()); - for (i, val) in row.iter().enumerate() { - insert = insert.set(i, *val).unwrap(); - } - builder = builder.insert(insert); - } - - // Tag inserts - for row in TAG_ROWS { - let mut insert: Insert<_, &str, &[u8]> = Insert::from(tags.clone()); - for (i, val) in row.iter().enumerate() { - insert = insert.set(i, *val).unwrap(); - } - builder = builder.insert(insert); - } - - // Post-tag inserts - for row in POST_TAG_ROWS { - let mut insert: Insert<_, &str, &[u8]> = Insert::from(post_tags.clone()); - for (i, val) in row.iter().enumerate() { - insert = insert.set(i, *val).unwrap(); - } - builder = builder.insert(insert); - } - - // Comment inserts - for row in COMMENT_ROWS { - let mut insert: Insert<_, &str, &[u8]> = Insert::from(comments.clone()); - for (i, val) in row.iter().enumerate() { - insert = insert.set(i, *val).unwrap(); + for cols in updates { + let mut update: PatchUpdate<_, &str, &[u8]> = PatchUpdate::from(table.clone()); + for (i, val) in *cols { + update = update.set(*i, *val).unwrap(); } - builder = builder.insert(insert); + builder = builder.update(update); } - builder } @@ -738,22 +669,10 @@ fn add_updates_to_patchset<'a>( posts: &TableSchema<&'a str>, ) -> PatchSet, &'a str, &'a [u8]> { // User updates - for cols in PATCH_USER_UPDATES { - let mut update: PatchUpdate<_, &str, &[u8]> = PatchUpdate::from(users.clone()); - for (i, val) in *cols { - update = update.set(*i, *val).unwrap(); - } - builder = builder.update(update); - } + builder = apply_patchset_updates(builder, users, PATCH_USER_UPDATES); // Post updates - for cols in PATCH_POST_UPDATES { - let mut update: PatchUpdate<_, &str, &[u8]> = PatchUpdate::from(posts.clone()); - for (i, val) in *cols { - update = update.set(*i, *val).unwrap(); - } - builder = builder.update(update); - } + builder = apply_patchset_updates(builder, posts, PATCH_POST_UPDATES); builder } @@ -789,7 +708,7 @@ fn builder_patchset() -> Vec { let mut builder: PatchSet, &str, &[u8]> = PatchSet::new(); // Add all operations - builder = add_inserts_to_patchset(builder, &users, &posts, &comments, &tags, &post_tags); + builder = add_inserts(builder, &users, &posts, &comments, &tags, &post_tags); builder = add_updates_to_patchset(builder, &users, &posts); builder = add_deletes_to_patchset(builder, &users, &comments, &post_tags); diff --git a/compile-bench/src/lib.rs b/compile-bench/src/lib.rs index eceeb3959..f0654b2d5 100644 --- a/compile-bench/src/lib.rs +++ b/compile-bench/src/lib.rs @@ -135,8 +135,8 @@ CREATE TABLE post_tags ( #[cfg(feature = "builder")] pub mod builder_approach { use sqlite_diff_rs::{ - ChangeDelete, ChangeSet, ChangeUpdate, DiffOps, Insert, PatchDelete, PatchSet, PatchUpdate, - TableSchema, Value, + ChangeDelete, ChangeSet, ChangeUpdate, DiffFormat, DiffOps, DiffSetBuilder, Insert, + PatchDelete, PatchSet, PatchUpdate, TableSchema, Value, }; // Type aliases for cleaner code @@ -327,14 +327,18 @@ pub mod builder_approach { ] } - fn add_inserts( - mut builder: ChangeSet>, + fn add_inserts( + mut builder: DiffSetBuilder>, users: &Schema, posts: &Schema, comments: &Schema, tags: &Schema, post_tags: &Schema, - ) -> ChangeSet> { + ) -> DiffSetBuilder> + where + F: DiffFormat>, + DiffSetBuilder>: DiffOps, Format = F>, + { for (schema, rows) in [ (users, get_user_rows()), (posts, get_post_rows()), @@ -353,6 +357,38 @@ pub mod builder_approach { builder } + /// Apply a batch of `(old_row, new_row)` updates against one table. + fn apply_changeset_updates( + mut builder: ChangeSet>, + table: &Schema, + updates: &[(&[Val], &[Val])], + ) -> ChangeSet> { + for (old, new) in updates { + let mut update = ChangeUpdate::from(table.clone()); + for (i, (o, n)) in old.iter().zip(new.iter()).enumerate() { + update = update.set(i, o.clone(), n.clone()).unwrap(); + } + builder = builder.update(update); + } + builder + } + + /// Apply a batch of `(column, new_value)` updates against one table. + fn apply_patchset_updates( + mut builder: PatchSet>, + table: &Schema, + updates: &[&[(usize, Val)]], + ) -> PatchSet> { + for cols in updates { + let mut update = PatchUpdate::from(table.clone()); + for (i, val) in *cols { + update = update.set(*i, val.clone()).unwrap(); + } + builder = builder.update(update); + } + builder + } + fn add_changeset_updates( mut builder: ChangeSet>, users: &Schema, @@ -400,13 +436,7 @@ pub mod builder_approach { ], ), ]; - for (old, new) in user_updates { - let mut update = ChangeUpdate::from(users.clone()); - for (i, (o, n)) in old.iter().zip(new.iter()).enumerate() { - update = update.set(i, o.clone(), n.clone()).unwrap(); - } - builder = builder.update(update); - } + builder = apply_changeset_updates(builder, users, user_updates); let post_updates: &[(&[Val], &[Val])] = &[ ( @@ -476,13 +506,7 @@ pub mod builder_approach { ], ), ]; - for (old, new) in post_updates { - let mut update = ChangeUpdate::from(posts.clone()); - for (i, (o, n)) in old.iter().zip(new.iter()).enumerate() { - update = update.set(i, o.clone(), n.clone()).unwrap(); - } - builder = builder.update(update); - } + builder = apply_changeset_updates(builder, posts, post_updates); builder } @@ -547,32 +571,6 @@ pub mod builder_approach { builder.build() } - fn add_patchset_inserts( - mut builder: PatchSet>, - users: &Schema, - posts: &Schema, - comments: &Schema, - tags: &Schema, - post_tags: &Schema, - ) -> PatchSet> { - for (schema, rows) in [ - (users, get_user_rows()), - (posts, get_post_rows()), - (tags, get_tag_rows()), - (post_tags, get_post_tag_rows()), - (comments, get_comment_rows()), - ] { - for row in rows { - let mut insert = Insert::from(schema.clone()); - for (i, val) in row.iter().enumerate() { - insert = insert.set(i, val.clone()).unwrap(); - } - builder = builder.insert(insert); - } - } - builder - } - fn add_patchset_updates( mut builder: PatchSet>, users: &Schema, @@ -582,13 +580,7 @@ pub mod builder_approach { &[(0, 1i64.into()), (4, 1002000i64.into())], &[(0, 2i64.into()), (4, 1002100i64.into())], ]; - for cols in user_updates { - let mut update = PatchUpdate::from(users.clone()); - for (i, val) in *cols { - update = update.set(*i, val.clone()).unwrap(); - } - builder = builder.update(update); - } + builder = apply_patchset_updates(builder, users, user_updates); let post_updates: &[&[(usize, Val)]] = &[ &[(0, 1i64.into()), (6, 10i64.into())], @@ -599,13 +591,7 @@ pub mod builder_approach { (5, 1002200i64.into()), ], ]; - for cols in post_updates { - let mut update = PatchUpdate::from(posts.clone()); - for (i, val) in *cols { - update = update.set(*i, val.clone()).unwrap(); - } - builder = builder.update(update); - } + builder = apply_patchset_updates(builder, posts, post_updates); builder } @@ -628,7 +614,7 @@ pub mod builder_approach { pub fn patchset() -> Vec { let (users, posts, comments, tags, post_tags) = create_schemas(); let builder = PatchSet::new(); - let builder = add_patchset_inserts(builder, &users, &posts, &comments, &tags, &post_tags); + let builder = add_inserts(builder, &users, &posts, &comments, &tags, &post_tags); let builder = add_patchset_updates(builder, &users, &posts); let builder = add_patchset_deletes(builder, &users, &comments, &post_tags); builder.build() diff --git a/integration-tests/diesel-e2e/Cargo.toml b/integration-tests/diesel-e2e/Cargo.toml index 79a6bac68..7d570f29d 100644 --- a/integration-tests/diesel-e2e/Cargo.toml +++ b/integration-tests/diesel-e2e/Cargo.toml @@ -32,6 +32,11 @@ diesel = { version = "2", default-features = false, features = [ "mysql", "sqlite", ] } +diesel-async = { version = "0.9", default-features = false, features = [ + "postgres", + "mysql", + "sqlite", +] } [dev-dependencies] testcontainers-modules = { version = "0.15", features = [ @@ -39,11 +44,6 @@ testcontainers-modules = { version = "0.15", features = [ "mysql", "blocking", ] } -diesel-async = { version = "0.9", default-features = false, features = [ - "postgres", - "mysql", - "sqlite", -] } tokio = { version = "1", features = ["macros", "rt", "rt-multi-thread"] } [patch.crates-io] diff --git a/integration-tests/diesel-e2e/src/lib.rs b/integration-tests/diesel-e2e/src/lib.rs index 946343e5f..df18a3f19 100644 --- a/integration-tests/diesel-e2e/src/lib.rs +++ b/integration-tests/diesel-e2e/src/lib.rs @@ -1,11 +1,260 @@ //! Backend-agnostic helpers for the E2E test suite. //! -//! Every function here returns a fully constructed [`PatchSet`] or -//! [`SimpleTable`] schema. The per-backend test files take care of DDL, -//! container startup, connection, patchset application, and verification. +//! The schema and patchset builders below return fully constructed +//! [`SimpleTable`] and [`PatchSet`] values. The per-backend test files keep +//! the DDL and container startup. Scenario application, row loading, and +//! verification happen here, generic over the diesel connection, with async +//! twins over `diesel_async`. +use diesel::QueryableByName; +use diesel::sql_types::{BigInt, Binary, Double, Nullable, Text}; use sqlite_diff_rs::{DiffOps, Insert, PatchDelete, PatchSet, PatchUpdate, SimpleTable}; +/// A `users` row as loaded by `sql_query`. +#[derive(QueryableByName, Debug, PartialEq)] +pub struct UserRow { + /// `id` column. + #[diesel(sql_type = BigInt)] + pub id: i64, + /// `name` column. + #[diesel(sql_type = Text)] + pub name: String, + /// `email` column. + #[diesel(sql_type = Nullable)] + pub email: Option, + /// `score` column. + #[diesel(sql_type = Nullable)] + pub score: Option, +} + +/// A `blobs` row as loaded by `sql_query`. +#[derive(QueryableByName, Debug, PartialEq)] +pub struct BlobRow { + /// `id` column. + #[diesel(sql_type = BigInt)] + pub id: i64, + /// `payload` column. + #[diesel(sql_type = Nullable)] + pub payload: Option>, +} + +/// A `kv` row as loaded by `sql_query`. +#[derive(QueryableByName, Debug, PartialEq)] +pub struct KvRow { + /// `tenant_id` column. + #[diesel(sql_type = BigInt)] + pub tenant_id: i64, + /// `user_id` column. + #[diesel(sql_type = BigInt)] + pub user_id: i64, + /// `value` column. + #[diesel(sql_type = Nullable)] + pub value: Option, +} + +// The sync and async scenario helpers live in separate modules because +// `diesel::RunQueryDsl` and `diesel_async::RunQueryDsl` are both blanket +// traits and cannot share one import scope without making `.load` ambiguous. +mod sync { + use diesel::query_builder::{QueryFragment, QueryId, SqlQuery}; + use diesel::query_dsl::LoadQuery; + use diesel::sql_query; + use diesel::{Connection, RunQueryDsl}; + use sqlite_diff_rs::{ApplyOps, PatchsetOp, SimpleTable}; + + use crate::{ + BlobRow, KvRow, UserRow, blobs_schema, insert_blob_row, insert_three_users, kv_full_cycle, + kv_schema, update_alice_delete_bob, users_schema, + }; + + /// Create the `users` table, insert three users, then apply the + /// update/delete patchset, verifying the table after each step. + pub fn users_full_cycle(conn: &mut Conn, ddl: &str) + where + Conn: Connection, + SqlQuery: LoadQuery<'static, Conn, UserRow>, + for<'a> PatchsetOp<'a, SimpleTable, String, Vec>: + QueryFragment + QueryId + RunQueryDsl, + { + conn.batch_execute(ddl).unwrap(); + + let schema = users_schema(); + assert_eq!(insert_three_users(&schema).iter().apply(conn).unwrap(), 3); + + let rows: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") + .load(conn) + .unwrap(); + assert_eq!(rows.len(), 3); + assert_eq!(rows[0].id, 1); + assert_eq!(rows[0].name, "Alice"); + assert_eq!(rows[0].email.as_deref(), Some("alice@example.com")); + assert_eq!(rows[0].score, Some(95.5)); + assert_eq!(rows[1].id, 2); + assert_eq!(rows[1].email, None); + assert_eq!(rows[2].name, "Carol'); DROP TABLE users; --"); + assert_eq!(rows[2].score, None); + + update_alice_delete_bob(&schema).iter().apply(conn).unwrap(); + + let after: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") + .load(conn) + .unwrap(); + assert_eq!(after.len(), 2); + assert_eq!(after[0].id, 1); + assert_eq!(after[0].name, "Alice"); + assert_eq!(after[0].email.as_deref(), Some("alice+new@example.com")); + assert_eq!(after[0].score, Some(99.0)); + assert_eq!(after[1].id, 3); + } + + /// Create the `blobs` table, insert the raw-bytes row, and verify the + /// payload round-trips byte for byte. + pub fn blobs_roundtrip(conn: &mut Conn, ddl: &str) + where + Conn: Connection, + SqlQuery: LoadQuery<'static, Conn, BlobRow>, + for<'a> PatchsetOp<'a, SimpleTable, String, Vec>: + QueryFragment + QueryId + RunQueryDsl, + { + conn.batch_execute(ddl).unwrap(); + + let schema = blobs_schema(); + insert_blob_row(&schema).iter().apply(conn).unwrap(); + + let rows: Vec = sql_query("SELECT id, payload FROM blobs ORDER BY id") + .load(conn) + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].id, 1); + assert_eq!( + rows[0].payload.as_deref(), + Some(&[0x00, b'\'', 0x7F, 0x80, 0xFE, 0xFF][..]) + ); + } + + /// Create the `kv` table, run the composite-PK insert/update/delete + /// cycle, and verify the surviving row. + pub fn kv_lifecycle(conn: &mut Conn, ddl: &str) + where + Conn: Connection, + SqlQuery: LoadQuery<'static, Conn, KvRow>, + for<'a> PatchsetOp<'a, SimpleTable, String, Vec>: + QueryFragment + QueryId + RunQueryDsl, + { + conn.batch_execute(ddl).unwrap(); + + let schema = kv_schema(); + kv_full_cycle(&schema).iter().apply(conn).unwrap(); + + let rows: Vec = + sql_query("SELECT tenant_id, user_id, value FROM kv ORDER BY tenant_id, user_id") + .load(conn) + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].tenant_id, 1); + assert_eq!(rows[0].user_id, 20); + assert_eq!(rows[0].value.as_deref(), Some("two-updated")); + } +} + +mod async_ { + use diesel::query_builder::{QueryFragment, QueryId, SqlQuery}; + use diesel::sql_query; + use diesel_async::{AsyncConnection, RunQueryDsl}; + use sqlite_diff_rs::{ApplyOpsAsync, PatchsetOp, SimpleTable}; + + use crate::{ + KvRow, UserRow, insert_three_users, kv_full_cycle, kv_schema, update_alice_delete_bob, + users_schema, + }; + + /// Async twin of [`crate::users_full_cycle`]. + pub async fn users_full_cycle_async(conn: &mut Conn, ddl: &str) + where + Conn: AsyncConnection, + SqlQuery: diesel_async::methods::LoadQuery<'static, Conn, UserRow>, + for<'a> PatchsetOp<'a, SimpleTable, String, Vec>: + QueryFragment + QueryId + Send, + { + conn.batch_execute(ddl).await.unwrap(); + + let schema = users_schema(); + assert_eq!( + insert_three_users(&schema) + .iter() + .apply_async(conn) + .await + .unwrap(), + 3 + ); + + let rows: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") + .load(conn) + .await + .unwrap(); + assert_eq!(rows.len(), 3); + assert_eq!(rows[0].id, 1); + assert_eq!(rows[0].name, "Alice"); + assert_eq!(rows[0].email.as_deref(), Some("alice@example.com")); + assert_eq!(rows[0].score, Some(95.5)); + assert_eq!(rows[1].id, 2); + assert_eq!(rows[1].email, None); + assert_eq!(rows[2].name, "Carol'); DROP TABLE users; --"); + assert_eq!(rows[2].score, None); + + update_alice_delete_bob(&schema) + .iter() + .apply_transactional_async(conn) + .await + .unwrap(); + + let after: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") + .load(conn) + .await + .unwrap(); + assert_eq!(after.len(), 2); + assert_eq!(after[0].id, 1); + assert_eq!(after[0].name, "Alice"); + assert_eq!(after[0].email.as_deref(), Some("alice+new@example.com")); + assert_eq!(after[0].score, Some(99.0)); + assert_eq!(after[1].id, 3); + } + + /// Async twin of [`crate::kv_lifecycle`]. + pub async fn kv_lifecycle_async(conn: &mut Conn, ddl: &str) + where + Conn: AsyncConnection, + SqlQuery: diesel_async::methods::LoadQuery<'static, Conn, KvRow>, + for<'a> PatchsetOp<'a, SimpleTable, String, Vec>: + QueryFragment + QueryId + Send, + { + conn.batch_execute(ddl).await.unwrap(); + + let schema = kv_schema(); + kv_full_cycle(&schema) + .iter() + .apply_async(conn) + .await + .unwrap(); + + let rows: Vec = + sql_query("SELECT tenant_id, user_id, value FROM kv ORDER BY tenant_id, user_id") + .load(conn) + .await + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].tenant_id, 1); + assert_eq!(rows[0].user_id, 20); + assert_eq!(rows[0].value.as_deref(), Some("two-updated")); + } +} + +pub use async_::kv_lifecycle_async; +pub use async_::users_full_cycle_async; +pub use sync::blobs_roundtrip; +pub use sync::kv_lifecycle; +pub use sync::users_full_cycle; + /// `users(id INT PK, name TEXT, email TEXT NULL, score REAL NULL)`. #[must_use] pub fn users_schema() -> SimpleTable { diff --git a/integration-tests/diesel-e2e/tests/mysql_async_e2e.rs b/integration-tests/diesel-e2e/tests/mysql_async_e2e.rs index 84d37b488..6c7c29904 100644 --- a/integration-tests/diesel-e2e/tests/mysql_async_e2e.rs +++ b/integration-tests/diesel-e2e/tests/mysql_async_e2e.rs @@ -4,24 +4,12 @@ //! //! Requires a running Docker daemon. Without it, `Mysql::default().start()` //! errors and the tests fail loudly. -//! -//! Only the async `diesel_async::RunQueryDsl` is imported (no -//! `diesel::prelude::*` glob), so `.load`/`.execute` resolve unambiguously to -//! the async connection. -use diesel::prelude::QueryableByName; -use diesel::sql_query; -use diesel::sql_types::{BigInt, Double, Nullable, Text}; -use diesel_async::{AsyncConnection, AsyncMysqlConnection, RunQueryDsl, SimpleAsyncConnection}; +use diesel_async::{AsyncConnection, AsyncMysqlConnection}; use testcontainers_modules::mysql; use testcontainers_modules::testcontainers::ContainerAsync; use testcontainers_modules::testcontainers::runners::AsyncRunner; -use diesel_e2e::{ - insert_three_users, kv_full_cycle, kv_schema, update_alice_delete_bob, users_schema, -}; -use sqlite_diff_rs::ApplyOpsAsync; - // MySQL needs an explicit length on indexed text, so `name`/`email` use // `VARCHAR(255)`; otherwise the layout mirrors the SQLite and Postgres cases. const USERS_DDL: &str = "\ @@ -40,28 +28,6 @@ CREATE TABLE kv ( PRIMARY KEY (tenant_id, user_id) )"; -#[derive(QueryableByName, Debug, PartialEq)] -struct UserRow { - #[diesel(sql_type = BigInt)] - id: i64, - #[diesel(sql_type = Text)] - name: String, - #[diesel(sql_type = Nullable)] - email: Option, - #[diesel(sql_type = Nullable)] - score: Option, -} - -#[derive(QueryableByName, Debug, PartialEq)] -struct KvRow { - #[diesel(sql_type = BigInt)] - tenant_id: i64, - #[diesel(sql_type = BigInt)] - user_id: i64, - #[diesel(sql_type = Nullable)] - value: Option, -} - async fn boot() -> (ContainerAsync, AsyncMysqlConnection) { let container = mysql::Mysql::default() .start() @@ -93,59 +59,11 @@ async fn boot() -> (ContainerAsync, AsyncMysqlConnection) { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn users_insert_update_delete_full_cycle_async() { let (_container, mut conn) = boot().await; - conn.batch_execute(USERS_DDL).await.unwrap(); - - let schema = users_schema(); - insert_three_users(&schema) - .iter() - .apply_async(&mut conn) - .await - .unwrap(); - - let rows: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .await - .unwrap(); - assert_eq!(rows.len(), 3); - assert_eq!(rows[0].email.as_deref(), Some("alice@example.com")); - assert_eq!(rows[0].score, Some(95.5)); - assert_eq!(rows[1].email, None); - assert_eq!(rows[2].name, "Carol'); DROP TABLE users; --"); - - update_alice_delete_bob(&schema) - .iter() - .apply_transactional_async(&mut conn) - .await - .unwrap(); - - let after: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .await - .unwrap(); - assert_eq!(after.len(), 2); - assert_eq!(after[0].email.as_deref(), Some("alice+new@example.com")); - assert_eq!(after[1].id, 3); + diesel_e2e::users_full_cycle_async(&mut conn, USERS_DDL).await; } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn kv_composite_pk_lifecycle_async() { let (_container, mut conn) = boot().await; - conn.batch_execute(KV_DDL).await.unwrap(); - - let schema = kv_schema(); - kv_full_cycle(&schema) - .iter() - .apply_transactional_async(&mut conn) - .await - .unwrap(); - - let rows: Vec = - sql_query("SELECT tenant_id, user_id, value FROM kv ORDER BY tenant_id, user_id") - .load(&mut conn) - .await - .unwrap(); - assert_eq!(rows.len(), 1); - assert_eq!(rows[0].tenant_id, 1); - assert_eq!(rows[0].user_id, 20); - assert_eq!(rows[0].value.as_deref(), Some("two-updated")); + diesel_e2e::kv_lifecycle_async(&mut conn, KV_DDL).await; } diff --git a/integration-tests/diesel-e2e/tests/mysql_e2e.rs b/integration-tests/diesel-e2e/tests/mysql_e2e.rs index 767caeda2..8c1ef19d9 100644 --- a/integration-tests/diesel-e2e/tests/mysql_e2e.rs +++ b/integration-tests/diesel-e2e/tests/mysql_e2e.rs @@ -4,18 +4,11 @@ //! Requires a running Docker daemon. Without it, `Mysql::default().start()` //! errors and the tests fail loudly. -use diesel::connection::SimpleConnection; -use diesel::prelude::*; -use diesel::sql_query; -use diesel::sql_types::{BigInt, Binary, Double, Nullable, Text}; +use diesel::Connection; +use diesel::mysql::MysqlConnection; use testcontainers_modules::mysql; use testcontainers_modules::testcontainers::runners::SyncRunner; -use diesel_e2e::{ - blobs_schema, insert_blob_row, insert_three_users, kv_full_cycle, kv_schema, - update_alice_delete_bob, users_schema, -}; - // MySQL identifiers: `LONGBLOB` for large binary, `DOUBLE` for floats, `TEXT` // for variable text. `name` uses `VARCHAR(255)` because MySQL requires an // explicit length on indexed text; the schema layout otherwise mirrors the @@ -42,36 +35,6 @@ CREATE TABLE kv ( PRIMARY KEY (tenant_id, user_id) )"; -#[derive(QueryableByName, Debug, PartialEq)] -struct UserRow { - #[diesel(sql_type = BigInt)] - id: i64, - #[diesel(sql_type = Text)] - name: String, - #[diesel(sql_type = Nullable)] - email: Option, - #[diesel(sql_type = Nullable)] - score: Option, -} - -#[derive(QueryableByName, Debug, PartialEq)] -struct BlobRow { - #[diesel(sql_type = BigInt)] - id: i64, - #[diesel(sql_type = Nullable)] - payload: Option>, -} - -#[derive(QueryableByName, Debug, PartialEq)] -struct KvRow { - #[diesel(sql_type = BigInt)] - tenant_id: i64, - #[diesel(sql_type = BigInt)] - user_id: i64, - #[diesel(sql_type = Nullable)] - value: Option, -} - fn boot() -> ( testcontainers_modules::testcontainers::Container, MysqlConnection, @@ -96,80 +59,20 @@ fn boot() -> ( (container, conn) } -fn apply( - conn: &mut MysqlConnection, - patchset: &sqlite_diff_rs::PatchSet>, -) { - for op in patchset.iter() { - op.execute(conn).expect("execute patchset op against MySQL"); - } -} - #[test] fn users_insert_update_delete_full_cycle() { let (_container, mut conn) = boot(); - conn.batch_execute(USERS_DDL).unwrap(); - - let schema = users_schema(); - apply(&mut conn, &insert_three_users(&schema)); - - let rows: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .unwrap(); - assert_eq!(rows.len(), 3); - assert_eq!(rows[0].id, 1); - assert_eq!(rows[0].name, "Alice"); - assert_eq!(rows[0].email.as_deref(), Some("alice@example.com")); - assert_eq!(rows[0].score, Some(95.5)); - assert_eq!(rows[1].id, 2); - assert_eq!(rows[1].email, None); - assert_eq!(rows[2].name, "Carol'); DROP TABLE users; --"); - assert_eq!(rows[2].score, None); - - apply(&mut conn, &update_alice_delete_bob(&schema)); - - let after: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .unwrap(); - assert_eq!(after.len(), 2); - assert_eq!(after[0].id, 1); - assert_eq!(after[0].email.as_deref(), Some("alice+new@example.com")); - assert_eq!(after[0].score, Some(99.0)); - assert_eq!(after[1].id, 3); + diesel_e2e::users_full_cycle(&mut conn, USERS_DDL); } #[test] fn blobs_null_and_raw_bytes_roundtrip() { let (_container, mut conn) = boot(); - conn.batch_execute(BLOBS_DDL).unwrap(); - - let schema = blobs_schema(); - apply(&mut conn, &insert_blob_row(&schema)); - - let rows: Vec = sql_query("SELECT id, payload FROM blobs ORDER BY id") - .load(&mut conn) - .unwrap(); - assert_eq!(rows.len(), 1); - assert_eq!( - rows[0].payload.as_deref(), - Some(&[0x00, b'\'', 0x7F, 0x80, 0xFE, 0xFF][..]) - ); + diesel_e2e::blobs_roundtrip(&mut conn, BLOBS_DDL); } #[test] fn kv_composite_pk_lifecycle() { let (_container, mut conn) = boot(); - conn.batch_execute(KV_DDL).unwrap(); - - let schema = kv_schema(); - apply(&mut conn, &kv_full_cycle(&schema)); - - let rows: Vec = - sql_query("SELECT tenant_id, user_id, value FROM kv ORDER BY tenant_id, user_id") - .load(&mut conn) - .unwrap(); - assert_eq!(rows.len(), 1); - assert_eq!(rows[0].tenant_id, 1); - assert_eq!(rows[0].user_id, 20); - assert_eq!(rows[0].value.as_deref(), Some("two-updated")); + diesel_e2e::kv_lifecycle(&mut conn, KV_DDL); } diff --git a/integration-tests/diesel-e2e/tests/postgres_async_e2e.rs b/integration-tests/diesel-e2e/tests/postgres_async_e2e.rs index 552a1e325..bee61ac25 100644 --- a/integration-tests/diesel-e2e/tests/postgres_async_e2e.rs +++ b/integration-tests/diesel-e2e/tests/postgres_async_e2e.rs @@ -14,15 +14,12 @@ use diesel::prelude::QueryableByName; use diesel::query_builder::AstPass; use diesel::result::QueryResult; use diesel::sql_query; -use diesel::sql_types::{BigInt, Bool, Double, Nullable, Text}; +use diesel::sql_types::{BigInt, Bool}; use diesel_async::{AsyncConnection, AsyncPgConnection, RunQueryDsl, SimpleAsyncConnection}; use testcontainers_modules::postgres; use testcontainers_modules::testcontainers::ContainerAsync; use testcontainers_modules::testcontainers::runners::AsyncRunner; -use diesel_e2e::{ - insert_three_users, kv_full_cycle, kv_schema, update_alice_delete_bob, users_schema, -}; use sqlite_diff_rs::{ Adapter, ApplyOpsAsync, Binder, DefaultBinder, DiffOps, Insert, PatchSet, SimpleTable, Value, }; @@ -43,28 +40,6 @@ CREATE TABLE kv ( PRIMARY KEY (tenant_id, user_id) )"; -#[derive(QueryableByName, Debug, PartialEq)] -struct UserRow { - #[diesel(sql_type = BigInt)] - id: i64, - #[diesel(sql_type = Text)] - name: String, - #[diesel(sql_type = Nullable)] - email: Option, - #[diesel(sql_type = Nullable)] - score: Option, -} - -#[derive(QueryableByName, Debug, PartialEq)] -struct KvRow { - #[diesel(sql_type = BigInt)] - tenant_id: i64, - #[diesel(sql_type = BigInt)] - user_id: i64, - #[diesel(sql_type = Nullable)] - value: Option, -} - async fn boot() -> (ContainerAsync, AsyncPgConnection) { let container = postgres::Postgres::default() .start() @@ -85,61 +60,13 @@ async fn boot() -> (ContainerAsync, AsyncPgConnection) { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn users_insert_update_delete_full_cycle_async() { let (_container, mut conn) = boot().await; - conn.batch_execute(USERS_DDL).await.unwrap(); - - let schema = users_schema(); - insert_three_users(&schema) - .iter() - .apply_async(&mut conn) - .await - .unwrap(); - - let rows: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .await - .unwrap(); - assert_eq!(rows.len(), 3); - assert_eq!(rows[0].email.as_deref(), Some("alice@example.com")); - assert_eq!(rows[0].score, Some(95.5)); - assert_eq!(rows[1].email, None); - assert_eq!(rows[2].name, "Carol'); DROP TABLE users; --"); - - update_alice_delete_bob(&schema) - .iter() - .apply_transactional_async(&mut conn) - .await - .unwrap(); - - let after: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .await - .unwrap(); - assert_eq!(after.len(), 2); - assert_eq!(after[0].email.as_deref(), Some("alice+new@example.com")); - assert_eq!(after[1].id, 3); + diesel_e2e::users_full_cycle_async(&mut conn, USERS_DDL).await; } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn kv_composite_pk_lifecycle_async() { let (_container, mut conn) = boot().await; - conn.batch_execute(KV_DDL).await.unwrap(); - - let schema = kv_schema(); - kv_full_cycle(&schema) - .iter() - .apply_transactional_async(&mut conn) - .await - .unwrap(); - - let rows: Vec = - sql_query("SELECT tenant_id, user_id, value FROM kv ORDER BY tenant_id, user_id") - .load(&mut conn) - .await - .unwrap(); - assert_eq!(rows.len(), 1); - assert_eq!(rows[0].tenant_id, 1); - assert_eq!(rows[0].user_id, 20); - assert_eq!(rows[0].value.as_deref(), Some("two-updated")); + diesel_e2e::kv_lifecycle_async(&mut conn, KV_DDL).await; } // Native BOOLEAN bind through the adapter over the async connection: the diff --git a/integration-tests/diesel-e2e/tests/postgres_e2e.rs b/integration-tests/diesel-e2e/tests/postgres_e2e.rs index 86be6f845..f9aed6a64 100644 --- a/integration-tests/diesel-e2e/tests/postgres_e2e.rs +++ b/integration-tests/diesel-e2e/tests/postgres_e2e.rs @@ -7,15 +7,10 @@ use diesel::connection::SimpleConnection; use diesel::prelude::*; use diesel::sql_query; -use diesel::sql_types::{BigInt, Binary, Double, Nullable, Text}; +use diesel::sql_types::{BigInt, Binary, Double, Text}; use testcontainers_modules::postgres; use testcontainers_modules::testcontainers::runners::SyncRunner; -use diesel_e2e::{ - blobs_schema, insert_blob_row, insert_three_users, kv_full_cycle, kv_schema, - update_alice_delete_bob, users_schema, -}; - const USERS_DDL: &str = "\ CREATE TABLE users ( id BIGINT PRIMARY KEY, @@ -38,36 +33,6 @@ CREATE TABLE kv ( PRIMARY KEY (tenant_id, user_id) )"; -#[derive(QueryableByName, Debug, PartialEq)] -struct UserRow { - #[diesel(sql_type = BigInt)] - id: i64, - #[diesel(sql_type = Text)] - name: String, - #[diesel(sql_type = Nullable)] - email: Option, - #[diesel(sql_type = Nullable)] - score: Option, -} - -#[derive(QueryableByName, Debug, PartialEq)] -struct BlobRow { - #[diesel(sql_type = BigInt)] - id: i64, - #[diesel(sql_type = Nullable)] - payload: Option>, -} - -#[derive(QueryableByName, Debug, PartialEq)] -struct KvRow { - #[diesel(sql_type = BigInt)] - tenant_id: i64, - #[diesel(sql_type = BigInt)] - user_id: i64, - #[diesel(sql_type = Nullable)] - value: Option, -} - fn boot() -> ( testcontainers_modules::testcontainers::Container, PgConnection, @@ -84,83 +49,22 @@ fn boot() -> ( (container, conn) } -fn apply( - conn: &mut PgConnection, - patchset: &sqlite_diff_rs::PatchSet>, -) { - for op in patchset.iter() { - op.execute(conn) - .expect("execute patchset op against Postgres"); - } -} - #[test] fn users_insert_update_delete_full_cycle() { let (_container, mut conn) = boot(); - conn.batch_execute(USERS_DDL).unwrap(); - - let schema = users_schema(); - apply(&mut conn, &insert_three_users(&schema)); - - let rows: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .unwrap(); - assert_eq!(rows.len(), 3); - assert_eq!(rows[0].id, 1); - assert_eq!(rows[0].name, "Alice"); - assert_eq!(rows[0].email.as_deref(), Some("alice@example.com")); - assert_eq!(rows[0].score, Some(95.5)); - assert_eq!(rows[1].id, 2); - assert_eq!(rows[1].email, None); - assert_eq!(rows[2].name, "Carol'); DROP TABLE users; --"); - assert_eq!(rows[2].score, None); - - apply(&mut conn, &update_alice_delete_bob(&schema)); - - let after: Vec = sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .unwrap(); - assert_eq!(after.len(), 2); - assert_eq!(after[0].id, 1); - assert_eq!(after[0].email.as_deref(), Some("alice+new@example.com")); - assert_eq!(after[0].score, Some(99.0)); - assert_eq!(after[1].id, 3); + diesel_e2e::users_full_cycle(&mut conn, USERS_DDL); } #[test] fn blobs_null_and_raw_bytes_roundtrip() { let (_container, mut conn) = boot(); - conn.batch_execute(BLOBS_DDL).unwrap(); - - let schema = blobs_schema(); - apply(&mut conn, &insert_blob_row(&schema)); - - let rows: Vec = sql_query("SELECT id, payload FROM blobs ORDER BY id") - .load(&mut conn) - .unwrap(); - assert_eq!(rows.len(), 1); - assert_eq!( - rows[0].payload.as_deref(), - Some(&[0x00, b'\'', 0x7F, 0x80, 0xFE, 0xFF][..]) - ); + diesel_e2e::blobs_roundtrip(&mut conn, BLOBS_DDL); } #[test] fn kv_composite_pk_lifecycle() { let (_container, mut conn) = boot(); - conn.batch_execute(KV_DDL).unwrap(); - - let schema = kv_schema(); - apply(&mut conn, &kv_full_cycle(&schema)); - - let rows: Vec = - sql_query("SELECT tenant_id, user_id, value FROM kv ORDER BY tenant_id, user_id") - .load(&mut conn) - .unwrap(); - assert_eq!(rows.len(), 1); - assert_eq!(rows[0].tenant_id, 1); - assert_eq!(rows[0].user_id, 20); - assert_eq!(rows[0].value.as_deref(), Some("two-updated")); + diesel_e2e::kv_lifecycle(&mut conn, KV_DDL); } // Adapter path: native BOOLEAN bind against a real Pg column --------------- diff --git a/integration-tests/diesel-e2e/tests/sqlite_async_e2e.rs b/integration-tests/diesel-e2e/tests/sqlite_async_e2e.rs index 4b1d50fc8..1a270a990 100644 --- a/integration-tests/diesel-e2e/tests/sqlite_async_e2e.rs +++ b/integration-tests/diesel-e2e/tests/sqlite_async_e2e.rs @@ -10,13 +10,13 @@ use diesel::prelude::QueryableByName; use diesel::result::QueryResult; use diesel::sql_query; -use diesel::sql_types::{BigInt, Double, Nullable, Text}; +use diesel::sql_types::BigInt; use diesel::sqlite::{Sqlite, SqliteConnection}; use diesel_async::sync_connection_wrapper::SyncConnectionWrapper; use diesel_async::{AsyncConnection, RunQueryDsl, SimpleAsyncConnection}; use diesel_e2e::{ - duplicate_email_patchset, insert_three_users, kv_full_cycle, kv_schema, - update_alice_delete_bob, users_schema, + KvRow, UserRow, duplicate_email_patchset, insert_three_users, kv_full_cycle, kv_schema, + users_schema, }; use sqlite_diff_rs::{ Adapter, ApplyOpsAsync, Binder, DefaultBinder, DiffOps, Insert, PatchSet, SimpleTable, Value, @@ -48,28 +48,6 @@ CREATE TABLE users ( email TEXT NOT NULL UNIQUE )"; -#[derive(QueryableByName, Debug, PartialEq)] -struct UserRow { - #[diesel(sql_type = BigInt)] - id: i64, - #[diesel(sql_type = Text)] - name: String, - #[diesel(sql_type = Nullable)] - email: Option, - #[diesel(sql_type = Nullable)] - score: Option, -} - -#[derive(QueryableByName, Debug, PartialEq)] -struct KvRow { - #[diesel(sql_type = BigInt)] - tenant_id: i64, - #[diesel(sql_type = BigInt)] - user_id: i64, - #[diesel(sql_type = Nullable)] - value: Option, -} - #[derive(QueryableByName, Debug, PartialEq)] struct CountRow { #[diesel(sql_type = BigInt)] @@ -85,44 +63,7 @@ async fn conn() -> Db { #[tokio::test] async fn users_insert_update_delete_full_cycle_async() { let mut conn = conn().await; - conn.batch_execute(USERS_DDL).await.unwrap(); - - let schema = users_schema(); - let inserted = insert_three_users(&schema); - let count = inserted.iter().apply_async(&mut conn).await.unwrap(); - assert_eq!(count, 3); - - let after_insert: Vec = - sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .await - .unwrap(); - assert_eq!(after_insert.len(), 3); - assert_eq!(after_insert[0].id, 1); - assert_eq!(after_insert[0].email.as_deref(), Some("alice@example.com")); - assert_eq!(after_insert[0].score, Some(95.5)); - assert_eq!(after_insert[1].email, None); - assert_eq!(after_insert[2].name, "Carol'); DROP TABLE users; --"); - assert_eq!(after_insert[2].score, None); - - update_alice_delete_bob(&schema) - .iter() - .apply_async(&mut conn) - .await - .unwrap(); - - let after_update: Vec = - sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .await - .unwrap(); - assert_eq!(after_update.len(), 2); - assert_eq!( - after_update[0].email.as_deref(), - Some("alice+new@example.com") - ); - assert_eq!(after_update[0].score, Some(99.0)); - assert_eq!(after_update[1].id, 3); // Bob deleted + diesel_e2e::users_full_cycle_async(&mut conn, USERS_DDL).await; } #[tokio::test] diff --git a/integration-tests/diesel-e2e/tests/sqlite_e2e.rs b/integration-tests/diesel-e2e/tests/sqlite_e2e.rs index a3658ea4e..88d2a9808 100644 --- a/integration-tests/diesel-e2e/tests/sqlite_e2e.rs +++ b/integration-tests/diesel-e2e/tests/sqlite_e2e.rs @@ -4,13 +4,10 @@ use diesel::connection::SimpleConnection; use diesel::prelude::*; use diesel::sql_query; -use diesel::sql_types::{BigInt, Binary, Double, Nullable, Text}; +use diesel::sql_types::{BigInt, Text}; use sqlite_diff_rs::DiffOps; -use diesel_e2e::{ - blobs_schema, duplicate_email_patchset, insert_blob_row, insert_three_users, kv_full_cycle, - kv_schema, update_alice_delete_bob, users_schema, -}; +use diesel_e2e::{KvRow, UserRow, duplicate_email_patchset, insert_three_users, users_schema}; const USERS_DDL: &str = "\ CREATE TABLE users ( @@ -34,36 +31,6 @@ CREATE TABLE kv ( PRIMARY KEY (tenant_id, user_id) )"; -#[derive(QueryableByName, Debug, PartialEq)] -struct UserRow { - #[diesel(sql_type = BigInt)] - id: i64, - #[diesel(sql_type = Text)] - name: String, - #[diesel(sql_type = Nullable)] - email: Option, - #[diesel(sql_type = Nullable)] - score: Option, -} - -#[derive(QueryableByName, Debug, PartialEq)] -struct BlobRow { - #[diesel(sql_type = BigInt)] - id: i64, - #[diesel(sql_type = Nullable)] - payload: Option>, -} - -#[derive(QueryableByName, Debug, PartialEq)] -struct KvRow { - #[diesel(sql_type = BigInt)] - tenant_id: i64, - #[diesel(sql_type = BigInt)] - user_id: i64, - #[diesel(sql_type = Nullable)] - value: Option, -} - fn conn() -> SqliteConnection { SqliteConnection::establish(":memory:").expect("open in-memory SQLite") } @@ -82,79 +49,19 @@ fn apply( #[test] fn users_insert_update_delete_full_cycle() { let mut conn = conn(); - conn.batch_execute(USERS_DDL).unwrap(); - - let schema = users_schema(); - apply(&mut conn, &insert_three_users(&schema)); - - let after_insert: Vec = - sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .unwrap(); - assert_eq!(after_insert.len(), 3); - assert_eq!(after_insert[0].id, 1); - assert_eq!(after_insert[0].name, "Alice"); - assert_eq!(after_insert[0].email.as_deref(), Some("alice@example.com")); - assert_eq!(after_insert[0].score, Some(95.5)); - assert_eq!(after_insert[1].id, 2); - assert_eq!(after_insert[1].email, None); // set_null landed - assert_eq!(after_insert[2].name, "Carol'); DROP TABLE users; --"); // injection payload stored verbatim - assert_eq!(after_insert[2].score, None); - - apply(&mut conn, &update_alice_delete_bob(&schema)); - - let after_update: Vec = - sql_query("SELECT id, name, email, score FROM users ORDER BY id") - .load(&mut conn) - .unwrap(); - assert_eq!(after_update.len(), 2); - assert_eq!(after_update[0].id, 1); - assert_eq!( - after_update[0].email.as_deref(), - Some("alice+new@example.com") - ); - assert_eq!(after_update[0].score, Some(99.0)); - assert_eq!(after_update[0].name, "Alice"); // untouched - assert_eq!(after_update[1].id, 3); // Bob is gone + diesel_e2e::users_full_cycle(&mut conn, USERS_DDL); } #[test] fn blobs_null_and_raw_bytes_roundtrip() { let mut conn = conn(); - conn.batch_execute(BLOBS_DDL).unwrap(); - - let schema = blobs_schema(); - apply(&mut conn, &insert_blob_row(&schema)); - - let rows: Vec = sql_query("SELECT id, payload FROM blobs ORDER BY id") - .load(&mut conn) - .unwrap(); - assert_eq!(rows.len(), 1); - assert_eq!(rows[0].id, 1); - assert_eq!( - rows[0].payload.as_deref(), - Some(&[0x00, b'\'', 0x7F, 0x80, 0xFE, 0xFF][..]), - "raw bytes should survive round trip through a bind parameter" - ); + diesel_e2e::blobs_roundtrip(&mut conn, BLOBS_DDL); } #[test] fn kv_composite_pk_lifecycle() { let mut conn = conn(); - conn.batch_execute(KV_DDL).unwrap(); - - let schema = kv_schema(); - apply(&mut conn, &kv_full_cycle(&schema)); - - let rows: Vec = - sql_query("SELECT tenant_id, user_id, value FROM kv ORDER BY tenant_id, user_id") - .load(&mut conn) - .unwrap(); - // (1, 10) inserted then deleted; (1, 20) inserted then updated. - assert_eq!(rows.len(), 1); - assert_eq!(rows[0].tenant_id, 1); - assert_eq!(rows[0].user_id, 20); - assert_eq!(rows[0].value.as_deref(), Some("two-updated")); + diesel_e2e::kv_lifecycle(&mut conn, KV_DDL); } #[test] diff --git a/integration-tests/payload-size-bench/src/plots.rs b/integration-tests/payload-size-bench/src/plots.rs index 9330c2705..8e3e63a16 100644 --- a/integration-tests/payload-size-bench/src/plots.rs +++ b/integration-tests/payload-size-bench/src/plots.rs @@ -40,6 +40,41 @@ fn y_fmt(y: &f64) -> String { } } +/// Draw one line series (line, point markers, legend entry) per palette slot. +fn draw_line_series>( + chart: &mut ChartContext<'_, DB, CT>, + data: &[(f64, Vec)], + names: &[&str], +) -> Result<(), Box> +where + ::ErrorType: 'static, +{ + let n = names.len().min(PALETTE.len()); + for i in 0..n { + let color = PALETTE[i]; + let width = STROKE_W[i]; + + let points: Vec<(f64, f64)> = data + .iter() + .filter_map(|(x, ys)| ys.get(i).map(|y| (*x, *y))) + .collect(); + + chart + .draw_series(LineSeries::new(points.clone(), color.stroke_width(width)))? + .label(names[i]) + .legend(move |(x, y)| { + PathElement::new(vec![(x, y), (x + 20, y)], color.stroke_width(width)) + }); + + chart.draw_series( + points + .iter() + .map(|&(x, y)| Circle::new((x, y), 3, color.filled())), + )?; + } + Ok(()) +} + // --------------------------------------------------------------------------- // Public API // --------------------------------------------------------------------------- @@ -89,29 +124,7 @@ pub fn line_chart( .y_label_formatter(&y_fmt) .draw()?; - let n = names.len().min(PALETTE.len()); - for i in 0..n { - let color = PALETTE[i]; - let width = STROKE_W[i]; - - let points: Vec<(f64, f64)> = data - .iter() - .filter_map(|(x, ys)| ys.get(i).map(|y| (*x, *y))) - .collect(); - - chart - .draw_series(LineSeries::new(points.clone(), color.stroke_width(width)))? - .label(names[i]) - .legend(move |(x, y)| { - PathElement::new(vec![(x, y), (x + 20, y)], color.stroke_width(width)) - }); - - chart.draw_series( - points - .iter() - .map(|&(x, y)| Circle::new((x, y), 3, color.filled())), - )?; - } + draw_line_series(&mut chart, data, names)?; chart .configure_series_labels() @@ -188,28 +201,7 @@ pub fn stacked_size_and_time( .draw()?; let n = names.len().min(PALETTE.len()); - for i in 0..n { - let color = PALETTE[i]; - let width = STROKE_W[i]; - - let points: Vec<(f64, f64)> = size_data - .iter() - .filter_map(|(x, ys)| ys.get(i).map(|y| (*x, *y))) - .collect(); - - size_chart - .draw_series(LineSeries::new(points.clone(), color.stroke_width(width)))? - .label(names[i]) - .legend(move |(x, y)| { - PathElement::new(vec![(x, y), (x + 20, y)], color.stroke_width(width)) - }); - - size_chart.draw_series( - points - .iter() - .map(|&(x, y)| Circle::new((x, y), 3, color.filled())), - )?; - } + draw_line_series(&mut size_chart, size_data, names)?; size_chart .configure_series_labels() diff --git a/src/builders.rs b/src/builders.rs index 2dc32ac42..575df8e5a 100644 --- a/src/builders.rs +++ b/src/builders.rs @@ -14,6 +14,8 @@ mod sql_output; mod update_operation; mod view; +#[cfg(any(feature = "wal2json", feature = "maxwell", feature = "pg-walstream"))] +pub(crate) use change::DeleteArg; pub use change::{ChangeSet, DiffOps, DiffSet, DiffSetBuilder, PatchSet}; pub use delete_operation::{ChangeDelete, PatchDelete}; #[cfg(feature = "diesel-async")] diff --git a/src/builders/change.rs b/src/builders/change.rs index 5071671f6..1681823f9 100644 --- a/src/builders/change.rs +++ b/src/builders/change.rs @@ -373,6 +373,16 @@ pub struct DiffSetBuilder, T: SchemaWithPK, S, B> { pub(crate) tables: TableMap, } +/// Whether two table sequences agree once tables without rows are skipped. +fn same_nonempty_tables<'a, T: PartialEq + 'a, R: PartialEq + 'a>( + lhs: impl Iterator, + rhs: impl Iterator, + is_empty: impl Fn(&R) -> bool, +) -> bool { + lhs.filter(|(_, rows)| !is_empty(rows)) + .eq(rhs.filter(|(_, rows)| !is_empty(rows))) +} + /// Custom `PartialEq` that ignores tables with empty operations. /// /// Tables with no operations are not serialized (skipped in `build()`), so after @@ -391,12 +401,12 @@ where F::DeleteData: PartialEq, { fn eq(&self, other: &Self) -> bool { - // Filter out tables with empty operations, then compare element by element. // IndexMap preserves insertion order, so this also checks table ordering. - self.tables - .iter() - .filter(|(_, ops)| !ops.is_empty()) - .eq(other.tables.iter().filter(|(_, ops)| !ops.is_empty())) + same_nonempty_tables( + self.tables.iter(), + other.tables.iter(), + RowMap::::is_empty, + ) } } @@ -564,6 +574,26 @@ impl, T: SchemaWithPK, S: AsRef + Hash + Eq, B: AsRef<[u8]> self } + + /// Add an INSERT, which every format stores the same way. + pub(crate) fn add_insert(&mut self, insert: Insert) -> &mut Self + where + S: Clone, + B: Clone, + Operation: core::ops::Add>>, + { + let pk = insert.extract_pk(); + let table = insert.as_ref().clone(); + let indirect = insert.indirect; + self.add_operation( + &table, + pk, + Operation::Insert { + values: insert.into_values(), + indirect, + }, + ) + } } // Unified digest entry point for wire events (0.2.0+). @@ -622,6 +652,10 @@ pub trait DiffOps: Sized { -> DiffSetBuilder; } +/// The DELETE argument of a builder in format `F`. +#[cfg(any(feature = "wal2json", feature = "maxwell", feature = "pg-walstream"))] +pub(crate) type DeleteArg = as DiffOps>::DeleteArg; + // -- DiffOps for DiffSetBuilder ------------------------------ impl< @@ -634,17 +668,7 @@ impl< type DeleteArg = ChangeDelete; fn insert(mut self, insert: Insert) -> Self { - let pk = insert.extract_pk(); - let table = insert.as_ref().clone(); - let indirect = insert.indirect; - self.add_operation( - &table, - pk, - Operation::Insert { - values: insert.into_values(), - indirect, - }, - ); + self.add_insert(insert); self } @@ -687,17 +711,7 @@ impl, B: Clone + Hash + Eq + type DeleteArg = PatchDelete; fn insert(mut self, insert: Insert) -> Self { - let pk = insert.extract_pk(); - let table = insert.as_ref().clone(); - let indirect = insert.indirect; - self.add_operation( - &table, - pk, - Operation::Insert { - values: insert.into_values(), - indirect, - }, - ); + self.add_insert(insert); self } @@ -922,23 +936,8 @@ impl, B: Clone + Debug + AsRef<[u /// time. pub fn iter(&self) -> impl Iterator> { self.tables.iter().flat_map(|(table, rows)| { - rows.iter().map(move |(_pk, op)| match op { - Operation::Insert { values, indirect } => ChangesetOp::Insert { - table, - values: values.as_slice(), - indirect: *indirect, - }, - Operation::Update { values, indirect } => ChangesetOp::Update { - table, - values: values.as_slice(), - indirect: *indirect, - }, - Operation::Delete { data, indirect } => ChangesetOp::Delete { - table, - old_values: data.as_slice(), - indirect: *indirect, - }, - }) + rows.iter() + .map(move |(_pk, op)| ChangesetOp::of_stored(table, op)) }) } } @@ -955,24 +954,8 @@ impl, B: Clone + AsRef<[u8]>> /// via [`RunQueryDsl`](diesel::RunQueryDsl). pub fn iter(&self) -> impl Iterator> { self.tables.iter().flat_map(|(table, rows)| { - rows.iter().map(move |(pk, op)| match op { - Operation::Insert { values, indirect } => PatchsetOp::Insert { - table, - values: values.as_slice(), - indirect: *indirect, - }, - Operation::Update { values, indirect } => PatchsetOp::Update { - table, - pk: pk.as_slice(), - entries: values.as_slice(), - indirect: *indirect, - }, - Operation::Delete { indirect, .. } => PatchsetOp::Delete { - table, - pk: pk.as_slice(), - indirect: *indirect, - }, - }) + rows.iter() + .map(move |(pk, op)| PatchsetOp::of_stored(table, pk, op)) }) } } @@ -1082,10 +1065,11 @@ where F::DeleteData: PartialEq, { fn eq(&self, other: &Self) -> bool { - self.tables - .iter() - .filter(|(_, ops)| !ops.is_empty()) - .eq(other.tables.iter().filter(|(_, ops)| !ops.is_empty())) + same_nonempty_tables( + self.tables.iter().map(|(table, rows)| (table, rows)), + other.tables.iter().map(|(table, rows)| (table, rows)), + RowVec::::is_empty, + ) } } @@ -1149,23 +1133,8 @@ impl, B: Clone + Debug + AsRef<[u /// iterator is invalidated when the `DiffSet` is dropped or mutated. pub fn iter(&self) -> impl Iterator> { self.tables.iter().flat_map(|(table, rows)| { - rows.iter().map(move |(_pk, op)| match op { - Operation::Insert { values, indirect } => ChangesetOp::Insert { - table, - values: values.as_slice(), - indirect: *indirect, - }, - Operation::Update { values, indirect } => ChangesetOp::Update { - table, - values: values.as_slice(), - indirect: *indirect, - }, - Operation::Delete { data, indirect } => ChangesetOp::Delete { - table, - old_values: data.as_slice(), - indirect: *indirect, - }, - }) + rows.iter() + .map(move |(_pk, op)| ChangesetOp::of_stored(table, op)) }) } } @@ -1181,24 +1150,8 @@ impl, B: Clone + AsRef<[u8]>> /// does not carry full old-row values). pub fn iter(&self) -> impl Iterator> { self.tables.iter().flat_map(|(table, rows)| { - rows.iter().map(move |(pk, op)| match op { - Operation::Insert { values, indirect } => PatchsetOp::Insert { - table, - values: values.as_slice(), - indirect: *indirect, - }, - Operation::Update { values, indirect } => PatchsetOp::Update { - table, - pk: pk.as_slice(), - entries: values.as_slice(), - indirect: *indirect, - }, - Operation::Delete { indirect, .. } => PatchsetOp::Delete { - table, - pk: pk.as_slice(), - indirect: *indirect, - }, - }) + rows.iter() + .map(move |(pk, op)| PatchsetOp::of_stored(table, pk, op)) }) } } diff --git a/src/builders/delete_operation.rs b/src/builders/delete_operation.rs index 52d9f312c..f6a47d627 100644 --- a/src/builders/delete_operation.rs +++ b/src/builders/delete_operation.rs @@ -184,21 +184,14 @@ mod tests { } #[test] - fn test_change_delete_set_out_of_bounds() { - let err = ChangeDelete::<_, String, Vec>::from(users()) - .set(5, 1i64) - .unwrap_err(); + fn test_change_delete_out_of_bounds_columns() { + let delete = ChangeDelete::<_, String, Vec>::from(users()); + let err = delete.clone().set(5, 1i64).unwrap_err(); assert!( matches!(err, Error::ColumnIndexOutOfBounds(5, 2)), "got {err:?}" ); - } - - #[test] - fn test_change_delete_set_null_out_of_bounds() { - let err = ChangeDelete::<_, String, Vec>::from(users()) - .set_null(2) - .unwrap_err(); + let err = delete.set_null(2).unwrap_err(); assert!( matches!(err, Error::ColumnIndexOutOfBounds(2, 2)), "got {err:?}" diff --git a/src/builders/operation.rs b/src/builders/operation.rs index e786b1704..4bfe17885 100644 --- a/src/builders/operation.rs +++ b/src/builders/operation.rs @@ -61,6 +61,33 @@ impl, S, B> Operation { } } +/// The `(old, new)` pairs of an UPDATE in format `F`. +type UpdatePairs = Vec<(>::Old, MaybeValue)>; + +/// The row of an INSERT with every defined new value of a following UPDATE applied. +fn apply_update, S, B>( + mut values: Vec>, + updates: UpdatePairs, +) -> Vec> { + for (idx, (_old, new)) in updates.into_iter().enumerate() { + if let Some(new_val) = new { + values[idx] = new_val; + } + } + values +} + +/// Two consecutive UPDATEs as one, keeping the first's old values and the second's new ones. +fn merge_updates, S, B>( + lhs: UpdatePairs, + rhs: UpdatePairs, +) -> UpdatePairs { + lhs.into_iter() + .zip(rhs) + .map(|((old, _mid), (_mid2, new))| (old, new)) + .collect() +} + impl, S: PartialEq + AsRef, B: PartialEq + AsRef<[u8]>> PartialEq for Operation where @@ -166,18 +193,14 @@ impl, B: Clone + Debug + PartialEq + A // INSERT + UPDATE: apply update to insert values ( - Self::Insert { mut values, .. }, + Self::Insert { values, .. }, Self::Update { values: updates, .. }, - ) => { - for (idx, (_old, new)) in updates.into_iter().enumerate() { - if let Some(new_val) = new { - values[idx] = new_val; - } - } - Some(Self::Insert { values, indirect }) - } + ) => Some(Self::Insert { + values: apply_update::(values, updates), + indirect, + }), // INSERT + DELETE: cancel out (Self::Insert { .. }, Self::Delete { .. }) => None, @@ -190,13 +213,8 @@ impl, B: Clone + Debug + PartialEq + A // UPDATE + UPDATE: keep original old, use final new (Self::Update { values: lhs, .. }, Self::Update { values: rhs, .. }) => { - let merged = lhs - .into_iter() - .zip(rhs) - .map(|((old, _mid), (_mid2, new))| (old, new)) - .collect(); Some(Self::Update { - values: merged, + values: merge_updates::(lhs, rhs), indirect, }) } @@ -262,18 +280,14 @@ impl, B: Clone + PartialEq + AsRef<[u8]>> core // INSERT + UPDATE: apply update to insert values ( - Self::Insert { mut values, .. }, + Self::Insert { values, .. }, Self::Update { values: updates, .. }, - ) => { - for (idx, ((), new)) in updates.into_iter().enumerate() { - if let Some(new_val) = new { - values[idx] = new_val; - } - } - Some(Self::Insert { values, indirect }) - } + ) => Some(Self::Insert { + values: apply_update::(values, updates), + indirect, + }), // INSERT + DELETE: cancel out (Self::Insert { .. }, Self::Delete { data: (), .. }) => None, @@ -286,13 +300,8 @@ impl, B: Clone + PartialEq + AsRef<[u8]>> core // UPDATE + UPDATE: keep original old (unit), use final new (Self::Update { values: lhs, .. }, Self::Update { values: rhs, .. }) => { - let merged = lhs - .into_iter() - .zip(rhs) - .map(|((old, _mid), (_mid2, new))| (old, new)) - .collect(); Some(Self::Update { - values: merged, + values: merge_updates::(lhs, rhs), indirect, }) } diff --git a/src/builders/sql/parser.rs b/src/builders/sql/parser.rs index 014e13112..2ded9b46c 100644 --- a/src/builders/sql/parser.rs +++ b/src/builders/sql/parser.rs @@ -128,11 +128,10 @@ pub(crate) struct Parser<'input, 'builder, T: SchemaWithPK, S> { /// A parsed operation waiting to be applied: the table, its primary key values, and the /// operation itself. Nothing is applied to the builder until every statement has parsed. -type PendingOp = ( - T, - Vec>>, - Operation>, -); +type PendingOp = (T, KeyValues, Operation>); + +/// Primary key values in primary key order. +type KeyValues = Vec>>; impl<'input, 'builder, T: NamedColumns, S: Clone + Hash + Eq + AsRef + for<'a> From<&'a str>> Parser<'input, 'builder, T, S> @@ -326,36 +325,9 @@ impl<'input, 'builder, T: NamedColumns, S: Clone + Hash + Eq + AsRef + for< self.lexer.next()?; } - if self.lexer.peek()?.kind != TokenKind::Where { - return Err(ParseError::MissingWhere { - statement: "UPDATE", - }); - } - - let n_pk = table.number_of_primary_keys(); - let mut pk = vec![Value::Null; n_pk]; - let mut pk_seen = vec![false; n_pk]; - - let can_match = self.digest_where(&table, |col_idx, col_name, val| { - if let Some(primary_key_index) = table.primary_key_index(usize::from(col_idx)) { - pk[primary_key_index] = val.clone(); - pk_seen[primary_key_index] = true; - Ok(()) - } else { - Err(ParseError::WhereNonPKColumn { column: col_name }) - } - })?; - - let found = pk_seen.iter().filter(|&&s| s).count(); - if found != n_pk { - return Err(ParseError::IncompleteWhereKey { - expected: n_pk, - found, - }); - } - if !can_match { + let Some(pk) = self.digest_key_where(&table, "UPDATE")? else { return Ok(()); - } + }; // A session records a key change as a DELETE plus a full-row INSERT, which SQL text cannot supply. if let Some(&(_, _, column)) = key_sets.iter().find(|&&(col_idx, primary_key_index, _)| { @@ -383,19 +355,42 @@ impl<'input, 'builder, T: NamedColumns, S: Clone + Hash + Eq + AsRef + for< let (table, _) = self.expect_table()?; + let Some(pks) = self.digest_key_where(&table, "DELETE")? else { + return Ok(()); + }; + + self.pending.push(( + table, + pks, + Operation::Delete { + data: (), + indirect: false, + }, + )); + + Ok(()) + } + + /// Parse the WHERE clause of `statement`, which must bind every primary key + /// column and nothing else, into the key values in primary key order. + /// + /// Returns `None` when no row can satisfy the clause. + fn digest_key_where( + &mut self, + table: &T, + statement: &'static str, + ) -> Result>, ParseError<'input>> { if self.lexer.peek()?.kind != TokenKind::Where { - return Err(ParseError::MissingWhere { - statement: "DELETE", - }); + return Err(ParseError::MissingWhere { statement }); } let n_pk = table.number_of_primary_keys(); - let mut pks = vec![Value::Null; n_pk]; + let mut pk = vec![Value::Null; n_pk]; let mut pk_seen = vec![false; n_pk]; - let can_match = self.digest_where(&table, |col_idx, col_name, val| { + let can_match = self.digest_where(table, |col_idx, col_name, val| { if let Some(primary_key_index) = table.primary_key_index(usize::from(col_idx)) { - pks[primary_key_index] = val.clone(); + pk[primary_key_index] = val; pk_seen[primary_key_index] = true; Ok(()) } else { @@ -410,20 +405,7 @@ impl<'input, 'builder, T: NamedColumns, S: Clone + Hash + Eq + AsRef + for< found, }); } - if !can_match { - return Ok(()); - } - - self.pending.push(( - table, - pks, - Operation::Delete { - data: (), - indirect: false, - }, - )); - - Ok(()) + Ok(can_match.then_some(pk)) } /// Parse a WHERE clause, calling `digestor` for each `col = val` predicate diff --git a/src/builders/sql_output.rs b/src/builders/sql_output.rs index e03c2e81b..2b28ee6eb 100644 --- a/src/builders/sql_output.rs +++ b/src/builders/sql_output.rs @@ -99,6 +99,48 @@ fn format_insert, B: AsRef<[u8]>>( sql } +/// Appends `column = value` for each `(column index, value)` cell, joined by `separator`. +fn push_cells<'a, T: ColumnNames, S: AsRef + 'a, B: AsRef<[u8]> + 'a>( + sql: &mut String, + table: &T, + separator: &str, + cells: impl Iterator)>, +) { + for (i, (col_idx, value)) in cells.enumerate() { + if i > 0 { + sql.push_str(separator); + } + if let Some(name) = table.column_name(col_idx) { + sql.push_str("e_identifier(name)); + } else { + write!(sql, "\"col{col_idx}\"").unwrap(); + } + sql.push_str(" = "); + write!(sql, "{value}").unwrap(); + } +} + +/// The primary-key cells of a row whose key values are `pk`, in key order. +fn pk_cells<'a, T: ColumnNames, S, B>( + table: &'a T, + pk: &'a [Value], +) -> impl Iterator)> { + table + .primary_key_columns() + .enumerate() + .map(move |(pk_ordinal, col_idx)| (col_idx, &pk[pk_ordinal])) +} + +/// The non-key cells an UPDATE sets, from each column's new value. +fn set_cells<'a, T: ColumnNames, S: 'a, B: 'a>( + table: &'a T, + news: impl Iterator>, +) -> impl Iterator)> { + news.enumerate() + .filter_map(|(col_idx, new)| new.as_ref().map(|value| (col_idx, value))) + .filter(move |(col_idx, _)| table.primary_key_index(*col_idx).is_none()) +} + /// Format a DELETE statement (changeset format - has all old values). fn format_delete_changeset, B: AsRef<[u8]>>( table: &T, @@ -107,24 +149,12 @@ fn format_delete_changeset, B: AsRef<[u8]>>( let mut sql = String::new(); write!(sql, "DELETE FROM {}", quote_identifier(table.name())).unwrap(); sql.push_str(" WHERE "); - // Use PK columns for the WHERE clause - let mut first = true; - for (col_idx, value) in values.iter().enumerate() { - if table.primary_key_index(col_idx).is_some() { - if !first { - sql.push_str(" AND "); - } - first = false; - if let Some(name) = table.column_name(col_idx) { - sql.push_str("e_identifier(name)); - } else { - write!(sql, "\"col{col_idx}\"").unwrap(); - } - sql.push_str(" = "); - write!(sql, "{value}").unwrap(); - } - } + let keys = values + .iter() + .enumerate() + .filter(|(col_idx, _)| table.primary_key_index(*col_idx).is_some()); + push_cells(&mut sql, table, " AND ", keys); sql } @@ -136,21 +166,7 @@ fn format_delete_patchset, B: AsRef<[u8]>>( let mut sql = String::new(); write!(sql, "DELETE FROM {}", quote_identifier(table.name())).unwrap(); sql.push_str(" WHERE "); - - let mut first = true; - for (pk_ordinal, col_idx) in table.primary_key_columns().enumerate() { - if !first { - sql.push_str(" AND "); - } - first = false; - if let Some(name) = table.column_name(col_idx) { - sql.push_str("e_identifier(name)); - } else { - write!(sql, "\"col{col_idx}\"").unwrap(); - } - sql.push_str(" = "); - write!(sql, "{}", pk[pk_ordinal]).unwrap(); - } + push_cells(&mut sql, table, " AND ", pk_cells(table, pk)); sql } @@ -162,49 +178,21 @@ fn format_update_changeset, B: AsRef<[u8]>>( let mut sql = String::new(); write!(sql, "UPDATE {}", quote_identifier(table.name())).unwrap(); sql.push_str(" SET "); - - // SET clause: columns that changed (have new values) - let mut first_set = true; - for (col_idx, (_old, new)) in pairs.iter().enumerate() { - if let Some(new_val) = new { - // Skip PK columns in SET (they go in WHERE) - if table.primary_key_index(col_idx).is_some() { - continue; - } - if !first_set { - sql.push_str(", "); - } - first_set = false; - if let Some(name) = table.column_name(col_idx) { - sql.push_str("e_identifier(name)); - } else { - write!(sql, "\"col{col_idx}\"").unwrap(); - } - sql.push_str(" = "); - write!(sql, "{new_val}").unwrap(); - } - } - + // SET clause: columns that changed (have new values), PK columns go in WHERE + push_cells( + &mut sql, + table, + ", ", + set_cells(table, pairs.iter().map(|(_old, new)| new)), + ); // WHERE clause: use old PK values sql.push_str(" WHERE "); - let mut first_where = true; - for (col_idx, (old, _new)) in pairs.iter().enumerate() { - if table.primary_key_index(col_idx).is_some() { - if let Some(old_val) = old { - if !first_where { - sql.push_str(" AND "); - } - first_where = false; - if let Some(name) = table.column_name(col_idx) { - sql.push_str("e_identifier(name)); - } else { - write!(sql, "\"col{col_idx}\"").unwrap(); - } - sql.push_str(" = "); - write!(sql, "{old_val}").unwrap(); - } - } - } + let old_keys = pairs + .iter() + .enumerate() + .filter(|(col_idx, _)| table.primary_key_index(*col_idx).is_some()) + .filter_map(|(col_idx, (old, _new))| old.as_ref().map(|value| (col_idx, value))); + push_cells(&mut sql, table, " AND ", old_keys); sql } @@ -217,44 +205,15 @@ fn format_update_patchset, B: AsRef<[u8]>>( let mut sql = String::new(); write!(sql, "UPDATE {}", quote_identifier(table.name())).unwrap(); sql.push_str(" SET "); - // SET clause: columns that changed (have new values), excluding PK - let mut first_set = true; - for (col_idx, ((), new)) in pairs.iter().enumerate() { - if let Some(new_val) = new { - // Skip PK columns in SET - if table.primary_key_index(col_idx).is_some() { - continue; - } - if !first_set { - sql.push_str(", "); - } - first_set = false; - if let Some(name) = table.column_name(col_idx) { - sql.push_str("e_identifier(name)); - } else { - write!(sql, "\"col{col_idx}\"").unwrap(); - } - sql.push_str(" = "); - write!(sql, "{new_val}").unwrap(); - } - } - + push_cells( + &mut sql, + table, + ", ", + set_cells(table, pairs.iter().map(|((), new)| new)), + ); sql.push_str(" WHERE "); - let mut first_where = true; - for (pk_ordinal, col_idx) in table.primary_key_columns().enumerate() { - if !first_where { - sql.push_str(" AND "); - } - first_where = false; - if let Some(name) = table.column_name(col_idx) { - sql.push_str("e_identifier(name)); - } else { - write!(sql, "\"col{col_idx}\"").unwrap(); - } - sql.push_str(" = "); - write!(sql, "{}", pk[pk_ordinal]).unwrap(); - } + push_cells(&mut sql, table, " AND ", pk_cells(table, pk)); sql } diff --git a/src/builders/view.rs b/src/builders/view.rs index f6537be67..01f1e3be7 100644 --- a/src/builders/view.rs +++ b/src/builders/view.rs @@ -15,7 +15,10 @@ //! [`DiffSet::iter`]: super::DiffSet::iter use alloc::vec::Vec; +use core::fmt::Debug; +use super::format::{ChangesetFormat, PatchsetFormat}; +use super::operation::Operation; use crate::encoding::Value; use crate::schema::SchemaWithPK; @@ -115,6 +118,31 @@ impl<'a, T, S, B> ChangesetOp<'a, T, S, B> { } } +impl<'a, T, S: Clone + Debug + AsRef, B: Clone + Debug + AsRef<[u8]>> + ChangesetOp<'a, T, S, B> +{ + /// The view of `op`, an operation stored for `table`. + pub(crate) fn of_stored(table: &'a T, op: &'a Operation) -> Self { + match op { + Operation::Insert { values, indirect } => Self::Insert { + table, + values: values.as_slice(), + indirect: *indirect, + }, + Operation::Update { values, indirect } => Self::Update { + table, + values: values.as_slice(), + indirect: *indirect, + }, + Operation::Delete { data, indirect } => Self::Delete { + table, + old_values: data.as_slice(), + indirect: *indirect, + }, + } + } +} + impl ChangesetOp<'_, T, S, B> where Value: PartialEq, @@ -230,6 +258,34 @@ impl<'a, T, S, B> PatchsetOp<'a, T, S, B> { } } +impl<'a, T, S, B> PatchsetOp<'a, T, S, B> { + /// The view of `op`, an operation stored for `table` under the primary key `pk`. + pub(crate) fn of_stored( + table: &'a T, + pk: &'a [Value], + op: &'a Operation, + ) -> Self { + match op { + Operation::Insert { values, indirect } => Self::Insert { + table, + values: values.as_slice(), + indirect: *indirect, + }, + Operation::Update { values, indirect } => Self::Update { + table, + pk, + entries: values.as_slice(), + indirect: *indirect, + }, + Operation::Delete { indirect, .. } => Self::Delete { + table, + pk, + indirect: *indirect, + }, + } + } +} + impl PatchsetOp<'_, T, S, B> { /// Returns the primary-key cells of this operation, in key order. /// diff --git a/src/maxwell.rs b/src/maxwell.rs index e6a9be39b..490df15ed 100644 --- a/src/maxwell.rs +++ b/src/maxwell.rs @@ -48,7 +48,7 @@ use crate::wire::{ build_patch_delete, build_patchset_update, resolve_table, }; -use crate::builders::{DiffOps, DiffSetBuilder, PatchsetFormat}; +use crate::builders::{DeleteArg, DiffOps, DiffSetBuilder, Format, PatchsetFormat}; use crate::wire::WireAdapter; use core::fmt::Debug; use core::hash::Hash; @@ -145,31 +145,14 @@ where Sch: WireSchema, A: WireAdapter, { - match self { - Message::Insert(row) | Message::BootstrapInsert(row) => { - let table = resolve_table(schema, Some(row.database.as_str()), row.table.as_str())?; - let insert = build_insert_from_maxwell(&row.data, table, adapter)?; - Ok(DiffOps::insert(builder, insert)) - } - Message::Update(row) => { - let table = resolve_table(schema, Some(row.database.as_str()), row.table.as_str())?; - let update = build_changeset_update_from_maxwell( - &row.data, - row.old.as_ref(), - table, - adapter, - )?; - Ok(DiffOps::update(builder, update)) - } - Message::Delete(row) => { - let table = resolve_table(schema, Some(row.database.as_str()), row.table.as_str())?; - let delete = build_changeset_delete_from_maxwell(&row.data, table, adapter)?; - Ok(DiffOps::delete(builder, delete)) - } - // Bootstrap control events, DDL, and anything a newer Maxwell adds carry no - // row data to digest. `Message` is non-exhaustive, so this cannot enumerate. - _ => Ok(builder), - } + digest( + self, + builder, + schema, + adapter, + build_changeset_update_from_maxwell, + build_changeset_delete_from_maxwell, + ) } } @@ -192,26 +175,63 @@ where Sch: WireSchema
, A: WireAdapter, { - match self { - Message::Insert(row) | Message::BootstrapInsert(row) => { - let table = resolve_table(schema, Some(row.database.as_str()), row.table.as_str())?; - let insert = build_insert_from_maxwell(&row.data, table, adapter)?; - Ok(DiffOps::insert(builder, insert)) - } - Message::Update(row) => { - let table = resolve_table(schema, Some(row.database.as_str()), row.table.as_str())?; - let update = build_patchset_update_from_maxwell(&row.data, table, adapter)?; - Ok(DiffOps::update(builder, update)) - } - Message::Delete(row) => { - let table = resolve_table(schema, Some(row.database.as_str()), row.table.as_str())?; - let delete = build_patch_delete_from_maxwell(&row.data, table, adapter)?; - Ok(DiffOps::delete(builder, delete)) - } - // Bootstrap control events, DDL, and anything a newer Maxwell adds carry no - // row data to digest. `Message` is non-exhaustive, so this cannot enumerate. - _ => Ok(builder), + digest( + self, + builder, + schema, + adapter, + |data, _old, table, adapter| build_patchset_update_from_maxwell(data, table, adapter), + build_patch_delete_from_maxwell, + ) + } +} + +/// One Maxwell row image, column name to JSON value. +type RowImage = serde_json::Map; + +/// Digests one Maxwell message into `builder`, building its UPDATE and DELETE rows with +/// the format's `update` and `delete`. +fn digest( + message: &Message, + builder: DiffSetBuilder, + schema: &Sch, + adapter: &A, + update: impl FnOnce( + &RowImage, + Option<&RowImage>, + &T, + &A, + ) -> Result, ConversionError>, + delete: impl FnOnce(&RowImage, &T, &A) -> Result, ConversionError>, +) -> Result, ConversionError> +where + F: Format, + T: NamedColumns + WireColumnTypes, + S: Clone + Debug + Hash + Eq + AsRef + Default, + B: Clone + Debug + Hash + Eq + AsRef<[u8]> + Default, + Sch: WireSchema
, + A: WireAdapter, + DiffSetBuilder: DiffOps, +{ + match message { + Message::Insert(row) | Message::BootstrapInsert(row) => { + let table = resolve_table(schema, Some(row.database.as_str()), row.table.as_str())?; + let insert = build_insert_from_maxwell(&row.data, table, adapter)?; + Ok(DiffOps::insert(builder, insert)) + } + Message::Update(row) => { + let table = resolve_table(schema, Some(row.database.as_str()), row.table.as_str())?; + let update = update(&row.data, row.old.as_ref(), table, adapter)?; + Ok(DiffOps::update(builder, update)) + } + Message::Delete(row) => { + let table = resolve_table(schema, Some(row.database.as_str()), row.table.as_str())?; + let delete = delete(&row.data, table, adapter)?; + Ok(DiffOps::delete(builder, delete)) } + // Bootstrap control events, DDL, and anything a newer Maxwell adds carry no + // row data to digest. `Message` is non-exhaustive, so this cannot enumerate. + _ => Ok(builder), } } diff --git a/src/parser.rs b/src/parser.rs index efdec707a..18b272d4f 100644 --- a/src/parser.rs +++ b/src/parser.rs @@ -33,7 +33,9 @@ type UpdateValues = Vec<(MaybeValue>, MaybeValue /// Type alias for parsed values result. type ParsedValues = (Vec>>, usize); -use crate::builders::{ChangesetFormat, DiffSet, DiffSetBuilder, Operation, PatchsetFormat}; +use crate::builders::{ + ChangesetFormat, DiffSet, DiffSetBuilder, Format, Operation, PatchsetFormat, +}; use crate::encoding::varint::decode_varint; use crate::encoding::{MaybeValue, Value, decode_value, markers, op_codes}; use crate::schema::{DynTable, SchemaWithPK}; @@ -372,15 +374,31 @@ impl ParsedDiffSet { fn parse_as_changeset( data: &[u8], ) -> Result, String, Vec>, ParseError> { - let mut builder: DiffSetBuilder, String, Vec> = - DiffSetBuilder::new(); + parse_as(data, FormatMarker::Changeset, parse_changeset_operation) +} + +/// Parse every table of `data`, which must all be in the `expected` format, adding +/// each operation with `parse_operation`. +/// +/// # Errors +/// +/// Returns a `ParseError` if the data is malformed or mixes formats. +fn parse_as>>( + data: &[u8], + expected: FormatMarker, + parse_operation: ParseOperation, +) -> Result, ParseError> +where + ParsedSet: From>, +{ + let mut builder = DiffSetBuilder::new(); let mut pos = 0; while pos < data.len() { let (schema, format, header_len) = parse_table_header(&data[pos..], pos)?; - if format != FormatMarker::Changeset { + if format != expected { return Err(ParseError::MixedFormats { - expected: FormatMarker::Changeset, + expected, found: format, position: pos, }); @@ -392,7 +410,7 @@ fn parse_as_changeset( if byte == markers::CHANGESET || byte == markers::PATCHSET { break; } - let op_len = parse_changeset_operation(&data[pos..], pos, &schema, &mut builder)?; + let op_len = parse_operation(&data[pos..], pos, &schema, &mut builder)?; pos += op_len; } } @@ -408,32 +426,7 @@ fn parse_as_changeset( fn parse_as_patchset( data: &[u8], ) -> Result, String, Vec>, ParseError> { - let mut builder: DiffSetBuilder, String, Vec> = - DiffSetBuilder::new(); - let mut pos = 0; - - while pos < data.len() { - let (schema, format, header_len) = parse_table_header(&data[pos..], pos)?; - if format != FormatMarker::Patchset { - return Err(ParseError::MixedFormats { - expected: FormatMarker::Patchset, - found: format, - position: pos, - }); - } - pos += header_len; - - while pos < data.len() { - let byte = data[pos]; - if byte == markers::CHANGESET || byte == markers::PATCHSET { - break; - } - let op_len = parse_patchset_operation(&data[pos..], pos, &schema, &mut builder)?; - pos += op_len; - } - } - - Ok(builder.into()) + parse_as(data, FormatMarker::Patchset, parse_patchset_operation) } /// Parse a table header and return the schema. @@ -528,6 +521,48 @@ fn parse_operation_header(data: &[u8], base_pos: usize) -> Result<(u8, bool, usi Ok((data[0], data[1] != 0, 2)) } +/// A builder of parsed tables in format `F`. +type ParsedBuilder = DiffSetBuilder, String, Vec>; + +/// A diff set of parsed tables in format `F`. +type ParsedSet = DiffSet, String, Vec>; + +/// Parses one operation at a byte offset into the builder, returning its length. +type ParseOperation = + fn(&[u8], usize, &TableSchema, &mut ParsedBuilder) -> Result; + +/// A full row image and the number of bytes it took. +type ParsedRow = (Vec>>, usize); + +/// Parse `count` values with undefined ones read as NULL, a full row image. +fn parse_row(data: &[u8], base_pos: usize, count: usize) -> Result { + let (values, len) = parse_values(data, base_pos, count)?; + Ok(( + values + .into_iter() + .map(|v| v.unwrap_or(Value::Null)) + .collect(), + len, + )) +} + +/// Parse an INSERT body, which both formats store as a full row, into `builder`. +fn parse_insert>>( + data: &[u8], + base_pos: usize, + schema: &TableSchema, + builder: &mut ParsedBuilder, + indirect: bool, +) -> Result +where + Operation>: core::ops::Add>>>, +{ + let (values, len) = parse_row(data, base_pos, schema.column_count)?; + let pk = schema.extract_pk(&values); + builder.add_operation(schema, pk, Operation::Insert { values, indirect }); + Ok(len) +} + /// Parse a changeset operation. fn parse_changeset_operation( data: &[u8], @@ -539,22 +574,11 @@ fn parse_changeset_operation( match op_code { op_codes::INSERT => { - let (values, len) = parse_values(&data[pos..], base_pos + pos, schema.column_count)?; - pos += len; - let values: Vec>> = values - .into_iter() - .map(|v| v.unwrap_or(Value::Null)) - .collect(); - let pk = schema.extract_pk(&values); - builder.add_operation(schema, pk, Operation::Insert { values, indirect }); + pos += parse_insert(&data[pos..], base_pos + pos, schema, builder, indirect)?; } op_codes::DELETE => { - let (values, len) = parse_values(&data[pos..], base_pos + pos, schema.column_count)?; + let (values, len) = parse_row(&data[pos..], base_pos + pos, schema.column_count)?; pos += len; - let values: Vec>> = values - .into_iter() - .map(|v| v.unwrap_or(Value::Null)) - .collect(); let pk = schema.extract_pk(&values); builder.add_operation( schema, @@ -598,14 +622,7 @@ fn parse_patchset_operation( match op_code { op_codes::INSERT => { - let (values, len) = parse_values(&data[pos..], base_pos + pos, schema.column_count)?; - pos += len; - let values: Vec>> = values - .into_iter() - .map(|v| v.unwrap_or(Value::Null)) - .collect(); - let pk = schema.extract_pk(&values); - builder.add_operation(schema, pk, Operation::Insert { values, indirect }); + pos += parse_insert(&data[pos..], base_pos + pos, schema, builder, indirect)?; } op_codes::DELETE => { // Patchset DELETE: only PK values in column order diff --git a/src/pg_walstream.rs b/src/pg_walstream.rs index 3c94dad9c..143699d6e 100644 --- a/src/pg_walstream.rs +++ b/src/pg_walstream.rs @@ -11,7 +11,8 @@ pub use pg_walstream::{ChangeEvent, ColumnValue, EventType, Lsn, ReplicaIdentity use crate::ChangesetFormat; use crate::builders::{ - ChangeDelete, DiffOps, DiffSetBuilder, Insert, PatchDelete, PatchsetFormat, Update, + ChangeDelete, DeleteArg, DiffOps, DiffSetBuilder, Format, Insert, PatchDelete, PatchsetFormat, + Update, }; use crate::schema::NamedColumns; use crate::wire::{ @@ -118,41 +119,14 @@ where Sch: WireSchema
, A: WireAdapter, { - match self { - EventType::Insert { - schema: source_schema, - table: name, - data, - .. - } => { - let table = resolve_table(schema, Some(source_schema.as_ref()), name.as_ref())?; - let insert = build_insert_from_pg(data, table, adapter)?; - Ok(DiffOps::insert(builder, insert)) - } - EventType::Update { - schema: source_schema, - table: name, - old_data, - new_data, - .. - } => { - let table = resolve_table(schema, Some(source_schema.as_ref()), name.as_ref())?; - let update = - build_changeset_update_from_pg(old_data.as_ref(), new_data, table, adapter)?; - Ok(DiffOps::update(builder, update)) - } - EventType::Delete { - schema: source_schema, - table: name, - old_data, - .. - } => { - let table = resolve_table(schema, Some(source_schema.as_ref()), name.as_ref())?; - let delete = build_changeset_delete_from_pg(old_data, table, adapter)?; - Ok(DiffOps::delete(builder, delete)) - } - _ => Ok(builder), - } + digest( + self, + builder, + schema, + adapter, + build_changeset_update_from_pg, + build_changeset_delete_from_pg, + ) } } @@ -175,39 +149,76 @@ where Sch: WireSchema
, A: WireAdapter, { - match self { - EventType::Insert { - schema: source_schema, - table: name, - data, - .. - } => { - let table = resolve_table(schema, Some(source_schema.as_ref()), name.as_ref())?; - let insert = build_insert_from_pg(data, table, adapter)?; - Ok(DiffOps::insert(builder, insert)) - } - EventType::Update { - schema: source_schema, - table: name, - new_data, - .. - } => { - let table = resolve_table(schema, Some(source_schema.as_ref()), name.as_ref())?; - let update = build_patchset_update_from_pg(new_data, table, adapter)?; - Ok(DiffOps::update(builder, update)) - } - EventType::Delete { - schema: source_schema, - table: name, - old_data, - .. - } => { - let table = resolve_table(schema, Some(source_schema.as_ref()), name.as_ref())?; - let delete = build_patch_delete_from_pg(old_data, table, adapter)?; - Ok(DiffOps::delete(builder, delete)) - } - _ => Ok(builder), + digest( + self, + builder, + schema, + adapter, + |_old, new_data, table, adapter| { + build_patchset_update_from_pg(new_data, table, adapter) + }, + build_patch_delete_from_pg, + ) + } +} + +/// Digests one `pg_walstream` event into `builder`, building its UPDATE and DELETE rows +/// with the format's `update` and `delete`. +fn digest( + event: &EventType, + builder: DiffSetBuilder, + schema: &Sch, + adapter: &A, + update: impl FnOnce( + Option<&RowData>, + &RowData, + &T, + &A, + ) -> Result, ConversionError>, + delete: impl FnOnce(&RowData, &T, &A) -> Result, ConversionError>, +) -> Result, ConversionError> +where + F: Format, + T: NamedColumns + WireColumnTypes, + S: Clone + Debug + Hash + Eq + AsRef + Default, + B: Clone + Debug + Hash + Eq + AsRef<[u8]> + Default, + Sch: WireSchema
, + A: WireAdapter, + DiffSetBuilder: DiffOps, +{ + match event { + EventType::Insert { + schema: source_schema, + table: name, + data, + .. + } => { + let table = resolve_table(schema, Some(source_schema.as_ref()), name.as_ref())?; + let insert = build_insert_from_pg(data, table, adapter)?; + Ok(DiffOps::insert(builder, insert)) + } + EventType::Update { + schema: source_schema, + table: name, + old_data, + new_data, + .. + } => { + let table = resolve_table(schema, Some(source_schema.as_ref()), name.as_ref())?; + let update = update(old_data.as_ref(), new_data, table, adapter)?; + Ok(DiffOps::update(builder, update)) + } + EventType::Delete { + schema: source_schema, + table: name, + old_data, + .. + } => { + let table = resolve_table(schema, Some(source_schema.as_ref()), name.as_ref())?; + let delete = delete(old_data, table, adapter)?; + Ok(DiffOps::delete(builder, delete)) } + _ => Ok(builder), } } diff --git a/src/wal2json.rs b/src/wal2json.rs index b03b548fb..e5af12fbd 100644 --- a/src/wal2json.rs +++ b/src/wal2json.rs @@ -108,8 +108,8 @@ impl Wal2JsonColumn<'_> { } use crate::builders::{ - ChangeDelete, ChangesetFormat, DiffOps, DiffSetBuilder, Insert, PatchDelete, PatchsetFormat, - Update, + ChangeDelete, ChangesetFormat, DeleteArg, DiffOps, DiffSetBuilder, Format, Insert, PatchDelete, + PatchsetFormat, Update, }; use crate::encoding::Value; use crate::schema::NamedColumns; @@ -136,44 +136,14 @@ where Sch: WireSchema
, A: WireAdapter, { - match self { - MessageV2::Insert(row) => { - let table = resolve_table(schema, row.schema.as_deref(), row.table.as_str())?; - let columns = row - .columns - .as_deref() - .ok_or(ConversionError::MissingColumns)?; - let insert = build_insert_from_v2(columns, table, adapter)?; - Ok(DiffOps::insert(builder, insert)) - } - MessageV2::Update(row) => { - let table = resolve_table(schema, row.schema.as_deref(), row.table.as_str())?; - let columns = row - .columns - .as_deref() - .ok_or(ConversionError::MissingColumns)?; - let update = build_changeset_update_from_v2( - columns, - row.identity.as_deref(), - table, - adapter, - )?; - Ok(DiffOps::update(builder, update)) - } - MessageV2::Delete(row) => { - let table = resolve_table(schema, row.schema.as_deref(), row.table.as_str())?; - let identity = row - .identity - .as_deref() - .ok_or(ConversionError::MissingColumns)?; - let delete = build_changeset_delete_from_columns(identity, table, adapter)?; - Ok(DiffOps::delete(builder, delete)) - } - MessageV2::Begin(_) - | MessageV2::Commit(_) - | MessageV2::Truncate(_) - | MessageV2::Message(_) => Ok(builder), - } + digest_v2( + self, + builder, + schema, + adapter, + build_changeset_update_from_v2, + build_changeset_delete_from_columns, + ) } } @@ -196,39 +166,16 @@ where Sch: WireSchema
, A: WireAdapter, { - match self { - MessageV2::Insert(row) => { - let table = resolve_table(schema, row.schema.as_deref(), row.table.as_str())?; - let columns = row - .columns - .as_deref() - .ok_or(ConversionError::MissingColumns)?; - let insert = build_insert_from_v2(columns, table, adapter)?; - Ok(DiffOps::insert(builder, insert)) - } - MessageV2::Update(row) => { - let table = resolve_table(schema, row.schema.as_deref(), row.table.as_str())?; - let columns = row - .columns - .as_deref() - .ok_or(ConversionError::MissingColumns)?; - let update = build_patchset_update_from_v2(columns, table, adapter)?; - Ok(DiffOps::update(builder, update)) - } - MessageV2::Delete(row) => { - let table = resolve_table(schema, row.schema.as_deref(), row.table.as_str())?; - let identity = row - .identity - .as_deref() - .ok_or(ConversionError::MissingColumns)?; - let delete = build_patch_delete_from_columns(identity, table, adapter)?; - Ok(DiffOps::delete(builder, delete)) - } - MessageV2::Begin(_) - | MessageV2::Commit(_) - | MessageV2::Truncate(_) - | MessageV2::Message(_) => Ok(builder), - } + digest_v2( + self, + builder, + schema, + adapter, + |columns, _identity, table, adapter| { + build_patchset_update_from_v2(columns, table, adapter) + }, + build_patch_delete_from_columns, + ) } } @@ -251,40 +198,14 @@ where Sch: WireSchema
, A: WireAdapter, { - match self { - ChangeV1::Insert { - schema: source_schema, - table, - columns, - .. - } => { - let table = resolve_table(schema, source_schema.as_deref(), table.as_str())?; - let insert = build_insert_from_v1(columns, table, adapter)?; - Ok(DiffOps::insert(builder, insert)) - } - ChangeV1::Update { - schema: source_schema, - table, - columns, - oldkeys, - .. - } => { - let table = resolve_table(schema, source_schema.as_deref(), table.as_str())?; - let update = build_changeset_update_from_v1(columns, oldkeys, table, adapter)?; - Ok(DiffOps::update(builder, update)) - } - ChangeV1::Delete { - schema: source_schema, - table, - oldkeys, - .. - } => { - let table = resolve_table(schema, source_schema.as_deref(), table.as_str())?; - let delete = build_changeset_delete_from_v1(oldkeys, table, adapter)?; - Ok(DiffOps::delete(builder, delete)) - } - ChangeV1::Message { .. } => Ok(builder), - } + digest_v1( + self, + builder, + schema, + adapter, + build_changeset_update_from_v1, + build_changeset_delete_from_v1, + ) } } @@ -307,39 +228,130 @@ where Sch: WireSchema
, A: WireAdapter, { - match self { - ChangeV1::Insert { - schema: source_schema, - table, - columns, - .. - } => { - let table = resolve_table(schema, source_schema.as_deref(), table.as_str())?; - let insert = build_insert_from_v1(columns, table, adapter)?; - Ok(DiffOps::insert(builder, insert)) - } - ChangeV1::Update { - schema: source_schema, - table, - columns, - .. - } => { - let table = resolve_table(schema, source_schema.as_deref(), table.as_str())?; - let update = build_patchset_update_from_v1(columns, table, adapter)?; - Ok(DiffOps::update(builder, update)) - } - ChangeV1::Delete { - schema: source_schema, - table, - oldkeys, - .. - } => { - let table = resolve_table(schema, source_schema.as_deref(), table.as_str())?; - let delete = build_patch_delete_from_v1(oldkeys, table, adapter)?; - Ok(DiffOps::delete(builder, delete)) - } - ChangeV1::Message { .. } => Ok(builder), + digest_v1( + self, + builder, + schema, + adapter, + |columns, _oldkeys, table, adapter| { + build_patchset_update_from_v1(columns, table, adapter) + }, + build_patch_delete_from_v1, + ) + } +} + +/// Digests one wal2json v2 message into `builder`, building its UPDATE and DELETE rows +/// with the format's `update` and `delete`. +fn digest_v2( + message: &MessageV2, + builder: DiffSetBuilder, + schema: &Sch, + adapter: &A, + update: impl FnOnce( + &[Column], + Option<&[Column]>, + &T, + &A, + ) -> Result, ConversionError>, + delete: impl FnOnce(&[Column], &T, &A) -> Result, ConversionError>, +) -> Result, ConversionError> +where + F: Format, + T: NamedColumns + WireColumnTypes, + S: Clone + Debug + Hash + Eq + AsRef + Default, + B: Clone + Debug + Hash + Eq + AsRef<[u8]> + Default, + Sch: WireSchema
, + A: WireAdapter, + DiffSetBuilder: DiffOps, +{ + match message { + MessageV2::Insert(row) => { + let table = resolve_table(schema, row.schema.as_deref(), row.table.as_str())?; + let columns = row + .columns + .as_deref() + .ok_or(ConversionError::MissingColumns)?; + let insert = build_insert_from_v2(columns, table, adapter)?; + Ok(DiffOps::insert(builder, insert)) + } + MessageV2::Update(row) => { + let table = resolve_table(schema, row.schema.as_deref(), row.table.as_str())?; + let columns = row + .columns + .as_deref() + .ok_or(ConversionError::MissingColumns)?; + let update = update(columns, row.identity.as_deref(), table, adapter)?; + Ok(DiffOps::update(builder, update)) + } + MessageV2::Delete(row) => { + let table = resolve_table(schema, row.schema.as_deref(), row.table.as_str())?; + let identity = row + .identity + .as_deref() + .ok_or(ConversionError::MissingColumns)?; + let delete = delete(identity, table, adapter)?; + Ok(DiffOps::delete(builder, delete)) + } + MessageV2::Begin(_) + | MessageV2::Commit(_) + | MessageV2::Truncate(_) + | MessageV2::Message(_) => Ok(builder), + } +} + +/// Digests one wal2json v1 change into `builder`, building its UPDATE and DELETE rows +/// with the format's `update` and `delete`. +fn digest_v1( + change: &ChangeV1, + builder: DiffSetBuilder, + schema: &Sch, + adapter: &A, + update: impl FnOnce(&ColumnArrays, &OldKeys, &T, &A) -> Result, ConversionError>, + delete: impl FnOnce(&OldKeys, &T, &A) -> Result, ConversionError>, +) -> Result, ConversionError> +where + F: Format, + T: NamedColumns + WireColumnTypes, + S: Clone + Debug + Hash + Eq + AsRef + Default, + B: Clone + Debug + Hash + Eq + AsRef<[u8]> + Default, + Sch: WireSchema
, + A: WireAdapter, + DiffSetBuilder: DiffOps, +{ + match change { + ChangeV1::Insert { + schema: source_schema, + table, + columns, + .. + } => { + let table = resolve_table(schema, source_schema.as_deref(), table.as_str())?; + let insert = build_insert_from_v1(columns, table, adapter)?; + Ok(DiffOps::insert(builder, insert)) + } + ChangeV1::Update { + schema: source_schema, + table, + columns, + oldkeys, + .. + } => { + let table = resolve_table(schema, source_schema.as_deref(), table.as_str())?; + let update = update(columns, oldkeys, table, adapter)?; + Ok(DiffOps::update(builder, update)) + } + ChangeV1::Delete { + schema: source_schema, + table, + oldkeys, + .. + } => { + let table = resolve_table(schema, source_schema.as_deref(), table.as_str())?; + let delete = delete(oldkeys, table, adapter)?; + Ok(DiffOps::delete(builder, delete)) } + ChangeV1::Message { .. } => Ok(builder), } } diff --git a/src/wire/bytes_helpers.rs b/src/wire/bytes_helpers.rs index 6f5e0f0fc..48c0f3ca2 100644 --- a/src/wire/bytes_helpers.rs +++ b/src/wire/bytes_helpers.rs @@ -7,6 +7,9 @@ use alloc::vec::Vec; +#[cfg(any(feature = "wal2json", feature = "pg-walstream"))] +use super::uuid_helpers::hex_nibble; + /// Decode a PG `\xHEX` escape (produced by `\x` prefix + even-length /// lowercase or uppercase hex) into raw bytes. /// @@ -61,17 +64,6 @@ fn decode_hex(hex: &[u8]) -> Result, usize> { Ok(out) } -#[cfg(any(feature = "wal2json", feature = "pg-walstream"))] -#[inline] -fn hex_nibble(b: u8) -> Option { - match b { - b'0'..=b'9' => Some(b - b'0'), - b'a'..=b'f' => Some(b - b'a' + 10), - b'A'..=b'F' => Some(b - b'A' + 10), - _ => None, - } -} - /// Decode a standard base64 (RFC 4648) string into raw bytes. /// /// Accepts padded standard base64 (the `+`, `/` alphabet with `=` diff --git a/src/wire/impls_maxwell.rs b/src/wire/impls_maxwell.rs index 1642ea293..84fcf6ce2 100644 --- a/src/wire/impls_maxwell.rs +++ b/src/wire/impls_maxwell.rs @@ -10,7 +10,6 @@ use super::decoder::{ TimestampDecoder, TimestampTzDecoder, UuidBlob16Decoder, UuidText36Decoder, }; use super::error::DecodeError; -use super::scalar_helpers::normalize_real; use super::temporal::{Temporal, decode_temporal_text, decode_utc_timestamptz_text}; use super::type_map::{TypeMap, TypeMapDefaults}; use super::wire_type::WireType; @@ -77,33 +76,7 @@ impl Decoder for BoolDecoder { impl Decoder for IntDecoder { fn decode(&self, payload: MaxwellColumn<'_>) -> Result, DecodeError> { - match payload.value { - serde_json::Value::Null => Ok(Value::Null), - serde_json::Value::Number(n) => match n.as_i64() { - Some(i) => Ok(Value::Integer(i)), - None => Err(DecodeError::IntegerOverflow { - column: payload.column_name.to_string(), - digits: n.to_string(), - }), - }, - serde_json::Value::Bool(_) => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "JSON integer number", - actual: "JSON boolean", - }), - serde_json::Value::String(_) => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "JSON integer number", - actual: "JSON string", - }), - serde_json::Value::Array(_) | serde_json::Value::Object(_) => { - Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "JSON integer number", - actual: "JSON array or object", - }) - } - } + super::json_decoders::decode_json_int(&payload) } } @@ -119,28 +92,7 @@ where S: From, { fn decode(&self, payload: MaxwellColumn<'_>) -> Result, DecodeError> { - match payload.value { - serde_json::Value::Null => Ok(Value::Null), - serde_json::Value::Number(n) => match n.as_i64() { - Some(i) => Ok(Value::Integer(i)), - None => Ok(Value::Text(S::from(n.to_string()))), - }, - serde_json::Value::String(s) - if s.trim_start_matches('-') - .chars() - .all(|c| c.is_ascii_digit()) => - { - match s.parse::() { - Ok(i) => Ok(Value::Integer(i)), - Err(_) => Ok(Value::Text(S::from(s.clone()))), - } - } - _ => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "JSON integer number or numeric string", - actual: "other JSON shape", - }), - } + super::json_decoders::decode_json_int64_or_text(&payload) } } @@ -150,34 +102,7 @@ where impl Decoder for RealDecoder { fn decode(&self, payload: MaxwellColumn<'_>) -> Result, DecodeError> { - match payload.value { - serde_json::Value::Null => Ok(Value::Null), - serde_json::Value::Number(n) => match n.as_f64() { - Some(f) => Ok(normalize_real(f)), - None => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "IEEE 754 float number", - actual: "arbitrary-precision JSON number", - }), - }, - serde_json::Value::Bool(_) => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "IEEE 754 float number", - actual: "JSON boolean", - }), - serde_json::Value::String(_) => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "IEEE 754 float number", - actual: "JSON string", - }), - serde_json::Value::Array(_) | serde_json::Value::Object(_) => { - Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "IEEE 754 float number", - actual: "JSON array or object", - }) - } - } + super::json_decoders::decode_json_real(&payload) } } diff --git a/src/wire/impls_wal2json.rs b/src/wire/impls_wal2json.rs index 74d0057a6..ffc98159c 100644 --- a/src/wire/impls_wal2json.rs +++ b/src/wire/impls_wal2json.rs @@ -10,7 +10,6 @@ use super::decoder::{ TimestampDecoder, TimestampTzDecoder, UuidBlob16Decoder, UuidText36Decoder, }; use super::error::DecodeError; -use super::scalar_helpers::normalize_real; use super::temporal::{Temporal, decode_temporal_text}; use super::type_map::{TypeMap, TypeMapDefaults}; use super::wire_type::WireType; @@ -76,33 +75,7 @@ impl Decoder for BoolDecoder { impl Decoder for IntDecoder { fn decode(&self, payload: Wal2JsonColumn<'_>) -> Result, DecodeError> { - match payload.value { - serde_json::Value::Null => Ok(Value::Null), - serde_json::Value::Number(n) => match n.as_i64() { - Some(i) => Ok(Value::Integer(i)), - None => Err(DecodeError::IntegerOverflow { - column: payload.column_name.to_string(), - digits: n.to_string(), - }), - }, - serde_json::Value::Bool(_) => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "JSON integer number", - actual: "JSON boolean", - }), - serde_json::Value::String(_) => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "JSON integer number", - actual: "JSON string", - }), - serde_json::Value::Array(_) | serde_json::Value::Object(_) => { - Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "JSON integer number", - actual: "JSON array or object", - }) - } - } + super::json_decoders::decode_json_int(&payload) } } @@ -115,28 +88,7 @@ where S: From, { fn decode(&self, payload: Wal2JsonColumn<'_>) -> Result, DecodeError> { - match payload.value { - serde_json::Value::Null => Ok(Value::Null), - serde_json::Value::Number(n) => match n.as_i64() { - Some(i) => Ok(Value::Integer(i)), - None => Ok(Value::Text(S::from(n.to_string()))), - }, - serde_json::Value::String(s) - if s.trim_start_matches('-') - .chars() - .all(|c| c.is_ascii_digit()) => - { - match s.parse::() { - Ok(i) => Ok(Value::Integer(i)), - Err(_) => Ok(Value::Text(S::from(s.clone()))), - } - } - _ => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "JSON integer number or numeric string", - actual: "other JSON shape", - }), - } + super::json_decoders::decode_json_int64_or_text(&payload) } } @@ -148,34 +100,7 @@ where impl Decoder for RealDecoder { fn decode(&self, payload: Wal2JsonColumn<'_>) -> Result, DecodeError> { - match payload.value { - serde_json::Value::Null => Ok(Value::Null), - serde_json::Value::Number(n) => match n.as_f64() { - Some(f) => Ok(normalize_real(f)), - None => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "IEEE 754 float number", - actual: "arbitrary-precision JSON number", - }), - }, - serde_json::Value::Bool(_) => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "IEEE 754 float number", - actual: "JSON boolean", - }), - serde_json::Value::String(_) => Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "IEEE 754 float number", - actual: "JSON string", - }), - serde_json::Value::Array(_) | serde_json::Value::Object(_) => { - Err(DecodeError::WrongPayloadKind { - column: payload.column_name.to_string(), - expected: "IEEE 754 float number", - actual: "JSON array or object", - }) - } - } + super::json_decoders::decode_json_real(&payload) } } diff --git a/src/wire/json_decoders.rs b/src/wire/json_decoders.rs index df661d174..85b71c75b 100644 --- a/src/wire/json_decoders.rs +++ b/src/wire/json_decoders.rs @@ -9,6 +9,7 @@ use alloc::string::{String, ToString}; use alloc::vec::Vec; use super::error::DecodeError; +use super::scalar_helpers::normalize_real; use crate::encoding::Value; /// Abstracts access to the JSON payload value and column name for a CDC column. @@ -66,6 +67,104 @@ pub(crate) fn decode_json_str_with( } } +/// Hands a JSON number column to `decode`, and maps JSON null to [`Value::Null`]. +/// +/// # Errors +/// +/// Returns [`DecodeError::WrongPayloadKind`] naming `expected` if the JSON value is not a +/// number or null, and propagates `decode`'s error. +pub(crate) fn decode_json_number_with( + payload: &P, + expected: &'static str, + decode: impl FnOnce(&serde_json::Number) -> Result, DecodeError>, +) -> Result, DecodeError> { + let actual = match payload.json_value() { + serde_json::Value::Null => return Ok(Value::Null), + serde_json::Value::Number(n) => return decode(n), + serde_json::Value::Bool(_) => "JSON boolean", + serde_json::Value::String(_) => "JSON string", + serde_json::Value::Array(_) | serde_json::Value::Object(_) => "JSON array or object", + }; + Err(DecodeError::WrongPayloadKind { + column: payload.column_name().to_string(), + expected, + actual, + }) +} + +/// Decodes a JSON integer column into [`Value::Integer`], or [`Value::Null`]. +/// +/// # Errors +/// +/// Returns [`DecodeError::IntegerOverflow`] if the number does not fit an `i64`, and +/// [`DecodeError::WrongPayloadKind`] if the JSON value is not a number or null. +pub(crate) fn decode_json_int( + payload: &P, +) -> Result, DecodeError> { + decode_json_number_with(payload, "JSON integer number", |n| { + n.as_i64() + .map(Value::Integer) + .ok_or_else(|| DecodeError::IntegerOverflow { + column: payload.column_name().to_string(), + digits: n.to_string(), + }) + }) +} + +/// Decodes a JSON float column into a normalized [`Value::Real`], or [`Value::Null`]. +/// +/// # Errors +/// +/// Returns [`DecodeError::WrongPayloadKind`] if the number has no `f64` form, or if the +/// JSON value is not a number or null. +pub(crate) fn decode_json_real( + payload: &P, +) -> Result, DecodeError> { + decode_json_number_with(payload, "IEEE 754 float number", |n| { + n.as_f64() + .map(normalize_real) + .ok_or_else(|| DecodeError::WrongPayloadKind { + column: payload.column_name().to_string(), + expected: "IEEE 754 float number", + actual: "arbitrary-precision JSON number", + }) + }) +} + +/// Decodes a JSON integer, or a numeric string, into [`Value::Integer`] when it fits an +/// `i64` and into base-10 [`Value::Text`] otherwise, or [`Value::Null`]. +/// +/// # Errors +/// +/// Returns [`DecodeError::WrongPayloadKind`] if the JSON value is neither a number, a +/// numeric string, nor null. +pub(crate) fn decode_json_int64_or_text( + payload: &P, +) -> Result, DecodeError> +where + S: From, +{ + match payload.json_value() { + serde_json::Value::Null => Ok(Value::Null), + serde_json::Value::Number(n) => Ok(n + .as_i64() + .map_or_else(|| Value::Text(S::from(n.to_string())), Value::Integer)), + serde_json::Value::String(s) + if s.trim_start_matches('-') + .chars() + .all(|c| c.is_ascii_digit()) => + { + Ok(s.parse::() + .map_or_else(|_| Value::Text(S::from(s.clone())), Value::Integer)) + } + _ => Err(DecodeError::WrongPayloadKind { + column: payload.column_name().to_string(), + expected: "JSON integer number or numeric string", + actual: "other JSON shape", + }), + } +} + /// Decodes a JSON string or number decimal column into [`Value::Text`], or [`Value::Null`]. /// /// # Errors diff --git a/src/wire/shared_builders.rs b/src/wire/shared_builders.rs index decf3abe1..dfa8db22e 100644 --- a/src/wire/shared_builders.rs +++ b/src/wire/shared_builders.rs @@ -74,6 +74,28 @@ where Ok((col_idx, value)) } +/// Decode every column of `items` and set it on `row` with `set`. +fn set_items( + items: I, + table: &T, + adapter: &A, + row: R, + set: impl Fn(R, usize, Value) -> Result, +) -> Result +where + Src: WireSource, + T: NamedColumns + WireColumnTypes, + A: WireAdapter, + I: IntoIterator, + C: WireColumnItem, +{ + items.into_iter().try_fold(row, |row, item| { + let (col_idx, value) = decode_item(&item, table, adapter)?; + set(row, col_idx, value) + .map_err(|_| ConversionError::ColumnNotFound(String::from(item.name()))) + }) +} + /// Build an INSERT from every column in the event. pub(crate) fn build_insert( items: I, @@ -89,14 +111,13 @@ where I: IntoIterator, C: WireColumnItem, { - let mut insert = Insert::from(table.clone()); - for item in items { - let (col_idx, value) = decode_item(&item, table, adapter)?; - insert = insert - .set(col_idx, value) - .map_err(|_| ConversionError::ColumnNotFound(String::from(item.name())))?; - } - Ok(insert) + set_items( + items, + table, + adapter, + Insert::from(table.clone()), + Insert::set, + ) } /// Build a patchset UPDATE from the event's new-image columns. @@ -114,14 +135,10 @@ where I: IntoIterator, C: WireColumnItem, { - let mut update: Update = Update::from(table.clone()); - for item in items { - let (col_idx, value) = decode_item(&item, table, adapter)?; - update = update - .set(col_idx, value) - .map_err(|_| ConversionError::ColumnNotFound(String::from(item.name())))?; - } - Ok(update) + let update: Update = Update::from(table.clone()); + set_items(items, table, adapter, update, |row, col_idx, value| { + row.set(col_idx, value) + }) } /// Build a changeset DELETE from the event's old-image (identity) columns. @@ -139,14 +156,13 @@ where I: IntoIterator, C: WireColumnItem, { - let mut delete = ChangeDelete::from(table.clone()); - for item in items { - let (col_idx, value) = decode_item(&item, table, adapter)?; - delete = delete - .set(col_idx, value) - .map_err(|_| ConversionError::ColumnNotFound(String::from(item.name())))?; - } - Ok(delete) + set_items( + items, + table, + adapter, + ChangeDelete::from(table.clone()), + ChangeDelete::set, + ) } /// Build a patchset DELETE, extracting only the primary-key columns. diff --git a/src/wire/uuid_helpers.rs b/src/wire/uuid_helpers.rs index dde689388..15e5e8d19 100644 --- a/src/wire/uuid_helpers.rs +++ b/src/wire/uuid_helpers.rs @@ -25,8 +25,8 @@ pub(crate) fn parse_uuid(s: &str) -> Result<[u8; 16], usize> { let mut out = [0u8; 16]; let hex_positions: [usize; 16] = [0, 2, 4, 6, 9, 11, 14, 16, 19, 21, 24, 26, 28, 30, 32, 34]; for (i, pos) in hex_positions.into_iter().enumerate() { - let hi = nibble(bytes[pos]).ok_or(s.len())?; - let lo = nibble(bytes[pos + 1]).ok_or(s.len())?; + let hi = hex_nibble(bytes[pos]).ok_or(s.len())?; + let lo = hex_nibble(bytes[pos + 1]).ok_or(s.len())?; out[i] = (hi << 4) | lo; } Ok(out) @@ -50,8 +50,9 @@ fn strip_uuid_braces(s: &str) -> &str { } } +/// The value of one ASCII hex digit, either case. #[inline] -fn nibble(b: u8) -> Option { +pub(crate) fn hex_nibble(b: u8) -> Option { match b { b'0'..=b'9' => Some(b - b'0'), b'a'..=b'f' => Some(b - b'a' + 10), diff --git a/tests/common/mod.rs b/tests/common/mod.rs index f331a53f1..51fb6f7ba 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -1,7 +1,9 @@ -//! Shared schema fixtures for CDC digestion integration tests. +//! Shared schema fixtures and operation assertions for the CDC digestion +//! integration tests. //! //! Provides a three-column `users` table (`id` INT PK, `name` TEXT, `active` -//! BOOL) with all required trait impls, reused by `digestion_pg_walstream`, +//! BOOL) with all required trait impls, plus the assertions every digestion +//! test shares for the canonical row images, reused by `digestion_pg_walstream`, //! `digestion_wal2json`, and `digestion_maxwell`. #![allow(dead_code, unused_imports)] @@ -11,8 +13,8 @@ extern crate alloc; use alloc::vec::Vec; use sqlite_diff_rs::{ - DynTable, IndexableValues, NamedColumns, SchemaWithPK, SimpleTable, Value, WireColumnTypes, - WireSchema, WireType, + ChangesetUpdatePair, DynTable, IndexableValues, NamedColumns, PatchsetUpdateEntry, + SchemaWithPK, SimpleTable, Value, WireColumnTypes, WireSchema, WireType, }; /// A single-table schema holding the `users` test table. @@ -122,3 +124,49 @@ pub fn source_scoped_test_schema(source_schema: &'static str) -> SourceScopedTes source_schema, } } + +/// The encoded image of the canonical `users` row (`id` 1, `name`, `active` 1). +pub fn assert_users_row(values: &[Value>], name: &str) { + assert_eq!(values.len(), 3, "three columns"); + assert_eq!(values[0], Value::Integer(1), "id"); + assert_eq!(values[1], Value::Text(name.to_string()), "name"); + assert_eq!(values[2], Value::Integer(1), "active=true encodes as 1"); +} + +/// The `(old, new)` image pairs of a canonical `users` `UPDATE`. +pub fn assert_users_update(values: &[ChangesetUpdatePair>]) { + assert_eq!(values.len(), 3, "three columns"); + assert_eq!(values[0].0, Some(Value::Integer(1)), "old id"); + assert_eq!( + values[1].0, + Some(Value::Text("Alice".to_string())), + "old name" + ); + assert_eq!( + values[1].1, + Some(Value::Text("Alicia".to_string())), + "new name" + ); +} + +/// The primary key and new images of a canonical `users` patchset `UPDATE`. +pub fn assert_users_patchset_update( + pk: &[Value>], + entries: &[PatchsetUpdateEntry>], +) { + assert_eq!(pk, &[Value::Integer(1)], "primary key"); + assert_eq!(entries.len(), 3, "three column entries"); + assert_eq!( + entries[1].1, + Some(Value::Text("Alicia".to_string())), + "new name" + ); + assert_eq!(entries[2].1, Some(Value::Integer(1)), "new active"); +} + +/// The re-parsed image of a canonical `users` changeset `UPDATE`. +pub fn assert_parsed_users_update(values: &[ChangesetUpdatePair>]) { + assert_eq!(values.len(), 3, "column count in encoded bytes"); + assert_eq!(values[1].0, Some(Value::Text("Alice".to_string()))); + assert_eq!(values[1].1, Some(Value::Text("Alicia".to_string()))); +} diff --git a/tests/decoder_edge_cases.rs b/tests/decoder_edge_cases.rs index 0764de576..335edc842 100644 --- a/tests/decoder_edge_cases.rs +++ b/tests/decoder_edge_cases.rs @@ -1051,23 +1051,6 @@ fn mx_int64_overflow_wrong_payload() { } } -/// NaN cannot be represented in serde_json::Value::Number, so the -/// NaN→Null path in RealDecoder is unreachable for maxwell. -/// The NaN normalization is exercised via pg_walstream text-mode "NaN". -#[test] -fn mx_real_decoder_nan_unreachable() { - let dec = sqlite_diff_rs::RealDecoder; - // Verify that -0.0 keeps its sign instead. - let val = as_mx_dec!(dec) - .decode(MaxwellColumn { - column_name: "c", - wire_type: WireType::Real, - value: &serde_json::Value::Number(serde_json::Number::from_f64(-0.0).unwrap()), - }) - .unwrap(); - assert_eq!(val, Value::Real(-0.0), "-0.0 keeps its sign"); -} - #[test] fn mx_real_decoder_negative_zero_keeps_sign() { let dec = sqlite_diff_rs::RealDecoder; diff --git a/tests/digestion_maxwell.rs b/tests/digestion_maxwell.rs index 18b3693b2..68179967b 100644 --- a/tests/digestion_maxwell.rs +++ b/tests/digestion_maxwell.rs @@ -21,7 +21,10 @@ use sqlite_diff_rs::{ }; mod common; -use common::{TestUsersTable, source_scoped_test_schema, test_schema}; +use common::{ + TestUsersTable, assert_parsed_users_update, assert_users_patchset_update, assert_users_row, + assert_users_update, source_scoped_test_schema, test_schema, +}; fn default_adapter() -> TypeMap> { TypeMap::defaults() @@ -124,10 +127,7 @@ fn maxwell_changeset_insert() { match &ops[0] { ChangesetOp::Insert { table, values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(values.len(), 3, "three columns"); - assert_eq!(values[0], Value::Integer(1), "id"); - assert_eq!(values[1], Value::Text("Alice".to_string()), "name"); - assert_eq!(values[2], Value::Integer(1), "active=true encodes as 1"); + assert_users_row(values, "Alice"); } other => panic!("expected Insert, got {other:?}"), } @@ -141,10 +141,7 @@ fn maxwell_changeset_insert() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { ChangesetOp::Insert { values, .. } => { - assert_eq!(values.len(), 3, "column count in encoded bytes"); - assert_eq!(values[0], Value::Integer(1)); - assert_eq!(values[1], Value::Text("Alice".to_string())); - assert_eq!(values[2], Value::Integer(1)); + assert_users_row(values, "Alice"); } other => panic!("expected Insert in parsed bytes, got {other:?}"), } @@ -174,19 +171,8 @@ fn maxwell_changeset_update() { match &ops[0] { ChangesetOp::Update { table, values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(values.len(), 3, "three columns"); - assert_eq!(values[0].0, Some(Value::Integer(1)), "old id"); assert_eq!(values[0].1, Some(Value::Integer(1)), "new id (unchanged)"); - assert_eq!( - values[1].0, - Some(Value::Text("Alice".to_string())), - "old name" - ); - assert_eq!( - values[1].1, - Some(Value::Text("Alicia".to_string())), - "new name" - ); + assert_users_update(values); } other => panic!("expected Update, got {other:?}"), } @@ -199,9 +185,7 @@ fn maxwell_changeset_update() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { ChangesetOp::Update { values, .. } => { - assert_eq!(values.len(), 3, "column count in encoded bytes"); - assert_eq!(values[1].0, Some(Value::Text("Alice".to_string()))); - assert_eq!(values[1].1, Some(Value::Text("Alicia".to_string()))); + assert_parsed_users_update(values); } other => panic!("expected Update in parsed bytes, got {other:?}"), } @@ -223,10 +207,7 @@ fn maxwell_changeset_delete() { table, old_values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(old_values.len(), 3, "three columns"); - assert_eq!(old_values[0], Value::Integer(1), "id"); - assert_eq!(old_values[1], Value::Text("Alice".to_string()), "name"); - assert_eq!(old_values[2], Value::Integer(1), "active=true encodes as 1"); + assert_users_row(old_values, "Alice"); } other => panic!("expected Delete, got {other:?}"), } @@ -239,10 +220,7 @@ fn maxwell_changeset_delete() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { ChangesetOp::Delete { old_values, .. } => { - assert_eq!(old_values.len(), 3, "column count in encoded bytes"); - assert_eq!(old_values[0], Value::Integer(1)); - assert_eq!(old_values[1], Value::Text("Alice".to_string())); - assert_eq!(old_values[2], Value::Integer(1)); + assert_users_row(old_values, "Alice"); } other => panic!("expected Delete in parsed bytes, got {other:?}"), } @@ -264,10 +242,7 @@ fn maxwell_patchset_insert() { match &ops[0] { PatchsetOp::Insert { table, values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(values.len(), 3, "three columns"); - assert_eq!(values[0], Value::Integer(1), "id"); - assert_eq!(values[1], Value::Text("Alice".to_string()), "name"); - assert_eq!(values[2], Value::Integer(1), "active=true encodes as 1"); + assert_users_row(values, "Alice"); } other => panic!("expected Insert, got {other:?}"), } @@ -280,10 +255,7 @@ fn maxwell_patchset_insert() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { PatchsetOp::Insert { values, .. } => { - assert_eq!(values.len(), 3, "column count in encoded bytes"); - assert_eq!(values[0], Value::Integer(1)); - assert_eq!(values[1], Value::Text("Alice".to_string())); - assert_eq!(values[2], Value::Integer(1)); + assert_users_row(values, "Alice"); } other => panic!("expected Insert in parsed bytes, got {other:?}"), } @@ -314,14 +286,7 @@ fn maxwell_patchset_update() { table, pk, entries, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(pk, &[Value::Integer(1)], "primary key"); - assert_eq!(entries.len(), 3, "three column entries"); - assert_eq!( - entries[1].1, - Some(Value::Text("Alicia".to_string())), - "new name" - ); - assert_eq!(entries[2].1, Some(Value::Integer(1)), "new active"); + assert_users_patchset_update(pk, entries); } other => panic!("expected Update, got {other:?}"), } @@ -334,9 +299,7 @@ fn maxwell_patchset_update() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { PatchsetOp::Update { pk, entries, .. } => { - assert_eq!(pk, &[Value::Integer(1)]); - assert_eq!(entries.len(), 3, "column count in encoded bytes"); - assert_eq!(entries[1].1, Some(Value::Text("Alicia".to_string()))); + assert_users_patchset_update(pk, entries); } other => panic!("expected Update in parsed bytes, got {other:?}"), } diff --git a/tests/digestion_pg_walstream.rs b/tests/digestion_pg_walstream.rs index f79e9d8ce..fc5fd794b 100644 --- a/tests/digestion_pg_walstream.rs +++ b/tests/digestion_pg_walstream.rs @@ -18,7 +18,10 @@ use sqlite_diff_rs::{ }; mod common; -use common::{TestUsersTable, source_scoped_test_schema, test_schema}; +use common::{ + TestUsersTable, assert_parsed_users_update, assert_users_patchset_update, assert_users_row, + assert_users_update, source_scoped_test_schema, test_schema, +}; fn default_adapter() -> TypeMap> { TypeMap::defaults() @@ -71,10 +74,7 @@ fn pg_changeset_insert() { match &ops[0] { ChangesetOp::Insert { table, values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(values.len(), 3, "three columns"); - assert_eq!(values[0], Value::Integer(1), "id"); - assert_eq!(values[1], Value::Text("Alice".to_string()), "name"); - assert_eq!(values[2], Value::Integer(1), "active=true encodes as 1"); + assert_users_row(values, "Alice"); } other => panic!("expected Insert, got {other:?}"), } @@ -87,10 +87,7 @@ fn pg_changeset_insert() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { ChangesetOp::Insert { values, .. } => { - assert_eq!(values.len(), 3, "column count in encoded bytes"); - assert_eq!(values[0], Value::Integer(1)); - assert_eq!(values[1], Value::Text("Alice".to_string())); - assert_eq!(values[2], Value::Integer(1)); + assert_users_row(values, "Alice"); } other => panic!("expected Insert in parsed bytes, got {other:?}"), } @@ -126,18 +123,7 @@ fn pg_changeset_update() { match &ops[0] { ChangesetOp::Update { table, values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(values.len(), 3, "three columns"); - assert_eq!(values[0].0, Some(Value::Integer(1)), "old id"); - assert_eq!( - values[1].0, - Some(Value::Text("Alice".to_string())), - "old name" - ); - assert_eq!( - values[1].1, - Some(Value::Text("Alicia".to_string())), - "new name" - ); + assert_users_update(values); } other => panic!("expected Update, got {other:?}"), } @@ -150,9 +136,7 @@ fn pg_changeset_update() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { ChangesetOp::Update { values, .. } => { - assert_eq!(values.len(), 3, "column count in encoded bytes"); - assert_eq!(values[1].0, Some(Value::Text("Alice".to_string()))); - assert_eq!(values[1].1, Some(Value::Text("Alicia".to_string()))); + assert_parsed_users_update(values); } other => panic!("expected Update in parsed bytes, got {other:?}"), } @@ -180,10 +164,7 @@ fn pg_changeset_delete() { table, old_values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(old_values.len(), 3, "three columns"); - assert_eq!(old_values[0], Value::Integer(1), "id"); - assert_eq!(old_values[1], Value::Text("Alice".to_string()), "name"); - assert_eq!(old_values[2], Value::Integer(1), "active=true encodes as 1"); + assert_users_row(old_values, "Alice"); } other => panic!("expected Delete, got {other:?}"), } @@ -196,10 +177,7 @@ fn pg_changeset_delete() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { ChangesetOp::Delete { old_values, .. } => { - assert_eq!(old_values.len(), 3, "column count in encoded bytes"); - assert_eq!(old_values[0], Value::Integer(1)); - assert_eq!(old_values[1], Value::Text("Alice".to_string())); - assert_eq!(old_values[2], Value::Integer(1)); + assert_users_row(old_values, "Alice"); } other => panic!("expected Delete in parsed bytes, got {other:?}"), } @@ -225,10 +203,7 @@ fn pg_patchset_insert() { match &ops[0] { PatchsetOp::Insert { table, values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(values.len(), 3, "three columns"); - assert_eq!(values[0], Value::Integer(1), "id"); - assert_eq!(values[1], Value::Text("Alice".to_string()), "name"); - assert_eq!(values[2], Value::Integer(1), "active=true encodes as 1"); + assert_users_row(values, "Alice"); } other => panic!("expected Insert, got {other:?}"), } @@ -241,10 +216,7 @@ fn pg_patchset_insert() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { PatchsetOp::Insert { values, .. } => { - assert_eq!(values.len(), 3, "column count in encoded bytes"); - assert_eq!(values[0], Value::Integer(1)); - assert_eq!(values[1], Value::Text("Alice".to_string())); - assert_eq!(values[2], Value::Integer(1)); + assert_users_row(values, "Alice"); } other => panic!("expected Insert in parsed bytes, got {other:?}"), } @@ -282,14 +254,7 @@ fn pg_patchset_update() { table, pk, entries, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(pk, &[Value::Integer(1)], "primary key"); - assert_eq!(entries.len(), 3, "three column entries"); - assert_eq!( - entries[1].1, - Some(Value::Text("Alicia".to_string())), - "new name" - ); - assert_eq!(entries[2].1, Some(Value::Integer(1)), "new active"); + assert_users_patchset_update(pk, entries); } other => panic!("expected Update, got {other:?}"), } @@ -302,9 +267,7 @@ fn pg_patchset_update() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { PatchsetOp::Update { pk, entries, .. } => { - assert_eq!(pk, &[Value::Integer(1)]); - assert_eq!(entries.len(), 3, "column count in encoded bytes"); - assert_eq!(entries[1].1, Some(Value::Text("Alicia".to_string()))); + assert_users_patchset_update(pk, entries); } other => panic!("expected Update in parsed bytes, got {other:?}"), } diff --git a/tests/digestion_wal2json.rs b/tests/digestion_wal2json.rs index e7bb25fc1..4ef188510 100644 --- a/tests/digestion_wal2json.rs +++ b/tests/digestion_wal2json.rs @@ -20,7 +20,10 @@ use sqlite_diff_rs::{ }; mod common; -use common::{TestUsersTable, source_scoped_test_schema, test_schema}; +use common::{ + TestUsersTable, assert_parsed_users_update, assert_users_patchset_update, assert_users_row, + assert_users_update, source_scoped_test_schema, test_schema, +}; fn default_adapter() -> TypeMap> { TypeMap::defaults() @@ -191,10 +194,7 @@ fn w2j_v2_changeset_insert() { match &ops[0] { ChangesetOp::Insert { table, values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(values.len(), 3, "three columns"); - assert_eq!(values[0], Value::Integer(1), "id"); - assert_eq!(values[1], Value::Text("Alice".to_string()), "name"); - assert_eq!(values[2], Value::Integer(1), "active=true encodes as 1"); + assert_users_row(values, "Alice"); } other => panic!("expected Insert, got {other:?}"), } @@ -207,10 +207,7 @@ fn w2j_v2_changeset_insert() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { ChangesetOp::Insert { values, .. } => { - assert_eq!(values.len(), 3, "column count in encoded bytes"); - assert_eq!(values[0], Value::Integer(1)); - assert_eq!(values[1], Value::Text("Alice".to_string())); - assert_eq!(values[2], Value::Integer(1)); + assert_users_row(values, "Alice"); } other => panic!("expected Insert in parsed bytes, got {other:?}"), } @@ -242,18 +239,7 @@ fn w2j_v2_changeset_update() { match &ops[0] { ChangesetOp::Update { table, values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(values.len(), 3, "three columns"); - assert_eq!(values[0].0, Some(Value::Integer(1)), "old id"); - assert_eq!( - values[1].0, - Some(Value::Text("Alice".to_string())), - "old name" - ); - assert_eq!( - values[1].1, - Some(Value::Text("Alicia".to_string())), - "new name" - ); + assert_users_update(values); } other => panic!("expected Update, got {other:?}"), } @@ -266,9 +252,7 @@ fn w2j_v2_changeset_update() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { ChangesetOp::Update { values, .. } => { - assert_eq!(values.len(), 3, "column count in encoded bytes"); - assert_eq!(values[1].0, Some(Value::Text("Alice".to_string()))); - assert_eq!(values[1].1, Some(Value::Text("Alicia".to_string()))); + assert_parsed_users_update(values); } other => panic!("expected Update in parsed bytes, got {other:?}"), } @@ -294,10 +278,7 @@ fn w2j_v2_changeset_delete() { table, old_values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(old_values.len(), 3, "three columns"); - assert_eq!(old_values[0], Value::Integer(1), "id"); - assert_eq!(old_values[1], Value::Text("Alice".to_string()), "name"); - assert_eq!(old_values[2], Value::Integer(1), "active"); + assert_users_row(old_values, "Alice"); } other => panic!("expected Delete, got {other:?}"), } @@ -310,10 +291,7 @@ fn w2j_v2_changeset_delete() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { ChangesetOp::Delete { old_values, .. } => { - assert_eq!(old_values.len(), 3, "column count in encoded bytes"); - assert_eq!(old_values[0], Value::Integer(1)); - assert_eq!(old_values[1], Value::Text("Alice".to_string())); - assert_eq!(old_values[2], Value::Integer(1)); + assert_users_row(old_values, "Alice"); } other => panic!("expected Delete in parsed bytes, got {other:?}"), } @@ -339,10 +317,7 @@ fn w2j_v2_patchset_insert() { match &ops[0] { PatchsetOp::Insert { table, values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(values.len(), 3, "three columns"); - assert_eq!(values[0], Value::Integer(1), "id"); - assert_eq!(values[1], Value::Text("Alice".to_string()), "name"); - assert_eq!(values[2], Value::Integer(1), "active=true encodes as 1"); + assert_users_row(values, "Alice"); } other => panic!("expected Insert, got {other:?}"), } @@ -355,10 +330,7 @@ fn w2j_v2_patchset_insert() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { PatchsetOp::Insert { values, .. } => { - assert_eq!(values.len(), 3, "column count in encoded bytes"); - assert_eq!(values[0], Value::Integer(1)); - assert_eq!(values[1], Value::Text("Alice".to_string())); - assert_eq!(values[2], Value::Integer(1)); + assert_users_row(values, "Alice"); } other => panic!("expected Insert in parsed bytes, got {other:?}"), } @@ -392,14 +364,7 @@ fn w2j_v2_patchset_update() { table, pk, entries, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(pk, &[Value::Integer(1)], "primary key"); - assert_eq!(entries.len(), 3, "three column entries"); - assert_eq!( - entries[1].1, - Some(Value::Text("Alicia".to_string())), - "new name" - ); - assert_eq!(entries[2].1, Some(Value::Integer(1)), "new active"); + assert_users_patchset_update(pk, entries); } other => panic!("expected Update, got {other:?}"), } @@ -412,9 +377,7 @@ fn w2j_v2_patchset_update() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { PatchsetOp::Update { pk, entries, .. } => { - assert_eq!(pk, &[Value::Integer(1)]); - assert_eq!(entries.len(), 3, "column count in encoded bytes"); - assert_eq!(entries[1].1, Some(Value::Text("Alicia".to_string()))); + assert_users_patchset_update(pk, entries); } other => panic!("expected Update in parsed bytes, got {other:?}"), } @@ -480,10 +443,7 @@ fn w2j_v1_changeset_insert() { match &ops[0] { ChangesetOp::Insert { table, values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(values.len(), 3, "three columns"); - assert_eq!(values[0], Value::Integer(1), "id"); - assert_eq!(values[1], Value::Text("Alice".to_string()), "name"); - assert_eq!(values[2], Value::Integer(1), "active=true encodes as 1"); + assert_users_row(values, "Alice"); } other => panic!("expected Insert, got {other:?}"), } @@ -496,10 +456,7 @@ fn w2j_v1_changeset_insert() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { ChangesetOp::Insert { values, .. } => { - assert_eq!(values.len(), 3, "column count in encoded bytes"); - assert_eq!(values[0], Value::Integer(1)); - assert_eq!(values[1], Value::Text("Alice".to_string())); - assert_eq!(values[2], Value::Integer(1)); + assert_users_row(values, "Alice"); } other => panic!("expected Insert in parsed bytes, got {other:?}"), } @@ -637,10 +594,7 @@ fn w2j_v1_patchset_insert() { match &ops[0] { PatchsetOp::Insert { table, values, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(values.len(), 3, "three columns"); - assert_eq!(values[0], Value::Integer(1), "id"); - assert_eq!(values[1], Value::Text("Alice".to_string()), "name"); - assert_eq!(values[2], Value::Integer(1), "active=true encodes as 1"); + assert_users_row(values, "Alice"); } other => panic!("expected Insert, got {other:?}"), } @@ -653,10 +607,7 @@ fn w2j_v1_patchset_insert() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { PatchsetOp::Insert { values, .. } => { - assert_eq!(values.len(), 3, "column count in encoded bytes"); - assert_eq!(values[0], Value::Integer(1)); - assert_eq!(values[1], Value::Text("Alice".to_string())); - assert_eq!(values[2], Value::Integer(1)); + assert_users_row(values, "Alice"); } other => panic!("expected Insert in parsed bytes, got {other:?}"), } @@ -698,13 +649,7 @@ fn w2j_v1_patchset_update() { table, pk, entries, .. } => { assert_eq!(table.name(), "users"); - assert_eq!(pk, &[Value::Integer(1)], "primary key"); - assert_eq!(entries.len(), 3, "three column entries"); - assert_eq!( - entries[1].1, - Some(Value::Text("Alicia".to_string())), - "new name" - ); + assert_users_patchset_update(pk, entries); } other => panic!("expected Update, got {other:?}"), } @@ -717,9 +662,7 @@ fn w2j_v1_patchset_update() { assert_eq!(parsed_ops.len(), 1); match &parsed_ops[0] { PatchsetOp::Update { pk, entries, .. } => { - assert_eq!(pk, &[Value::Integer(1)]); - assert_eq!(entries.len(), 3, "column count in encoded bytes"); - assert_eq!(entries[1].1, Some(Value::Text("Alicia".to_string()))); + assert_users_patchset_update(pk, entries); } other => panic!("expected Update in parsed bytes, got {other:?}"), } diff --git a/tests/sql_parsing.rs b/tests/sql_parsing.rs index 02d58c8ef..46dc3dd41 100644 --- a/tests/sql_parsing.rs +++ b/tests/sql_parsing.rs @@ -399,21 +399,6 @@ fn parity_insert_and_update() { ); } -#[test] -fn parity_multi_table() { - let users = SimpleTable::with_rowid_alias("users", &[("id", "INTEGER"), ("name", "TEXT")], 0); - let posts = SimpleTable::with_rowid_alias("posts", &[("id", "INTEGER"), ("title", "TEXT")], 0); - assert_patchset_sql_parity( - &[users, posts], - &[ - "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)", - "CREATE TABLE posts (id INTEGER PRIMARY KEY, title TEXT)", - "INSERT INTO users (id, name) VALUES (1, 'Alice')", - "INSERT INTO posts (id, title) VALUES (1, 'Hello')", - ], - ); -} - #[test] fn parity_composite_pk() { let order_items = SimpleTable::new(