feat(model-client): new filter parameter for pull

Clients can now specify which objects they need, instead of receiving the whole history.

Author: slisson

model-client/src/commonMain/kotlin/org/modelix/model/client2/IModelClientV2.kt

@@ -2,6 +2,7 @@ package org.modelix.model.client2
 
 import org.modelix.kotlin.utils.DeprecationInfo
 import org.modelix.model.IVersion
+import org.modelix.model.ObjectDeltaFilter
 import org.modelix.model.api.IIdGenerator
 import org.modelix.model.api.INode
 import org.modelix.model.lazy.BranchReference
@@ -55,7 +56,11 @@ interface IModelClientV2 {
      */
     suspend fun push(branch: BranchReference, version: IVersion, baseVersion: IVersion?): IVersion
 
-    suspend fun pull(branch: BranchReference, lastKnownVersion: IVersion?): IVersion
+    suspend fun pull(
+        branch: BranchReference,
+        lastKnownVersion: IVersion?,
+        filter: ObjectDeltaFilter = ObjectDeltaFilter(),
+    ): IVersion
 
     suspend fun pullIfExists(branch: BranchReference): IVersion?
 

model-client/src/commonMain/kotlin/org/modelix/model/client2/ModelClientV2.kt

@@ -38,6 +38,7 @@ import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.launch
 import org.modelix.kotlin.utils.DeprecationInfo
 import org.modelix.model.IVersion
+import org.modelix.model.ObjectDeltaFilter
 import org.modelix.model.api.IBranch
 import org.modelix.model.api.IIdGenerator
 import org.modelix.model.api.INode
@@ -371,7 +372,7 @@ class ModelClientV2(
         return chunkEntries
     }
 
-    override suspend fun pull(branch: BranchReference, lastKnownVersion: IVersion?): IVersion {
+    override suspend fun pull(branch: BranchReference, lastKnownVersion: IVersion?, filter: ObjectDeltaFilter): IVersion {
         require(lastKnownVersion is CLVersion?)
         return httpClient.prepareGet {
             url {
@@ -380,6 +381,9 @@ class ModelClientV2(
                 if (lastKnownVersion != null) {
                     parameters["lastKnown"] = lastKnownVersion.getContentHash()
                 }
+                if (filter != ObjectDeltaFilter()) {
+                    parameters["filter"] = filter.toJson()
+                }
             }
             useVersionStreamFormat()
         }.execute { response ->

model-datastructure/build.gradle.kts

@@ -1,6 +1,7 @@
 plugins {
     `maven-publish`
     `modelix-kotlin-multiplatform`
+    alias(libs.plugins.kotlin.serialization)
 }
 
 kotlin {
@@ -12,6 +13,7 @@ kotlin {
                 implementation(libs.kotlin.coroutines.core)
                 implementation(libs.kotlin.logging)
                 implementation(libs.kotlin.datetime)
+                implementation(libs.kotlin.serialization.json)
             }
         }
         val commonTest by getting {

model-datastructure/src/commonMain/kotlin/org/modelix/model/ObjectDeltaFilter.kt

@@ -0,0 +1,38 @@
+package org.modelix.model
+
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.Json
+
+@Serializable
+data class ObjectDeltaFilter(
+    /**
+     * Hashes of version objects that are already available and should be excluded.
+     * The traversal stops at these hashes and all transitively referenced objects are also excluded.
+     */
+    val knownVersions: Set<String> = emptySet(),
+
+    /**
+     * Objects required to access the operations of a version. They are usually only required for merging versions.
+     * If a client doesn't do local merges, but let the server do the merge, the operations are never accessed and can
+     * be excluded.
+     */
+    val includeOperations: Boolean = true,
+
+    /**
+     * If false, then the requested version is the only included version object.
+     */
+    val includeHistory: Boolean = true,
+
+    /**
+     * If false, then only the version objects are included, but not their tree data.
+     */
+    val includeTrees: Boolean = true,
+) {
+    constructor(baseVersion: String?) : this(knownVersions = setOfNotNull(baseVersion))
+
+    fun toJson() = Json.encodeToString(this)
+
+    companion object {
+        fun fromJson(serializedJson: String): ObjectDeltaFilter = Json.decodeFromString(serializedJson)
+    }
+}

model-datastructure/src/commonMain/kotlin/org/modelix/model/lazy/CLVersion.kt

@@ -5,6 +5,7 @@ import kotlinx.datetime.Instant
 import kotlinx.datetime.TimeZone
 import kotlinx.datetime.toInstant
 import org.modelix.model.IVersion
+import org.modelix.model.ObjectDeltaFilter
 import org.modelix.model.VersionMerger
 import org.modelix.model.api.IIdGenerator
 import org.modelix.model.api.INodeReference
@@ -26,7 +27,6 @@ import org.modelix.model.persistent.OperationsList
 import org.modelix.model.persistent.getAllObjects
 import org.modelix.streams.IStream
 import org.modelix.streams.flatten
-import org.modelix.streams.notNull
 import org.modelix.streams.plus
 import kotlin.jvm.JvmName
 
@@ -303,52 +303,51 @@ class CLVersion : IVersion {
         }
     }
 
-    fun getParents(stopAt: CLVersion?): List<CLVersion> {
-        if (stopAt != null && this.getContentHash() == stopAt.getContentHash()) {
-            return emptyList()
-        }
-        val ancestors = if (isMerge()) {
-            listOf(getMergedVersion1()!!, getMergedVersion2()!!)
-        } else {
-            listOfNotNull(baseVersion)
-        }
-        return ancestors.filter { stopAt == null || it.getContentHash() != stopAt.getContentHash() }
-    }
-
-    fun collectAncestors(stopAt: CLVersion?, result: MutableMap<String, CLVersion>) {
-        if (stopAt != null && this.getContentHash() == stopAt.getContentHash()) return
+    fun collectAncestors(stopAt: Set<String>, result: MutableMap<String, CLVersion>) {
+        if (stopAt.contains(this.getContentHash())) return
         if (result.contains(getContentHash())) return
         result[getContentHash()] = this
-        for (parent in getParents(stopAt)) {
+        for (parent in listOfNotNull(baseVersion, getMergedVersion1(), getMergedVersion2())) {
             parent.collectAncestors(stopAt, result)
         }
     }
 }
 
 fun CLVersion.fullDiff(baseVersion: CLVersion?): IStream.Many<IKVValue> {
-    val history = historyDiff(baseVersion)
-    return history.plus(
-        history.flatMap { version ->
+    return diff(ObjectDeltaFilter(knownVersions = setOfNotNull(baseVersion?.getContentHash())))
+}
+
+fun CLVersion.diff(filter: ObjectDeltaFilter): IStream.Many<IKVValue> {
+    val history = historyDiff(filter)
+    return history.flatMap { version ->
+        var result: IStream.Many<IKVValue> = IStream.of(version)
+        if (filter.includeTrees) {
             val baseVersion = version.baseVersion?.getValue(asyncStore) ?: IStream.of(null)
             val currentVersion = version.treeHash.getValue(asyncStore)
-            val treeDiff = currentVersion.zipWith(baseVersion) { v, b ->
+            result += currentVersion.zipWith(baseVersion) { v, b ->
                 if (b == null) v.getAllObjects(asyncStore) else v.objectDiff(b, asyncStore)
             }.flatten()
-            if (version.operationsHash != null) {
-                val operations = version.operationsHash.getValue(asyncStore).flatMap { it.getAllObjects(asyncStore) }
-                treeDiff.plus(operations)
-            } else {
-                treeDiff
-            }
-        },
-    )
+        }
+        if (filter.includeOperations && version.operationsHash != null) {
+            result += version.operationsHash.getValue(asyncStore).flatMap { it.getAllObjects(asyncStore) }
+        }
+        result
+    }
 }
 
-fun CLVersion.historyDiff(baseVersion: CLVersion?): IStream.Many<CPVersion> {
-    val commonBase = VersionMerger.commonBaseVersion(this, baseVersion)
-    val history = LinkedHashMap<String, CLVersion>()
-    collectAncestors(commonBase, history)
-    return IStream.many(history.values.map { it.data })
+fun CLVersion.historyDiff(filter: ObjectDeltaFilter): IStream.Many<CPVersion> {
+    if (filter.includeHistory) {
+        val knownVersions = asyncStore.getStreamExecutor().query {
+            IStream.many(filter.knownVersions).flatMap { CLVersion.tryLoadFromHash(it, asyncStore) }.toList()
+        }
+        val commonBases = knownVersions.mapNotNull { VersionMerger.commonBaseVersion(this, it) }
+            .map { it.getContentHash() }.toSet()
+        val history = LinkedHashMap<String, CLVersion>()
+        collectAncestors(stopAt = commonBases, result = history)
+        return IStream.many(history.values.map { it.data })
+    } else {
+        return IStream.of(this.data)
+    }
 }
 
 @Suppress("UNCHECKED_CAST")

model-server-openapi/specifications/model-server-v2.yaml

@@ -145,6 +145,11 @@ paths:
           required: false
           schema:
             type: string
+        - name: filter
+          in: "query"
+          required: false
+          schema:
+            type: string
         - name: repository
           in: "path"
           required: true

model-server/src/main/kotlin/org/modelix/model/server/handlers/IRepositoriesManager.kt

@@ -1,5 +1,6 @@
 package org.modelix.model.server.handlers
 
+import org.modelix.model.ObjectDeltaFilter
 import org.modelix.model.async.IAsyncObjectStore
 import org.modelix.model.lazy.BranchReference
 import org.modelix.model.lazy.CLVersion
@@ -52,7 +53,9 @@ interface IRepositoriesManager {
 
     @RequiresTransaction
     fun mergeChanges(branch: BranchReference, newVersionHash: String): String
-    suspend fun computeDelta(repository: RepositoryId?, versionHash: String, baseVersionHash: String?): ObjectData
+    suspend fun computeDelta(repository: RepositoryId?, versionHash: String, baseVersionHash: String?): ObjectData =
+        computeDelta(repository, versionHash, ObjectDeltaFilter(knownVersions = setOfNotNull(baseVersionHash)))
+    suspend fun computeDelta(repository: RepositoryId?, versionHash: String, filter: ObjectDeltaFilter): ObjectData
 
     /**
      * The data of a repository is stored separately from other repositories, but that wasn't always the case.

model-server/src/main/kotlin/org/modelix/model/server/handlers/ModelReplicationServer.kt

@@ -27,6 +27,7 @@ import org.modelix.authorization.checkPermission
 import org.modelix.authorization.getUserName
 import org.modelix.authorization.hasPermission
 import org.modelix.authorization.requiresLogin
+import org.modelix.model.ObjectDeltaFilter
 import org.modelix.model.api.IBranch
 import org.modelix.model.api.PBranch
 import org.modelix.model.api.TreeAsBranch
@@ -110,6 +111,7 @@ class ModelReplicationServer(
         repository: String,
         branch: String,
         lastKnown: String?,
+        filter: String?,
     ) {
         checkPermission(ModelServerPermissionSchema.repository(repository).branch(branch).pull)
         val branchRef = repositoryId(repository).getBranchReference(branch)
@@ -118,13 +120,20 @@ class ModelReplicationServer(
         val versionHash = runRead {
             repositoriesManager.getVersionHash(branchRef) ?: throw BranchNotFoundException(branchRef)
         }
-        call.respondDelta(RepositoryId(repository), versionHash, lastKnown)
+        call.respondDelta(RepositoryId(repository), versionHash, parseFilter(filter, lastKnown))
+    }
+
+    private fun parseFilter(filterAsJson: String?, lastKnown: String?): ObjectDeltaFilter {
+        return (filterAsJson?.let { ObjectDeltaFilter.fromJson(it) } ?: ObjectDeltaFilter()).let {
+            if (lastKnown == null) it else it.copy(knownVersions = it.knownVersions + lastKnown)
+        }
     }
 
     override suspend fun RoutingContext.getRepositoryBranchV1(
         repository: String,
         branch: String,
         lastKnown: String?,
+        filter: String?,
     ) {
         checkPermission(ModelServerPermissionSchema.repository(repository).branch(branch).pull)
         val branchRef = repositoryId(repository).getBranchReference(branch)
@@ -189,7 +198,7 @@ class ModelReplicationServer(
                 legacyGlobalStorage ?: false,
             )
         }
-        call.respondDelta(RepositoryId(repository), initialVersion.getContentHash(), null)
+        call.respondDelta(RepositoryId(repository), initialVersion.getContentHash(), ObjectDeltaFilter())
     }
 
     override suspend fun RoutingContext.deleteRepository(repository: String) {
@@ -223,7 +232,7 @@ class ModelReplicationServer(
         val mergedHash = runWrite {
             repositoriesManager.mergeChanges(branchRef, deltaFromClient.versionHash)
         }
-        call.respondDelta(RepositoryId(repository), mergedHash, deltaFromClient.versionHash)
+        call.respondDelta(RepositoryId(repository), mergedHash, ObjectDeltaFilter(deltaFromClient.versionHash))
     }
 
     override suspend fun RoutingContext.pollRepositoryBranch(
@@ -234,7 +243,7 @@ class ModelReplicationServer(
         checkPermission(ModelServerPermissionSchema.repository(repository).branch(branch).pull)
         val branchRef = repositoryId(repository).getBranchReference(branch)
         val newVersionHash = repositoriesManager.pollVersionHash(branchRef, lastKnown)
-        call.respondDelta(RepositoryId(repository), newVersionHash, lastKnown)
+        call.respondDelta(RepositoryId(repository), newVersionHash, ObjectDeltaFilter(lastKnown))
     }
 
     override suspend fun RoutingContext.postRepositoryObjectsGetAll(repository: String) {
@@ -282,7 +291,7 @@ class ModelReplicationServer(
         if (repositoriesManager.getVersion(repositoryId(repository), versionHash) == null) {
             throw VersionNotFoundException(versionHash)
         }
-        call.respondDelta(RepositoryId(repository), versionHash, lastKnown)
+        call.respondDelta(RepositoryId(repository), versionHash, ObjectDeltaFilter(lastKnown))
     }
 
     override suspend fun RoutingContext.postRepositoryBranchQuery(
@@ -384,40 +393,40 @@ class ModelReplicationServer(
         if (runRead { stores.getGlobalStoreClient()[versionHash] } == null) {
             throw VersionNotFoundException(versionHash)
         }
-        call.respondDelta(null, versionHash, lastKnown)
+        call.respondDelta(null, versionHash, ObjectDeltaFilter(lastKnown))
     }
 
-    private suspend fun ApplicationCall.respondDelta(repositoryId: RepositoryId?, versionHash: String, baseVersionHash: String?) {
+    private suspend fun ApplicationCall.respondDelta(repositoryId: RepositoryId?, versionHash: String, filter: ObjectDeltaFilter) {
         val expectedTypes = request.acceptItems().map { ContentType.parse(it.value) }
         return if (expectedTypes.any { it.match(VersionDeltaStreamV2.CONTENT_TYPE) }) {
-            respondDeltaAsObjectStreamV2(repositoryId, versionHash, baseVersionHash)
+            respondDeltaAsObjectStreamV2(repositoryId, versionHash, filter)
         } else if (expectedTypes.any { it.match(VersionDeltaStream.CONTENT_TYPE) }) {
-            respondDeltaAsObjectStreamV1(repositoryId, versionHash, baseVersionHash, false)
+            respondDeltaAsObjectStreamV1(repositoryId, versionHash, filter, false)
         } else if (expectedTypes.any { it.match(ContentType.Application.Json) }) {
-            respondDeltaAsJson(repositoryId, versionHash, baseVersionHash)
+            respondDeltaAsJson(repositoryId, versionHash, filter)
         } else {
-            respondDeltaAsObjectStreamV1(repositoryId, versionHash, baseVersionHash, true)
+            respondDeltaAsObjectStreamV1(repositoryId, versionHash, filter, true)
         }
     }
 
-    private suspend fun ApplicationCall.respondDeltaAsJson(repositoryId: RepositoryId?, versionHash: String, baseVersionHash: String?) {
+    private suspend fun ApplicationCall.respondDeltaAsJson(repositoryId: RepositoryId?, versionHash: String, filter: ObjectDeltaFilter) {
         val delta = VersionDelta(
             versionHash,
-            baseVersionHash,
-            objectsMap = repositoriesManager.computeDelta(repositoryId, versionHash, baseVersionHash).asMap(),
+            filter.knownVersions.firstOrNull(),
+            objectsMap = repositoriesManager.computeDelta(repositoryId, versionHash, filter).asMap(),
         )
         respond(delta)
     }
 
     private suspend fun ApplicationCall.respondDeltaAsObjectStreamV1(
         repositoryId: RepositoryId?,
         versionHash: String,
-        baseVersionHash: String?,
+        filter: ObjectDeltaFilter,
         plainText: Boolean,
     ) {
         // Call `computeDelta` before starting to respond.
         // It could already throw an exception, and in that case we do not want a successful response status.
-        val objectData = repositoriesManager.computeDelta(repositoryId, versionHash, baseVersionHash)
+        val objectData = repositoriesManager.computeDelta(repositoryId, versionHash, filter)
         val contentType = if (plainText) ContentType.Text.Plain else VersionDeltaStream.CONTENT_TYPE
         respondBytesWriter(contentType) {
             this.useClosingWithoutCause {
@@ -434,8 +443,8 @@ class ModelReplicationServer(
         }
     }
 
-    private suspend fun ApplicationCall.respondDeltaAsObjectStreamV2(repositoryId: RepositoryId?, versionHash: String, baseVersionHash: String?) {
-        val objectData = repositoriesManager.computeDelta(repositoryId, versionHash, baseVersionHash)
+    private suspend fun ApplicationCall.respondDeltaAsObjectStreamV2(repositoryId: RepositoryId?, versionHash: String, filter: ObjectDeltaFilter) {
+        val objectData = repositoriesManager.computeDelta(repositoryId, versionHash, filter)
         respondBytesWriter(VersionDeltaStreamV2.CONTENT_TYPE) {
             this.useClosingWithoutCause {
                 VersionDeltaStreamV2.encodeVersionDeltaStreamV2(this, versionHash, objectData.asStream())