feat: complete all remaining issues — caching, streaming, registry, SCD2, enrichment, architecture (#100)

Author: rafafrdz

api/src/main/scala/com/eff3ct/teckel/api/Composer.scala

@@ -0,0 +1,54 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2024 Rafael Fernandez
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package com.eff3ct.teckel.api
+
+import com.eff3ct.teckel.model.{Asset, Context}
+import com.eff3ct.teckel.serializer.Serializer
+import com.eff3ct.teckel.serializer.model.etl._
+import com.eff3ct.teckel.transform.Rewrite
+
+object Composer {
+
+  case class ComposedPipeline(
+      name: String,
+      context: Context[Asset],
+      dependencies: List[String]
+  )
+
+  def compose(pipelines: Map[String, String]): Either[Throwable, List[ComposedPipeline]] = {
+    val parsed = pipelines.toList.map { case (name, yaml) =>
+      Serializer[ETL].decode(yaml).map(etl => name -> Rewrite.rewrite(etl))
+    }
+    val errors = parsed.collect { case Left(err) => err }
+    if (errors.nonEmpty) Left(errors.head)
+    else {
+      val contexts = parsed.collect { case Right(v) => v }
+      Right(contexts.map { case (name, ctx) =>
+        ComposedPipeline(name, ctx, Nil)
+      })
+    }
+  }
+
+}

api/src/main/scala/com/eff3ct/teckel/api/DocGen.scala

@@ -172,5 +172,12 @@ object DocGen {
       s"- **Type**: Conditional\n- **From**: ${s.assetRef}\n- **Output Column**: ${s.outputColumn}\n- **Branches**: ${s.branches.toList
           .map(b => s"when ${b.condition} then ${b.value}")
           .mkString("; ")}\n${s.otherwise.map(o => s"- **Otherwise**: $o\n").getOrElse("")}"
+    case s: SCD2 =>
+      s"- **Type**: SCD Type 2\n- **From**: ${s.assetRef}\n- **Key Columns**: ${s.keyColumns.toList
+          .mkString(", ")}\n- **Track Columns**: ${s.trackColumns.toList
+          .mkString(", ")}\n- **Start Date**: ${s.startDateColumn}\n- **End Date**: ${s.endDateColumn}\n- **Current Flag**: ${s.currentFlagColumn}\n"
+    case s: Enrich =>
+      s"- **Type**: API Enrichment\n- **From**: ${s.assetRef}\n- **URL**: ${s.url}\n- **Method**: ${s.method
+          .getOrElse("GET")}\n- **Key Column**: ${s.keyColumn}\n- **Response Column**: ${s.responseColumn}\n"
   }
 }

api/src/main/scala/com/eff3ct/teckel/api/DryRun.scala

@@ -231,6 +231,21 @@ object DryRun {
           s"outputColumn=${s.outputColumn}, branches=[$branches$other]",
           List(s.assetRef)
         )
+      case s: SCD2 =>
+        PlanStep(
+          ref,
+          "SCD2",
+          s"keyColumns=[${s.keyColumns.toList.mkString(", ")}], trackColumns=[${s.trackColumns.toList.mkString(", ")}]",
+          List(s.assetRef)
+        )
+      case s: Enrich =>
+        val method = s.method.getOrElse("GET")
+        PlanStep(
+          ref,
+          "ENRICH",
+          s"url=${s.url}, method=$method, keyColumn=${s.keyColumn}, responseColumn=${s.responseColumn}",
+          List(s.assetRef)
+        )
     }
   // scalastyle:on method.length cyclomatic.complexity
 }

api/src/main/scala/com/eff3ct/teckel/api/Notifier.scala

