A few sync_local simplifications

crates/core/src/schema/table_info.rs

@@ -336,6 +336,7 @@ impl<'de> Deserialize<'de> for PendingStatement {
                 set.insert(match column {
                     PendingStatementValue::Id => "id".to_string(),
                     PendingStatementValue::Column(name) => name.clone(),
+                    PendingStatementValue::Row => continue,
                     PendingStatementValue::Rest => {
                         rest_parameter_positions.push(i);
                         continue;
@@ -366,5 +367,6 @@ pub enum PendingStatementValue {
     /// Bind to a JSON object containing all columns from the synced row that haven't been matched
     /// by other statement values.
     Rest,
-    // TODO: Stuff like a raw object of put data?
+    /// The full JSON object for the row, as received from the PowerSync service.
+    Row,
 }

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

@@ -1009,6 +1009,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]