Replace streams Sequence/Flow/Reaktive backends with a round-based Step engine

.github/workflows/build.yaml

@@ -67,7 +67,7 @@ jobs:
           node-version-file: '.nvmrc'
       - name: Combine SARIF files
         run: |
-          npx @microsoft/sarif-multitool merge --merge-runs --output-file merged.sarif  $(find . -iname '*.sarif*')
+          npx @microsoft/sarif-multitool merge --output-file merged.sarif  $(find . -iname '*.sarif*')
         env:
           # Disables globalization support.
           # This makes the @microsoft/sarif-multitool work without ICU package installed.

datastructures/src/commonMain/kotlin/org/modelix/datastructures/btree/BTree.kt

@@ -13,10 +13,10 @@ data class BTree<K, V>(val root: BTreeNode<K, V>) : IStreamExecutorProvider by r
     fun validate() {
         graph.getStreamExecutor().query {
             root.validate(true)
-            check(root.getEntries().toList().getBlocking(graph).map { it.key }.toSet().size == root.getEntries().map { it.key }.count().getBlocking(graph)) {
+            check(root.getEntries().toList().getBlocking().map { it.key }.toSet().size == root.getEntries().map { it.key }.count().getBlocking()) {
                 "duplicate entries: $root"
             }
-            check(root.getEntries().map { it.key }.toList().getBlocking(graph).sortedWith(root.config.keyConfiguration) == root.getEntries().map { it.key }.toList().getBlocking(graph)) {
+            check(root.getEntries().map { it.key }.toList().getBlocking().sortedWith(root.config.keyConfiguration) == root.getEntries().map { it.key }.toList().getBlocking()) {
                 "not sorted: $this"
             }
             IStream.of(Unit)

datastructures/src/commonTest/kotlin/org/modelix/datastructures/BTreeTest.kt

@@ -72,7 +72,7 @@ class BTreeTest {
 
         assertEquals(
             (100L..200L).map { it to it * 2 },
-            tree.getAll((100L..200L)).toList().getBlocking(tree),
+            tree.getAll((100L..200L)).toList().getBlocking(),
         )
     }
 }

datastructures/src/commonTest/kotlin/org/modelix/datastructures/HamtCollisionTest.kt

@@ -27,19 +27,19 @@ class HamtCollisionTest {
         )
         var tree: HamtTree<Long, String> = HamtTree(HamtInternalNode.createEmpty(config))
 
-        tree = tree.put(0b000000000, "a").getBlocking(tree)
-        tree = tree.put(0b000000001, "b").getBlocking(tree)
-        tree = tree.put(0b000000011, "c").getBlocking(tree)
-        tree = tree.put(0b100000000, "d").getBlocking(tree)
-        tree = tree.put(0b100000001, "e").getBlocking(tree)
-        tree = tree.put(0b100000011, "f").getBlocking(tree)
-
-        assertEquals("a", tree.get(0b000000000).getBlocking(tree))
-        assertEquals("b", tree.get(0b000000001).getBlocking(tree))
-        assertEquals("c", tree.get(0b000000011).getBlocking(tree))
-        assertEquals("d", tree.get(0b100000000).getBlocking(tree))
-        assertEquals("e", tree.get(0b100000001).getBlocking(tree))
-        assertEquals("f", tree.get(0b100000011).getBlocking(tree))
+        tree = tree.put(0b000000000, "a").getBlocking()
+        tree = tree.put(0b000000001, "b").getBlocking()
+        tree = tree.put(0b000000011, "c").getBlocking()
+        tree = tree.put(0b100000000, "d").getBlocking()
+        tree = tree.put(0b100000001, "e").getBlocking()
+        tree = tree.put(0b100000011, "f").getBlocking()
+
+        assertEquals("a", tree.get(0b000000000).getBlocking())
+        assertEquals("b", tree.get(0b000000001).getBlocking())
+        assertEquals("c", tree.get(0b000000011).getBlocking())
+        assertEquals("d", tree.get(0b100000000).getBlocking())
+        assertEquals("e", tree.get(0b100000001).getBlocking())
+        assertEquals("f", tree.get(0b100000011).getBlocking())
     }
 
     /**
@@ -60,22 +60,22 @@ class HamtCollisionTest {
         )
         var tree: HamtTree<Long, String> = HamtTree(HamtInternalNode.createEmpty(config))
 
-        tree = tree.put(0b00, "a").getBlocking(tree)
-        tree = tree.put(0b01, "b").getBlocking(tree)
-        tree = tree.put(0b10, "c").getBlocking(tree)
-        tree = tree.put(0b11, "d").getBlocking(tree)
+        tree = tree.put(0b00, "a").getBlocking()
+        tree = tree.put(0b01, "b").getBlocking()
+        tree = tree.put(0b10, "c").getBlocking()
+        tree = tree.put(0b11, "d").getBlocking()
 
-        assertEquals("a", tree.get(0b00).getBlocking(tree))
-        assertEquals("b", tree.get(0b01).getBlocking(tree))
-        assertEquals("c", tree.get(0b10).getBlocking(tree))
-        assertEquals("d", tree.get(0b11).getBlocking(tree))
+        assertEquals("a", tree.get(0b00).getBlocking())
+        assertEquals("b", tree.get(0b01).getBlocking())
+        assertEquals("c", tree.get(0b10).getBlocking())
+        assertEquals("d", tree.get(0b11).getBlocking())
 
-        tree = tree.put(0b01, "changed").getBlocking(tree)
+        tree = tree.put(0b01, "changed").getBlocking()
 
-        assertEquals("a", tree.get(0b00).getBlocking(tree))
-        assertEquals("changed", tree.get(0b01).getBlocking(tree))
-        assertEquals("c", tree.get(0b10).getBlocking(tree))
-        assertEquals("d", tree.get(0b11).getBlocking(tree))
+        assertEquals("a", tree.get(0b00).getBlocking())
+        assertEquals("changed", tree.get(0b01).getBlocking())
+        assertEquals("c", tree.get(0b10).getBlocking())
+        assertEquals("d", tree.get(0b11).getBlocking())
     }
 }
 

datastructures/src/commonTest/kotlin/org/modelix/datastructures/HamtGetChangesTest.kt

@@ -0,0 +1,43 @@
+package org.modelix.datastructures
+
+import org.modelix.datastructures.hamt.HamtInternalNode
+import org.modelix.datastructures.hamt.HamtNode
+import org.modelix.datastructures.hamt.HamtTree
+import org.modelix.datastructures.objects.IObjectGraph
+import org.modelix.datastructures.objects.StringDataTypeConfiguration
+import org.modelix.streams.getBlocking
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+/**
+ * Direct coverage of [org.modelix.datastructures.IPersistentMap.getChanges] on the HAMT. Guards that a value change is
+ * reported as [EntryChangedEvent] (not add/remove): this relies on the streams engine evaluating a synchronous prefix
+ * eagerly, which `HamtLeafNode.getChanges` depends on via a `var` set by one stream and read by a later deferred one.
+ */
+class HamtGetChangesTest {
+    private fun newTree(): HamtTree<String, String> {
+        val config = HamtNode.Config(
+            graph = IObjectGraph.FREE_FLOATING,
+            keyConfig = StringDataTypeConfiguration(),
+            valueConfig = StringDataTypeConfiguration(),
+        )
+        return HamtTree(HamtInternalNode.createEmpty(config))
+    }
+
+    @Test
+    fun single_entry_value_change_is_a_change_not_add() {
+        val old = newTree().put("k1", "a").getBlocking()
+        val new = old.put("k1", "b").getBlocking()
+        val changes = new.getChanges(old, changesOnly = false).toList().getBlocking()
+        assertEquals(listOf(EntryChangedEvent("k1", "a", "b")), changes)
+    }
+
+    @Test
+    fun one_changed_among_several() {
+        var old = newTree()
+        for (i in 0 until 20) old = old.put("k$i", "v$i").getBlocking()
+        val new = old.put("k7", "changed").getBlocking()
+        val changes = new.getChanges(old, changesOnly = false).toList().getBlocking()
+        assertEquals(listOf(EntryChangedEvent("k7", "v7", "changed")), changes)
+    }
+}

datastructures/src/commonTest/kotlin/org/modelix/datastructures/HamtTest.kt

@@ -33,21 +33,21 @@ class HamtTest {
                     if (expected.isNotEmpty()) {
                         val key = expected.keys.random(rand)
                         // println("remove $key")
-                        tree = tree.remove(key).getBlocking(tree)
+                        tree = tree.remove(key).getBlocking()
                         expected.remove(key)
                     }
                 }
                 else -> {
                     val key = "k" + rand.nextInt(keyRange).toString()
                     val value = "v" + rand.nextInt(valueRange).toString()
                     // println("insert $key -> $value")
-                    tree = tree.put(key, value).getBlocking(tree)
+                    tree = tree.put(key, value).getBlocking()
                     expected[key] = value
                 }
             }
 
             for (entry in expected.entries) {
-                assertEquals(entry.value, tree.get(entry.key).getBlocking(tree), "for key ${entry.key}")
+                assertEquals(entry.value, tree.get(entry.key).getBlocking(), "for key ${entry.key}")
             }
         }
     }

