Support direct tables

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## 0.1.2 (unreleased)
 
 - Update PowerSync SQLite core extension to 0.5.4 ([release notes](https://github.com/powersync-ja/powersync-sqlite-core/releases/tag/v0.5.4)).
+- Support [direct tables](https://docs.powersync.com/client-sdks/advanced/performance-optimization#direct-tables).
 
 ## 0.1.1
 

powersync/src/db/schema.rs

@@ -77,6 +77,19 @@ pub struct Table {
     pub indexes: Vec<Index>,
     #[serde(flatten)]
     pub options: TableOptions,
+    /// Marks this table as direct, meaning that it stores data directly as SQLite rows instead of
+    /// using an intermediate view around a JSON store.
+    ///
+    /// Direct tables are still managed by the PowerSync SDK, which is responsible for creating,
+    /// altering and dropping them.
+    /// You can query raw tables like JSON-based tables, and mutations are consistently tracked as
+    /// pending uploads.
+    /// For more information on direct tables, see [the documentation](https://docs.powersync.com/client-sdks/advanced/performance-optimization#direct-tables).
+    ///
+    /// Direct tables can't use a view name override.
+    ///
+    /// Direct tables are currently experimental.
+    pub direct: bool,
 }
 
 impl Table {
@@ -94,6 +107,7 @@ impl Table {
             columns,
             indexes: vec![],
             options: TableOptions::default(),
+            direct: false,
         };
         build(&mut table);
         table
@@ -153,6 +167,13 @@ impl Table {
             }
         }
 
+        if self.direct && self.view_name_override.is_some() {
+            return Err(PowerSyncError::argument_error(format!(
+                "Direct table {} can't use view name override",
+                self.name
+            )));
+        }
+
         Ok(())
     }
 
@@ -624,6 +645,15 @@ mod test {
         assert!(table.validate().is_err());
     }
 
+    #[test]
+    fn direct_table() {
+        let mut table = Table::create("users", vec![Column::text("name")], |tbl| tbl.direct = true);
+        assert!(table.validate().is_ok());
+
+        table.view_name_override = Some("forbidden".into());
+        assert!(table.validate().is_err());
+    }
+
     #[test]
     fn invalid_raw_table_missing_statements() {
         let mut table = RawTable::with_schema("users", RawTableSchema::default());

powersync/tests/sync_test.rs

@@ -17,6 +17,7 @@ use powersync::{
     PowerSyncCredentials, PowerSyncDatabase, RequestsCheckpointMode, StreamPriority,
     StreamSubscription, StreamSubscriptionOptions, SyncOptions, SyncStatusData,
     error::PowerSyncError,
+    schema::{Column, Schema, Table},
 };
 use powersync_test_utils::{
     DatabaseTest, execute,
@@ -35,8 +36,12 @@ struct SyncStreamTest {
 
 impl SyncStreamTest {
     fn new() -> Self {
+        Self::with_schema(DatabaseTest::default_schema())
+    }
+
+    fn with_schema(schema: Schema) -> Self {
         let test = DatabaseTest::new();
-        let db = test.in_memory_database();
+        let db = PowerSyncDatabase::new(test.in_memory(), schema);
 
         Self { db, test }
     }
@@ -555,6 +560,80 @@ fn does_not_apply_checkpoint_with_pending_local_data() {
     });
 }
 
+#[test]
+fn direct_tables() {
+    // This is mostly a smoke test, the core extension manages direct tables and covers things like
+    // migrations.
+    let mut schema = Schema::default();
+    schema
+        .tables
+        .push(Table::create("users", vec![Column::text("name")], |tbl| {
+            tbl.direct = true
+        }));
+    let sync = SyncStreamTest::with_schema(schema);
+
+    sync.run(async {
+        assert_eq!(
+            query_all(
+                &sync.db,
+                "SELECT type FROM pragma_table_list('users')",
+                params![]
+            )
+            .await,
+            json!([{"type": "table"}])
+        );
+    });
+
+    sync.connect();
+    sync.run(async {
+        let request = sync.test.http.receive_requests.recv().await.unwrap();
+        sync.wait_for_status(|s| s.is_connected()).await;
+
+        request
+            .send_checkpoint(Checkpoint::single_bucket("a", 1, None))
+            .await;
+        request
+            .channel
+            .send(SyncLine::Data(DataLine {
+                bucket: "a",
+                data: vec![OplogEntry {
+                    checksum: 0,
+                    op_id: 1,
+                    op: OpType::PUT,
+                    object_id: Some("user".to_string()),
+                    object_type: Some("users"),
+                    subkey: None,
+                    data: Some(r#"{"name": "synced user"}"#),
+                }],
+            }))
+            .await
+            .unwrap();
+        request.send_checkpoint_complete(1, None).await;
+        sync.wait_for_status(|s| s.has_synced()).await;
+
+        assert_eq!(
+            query_all(&sync.db, "SELECT * FROM users", params![]).await,
+            json!([{"id": "user", "name": "synced user"}])
+        );
+        sync.db.disconnect().await;
+
+        execute(&sync.db, "UPDATE users SET name = ?", params!["local user"]).await;
+        assert_eq!(
+            query_all(&sync.db, "SELECT * FROM users", params![]).await,
+            json!([{"id": "user", "name": "local user"}])
+        );
+
+        let tx = sync.db.next_crud_transaction().await.unwrap().unwrap();
+        assert_eq!(tx.crud.len(), 1);
+        let entry = &tx.crud[0];
+        assert_eq!(entry.table, "users");
+        assert_eq!(
+            entry.data.clone().map(serde_json::Value::Object),
+            Some(json!({"name": "local user"}))
+        );
+    });
+}
+
 #[test]
 fn fetching_credentials_does_not_hold_the_download_writer_lease() {
     struct WriterUsingConnector {