Allow using external tables

Author: simolus3

crates/core/src/schema/mod.rs

@@ -5,11 +5,14 @@ use alloc::vec::Vec;
 use serde::Deserialize;
 use sqlite::ResultCode;
 use sqlite_nostd as sqlite;
-pub use table_info::{DiffIncludeOld, Table, TableInfoFlags};
+pub use table_info::{
+    DiffIncludeOld, PendingStatement, PendingStatementValue, RawTableDefinition, Table,
+    TableInfoFlags,
+};
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Default)]
 pub struct Schema {
-    tables: Vec<table_info::Table>,
+    pub tables: Vec<table_info::Table>,
 }
 
 pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {

crates/core/src/schema/table_info.rs

@@ -17,6 +17,8 @@ pub struct Table {
     pub diff_include_old: Option<DiffIncludeOld>,
     #[serde(flatten)]
     pub flags: TableInfoFlags,
+    #[serde(default)]
+    pub raw: Option<RawTableDefinition>,
 }
 
 impl Table {
@@ -229,3 +231,23 @@ impl<'de> Deserialize<'de> for TableInfoFlags {
         )
     }
 }
+
+#[derive(Deserialize)]
+pub struct RawTableDefinition {
+    pub put: PendingStatement,
+    pub delete: PendingStatement,
+}
+
+#[derive(Deserialize)]
+pub struct PendingStatement {
+    pub sql: String,
+    /// This vec should contain an entry for each parameter in [sql].
+    pub params: Vec<PendingStatementValue>,
+}
+
+#[derive(Deserialize)]
+pub enum PendingStatementValue {
+    Id,
+    Column(String),
+    // TODO: Stuff like a raw object of put data?
+}

crates/core/src/sync/interface.rs

@@ -12,6 +12,7 @@ use sqlite_nostd::{self as sqlite, ColumnType};
 use sqlite_nostd::{Connection, Context};
 
 use crate::error::SQLiteError;
+use crate::schema::Schema;
 
 use super::streaming_sync::SyncClient;
 use super::sync_status::DownloadSyncStatus;
@@ -22,6 +23,8 @@ pub struct StartSyncStream {
     /// Bucket parameters to include in the request when opening a sync stream.
     #[serde(default)]
     pub parameters: Option<serde_json::Map<String, serde_json::Value>>,
+    #[serde(default)]
+    pub schema: Schema,
 }
 
 /// A request sent from a client SDK to the [SyncClient] with a `powersync_control` invocation.

crates/core/src/sync/storage_adapter.rs

@@ -9,6 +9,7 @@ use crate::{
     error::SQLiteError,
     ext::SafeManagedStmt,
     operations::delete_bucket,
+    schema::Schema,
     sync::checkpoint::{validate_checkpoint, ChecksumMismatch},
     sync_local::{PartialSyncOperation, SyncOperation},
 };
@@ -145,6 +146,7 @@ impl StorageAdapter {
         &self,
         checkpoint: &OwnedCheckpoint,
         priority: Option<BucketPriority>,
+        schema: &Schema,
     ) -> Result<SyncLocalResult, SQLiteError> {
         let mismatched_checksums =
             validate_checkpoint(checkpoint.buckets.values(), priority, self.db)?;
@@ -201,14 +203,15 @@ impl StorageAdapter {
 
                 // TODO: Avoid this serialization, it's currently used to bind JSON SQL parameters.
                 let serialized_args = serde_json::to_string(&args)?;
-                SyncOperation::new(
+                let mut sync = SyncOperation::new(
                     self.db,
                     Some(PartialSyncOperation {
                         priority,
                         args: &serialized_args,
                     }),
-                )
-                .apply()
+                );
+                sync.use_schema(schema);
+                sync.apply()
             }
         }?;
 

crates/core/src/sync/streaming_sync.rs

@@ -14,7 +14,12 @@ use alloc::{
 };
 use futures_lite::FutureExt;
 
-use crate::{bson, error::SQLiteError, kv::client_id, sync::checkpoint::OwnedBucketChecksum};
+use crate::{
+    bson,
+    error::SQLiteError,
+    kv::client_id,
+    sync::{checkpoint::OwnedBucketChecksum, interface::StartSyncStream},
+};
 use sqlite_nostd::{self as sqlite, ResultCode};
 
 use super::{
@@ -52,7 +57,7 @@ impl SyncClient {
             SyncControlRequest::StartSyncStream(options) => {
                 self.state.tear_down()?;
 
-                let mut handle = SyncIterationHandle::new(self.db, options.parameters)?;
+                let mut handle = SyncIterationHandle::new(self.db, options)?;
                 let instructions = handle.initialize()?;
                 self.state = ClientState::IterationActive(handle);
 
@@ -120,13 +125,10 @@ struct SyncIterationHandle {
 impl SyncIterationHandle {
     /// Creates a new sync iteration in a pending state by preparing statements for
     /// [StorageAdapter] and setting up the initial downloading state for [StorageAdapter] .
-    fn new(
-        db: *mut sqlite::sqlite3,
-        parameters: Option<serde_json::Map<String, serde_json::Value>>,
-    ) -> Result<Self, ResultCode> {
+    fn new(db: *mut sqlite::sqlite3, options: StartSyncStream) -> Result<Self, ResultCode> {
         let runner = StreamingSyncIteration {
             db,
-            parameters,
+            options,
             adapter: StorageAdapter::new(db)?,
             status: SyncStatusContainer::new(),
         };
@@ -191,7 +193,7 @@ impl<'a> ActiveEvent<'a> {
 struct StreamingSyncIteration {
     db: *mut sqlite::sqlite3,
     adapter: StorageAdapter,
-    parameters: Option<serde_json::Map<String, serde_json::Value>>,
+    options: StartSyncStream,
     status: SyncStatusContainer,
 }
 
@@ -244,7 +246,9 @@ impl StreamingSyncIteration {
                 SyncEvent::BinaryLine { data } => bson::from_bytes(data)?,
                 SyncEvent::UploadFinished => {
                     if let Some(checkpoint) = validated_but_not_applied.take() {
-                        let result = self.adapter.sync_local(&checkpoint, None)?;
+                        let result =
+                            self.adapter
+                                .sync_local(&checkpoint, None, &self.options.schema)?;
 
                         match result {
                             SyncLocalResult::ChangesApplied => {
@@ -320,7 +324,9 @@ impl StreamingSyncIteration {
                             ),
                         ));
                     };
-                    let result = self.adapter.sync_local(target, None)?;
+                    let result = self
+                        .adapter
+                        .sync_local(target, None, &self.options.schema)?;
 
                     match result {
                         SyncLocalResult::ChecksumFailure(checkpoint_result) => {
@@ -363,7 +369,9 @@ impl StreamingSyncIteration {
                             ),
                         ));
                     };
-                    let result = self.adapter.sync_local(target, Some(priority))?;
+                    let result =
+                        self.adapter
+                            .sync_local(target, Some(priority), &self.options.schema)?;
 
                     match result {
                         SyncLocalResult::ChecksumFailure(checkpoint_result) => {
@@ -459,7 +467,7 @@ impl StreamingSyncIteration {
             raw_data: true,
             binary_data: true,
             client_id: client_id(self.db)?,
-            parameters: self.parameters.take(),
+            parameters: self.options.parameters.take(),
         };
 
         event