datastructures/src/commonTest/kotlin/org/modelix/datastructures/PatriciaTrieTest.kt

@@ -25,10 +25,10 @@ class PatriciaTrieTest {
 
         fun assertTree() {
             val expectedEntries = addedEntries.associateWith { values[it]!! }
-            val actualEntries = tree.getAll().toList().getBlocking(tree).toMap()
+            val actualEntries = tree.getAll().toList().getBlocking().toMap()
             assertEquals(expectedEntries, actualEntries)
 
-            val expectedTree = addedEntries.fold(initialTree) { acc, it -> acc.put(it, values[it]!!).getBlocking(tree) }
+            val expectedTree = addedEntries.fold(initialTree) { acc, it -> acc.put(it, values[it]!!).getBlocking() }
             assertEquals(expectedTree, tree)
 
             assertEquals(expectedTree.asObject().getHash(), tree.asObject().getHash())
@@ -39,13 +39,13 @@ class PatriciaTrieTest {
                 val key = removedEntries.random(rand)
                 removedEntries.remove(key)
                 addedEntries.add(key)
-                tree = tree.put(key, values[key]!!).getBlocking(tree)
+                tree = tree.put(key, values[key]!!).getBlocking()
                 assertTree()
             } else {
                 val key = addedEntries.random(rand)
                 removedEntries.add(key)
                 addedEntries.remove(key)
-                tree = tree.remove(key).getBlocking(tree)
+                tree = tree.remove(key).getBlocking()
                 assertTree()
             }
         }
@@ -56,55 +56,55 @@ class PatriciaTrieTest {
     @Test
     fun `slice with shorter prefix and single entry`() {
         var tree: PatriciaTrie<String, String> = PatriciaTrie.withStrings(IObjectGraph.FREE_FLOATING)
-        tree = tree.put("abcdef", "1").getBlocking(tree)
+        tree = tree.put("abcdef", "1").getBlocking()
 
-        assertEquals("1", tree.slice("abc").flatMapZeroOrOne { it.get("abcdef") }.getBlocking(tree))
+        assertEquals("1", tree.slice("abc").flatMapZeroOrOne { it.get("abcdef") }.getBlocking())
     }
 
     @Test
     fun `slice with shorter prefix and two entries`() {
         var tree: PatriciaTrie<String, String> = PatriciaTrie.withStrings(IObjectGraph.FREE_FLOATING)
-        tree = tree.put("abcdef", "1").getBlocking(tree)
-        tree = tree.put("abcdeg", "2").getBlocking(tree)
+        tree = tree.put("abcdef", "1").getBlocking()
+        tree = tree.put("abcdeg", "2").getBlocking()
 
-        assertEquals("1", tree.slice("abc").flatMapZeroOrOne { it.get("abcdef") }.getBlocking(tree))
-        assertEquals("2", tree.slice("abc").flatMapZeroOrOne { it.get("abcdeg") }.getBlocking(tree))
+        assertEquals("1", tree.slice("abc").flatMapZeroOrOne { it.get("abcdef") }.getBlocking())
+        assertEquals("2", tree.slice("abc").flatMapZeroOrOne { it.get("abcdeg") }.getBlocking())
     }
 
     @Test
     fun `slice with prefix between two entries`() {
         var tree: PatriciaTrie<String, String> = PatriciaTrie.withStrings(IObjectGraph.FREE_FLOATING)
-        tree = tree.put("ab", "1").getBlocking(tree)
-        tree = tree.put("abcdef", "2").getBlocking(tree)
+        tree = tree.put("ab", "1").getBlocking()
+        tree = tree.put("abcdef", "2").getBlocking()
 
-        assertEquals(null, tree.slice("abcd").flatMapZeroOrOne { it.get("ab") }.getBlocking(tree))
-        assertEquals("2", tree.slice("abcd").flatMapZeroOrOne { it.get("abcdef") }.getBlocking(tree))
+        assertEquals(null, tree.slice("abcd").flatMapZeroOrOne { it.get("ab") }.getBlocking())
+        assertEquals("2", tree.slice("abcd").flatMapZeroOrOne { it.get("abcdef") }.getBlocking())
     }
 
     @Test
     fun `slice with one before and two after`() {
         var tree: PatriciaTrie<String, String> = PatriciaTrie.withStrings(IObjectGraph.FREE_FLOATING)
-        tree = tree.put("ab", "1").getBlocking(tree)
-        tree = tree.put("abcdef", "2").getBlocking(tree)
-        tree = tree.put("abcdeg", "3").getBlocking(tree)
+        tree = tree.put("ab", "1").getBlocking()
+        tree = tree.put("abcdef", "2").getBlocking()
+        tree = tree.put("abcdeg", "3").getBlocking()
 
-        assertEquals(null, tree.slice("abcd").flatMapZeroOrOne { it.get("ab") }.getBlocking(tree))
-        assertEquals("2", tree.slice("abcd").flatMapZeroOrOne { it.get("abcdef") }.getBlocking(tree))
-        assertEquals("3", tree.slice("abcd").flatMapZeroOrOne { it.get("abcdeg") }.getBlocking(tree))
+        assertEquals(null, tree.slice("abcd").flatMapZeroOrOne { it.get("ab") }.getBlocking())
+        assertEquals("2", tree.slice("abcd").flatMapZeroOrOne { it.get("abcdef") }.getBlocking())
+        assertEquals("3", tree.slice("abcd").flatMapZeroOrOne { it.get("abcdeg") }.getBlocking())
     }
 
     @Test
     fun `slice with two before and two after at existing split`() {
         var tree: PatriciaTrie<String, String> = PatriciaTrie.withStrings(IObjectGraph.FREE_FLOATING)
-        tree = tree.put("a", "0").getBlocking(tree)
-        tree = tree.put("ab", "1").getBlocking(tree)
-        tree = tree.put("abcdef", "2").getBlocking(tree)
-        tree = tree.put("abcdeg", "3").getBlocking(tree)
-
-        assertEquals(null, tree.slice("ab").flatMapZeroOrOne { it.get("a") }.getBlocking(tree))
-        assertEquals("1", tree.slice("ab").flatMapZeroOrOne { it.get("ab") }.getBlocking(tree))
-        assertEquals("2", tree.slice("ab").flatMapZeroOrOne { it.get("abcdef") }.getBlocking(tree))
-        assertEquals("3", tree.slice("ab").flatMapZeroOrOne { it.get("abcdeg") }.getBlocking(tree))
+        tree = tree.put("a", "0").getBlocking()
+        tree = tree.put("ab", "1").getBlocking()
+        tree = tree.put("abcdef", "2").getBlocking()
+        tree = tree.put("abcdeg", "3").getBlocking()
+
+        assertEquals(null, tree.slice("ab").flatMapZeroOrOne { it.get("a") }.getBlocking())
+        assertEquals("1", tree.slice("ab").flatMapZeroOrOne { it.get("ab") }.getBlocking())
+        assertEquals("2", tree.slice("ab").flatMapZeroOrOne { it.get("abcdef") }.getBlocking())
+        assertEquals("3", tree.slice("ab").flatMapZeroOrOne { it.get("abcdeg") }.getBlocking())
     }
 
     @Test
@@ -127,16 +127,16 @@ class PatriciaTrieTest {
             "abcdC",
         )
         for (key in initialKeys) {
-            tree = tree.put(key, "value of $key").getBlocking(tree)
+            tree = tree.put(key, "value of $key").getBlocking()
         }
 
         assertEquals(
             initialKeys.associateWith { "value of $it" },
-            tree.getAll().toList().getBlocking(tree).toMap(),
+            tree.getAll().toList().getBlocking().toMap(),
         )
 
         for (key in initialKeys) {
-            assertEquals("value of $key", tree.get(key).getBlocking(tree))
+            assertEquals("value of $key", tree.get(key).getBlocking())
         }
 
         var replacementTree = PatriciaTrie.withStrings(IObjectGraph.FREE_FLOATING)
@@ -151,16 +151,16 @@ class PatriciaTrieTest {
             "abcdB93",
         )
         for (key in replacementKeys) {
-            replacementTree = replacementTree.put(key, "new value of $key").getBlocking(tree)
+            replacementTree = replacementTree.put(key, "new value of $key").getBlocking()
         }
-        tree = tree.replaceSlice("abcdB", replacementTree).getBlocking(tree)
+        tree = tree.replaceSlice("abcdB", replacementTree).getBlocking()
 
         assertEquals(
             initialKeys.filterNot { it.startsWith("abcdB") }.map { it to "value of $it" }
                 .plus(replacementKeys.map { it to "new value of $it" })
                 .sortedBy { it.first }
                 .joinToString("\n") { "${it.first} -> ${it.second}" },
-            tree.getAll().toList().getBlocking(tree).sortedBy { it.first }.joinToString("\n") { "${it.first} -> ${it.second}" },
+            tree.getAll().toList().getBlocking().sortedBy { it.first }.joinToString("\n") { "${it.first} -> ${it.second}" },
         )
     }
 }

