feat(model-client): lazy loading support for JS

slisson · slisson · commit 71dc86bd823b · 2025-03-22T08:36:47.000+01:00
diff --git a/kotlin-utils/src/commonMain/kotlin/org/modelix/kotlin/utils/RunBlocking.kt b/kotlin-utils/src/commonMain/kotlin/org/modelix/kotlin/utils/RunBlocking.kt
@@ -0,0 +1,3 @@
+package org.modelix.kotlin.utils
+
+expect fun <R> runBlockingIfJvm(body: suspend () -> R): R
diff --git a/kotlin-utils/src/jsMain/kotlin/org/modelix/kotlin/utils/RunBlocking.js.kt b/kotlin-utils/src/jsMain/kotlin/org/modelix/kotlin/utils/RunBlocking.js.kt
@@ -0,0 +1,5 @@
+package org.modelix.kotlin.utils
+
+actual fun <R> runBlockingIfJvm(body: suspend () -> R): R {
+    throw UnsupportedOperationException("runBlocking not support by JS")
+}
diff --git a/kotlin-utils/src/jvmMain/kotlin/org/modelix/kotlin/utils/RunBlocking.jvm.kt b/kotlin-utils/src/jvmMain/kotlin/org/modelix/kotlin/utils/RunBlocking.jvm.kt
@@ -0,0 +1,9 @@
+package org.modelix.kotlin.utils
+
+import kotlinx.coroutines.runBlocking
+
+actual fun <R> runBlockingIfJvm(body: suspend () -> R): R {
+    return runBlocking {
+        body()
+    }
+}
diff --git a/model-client/src/commonMain/kotlin/org/modelix/model/client2/LazyLoading.kt b/model-client/src/commonMain/kotlin/org/modelix/model/client2/LazyLoading.kt
@@ -0,0 +1,37 @@
+package org.modelix.model.client2
+
+import org.modelix.model.IVersion
+import org.modelix.model.async.BulkAsyncStore
+import org.modelix.model.async.CachingAsyncStore
+import org.modelix.model.lazy.BranchReference
+import org.modelix.model.lazy.CLVersion
+import org.modelix.model.lazy.CacheConfiguration
+import org.modelix.model.lazy.RepositoryId
+
+/**
+ * This function loads parts of the model lazily while it is iterated and limits the amount of data that is cached on
+ * the client side.
+ *
+ * IModelClientV2#loadVersion eagerly loads the whole model. For large models this can be slow and requires lots of
+ * memory.
+ */
+fun IModelClientV2.lazyLoadVersion(repositoryId: RepositoryId, versionHash: String, config: CacheConfiguration = CacheConfiguration()): IVersion {
+    val store = BulkAsyncStore(
+        CachingAsyncStore(
+            ModelClientAsStore(this, repositoryId),
+            cacheSize = config.cacheSize,
+        ),
+        batchSize = config.requestBatchSize,
+    )
+    return store.getStreamExecutor().query {
+        CLVersion.Companion.tryLoadFromHash(versionHash, store).assertNotEmpty { "Version not found: $versionHash" }
+    }
+}
+
+/**
+ * An overload of [IModelClientV2.lazyLoadVersion] that reads the current version hash of the branch from the server and
+ * then loads that version with lazy loading support.
+ */
+suspend fun IModelClientV2.lazyLoadVersion(branchRef: BranchReference, config: CacheConfiguration = CacheConfiguration()): IVersion {
+    return lazyLoadVersion(branchRef.repositoryId, pullHash(branchRef), config)
+}
diff --git a/model-client/src/commonMain/kotlin/org/modelix/model/client2/ModelClientAsStore.kt b/model-client/src/commonMain/kotlin/org/modelix/model/client2/ModelClientAsStore.kt
@@ -0,0 +1,73 @@
+package org.modelix.model.client2
+
+import kotlinx.coroutines.flow.flow
+import org.modelix.model.IKeyValueStore
+import org.modelix.model.async.AsyncStoreAsLegacyDeserializingStore
+import org.modelix.model.async.IAsyncObjectStore
+import org.modelix.model.async.ObjectHash
+import org.modelix.model.lazy.IDeserializingKeyValueStore
+import org.modelix.model.lazy.RepositoryId
+import org.modelix.model.persistent.HashUtil
+import org.modelix.model.persistent.IKVValue
+import org.modelix.streams.IStream
+import org.modelix.streams.IStreamExecutor
+import org.modelix.streams.SimpleStreamExecutor
+import org.modelix.streams.withFlows
+
+class ModelClientAsStore(client: IModelClientV2, val repositoryId: RepositoryId) : IAsyncObjectStore {
+    private val client: IModelClientV2Internal = client as IModelClientV2Internal
+
+    override fun getLegacyKeyValueStore(): IKeyValueStore {
+        throw UnsupportedOperationException()
+    }
+
+    override fun getLegacyObjectStore(): IDeserializingKeyValueStore {
+        return AsyncStoreAsLegacyDeserializingStore(this)
+    }
+
+    override fun <T : Any> getIfCached(key: ObjectHash<T>): T? {
+        return null
+    }
+
+    override fun <T : Any> get(key: ObjectHash<T>): IStream.ZeroOrOne<T> {
+        return getAllAsStream(IStream.of(key)).map {
+            checkNotNull(it.second) { "Entry not found: ${key.hash}" } as T
+        }.exactlyOne()
+    }
+
+    override fun getAllAsStream(keys: IStream.Many<ObjectHash<*>>): IStream.Many<Pair<ObjectHash<*>, Any?>> {
+        return keys.toList().flatMap { keysAsList ->
+            getAllAsMap(keysAsList).flatMapIterable { it.entries }.map { it.key to it.value }
+        }
+    }
+
+    override fun getAllAsMap(keys: List<ObjectHash<*>>): IStream.One<Map<ObjectHash<*>, Any?>> {
+        return IStream.fromFlow(
+            flow {
+                val serializedObjects = client.getObjects(repositoryId, keys.asSequence().map { it.hash })
+                val deserializedObjects = keys.associateWith {
+                    serializedObjects[it.hash]?.let { p1 -> it.deserializer(p1) }
+                }
+                emit(deserializedObjects)
+            },
+        ).exactlyOne()
+    }
+
+    override fun putAll(entries: Map<ObjectHash<*>, IKVValue>): IStream.Zero {
+        return IStream.fromFlow<Nothing>(
+            flow {
+                client.pushObjects(
+                    repositoryId,
+                    entries.asSequence().map { (key, value) ->
+                        require(HashUtil.isSha256(key.hash)) { "Only immutable objects are allowed: $key -> $value" }
+                        key.hash to value.serialize()
+                    },
+                )
+            },
+        ).drainAll()
+    }
+
+    override fun getStreamExecutor(): IStreamExecutor {
+        return SimpleStreamExecutor().withFlows()
+    }
+}
diff --git a/model-client/src/jvmMain/kotlin/org/modelix/model/client2/LazyLoading.kt b/model-client/src/jvmMain/kotlin/org/modelix/model/client2/LazyLoading.kt
diff --git a/model-datastructure/src/commonMain/kotlin/org/modelix/model/async/BulkAsyncStore.kt b/model-datastructure/src/commonMain/kotlin/org/modelix/model/async/BulkAsyncStore.kt
@@ -7,8 +7,6 @@ import org.modelix.streams.BulkRequestStreamExecutor
 import org.modelix.streams.IBulkExecutor
 import org.modelix.streams.IStream
 import org.modelix.streams.IStreamExecutor
