[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 Literal(value: String) extends TemplateExpressionPart
+
+      trait Placeholder 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,129 @@
+package pl.touk.nussknacker.engine.flink
+
+import com.typesafe.config.ConfigFactory
+import org.apache.flink.api.connector.source.Boundedness
+import org.apache.flink.streaming.api.datastream.{DataStream, DataStreamSink}
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
+import org.apache.flink.streaming.api.functions.sink
+import org.apache.flink.streaming.api.functions.sink.SinkFunction
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+import pl.touk.nussknacker.engine.api.LazyParameter.TemplateLazyParameter
+import pl.touk.nussknacker.engine.api.LazyParameter.TemplateLazyParameter.TemplateExpressionPart.{Literal, Placeholder}
+import pl.touk.nussknacker.engine.api.{Context, LazyParameter, MethodToInvoke, NodeId, Params, ValueWithContext}
+import pl.touk.nussknacker.engine.api.component.{BoundedStreamComponent, Component, ComponentDefinition}
+import pl.touk.nussknacker.engine.api.context.{OutputVar, ValidationContext}
+import pl.touk.nussknacker.engine.api.context.transformation.{
+  DefinedLazyParameter,
+  NodeDependencyValue,
+  SingleInputDynamicComponent
+}
+import pl.touk.nussknacker.engine.api.definition.{
+  NodeDependency,
+  OutputVariableNameDependency,
+  ParameterDeclaration,
+  SpelTemplateParameterEditor
+}
+import pl.touk.nussknacker.engine.api.parameter.ParameterName
+import pl.touk.nussknacker.engine.api.process.{Sink, SinkFactory, Source, SourceFactory}
+import pl.touk.nussknacker.engine.api.typed.typing
+import pl.touk.nussknacker.engine.build.ScenarioBuilder
+import pl.touk.nussknacker.engine.flink.api.process.{FlinkCustomNodeContext, FlinkSink, StandardFlinkSource}
+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.util.test.TestScenarioRunner
+import pl.touk.nussknacker.test.ValidatedValuesDetailedMessage
+
+class TemplateLazyParameterTest extends AnyFunSuite with FlinkSpec with Matchers with ValidatedValuesDetailedMessage {
+
+  private lazy val runner = TestScenarioRunner
+    .flinkBased(ConfigFactory.empty(), flinkMiniCluster)
+    .withExecutionMode(ExecutionMode.Batch)
+    .withExtraComponents(
+      List(ComponentDefinition("templateAstOperationSink", SpelTemplateAstOperationSink))
+    )
+    .build()
+
+  test("should use spel template ast operation parameter") {
+    val scenario = ScenarioBuilder
+      .streaming("test")
+      .source("source", TestScenarioRunner.testDataSource)
+      .emptySink(
+        "end",
+        "templateAstOperationSink",
+        "templateOutput" -> Expression.spelTemplate(s"Hello#{#input}")
+      )
+
+    val result = runner.runWithData(scenario, List(1, 2, 3), Boundedness.BOUNDED)
+    println(result)
+    result.validValue.successes shouldBe List(
+      "[Hello]-literal[1]-templated",
+      "[Hello]-literal[2]-templated",
+      "[Hello]-literal[3]-templated"
+    )
+  }
+
+}
+
+object SpelTemplateAstOperationSink
+    extends SingleInputDynamicComponent[Sink]
+    with SinkFactory
+    with BoundedStreamComponent {
+
+  private val spelTemplateParameter = ParameterDeclaration
+    .lazyOptional[String](ParameterName("templateOutput"))
+    .withCreator(modify =
+      _.copy(
+        editor = Some(SpelTemplateParameterEditor)
+      )
+    )
+
+  override type State = Unit
+
+  override def contextTransformation(context: ValidationContext, dependencies: List[NodeDependencyValue])(
+      implicit nodeId: NodeId
+  ): SpelTemplateAstOperationSink.ContextTransformationDefinition = {
+    case TransformationStep(Nil, _) => NextParameters(List(spelTemplateParameter.createParameter()))
+    case TransformationStep((ParameterName("templateOutput"), DefinedLazyParameter(_)) :: Nil, _) =>
+      FinalResults(context, List.empty)
+  }
+
+  override def nodeDependencies: List[NodeDependency] = List.empty
+
+  override def implementation(params: Params, dependencies: List[NodeDependencyValue], finalState: Option[Unit]): Sink =
+    new FlinkSink {
+      override type Value = String
+
+      val valueFromParam: LazyParameter[String] = spelTemplateParameter.extractValueUnsafe(params)
+
+      override def prepareValue(
+          dataStream: DataStream[Context],
+          flinkCustomNodeContext: FlinkCustomNodeContext
+      ): DataStream[ValueWithContext[Value]] = {
+        dataStream.flatMap(
+          flinkCustomNodeContext.lazyParameterHelper.lazyMapFunction(valueFromParam),
+          flinkCustomNodeContext.valueWithContextInfo.forType(valueFromParam.returnType)
+        )
+      }
+
+      override def registerSink(
+          dataStream: DataStream[ValueWithContext[String]],
+          flinkNodeContext: FlinkCustomNodeContext
+      ): DataStreamSink[_] = {
+        println(dataStream)
+        dataStream.addSink(new SinkFunction[ValueWithContext[String]] {
+          override def invoke(value: ValueWithContext[String], context: SinkFunction.Context): Unit = {
+            println(value)
+            println("debug")
+
+          }
+        })
+      }
+
+    }
+
+}

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

@@ -1,10 +1,14 @@
 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.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.expression.ExpressionEvaluator
+import pl.touk.nussknacker.engine.graph.expression.Expression.Language
+import pl.touk.nussknacker.engine.spel.SpelExpression
+import pl.touk.nussknacker.engine.spel.SpelTemplateExpressionPart.{Literal, Placeholder}
 
 class EvaluableLazyParameter[T <: AnyRef](
     compiledParameter: BaseCompiledParameter,
@@ -14,19 +18,87 @@ 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 Placeholder(expression) => {
+              new TemplateExpressionPart.Placeholder {
+                override val evaluate: Evaluate[String] = context => {
+                  expressionEvaluator.evaluate[String](expression, "expressionId", nodeId.id, context)(jobData).value
+                }
+              }
+            }
+            case Literal(value) => TemplateExpressionPart.Literal(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,
+      typingResult: TypingResult
+  ): LazyParameter[T] = {
+    compiledParameter.expression.language 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,14 @@ 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,
+        typingResult = returnType
+      )
   }
 
 }
