Do not split transactions when uploading data

Author: Radiokot

core/src/commonMain/kotlin/com/powersync/bucket/BucketStorage.kt

@@ -14,6 +14,11 @@ internal interface BucketStorage {
 
     fun nextCrudItem(transaction: PowerSyncTransaction): CrudEntry?
 
+    fun getCrudItemsByTransactionId(
+        transactionId: Int,
+        transaction: PowerSyncTransaction,
+    ): List<CrudEntry>
+
     suspend fun hasCrud(): Boolean
 
     fun hasCrud(transaction: PowerSyncTransaction): Boolean

core/src/commonMain/kotlin/com/powersync/bucket/BucketStorageImpl.kt

@@ -41,13 +41,24 @@ internal class BucketStorageImpl(
         return id ?: throw IllegalStateException("Client ID not found")
     }
 
-    override suspend fun nextCrudItem(): CrudEntry? = db.getOptional(sql = nextCrudQuery, mapper = nextCrudMapper)
+    override suspend fun nextCrudItem(): CrudEntry? = db.getOptional(sql = nextCrudQuery, mapper = crudEntryMapper)
 
     override fun nextCrudItem(transaction: PowerSyncTransaction): CrudEntry? =
-        transaction.getOptional(sql = nextCrudQuery, mapper = nextCrudMapper)
+        transaction.getOptional(sql = nextCrudQuery, mapper = crudEntryMapper)
+
+    override fun getCrudItemsByTransactionId(
+        transactionId: Int,
+        transaction: PowerSyncTransaction,
+    ): List<CrudEntry> =
+        transaction.getAll(
+            sql = transactionCrudQuery,
+            parameters = listOf(transactionId),
+            mapper = crudEntryMapper,
+        )
 
     private val nextCrudQuery = "SELECT id, tx_id, data FROM ${InternalTable.CRUD} ORDER BY id ASC LIMIT 1"
-    private val nextCrudMapper: (SqlCursor) -> CrudEntry = { cursor ->
+    private val transactionCrudQuery = "SELECT id, tx_id, data FROM ${InternalTable.CRUD} WHERE tx_id = ? ORDER BY id ASC"
+    private val crudEntryMapper: (SqlCursor) -> CrudEntry = { cursor ->
         CrudEntry.fromRow(
             CrudRow(
                 id = cursor.getString(0)!!,

core/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt

@@ -298,15 +298,10 @@ internal class PowerSyncDatabaseImpl(
                 if (txId == null) {
                     listOf(entry)
                 } else {
-                    transaction.getAll("SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT 1") {
-                        CrudEntry.fromRow(
-                            CrudRow(
-                                id = it.getString("id"),
-                                data = it.getString("data"),
-                                txId = it.getLongOptional("tx_id")?.toInt(),
-                            ),
-                        )
-                    }
+                    bucketStorage.getCrudItemsByTransactionId(
+                        transactionId = txId,
+                        transaction = transaction,
+                    )
                 }
 
             return@readTransaction CrudTransaction(