Merge pull request #36 from powersync-ja/dapper-query-engine

Use Dapper as the internal query engine

Author: LucDeCaf

PowerSync/PowerSync.Common/CHANGELOG.md

@@ -1,5 +1,14 @@
 # PowerSync.Common Changelog
 
+## 0.0.8-alpha.1
+- Replaced the old JSON-based method of extracting type information from queries with using Dapper internally for queries, improving memory usage and execution time for querying.
+- Added non-generic overloads for `GetAll()`, `GetOptional()`, `Get()`, `Watch()` which return `dynamic`:
+
+```csharp
+dynamic asset = db.Get("SELECT id, description, make FROM assets");
+Console.WriteLine($"Asset ID: {asset.id}");
+```
+
 ## 0.0.7-alpha.1
 - Added fallback to check the application's root directory for the PowerSync extension - fixing compatibility with WPF/WAP, .NET Framework <= 4.8, and other platforms that flatten DLLs into the base folder.
 - Added `ExecuteBatch()` implementation.

PowerSync/PowerSync.Common/Client/PowerSyncDatabase.cs

@@ -63,10 +63,13 @@ public interface IPowerSyncDatabase : IEventStream<PowerSyncDBEvent>
     Task<NonQueryResult> ExecuteBatch(string query, object?[][]? parameters = null);
 
     Task<T[]> GetAll<T>(string sql, object?[]? parameters = null);
+    Task<dynamic[]> GetAll(string sql, object?[]? parameters = null);
 
     Task<T?> GetOptional<T>(string sql, object?[]? parameters = null);
+    Task<dynamic?> GetOptional(string sql, object?[]? parameters = null);
 
     Task<T> Get<T>(string sql, object?[]? parameters = null);
+    Task<dynamic> Get(string sql, object?[]? parameters = null);
 
     Task<T> ReadLock<T>(Func<ILockContext, Task<T>> fn, DBLockOptions? options = null);
 
@@ -422,7 +425,8 @@ await Database.WriteTransaction(async tx =>
         Closed = true;
     }
 
-    private record UploadQueueStatsResult(int size, int count);
+    private record UploadQueueStatsSizeCountResult(long size, long count);
+    private record UploadQueueStatsCountResult(long count);
     /// <summary>
     /// Get upload queue size estimate and count.
     /// </summary>