@@ -92,7 +93,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,13 @@ 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,
+          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.SpelTemplateExpressionPart.{Literal, Placeholder}
 import pl.touk.nussknacker.engine.spel.internal.EvaluationContextPreparer
 
 import scala.util.control.NonFatal
@@ -80,6 +81,13 @@ class SpelExpressionEvaluationException(val expression: String, val ctxId: Strin
       cause = cause
     )
 
+sealed trait SpelTemplateExpressionPart
+
+object SpelTemplateExpressionPart {
+  final case class Literal(value: String)                  extends SpelTemplateExpressionPart
+  final case class Placeholder(expression: SpelExpression) extends SpelTemplateExpressionPart
+}
+
 class SpelExpression(
     parsed: ParsedSpelExpression,
     expectedReturnType: TypingResult,
@@ -92,6 +100,27 @@ class SpelExpression(
 
   override val language: Language = flavour.languageId
 
+  def templateSubexpressions: Option[List[SpelTemplateExpressionPart]] = {
+    def parseTemplate(expression: Expression): List[SpelTemplateExpressionPart] = expression match {
+      case lit: LiteralExpression => List(Literal(lit.getExpressionString))
+      case spelExpr: org.springframework.expression.spel.standard.SpelExpression =>
+        val parsedTemplateExpr = ParsedSpelExpression(spelExpr.getExpressionString, parsed.parser, spelExpr)
+        val compiledExpr = new SpelExpression(
+          parsedTemplateExpr,
+          typing.Typed[String],
+          Standard,
+          evaluationContextPreparer
+        )
+        List(Placeholder(compiledExpr))
+      case compositeExpr: CompositeStringExpression => compositeExpr.getExpressions.toList.flatMap(parseTemplate)
+      case other => throw new IllegalArgumentException(s"Unsupported expression type: [${other.getClass.getName}]")
+    }
+    flavour.languageId match {
+      case Language.SpelTemplate => Some(parseTemplate(parsed.parsed))
+      case _                     => None
+    }
+  }
+
   private val expectedClass =
     expectedReturnType match {
       case r: SingleTypingResult =>