Avoid passing partial arguments as string to sync_local

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() {

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<BucketPriority>,
+    ) -> impl Iterator<Item = &'a str> {
+        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]

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