optional next node - changes in compiler

Author: mgoworko

engine/flink/executor/src/main/scala/pl/touk/nussknacker/engine/process/registrar/FlinkProcessRegistrar.scala

@@ -14,7 +14,6 @@ import pl.touk.nussknacker.engine.api.component.NodeComponentInfo
 import pl.touk.nussknacker.engine.api.context.{JoinContextTransformation, ValidationContext}
 import pl.touk.nussknacker.engine.api.process.ProcessName
 import pl.touk.nussknacker.engine.canonicalgraph.CanonicalProcess
-import pl.touk.nussknacker.engine.compile.ArtificialDeadEndSink
 import pl.touk.nussknacker.engine.compiledgraph.part._
 import pl.touk.nussknacker.engine.deployment.DeploymentData
 import pl.touk.nussknacker.engine.flink.FlinkScenarioCompilationDependencies
@@ -258,8 +257,6 @@ class FlinkProcessRegistrar(
       processPart match {
         case part @ SinkPart(sink: FlinkSink, _, contextBefore, _) =>
           registerSinkPark(start, part, sink, contextBefore)
-        case part @ SinkPart(ArtificialDeadEndSink, _, contextBefore, _) =>
-          registerSinkPark(start, part, EmptySink, contextBefore)
         case part: SinkPart =>
           // TODO: fixme "part.obj" is not stringified well
           //      (eg. Scenario can only use flink sinks, instead given: pl.touk.nussknacker.engine.management.sample.sink.LiteDeadEndSink$@21220fd7)

scenario-api/src/main/scala/pl/touk/nussknacker/engine/build/GraphBuilder.scala

@@ -20,34 +20,36 @@ trait GraphBuilder[R] {
   def build(inner: GraphBuilder.Creator[R]): GraphBuilder[R]
 
   def source(id: String, typ: String, params: (String, Expression)*): GraphBuilder[SourceNode] =
-    new SimpleGraphBuilder(
-      SourceNode(Source(id, SourceRef(typ, toNodeParameters(params))), _)
-    )
+    new SimpleGraphBuilder(node => SourceNode(Source(id, SourceRef(typ, toNodeParameters(params))), Some(node)))
 
   def buildVariable(id: String, varName: String, fields: (String, Expression)*): GraphBuilder[R] =
     build(node =>
-      creator(OneOutputSubsequentNode(VariableBuilder(id, varName, fields.map(f => Field(f._1, f._2)).toList), node))
+      creator(
+        OneOutputSubsequentNode(VariableBuilder(id, varName, fields.map(f => Field(f._1, f._2)).toList), Some(node))
+      )
     )
 
   def buildSimpleVariable(id: String, varName: String, value: Expression): GraphBuilder[R] =
-    build(node => creator(OneOutputSubsequentNode(Variable(id, varName, value), node)))
+    build(node => creator(OneOutputSubsequentNode(Variable(id, varName, value), Some(node))))
 
   def enricher(id: String, output: String, svcId: String, params: (String, Expression)*): GraphBuilder[R] =
     build(node =>
       creator(
-        OneOutputSubsequentNode(Enricher(id, ServiceRef(svcId, toNodeParameters(params)), output), node)
+        OneOutputSubsequentNode(Enricher(id, ServiceRef(svcId, toNodeParameters(params)), output), Some(node))
       )
     )
 
   def processor(id: String, svcId: String, params: (String, Expression)*): GraphBuilder[R] =
-    build(node => creator(OneOutputSubsequentNode(Processor(id, ServiceRef(svcId, toNodeParameters(params))), node)))
+    build(node =>
+      creator(OneOutputSubsequentNode(Processor(id, ServiceRef(svcId, toNodeParameters(params))), Some(node)))
+    )
 
   def disabledProcessor(id: String, svcId: String, params: (String, Expression)*): GraphBuilder[R] =
     build(node =>
       creator(
         OneOutputSubsequentNode(
           Processor(id, ServiceRef(svcId, toNodeParameters(params)), isDisabled = Some(true)),
-          node
+          Some(node)
         )
       )
     )
@@ -98,24 +100,24 @@ trait GraphBuilder[R] {
     )
 
   def fragmentInput(id: String, params: (String, Class[_])*): GraphBuilder[SourceNode] =
-    new SimpleGraphBuilder(
+    new SimpleGraphBuilder(node =>
       SourceNode(
         FragmentInputDefinition(
           id = id,
           parameters = params.map(kv => FragmentParameter(ParameterName(kv._1), FragmentClazzRef(kv._2.getName))).toList
         ),
-        _
+        Some(node)
       )
     )
 
   def fragmentInputWithRawParameters(id: String, params: FragmentParameter*): GraphBuilder[SourceNode] =
-    new SimpleGraphBuilder(
+    new SimpleGraphBuilder(node =>
       SourceNode(
         FragmentInputDefinition(
           id = id,
           parameters = params.toList
         ),
-        _
+        Some(node)
       )
     )
 
@@ -172,7 +174,7 @@ trait GraphBuilder[R] {
       creator(
         OneOutputSubsequentNode(
           CustomNode(id, Some(outputVar), customNodeRef, toNodeParameters(params)),
-          node
+          Some(node)
         )
       )
     )
@@ -184,7 +186,7 @@ trait GraphBuilder[R] {
   def customNodeNoOutput(id: String, customNodeRef: String, params: (String, Expression)*): GraphBuilder[R] =
     build(node =>
       creator(
-        OneOutputSubsequentNode(CustomNode(id, None, customNodeRef, toNodeParameters(params)), node)
+        OneOutputSubsequentNode(CustomNode(id, None, customNodeRef, toNodeParameters(params)), Some(node))
       )
     )
 
@@ -203,7 +205,9 @@ trait GraphBuilder[R] {
     val branchParameters = branchParams.map { case (branchId, bParams) =>
       BranchParameters(branchId, toNodeParameters(bParams))
     }
-    new SimpleGraphBuilder(SourceNode(node.Join(id, output, typ, toNodeParameters(params), branchParameters), _))
+    new SimpleGraphBuilder(n =>
+      SourceNode(node.Join(id, output, typ, toNodeParameters(params), branchParameters), Some(n))
+    )
   }
 
   def decisionTable(

scenario-api/src/main/scala/pl/touk/nussknacker/engine/canonize/MaybeArtificial.scala

@@ -45,9 +45,9 @@ private[engine] object MaybeArtificial {
   // we need to make sure it's unique to prevent weird errors
   private def generateArtificialName() = s"$DummyObjectNamePrefix-${UUID.randomUUID()}"
 
-  def missingSinkError(errors: ProcessUncanonizationError*): MaybeArtificial[node.SubsequentNode] =
+  def missingSinkError(errors: ProcessUncanonizationError*): MaybeArtificial[Option[node.SubsequentNode]] =
     new MaybeArtificial(
-      node.EndingNode(node.Sink(generateArtificialName(), SinkRef(artificalSourceSinkRef, Nil), None)),
+      Some(node.EndingNode(node.Sink(generateArtificialName(), SinkRef(artificalSourceSinkRef, Nil), None))),
       errors.toList
     )
 
@@ -56,13 +56,4 @@ private[engine] object MaybeArtificial {
       node.SourceNode(node.Source(generateArtificialName(), SourceRef(artificalSourceSinkRef, Nil)), _)
     )
 
-  def addedArtificialDeadEndSink(previousNodeId: String): MaybeArtificial[node.SubsequentNode] =
-    new MaybeArtificial(
-      node.EndingNode(new ArtificialDeadEndSink(previousNodeId)),
-      List.empty,
-    )
-
-  class ArtificialDeadEndSink(val previousNodeId: String)
-      extends node.Sink(s"artificialDeadEndSink-after-$previousNodeId", SinkRef(artificalSourceSinkRef, Nil), None)
-
 }

scenario-api/src/main/scala/pl/touk/nussknacker/engine/canonize/MissingSinkHandler.scala

@@ -0,0 +1,32 @@
+package pl.touk.nussknacker.engine.canonize
+
+import pl.touk.nussknacker.engine.graph.node
+
+// This trait and its 2 implementation encapsulate the logic of handling scenarios that do not end with sink
+//  - older behavior is that error is returned for missing sinks
+//  - new behavior, which we intend to become the default and only one in the future, is allowing missing sinks
+sealed trait MissingSinkHandler {
+  def handleMissingSink(previousNodeId: String): MaybeArtificial[Option[node.SubsequentNode]]
+}
+
+object MissingSinkHandler {
+
+  object AllowMissingSinkHandler extends MissingSinkHandler {
+
+    override def handleMissingSink(
+        previousNodeId: String
+    ): MaybeArtificial[Option[node.SubsequentNode]] =
+      new MaybeArtificial(None, Nil)
+
+  }
+
+  object DoNotAllowMissingSinkHandler extends MissingSinkHandler {
+
+    override def handleMissingSink(
+        previousNodeId: String
+    ): MaybeArtificial[Option[node.SubsequentNode]] =
+      new MaybeArtificial(None, InvalidTailOfBranch(previousNodeId) :: Nil)
+
+  }
+
+}

scenario-api/src/main/scala/pl/touk/nussknacker/engine/canonize/ProcessCanonizer.scala

@@ -24,29 +24,27 @@ object ProcessCanonizer {
 
   def uncanonize(
       canonicalProcess: CanonicalProcess,
-      allowEndingScenarioWithoutSink: Boolean,
+      missingSinkHandler: MissingSinkHandler,
   ): ValidatedNel[ProcessUncanonizationError, EspProcess] =
-    uncanonizeArtificial(canonicalProcess, allowEndingScenarioWithoutSink).toValidNel
+    uncanonizeArtificial(canonicalProcess, missingSinkHandler).toValidNel
 
   def uncanonizeArtificial(
       canonicalProcess: CanonicalProcess,
-      allowEndingScenarioWithoutSink: Boolean,
+      missingSinkHandler: MissingSinkHandler,
   ): MaybeArtificial[EspProcess] = {
 
     val branches: MaybeArtificial[NonEmptyList[pl.touk.nussknacker.engine.graph.node.SourceNode]] =
-      canonicalProcess.allStartNodes.map(uncanonizeSource(_, allowEndingScenarioWithoutSink)).sequence
+      canonicalProcess.allStartNodes.map(uncanonizeSource(_)(missingSinkHandler)).sequence
 
     branches.map(bList => EspProcess(canonicalProcess.metaData, bList))
   }
 
   private def uncanonizeSource(
       canonicalNode: List[canonicalnode.CanonicalNode],
-      allowEndingScenarioWithoutSink: Boolean,
-  ): MaybeArtificial[node.SourceNode] =
+  )(implicit missingSinkHandler: MissingSinkHandler): MaybeArtificial[node.SourceNode] =
     canonicalNode match {
       case (a @ canonicalnode.FlatNode(data: node.StartingNodeData)) :: tail =>
-        uncanonize(a, tail, allowEndingScenarioWithoutSink).map(node.SourceNode(data, _))
-
+        uncanonize(a, tail).map(node.SourceNode(data, _))
       case other :: _ =>
         MaybeArtificial.artificialSource(InvalidRootNode(other.id))
 
@@ -57,98 +55,79 @@ object ProcessCanonizer {
   private def uncanonize(
       previous: canonicalnode.CanonicalNode,
       canonicalNode: List[canonicalnode.CanonicalNode],
-      allowEndingScenarioWithoutSink: Boolean,
-  ): MaybeArtificial[node.SubsequentNode] =
+  )(implicit missingSinkHandler: MissingSinkHandler): MaybeArtificial[Option[node.SubsequentNode]] =
     canonicalNode match {
       case canonicalnode.FlatNode(data: node.BranchEndData) :: Nil =>
-        new MaybeArtificial(node.BranchEnd(data), Nil)
+        new MaybeArtificial(Some(node.BranchEnd(data)), Nil)
 
       case canonicalnode.FlatNode(data: node.EndingNodeData) :: Nil =>
-        new MaybeArtificial(node.EndingNode(data), Nil)
+        new MaybeArtificial(Some(node.EndingNode(data)), Nil)
 
       case (a @ canonicalnode.FlatNode(data: node.OneOutputSubsequentNodeData)) :: tail =>
-        uncanonize(a, tail, allowEndingScenarioWithoutSink).map(node.OneOutputSubsequentNode(data, _))
-
+        uncanonize(a, tail).map(node.OneOutputSubsequentNode(data, _)).map(Some(_): Option[node.SubsequentNode])
       case (a @ canonicalnode.FilterNode(data, nextFalse)) :: tail if nextFalse.isEmpty =>
-        uncanonize(a, tail, allowEndingScenarioWithoutSink).map(nextTrue => node.FilterNode(data, Some(nextTrue), None))
+        uncanonize(a, tail).map(nextTrue => Some(node.FilterNode(data, nextTrue, None)))
 
       case (a @ canonicalnode.FilterNode(data, nextFalse)) :: tail if tail.isEmpty =>
-        uncanonize(a, nextFalse, allowEndingScenarioWithoutSink).map { nextFalseV =>
-          node.FilterNode(data, None, Some(nextFalseV))
-        }
+        uncanonize(a, nextFalse).map { nextFalseV => Some(node.FilterNode(data, None, nextFalseV)) }
 
       case (a @ canonicalnode.FilterNode(data, nextFalse)) :: tail =>
-        (uncanonize(a, tail, allowEndingScenarioWithoutSink), uncanonize(a, nextFalse, allowEndingScenarioWithoutSink))
-          .mapN { (nextTrue, nextFalseV) =>
-            node.FilterNode(data, Some(nextTrue), Some(nextFalseV))
-          }
+        (uncanonize(a, tail), uncanonize(a, nextFalse)).mapN { (nextTrue, nextFalseV) =>
+          Some(node.FilterNode(data, nextTrue, nextFalseV))
+        }
 
       case (a @ canonicalnode.SwitchNode(data, Nil, defaultNext)) :: Nil =>
-        uncanonize(a, defaultNext, allowEndingScenarioWithoutSink).map { defaultNextV =>
-          node.SwitchNode(data, Nil, Some(defaultNextV))
-        }
+        uncanonize(a, defaultNext).map(defaultNextV => Some(node.SwitchNode(data, Nil, defaultNextV)))
 
       case (a @ canonicalnode.SwitchNode(data, nexts, defaultNext)) :: Nil if defaultNext.isEmpty =>
         nexts
-          .map { casee =>
-            uncanonize(a, casee.nodes, allowEndingScenarioWithoutSink).map(node.Case(casee.expression, _))
-          }
+          .map(casee => uncanonize(a, casee.nodes).map(node.Case(casee.expression, _)))
           .sequence[MaybeArtificial, node.Case]
           .map(node.SwitchNode(data, _, None))
+          .map(Some(_))
 
       case (a @ canonicalnode.SwitchNode(data, nexts, defaultNext)) :: Nil =>
         val unFlattenNexts = nexts
-          .map { casee =>
-            uncanonize(a, casee.nodes, allowEndingScenarioWithoutSink).map(node.Case(casee.expression, _))
-          }
+          .map(casee => uncanonize(a, casee.nodes).map(node.Case(casee.expression, _)))
           .sequence[MaybeArtificial, node.Case]
 
-        (unFlattenNexts, uncanonize(a, defaultNext, allowEndingScenarioWithoutSink)).mapN { (nextsV, defaultNextV) =>
-          node.SwitchNode(data, nextsV, Some(defaultNextV))
+        (unFlattenNexts, uncanonize(a, defaultNext)).mapN { (nextsV, defaultNextV) =>
+          Some(node.SwitchNode(data, nextsV, defaultNextV))
         }
 
       case canonicalnode.SplitNode(bare, Nil) :: Nil =>
-        handleMissingSink(bare.id, allowEndingScenarioWithoutSink)
+        missingSinkHandler.handleMissingSink(bare.id)
 
       case (a @ canonicalnode.SplitNode(bare, nexts)) :: Nil =>
-        nexts.map(uncanonize(a, _, allowEndingScenarioWithoutSink)).sequence[MaybeArtificial, node.SubsequentNode].map {
-          uncanonized =>
-            node.SplitNode(bare, uncanonized)
-        }
+        nexts
+          .map(uncanonize(a, _))
+          .sequence[MaybeArtificial, Option[node.SubsequentNode]]
+          .map { uncanonized =>
+            Some(node.SplitNode(bare, uncanonized.flatten))
+          }
 
       case invalidHead :: _ =>
         MaybeArtificial.missingSinkError(InvalidTailOfBranch(invalidHead.id))
 
       case Nil =>
-        handleMissingSink(previous.id, allowEndingScenarioWithoutSink)
+        missingSinkHandler.handleMissingSink(previous.id)
     }
 
-  private def handleMissingSink(
-      previousNodeId: String,
-      allowEndingScenarioWithoutSink: Boolean
-  ): MaybeArtificial[node.SubsequentNode] = {
-    if (allowEndingScenarioWithoutSink) {
-      MaybeArtificial.addedArtificialDeadEndSink(previousNodeId)
-    } else {
-      MaybeArtificial.missingSinkError(InvalidTailOfBranch(previousNodeId))
-    }
-  }
-
 }
 
 object NodeCanonizer {
 
   def canonize(n: node.Node): List[canonicalnode.CanonicalNode] =
     n match {
       case oneOut: node.OneOutputNode =>
-        canonicalnode.FlatNode(oneOut.data) :: canonize(oneOut.next)
+        canonicalnode.FlatNode(oneOut.data) :: oneOut.next.map(canonize).getOrElse(Nil)
       case node.FilterNode(data, nextTrue, nextFalse) =>
         canonicalnode.FilterNode(data, nextFalse.toList.flatMap(canonize)) :: nextTrue.toList.flatMap(canonize)
       case node.SwitchNode(data, nexts, defaultNext) =>
         canonicalnode.SwitchNode(
           data = data,
           nexts = nexts.map { next =>
-            canonicalnode.Case(next.expression, canonize(next.node))
+            canonicalnode.Case(next.expression, next.node.map(canonize).getOrElse(Nil))
           },
           defaultNext = defaultNext.toList.flatMap(canonize)
         ) :: Nil

scenario-api/src/main/scala/pl/touk/nussknacker/engine/graph/node.scala

@@ -38,14 +38,14 @@ object node {
   }
 
   sealed trait OneOutputNode extends NodeWithData {
-    def next: SubsequentNode
+    def next: Option[SubsequentNode]
   }
 
-  case class SourceNode(data: StartingNodeData, next: SubsequentNode) extends OneOutputNode
+  case class SourceNode(data: StartingNodeData, next: Option[SubsequentNode]) extends OneOutputNode
 
   sealed trait SubsequentNode extends Node
 
-  case class OneOutputSubsequentNode(data: OneOutputSubsequentNodeData, next: SubsequentNode)
+  case class OneOutputSubsequentNode(data: OneOutputSubsequentNodeData, next: Option[SubsequentNode])
       extends OneOutputNode
       with SubsequentNode
 
@@ -61,7 +61,11 @@ object node {
 
   case class SplitNode(data: Split, nextParts: List[SubsequentNode]) extends SubsequentNode
 
-  case class Case(expression: Expression, node: SubsequentNode)
+  case class Case(expression: Expression, node: Option[SubsequentNode])
+
+  object Case {
+    def apply(expression: Expression, node: SubsequentNode): Case = Case(expression, Some(node))
+  }
 
   case class EndingNode(data: EndingNodeData) extends SubsequentNode
 

scenario-api/src/test/scala/pl/touk/nussknacker/engine/marshall/ProcessMarshallerSpec.scala

@@ -16,6 +16,7 @@ import pl.touk.nussknacker.engine.build.{GraphBuilder, ScenarioBuilder}
 import pl.touk.nussknacker.engine.canonicalgraph.{canonicalnode, CanonicalProcess}
 import pl.touk.nussknacker.engine.canonicalgraph.canonicalnode.{CanonicalNode, FlatNode}
 import pl.touk.nussknacker.engine.canonize.{MaybeArtificial, ProcessCanonizer}
+import pl.touk.nussknacker.engine.canonize.MissingSinkHandler.DoNotAllowMissingSinkHandler
 import pl.touk.nussknacker.engine.graph.expression.Expression
 import pl.touk.nussknacker.engine.graph.expression.Expression.Language
 import pl.touk.nussknacker.engine.graph.node
@@ -48,7 +49,7 @@ class ProcessMarshallerSpec
         .customNode("b", "alamakota == 'true'", "someRef")
         .buildVariable("c", "fooVar", "f1" -> "expr1".spel, "f2" -> "expr2".spel)
         .enricher("d", "barVar", "dService", "p1" -> "expr3".spel)
-        .switch("f", "expr4".spel, "eVar", nestedGraph("e"), Case("e1".spel, GraphBuilder.emptySink("endE1", "")))
+        .switch("f", "expr4".spel, "eVar", nestedGraph("e"), Case("e1".spel, Some(GraphBuilder.emptySink("endE1", ""))))
 
     val result = marshallAndUnmarshall(process)
 
@@ -227,7 +228,7 @@ class ProcessMarshallerSpec
       inside(
         ProcessCanonizer.uncanonize(
           CanonicalProcess(MetaData("1", StreamMetaData()), nodes.toList, List.empty),
-          allowEndingScenarioWithoutSink = false
+          DoNotAllowMissingSinkHandler,
         )
       ) { case Invalid(NonEmptyList(canonize.InvalidTailOfBranch(id), Nil)) =>
         id shouldBe expectedBadNodeId