diff --git a/crates/core/src/sync/storage_adapter.rs b/crates/core/src/sync/storage_adapter.rs index c79ccf9..6721a92 100644 --- a/crates/core/src/sync/storage_adapter.rs +++ b/crates/core/src/sync/storage_adapter.rs @@ -2,7 +2,6 @@ use core::fmt::Display; use alloc::{rc::Rc, string::ToString, vec::Vec}; use powersync_sqlite_nostd::{self as sqlite}; -use serde::Serialize; use crate::{ error::{PowerSyncError, Result}, @@ -273,51 +272,22 @@ WHERE bucket = ?1", self.persist_last_seen_checkpoint_request_id(*checkpoint_request_id)?; } - #[derive(Serialize)] - struct PartialArgs<'a> { - priority: BucketPriority, - buckets: Vec<&'a str>, - } let now = self.now()?; - let sync_result = match priority { - None => { - let mut sync = SyncOperation::new(state, self.db, None, now); - sync.use_schema(schema); - sync.apply() - } - Some(priority) => { - let args = PartialArgs { + let mut sync = match priority { + None => SyncOperation::new(state, self.db, None, now), + Some(priority) => SyncOperation::new( + state, + self.db, + Some(PartialSyncOperation { priority, - buckets: checkpoint - .buckets - .values() - .filter_map(|item| { - if item.is_in_priority(Some(priority)) { - Some(item.bucket.as_str()) - } else { - None - } - }) - .collect(), - }; - - // TODO: Avoid this serialization, it's currently used to bind JSON SQL parameters. - let serialized_args = - serde_json::to_string(&args).map_err(PowerSyncError::internal)?; - let mut sync = SyncOperation::new( - state, - self.db, - Some(PartialSyncOperation { - priority, - args: &serialized_args, - }), - now, - ); - sync.use_schema(schema); - sync.apply() - } - }?; + involved_buckets: checkpoint.list_buckets(Some(priority)).collect(), + }), + now, + ), + }; + sync.use_schema(schema); + let sync_result = sync.apply()?; if sync_result == 1 { if priority.is_none() { diff --git a/crates/core/src/sync/streaming_sync.rs b/crates/core/src/sync/streaming_sync.rs index 986d160..26035dd 100644 --- a/crates/core/src/sync/streaming_sync.rs +++ b/crates/core/src/sync/streaming_sync.rs @@ -1127,6 +1127,19 @@ impl OwnedCheckpoint { self.last_op_id = diff.last_op_id; self.write_checkpoint = diff.write_checkpoint; } + + pub fn list_buckets<'a>( + &'a self, + min_priority: Option, + ) -> impl Iterator { + self.buckets.values().filter_map(move |item| { + if item.is_in_priority(min_priority) { + Some(item.bucket.as_str()) + } else { + None + } + }) + } } /// A transition representing pending changes between [StreamingSyncIteration::prepare_handling_sync_line] diff --git a/crates/core/src/sync/sync_local.rs b/crates/core/src/sync/sync_local.rs index c973412..bfee930 100644 --- a/crates/core/src/sync/sync_local.rs +++ b/crates/core/src/sync/sync_local.rs @@ -2,6 +2,7 @@ use alloc::collections::btree_map::BTreeMap; use alloc::format; use alloc::rc::Rc; use alloc::string::{String, ToString}; +use alloc::vec::Vec; use serde::Serialize; use serde::ser::SerializeMap; @@ -24,9 +25,7 @@ use powersync_sqlite_nostd::{self as sqlite, Destructor}; pub struct PartialSyncOperation<'a> { /// The lowest priority part of the partial sync operation. pub priority: BucketPriority, - /// The JSON-encoded arguments passed by the client SDK. This includes the priority and a list - /// of bucket names in that (and higher) priorities. - pub args: &'a str, + pub involved_buckets: Vec<&'a str>, } pub struct SyncOperation<'a> { @@ -285,8 +284,8 @@ SELECT -- We filter out duplicates using the GROUP BY below. WITH involved_buckets (id) AS MATERIALIZED ( - SELECT id FROM ps_buckets WHERE ?1 IS NULL - OR name IN (SELECT value FROM json_each(json_extract(?1, '$.buckets'))) + SELECT id FROM ps_buckets + WHERE name IN (SELECT value FROM json_each(?1)) ), updated_rows AS ( SELECT b.row_type, b.row_id FROM ps_buckets AS buckets @@ -314,7 +313,10 @@ SELECT -- Group for (2) GROUP BY b.row_type, b.row_id;", )?; - stmt.bind_text(1, partial.args, Destructor::STATIC)?; + + let bucket_ids = serde_json::to_string(&partial.involved_buckets) + .map_err(PowerSyncError::internal)?; + stmt.bind_text(1, &bucket_ids, Destructor::TRANSIENT)?; stmt } @@ -325,16 +327,17 @@ SELECT match &self.partial { Some(partial) => { // language=SQLite - let updated = self - .db - .prepare_v2( "\ + let updated = self.db.prepare_v2( + "\ UPDATE ps_buckets SET last_applied_op = last_op - WHERE last_applied_op != last_op AND - name IN (SELECT value FROM json_each(json_extract(?1, '$.buckets')))", - )?; - updated.bind_text(1, partial.args, Destructor::STATIC)?; - updated.exec()?; + WHERE last_applied_op != last_op AND name = ?", + )?; + + for bucket in &partial.involved_buckets { + updated.bind_text(1, bucket, Destructor::STATIC)?; + updated.exec()?; + } } None => { // language=SQLite