[NU-1800] Add template lazy param

build.sbt

@@ -1816,7 +1816,8 @@ lazy val flinkBaseComponentsTests = (project in flink("components/base-tests"))
   )
   .dependsOn(
     flinkComponentsTestkit  % Test,
-    flinkTableApiComponents % Test
+    flinkTableApiComponents % Test,
+    scenarioCompiler        % "test->test"
   )
 
 lazy val flinkKafkaComponents = (project in flink("components/kafka"))

components-api/src/main/scala/pl/touk/nussknacker/engine/api/LazyParameter.scala

@@ -1,5 +1,6 @@
 package pl.touk.nussknacker.engine.api
 
+import pl.touk.nussknacker.engine.api.LazyParameter.TemplateLazyParameter.TemplateExpression
 import pl.touk.nussknacker.engine.api.LazyParameter.{Evaluate, MappedLazyParameter, ProductLazyParameter}
 import pl.touk.nussknacker.engine.api.typed.typing.{Typed, TypingResult}
 
@@ -68,6 +69,25 @@ object LazyParameter {
 
   trait CustomLazyParameter[+T <: AnyRef] extends LazyParameter[T]
 
+  trait TemplateLazyParameter[T <: AnyRef] extends LazyParameter[T] {
+    def templateExpression: TemplateExpression
+  }
+
+  object TemplateLazyParameter {
+    case class TemplateExpression(parts: List[TemplateExpressionPart])
+    sealed trait TemplateExpressionPart
+
+    object TemplateExpressionPart {
+      case class NonTemplatedPart(value: String) extends TemplateExpressionPart
+
+      trait TemplatedPart extends TemplateExpressionPart {
+        val evaluate: Evaluate[String]
+      }
+
+    }
+
+  }
+
   final class ProductLazyParameter[T <: AnyRef, Y <: AnyRef](
       val arg1: LazyParameter[T],
       val arg2: LazyParameter[Y]

engine/flink/components/base-tests/src/test/scala/pl/touk/nussknacker/engine/flink/TemplateLazyParameterTest.scala

@@ -0,0 +1,55 @@
+package pl.touk.nussknacker.engine.flink
+
+import com.typesafe.config.ConfigFactory
+import org.apache.flink.api.connector.source.Boundedness
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.{Inside, LoneElement}
+import pl.touk.nussknacker.engine.api.component.ComponentDefinition
+import pl.touk.nussknacker.engine.build.ScenarioBuilder
+import pl.touk.nussknacker.engine.flink.test.FlinkSpec
+import pl.touk.nussknacker.engine.flink.util.test.FlinkTestScenarioRunner._
+import pl.touk.nussknacker.engine.graph.expression.Expression
+import pl.touk.nussknacker.engine.process.FlinkJobConfig.ExecutionMode
+import pl.touk.nussknacker.engine.spel.SpelExtension._
+import pl.touk.nussknacker.engine.testcomponents.SpelTemplateAstOperationService
+import pl.touk.nussknacker.engine.util.test.TestScenarioRunner
+import pl.touk.nussknacker.test.ValidatedValuesDetailedMessage
+
+class TemplateLazyParameterTest
+    extends AnyFunSuite
+    with FlinkSpec
+    with Matchers
+    with Inside
+    with ValidatedValuesDetailedMessage
+    with LoneElement {
+
+  private lazy val runner = TestScenarioRunner
+    .flinkBased(ConfigFactory.empty(), flinkMiniCluster)
+    .withExecutionMode(ExecutionMode.Batch)
+    .withExtraComponents(
+      List(ComponentDefinition("templateAstOperationService", SpelTemplateAstOperationService))
+    )
+    .build()
+
+  test("should use spel template ast operation parameter") {
+    val scenario = ScenarioBuilder
+      .streaming("test")
+      .source("source", TestScenarioRunner.testDataSource)
+      .enricher(
+        "customService",
+        "output",
+        "templateAstOperationService",
+        "template" -> Expression.spelTemplate(s"Hello#{#input}")
+      )
+      .emptySink("end", TestScenarioRunner.testResultSink, "value" -> "#output".spel)
+
+    val result = runner.runWithData(scenario, List(1, 2, 3), Boundedness.BOUNDED)
+    result.validValue.successes shouldBe List(
+      "[Hello]-literal[1]-templated",
+      "[Hello]-literal[2]-templated",
+      "[Hello]-literal[3]-templated"
+    )
+  }
+
+}

scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/compile/nodecompilation/EvaluableLazyParameter.scala

@@ -1,10 +1,16 @@
 package pl.touk.nussknacker.engine.compile.nodecompilation
 
-import pl.touk.nussknacker.engine.api.LazyParameter.{CustomLazyParameter, Evaluate}
-import pl.touk.nussknacker.engine.api.typed.typing.TypingResult
-import pl.touk.nussknacker.engine.api.{Context, JobData, MetaData, NodeId}
-import pl.touk.nussknacker.engine.compiledgraph.{BaseCompiledParameter, CompiledParameter}
+import pl.touk.nussknacker.engine.api.LazyParameter.TemplateLazyParameter.{TemplateExpression, TemplateExpressionPart}
+import pl.touk.nussknacker.engine.api.LazyParameter.{CustomLazyParameter, Evaluate, TemplateLazyParameter}
+import pl.touk.nussknacker.engine.api.definition.{Parameter => ParameterDef}
+import pl.touk.nussknacker.engine.api.typed.typing.{Typed, TypingResult}
+import pl.touk.nussknacker.engine.api.{Context, JobData, LazyParameter, NodeId}
+import pl.touk.nussknacker.engine.compiledgraph.BaseCompiledParameter
+import pl.touk.nussknacker.engine.definition.component.parameter.defaults.EditorBasedLanguageDeterminer
 import pl.touk.nussknacker.engine.expression.ExpressionEvaluator
+import pl.touk.nussknacker.engine.graph.expression.Expression.Language
+import pl.touk.nussknacker.engine.spel.SpelExpression
+import pl.touk.nussknacker.engine.spel.SpelTemplateSubexpression.{NonTemplatedValue, TemplatedExpression}
 
 class EvaluableLazyParameter[T <: AnyRef](
     compiledParameter: BaseCompiledParameter,
@@ -14,19 +20,88 @@ class EvaluableLazyParameter[T <: AnyRef](
     override val returnType: TypingResult
 ) extends CustomLazyParameter[T] {
 
-  def this(
-      compiledParameter: CompiledParameter,
+  override val evaluate: Evaluate[T] =
+    LazyParameterEvaluator.evaluate(compiledParameter, expressionEvaluator, nodeId, jobData)
+
+}
+
+class SpelTemplateEvaluableLazyParameter[T <: AnyRef](
+    compiledParameter: BaseCompiledParameter,
+    expressionEvaluator: ExpressionEvaluator,
+    nodeId: NodeId,
+    jobData: JobData
+) extends TemplateLazyParameter[T] {
+
+  override val evaluate: Evaluate[T] =
+    LazyParameterEvaluator.evaluate(compiledParameter, expressionEvaluator, nodeId, jobData)
+
+  override def templateExpression: TemplateExpression = compiledParameter.expression match {
+    case expression: SpelExpression =>
+      expression.templateSubexpressions match {
+        case Some(subexpressions) =>
+          val templateParts = subexpressions.map {
+            case TemplatedExpression(expression) => {
+              new TemplateExpressionPart.TemplatedPart {
+                override val evaluate: Evaluate[String] = context => {
+                  expressionEvaluator.evaluate[String](expression, "expressionId", nodeId.id, context)(jobData).value
+                }
+              }
+            }
+            case NonTemplatedValue(value) => TemplateExpressionPart.NonTemplatedPart(value)
+          }
+          TemplateExpression(templateParts)
+        case None =>
+          throw new IllegalStateException("Non SpEL-template expression received in SpelTemplateLazyParameter")
+      }
+    case _ => throw new IllegalStateException("Non SpEL expression received in SpelTemplateLazyParameter")
+  }
+
+  override def returnType: TypingResult = Typed[String]
+}
+
+private[this] object LazyParameterEvaluator {
+
+  def evaluate[T <: AnyRef](
+      compiledParameter: BaseCompiledParameter,
       expressionEvaluator: ExpressionEvaluator,
       nodeId: NodeId,
       jobData: JobData
-  ) =
-    this(compiledParameter, expressionEvaluator, nodeId, jobData, compiledParameter.typingInfo.typingResult)
-
-  override val evaluate: Evaluate[T] = { ctx: Context =>
+  ): Evaluate[T] = { ctx: Context =>
     expressionEvaluator
       .evaluateParameter(compiledParameter, ctx)(nodeId, jobData)
       .value
       .asInstanceOf[T]
   }
 
 }
+
+object EvaluableLazyParameterFactory {
+
+  def build[T <: AnyRef](
+      compiledParameter: BaseCompiledParameter,
+      expressionEvaluator: ExpressionEvaluator,
+      nodeId: NodeId,
+      jobData: JobData,
+      parameterDefinition: ParameterDef,
+      typingResult: TypingResult
+  ): LazyParameter[T] = {
+    EditorBasedLanguageDeterminer.determineLanguageOf(parameterDefinition.editor) match {
+      case Language.SpelTemplate =>
+        new SpelTemplateEvaluableLazyParameter[T](
+          compiledParameter,
+          expressionEvaluator,
+          nodeId,
+          jobData
+        )
+      case _ =>
+        new EvaluableLazyParameter[T](
+          compiledParameter,
+          expressionEvaluator,
+          nodeId,
+          jobData,
+          typingResult
+        )
+    }
+  }
+
+}

scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/compile/nodecompilation/EvaluableLazyParameterCreator.scala

@@ -20,7 +20,7 @@ final class EvaluableLazyParameterCreator[T <: AnyRef](
     override val returnType: TypingResult
 ) extends CustomLazyParameter[T] {
 
-  def create(deps: EvaluableLazyParameterCreatorDeps): EvaluableLazyParameter[T] = {
+  def create(deps: EvaluableLazyParameterCreatorDeps): LazyParameter[T] = {
     createEvaluableLazyParameter(deps)
   }
 
@@ -42,13 +42,15 @@ final class EvaluableLazyParameterCreator[T <: AnyRef](
       override val shouldBeWrappedWithScalaOption: Boolean  = parameterDef.scalaOptionParameter
       override val shouldBeWrappedWithJavaOptional: Boolean = parameterDef.javaOptionalParameter
     }
-    new EvaluableLazyParameter[T](
-      compiledParameter,
-      deps.expressionEvaluator,
-      nodeId,
-      deps.jobData,
-      returnType
-    )
+    EvaluableLazyParameterFactory
+      .build[T](
+        compiledParameter = compiledParameter,
+        expressionEvaluator = deps.expressionEvaluator,
+        nodeId = nodeId,
+        jobData = deps.jobData,
+        parameterDefinition = parameterDef,
+        typingResult = returnType
+      )
   }
 
 }
@@ -92,7 +94,8 @@ class DefaultToEvaluateFunctionConverter(deps: EvaluableLazyParameterCreatorDeps
           p.fun,
           p.transformTypingResult
         )
-      case p: CustomLazyParameter[T] => p
+      case p: CustomLazyParameter[T]   => p
+      case p: TemplateLazyParameter[T] => p
     }
   }
 

scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/compile/nodecompilation/ParameterEvaluator.scala

@@ -95,11 +95,14 @@ class ParameterEvaluator(
   ): LazyParameter[Nothing] = {
     lazyParameterCreationStrategy match {
       case EvaluableLazyParameterStrategy =>
-        new EvaluableLazyParameter(
-          CompiledParameter(exprValue, definition),
-          runtimeExpressionEvaluator,
-          nodeId,
-          jobData
+        val compiledParameter = CompiledParameter(exprValue, definition)
+        EvaluableLazyParameterFactory.build(
+          compiledParameter = compiledParameter,
+          expressionEvaluator = runtimeExpressionEvaluator,
+          nodeId = nodeId,
+          jobData = jobData,
+          parameterDefinition = definition,
+          typingResult = compiledParameter.typingInfo.typingResult
         )
       case PostponedEvaluatorLazyParameterStrategy =>
         new EvaluableLazyParameterCreator(

scenario-compiler/src/main/scala/pl/touk/nussknacker/engine/spel/SpelExpression.scala

@@ -28,7 +28,8 @@ import pl.touk.nussknacker.engine.expression.parse.{CompiledExpression, Expressi
 import pl.touk.nussknacker.engine.graph.expression.Expression.Language
 import pl.touk.nussknacker.engine.graph.expression.{Expression => GraphExpression}
 import pl.touk.nussknacker.engine.spel.SpelExpressionParseError.ExpressionCompilationError
-import pl.touk.nussknacker.engine.spel.SpelExpressionParser.Flavour
+import pl.touk.nussknacker.engine.spel.SpelExpressionParser.{Flavour, Standard}
+import pl.touk.nussknacker.engine.spel.SpelTemplateSubexpression.{NonTemplatedValue, TemplatedExpression}
 import pl.touk.nussknacker.engine.spel.internal.EvaluationContextPreparer
 
 import scala.util.control.NonFatal
@@ -80,6 +81,17 @@ class SpelExpressionEvaluationException(val expression: String, val ctxId: Strin
       cause = cause
     )
 
+sealed trait SpelTemplateSubexpression
+
+object SpelTemplateSubexpression {
+  final case class NonTemplatedValue(val value: String) extends SpelTemplateSubexpression
+
+  final case class TemplatedExpression(expression: SpelExpression) extends SpelTemplateSubexpression {
+    def evaluate: (Context, Map[String, Any]) => String = expression.evaluate[String]
+  }
+
+}
+
 class SpelExpression(
     parsed: ParsedSpelExpression,
     expectedReturnType: TypingResult,
@@ -92,6 +104,37 @@ class SpelExpression(
 
   override val language: Language = flavour.languageId
 
+  def templateSubexpressions: Option[List[SpelTemplateSubexpression]] = {
+    def createTemplatedExpression(expression: org.springframework.expression.spel.standard.SpelExpression) = {
+      val parsedTemplateExpr = ParsedSpelExpression(expression.getExpressionString, parsed.parser, expression)
+      val compiledExpr = new SpelExpression(
+        parsedTemplateExpr,
+        typing.Typed[String],
+        Standard,
+        evaluationContextPreparer
+      )
+      TemplatedExpression(compiledExpr)
+    }
+    flavour.languageId match {
+      case Language.SpelTemplate =>
+        Some(parsed.parsed match {
+          case compositeExpr: CompositeStringExpression =>
+            compositeExpr.getExpressions.toList.map {
+              case lit: LiteralExpression => NonTemplatedValue(lit.getExpressionString)
+              case spelExpr: org.springframework.expression.spel.standard.SpelExpression =>
+                createTemplatedExpression(spelExpr)
+            }
+          case spelExpr: org.springframework.expression.spel.standard.SpelExpression =>
+            List(createTemplatedExpression(spelExpr))
+          case litExpr: LiteralExpression => List(NonTemplatedValue(litExpr.getExpressionString))
+          case other =>
+            throw new IllegalArgumentException(s"Unsupported expression type: [${other.getClass.getName}]")
+
+        })
+      case _ => None
+    }
+  }
+
   private val expectedClass =
     expectedReturnType match {
       case r: SingleTypingResult =>