feat(compiler): CRDT-maps implementation

aqua-src/antithesis.aqua

@@ -1,10 +1,128 @@
-aqua Job declares *
+aqua StreamMapTest declares *
 
-use "declare"
+export testGetFunc, testGetStreamFunc, testKeysFunc, testKeysStreamFunc
+export testContainsFunc, testForFunc, testParSeqMap, testForTupleFunc
 
-export timeout
+import "builtin.aqua"
 
-func timeout() -> AquaName.Worker:
-  w <- AquaName.getWorker()
-  a = w.host_id
-  <- w
+func testGetFunc() -> []string, []string, []string, u32:
+  streamMap: %string
+  key = "key"
+  resEmpty = streamMap.get(key)
+  streamMap <<- key, "first value"
+  resFirst = streamMap.get(key)
+  streamMap <<- key, "second value"
+  resSecond = streamMap.get(key)
+  <- resEmpty, resFirst, resSecond, resSecond.length
+
+func testGetStreamFunc() -> []string, string, string:
+  streamMap: %string
+  key = "key"
+  resEmptyStream = streamMap.getStream(key)
+  streamMap <<- key, "first value"
+  resFirstStream = streamMap.getStream(key)
+  streamMap <<- key, "second value"
+  resSecondStream = streamMap.getStream(key)
+  resFirst = resFirstStream[0]
+  resSecond = resSecondStream[1]
+  <- resEmptyStream, resFirst, resSecond
+
+func testKeysFunc() -> []string, []string, []string:
+  streamMap: %string
+  resEmpty = streamMap.keys()
+  streamMap <<- "key one", ""
+  resFirst = streamMap.keys()
+  streamMap <<- "key two", ""
+  streamMap <<- "key one", ""
+  streamMap <<- "key one", "text"
+  resSecond = streamMap.keys()
+  <- resEmpty, resFirst, resSecond
+
+func testKeysStreamFunc() -> []string, []string, []string:
+  streamMap: %string
+  resEmpty = streamMap.keysStream()
+  streamMap <<- "key one", ""
+  resFirst = streamMap.keysStream()
+  streamMap <<- "key one", "new"
+  streamMap <<- "key two", ""
+  resSecond = streamMap.keysStream()
+  <- resEmpty, resFirst, resSecond
+
+func testContainsFunc() -> bool, bool, bool, bool, bool:
+  keys = ["key one", "key two"]
+  streamMap: %string
+  resFirst = streamMap.contains(keys[0])
+  streamMap <<- keys[0], ""
+  resSecond = streamMap.contains(keys[0])
+  resThird = streamMap.contains(keys[1])
+  streamMap <<- keys[0], "new"
+  streamMap <<- keys[1], ""
+  resFourth = streamMap.contains(keys[0])
+  resFifth = streamMap.contains(keys[1])
+  <- resFirst, resSecond, resThird, resFourth, resFifth
+
+func testForFunc() -> []string, []string:
+  streamMap: %string
+  streamMap <<- "key one", "1"
+  streamMap <<- "key one", "2"
+
+  streamMap <<- "key two", "3"
+  streamMap <<- "key two", "4"
+  streamMap <<- "key two", "5"
+
+  streamMap <<- "key three", "6"
+
+  streamMap <<- "key four", "7"
+
+  streamKeys: *string
+  streamValues: *string
+
+  for kv <- streamMap:
+    streamKeys <<- kv.key
+    streamValues <<- kv.value
+  <- streamKeys, streamValues
+
+func testParSeqMap(relay1: string, relay2: string, relay3: string) -> string:
+  relays = [relay1, relay2, relay3]
+  map: %u64
+  map2: %u64
+  parseq r <- relays on r:
+    map <<- "time", Peer.timestamp_ms()
+
+  for r <- relays par:
+    on r:
+      join map.get("time")[relays.length - 1]
+      map2 <<- "time", Peer.timestamp_ms()
+
+  join map2.get("time")[relays.length - 1]
+  <- "ok"
+
+func testForTupleFunc() -> []string, []string, []string:
+  streamMap: %string
+  streamMap <<- "key one", "1"
+  streamMap <<- "key one", "2"
+
+  streamMap <<- "key two", "3"
+  streamMap <<- "key two", "4"
+  streamMap <<- "key two", "5"
+
+  streamMap <<- "key three", "6"
+
+  streamMap <<- "key four", "7"
+
+  streamFirst: *string
+  streamSecond: *string
+  streamThird: *string
+
+  for k, v <- streamMap:
+    streamFirst <<- k
+    streamSecond <<- v
+
+  for k, v <- streamMap:
+    streamFirst <<- v
+    streamSecond <<- k
+
+  for k, v <- streamMap:
+    streamThird <<- streamMap.get(k)!
+
+  <- streamFirst, streamSecond, streamThird