@@ -432,23 +436,22 @@ public async Task<UploadQueueStats> GetUploadQueueStats(bool includeSize = false
         {
             if (includeSize)
             {
-                var result = await tx.Get<UploadQueueStatsResult>(
+                var result = await tx.Get<UploadQueueStatsSizeCountResult>(
                     $"SELECT SUM(cast(data as blob) + 20) as size, count(*) as count FROM {PSInternalTable.CRUD}"
                 );
 
                 return new UploadQueueStats(result.count, result.size);
             }
             else
             {
-                var result = await tx.Get<UploadQueueStatsResult>(
+                var result = await tx.Get<UploadQueueStatsCountResult>(
                     $"SELECT count(*) as count FROM {PSInternalTable.CRUD}"
                 );
                 return new UploadQueueStats(result.count);
             }
         });
     }
 
-
     /// <summary>
     /// Get a batch of crud data to upload.
     /// <para />
@@ -468,7 +471,7 @@ public async Task<UploadQueueStats> GetUploadQueueStats(bool includeSize = false
     /// </summary>
     public async Task<CrudBatch?> GetCrudBatch(int limit = 100)
     {
-        var crudResult = await GetAll<CrudEntryJSON>($"SELECT id, tx_id, data FROM {PSInternalTable.CRUD} ORDER BY id ASC LIMIT ?", [limit + 1]);
+        var crudResult = await GetAll<CrudEntryJSON>($"SELECT id, tx_id as transactionId, data FROM {PSInternalTable.CRUD} ORDER BY id ASC LIMIT ?", [limit + 1]);
 
         var all = crudResult.Select(CrudEntry.FromRow).ToList();
 
@@ -510,7 +513,7 @@ public async Task<UploadQueueStats> GetUploadQueueStats(bool includeSize = false
         return await ReadTransaction(async tx =>
         {
             var first = await tx.GetOptional<CrudEntryJSON>(
-            $"SELECT id, tx_id, data FROM {PSInternalTable.CRUD} ORDER BY id ASC LIMIT 1");
+            $"SELECT id, tx_id AS transactionId, data FROM {PSInternalTable.CRUD} ORDER BY id ASC LIMIT 1");
 
             if (first == null)
             {
@@ -528,7 +531,7 @@ public async Task<UploadQueueStats> GetUploadQueueStats(bool includeSize = false
             else
             {
                 var result = await tx.GetAll<CrudEntryJSON>(
-                    $"SELECT id, tx_id, data FROM {PSInternalTable.CRUD} WHERE tx_id = ? ORDER BY id ASC",
+                    $"SELECT id, tx_id as transactionId, data FROM {PSInternalTable.CRUD} WHERE tx_id = ? ORDER BY id ASC",
                     [txId]);
 
                 all = result.Select(CrudEntry.FromRow).ToList();
@@ -596,17 +599,36 @@ public async Task<T[]> GetAll<T>(string query, object?[]? parameters = null)
         return await Database.GetAll<T>(query, parameters);
     }
 
+    public async Task<dynamic[]> GetAll(string query, object?[]? parameters = null)
+    {
+        await WaitForReady();
+        return await Database.GetAll(query, parameters);
+    }
+
     public async Task<T?> GetOptional<T>(string query, object?[]? parameters = null)
     {
         await WaitForReady();
         return await Database.GetOptional<T>(query, parameters);
     }
+
+    public async Task<dynamic?> GetOptional(string query, object?[]? parameters = null)
+    {
+        await WaitForReady();
+        return await Database.GetOptional(query, parameters);
+    }
+
     public async Task<T> Get<T>(string query, object?[]? parameters = null)
     {
         await WaitForReady();
         return await Database.Get<T>(query, parameters);
     }
 
+    public async Task<dynamic> Get(string query, object?[]? parameters = null)
+    {
+        await WaitForReady();
+        return await Database.Get(query, parameters);
+    }
+
     public async Task<T> ReadLock<T>(Func<ILockContext, Task<T>> fn, DBLockOptions? options = null)
     {
         await WaitForReady();
@@ -650,14 +672,32 @@ public async Task<T> WriteTransaction<T>(Func<ITransaction, Task<T>> fn, DBLockO
     /// Source tables are automatically detected using <c>EXPLAIN QUERY PLAN</c>.
     /// </summary>
     public Task Watch<T>(string query, object?[]? parameters, WatchHandler<T> handler, SQLWatchOptions? options = null)
+        => WatchInternal(query, parameters, handler, options, GetAll<T>);
+
+    /// <summary>
+    /// Executes a read query every time the source tables are modified.
+    /// <para />
+    /// Use <see cref="SQLWatchOptions.ThrottleMs"/> to specify the minimum interval between queries.
+    /// Source tables are automatically detected using <c>EXPLAIN QUERY PLAN</c>.
+    /// </summary>
+    public Task Watch(string query, object?[]? parameters, WatchHandler<dynamic> handler, SQLWatchOptions? options = null)
+        => WatchInternal(query, parameters, handler, options, GetAll);
+
+    private Task WatchInternal<T>(
+        string query,
+        object?[]? parameters,
+        WatchHandler<T> handler,
+        SQLWatchOptions? options,
+        Func<string, object?[]?, Task<T[]>> getter
+    )
     {
         var tcs = new TaskCompletionSource<bool>();
         Task.Run(async () =>
         {
             try
             {
                 var resolvedTables = await ResolveTables(query, parameters, options);
-                var result = await GetAll<T>(query, parameters);
+                var result = await getter(query, parameters);
                 handler.OnResult(result);
 
                 OnChange(new WatchOnChangeHandler
@@ -666,7 +706,7 @@ public Task Watch<T>(string query, object?[]? parameters, WatchHandler<T> handle
                     {
                         try
                         {
-                            var result = await GetAll<T>(query, parameters);
+                            var result = await getter(query, parameters);
                             handler.OnResult(result);
                         }
                         catch (Exception ex)
@@ -691,7 +731,16 @@ public Task Watch<T>(string query, object?[]? parameters, WatchHandler<T> handle
         return tcs.Task;
     }
 
-    private record ExplainedResult(string opcode, int p2, int p3);
+    private class ExplainedResult
+    {
+        public int addr = 0;
+        public string opcode = "";
+        public int p1 = 0;
+        public int p2 = 0;
+        public int p3 = 0;
+        public string p4 = "";
+        public int p5 = 0;
+    }
     private record TableSelectResult(string tbl_name);
     public async Task<string[]> ResolveTables(string sql, object?[]? parameters = null, SQLWatchOptions? options = null)
     {
@@ -718,7 +767,6 @@ public async Task<string[]> ResolveTables(string sql, object?[]? parameters = nu
                 resolvedTables.Add(POWERSYNC_TABLE_MATCH.Replace(table.tbl_name, ""));
             }
         }
-
         return [.. resolvedTables];
     }
 

PowerSync/PowerSync.Common/DB/Crud/UploadQueueStatus.cs

@@ -1,8 +1,8 @@
 namespace PowerSync.Common.DB.Crud;
 
-public class UploadQueueStats(int count, long? size = null)
+public class UploadQueueStats(long count, long? size = null)
 {
-    public int Count { get; set; } = count;
+    public long Count { get; set; } = count;
 
     public long? Size { get; set; } = size;
 

PowerSync/PowerSync.Common/DB/IDBAdapter.cs

@@ -31,12 +31,15 @@ public interface IDBGetUtils
 {
     // Execute a read-only query and return results.
     Task<T[]> GetAll<T>(string sql, object?[]? parameters = null);
+    Task<dynamic[]> GetAll(string sql, object?[]? parameters = null);
 
     // Execute a read-only query and return the first result, or null if the ResultSet is empty.
     Task<T?> GetOptional<T>(string sql, object?[]? parameters = null);
+    Task<dynamic?> GetOptional(string sql, object?[]? parameters = null);
 
     // Execute a read-only query and return the first result, error if the ResultSet is empty.
     Task<T> Get<T>(string sql, object?[]? parameters = null);
+    Task<dynamic> Get(string sql, object?[]? parameters = null);
 }
 
 public interface ILockContext : IDBGetUtils

PowerSync/PowerSync.Common/MDSQLite/MDSQLiteAdapter.cs

@@ -154,22 +154,36 @@ public async Task<NonQueryResult> ExecuteBatch(string query, object?[][]? parame
         return await WriteTransaction((ctx) => ctx.ExecuteBatch(query, parameters));
     }
 
-    public async Task<T> Get<T>(string sql, object?[]? parameters = null)
+    public async Task<T[]> GetAll<T>(string sql, object?[]? parameters = null)
     {
-        return await ReadLock((ctx) => ctx.Get<T>(sql, parameters));
-        ;
+        return await ReadLock((ctx) => ctx.GetAll<T>(sql, parameters));
     }
 
-    public async Task<T[]> GetAll<T>(string sql, object?[]? parameters = null)
+    public async Task<dynamic[]> GetAll(string sql, object?[]? parameters = null)
     {
-        return await ReadLock((ctx) => ctx.GetAll<T>(sql, parameters));
+        return await ReadLock((ctx) => ctx.GetAll(sql, parameters));
     }
 
     public async Task<T?> GetOptional<T>(string sql, object?[]? parameters = null)
     {
         return await ReadLock((ctx) => ctx.GetOptional<T>(sql, parameters));
     }
 
+    public async Task<dynamic?> GetOptional(string sql, object?[]? parameters = null)
+    {
+        return await ReadLock((ctx) => ctx.GetOptional(sql, parameters));
+    }
+
+    public async Task<T> Get<T>(string sql, object?[]? parameters = null)
+    {
+        return await ReadLock((ctx) => ctx.Get<T>(sql, parameters));
+    }
+
+    public async Task<dynamic> Get(string sql, object?[]? parameters = null)
+    {
+        return await ReadLock((ctx) => ctx.Get(sql, parameters));
+    }
+
     public async Task<T> ReadTransaction<T>(Func<ITransaction, Task<T>> fn, DBLockOptions? options = null)
     {
         return await ReadLock((ctx) => InternalTransaction(new MDSQLiteTransaction(readConnection!)!, fn));
@@ -311,18 +325,33 @@ public Task<NonQueryResult> ExecuteBatch(string query, object?[][]? parameters =
         return connection.ExecuteBatch(query, parameters);
     }
 
-    public Task<T> Get<T>(string sql, object?[]? parameters = null)
+    public Task<T[]> GetAll<T>(string sql, object?[]? parameters = null)
     {
-        return connection.Get<T>(sql, parameters);
+        return connection.GetAll<T>(sql, parameters);
     }
 
-    public Task<T[]> GetAll<T>(string sql, object?[]? parameters = null)
+    public Task<dynamic[]> GetAll(string sql, object?[]? parameters = null)
     {
-        return connection.GetAll<T>(sql, parameters);
+        return connection.GetAll(sql, parameters);
     }
 
     public Task<T?> GetOptional<T>(string sql, object?[]? parameters = null)
     {
         return connection.GetOptional<T>(sql, parameters);
     }
+
+    public Task<dynamic?> GetOptional(string sql, object?[]? parameters = null)
+    {
+        return connection.GetOptional(sql, parameters);
+    }
+
+    public Task<T> Get<T>(string sql, object?[]? parameters = null)
+    {
+        return connection.Get<T>(sql, parameters);
+    }
+
+    public Task<dynamic> Get(string sql, object?[]? parameters = null)
+    {
+        return connection.Get(sql, parameters);
+    }
 }