Merge pull request #1 from somtochiama/fix-pk-update

Update columns in clock table when changing primary keys.

Author: somtochiama

.github/workflows/py-tests.yaml

@@ -28,7 +28,7 @@ jobs:
           cd core
           make loadable
 
-      - uses: conda-incubator/setup-miniconda@v2
+      - uses: conda-incubator/setup-miniconda@v3
         with:
           auto-update-conda: true
           auto-activate-base: true

core/Makefile

@@ -72,6 +72,8 @@ dbg_prefix=./dbg
 bundle=bundle_static
 valgrind: bundle = integration_check
 test: bundle = integration_check
+ubsan: bundle = integration_check
+asan: bundle = integration_check
 
 TARGET_LOADABLE=$(prefix)/crsqlite.$(LOADABLE_EXTENSION)
 TARGET_DBG_LOADABLE=$(dbg_prefix)/crsqlite.$(LOADABLE_EXTENSION)
@@ -124,7 +126,10 @@ valgrind: $(TARGET_TEST)
 	valgrind $(prefix)/test
 analyzer:
 	scan-build $(MAKE) clean loadable
-ubsan: CC=clang
+asan: export RUSTFLAGS=-Zsanitizer=address
+asan: rs_build_flags=--target x86_64-unknown-linux-gnu -Zbuild-std
+asan: rs_lib_dbg_static=./rs/$(bundle)/target/x86_64-unknown-linux-gnu/debug/libcrsql_$(bundle).a
+asan: CC=clang
 ubsan: LDLIBS += -lubsan
 ubsan: clean $(TARGET_TEST)
 	$(prefix)/test
