Support custom schemas

Author: simolus3

powersync/src/db/internal.rs

@@ -9,13 +9,13 @@ use std::{
     task::{Context, Poll},
 };
 
+use crate::schema::SchemaOrCustom;
 use crate::{
     db::{
         core_extension::CoreExtensionVersion, pool::LeasedConnection, streams::SyncStreamTracker,
     },
     env::PowerSyncEnvironment,
     error::PowerSyncError,
-    schema::Schema,
     sync::{MAX_OP_ID, coordinator::SyncCoordinator, status::SyncStatus, status::SyncStatusData},
     util::SharedFuture,
 };
@@ -28,7 +28,7 @@ pub struct InnerPowerSyncState {
     /// The schema passed to the database.
     ///
     /// This is forwarded to the sync client for raw tables.
-    pub schema: Arc<Schema>,
+    pub schema: Arc<SchemaOrCustom>,
     /// A container for the current sync status.
     pub status: SyncStatus,
     /// A collection of currently-referenced sync stream subscriptions.
@@ -41,7 +41,11 @@ pub struct InnerPowerSyncState {
 }
 
 impl InnerPowerSyncState {
-    pub fn new(env: PowerSyncEnvironment, schema: Schema, sync: &Arc<SyncCoordinator>) -> Self {
+    pub fn new(
+        env: PowerSyncEnvironment,
+        schema: SchemaOrCustom,
+        sync: &Arc<SyncCoordinator>,
+    ) -> Self {
         Self {
             env,
             did_initialize: SharedFuture::new(),
@@ -73,7 +77,9 @@ impl InnerPowerSyncState {
     }
 
     fn update_schema_internal(&self, conn: &Connection) -> Result<(), PowerSyncError> {
-        self.schema.validate()?;
+        if let SchemaOrCustom::Schema(schema) = self.schema.as_ref() {
+            schema.validate()?;
+        };
 
         let serialized_schema = serde_json::to_string(&self.schema)?;
         conn.prepare("SELECT powersync_replace_schema(?)")?

powersync/src/db/mod.rs

@@ -4,6 +4,7 @@ use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
 use crate::db::async_support::AsyncDatabaseTasks;
+use crate::schema::SchemaOrCustom;
 use crate::sync::coordinator::SyncCoordinator;
 use crate::{
     CrudTransaction, SyncOptions,
@@ -13,7 +14,6 @@ use crate::{
     },
     env::PowerSyncEnvironment,
     error::PowerSyncError,
-    schema::Schema,
     sync::{download::DownloadActor, status::SyncStatusData, upload::UploadActor},
 };
 use futures_lite::stream::{once, once_future};
@@ -47,11 +47,18 @@ impl PowerSyncDatabase {
     const PS_DATA_PREFIX: &'static str = "ps_data__";
     const PS_DATA_LOCAL_PREFIX: &'static str = "ps_data_local__";
 
-    pub fn new(env: PowerSyncEnvironment, schema: Schema) -> Self {
+    /// Creates a new PowerSync database using the connection pool and HTTP client from the given
+    /// [PowerSyncEnvironment].
+    ///
+    /// On first use, the database will instantiate a PowerSync schema. Typically, a [Schema]
+    /// instance would be passed for schemas. For cases where the schema is defined externally as a
+    /// JSON object understood by the PowerSync SQLite core extension, it can also be passed as a
+    /// [serde_json::value::RawValue] reference.
+    pub fn new(env: PowerSyncEnvironment, schema: impl Into<SchemaOrCustom>) -> Self {
         let coordinator = Arc::new(SyncCoordinator::default());
 
         Self {
-            inner: Arc::new(InnerPowerSyncState::new(env, schema, &coordinator)),
+            inner: Arc::new(InnerPowerSyncState::new(env, schema.into(), &coordinator)),
             sync: coordinator,
         }
     }

powersync/src/db/schema.rs

@@ -4,6 +4,7 @@ use std::collections::HashSet;
 use serde::{Serialize, Serializer, ser::SerializeStruct};
 
 use crate::error::PowerSyncError;
+use crate::util::SerializedJsonObject;
 
 type SchemaString = Cow<'static, str>;
 
@@ -502,6 +503,30 @@ impl TrackPreviousValues {
     }
 }
 
+#[derive(Serialize, Debug)]
+#[serde(untagged)]
+pub enum SchemaOrCustom {
+    /// A schema created in Rust.
+    Schema(Schema),
+    /// A pre-serialized schema we just forward to the core extension.
+    Custom(Box<SerializedJsonObject>),
+}
+
+impl From<Schema> for SchemaOrCustom {
+    fn from(value: Schema) -> Self {
+        Self::Schema(value)
+    }
+}
+
+impl From<&serde_json::value::RawValue> for SchemaOrCustom {
+    fn from(value: &serde_json::value::RawValue) -> Self {
+        Self::Custom(unsafe {
+            // Safety: The core extension will report an error if this isn't a JSON object.
+            SerializedJsonObject::from_owned_value(value.to_owned())
+        })
+    }
+}
+
 #[cfg(test)]
 mod test {
     use crate::schema::{Column, RawTable, RawTableSchema, Schema, Table, TrackPreviousValues};

powersync/src/sync/download/sync_iteration.rs

@@ -9,11 +9,11 @@ use rusqlite::{
 use serde::Serialize;
 use serde_json::value::RawValue;
 
+use crate::schema::SchemaOrCustom;
 use crate::{
     SyncOptions,
     db::internal::InnerPowerSyncState,
     error::PowerSyncError,
-    schema::Schema,
     sync::{
         download::http::sync_stream,
         instruction::{CloseSyncStream, Instruction, LogSeverity},
@@ -226,7 +226,7 @@ impl ToSql for PowerSyncControlArgument {
 #[derive(Debug, Serialize)]
 pub struct StartDownloadIteration {
     pub parameters: serde_json::Value,
-    pub schema: Arc<Schema>,
+    pub schema: Arc<SchemaOrCustom>,
     pub include_defaults: bool,
     pub active_streams: Vec<StreamKey>,
 }

powersync/src/util/mod.rs

@@ -19,7 +19,7 @@ pub struct SerializedJsonObject {
 
 impl SerializedJsonObject {
     /// Safety: This must only be called for raw values that are known to be objects.
-    unsafe fn from_owned_value(raw: Box<RawValue>) -> Box<Self> {
+    pub unsafe fn from_owned_value(raw: Box<RawValue>) -> Box<Self> {
         unsafe {
             // Safety: Identical representation.
             std::mem::transmute(raw)

powersync/tests/crud_test.rs

@@ -330,7 +330,7 @@ fn raw_table_crud_trigger() {
         let serialized_table = serde_json::to_string(&schema.raw_tables[0]).unwrap();
 
         let test = DatabaseTest::new();
-        let db = PowerSyncDatabase::new(test.in_memory(), Default::default());
+        let db = PowerSyncDatabase::new(test.in_memory(), Schema::default());
 
         {
             let mut writer = db.writer().await.unwrap();

powersync/tests/database_test.rs

@@ -2,9 +2,12 @@ use std::sync::Arc;
 
 use async_oneshot::oneshot;
 use futures_lite::{StreamExt, future};
+use powersync::PowerSyncDatabase;
 use powersync::error::PowerSyncError;
+use powersync::schema::{Column, Schema, Table};
 use powersync_test_utils::{DatabaseTest, UserRow, execute, query_all};
 use rusqlite::params;
+use serde_json::value::RawValue;
 use serde_json::{Value, json};
 
 #[test]
@@ -232,3 +235,31 @@ fn test_watch_statement() {
         );
     });
 }
+
+#[test]
+fn test_external_schema() {
+    let test = DatabaseTest::new();
+    let schema = {
+        let mut original = Schema::default();
+        original
+            .tables
+            .push(Table::create("users", vec![Column::text("name")], |_| {}));
+
+        let serialized = serde_json::to_string(&original).unwrap();
+        let custom: Box<RawValue> = serde_json::from_str(&serialized).unwrap();
+        custom
+    };
+
+    // Passing a pre-serialized schema should still work correctly.
+    let db = PowerSyncDatabase::new(test.in_memory(), schema.as_ref());
+    future::block_on(execute(
+        &db,
+        "INSERT INTO users (id, name) VALUES (uuid(), ?)",
+        params!["User"],
+    ));
+
+    future::block_on(async {
+        let rows = query_all(&db, "SELECT name FROM users", params![]).await;
+        assert_eq!(rows, json!([{"name": "User"}]));
+    });
+}