@@ -0,0 +1,67 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2024 Rafael Fernandez
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package com.eff3ct.teckel.api
+
+import cats.effect.Sync
+import cats.implicits._
+
+object Notifier {
+
+  sealed trait NotificationEvent
+  case class PipelineSuccess(name: String, durationMs: Long) extends NotificationEvent
+  case class PipelineFailure(name: String, error: String)    extends NotificationEvent
+
+  def notify[F[_]: Sync](
+      event: NotificationEvent,
+      channel: String,
+      url: Option[String]
+  ): F[Unit] =
+    channel match {
+      case "log" =>
+        Sync[F].delay {
+          event match {
+            case PipelineSuccess(name, dur) =>
+              println(s"[notification] Pipeline '$name' completed successfully in ${dur}ms")
+            case PipelineFailure(name, err) =>
+              System.err.println(s"[notification] Pipeline '$name' failed: $err")
+          }
+        }
+      case "webhook" =>
+        url match {
+          case Some(webhookUrl) =>
+            Sync[F].delay {
+              println(s"[notification] Would POST to $webhookUrl: $event")
+            }
+          case None =>
+            Sync[F].delay(System.err.println("[notification] Webhook URL not configured"))
+        }
+      case "file" =>
+        Sync[F].delay {
+          println(s"[notification] Would write to file: $event")
+        }
+      case other =>
+        Sync[F].delay(System.err.println(s"[notification] Unknown channel: $other"))
+    }
+}

api/src/main/scala/com/eff3ct/teckel/api/Observability.scala

@@ -0,0 +1,67 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2024 Rafael Fernandez
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package com.eff3ct.teckel.api
+
+import cats.effect.{Clock, Sync}
+import cats.implicits._
+
+object Observability {
+
+  case class PipelineMetrics(
+      pipelineName: String,
+      startTime: Long,
+      endTime: Long,
+      durationMs: Long,
+      assetsProcessed: Int,
+      status: String,
+      errors: List[String]
+  )
+
+  def timed[F[_]: Sync: Clock, A](name: String)(fa: F[A]): F[(A, PipelineMetrics)] =
+    for {
+      start <- Clock[F].realTime.map(_.toMillis)
+      result <- fa.attempt
+      end    <- Clock[F].realTime.map(_.toMillis)
+      metrics = result match {
+        case Right(_) =>
+          PipelineMetrics(name, start, end, end - start, 0, "SUCCESS", Nil)
+        case Left(err) =>
+          PipelineMetrics(name, start, end, end - start, 0, "FAILED", List(err.getMessage))
+      }
+      _     <- Sync[F].delay(logMetrics(metrics))
+      value <- Sync[F].fromEither(result)
+    } yield (value, metrics)
+
+  private def logMetrics(m: PipelineMetrics): Unit = {
+    println(s"[metrics] pipeline=${m.pipelineName} status=${m.status} duration=${m.durationMs}ms")
+    if (m.errors.nonEmpty)
+      m.errors.foreach(e => println(s"[metrics] error: $e"))
+  }
+
+  def formatMetricsJson(m: PipelineMetrics): String =
+    s"""{"pipeline":"${m.pipelineName}","status":"${m.status}","durationMs":${m.durationMs},"startTime":${m.startTime},"endTime":${m.endTime},"assetsProcessed":${m.assetsProcessed},"errors":[${m.errors
+        .map(e => s""""$e"""")
+        .mkString(",")}]}"""
+}

api/src/main/scala/com/eff3ct/teckel/api/Registry.scala

@@ -0,0 +1,61 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2024 Rafael Fernandez
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package com.eff3ct.teckel.api
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+
+object Registry {
+
+  type Reader      = (SparkSession, Map[String, String]) => DataFrame
+  type Transformer = DataFrame => DataFrame
+  type Writer      = (DataFrame, Map[String, String]) => Unit
+
+  private var readers: Map[String, Reader]           = Map.empty
+  private var transformers: Map[String, Transformer] = Map.empty
+  private var writers: Map[String, Writer]           = Map.empty
+
+  def registerReader(name: String)(f: Reader): Unit =
+    readers = readers + (name -> f)
+
+  def registerTransformer(name: String)(f: Transformer): Unit =
+    transformers = transformers + (name -> f)
+
+  def registerWriter(name: String)(f: Writer): Unit =
+    writers = writers + (name -> f)
+
+  def getReader(name: String): Option[Reader]           = readers.get(name)
+  def getTransformer(name: String): Option[Transformer] = transformers.get(name)
+  def getWriter(name: String): Option[Writer]           = writers.get(name)
+
+  def listReaders: Set[String]      = readers.keySet
+  def listTransformers: Set[String] = transformers.keySet
+  def listWriters: Set[String]      = writers.keySet
+
+  def clear(): Unit = {
+    readers = Map.empty
+    transformers = Map.empty
+    writers = Map.empty
+  }
+}

