feat(compiler): Make topology hop with non-FFI snippet [fixes LNG-125]

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

@@ -46,8 +46,6 @@ object DataView {
 
   case class Variable(name: String) extends DataView
 
-  case class Stream(name: String) extends DataView
-
   case class VarLens(name: String, lens: String, isField: Boolean = true) extends DataView {
     def append(sublens: String): VarLens = copy(lens = lens + sublens)
   }
@@ -57,7 +55,6 @@ object DataView {
     case InitPeerId ⇒ "%init_peer_id%"
     case LastError ⇒ "%last_error%"
     case Variable(name) ⇒ name
-    case Stream(name) ⇒ name
     case VarLens(name, lens, isField) ⇒
       if (isField) name + ".$" + lens
       else name + lens
@@ -90,7 +87,12 @@ object Air {
 
   case class Next(label: String) extends Air(Keyword.Next)
 
-  case class Fold(iterable: DataView, label: String, instruction: Air, lastNextInstruction: Option[Air]) extends Air(Keyword.Fold)
+  case class Fold(
+    iterable: DataView,
+    label: String,
+    instruction: Air,
+    lastNextInstruction: Option[Air]
+  ) extends Air(Keyword.Fold)
 
   case class Match(left: DataView, right: DataView, instruction: Air) extends Air(Keyword.Match)
 

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, StreamType}
+import aqua.types.{ArrayType, CanonStreamType, StreamType, Type}
 import cats.Eval
 import cats.data.Chain
 import cats.free.Cofree
@@ -26,14 +26,17 @@ object AirGen extends Logging {
       s".[$idx]${propertyToString(tail)}"
   }
 
+  def varNameToString(name: String, `type`: Type): String =
+    (`type` match {
+      case _: StreamType => "$" + name
+      case _: CanonStreamType => "#" + name
+      case _ => name
+    }).replace('.', '_')
+
   def valueToData(vm: ValueModel): DataView = vm match {
     case LiteralModel(value, _) => DataView.StringScalar(value)
     case VarModel(name, t, property) =>
-      val n = (t match {
-        case _: StreamType => "$" + name
-        case _: CanonStreamType => "#" + name
-        case _ => name
-      }).replace('.', '_')
+      val n = varNameToString(name, t)
       if (property.isEmpty) DataView.Variable(n)
       else {
         val functors = property.find {
@@ -97,8 +100,8 @@ object AirGen extends Logging {
           case ForModel.NeverMode => NeverGen
         }
         Eval later ForGen(valueToData(iterable), item, opsToSingle(ops), m)
-      case RestrictionRes(item, isStream) =>
-        Eval later NewGen(item, isStream, opsToSingle(ops))
+      case RestrictionRes(item, itemType) =>
+        Eval later NewGen(varNameToString(item, itemType), opsToSingle(ops))
       case CallServiceRes(serviceId, funcName, CallRes(args, exportTo), peerId) =>
         Eval.later(
           ServiceCallGen(
@@ -179,14 +182,17 @@ case class MatchMismatchGen(
     else Air.Mismatch(left, right, body.generate)
 }
 
-case class ForGen(iterable: DataView, item: String, body: AirGen, mode: Option[AirGen]) extends AirGen {
+case class ForGen(iterable: DataView, item: String, body: AirGen, mode: Option[AirGen])
+    extends AirGen {
   override def generate: Air = Air.Fold(iterable, item, body.generate, mode.map(_.generate))
 }
 
-case class NewGen(item: String, isStream: Boolean, body: AirGen) extends AirGen {
+case class NewGen(name: String, body: AirGen) extends AirGen {
 
-  override def generate: Air =
-    Air.New(if (isStream) DataView.Stream("$" + item) else DataView.Variable(item), body.generate)
+  override def generate: Air = Air.New(
+    DataView.Variable(name),
+    body.generate
+  )
 }
 
 case class NextGen(item: String) extends AirGen {

build.sbt

@@ -233,7 +233,7 @@ lazy val transform = crossProject(JVMPlatform, JSPlatform)
   .crossType(CrossType.Pure)
   .in(file("model/transform"))
   .settings(commons: _*)
-  .dependsOn(model, res, inline)
+  .dependsOn(model, res, inline, res % "test->test")
 
 lazy val semantics = crossProject(JVMPlatform, JSPlatform)
   .withoutSuffixFor(JVMPlatform)
@@ -252,7 +252,7 @@ lazy val compiler = crossProject(JVMPlatform, JSPlatform)
   .crossType(CrossType.Pure)
   .in(file("compiler"))
   .settings(commons: _*)
-  .dependsOn(semantics, linker, backend, transform % Test)
+  .dependsOn(semantics, linker, backend, transform % Test, res % "test->test")
 
 lazy val backend = crossProject(JVMPlatform, JSPlatform)
   .withoutSuffixFor(JVMPlatform)

compiler/src/test/scala/aqua/compiler/AquaCompilerSpec.scala

@@ -1,6 +1,14 @@
 package aqua.compiler
 
-import aqua.model.{CallModel, ForModel, FunctorModel, IntoIndexModel, LiteralModel, ValueModel, VarModel}
+import aqua.model.{
+  CallModel,
+  ForModel,
+  FunctorModel,
+  IntoIndexModel,
+  LiteralModel,
+  ValueModel,
+  VarModel
+}
 import aqua.model.transform.TransformConfig
 import aqua.model.transform.Transform
 import aqua.parser.ParserError
@@ -10,7 +18,21 @@ import aqua.parser.lift.Span
 import aqua.parser.lift.Span.S
 import aqua.raw.ConstantRaw
 import aqua.raw.value.{LiteralRaw, ValueRaw, VarRaw}
-import aqua.res.{ApRes, CallRes, CallServiceRes, CanonRes, FoldRes, MakeRes, MatchMismatchRes, NextRes, ParRes, RestrictionRes, SeqRes, XorRes}
+import aqua.res.{
+  ApRes,
+  CallRes,
+  CallServiceRes,
+  CanonRes,
+  FoldRes,
+  MakeRes,
+  MatchMismatchRes,
+  NextRes,
+  ParRes,
+  RestrictionRes,
+  SeqRes,
+  XorRes
+}
+import aqua.res.ResBuilder
 import aqua.types.{ArrayType, CanonStreamType, LiteralType, ScalarType, StreamType, Type}
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
@@ -83,8 +105,8 @@ class AquaCompilerSpec extends AnyFlatSpec with Matchers {
 
   }
 
-  def through(peer: ValueModel, log: String = null) =
-    MakeRes.noop(peer, log)
+  def through(peer: ValueModel) =
+    MakeRes.hop(peer)
 
   val relay = VarRaw("-relay-", ScalarType.string)
 
@@ -97,42 +119,10 @@ class AquaCompilerSpec extends AnyFlatSpec with Matchers {
     ).leaf
   }
 
-  val init = LiteralModel.fromRaw(ValueRaw.InitPeerId)
-
-  private def join(vm: VarModel, length: ValueModel) = {
-    val testVM = VarModel(vm.name + "_test", vm.`type`)
-    val iter = VarModel(vm.name + "_fold_var", ScalarType.string)
-    val canon = VarModel(vm.name + "_iter_canon", CanonStreamType(ScalarType.string))
-    val canonRes = VarModel(vm.name + "_result_canon", CanonStreamType(ScalarType.string))
-    val arrayRes = VarModel(vm.name + "_gate", ArrayType(ScalarType.string))
-    val idx = VarModel(vm.name + "_incr", ScalarType.u32)
-
-    RestrictionRes(testVM.name, true).wrap(
-      CallServiceRes(
-        LiteralModel("\"math\"", ScalarType.string),
-        "add",
-        CallRes(
-          length :: LiteralModel.fromRaw(LiteralRaw.number(1)) :: Nil,
-          Some(CallModel.Export(idx.name, idx.`type`))
-        ),
-        init
-      ).leaf,
-      FoldRes(iter.name, vm, Some(ForModel.NeverMode)).wrap(
-        ApRes(iter, CallModel.Export(testVM.name, testVM.`type`)).leaf,
-        CanonRes(testVM, init, CallModel.Export(canon.name, canon.`type`)).leaf,
-        XorRes.wrap(
-          MatchMismatchRes(
-            canon.copy(properties = Chain.one(FunctorModel("length", ScalarType.u32))),
-            idx,
-            true
-          ).leaf,
-          NextRes(iter.name).leaf
-        )
-      ),
-      CanonRes(testVM, init, CallModel.Export(canonRes.name, canonRes.`type`)).leaf,
-      ApRes(canonRes, CallModel.Export(arrayRes.name, arrayRes.`type`)).leaf
-    )
-  }
+  private val init = LiteralModel.fromRaw(ValueRaw.InitPeerId)
+
+  private def join(vm: VarModel, idx: ValueModel) =
+    ResBuilder.join(vm, idx, init)
 
   "aqua compiler" should "create right topology" in {
 
@@ -177,7 +167,7 @@ class AquaCompilerSpec extends AnyFlatSpec with Matchers {
       SeqRes.wrap(
         getDataSrv("-relay-", ScalarType.string),
         getDataSrv(peers.name, peers.`type`),
-        RestrictionRes("results", true).wrap(
+        RestrictionRes(results.name, resultsType).wrap(
           SeqRes.wrap(
             ParRes.wrap(
               FoldRes(peer.name, peers, Some(ForModel.NeverMode)).wrap(
@@ -203,10 +193,10 @@ class AquaCompilerSpec extends AnyFlatSpec with Matchers {
             ),
             join(results, LiteralModel.fromRaw(LiteralRaw.number(2))),
             CanonRes(results, init, CallModel.Export(canonResult.name, canonResult.`type`)).leaf,
-          ApRes(
-            canonResult,
-            CallModel.Export(flatResult.name, flatResult.`type`)
-          ).leaf
+            ApRes(
+              canonResult,
+              CallModel.Export(flatResult.name, flatResult.`type`)
+            ).leaf
           )
         ),
         CallServiceRes(
@@ -283,23 +273,24 @@ class AquaCompilerSpec extends AnyFlatSpec with Matchers {
     val Some(funcWrap) = aquaRes.funcs.find(_.funcName == "wrap")
     val Some(barfoo) = aquaRes.funcs.find(_.funcName == "barfoo")
 
-    val resVM = VarModel("res", StreamType(ScalarType.string))
+    val resStreamType = StreamType(ScalarType.string)
+    val resVM = VarModel("res", resStreamType)
     val resCanonVM = VarModel("-res-fix-0", CanonStreamType(ScalarType.string))
     val resFlatVM = VarModel("-res-flat-0", ArrayType(ScalarType.string))
 
     barfoo.body.equalsOrShowDiff(
       SeqRes.wrap(
-        RestrictionRes("res", true).wrap(
+        RestrictionRes(resVM.name, resStreamType).wrap(
           SeqRes.wrap(
             // res <- foo()
             ApRes(
               LiteralModel.fromRaw(LiteralRaw.quote("I am MyFooBar foo")),
-              CallModel.Export("res", StreamType(ScalarType.string))
+              CallModel.Export(resVM.name, resVM.`type`)
             ).leaf,
             // res <- bar()
             ApRes(
               LiteralModel.fromRaw(LiteralRaw.quote(" I am MyFooBar bar")),
-              CallModel.Export("res", StreamType(ScalarType.string))
+              CallModel.Export(resVM.name, resVM.`type`)
             ).leaf,
             // canonicalization
             CanonRes(

integration-tests/aqua/examples/foldJoin.aqua

@@ -15,7 +15,5 @@ func getTwoResults(node: string) -> []u64:
       on n:
         try:
           res <- Peer.timestamp_sec()
-    Op2.identity(res!)
-    Op2.identity(res!1)
-    Op2.identity(res!2)
+    join res!2
   <- res

model/inline/src/main/scala/aqua/model/inline/TagInliner.scala

@@ -351,8 +351,8 @@ object TagInliner extends Logging {
           }
         } yield model.fold(TagInlined.Empty())(m => TagInlined.Single(model = m))
 
-      case RestrictionTag(name, isStream) =>
-        pure(RestrictionModel(name, isStream))
+      case RestrictionTag(name, typ) =>
+        pure(RestrictionModel(name, typ))
 
       case DeclareStreamTag(value) =>
         value match