datastructures/src/commonTest/kotlin/org/modelix/datastructures/TreeDiffTest.kt

@@ -29,7 +29,7 @@ class TreeDiffTest {
         val restoringGraph = FullyLoadedObjectGraph()
         val initialDiff = tree.asObject().getDescendantsAndSelf()
             .map { it.getHash() to it.data.serialize() }
-            .toList().getBlocking(tree).toMap()
+            .toList().getBlocking().toMap()
         val restoringConfig = PatriciaTrieConfig(
             restoringGraph,
             StringDataTypeConfiguration(),
@@ -40,11 +40,11 @@ class TreeDiffTest {
         fun assertTree() {
             val diff = tree.asObject().objectDiff(restoredTree.asObject())
                 .map { it.getHash() to it.data.serialize() }
-                .toList().getBlocking(tree).toMap()
+                .toList().getBlocking().toMap()
             restoredTree = PatriciaTrie(restoringGraph.loadObjects(tree.asObject().getHash(), restoringDeserializer, diff))
 
             val expectedEntries = addedEntries.associateWith { values[it]!! }
-            val actualEntries = restoredTree.getAll().toList().getBlocking(tree).toMap()
+            val actualEntries = restoredTree.getAll().toList().getBlocking().toMap()
             assertEquals(expectedEntries, actualEntries)
         }
 
@@ -53,13 +53,13 @@ class TreeDiffTest {
                 val key = removedEntries.random(rand)
                 removedEntries.remove(key)
                 addedEntries.add(key)
-                tree = tree.put(key, values[key]!!).getBlocking(tree)
+                tree = tree.put(key, values[key]!!).getBlocking()
                 assertTree()
             } else {
                 val key = addedEntries.random(rand)
                 removedEntries.add(key)
                 addedEntries.remove(key)
-                tree = tree.remove(key).getBlocking(tree)
+                tree = tree.remove(key).getBlocking()
                 assertTree()
             }
         }
@@ -85,7 +85,7 @@ class TreeDiffTest {
             for (previousTree in (1..5).map { previousTrees.random(rand) }) {
                 val diff = tree.asObject().objectDiff(previousTree.asObject())
                     .map { it.getHash() to it.data.serialize() }
-                    .toList().getBlocking(tree)
+                    .toList().getBlocking()
 
                 val duplicateObjects = diff.groupBy { it.first }.filter { it.value.size > 1 }.map { it.value.first() }
                 assertEquals(emptyList(), duplicateObjects)
@@ -98,13 +98,13 @@ class TreeDiffTest {
                 val key = removedEntries.random(rand)
                 removedEntries.remove(key)
                 addedEntries.add(key)
-                tree = tree.put(key, values[key]!!).getBlocking(tree)
+                tree = tree.put(key, values[key]!!).getBlocking()
                 assertTree()
             } else {
                 val key = addedEntries.random(rand)
                 removedEntries.add(key)
                 addedEntries.remove(key)
-                tree = tree.remove(key).getBlocking(tree)
+                tree = tree.remove(key).getBlocking()
                 assertTree()
             }
         }