POC: Fan-in fan-out on top of Scala Graph

Author: franbh

build.sbt

@@ -63,7 +63,8 @@ lazy val `pipeline-sdk` =
           ZIO.core,
           ZIO.schema,
           ZIO.schemaDerivation,
-          Scalaland.chimney
+          Scalaland.chimney,
+          "com.chuusai" %% "shapeless" % "2.3.10"
         )
     )
     .dependsOn(`pipeline-dsl-macros`)

pipeline-sdk/src/main/scala/ai/nikin/pipeline/model/dsl/package.scala

@@ -2,6 +2,7 @@ package ai.nikin.pipeline
 package model
 
 import io.scalaland.chimney.dsl._
+import shapeless.{::, HList, HNil}
 
 package object dsl {
 
@@ -23,9 +24,20 @@ package object dsl {
 
     lazy final val schema: zio.schema.Schema[DATA] = s
 
+    def &[T <: Product](l: Lake[T]): CombinedLake[T :: DATA :: HNil] =
+      CombinedLake[T :: DATA :: HNil](Set(this, l))
+
     def toUntyped: UntypedLake = this.transformInto[UntypedLake]
   }
 
+  case class CombinedLake[L <: HList](lakes: Set[Lake[_]])
+      extends Vertex[CombinedLake[L]](s"combined-${lakes.map(_.name).mkString("&")}") {
+    final override type IN  = L
+    final override type OUT = L
+
+    def &[T <: Product](l: Lake[T]): CombinedLake[T :: L] = CombinedLake[T :: L](lakes + l)
+  }
+
   sealed abstract class Transformation[_IN, _OUT](n: String)
       extends Vertex[Transformation[_IN, _OUT]](n) {
     final override type IN  = _IN

pipeline-sdk/src/main/scala/ai/nikin/pipeline/sdk/package.scala

@@ -21,14 +21,19 @@ package object sdk {
   implicit def toPipelineBuilder[V <: Vertex[V]](v: V): PipelineBuilder[V] =
     PipelineBuilder(v, Graph.empty[Vertex[_], DiEdge])
 
+  import shapeless.HList
+
   class PipelineBuilder[SELF <: Vertex[SELF]](
       private[sdk] val v:     SELF,
       private[sdk] val graph: PipelineDef
   ) {
     def >>>[
         V <: VertexTO[SELF, V]
     ](next: V)(implicit @unused ev: CanMakeEdge[SELF, V]): PipelineBuilder[V] =
-      PipelineBuilder(next, graph ++ Set(v ~> next))
+      v match {
+        case cb: CombinedLake[_] => PipelineBuilder(next, graph ++ cb.lakes.map(_ ~> next))
+        case _ => PipelineBuilder(next, graph ++ Set(v ~> next))
+      }
   }
 
   private object PipelineBuilder {
@@ -45,6 +50,12 @@ package object sdk {
   implicit def lakeToTransform[DATA <: Product, OUT]: CanMakeEdge[Lake[DATA], Transformation[DATA, OUT]] =
     CanMakeEdge[Lake[DATA], Transformation[DATA, OUT]]()
 
+  implicit def combinedLakeToTransform[L <: HList, OUT]: CanMakeEdge[CombinedLake[L], Transformation[L, OUT]] =
+    CanMakeEdge[CombinedLake[L], Transformation[L, OUT]]()
+
+  implicit def transformToCombinedLake[DATA <: HList, IN]: CanMakeEdge[Transformation[IN, DATA], CombinedLake[DATA]] =
+    CanMakeEdge[Transformation[IN, DATA], CombinedLake[DATA]]()
+
   def aggregation[IN <: Product, OUT <: Product](name: String, f: AggregationFunction)(implicit
       inTypeTag:  WeakTypeTag[IN],
       outTypeTag: WeakTypeTag[OUT]
@@ -54,7 +65,7 @@ package object sdk {
       typeTag: WeakTypeTag[DATA]
   ): Lake[DATA] = Lake(name, extractFQN(typeTag))
 
-  private[sdk] def extractFQN[T](typeTag: WeakTypeTag[T]): String = typeTag.tpe.typeSymbol.fullName
+  private[sdk] def extractFQN[T](typeTag: WeakTypeTag[T]): String = typeTag.tpe.toString
 
   @compileTimeOnly("enable macro paradise")
   class Schema extends StaticAnnotation {

pipeline-sdk/src/test/scala/ai/nikin/pipeline/sdk/SdkSpec.scala

@@ -2,7 +2,9 @@ package ai.nikin.pipeline.sdk
 
 import ai.nikin.pipeline.model.dsl._
 import AggregationFunction.{Avg, Sum}
-import ai.nikin.pipeline.sdk.schemas.{RecordA, RecordB}
+import ai.nikin.pipeline.sdk.schemas.{RecordA, RecordB, RecordC}
+import shapeless._
+import zio.schema.DeriveSchema.gen
 
 class SdkSpec extends TestUtils {
   test("SDK - aggregation to lake") {
@@ -30,8 +32,9 @@ class SdkSpec extends TestUtils {
 
   test("SDK - lake to aggregation to lake") {
     val pipeline =
-      lake[RecordA]("lA") >>> aggregation[RecordA, RecordB]("tAB", Avg("col1", "col2")) >>>
-        lake[RecordB]("lB")
+      (lake[RecordA]("lA") & lake[RecordB]("lB") & lake[RecordC]("lC")) >>>
+        aggregation[RecordC :: RecordB :: RecordA :: HNil, RecordC :: RecordB :: HNil]("tAB", Avg("col1", "col2")) >>>
+        (lake[RecordB]("lB") & lake[RecordC]("lC"))
 
     println(pipeline.graph)
   }