-import org.modelix.streams.SimpleStreamExecutor
-import org.modelix.streams.withSequences
 
 class BulkAsyncStore(
     val store: IAsyncObjectStore,
@@ -18,13 +16,11 @@ class BulkAsyncStore(
     private val bulkExecutor = BulkRequestStreamExecutor<ObjectHash<*>, Any?>(
         object : IBulkExecutor<ObjectHash<*>, Any?> {
             override fun execute(keys: List<ObjectHash<*>>): Map<ObjectHash<*>, Any?> {
-                @Suppress("DEPRECATION")
-                return SimpleStreamExecutor().withSequences().query { store.getAllAsMap(keys) }
+                return store.getStreamExecutor().query { store.getAllAsMap(keys) }
             }
 
             override suspend fun executeSuspending(keys: List<ObjectHash<*>>): Map<ObjectHash<*>, Any?> {
-                @Suppress("DEPRECATION")
-                return SimpleStreamExecutor().withSequences().querySuspending { store.getAllAsMap(keys) }
+                return store.getStreamExecutor().querySuspending { store.getAllAsMap(keys) }
             }
         },
         batchSize = batchSize,
diff --git a/model-server/src/test/kotlin/org/modelix/model/server/LazyLoadingTest.kt b/model-server/src/test/kotlin/org/modelix/model/server/LazyLoadingTest.kt
@@ -81,7 +81,7 @@ class LazyLoadingTest {
     @Ignore @Test fun compare_cache_size_3200() = compare_cache_size(100_000, 22, 48, 0, 212, 1859, 0)
     private fun compare_cache_size(cacheSize: Int, vararg expected: Int) = runLazyLoadingTest(DepthFirstSearchPattern, 1_000, cacheSize, 50, 50, *expected)
 
-    @Ignore @Test fun compare_prefetch_size_0() = compare_prefetch_size(0, 22, 2055, 2073, 22, 2055, 2073)
+    @Test fun compare_prefetch_size_0() = compare_prefetch_size(0, 22, 2055, 2073, 22, 2055, 2073)
     @Ignore @Test fun compare_prefetch_size_2() = compare_prefetch_size(2, 22, 1028, 1046, 38, 2056, 2092)
     @Ignore @Test fun compare_prefetch_size_4() = compare_prefetch_size(3, 22, 707, 717, 53, 2121, 2151)
     @Ignore @Test fun compare_prefetch_size_10() = compare_prefetch_size(10, 22, 379, 406, 115, 3773, 4013)
diff --git a/streams/src/commonMain/kotlin/org/modelix/streams/FlowStreamBuilder.kt b/streams/src/commonMain/kotlin/org/modelix/streams/FlowStreamBuilder.kt
@@ -27,6 +27,7 @@ import kotlinx.coroutines.flow.take
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.flow.withIndex
 import kotlinx.coroutines.flow.zip
+import org.modelix.kotlin.utils.runBlockingIfJvm
 import kotlin.coroutines.coroutineContext
 
 class FlowStreamBuilder(executor: IStreamExecutorProvider) : IStreamBuilder, IStreamExecutorProvider by executor {
@@ -91,7 +92,11 @@ class FlowStreamBuilder(executor: IStreamExecutorProvider) : IStreamBuilder, ISt
         override fun toList(): IStream.One<List<E>> = Wrapper(flow { emit(wrapped.toList()) })
         override fun asSequence(): Sequence<E> = throw UnsupportedOperationException()
 
-        override fun iterateSynchronous(visitor: (E) -> Unit) = throw UnsupportedOperationException()
+        override fun iterateSynchronous(visitor: (E) -> Unit) {
+            runBlockingIfJvm {
+                wrapped.collect { visitor(it) }
+            }
+        }
         override suspend fun iterateSuspending(visitor: suspend (E) -> Unit) {
             wrapped.collect(visitor)
         }
@@ -107,7 +112,7 @@ class FlowStreamBuilder(executor: IStreamExecutorProvider) : IStreamBuilder, ISt
             )
         }
         override fun executeSynchronous() {
-            throw UnsupportedOperationException()
+            runBlockingIfJvm { wrapped.collect() }
         }
 
         override fun andThen(other: IStream.Zero): IStream.Zero {
@@ -147,7 +152,15 @@ class FlowStreamBuilder(executor: IStreamExecutorProvider) : IStreamBuilder, ISt
 
     inner class Wrapper<E>(wrapped: Flow<E>) : WrapperBase<E>(wrapped), IStream.One<E> {
         override fun getAsync(onError: ((Throwable) -> Unit)?, onSuccess: ((E) -> Unit)?) {
-            throw UnsupportedOperationException()
+            runBlockingIfJvm {
+                try {
+                    wrapped.collect {
+                        onSuccess?.invoke(it)
+                    }
+                } catch (ex: Throwable) {
+                    onError?.invoke(ex)
+                }
+            }
         }
 
         override fun <R> flatMapOne(mapper: (E) -> IStream.One<R>): IStream.One<R> {
@@ -191,7 +204,7 @@ class FlowStreamBuilder(executor: IStreamExecutorProvider) : IStreamBuilder, ISt
         }
 
         override fun getSynchronous(): E {
-            return throw UnsupportedOperationException()
+            return runBlockingIfJvm { wrapped.toList().single() }
         }
 
         override suspend fun getSuspending(): E {

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+package org.modelix.kotlin.utils`
	`2`	`+`
	`3`	`+expect fun <R> runBlockingIfJvm(body: suspend () -> R): R`