aqua-src/builtin.aqua

@@ -1,3 +1,5 @@
+aqua Builtin declares *
+
 -- Default public interface of Fluence nodes
 
 alias Field : []string

backend/air/src/main/scala/aqua/backend/air/AirGen.scala

@@ -3,7 +3,7 @@ package aqua.backend.air
 import aqua.model.*
 import aqua.raw.ops.Call
 import aqua.res.*
-import aqua.types.{ArrayType, CanonStreamType, StreamMapType, StreamType, Type}
+import aqua.types.{ArrayType, CanonStreamMapType, CanonStreamType, StreamMapType, StreamType, Type}
 import cats.Eval
 import cats.data.Chain
 import cats.free.Cofree
@@ -29,7 +29,8 @@ object AirGen extends Logging {
   def varNameToString(name: String, `type`: Type): String =
     (`type` match {
       case _: StreamType => "$" + name
-      case _: CanonStreamType => "#" + name
+      case _: CanonStreamType => "#$" + name
+      case _: CanonStreamMapType => "#%" + name
       case _: StreamMapType => "%" + name
       case _ => name
     }).replace('.', '_')

integration-tests/aqua/examples/parseq.aqua

@@ -4,27 +4,17 @@ export testParSeq
 
 import "@fluencelabs/aqua-lib/builtin.aqua"
 
-service NumOp("op"):
-  identity(n: u64) -> u64
-
-data PeerRelay:
-  peer: string
-  relay: string
-
-func testParSeq(peer1: string, peer2: string, peer3: string, relay1: string, relay2: string, relay3: string) -> string:
-  pr1 = PeerRelay(peer = peer1, relay = relay1)
-  pr2 = PeerRelay(peer = peer2, relay = relay2)
-  pr3 = PeerRelay(peer = peer3, relay = relay3)
-  peers = [pr1, pr2, pr3]
+func testParSeq(relay1: string, relay2: string, relay3: string) -> string:
+  relays = [relay1, relay2, relay3]
   stream: *u64
   stream2: *u64
-  parseq p <- peers on p.peer via p.relay:
+  parseq r <- relays on r:
     stream <- Peer.timestamp_ms()
 
-  for p <- peers par:
-    on p.peer via p.relay:
-      join stream[peers.length - 1]
+  for r <- relays par:
+    on r:
+      join stream[relays.length - 1]
       stream2 <<- Peer.timestamp_ms()
 
-  join stream2[peers.length - 1]
+  join stream2[relays.length - 1]
   <- "ok"

integration-tests/aqua/examples/streamMap.aqua

@@ -0,0 +1,130 @@
+aqua StreamMapTest declares *
+
+export testGetFunc, testGetStreamFunc, testKeysFunc, testKeysStreamFunc
+export testContainsFunc, testForFunc, testParSeqMap, testForTupleFunc
+
+import "@fluencelabs/aqua-lib/builtin.aqua"
+
+func testGetFunc() -> []string, []string, []string, u32:
+  streamMap: %string
+  key = "key"
+  resEmpty = streamMap.get(key)
+  streamMap <<- key, "first value"
+  resFirst = streamMap.get(key)
+  streamMap <<- key, "second value"
+  resSecond = streamMap.get(key)
+  <- resEmpty, resFirst, resSecond, resSecond.length
+
+func testGetStreamFunc() -> []string, string, string:
+  streamMap: %string
+  key = "key"
+  resEmptyStream = streamMap.getStream(key)
+  streamMap <<- key, "first value"
+  resFirstStream = streamMap.getStream(key)
+  streamMap <<- key, "second value"
+  resSecondStream = streamMap.getStream(key)
+  resFirst = resFirstStream[0]
+  resSecond = resSecondStream[1]
+  <- resEmptyStream, resFirst, resSecond
+
+func testKeysFunc() -> []string, []string, []string:
+  streamMap: %string
+  resEmpty = streamMap.keys()
+  streamMap <<- "key one", ""
+  resFirst = streamMap.keys()
+  streamMap <<- "key two", ""
+  streamMap <<- "key one", ""
+  streamMap <<- "key one", "text"
+  resSecond = streamMap.keys()
+  <- resEmpty, resFirst, resSecond
+
+func testKeysStreamFunc() -> []string, []string, []string:
+  streamMap: %string
+  resEmpty = streamMap.keysStream()
+  streamMap <<- "key one", ""
+  resFirst = streamMap.keysStream()
+  streamMap <<- "key one", "new"
+  streamMap <<- "key two", ""
+  resSecond = streamMap.keysStream()
+  join resFirst[0]
+  join resSecond[2]
+  <- resEmpty, resFirst, resSecond
+
+func testContainsFunc() -> bool, bool, bool, bool, bool:
+  keys = ["key one", "key two"]
+  streamMap: %string
+  resFirst = streamMap.contains(keys[0])
+  streamMap <<- keys[0], ""
+  resSecond = streamMap.contains(keys[0])
+  resThird = streamMap.contains(keys[1])
+  streamMap <<- keys[0], "new"
+  streamMap <<- keys[1], ""
+  resFourth = streamMap.contains(keys[0])
+  resFifth = streamMap.contains(keys[1])
+  <- resFirst, resSecond, resThird, resFourth, resFifth
+
+func testForFunc() -> []string, []string:
+  streamMap: %string
+  streamMap <<- "key one", "1"
+  streamMap <<- "key one", "2"
+
+  streamMap <<- "key two", "3"
+  streamMap <<- "key two", "4"
+  streamMap <<- "key two", "5"
+
+  streamMap <<- "key three", "6"
+
+  streamMap <<- "key four", "7"
+
+  streamKeys: *string
+  streamValues: *string
+
+  for kv <- streamMap:
+    streamKeys <<- kv.key
+    streamValues <<- kv.value
+  <- streamKeys, streamValues
+
+func testParSeqMap(relay1: string, relay2: string, relay3: string) -> string:
+  relays = [relay1, relay2, relay3]
+  map: %u64
+  map2: %u64
+  parseq r <- relays on r:
+    map <<- "time", Peer.timestamp_ms()
+
+  for r <- relays par:
+    on r:
+      join map.getStream("time")[relays.length - 1]
+      map2 <<- "time", Peer.timestamp_ms()
+
+  join map2.getStream("time")[relays.length - 1]
+  <- "ok"
+
+func testForTupleFunc() -> []string, []string, []string:
+  streamMap: %string
+  streamMap <<- "key one", "1"
+  streamMap <<- "key one", "2"
+
+  streamMap <<- "key two", "3"
+  streamMap <<- "key two", "4"
+  streamMap <<- "key two", "5"
+
+  streamMap <<- "key three", "6"
+
+  streamMap <<- "key four", "7"
+
+  streamFirst: *string
+  streamSecond: *string
+  streamThird: *string
+
+  for k, v <- streamMap:
+    streamFirst <<- k
+    streamSecond <<- v
+
+  for k, v <- streamMap:
+    streamFirst <<- v
+    streamSecond <<- k
+
+  for k, v <- streamMap:
+    streamThird <<- streamMap.get(k)!
+
+  <- streamFirst, streamSecond, streamThird

integration-tests/aqua/examples/streamMapAbilities.aqua

@@ -0,0 +1,41 @@
+aqua StreamMapAbilities
+
+export streamMapAbilityTest
+
+ability Streams:
+  stream: *string
+  map: %string
+
+ability Adds:
+  addToStream(s: string)
+  addToMap(k: string, v: string)
+
+func addToStreamClosure(str: *string) -> string -> ():
+  cl = func (s: string):
+    str <<- s
+  <- cl
+
+func addToMapClosure(str: %string) -> string, string -> ():
+  cl = func (k: string, v: string):
+    str <<- k, v
+  <- cl
+
+func addTo{Streams}() -> Adds:
+  addStream = addToStreamClosure(Streams.stream)
+  addMap = addToMapClosure(Streams.map)
+  adds = Adds(addToStream = addStream, addToMap = addMap)
+  <- adds
+
+func add{Adds}(s: string, k: string):
+  Adds.addToStream(s)
+  Adds.addToMap(k, k)
+
+func streamMapAbilityTest() -> []string, []string:
+  stream: *string
+  map: %string
+  ab = Streams(stream = stream, map = map)
+  adds <- addTo{ab}()
+  add{adds}("one", "1")
+  add{adds}("two", "2")
+  add{adds}("three", "3")
+  <- stream, map.keys()