@@ -255,7 +260,7 @@ $(TARGET_TEST): $(prefix) $(TARGET_SQLITE3_EXTRA_C) src/tests.c src/*.test.c $(e
 	$(TARGET_SQLITE3_EXTRA_C) src/tests.c src/*.test.c $(ext_files) $(rs_lib_dbg_static_cpy) \
 	$(LDLIBS) -o $@
 
-$(TARGET_TEST_ASAN): $(prefix) $(TARGET_SQLITE3_EXTRA_C) src/tests.c src/*.test.c $(ext_files)
+$(TARGET_TEST_ASAN): $(prefix) $(TARGET_SQLITE3_EXTRA_C) src/tests.c src/*.test.c $(ext_files) $(rs_lib_dbg_static_cpy)
 	$(CC) -fsanitize=address -g -fno-omit-frame-pointer -Wall \
 	-DSQLITE_THREADSAFE=0 \
 	-DSQLITE_OMIT_LOAD_EXTENSION=1 \

core/rs/core/src/lib.rs

@@ -548,7 +548,7 @@ unsafe extern "C" fn x_crsql_as_crr(
 ) {
     if argc == 0 {
         ctx.result_error(
-            "Wrong number of args provided to crsql_as_crr. Provide the schema 
+            "Wrong number of args provided to crsql_as_crr. Provide the schema
           name and table name or just the table name.",
         );
         return;
@@ -609,7 +609,7 @@ unsafe extern "C" fn x_crsql_begin_alter(
 ) {
     if argc == 0 {
         ctx.result_error(
-            "Wrong number of args provided to crsql_begin_alter. Provide the 
+            "Wrong number of args provided to crsql_begin_alter. Provide the
           schema name and table name or just the table name.",
         );
         return;
@@ -620,7 +620,7 @@ unsafe extern "C" fn x_crsql_begin_alter(
     let (_schema_name, table_name) = if argc == 2 {
         (args[0].text(), args[1].text())
     } else {
-        ("main", args[0].text())
+        ("main\0", args[0].text())
     };
 
     let db = ctx.db_handle();
@@ -645,7 +645,7 @@ unsafe extern "C" fn x_crsql_commit_alter(
 ) {
     if argc == 0 {
         ctx.result_error(
-            "Wrong number of args provided to crsql_commit_alter. Provide the 
+            "Wrong number of args provided to crsql_commit_alter. Provide the
           schema name and table name or just the table name.",
         );
         return;
@@ -655,7 +655,7 @@ unsafe extern "C" fn x_crsql_commit_alter(
     let (schema_name, table_name) = if argc >= 2 {
         (args[0].text(), args[1].text())
     } else {
-        ("main", args[0].text())
+        ("main\0", args[0].text())
     };
 
     let non_destructive = if argc >= 3 { args[2].int() == 1 } else { false };

core/rs/core/src/local_writes/after_update.rs

@@ -74,7 +74,7 @@ fn after_update(
     let next_db_version = crate::db_version::next_db_version(db, ext_data, None)?;
     let new_key = tbl_info
         .get_or_create_key_via_raw_values(db, pks_new)
-        .or_else(|_| Err("failed geteting or creating lookaside key"))?;
+        .or_else(|_| Err("failed getting or creating lookaside key"))?;
 
     // Changing a primary key column to a new value is the same thing as deleting the row
     // previously identified by the primary key.
@@ -85,16 +85,21 @@ fn after_update(
         let next_seq = super::bump_seq(ext_data);
         // Record the delete of the row identified by the old primary keys
         after_update__mark_old_pk_row_deleted(db, tbl_info, old_key, next_db_version, next_seq)?;
-        // TODO: each non sentinel needs a unique seq on the move?
-        after_update__move_non_sentinels(db, tbl_info, new_key, old_key)?;
-        // Record a create of the row identified by the new primary keys
-        // if no rows were moved. This is related to the optimization to not save
-        // sentinels unless required.
-        // if db.changes64() == 0 { <-- an optimization if we can get to it. we'd need to know to increment causal length.
-        // so we can get to this when CL is stored in the lookaside.
         let next_seq = super::bump_seq(ext_data);
+        // todo: we don't need to this, if there's no existing row (cl is assumed to be 1).
         super::mark_new_pk_row_created(db, tbl_info, new_key, next_db_version, next_seq)?;
-        // }
+        for col in tbl_info.non_pks.iter() {
+            let next_seq = super::bump_seq(ext_data);
+            after_update__move_non_pk_col(
+                db,
+                tbl_info,
+                new_key,
+                old_key,
+                &col.name,
+                next_db_version,
+                next_seq,
+            )?;
+        }
     }
 
     // now for each non_pk_col we need to do an insert
@@ -146,6 +151,32 @@ fn after_update__mark_old_pk_row_deleted(
     super::step_trigger_stmt(mark_locally_deleted_stmt)
 }
 
+#[allow(non_snake_case)]
+fn after_update__move_non_pk_col(
+    db: *mut sqlite3,
+    tbl_info: &TableInfo,
+    new_key: sqlite::int64,
+    old_key: sqlite::int64,
+    col_name: &str,
+    db_version: sqlite::int64,
+    seq: i32,
+) -> Result<ResultCode, String> {
+    let move_non_pk_col_stmt_ref = tbl_info
+        .get_move_non_pk_col_stmt(db)
+        .or_else(|_e| Err("failed to get move_non_pk_col_stmt"))?;
+    let move_non_pk_col_stmt = move_non_pk_col_stmt_ref
+        .as_ref()
+        .ok_or("Failed to deref sentinel stmt")?;
+    move_non_pk_col_stmt
+        .bind_int64(1, new_key)
+        .and_then(|_| move_non_pk_col_stmt.bind_int64(2, db_version))
+        .and_then(|_| move_non_pk_col_stmt.bind_int(3, seq))
+        .and_then(|_| move_non_pk_col_stmt.bind_int64(4, old_key))
+        .and_then(|_| move_non_pk_col_stmt.bind_text(5, col_name, sqlite::Destructor::STATIC))
+        .or_else(|_| Err("failed binding to move_non_pk_col_stmt"))?;
+    super::step_trigger_stmt(move_non_pk_col_stmt)
+}
+
 // TODO: in the future we can keep sentinel information in the lookaside
 #[allow(non_snake_case)]
 fn after_update__move_non_sentinels(

core/rs/core/src/tableinfo.rs

@@ -468,6 +468,32 @@ impl TableInfo {
         Ok(self.move_non_sentinels_stmt.try_borrow()?)
     }
 
+    pub fn get_move_non_pk_col_stmt(
+        &self,
+        db: *mut sqlite3,
+    ) -> Result<Ref<Option<ManagedStmt>>, ResultCode> {
+        if self.move_non_sentinels_stmt.try_borrow()?.is_none() {
+            // Incrementing col_version is especially important for the case where we
+            // are updating to a currently existing pk, so that the columns
+            // from the old pk can override the ones from the new at a node
+            // following our changes.
+            let sql = format!(
+            "UPDATE OR REPLACE \"{table_name}__crsql_clock\" SET
+                key = ?,
+                db_version = ?,
+                seq = ?,
+                col_version = col_version + 1,
+                site_id = 0
+            WHERE
+                key = ? AND col_name = ?",
+              table_name = crate::util::escape_ident(&self.tbl_name),
+            );
+            let ret = db.prepare_v3(&sql, sqlite::PREPARE_PERSISTENT)?;
+            *self.move_non_sentinels_stmt.try_borrow_mut()? = Some(ret);
+        }
+        Ok(self.move_non_sentinels_stmt.try_borrow()?)
+    }
+
     pub fn get_mark_locally_created_stmt(
         &self,
         db: *mut sqlite3,