Parse checkpoints and data lines with serde

Author: simolus3

crates/core/src/checkpoint.rs

@@ -5,7 +5,7 @@ use alloc::string::String;
 use alloc::vec::Vec;
 use core::ffi::c_int;
 
-use serde::{Deserialize, Serialize};
+use serde::Serialize;
 use serde_json as json;
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;

crates/core/src/operations.rs

@@ -1,8 +1,12 @@
 use alloc::format;
 use alloc::string::String;
+use alloc::vec::Vec;
 use num_traits::Zero;
+use serde::Deserialize;
 
 use crate::error::{PSResult, SQLiteError};
+use crate::sync::line::DataLine;
+use crate::sync::operations::insert_bucket_operations;
 use crate::sync::Checksum;
 use sqlite_nostd as sqlite;
 use sqlite_nostd::{Connection, ResultCode};
@@ -11,246 +15,15 @@ use crate::ext::SafeManagedStmt;
 
 // Run inside a transaction
 pub fn insert_operation(db: *mut sqlite::sqlite3, data: &str) -> Result<(), SQLiteError> {
-    // language=SQLite
-    let statement = db.prepare_v2(
-        "\
-SELECT
-    json_extract(e.value, '$.bucket') as bucket,
-    json_extract(e.value, '$.data') as data,
-    json_extract(e.value, '$.has_more') as has_more,
-    json_extract(e.value, '$.after') as after,
-    json_extract(e.value, '$.next_after') as next_after
-FROM json_each(json_extract(?1, '$.buckets')) e",
-    )?;
-    statement.bind_text(1, data, sqlite::Destructor::STATIC)?;
-
-    while statement.step()? == ResultCode::ROW {
-        let bucket = statement.column_text(0)?;
-        let data = statement.column_text(1)?;
-        // let _has_more = statement.column_int(2)? != 0;
-        // let _after = statement.column_text(3)?;
-        // let _next_after = statement.column_text(4)?;
-
-        insert_bucket_operations(db, bucket, data)?;
-    }
-
-    Ok(())
-}
-
-pub fn insert_bucket_operations(
-    db: *mut sqlite::sqlite3,
-    bucket: &str,
-    data: &str,
-) -> Result<(), SQLiteError> {
-    // Statement to insert new operations (only for PUT and REMOVE).
-    // language=SQLite
-    let iterate_statement = db.prepare_v2(
-        "\
-SELECT
-    json_extract(e.value, '$.op_id') as op_id,
-    json_extract(e.value, '$.op') as op,
-    json_extract(e.value, '$.object_type') as object_type,
-    json_extract(e.value, '$.object_id') as object_id,
-    json_extract(e.value, '$.checksum') as checksum,
-    json_extract(e.value, '$.data') as data,
-    json_extract(e.value, '$.subkey') as subkey
-FROM json_each(?) e",
-    )?;
-    iterate_statement.bind_text(1, data, sqlite::Destructor::STATIC)?;
-
-    // We do an ON CONFLICT UPDATE simply so that the RETURNING bit works for existing rows.
-    // We can consider splitting this into separate SELECT and INSERT statements.
-    // language=SQLite
-    let bucket_statement = db.prepare_v2(
-        "INSERT INTO ps_buckets(name)
-                            VALUES(?)
-                        ON CONFLICT DO UPDATE
-                            SET last_applied_op = last_applied_op
-                        RETURNING id, last_applied_op",
-    )?;
-    bucket_statement.bind_text(1, bucket, sqlite::Destructor::STATIC)?;
-    bucket_statement.step()?;
-
-    let bucket_id = bucket_statement.column_int64(0);
-
-    // This is an optimization for initial sync - we can avoid persisting individual REMOVE
-    // operations when last_applied_op = 0.
-    // We do still need to do the "supersede_statement" step for this case, since a REMOVE
-    // operation can supersede another PUT operation we're syncing at the same time.
-    let mut is_empty = bucket_statement.column_int64(1) == 0;
-
-    // Statement to supersede (replace) operations with the same key.
-    // language=SQLite
-    let supersede_statement = db.prepare_v2(
-        "\
-DELETE FROM ps_oplog
-    WHERE unlikely(ps_oplog.bucket = ?1)
-    AND ps_oplog.key = ?2
-RETURNING op_id, hash",
-    )?;
-    supersede_statement.bind_int64(1, bucket_id)?;
-
-    // language=SQLite
-    let insert_statement = db.prepare_v2("\
-INSERT INTO ps_oplog(bucket, op_id, key, row_type, row_id, data, hash) VALUES (?, ?, ?, ?, ?, ?, ?)")?;
-    insert_statement.bind_int64(1, bucket_id)?;
-
-    let updated_row_statement = db.prepare_v2(
-        "\
-INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id) VALUES(?1, ?2)",
-    )?;
-
-    bucket_statement.reset()?;
-
-    let mut last_op: Option<i64> = None;
-    let mut add_checksum = Checksum::zero();
-    let mut op_checksum = Checksum::zero();
-    let mut added_ops: i32 = 0;
-
-    while iterate_statement.step()? == ResultCode::ROW {
-        let op_id = iterate_statement.column_int64(0);
-        let op = iterate_statement.column_text(1)?;
-        let object_type = iterate_statement.column_text(2);
-        let object_id = iterate_statement.column_text(3);
-        let checksum = Checksum::from_i32(iterate_statement.column_int(4));
-        let op_data = iterate_statement.column_text(5);
-
-        last_op = Some(op_id);
-        added_ops += 1;
-
-        if op == "PUT" || op == "REMOVE" {
-            let key: String;
-            if let (Ok(object_type), Ok(object_id)) = (object_type.as_ref(), object_id.as_ref()) {
-                let subkey = iterate_statement.column_text(6).unwrap_or("null");
-                key = format!("{}/{}/{}", &object_type, &object_id, subkey);
-            } else {
-                key = String::from("");
-            }
-
-            supersede_statement.bind_text(2, &key, sqlite::Destructor::STATIC)?;
-
-            let mut superseded = false;
-
-            while supersede_statement.step()? == ResultCode::ROW {
-                // Superseded (deleted) a previous operation, add the checksum
-                let supersede_checksum = Checksum::from_i32(supersede_statement.column_int(1));
-                add_checksum += supersede_checksum;
-                op_checksum -= supersede_checksum;
-
-                // Superseded an operation, only skip if the bucket was empty
-                // Previously this checked "superseded_op <= last_applied_op".
-                // However, that would not account for a case where a previous
-                // PUT operation superseded the original PUT operation in this
-                // same batch, in which case superseded_op is not accurate for this.
-                if !is_empty {
-                    superseded = true;
-                }
-            }
-            supersede_statement.reset()?;
-
-            if op == "REMOVE" {
-                let should_skip_remove = !superseded;
-
-                add_checksum += checksum;
-
-                if !should_skip_remove {
-                    if let (Ok(object_type), Ok(object_id)) = (object_type, object_id) {
-                        updated_row_statement.bind_text(
-                            1,
-                            object_type,
-                            sqlite::Destructor::STATIC,
-                        )?;
-                        updated_row_statement.bind_text(
-                            2,
-                            object_id,
-                            sqlite::Destructor::STATIC,
-                        )?;
-                        updated_row_statement.exec()?;
-                    }
-                }
-
-                continue;
-            }
-
-            insert_statement.bind_int64(2, op_id)?;
-            if key != "" {
-                insert_statement.bind_text(3, &key, sqlite::Destructor::STATIC)?;
-            } else {
-                insert_statement.bind_null(3)?;
-            }
-
-            if let (Ok(object_type), Ok(object_id)) = (object_type, object_id) {
-                insert_statement.bind_text(4, object_type, sqlite::Destructor::STATIC)?;
-                insert_statement.bind_text(5, object_id, sqlite::Destructor::STATIC)?;
-            } else {
-                insert_statement.bind_null(4)?;
-                insert_statement.bind_null(5)?;
-            }
-            if let Ok(data) = op_data {
-                insert_statement.bind_text(6, data, sqlite::Destructor::STATIC)?;
-            } else {
-                insert_statement.bind_null(6)?;
-            }
-
-            insert_statement.bind_int(7, checksum.bitcast_i32())?;
-            insert_statement.exec()?;
-
-            op_checksum += checksum;
-        } else if op == "MOVE" {
-            add_checksum += checksum;
-        } else if op == "CLEAR" {
-            // Any remaining PUT operations should get an implicit REMOVE
-            // language=SQLite
-            let clear_statement1 = db
-                .prepare_v2(
-                    "INSERT OR IGNORE INTO ps_updated_rows(row_type, row_id)
-SELECT row_type, row_id
-FROM ps_oplog
-WHERE bucket = ?1",
-                )
-                .into_db_result(db)?;
-            clear_statement1.bind_int64(1, bucket_id)?;
-            clear_statement1.exec()?;
-
-            let clear_statement2 = db
-                .prepare_v2("DELETE FROM ps_oplog WHERE bucket = ?1")
-                .into_db_result(db)?;
-            clear_statement2.bind_int64(1, bucket_id)?;
-            clear_statement2.exec()?;
-
-            // And we need to re-apply all of those.
-            // We also replace the checksum with the checksum of the CLEAR op.
-            // language=SQLite
-            let clear_statement2 = db.prepare_v2(
-                "UPDATE ps_buckets SET last_applied_op = 0, add_checksum = ?1, op_checksum = 0 WHERE id = ?2",
-            )?;
-            clear_statement2.bind_int64(2, bucket_id)?;
-            clear_statement2.bind_int(1, checksum.bitcast_i32())?;
-            clear_statement2.exec()?;
-
-            add_checksum = Checksum::zero();
-            is_empty = true;
-            op_checksum = Checksum::zero();
-        }
+    #[derive(Deserialize)]
+    struct BucketBatch<'a> {
+        #[serde(borrow)]
+        buckets: Vec<DataLine<'a>>,
     }
 
-    if let Some(last_op) = &last_op {
-        // language=SQLite
-        let statement = db.prepare_v2(
-            "UPDATE ps_buckets
-                SET last_op = ?2,
-                    add_checksum = (add_checksum + ?3) & 0xffffffff,
-                    op_checksum = (op_checksum + ?4) & 0xffffffff,
-                    count_since_last = count_since_last + ?5
-            WHERE id = ?1",
-        )?;
-        statement.bind_int64(1, bucket_id)?;
-        statement.bind_int64(2, *last_op)?;
-        statement.bind_int(3, add_checksum.bitcast_i32())?;
-        statement.bind_int(4, op_checksum.bitcast_i32())?;
-        statement.bind_int(5, added_ops)?;
-
-        statement.exec()?;
+    let batch: BucketBatch = serde_json::from_str(data)?;
+    for line in &batch.buckets {
+        insert_bucket_operations(db, line)?;
     }
 
     Ok(())

crates/core/src/sync/checkpoint.rs

@@ -1,13 +1,9 @@
 use alloc::{string::String, vec::Vec};
 use num_traits::Zero;
-use serde::Deserialize;
 
 use crate::{
     error::SQLiteError,
-    sync::{
-        line::{BucketChecksum, Checkpoint},
-        BucketPriority, Checksum,
-    },
+    sync::{line::BucketChecksum, BucketPriority, Checksum},
 };
 use sqlite_nostd::{self as sqlite, Connection, ResultCode};
 

crates/core/src/sync/line.rs

@@ -1,6 +1,4 @@
 use alloc::borrow::Cow;
-use alloc::string::String;
-use alloc::string::ToString;
 use alloc::vec::Vec;
 use serde::Deserialize;
 

crates/core/src/sync/mod.rs

@@ -2,6 +2,7 @@ mod bucket_priority;
 pub mod checkpoint;
 mod checksum;
 pub mod line;
+pub mod operations;
 
 pub use bucket_priority::BucketPriority;
 pub use checksum::Checksum;