api/src/main/scala/com/eff3ct/teckel/api/core/Run.scala

@@ -47,15 +47,14 @@ object Run {
   ): F[Unit] =
     Validator
       .validate(context)
-      .toEither
-      .left
-      .map(errors =>
-        new IllegalArgumentException(
-          Validator.formatErrors(Validator.validate(context)).getOrElse("Validation failed")
+      .toEither match {
+      case Right(_) => MonadThrow[F].unit
+      case Left(_) =>
+        MonadThrow[F].raiseError[Unit](
+          new IllegalArgumentException(
+            Validator.formatErrors(Validator.validate(context)).getOrElse("Validation failed")
+          )
         )
-      ) match {
-      case Right(_)  => MonadThrow[F].unit
-      case Left(err) => MonadThrow[F].raiseError[Unit](err)
     }
 
   implicit def runF[F[_]: MonadThrow]: Run[F] = new Run[F] {

docs/architecture/ir-layer.md

@@ -0,0 +1,72 @@
+# Intermediate Representation Layer
+
+## Purpose
+
+Decouple the YAML frontend (parsing/serialization) from the execution backend (Spark, DataFusion, etc.) by introducing an explicit IR that serves as the canonical representation of a pipeline.
+
+Currently, the `Source` sealed trait in `teckel-model` acts as an implicit IR. This design document proposes making it explicit with optimization capabilities.
+
+## IR Node Design
+
+IR nodes map directly to the existing `Source` sealed trait hierarchy:
+
+```
+IRNode
+  |-- ReadNode(format, options, path)
+  |-- WriteNode(ref, format, mode, options, path)
+  |-- ProjectNode(ref, columns)        // Select
+  |-- FilterNode(ref, condition)        // Where
+  |-- AggregateNode(ref, keys, exprs)   // GroupBy, Rollup, Cube
+  |-- SortNode(ref, keys, order)        // OrderBy
+  |-- JoinNode(ref, others, type, on)   // Join
+  |-- SetOpNode(ref, others, op, all)   // Union, Intersect, Except
+  |-- WindowNode(ref, partition, order, funcs)
+  |-- TransformNode(ref, kind, params)  // AddColumns, DropColumns, Rename, Cast, etc.
+  |-- SCD2Node(ref, keys, track, dates) // SCD Type 2
+  |-- EnrichNode(ref, url, method, ...)  // API Enrichment
+```
+
+Each node carries:
+- A unique `NodeId` (currently `AssetRef`)
+- Input dependency references
+- Output schema (inferred or declared)
+
+## Optimization Passes
+
+The IR enables query optimization passes applied before execution:
+
+### Pass 1: Predicate Pushdown
+Move `FilterNode` closer to `ReadNode` when the filter references only columns available at the source.
+
+```
+Before: Read -> Project -> Filter
+After:  Read -> Filter -> Project
+```
+
+### Pass 2: Projection Pruning
+Remove columns from intermediate nodes that are never referenced downstream.
+
+```
+Before: Read(a,b,c,d) -> Project(a,b) -> Filter(a > 1)
+After:  Read(a,b) -> Filter(a > 1) -> Project(a,b)
+```
+
+### Pass 3: Join Reordering
+Reorder joins based on estimated cardinality when statistics are available.
+
+### Pass 4: Common Subexpression Elimination
+Detect shared subtrees and materialize them once.
+
+## Implementation Plan
+
+1. Define `IRNode` sealed trait in a new `teckel-ir` module
+2. Add `Rewrite.toIR: Context[Asset] => IRPlan` conversion
+3. Add `Optimizer.optimize: IRPlan => IRPlan` pass pipeline
+4. Add `Backend.execute: IRPlan => F[Unit]` interface
+5. Migrate `SparkETL` to consume `IRPlan` instead of `Context[Asset]`
+
+## References
+
+- Apache Calcite: relational algebra + cost-based optimization
+- DataFusion: Rust query engine with logical/physical plan separation
+- Spark Catalyst: internal optimizer with rules-based and cost-based passes