[NU-1679] table join component

docs/Changelog.md

@@ -22,6 +22,7 @@
   in table name
 * [#6353](https://github.com/TouK/nussknacker/pull/6353) Performance improvement: simple types such as numbers, boolean, string, date types
   and arrays are serialized/deserialized more optimal in aggregates
+* [#6353](https://github.com/TouK/nussknacker/pull/6353) Added `join` component available in Batch processing mode
 
 1.16.1 (16 July 2024)
 -------------------------

engine/flink/components-api/src/main/scala/pl/touk/nussknacker/engine/flink/api/process/FlinkCustomNodeContext.scala

@@ -58,10 +58,23 @@ case class FlinkCustomNodeContext(
     lazy val forUnknown: TypeInformation[ValueWithContext[AnyRef]] = forType[AnyRef](Unknown)
   }
 
+  def branchValidationContext(branchId: String): ValidationContext = asJoinContext.getOrElse(
+    branchId,
+    throw new IllegalArgumentException(s"No validation context for branchId [$branchId] is defined")
+  )
+
   private def asOneOutputContext: ValidationContext =
-    validationContext.left.getOrElse(throw new IllegalArgumentException("This node is a join, use asJoinContext"))
+    validationContext.left.getOrElse(
+      throw new IllegalArgumentException(
+        "This node is a join, asJoinContext should be used to extract validation context"
+      )
+    )
 
   private def asJoinContext: Map[String, ValidationContext] =
-    validationContext.getOrElse(throw new IllegalArgumentException())
+    validationContext.getOrElse(
+      throw new IllegalArgumentException(
+        "This node is a single input node. asOneOutputContext should be used to extract validation context"
+      )
+    )
 
 }

engine/flink/components/base-tests/src/test/scala/pl/touk/nussknacker/engine/flink/table/join/TableJoinTest.scala

@@ -0,0 +1,105 @@
+package pl.touk.nussknacker.engine.flink.table.join
+
+import com.typesafe.config.ConfigFactory
+import org.apache.flink.api.common.RuntimeExecutionMode
+import org.apache.flink.api.connector.source.Boundedness
+import org.scalatest.Inside
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+import pl.touk.nussknacker.engine.api.component.ComponentDefinition
+import pl.touk.nussknacker.engine.build.{GraphBuilder, ScenarioBuilder}
+import pl.touk.nussknacker.engine.flink.table.FlinkTableComponentProvider
+import pl.touk.nussknacker.engine.flink.table.join.TableJoinTest.OrderProduct
+import pl.touk.nussknacker.engine.flink.test.FlinkSpec
+import pl.touk.nussknacker.engine.flink.util.transformer.join.BranchType
+import pl.touk.nussknacker.engine.util.test.TestScenarioRunner
+import pl.touk.nussknacker.test.ValidatedValuesDetailedMessage
+
+import scala.beans.BeanProperty
+
+class TableJoinTest extends AnyFunSuite with FlinkSpec with Matchers with Inside with ValidatedValuesDetailedMessage {
+
+  import pl.touk.nussknacker.engine.flink.util.test.FlinkTestScenarioRunner._
+  import pl.touk.nussknacker.engine.spel.SpelExtension._
+
+  import scala.jdk.CollectionConverters._
+
+  private lazy val additionalComponents: List[ComponentDefinition] =
+    FlinkTableComponentProvider.configIndependentComponents ::: Nil
+
+  private lazy val runner = TestScenarioRunner
+    .flinkBased(ConfigFactory.empty(), flinkMiniCluster)
+    .withExtraComponents(additionalComponents)
+    .build()
+
+  private val MainBranchId = "main"
+
+  private val JoinedBranchId = "joined"
+
+  private val JoinNodeId = "joined-node-id"
+
+  test("should be able to join") {
+    val scenario = ScenarioBuilder
+      .streaming("sample-join-last")
+      .sources(
+        GraphBuilder
+          .source("orders-source", TestScenarioRunner.testDataSource)
+          .filter("orders-filter", "#input.type == 'order'".spel)
+          .branchEnd(MainBranchId, JoinNodeId),
+        GraphBuilder
+          .source("products-source", TestScenarioRunner.testDataSource)
+          .filter("product-filter", "#input.type == 'product'".spel)
+          .branchEnd(JoinedBranchId, JoinNodeId),
+        GraphBuilder
+          .join(
+            JoinNodeId,
+            "join",
+            Some("product"),
+            List(
+              MainBranchId -> List(
+                "branchType" -> s"T(${classOf[BranchType].getName}).MAIN".spel,
+                "key"        -> s"#input.productId.toString".spel
+              ),
+              JoinedBranchId -> List(
+                "branchType" -> s"T(${classOf[BranchType].getName}).JOINED".spel,
+                "key"        -> s"#input.id.toString".spel
+              )
+            ),
+            "output" -> "#input".spel,
+          )
+          .emptySink("end", TestScenarioRunner.testResultSink, "value" -> "{#input, #product}".spel)
+      )
+
+    val result = runner.runWithData(
+      scenario,
+      List(
+        OrderProduct("product", 1, -1),
+        OrderProduct("order", 10, 1),
+      ),
+      Boundedness.BOUNDED,
+      Some(RuntimeExecutionMode.BATCH)
+    )
+
+    result.validValue.successes shouldBe List(
+      List(OrderProduct("order", 10, 1), OrderProduct("product", 1, -1)).asJava,
+    )
+  }
+
+}
+
+object TableJoinTest {
+
+  // TODO: split into separate classes and pass two streams to separate source nodes
+  // productId is dedicated only for order events
+  // It have to by POJO in order by acceptable by table api operators
+  case class OrderProduct(
+      @BeanProperty var `type`: String,
+      @BeanProperty var id: Int,
+      @BeanProperty var productId: Int
+  ) {
+
+    def this() = this(null, -1, -1)
+
+  }
+
+}

engine/flink/components/table/src/main/scala/pl/touk/nussknacker/engine/flink/table/FlinkTableComponentProvider.scala

@@ -10,6 +10,7 @@ import pl.touk.nussknacker.engine.flink.table.aggregate.TableAggregationFactory
 import pl.touk.nussknacker.engine.flink.table.extractor.TableExtractor.extractTablesFromFlinkRuntime
 import pl.touk.nussknacker.engine.flink.table.extractor.SqlStatementReader
 import pl.touk.nussknacker.engine.flink.table.extractor.SqlStatementReader.SqlStatement
+import pl.touk.nussknacker.engine.flink.table.join.TableJoinComponent
 import pl.touk.nussknacker.engine.flink.table.sink.TableSinkFactory
 import pl.touk.nussknacker.engine.flink.table.source.TableSourceFactory
 import pl.touk.nussknacker.engine.util.ResourceLoader
@@ -88,6 +89,10 @@ object FlinkTableComponentProvider {
     ComponentDefinition(
       "aggregate",
       new TableAggregationFactory()
+    ),
+    ComponentDefinition(
+      "join",
+      TableJoinComponent
     )
   )