Template parts handling approach change: expression level instead of LazyParameter level

build.sbt

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

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

@@ -1,6 +1,5 @@
 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}
 
@@ -69,25 +68,6 @@ 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]

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

@@ -0,0 +1,15 @@
+package pl.touk.nussknacker.engine.api
+
+case class TemplateEvaluationResult(renderedParts: List[TemplateRenderedPart]) {
+  def renderedTemplate: String = renderedParts.map(_.value).mkString("")
+}
+
+sealed trait TemplateRenderedPart {
+  def value: String
+}
+
+object TemplateRenderedPart {
+  case class RenderedLiteral(value: String) extends TemplateRenderedPart
+
+  case class RenderedSubExpression(value: String) extends TemplateRenderedPart
+}

docs/MigrationGuide.md

@@ -44,6 +44,9 @@ To see the biggest differences please consult the [changelog](Changelog.md).
 * [#6988](https://github.com/TouK/nussknacker/pull/6988) Removed unused API classes: `MultiMap`, `TimestampedEvictableStateFunction`.
   `MultiMap` was incorrectly handled by Flink's default Kryo serializer, so if you want to copy it to your code
   you should write and register a proper serializer.
+* [#7162](https://github.com/TouK/nussknacker/pull/7162) When component declares that requires parameter with either `SpelTemplateParameterEditor` 
+  or `SqlParameterEditor` editor, in the runtime, for the expression evaluation result, will be used the new `TemplateEvaluationResult` 
+  class instead of `String` class. To access the previous `String` use `TemplateEvaluationResult.renderedTemplate` method.
 
 ### REST API changes
 

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

@@ -0,0 +1,118 @@
+package pl.touk.nussknacker.engine.flink
+
+import com.typesafe.config.ConfigFactory
+import org.apache.flink.api.common.functions.FlatMapFunction
+import org.apache.flink.api.connector.source.Boundedness
+import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.util.Collector
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+import pl.touk.nussknacker.engine.api.TemplateRenderedPart.{RenderedLiteral, RenderedSubExpression}
+import pl.touk.nussknacker.engine.api._
+import pl.touk.nussknacker.engine.api.component.{BoundedStreamComponent, ComponentDefinition}
+import pl.touk.nussknacker.engine.api.context.ValidationContext
+import pl.touk.nussknacker.engine.api.context.transformation.{DefinedLazyParameter, NodeDependencyValue, SingleInputDynamicComponent}
+import pl.touk.nussknacker.engine.api.definition.{NodeDependency, OutputVariableNameDependency, Parameter, SpelTemplateParameterEditor}
+import pl.touk.nussknacker.engine.api.parameter.ParameterName
+import pl.touk.nussknacker.engine.api.typed.typing.Typed
+import pl.touk.nussknacker.engine.build.ScenarioBuilder
+import pl.touk.nussknacker.engine.flink.api.process.{AbstractOneParamLazyParameterFunction, FlinkCustomNodeContext, FlinkCustomStreamTransformation}
+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 SpelTemplateLazyParameterTest extends AnyFunSuite with FlinkSpec with Matchers with ValidatedValuesDetailedMessage {
+
+  private lazy val runner = TestScenarioRunner
+    .flinkBased(ConfigFactory.empty(), flinkMiniCluster)
+    .withExecutionMode(ExecutionMode.Batch)
+    .withExtraComponents(
+      List(ComponentDefinition("spelTemplatePartsCustomTransformer", SpelTemplatePartsCustomTransformer))
+    )
+    .build()
+
+  test("flink custom transformer using spel template rendered parts") {
+    val scenario = ScenarioBuilder
+      .streaming("test")
+      .source("source", TestScenarioRunner.testDataSource)
+      .customNode(
+        "custom",
+        "output",
+        "spelTemplatePartsCustomTransformer",
+        "template" -> Expression.spelTemplate(s"Hello#{#input}")
+      )
+      .emptySink("sink", TestScenarioRunner.testResultSink, "value" -> "#output".spel)
+
+    val result = runner.runWithData(scenario, List(1, 2, 3), Boundedness.BOUNDED)
+    result.validValue.errors shouldBe empty
+    result.validValue.successes shouldBe List(
+      "[Hello]-literal[1]-subexpression",
+      "[Hello]-literal[2]-subexpression",
+      "[Hello]-literal[3]-subexpression"
+    )
+  }
+
+}
+
+object SpelTemplatePartsCustomTransformer
+    extends CustomStreamTransformer
+    with SingleInputDynamicComponent[FlinkCustomStreamTransformation]
+    with BoundedStreamComponent {
+
+  private val spelTemplateParameterName = ParameterName("template")
+
+  private val spelTemplateParameter = Parameter
+    .optional[String](spelTemplateParameterName)
+    .copy(
+      isLazyParameter = true,
+      editor = Some(SpelTemplateParameterEditor)
+    )
+
+  override type State = Unit
+
+  override def contextTransformation(context: ValidationContext, dependencies: List[NodeDependencyValue])(
+      implicit nodeId: NodeId
+  ): SpelTemplatePartsCustomTransformer.ContextTransformationDefinition = {
+    case TransformationStep(Nil, _) => NextParameters(List(spelTemplateParameter))
+    case TransformationStep((`spelTemplateParameterName`, DefinedLazyParameter(_)) :: Nil, _) =>
+      val outName = OutputVariableNameDependency.extract(dependencies)
+      FinalResults(context.withVariableUnsafe(outName, Typed[String]), List.empty)
+  }
+
+  override def nodeDependencies: List[NodeDependency] = List(OutputVariableNameDependency)
+
+  override def implementation(
+      params: Params,
+      dependencies: List[NodeDependencyValue],
+      finalState: Option[Unit]
+  ): FlinkCustomStreamTransformation = {
+    val templateLazyParam: LazyParameter[TemplateEvaluationResult] =
+      params.extractUnsafe[LazyParameter[TemplateEvaluationResult]](spelTemplateParameterName)
+    FlinkCustomStreamTransformation {
+      (dataStream: DataStream[Context], flinkCustomNodeContext: FlinkCustomNodeContext) =>
+        dataStream.flatMap(
+          new AbstractOneParamLazyParameterFunction[TemplateEvaluationResult](
+            templateLazyParam,
+            flinkCustomNodeContext.lazyParameterHelper
+          ) with FlatMapFunction[Context, ValueWithContext[String]] {
+            override def flatMap(value: Context, out: Collector[ValueWithContext[String]]): Unit = {
+              collectHandlingErrors(value, out) {
+                val templateResult = evaluateParameter(value)
+                val result = templateResult.renderedParts.map {
+                  case RenderedLiteral(value)       => s"[$value]-literal"
+                  case RenderedSubExpression(value) => s"[$value]-subexpression"
+                }.mkString
+                ValueWithContext(result, value)
+              }
+            }
+          },
+          flinkCustomNodeContext.valueWithContextInfo.forClass[String]
+        ).asInstanceOf[DataStream[ValueWithContext[AnyRef]]]
+    }
+  }
+
+}

engine/flink/management/dev-model/src/main/scala/pl/touk/nussknacker/engine/management/sample/LoggingService.scala

@@ -18,7 +18,9 @@ object LoggingService extends EagerService {
   def prepare(
       @ParamName("logger") @Nullable loggerName: String,
       @ParamName("level") @DefaultValue("T(org.slf4j.event.Level).DEBUG") level: Level,
-      @ParamName("message") @SimpleEditor(`type` = SimpleEditorType.SPEL_TEMPLATE_EDITOR) message: LazyParameter[String]
+      @ParamName("message") @SimpleEditor(`type` = SimpleEditorType.SPEL_TEMPLATE_EDITOR) message: LazyParameter[
+        TemplateEvaluationResult
+      ]
   )(implicit metaData: MetaData, nodeId: NodeId): ServiceInvoker =
     new ServiceInvoker {
 
@@ -31,7 +33,7 @@ object LoggingService extends EagerService {
           collector: ServiceInvocationCollector,
           componentUseCase: ComponentUseCase
       ): Future[Any] = {
-        val msg = message.evaluate(context)
+        val msg = message.evaluate(context).renderedTemplate
         level match {
           case Level.TRACE => logger.trace(msg)
           case Level.DEBUG => logger.debug(msg)