From c7fbf3894f6059cd198fc6fcf2af49b8899004bf Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Sun, 10 May 2026 23:25:56 +0900
Subject: [PATCH 1/2] Add Step DSL (Source/Transform/Sink) for Spark jobs

Introduces the in-job Step model and its YAML inline form for a single
Spark job's data flow, complementing the workflow DSL (#310) which covers
job-to-job relationships.

- Step sealed trait: Source / Transform / Sink with arities 0->1, N->1, 1->0
- Plan type-state DSL: ~>, +, .as, fanOut
- Executor: AST-level memoization, fanOut cache/unpersist
- StepsRunnerJob + StepsBuilder + ClassResolver for inline YAML chains
- Built-ins: FileSource, SampleSource, SqlTransform, CacheTransform,
  FileSink, ShowSink

Generated with [Claude Code](https://claude.ai/code)
via [Happy](https://happy.engineering)

Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Happy <yesreply@happy.engineering>
---
 .../kakao/actionbase/pipeline/dsl/Job.scala   |  76 ++++++++++
 .../kakao/actionbase/pipeline/dsl/Plan.scala  | 132 ++++++++++++++++
 .../kakao/actionbase/pipeline/dsl/Step.scala  |  27 ++++
 .../actionbase/pipeline/dsl/package.scala     |  10 ++
 .../pipeline/jobs/StepsRunnerJob.scala        |  29 ++++
 .../pipeline/runner/ClassResolver.scala       |  32 ++++
 .../pipeline/runner/StepsBuilder.scala        | 126 ++++++++++++++++
 .../pipeline/steps/sink/FileSink.scala        |  17 +++
 .../pipeline/steps/sink/ShowSink.scala        |  10 ++
 .../pipeline/steps/source/FileSource.scala    |  16 ++
 .../pipeline/steps/source/SampleSource.scala  |  13 ++
 .../steps/transform/CacheTransform.scala      |  18 +++
 .../steps/transform/SqlTransform.scala        |  18 +++
 .../pipeline/workflow/StepSpec.scala          |  17 +++
 .../kakao/actionbase/pipeline/SparkTest.scala |  16 ++
 .../pipeline/runner/StepsBuilderTest.scala    | 141 ++++++++++++++++++
 .../pipeline/steps/sink/FileSinkTest.scala    |  49 ++++++
 .../pipeline/steps/sink/ShowSinkTest.scala    |  16 ++
 .../steps/source/FileSourceTest.scala         |  58 +++++++
 .../steps/source/SampleSourceTest.scala       |  38 +++++
 .../steps/transform/CacheTransformTest.scala  |  42 ++++++
 .../steps/transform/SqlTransformTest.scala    |  54 +++++++
 22 files changed, 955 insertions(+)
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Job.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Plan.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Step.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/package.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/jobs/StepsRunnerJob.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/ClassResolver.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/StepsBuilder.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/sink/FileSink.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/sink/ShowSink.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/source/FileSource.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/source/SampleSource.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/transform/CacheTransform.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/transform/SqlTransform.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/StepSpec.scala
 create mode 100644 pipeline/src/test/scala/com/kakao/actionbase/pipeline/SparkTest.scala
 create mode 100644 pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/StepsBuilderTest.scala
 create mode 100644 pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/sink/FileSinkTest.scala
 create mode 100644 pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/sink/ShowSinkTest.scala
 create mode 100644 pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/source/FileSourceTest.scala
 create mode 100644 pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/source/SampleSourceTest.scala
 create mode 100644 pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/transform/CacheTransformTest.scala
 create mode 100644 pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/transform/SqlTransformTest.scala

diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Job.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Job.scala
new file mode 100644
index 000000000..f4fd9d883
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Job.scala
@@ -0,0 +1,76 @@
+package com.kakao.actionbase.pipeline.dsl
+
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import org.apache.spark.sql.SparkSession
+
+import scala.reflect.ClassTag
+
+/** One spark-submit unit. Subclass and implement `plan(cfg)`.
+  *
+  * Two entry points share the same Cfg-binding mapper:
+  *   - `main(argv)` — for spark-submit / CLI: parses `--key=value` argv into Cfg.
+  *   - `planFromMap(args)` — for in-process runners: binds an arbitrary YAML-shaped Map onto Cfg, supporting nested
+  *     fields like `Seq[StepSpec]` that argv cannot represent.
+  *
+  * {{{
+  * case class MyConfig(in: String, out: String)
+  *
+  * object MyJob extends Job[MyConfig] {
+  *   override def plan(cfg: MyConfig): Plan.Closed =
+  *     FileSource(cfg.in, "parquet") ~> MyTransform() ~> FileSink(cfg.out, "parquet")
+  * }
+  * }}}
+  */
+abstract class Job[C <: Product: ClassTag] {
+
+  def plan(cfg: C): Plan.Closed
+
+  /** Hook for subclasses to tweak the SparkSession builder (master, configs). */
+  protected def configure(builder: SparkSession.Builder): SparkSession.Builder = builder
+
+  /** Bind `args` onto Cfg via Jackson and return the resulting Plan. Used by runners that already manage Spark. */
+  def planFromMap(args: Map[String, Any]): Plan.Closed = {
+    val cls = implicitly[ClassTag[C]].runtimeClass.asInstanceOf[Class[C]]
+    val cfg = Job.mapper.convertValue(args, cls)
+    plan(cfg)
+  }
+
+  def main(argv: Array[String]): Unit = {
+    val args = Job.parseArgv(argv)
+    println(s"Running ${getClass.getSimpleName}: $args")
+
+    val spark = configure(
+      SparkSession.builder().appName(getClass.getCanonicalName.stripSuffix("$"))
+    ).getOrCreate()
+
+    try planFromMap(args).run()(spark)
+    finally {
+      println("Stopping Spark session...")
+      spark.stop()
+    }
+  }
+}
+
+object Job {
+
+  // Strict mapper: missing required primitives must throw rather than silently
+  // become 0/false. Unknown keys are tolerated so `--extra=...` doesn't fail.
+  @transient private[pipeline] lazy val mapper: ObjectMapper with ClassTagExtensions = {
+    val m = new ObjectMapper() with ClassTagExtensions
+    m.registerModule(DefaultScalaModule)
+    m.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+    m.configure(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES, true)
+    m
+  }
+
+  private def parseArgv(argv: Array[String]): Map[String, Any] =
+    argv.iterator.flatMap { s =>
+      if (!s.startsWith("--")) None
+      else
+        s.drop(2).split("=", 2) match {
+          case Array(k, v) => Some(k -> v)
+          case _           => None
+        }
+    }.toMap
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Plan.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Plan.scala
new file mode 100644
index 000000000..b60ba166c
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Plan.scala
@@ -0,0 +1,132 @@
+package com.kakao.actionbase.pipeline.dsl
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+
+import scala.collection.mutable
+
+// Type-state wrapper around the internal Ast.
+//   Open   = chain ends in a Source or Transform; not yet runnable.
+//   Closed = chain ends in a Sink (or Fork of Sinks); runnable.
+sealed trait Plan {
+  private[dsl] def ast: Ast
+}
+
+object Plan {
+
+  final class Open private[dsl] (
+      private[dsl] val ast: Ast,
+      private[dsl] val edgeLabel: String = "_0"
+  ) extends Plan {
+
+    /** Name this step's output for the consumer's view of it (e.g., the temp view name a `SqlTransform` will see).
+      * Defaults to `"_0"` if not called.
+      */
+    def as(label: String): Open = new Open(ast, label)
+
+    def ~>(t: Transform): Open = new Open(Ast.Tx(Seq(edgeLabel -> ast), t))
+
+    def ~>(s: Sink): Closed = new Closed(Ast.Snk(ast, s))
+
+    /** Combine with another labeled output to feed a multi-input Transform:
+      *
+      * `(users.as("u") + events.as("e")) ~> SqlTransform("... FROM u JOIN e ...")`.
+      */
+    def +(other: Open): MultiOpen =
+      new MultiOpen(Seq(edgeLabel -> ast, other.edgeLabel -> other.ast))
+
+    def fanOut(branches: (Open => Closed)*): Closed = {
+      require(branches.nonEmpty, "fanOut requires at least one branch")
+      val branchAsts = branches.map { b =>
+        val branchOpen = new Open(Ast.Ref)
+        b(branchOpen).ast
+      }
+      new Closed(Ast.Fork(ast, branchAsts))
+    }
+  }
+
+  /** Two or more labeled outputs combined for a multi-input Transform. Build via `Open.+`; finish with `~>`. */
+  final class MultiOpen private[dsl] (private[dsl] val parts: Seq[(String, Ast)]) {
+    def +(other: Open): MultiOpen = new MultiOpen(parts :+ (other.edgeLabel -> other.ast))
+    def ~>(t: Transform): Open    = new Open(Ast.Tx(parts, t))
+  }
+
+  final class Closed private[dsl] (private[dsl] val ast: Ast) extends Plan {
+    def run()(implicit spark: SparkSession): Unit = Executor.run(ast)
+  }
+
+  /** Escape hatch for runners that build the AST directly (e.g., `StepsBuilder` assembling a DAG from YAML). */
+  private[pipeline] def closed(ast: Ast): Closed = new Closed(ast)
+  private[pipeline] def open(ast: Ast): Open     = new Open(ast)
+}
+
+// Internal AST. Hidden from users so the DSL surface stays small.
+private[pipeline] sealed trait Ast
+private[pipeline] object Ast {
+  case class Src(s: Source)                                  extends Ast
+  case class Tx(upstreams: Seq[(String, Ast)], t: Transform) extends Ast
+  case class Snk(upstream: Ast, s: Sink)                     extends Ast
+  case class Fork(upstream: Ast, branches: Seq[Ast])         extends Ast
+  // Multiple terminal roots executed under a shared materialization memo, so a common upstream is built once even
+  // when several sinks consume it.
+  case class Group(roots: Seq[Ast]) extends Ast
+  case object Ref                   extends Ast
+}
+
+private[dsl] object Executor {
+
+  def run(ast: Ast)(implicit spark: SparkSession): Unit = {
+    val memo = mutable.Map.empty[Ast, DataFrame]
+    runRoot(ast, memo)
+  }
+
+  private def runRoot(ast: Ast, memo: mutable.Map[Ast, DataFrame])(implicit
+      spark: SparkSession
+  ): Unit = ast match {
+    case Ast.Snk(up, sink) =>
+      sink.write(materialize(up, memo))
+
+    case Ast.Fork(up, branches) =>
+      val df = materialize(up, memo).cache()
+      try branches.foreach(b => runBranch(b, df))
+      finally df.unpersist(blocking = false)
+
+    case Ast.Group(roots) =>
+      roots.foreach(r => runRoot(r, memo))
+
+    case other =>
+      throw new IllegalStateException(
+        s"Top-level Plan must end in a Sink, Fork, or Group, got: $other"
+      )
+  }
+
+  private def materialize(ast: Ast, memo: mutable.Map[Ast, DataFrame])(implicit
+      spark: SparkSession
+  ): DataFrame = memo.getOrElseUpdate(
+    ast,
+    ast match {
+      case Ast.Src(s) => s.read()
+      case Ast.Tx(ups, t) =>
+        t.apply(ups.map { case (label, up) => label -> materialize(up, memo) })
+      case Ast.Ref =>
+        throw new IllegalStateException("Ast.Ref outside fanOut branch — invalid Plan")
+      case other =>
+        throw new IllegalStateException(s"materialize cannot handle: $other")
+    }
+  )
+
+  private def runBranch(ast: Ast, root: DataFrame)(implicit spark: SparkSession): Unit = {
+    def go(p: Ast): DataFrame = p match {
+      case Ast.Ref        => root
+      case Ast.Tx(ups, t) => t.apply(ups.map { case (label, up) => label -> go(up) })
+      case other =>
+        throw new IllegalStateException(
+          s"fanOut branch must be Transform*-then-Sink rooted at the fork, got: $other"
+        )
+    }
+    ast match {
+      case Ast.Snk(up, sink) => sink.write(go(up))
+      case other =>
+        throw new IllegalStateException(s"fanOut branch must end in Sink, got: $other")
+    }
+  }
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Step.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Step.scala
new file mode 100644
index 000000000..8f870eba3
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/Step.scala
@@ -0,0 +1,27 @@
+package com.kakao.actionbase.pipeline.dsl
+
+import org.apache.spark.sql.{DataFrame, SparkSession}
+
+sealed trait Step
+
+trait Source extends Step {
+  def read()(implicit spark: SparkSession): DataFrame
+}
+
+/** A Transform consumes one or more labeled input DataFrames and produces one. Each input arrives as a `(label, df)`
+  * pair so consumers (e.g., `SqlTransform`) can use the label as a temp view name without re-stating it. The label is
+  * the producer's `as:` (or `"_0"` for a single-input chain default).
+  */
+trait Transform extends Step {
+  def apply(inputs: Seq[(String, DataFrame)])(implicit spark: SparkSession): DataFrame
+
+  /** Single-input convenience for DSL `~>` and unit tests. The implicit label is `"_0"`. */
+  def apply(in: DataFrame)(implicit spark: SparkSession): DataFrame = apply(Seq("_0" -> in))
+}
+
+/** A Sink consumes a single DataFrame. Multi-input fan-in isn't a Sink concern — wire multiple sinks as siblings via
+  * `fanOut(...)` (Scala DSL) or repeat the sink step under different `inputs:` (YAML).
+  */
+trait Sink extends Step {
+  def write(in: DataFrame)(implicit spark: SparkSession): Unit
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/package.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/package.scala
new file mode 100644
index 000000000..6ede19d1a
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/dsl/package.scala
@@ -0,0 +1,10 @@
+package com.kakao.actionbase.pipeline
+
+import scala.language.implicitConversions
+
+package object dsl {
+  // Lets a Source start a chain: `MySource(...) ~> MyTransform() ~> MySink(...)`.
+  // `~>` lives on Plan.Open, so this conversion bridges the gap.
+  implicit def sourceToOpen(s: Source): Plan.Open =
+    new Plan.Open(Ast.Src(s))
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/jobs/StepsRunnerJob.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/jobs/StepsRunnerJob.scala
new file mode 100644
index 000000000..1cea02383
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/jobs/StepsRunnerJob.scala
@@ -0,0 +1,29 @@
+package com.kakao.actionbase.pipeline.jobs
+
+import com.kakao.actionbase.pipeline.dsl.{Job, Plan}
+import com.kakao.actionbase.pipeline.runner.StepsBuilder
+import com.kakao.actionbase.pipeline.workflow.StepSpec
+
+/** Generic Job whose Plan is built at runtime from an inline `steps:` list in its Cfg. Lets a workflow YAML express a
+  * Source ~> Transform* ~> Sink chain without requiring a hand-written Job class:
+  *
+  * {{{
+  * jobs:
+  *   pi:
+  *     kind: spark
+  *     artifact: "com.kakao.actionbase:pipeline:0.x"
+  *     mainClass: StepsRunnerJob
+  *     args:
+  *       steps:
+  *         - step: SampleSource
+  *           args: { n: 1000000, columns: [x, y] }
+  *         - step: SqlTransform
+  *           args: { query: "SELECT ... FROM _0" }
+  *         - step: ShowSink
+  * }}}
+  */
+case class StepsRunnerCfg(steps: Seq[StepSpec])
+
+object StepsRunnerJob extends Job[StepsRunnerCfg] {
+  override def plan(cfg: StepsRunnerCfg): Plan.Closed = StepsBuilder.build(cfg.steps)
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/ClassResolver.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/ClassResolver.scala
new file mode 100644
index 000000000..3bec224fe
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/ClassResolver.scala
@@ -0,0 +1,32 @@
+package com.kakao.actionbase.pipeline.runner
+
+/** Resolves a Job or Step class name from a workflow YAML. The name may be:
+  *   - a full FQN (`com.kakao.actionbase.pipeline.jobs.SparkPiJob`),
+  *   - a short name (`SparkPiJob`), resolved relative to one of `roots`,
+  *   - or a sub-package name (`subdir.MySource`), also resolved relative to a root.
+  *
+  * Roots are tried in order; first match wins.
+  */
+object ClassResolver {
+
+  val JobRoots: Seq[String] = Seq("com.kakao.actionbase.pipeline.jobs")
+
+  val StepRoots: Seq[String] = Seq(
+    "com.kakao.actionbase.pipeline.steps.source",
+    "com.kakao.actionbase.pipeline.steps.transform",
+    "com.kakao.actionbase.pipeline.steps.sink"
+  )
+
+  def resolve(name: String, roots: Seq[String]): Class[_] = {
+    val candidates = name +: roots.map(r => s"$r.$name")
+    candidates.view.flatMap(tryLoad).headOption.getOrElse {
+      throw new ClassNotFoundException(
+        s"Cannot resolve '$name' as a full FQN or under any of: ${roots.mkString(", ")}"
+      )
+    }
+  }
+
+  private def tryLoad(fqn: String): Option[Class[_]] =
+    try Some(Class.forName(fqn))
+    catch { case _: ClassNotFoundException => None }
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/StepsBuilder.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/StepsBuilder.scala
new file mode 100644
index 000000000..4fdd26d7a
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/StepsBuilder.scala
@@ -0,0 +1,126 @@
+package com.kakao.actionbase.pipeline.runner
+
+import com.kakao.actionbase.pipeline.dsl._
+import com.kakao.actionbase.pipeline.workflow.StepSpec
+
+import scala.collection.mutable
+
+/** Reflectively assembles a runnable `Plan.Closed` from a list of `StepSpec`s. Each spec's `args:` map is bound onto
+  * the Step's case-class fields via Jackson; the resulting AST is wired together as a DAG.
+  *
+  * Wiring rules:
+  *   - Source steps stand alone (no upstream).
+  *   - A step's upstream inputs are resolved from `inputs:` labels first; if absent, the previous step's output is used
+  *     (linear-chain default).
+  *   - `as: <label>` registers the step's output for later `inputs:` references.
+  *   - The terminal step must be a Sink; the result is a `Plan.Closed`.
+  *
+  * Expression resolution (`${{ ... }}`) is the caller's responsibility — `args:` values are bound as-is.
+  */
+object StepsBuilder {
+
+  def build(steps: Seq[StepSpec]): Plan.Closed = {
+    require(steps.nonEmpty, "`steps` must be non-empty")
+
+    val byLabel   = mutable.LinkedHashMap.empty[String, Ast]
+    val sinks     = mutable.ListBuffer.empty[Ast.Snk]
+    var prev: Ast = null // most recent step's AST (linear-chain default upstream)
+
+    steps.foreach { spec =>
+      val instance = instantiate(spec)
+      val ast: Ast = instance match {
+        case s: Source =>
+          require(spec.inputs.isEmpty, s"Source `${spec.step}` cannot declare `inputs:`")
+          Ast.Src(s)
+
+        case t: Transform =>
+          Ast.Tx(resolveUpstreams(spec, byLabel, prev), t)
+
+        case k: Sink =>
+          require(
+            spec.inputs.size <= 1,
+            s"Sink `${spec.step}` is single-input; `inputs:` must have at most one entry, got ${spec.inputs.size}"
+          )
+          val (_, up) = resolveUpstreams(spec, byLabel, prev).head
+          val snk     = Ast.Snk(up, k)
+          sinks += snk
+          snk
+
+        case other =>
+          throw new IllegalArgumentException(
+            s"${spec.step} is not a Step (must extend Source, Transform, or Sink); got ${other.getClass.getName}"
+          )
+      }
+
+      spec.as.foreach { label =>
+        if (byLabel.contains(label)) {
+          throw new IllegalArgumentException(s"duplicate step label `as: $label`")
+        }
+        byLabel(label) = ast
+      }
+      prev = ast
+    }
+
+    sinks.size match {
+      case 0 => throw new IllegalArgumentException("workflow must have at least one Sink")
+      case 1 => Plan.closed(sinks.head)
+      case _ => Plan.closed(Ast.Group(sinks.toSeq))
+    }
+  }
+
+  private def resolveUpstreams(
+      spec: StepSpec,
+      byLabel: collection.Map[String, Ast],
+      prev: Ast
+  ): Seq[(String, Ast)] = {
+    val ups: Seq[(String, Ast)] =
+      if (spec.inputs.nonEmpty) {
+        spec.inputs.map { label =>
+          val up = byLabel.getOrElse(
+            label,
+            throw new IllegalArgumentException(
+              s"step `${spec.step}` references unknown input `$label`; " +
+                s"known labels: ${if (byLabel.isEmpty) "<none>" else byLabel.keys.mkString(", ")}"
+            )
+          )
+          label -> up
+        }
+      } else {
+        if (prev == null) {
+          throw new IllegalArgumentException(
+            s"step `${spec.step}` has no upstream — the first step must be a Source, " +
+              "or specify `inputs:` referencing labeled upstream(s)"
+          )
+        }
+        Seq("_0" -> prev)
+      }
+
+    ups.foreach {
+      case (_, _: Ast.Src) | (_, _: Ast.Tx) => // ok — produces a DataFrame
+      case (_, other) =>
+        throw new IllegalArgumentException(
+          s"step `${spec.step}` upstream must be a Source or Transform, got ${describe(other)}"
+        )
+    }
+    ups
+  }
+
+  private def instantiate(spec: StepSpec): Step = {
+    val cls = ClassResolver.resolve(spec.step, ClassResolver.StepRoots)
+    val obj = Job.mapper.convertValue(spec.args, cls)
+    obj match {
+      case step: Step => step
+      case _ =>
+        throw new IllegalArgumentException(
+          s"${spec.step} is not a Step (must extend Source, Transform, or Sink)"
+        )
+    }
+  }
+
+  private def describe(ast: Ast): String = ast match {
+    case Ast.Src(s)    => s"Source(${s.getClass.getSimpleName})"
+    case Ast.Tx(_, t)  => s"Transform(${t.getClass.getSimpleName})"
+    case Ast.Snk(_, s) => s"Sink(${s.getClass.getSimpleName})"
+    case other         => other.toString
+  }
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/sink/FileSink.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/sink/FileSink.scala
new file mode 100644
index 000000000..2e54cbe29
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/sink/FileSink.scala
@@ -0,0 +1,17 @@
+package com.kakao.actionbase.pipeline.steps.sink
+
+import com.kakao.actionbase.pipeline.dsl.Sink
+import org.apache.spark.sql.{DataFrame, SparkSession}
+
+/** Writes a DataFrame to file storage using Spark's `DataFrameWriter`. `format` is any Spark-supported short name
+  * (`parquet`, `json`, `csv`, `orc`, ...); `options` are passed through to `DataFrameWriter.options`.
+  */
+case class FileSink(
+    path: String,
+    format: String,
+    mode: String = "overwrite",
+    options: Map[String, String] = Map.empty
+) extends Sink {
+  override def write(in: DataFrame)(implicit spark: SparkSession): Unit =
+    in.write.format(format).mode(mode).options(options).save(path)
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/sink/ShowSink.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/sink/ShowSink.scala
new file mode 100644
index 000000000..7f312136f
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/sink/ShowSink.scala
@@ -0,0 +1,10 @@
+package com.kakao.actionbase.pipeline.steps.sink
+
+import com.kakao.actionbase.pipeline.dsl.Sink
+import org.apache.spark.sql.{DataFrame, SparkSession}
+
+/** Prints rows to stdout via Spark's `DataFrame.show`. Useful as a debug/demo Sink. */
+case class ShowSink(numRows: Int = 20, truncate: Boolean = true) extends Sink {
+  override def write(in: DataFrame)(implicit spark: SparkSession): Unit =
+    in.show(numRows, truncate)
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/source/FileSource.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/source/FileSource.scala
new file mode 100644
index 000000000..574995e72
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/source/FileSource.scala
@@ -0,0 +1,16 @@
+package com.kakao.actionbase.pipeline.steps.source
+
+import com.kakao.actionbase.pipeline.dsl.Source
+import org.apache.spark.sql.{DataFrame, SparkSession}
+
+/** Reads a file-based dataset using Spark's `DataFrameReader`. `format` is any Spark-supported short name (`json`,
+  * `parquet`, `csv`, `orc`, ...); `options` are passed through to `DataFrameReader.options`.
+  */
+case class FileSource(
+    path: String,
+    format: String,
+    options: Map[String, String] = Map.empty
+) extends Source {
+  override def read()(implicit spark: SparkSession): DataFrame =
+    spark.read.format(format).options(options).load(path)
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/source/SampleSource.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/source/SampleSource.scala
new file mode 100644
index 000000000..1f9d01afa
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/source/SampleSource.scala
@@ -0,0 +1,13 @@
+package com.kakao.actionbase.pipeline.steps.source
+
+import com.kakao.actionbase.pipeline.dsl.Source
+import org.apache.spark.sql.functions.rand
+import org.apache.spark.sql.{DataFrame, SparkSession}
+
+/** Produces `n` rows of random values uniformly drawn from `[0.0, 1.0)`, one column per name in `columns`. */
+case class SampleSource(n: Long, columns: Seq[String]) extends Source {
+  require(columns.nonEmpty, "SampleSource requires at least one column")
+
+  override def read()(implicit spark: SparkSession): DataFrame =
+    spark.range(n).select(columns.map(name => rand().as(name)): _*)
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/transform/CacheTransform.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/transform/CacheTransform.scala
new file mode 100644
index 000000000..4c273a612
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/transform/CacheTransform.scala
@@ -0,0 +1,18 @@
+package com.kakao.actionbase.pipeline.steps.transform
+
+import com.kakao.actionbase.pipeline.dsl.Transform
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.storage.StorageLevel
+
+/** Materializes its input once and reuses it across downstream consumers. Useful when a single upstream feeds two or
+  * more transforms (e.g., a "split" expressed as two filter transforms over the same source) — without it, Spark
+  * re-executes the upstream lineage per consumer.
+  *
+  * `level` is any string accepted by `StorageLevel.fromString` (default `MEMORY_AND_DISK`).
+  */
+case class CacheTransform(level: String = "MEMORY_AND_DISK") extends Transform {
+  override def apply(inputs: Seq[(String, DataFrame)])(implicit spark: SparkSession): DataFrame = {
+    require(inputs.size == 1, s"CacheTransform expects 1 input, got ${inputs.size}")
+    inputs.head._2.persist(StorageLevel.fromString(level))
+  }
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/transform/SqlTransform.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/transform/SqlTransform.scala
new file mode 100644
index 000000000..ca4366c3b
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/steps/transform/SqlTransform.scala
@@ -0,0 +1,18 @@
+package com.kakao.actionbase.pipeline.steps.transform
+
+import com.kakao.actionbase.pipeline.dsl.Transform
+import org.apache.spark.sql.{DataFrame, SparkSession}
+
+/** Runs `query` against the labeled input DataFrames. Each input is registered as a temp view named after its label —
+  * the producer's `as:` (or `"_0"` for the chain default). No separate `views:` parameter — the SQL refers to inputs by
+  * the same names that wired them in.
+  *
+  * Single-input chain: `SqlTransform("SELECT ... FROM _0")`. Join: pair two labeled FileSources with `inputs: [users,
+  * events]`, then `SqlTransform("SELECT ... FROM users JOIN events USING (id)")`.
+  */
+case class SqlTransform(query: String) extends Transform {
+  override def apply(inputs: Seq[(String, DataFrame)])(implicit spark: SparkSession): DataFrame = {
+    inputs.foreach { case (name, df) => df.createOrReplaceTempView(name) }
+    spark.sql(query)
+  }
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/StepSpec.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/StepSpec.scala
new file mode 100644
index 000000000..89b80b552
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/StepSpec.scala
@@ -0,0 +1,17 @@
+package com.kakao.actionbase.pipeline.workflow
+
+/** A Step entry inside a Cfg whose Job interprets a `steps:` list (e.g., `StepsRunnerJob`).
+  *
+  *   - `step` — Step class name (Source / Transform / Sink); short, sub-package, and FQN forms all resolve via
+  *     `ClassResolver.StepRoots`.
+  *   - `args` — constructor args bound to the Step's case-class fields.
+  *   - `as` — optional label that names this step's output for downstream reference.
+  *   - `inputs` — labels of upstream steps to feed in. Empty = the previous step's output (linear-chain default, label
+  *     `"_0"`). Multiple = join semantics (e.g., `inputs: [users, events]` for a `SqlTransform` with two views).
+  */
+case class StepSpec(
+    step: String,
+    args: Map[String, Any] = Map.empty,
+    as: Option[String] = None,
+    inputs: Seq[String] = Seq.empty
+)
diff --git a/pipeline/src/test/scala/com/kakao/actionbase/pipeline/SparkTest.scala b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/SparkTest.scala
new file mode 100644
index 000000000..8c87438c6
--- /dev/null
+++ b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/SparkTest.scala
@@ -0,0 +1,16 @@
+package com.kakao.actionbase.pipeline
+
+import org.apache.spark.sql.SparkSession
+
+/** Mix-in for unit tests that need an `implicit SparkSession`. The session is `local[2]` and shared across all
+  * `SparkTest` mixers in the JVM (Spark's own `getOrCreate` semantics).
+  */
+trait SparkTest {
+  protected implicit lazy val spark: SparkSession = SparkSession
+    .builder()
+    .master("local[2]")
+    .config("spark.driver.bindAddress", "127.0.0.1")
+    .config("spark.ui.enabled", "false")
+    .appName(getClass.getSimpleName)
+    .getOrCreate()
+}
diff --git a/pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/StepsBuilderTest.scala b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/StepsBuilderTest.scala
new file mode 100644
index 000000000..1ef217217
--- /dev/null
+++ b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/StepsBuilderTest.scala
@@ -0,0 +1,141 @@
+package com.kakao.actionbase.pipeline.runner
+
+import com.kakao.actionbase.pipeline.SparkTest
+import com.kakao.actionbase.pipeline.workflow.StepSpec
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+
+class StepsBuilderTest extends SparkTest {
+
+  private val ShowSinkFqn     = "com.kakao.actionbase.pipeline.steps.sink.ShowSink"
+  private val SampleSourceFqn = "com.kakao.actionbase.pipeline.steps.source.SampleSource"
+  private val SqlTransformFqn = "com.kakao.actionbase.pipeline.steps.transform.SqlTransform"
+
+  @Test
+  def buildsAndRunsSourceSinkChain(): Unit = {
+    val steps = Seq(
+      StepSpec(SampleSourceFqn, Map("n" -> 50L, "columns" -> Seq("a", "b"))),
+      StepSpec(ShowSinkFqn)
+    )
+
+    StepsBuilder.build(steps).run()
+  }
+
+  @Test
+  def buildsAndRunsSourceTransformSinkChain(): Unit = {
+    val steps = Seq(
+      StepSpec(SampleSourceFqn, Map("n" -> 20L, "columns" -> Seq("x"))),
+      StepSpec(SqlTransformFqn, Map("query" -> "SELECT x * 2 AS y FROM _0")),
+      StepSpec(ShowSinkFqn)
+    )
+
+    StepsBuilder.build(steps).run()
+  }
+
+  @Test
+  def buildsAndRunsJoinAcrossLabeledSources(): Unit = {
+    val steps = Seq(
+      StepSpec(
+        SampleSourceFqn,
+        args = Map("n" -> 5L, "columns" -> Seq("u")),
+        as = Some("users")
+      ),
+      StepSpec(
+        SampleSourceFqn,
+        args = Map("n" -> 5L, "columns" -> Seq("e")),
+        as = Some("events")
+      ),
+      StepSpec(
+        SqlTransformFqn,
+        args = Map("query" -> "SELECT u.u, e.e FROM users u CROSS JOIN events e"),
+        inputs = Seq("users", "events")
+      ),
+      StepSpec(ShowSinkFqn)
+    )
+
+    StepsBuilder.build(steps).run()
+  }
+
+  @Test
+  def rejectsSinkWithoutUpstream(): Unit = {
+    val ex = assertThrows(
+      classOf[IllegalArgumentException],
+      () => StepsBuilder.build(Seq(StepSpec(ShowSinkFqn)))
+    )
+    assertTrue(ex.getMessage.contains("no upstream"), ex.getMessage)
+  }
+
+  @Test
+  def rejectsWorkflowWithNoSink(): Unit = {
+    val steps = Seq(StepSpec(SampleSourceFqn, Map("n" -> 5L, "columns" -> Seq("a"))))
+    val ex = assertThrows(
+      classOf[IllegalArgumentException],
+      () => StepsBuilder.build(steps)
+    )
+    assertTrue(ex.getMessage.contains("at least one Sink"), ex.getMessage)
+  }
+
+  @Test
+  def runsMultipleSinksOverSharedUpstream(): Unit = {
+    val FileSinkFqn = "com.kakao.actionbase.pipeline.steps.sink.FileSink"
+    val tmpA        = java.nio.file.Files.createTempDirectory("multi-sink-a-").resolve("out").toString
+    val tmpB        = java.nio.file.Files.createTempDirectory("multi-sink-b-").resolve("out").toString
+
+    val steps = Seq(
+      StepSpec(SampleSourceFqn, Map("n" -> 5L, "columns" -> Seq("x")), as = Some("data")),
+      StepSpec(FileSinkFqn, Map("path" -> tmpA, "format" -> "parquet"), inputs = Seq("data")),
+      StepSpec(FileSinkFqn, Map("path" -> tmpB, "format" -> "parquet"), inputs = Seq("data"))
+    )
+
+    StepsBuilder.build(steps).run()
+
+    assertTrue(java.nio.file.Files.exists(java.nio.file.Paths.get(tmpA)), s"sink A should write: $tmpA")
+    assertTrue(java.nio.file.Files.exists(java.nio.file.Paths.get(tmpB)), s"sink B should write: $tmpB")
+  }
+
+  @Test
+  def runsIndependentSinkSubTrees(): Unit = {
+    val FileSinkFqn = "com.kakao.actionbase.pipeline.steps.sink.FileSink"
+    val tmpA        = java.nio.file.Files.createTempDirectory("indep-a-").resolve("out").toString
+    val tmpB        = java.nio.file.Files.createTempDirectory("indep-b-").resolve("out").toString
+
+    val steps = Seq(
+      StepSpec(SampleSourceFqn, Map("n" -> 3L, "columns" -> Seq("a"))),
+      StepSpec(FileSinkFqn, Map("path" -> tmpA, "format" -> "parquet")),
+      StepSpec(SampleSourceFqn, Map("n" -> 3L, "columns" -> Seq("b"))),
+      StepSpec(FileSinkFqn, Map("path" -> tmpB, "format" -> "parquet"))
+    )
+
+    StepsBuilder.build(steps).run()
+
+    assertTrue(java.nio.file.Files.exists(java.nio.file.Paths.get(tmpA)), s"first subtree should write: $tmpA")
+    assertTrue(java.nio.file.Files.exists(java.nio.file.Paths.get(tmpB)), s"second subtree should write: $tmpB")
+  }
+
+  @Test
+  def rejectsUnknownInputLabel(): Unit = {
+    val steps = Seq(
+      StepSpec(SampleSourceFqn, Map("n" -> 5L, "columns" -> Seq("a"))),
+      StepSpec(ShowSinkFqn, inputs = Seq("ghost"))
+    )
+    val ex = assertThrows(
+      classOf[IllegalArgumentException],
+      () => StepsBuilder.build(steps)
+    )
+    assertTrue(ex.getMessage.contains("ghost"), ex.getMessage)
+  }
+
+  @Test
+  def rejectsDuplicateLabel(): Unit = {
+    val steps = Seq(
+      StepSpec(SampleSourceFqn, Map("n" -> 5L, "columns" -> Seq("a")), as = Some("dup")),
+      StepSpec(SampleSourceFqn, Map("n" -> 5L, "columns" -> Seq("b")), as = Some("dup")),
+      StepSpec(ShowSinkFqn)
+    )
+    val ex = assertThrows(
+      classOf[IllegalArgumentException],
+      () => StepsBuilder.build(steps)
+    )
+    assertTrue(ex.getMessage.contains("duplicate"), ex.getMessage)
+  }
+}
diff --git a/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/sink/FileSinkTest.scala b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/sink/FileSinkTest.scala
new file mode 100644
index 000000000..e2f240cbe
--- /dev/null
+++ b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/sink/FileSinkTest.scala
@@ -0,0 +1,49 @@
+package com.kakao.actionbase.pipeline.steps.sink
+
+import com.kakao.actionbase.pipeline.SparkTest
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+
+import java.nio.file.{Files, Paths}
+
+class FileSinkTest extends SparkTest {
+
+  @Test
+  def writesParquet(): Unit = {
+    import spark.implicits._
+    val df   = Seq(("a", 1), ("b", 2)).toDF("k", "v")
+    val tmp  = Files.createTempDirectory("file-sink-parquet-")
+    val path = tmp.resolve("out").toString
+
+    FileSink(path, "parquet").write(df)
+
+    assertTrue(Files.exists(Paths.get(path)))
+    assertEquals(2L, spark.read.parquet(path).count())
+  }
+
+  @Test
+  def writesJson(): Unit = {
+    import spark.implicits._
+    val df   = Seq(1, 2, 3).toDF("v")
+    val tmp  = Files.createTempDirectory("file-sink-json-")
+    val path = tmp.resolve("out").toString
+
+    FileSink(path, "json").write(df)
+
+    assertEquals(3L, spark.read.json(path).count())
+  }
+
+  @Test
+  def overwriteReplacesExisting(): Unit = {
+    import spark.implicits._
+    val df1  = Seq(1, 2).toDF("v")
+    val df2  = Seq(10, 20, 30).toDF("v")
+    val tmp  = Files.createTempDirectory("file-sink-mode-")
+    val path = tmp.resolve("out").toString
+
+    FileSink(path, "parquet").write(df1)
+    FileSink(path, "parquet").write(df2)
+
+    assertEquals(3L, spark.read.parquet(path).count())
+  }
+}
diff --git a/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/sink/ShowSinkTest.scala b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/sink/ShowSinkTest.scala
new file mode 100644
index 000000000..a27d427aa
--- /dev/null
+++ b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/sink/ShowSinkTest.scala
@@ -0,0 +1,16 @@
+package com.kakao.actionbase.pipeline.steps.sink
+
+import com.kakao.actionbase.pipeline.SparkTest
+import org.junit.jupiter.api.Test
+
+class ShowSinkTest extends SparkTest {
+
+  @Test
+  def writesDataframeWithoutError(): Unit = {
+    import spark.implicits._
+    val in = Seq((1, "a"), (2, "b")).toDF("n", "name")
+
+    ShowSink().write(in)
+    ShowSink(numRows = 1, truncate = false).write(in)
+  }
+}
diff --git a/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/source/FileSourceTest.scala b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/source/FileSourceTest.scala
new file mode 100644
index 000000000..86bcc1d6a
--- /dev/null
+++ b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/source/FileSourceTest.scala
@@ -0,0 +1,58 @@
+package com.kakao.actionbase.pipeline.steps.source
+
+import com.kakao.actionbase.pipeline.SparkTest
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Paths}
+
+class FileSourceTest extends SparkTest {
+
+  @Test
+  def readsJson(): Unit = {
+    val tmp  = Files.createTempDirectory("file-source-json-")
+    val path = tmp.resolve("data.json").toString
+    Files.write(
+      Paths.get(path),
+      "{\"a\":1}\n{\"a\":2}\n".getBytes(StandardCharsets.UTF_8)
+    )
+
+    val df = FileSource(path, "json").read()
+
+    assertEquals(2L, df.count())
+    assertTrue(df.columns.contains("a"))
+  }
+
+  @Test
+  def readsParquet(): Unit = {
+    val tmp = Files.createTempDirectory("file-source-parquet-")
+    val src = tmp.resolve("src.json").toString
+    Files.write(
+      Paths.get(src),
+      "{\"a\":1}\n{\"a\":2}\n{\"a\":3}\n".getBytes(StandardCharsets.UTF_8)
+    )
+    val parquet = tmp.resolve("out.parquet").toString
+    spark.read.json(src).write.parquet(parquet)
+
+    val df = FileSource(parquet, "parquet").read()
+
+    assertEquals(3L, df.count())
+  }
+
+  @Test
+  def passesOptionsToReader(): Unit = {
+    val tmp  = Files.createTempDirectory("file-source-csv-")
+    val path = tmp.resolve("data.csv").toString
+    Files.write(
+      Paths.get(path),
+      "name,age\nalice,30\nbob,25\n".getBytes(StandardCharsets.UTF_8)
+    )
+
+    val df = FileSource(path, "csv", Map("header" -> "true")).read()
+
+    assertEquals(2L, df.count())
+    assertTrue(df.columns.toSet.contains("name"))
+    assertTrue(df.columns.toSet.contains("age"))
+  }
+}
diff --git a/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/source/SampleSourceTest.scala b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/source/SampleSourceTest.scala
new file mode 100644
index 000000000..a9f6fe699
--- /dev/null
+++ b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/source/SampleSourceTest.scala
@@ -0,0 +1,38 @@
+package com.kakao.actionbase.pipeline.steps.source
+
+import com.kakao.actionbase.pipeline.SparkTest
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+
+class SampleSourceTest extends SparkTest {
+
+  @Test
+  def producesRowsForRequestedColumns(): Unit = {
+    val df = SampleSource(1000L, Seq("x", "y")).read()
+    assertEquals(1000L, df.count())
+    assertEquals(Set("x", "y"), df.columns.toSet)
+  }
+
+  @Test
+  def supportsArbitraryColumnSet(): Unit = {
+    val df = SampleSource(50L, Seq("a", "b", "c", "d")).read()
+    assertEquals(50L, df.count())
+    assertEquals(Set("a", "b", "c", "d"), df.columns.toSet)
+  }
+
+  @Test
+  def valuesAreInUnitInterval(): Unit = {
+    val df   = SampleSource(200L, Seq("x", "y")).read()
+    val rows = df.collect()
+    assertTrue(rows.forall { r =>
+      val x = r.getAs[Double]("x")
+      val y = r.getAs[Double]("y")
+      x >= 0.0 && x < 1.0 && y >= 0.0 && y < 1.0
+    })
+  }
+
+  @Test
+  def rejectsEmptyColumnList(): Unit = {
+    assertThrows(classOf[IllegalArgumentException], () => SampleSource(10L, Seq.empty))
+  }
+}
diff --git a/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/transform/CacheTransformTest.scala b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/transform/CacheTransformTest.scala
new file mode 100644
index 000000000..f501cceff
--- /dev/null
+++ b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/transform/CacheTransformTest.scala
@@ -0,0 +1,42 @@
+package com.kakao.actionbase.pipeline.steps.transform
+
+import com.kakao.actionbase.pipeline.SparkTest
+import org.apache.spark.storage.StorageLevel
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+
+class CacheTransformTest extends SparkTest {
+
+  @Test
+  def persistsAtDefaultLevel(): Unit = {
+    import spark.implicits._
+    val in = Seq(1, 2, 3).toDF("n")
+
+    val out = CacheTransform().apply(in)
+
+    assertEquals(StorageLevel.MEMORY_AND_DISK, out.storageLevel)
+    out.unpersist()
+  }
+
+  @Test
+  def respectsExplicitStorageLevel(): Unit = {
+    import spark.implicits._
+    val in = Seq(1, 2, 3).toDF("n")
+
+    val out = CacheTransform(level = "MEMORY_ONLY").apply(in)
+
+    assertEquals(StorageLevel.MEMORY_ONLY, out.storageLevel)
+    out.unpersist()
+  }
+
+  @Test
+  def passesRowsThroughUnchanged(): Unit = {
+    import spark.implicits._
+    val in = Seq(("a", 1), ("b", 2)).toDF("k", "v")
+
+    val out = CacheTransform().apply(in)
+
+    assertEquals(in.collect().toSet, out.collect().toSet)
+    out.unpersist()
+  }
+}
diff --git a/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/transform/SqlTransformTest.scala b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/transform/SqlTransformTest.scala
new file mode 100644
index 000000000..4c02ad5c3
--- /dev/null
+++ b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/steps/transform/SqlTransformTest.scala
@@ -0,0 +1,54 @@
+package com.kakao.actionbase.pipeline.steps.transform
+
+import com.kakao.actionbase.pipeline.SparkTest
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+
+class SqlTransformTest extends SparkTest {
+
+  @Test
+  def projectsAndComputesColumns(): Unit = {
+    import spark.implicits._
+    val in = Seq((1, 2), (3, 4)).toDF("a", "b")
+
+    val out = SqlTransform("SELECT a, b, a + b AS sum FROM _0").apply(in)
+
+    assertEquals(Set("a", "b", "sum"), out.columns.toSet)
+    val sums = out.collect().map(_.getAs[Int]("sum")).toSet
+    assertEquals(Set(3, 7), sums)
+  }
+
+  @Test
+  def usesProvidedLabelAsViewName(): Unit = {
+    import spark.implicits._
+    val in = Seq(1, 2, 3).toDF("n")
+
+    val out = SqlTransform("SELECT SUM(n) AS total FROM nums").apply(Seq("nums" -> in))
+
+    assertEquals(6L, out.first().getLong(0))
+  }
+
+  @Test
+  def filtersRowsViaWhere(): Unit = {
+    import spark.implicits._
+    val in = Seq(1, 2, 3, 4).toDF("n")
+
+    val out = SqlTransform("SELECT n FROM _0 WHERE n > 2").apply(in)
+
+    assertEquals(Set(3, 4), out.collect().map(_.getInt(0)).toSet)
+  }
+
+  @Test
+  def joinsTwoLabeledInputs(): Unit = {
+    import spark.implicits._
+    val users  = Seq((1, "alice"), (2, "bob")).toDF("id", "name")
+    val events = Seq((1, "login"), (2, "click"), (2, "logout")).toDF("user_id", "kind")
+
+    val out = SqlTransform(
+      "SELECT u.name, e.kind FROM users u JOIN events e ON u.id = e.user_id"
+    ).apply(Seq("users" -> users, "events" -> events))
+
+    val rows = out.collect().map(r => r.getString(0) -> r.getString(1)).toSet
+    assertEquals(Set("alice" -> "login", "bob" -> "click", "bob" -> "logout"), rows)
+  }
+}

From 3f00b8bae9b7b15180788c0794c7149084d33c6a Mon Sep 17 00:00:00 2001
From: Minseok Kim <mskim@apache.org>
Date: Sun, 10 May 2026 23:56:49 +0900
Subject: [PATCH 2/2] Add workflow YAML loader with $extends and ${{ }}
 expressions

Implements the parser layer of the workflow DSL (#310). Workflow YAML
binds onto Workflow + JobSpec case classes via Jackson, with $extends
deep-merge and ${{ env / presets / load }} expressions resolved at
load time. needs.* expressions and 'when:' operator forms are deferred
to runtime (handled by the EmbeddedRunner in a follow-up PR).

- Workflow + JobSpec (kind / artifact / mainClass / args / submit /
  run / needs / when)
- WorkflowLoader: YAML -> raw map -> resolve presets -> resolve
  \$extends -> resolve expressions (lenient on needs) -> Jackson bind
- Expression evaluator: env / presets / needs / load
- ExtendsResolver: deep merge with cycle protection
- Tests + sample YAMLs (spark-pi.yaml, extends-via-load.yaml)

Generated with [Claude Code](https://claude.ai/code)
via [Happy](https://happy.engineering)

Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Happy <yesreply@happy.engineering>
---
 .../pipeline/runner/Expression.scala          | 122 ++++++++++++++++++
 .../pipeline/runner/ExtendsResolver.scala     |  93 +++++++++++++
 .../pipeline/workflow/Workflow.scala          |  39 ++++++
 .../pipeline/workflow/WorkflowLoader.scala    | 107 +++++++++++++++
 .../resources/workflows/extends-via-load.yaml |  11 ++
 .../workflows/preset/spark-small.yaml         |   2 +
 .../test/resources/workflows/spark-pi.yaml    |  27 ++++
 .../pipeline/runner/ExpressionTest.scala      |  67 ++++++++++
 .../pipeline/runner/ExtendsResolverTest.scala |  96 ++++++++++++++
 .../workflow/WorkflowLoaderTest.scala         |  71 ++++++++++
 10 files changed, 635 insertions(+)
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/Expression.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/ExtendsResolver.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/Workflow.scala
 create mode 100644 pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/WorkflowLoader.scala
 create mode 100644 pipeline/src/test/resources/workflows/extends-via-load.yaml
 create mode 100644 pipeline/src/test/resources/workflows/preset/spark-small.yaml
 create mode 100644 pipeline/src/test/resources/workflows/spark-pi.yaml
 create mode 100644 pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/ExpressionTest.scala
 create mode 100644 pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/ExtendsResolverTest.scala
 create mode 100644 pipeline/src/test/scala/com/kakao/actionbase/pipeline/workflow/WorkflowLoaderTest.scala

diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/Expression.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/Expression.scala
new file mode 100644
index 000000000..04a99dfcc
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/Expression.scala
@@ -0,0 +1,122 @@
+package com.kakao.actionbase.pipeline.runner
+
+import scala.collection.JavaConverters._
+import scala.util.matching.Regex
+
+/** Resolves `${{ ... }}` expressions in a parsed YAML tree against a context.
+  *
+  * Vocabulary:
+  *   - `env.<key>` — workflow `env:` value (string)
+  *   - `needs.<id>.result` — upstream job result: `success` / `failure` / `skipped` / `cancelled`
+  *   - `needs.<id>.outputs.<key>` — value emitted by an upstream job
+  *   - `presets.<name>` — entry from the workflow `presets:` section (any value, often a map)
+  *   - `load('<path>')` — content of another YAML file at `<path>` relative to the workflow file's dir
+  *
+  * An expression that fills the entire string value (e.g. `"${{ presets.X }}"`) is replaced by the raw evaluated value
+  * (which may be a map / list, not just a string). Otherwise expressions are stringified and substituted within the
+  * surrounding text.
+  */
+object Expression {
+
+  /** Inputs the evaluator needs. `needs` is empty at YAML-load time and populated at runtime. `loadYaml` is used by the
+    * `load(...)` form; injected so test mocks don't have to touch the file system.
+    */
+  case class Context(
+      env: Map[String, String] = Map.empty,
+      presets: Map[String, Any] = Map.empty,
+      needs: Map[String, NeedsView] = Map.empty,
+      loadYaml: String => Any = _ => sys.error("load(...) not configured")
+  )
+
+  case class NeedsView(result: String, outputs: Map[String, String] = Map.empty)
+
+  private val Token: Regex      = """\$\{\{\s*(.+?)\s*\}\}""".r
+  private val WholeToken: Regex = """\A\s*\$\{\{\s*(.+?)\s*\}\}\s*\z""".r
+  private val LoadCall: Regex   = """\Aload\(\s*'([^']*)'\s*\)\z""".r
+
+  /** Resolve every string leaf in a YAML-shaped tree. Map / List structure preserved.
+    *
+    * `lenient = true` preserves the original `${{ ... }}` token when an expression cannot be evaluated against the
+    * current context (typically: `needs.*` at load-time, or operator forms like `needs.X.result == 'success'` inside
+    * `when:` that this evaluator does not parse). At runtime the same tree is walked again with full context to finish
+    * those.
+    */
+  def resolveDeep(value: Any, ctx: Context, lenient: Boolean = false): Any = value match {
+    case s: String =>
+      s match {
+        case WholeToken(expr) => evaluate(expr, ctx, lenient) // raw value, may be map / list
+        case other =>
+          Token.replaceAllIn(other, m => Regex.quoteReplacement(stringify(evaluate(m.group(1), ctx, lenient))))
+      }
+    case m: java.util.Map[_, _] =>
+      m.asInstanceOf[java.util.Map[String, Any]]
+        .asScala
+        .map { case (k, v) => k -> resolveDeep(v, ctx, lenient) }
+        .toMap
+        .asJava
+    case m: scala.collection.Map[_, _] =>
+      m.asInstanceOf[scala.collection.Map[String, Any]].map { case (k, v) => k -> resolveDeep(v, ctx, lenient) }.toMap
+    case l: java.util.List[_] =>
+      l.asInstanceOf[java.util.List[Any]].asScala.map(resolveDeep(_, ctx, lenient)).asJava
+    case l: scala.collection.Iterable[_] =>
+      l.map(resolveDeep(_, ctx, lenient)).toSeq
+    case other => other
+  }
+
+  /** Evaluate an expression body (without the outer `${{ }}`) and return the raw value.
+    *
+    * In lenient mode, an unresolvable expression (unknown key, unsupported syntax) returns the original `${{ ... }}`
+    * string unchanged so it can be evaluated later by a fuller context.
+    */
+  def evaluate(expr: String, ctx: Context, lenient: Boolean = false): Any = {
+    try evaluateImpl(expr, ctx)
+    catch {
+      case _: NoSuchElementException if lenient   => "${{ " + expr.trim + " }}"
+      case _: IllegalArgumentException if lenient => "${{ " + expr.trim + " }}"
+    }
+  }
+
+  private def evaluateImpl(expr: String, ctx: Context): Any = {
+    val e = expr.trim
+    if (e.startsWith("env.")) {
+      val k = e.stripPrefix("env.")
+      ctx.env.getOrElse(k, throw new NoSuchElementException(s"unknown env key: $k"))
+    } else if (e.startsWith("presets.")) {
+      val k = e.stripPrefix("presets.")
+      ctx.presets.getOrElse(k, throw new NoSuchElementException(s"unknown preset: $k"))
+    } else if (e.startsWith("needs.")) {
+      evalNeeds(e.stripPrefix("needs."), ctx)
+    } else if (e.startsWith("load(")) {
+      e match {
+        case LoadCall(path) => ctx.loadYaml(path)
+        case _              => throw new IllegalArgumentException(s"malformed load(): $e — expected `load('path')`")
+      }
+    } else {
+      throw new IllegalArgumentException(s"unknown expression: $e")
+    }
+  }
+
+  private def evalNeeds(rest: String, ctx: Context): Any = {
+    val parts = rest.split('.')
+    if (parts.length < 2) throw new IllegalArgumentException(s"malformed needs.* expression: needs.$rest")
+    val id   = parts(0)
+    val view = ctx.needs.getOrElse(id, throw new NoSuchElementException(s"unknown needs id: $id"))
+    parts(1) match {
+      case "result" => view.result
+      case "outputs" =>
+        if (parts.length != 3)
+          throw new IllegalArgumentException(s"needs.$id.outputs requires a key: needs.$id.outputs.<name>")
+        view.outputs.getOrElse(
+          parts(2),
+          throw new NoSuchElementException(s"unknown output: needs.$id.outputs.${parts(2)}")
+        )
+      case other => throw new IllegalArgumentException(s"unknown needs field: needs.$id.$other")
+    }
+  }
+
+  private def stringify(value: Any): String = value match {
+    case null      => ""
+    case s: String => s
+    case other     => other.toString
+  }
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/ExtendsResolver.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/ExtendsResolver.scala
new file mode 100644
index 000000000..c5c7f22db
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/runner/ExtendsResolver.scala
@@ -0,0 +1,93 @@
+package com.kakao.actionbase.pipeline.runner
+
+import com.kakao.actionbase.pipeline.runner.Expression.Context
+
+import scala.collection.JavaConverters._
+
+/** Resolves `$extends` directives in a parsed YAML tree.
+  *
+  * For every map containing `$extends:`:
+  *   1. Evaluate the value (must be a single `${{ ... }}` expression that resolves to a map). 2. Recursively resolve
+  *      `$extends` inside both the extended map and the surrounding sibling keys. 3. Deep-merge: the extended map
+  *      provides defaults; the sibling map's keys override.
+  *
+  * `$` keys are reserved as processor directives — only `$extends` is recognized today; unknown `$<name>` keys are left
+  * untouched (forward-compatible). Cycles are rejected via a depth limit.
+  */
+object ExtendsResolver {
+
+  private val ExtendsKey = "$extends"
+  private val MaxDepth   = 32
+  private val WholeToken = """\A\s*\$\{\{\s*(.+?)\s*\}\}\s*\z""".r
+
+  def resolve(value: Any, ctx: Context): Any = walk(value, ctx, depth = 0)
+
+  private def walk(value: Any, ctx: Context, depth: Int): Any = value match {
+    case m: java.util.Map[_, _] =>
+      walkMap(m.asInstanceOf[java.util.Map[String, Any]].asScala.toMap, ctx, depth)
+    case m: scala.collection.Map[_, _] =>
+      walkMap(m.asInstanceOf[scala.collection.Map[String, Any]].toMap, ctx, depth)
+    case l: java.util.List[_] =>
+      l.asInstanceOf[java.util.List[Any]].asScala.map(walk(_, ctx, depth)).asJava
+    case l: scala.collection.Iterable[_] =>
+      l.map(walk(_, ctx, depth)).toSeq
+    case other => other
+  }
+
+  private def walkMap(m: Map[String, Any], ctx: Context, depth: Int): java.util.Map[String, Any] = {
+    if (depth > MaxDepth) throw new IllegalStateException(s"$$extends nesting exceeds $MaxDepth (cycle?)")
+
+    m.get(ExtendsKey) match {
+      case None =>
+        m.map { case (k, v) => k -> walk(v, ctx, depth) }.toMap.asJava
+
+      case Some(expr) =>
+        val incoming = evalExtendsValue(expr, ctx)
+        val resolvedIncoming = walk(incoming, ctx, depth + 1) match {
+          case jm: java.util.Map[_, _] => jm.asInstanceOf[java.util.Map[String, Any]].asScala.toMap
+          case other =>
+            throw new IllegalArgumentException(s"$$extends must resolve to a map, got: ${other.getClass.getSimpleName}")
+        }
+        val sibling = (m - ExtendsKey).map { case (k, v) => k -> walk(v, ctx, depth) }
+        deepMerge(resolvedIncoming, sibling).asJava
+    }
+  }
+
+  private def evalExtendsValue(expr: Any, ctx: Context): Any = expr match {
+    case s: String =>
+      s.trim match {
+        case WholeToken(inner) => Expression.evaluate(inner, ctx)
+        case other =>
+          throw new IllegalArgumentException(
+            s"$$extends value must be a single `\\$${{ ... }}` expression, got: $other"
+          )
+      }
+    case other =>
+      throw new IllegalArgumentException(s"$$extends value must be a string expression, got: $other")
+  }
+
+  /** Recursive deep-merge: maps merge key-by-key (sibling wins on conflict at non-map leaves); other values are
+    * replaced wholesale by the sibling.
+    */
+  private def deepMerge(base: Map[String, Any], over: Map[String, Any]): Map[String, Any] = {
+    val keys = base.keySet ++ over.keySet
+    keys.iterator.map { k =>
+      (base.get(k), over.get(k)) match {
+        case (Some(a), Some(b)) =>
+          k -> mergeValue(a, b)
+        case (Some(a), None) => k -> a
+        case (None, Some(b)) => k -> b
+        case _               => sys.error("unreachable")
+      }
+    }.toMap
+  }
+
+  private def mergeValue(a: Any, b: Any): Any = (a, b) match {
+    case (am: java.util.Map[_, _], bm: java.util.Map[_, _]) =>
+      deepMerge(
+        am.asInstanceOf[java.util.Map[String, Any]].asScala.toMap,
+        bm.asInstanceOf[java.util.Map[String, Any]].asScala.toMap
+      ).asJava
+    case _ => b
+  }
+}
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/Workflow.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/Workflow.scala
new file mode 100644
index 000000000..09d5edcc9
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/Workflow.scala
@@ -0,0 +1,39 @@
+package com.kakao.actionbase.pipeline.workflow
+
+import com.fasterxml.jackson.annotation.JsonProperty
+
+/** Parsed workflow YAML.
+  *
+  * Job-to-job DAG. Each `jobs` entry is dispatched by its `kind:` to a runner-side handler. `presets:` is a free-form
+  * reusable map referenced from elsewhere via `${{ presets.<name> }}`. `$extends` (a `${{ ... }}` expression that
+  * resolves to a map) provides defaults via deep merge — see `ExtendsResolver`.
+  */
+case class Workflow(
+    name: String,
+    env: Map[String, String] = Map.empty,
+    presets: Map[String, Map[String, Any]] = Map.empty,
+    jobs: Map[String, JobSpec]
+)
+
+/** A single job in a workflow.
+  *
+  *   - `kind` — discriminator: `spark` | `bash`. Routes to a runner-side handler.
+  *   - `artifact` — Gradle coord `group:name:version` (spark only). The runner resolves the JAR.
+  *   - `mainClass` — class to invoke (spark only); short / sub-package / FQN forms all resolve via
+  *     `ClassResolver.JobRoots`.
+  *   - `args` — bound onto the Job's Cfg case class. Values may include `${{ ... }}` expressions.
+  *   - `submit` — passed through to `spark-submit` CLI (kebab keys → flags). Nested `conf` map → `--conf` repeats.
+  *   - `run` — shell command (bash only).
+  *   - `needs` — ids of jobs that must complete before this one runs.
+  *   - `when` — boolean expression; the job runs only when truthy. Defaults to true when absent.
+  */
+case class JobSpec(
+    kind: String,
+    artifact: Option[String] = None,
+    mainClass: Option[String] = None,
+    args: Map[String, Any] = Map.empty,
+    submit: Map[String, Any] = Map.empty,
+    run: Option[String] = None,
+    needs: Seq[String] = Seq.empty,
+    @JsonProperty("when") `when`: Option[String] = None
+)
diff --git a/pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/WorkflowLoader.scala b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/WorkflowLoader.scala
new file mode 100644
index 000000000..ce19fd197
--- /dev/null
+++ b/pipeline/src/main/scala/com/kakao/actionbase/pipeline/workflow/WorkflowLoader.scala
@@ -0,0 +1,107 @@
+package com.kakao.actionbase.pipeline.workflow
+
+import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory
+import com.fasterxml.jackson.module.scala.{ClassTagExtensions, DefaultScalaModule}
+import com.kakao.actionbase.pipeline.runner.{Expression, ExtendsResolver}
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path, Paths}
+import scala.collection.JavaConverters._
+
+/** Loads a workflow YAML into a `Workflow` case class.
+  *
+  * Pipeline:
+  *   1. Parse YAML to a raw `Map[String, Any]`. 2. Resolve `presets:` first (so its `$extends`/expressions resolve
+  *      before downstream uses). 3. Build an `Expression.Context` with the resolved presets and any `env:` block. 4.
+  *      Resolve `$extends` everywhere else in the tree. 5. Resolve remaining `${{ ... }}` expressions in string leaves
+  *      (env / presets / load — `needs.*` is deferred to runtime since job results aren't yet known at load time). 6.
+  *      Bind the resulting tree onto `Workflow` via Jackson.
+  */
+object WorkflowLoader {
+
+  private[pipeline] lazy val mapper: ObjectMapper with ClassTagExtensions = {
+    val m = new ObjectMapper(new YAMLFactory()) with ClassTagExtensions
+    m.registerModule(DefaultScalaModule)
+    m.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+    m.configure(DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES, true)
+    m
+  }
+
+  def load(path: Path): Workflow = {
+    val raw     = parseRaw(readUtf8(path))
+    val baseDir = path.toAbsolutePath.getParent
+    bind(raw, baseDir)
+  }
+
+  private def readUtf8(path: Path): String =
+    new String(Files.readAllBytes(path), StandardCharsets.UTF_8)
+
+  /** Used by tests / programmatic callers when YAML is already a string. `baseDir` is needed so `load(...)` expressions
+    * can resolve relative paths.
+    */
+  def loadString(yaml: String, baseDir: Path): Workflow = bind(parseRaw(yaml), baseDir)
+
+  private def parseRaw(yaml: String): Map[String, Any] = {
+    val node = mapper.readValue(yaml, classOf[java.util.Map[String, Any]])
+    if (node == null) Map.empty else node.asScala.toMap
+  }
+
+  private def bind(raw: Map[String, Any], baseDir: Path): Workflow = {
+    // 1. env (raw, no expression evaluation — env values are static strings)
+    val env: Map[String, String] = raw.get("env") match {
+      case Some(m: java.util.Map[_, _]) =>
+        m.asInstanceOf[java.util.Map[String, Any]].asScala.iterator.map { case (k, v) => k -> stringify(v) }.toMap
+      case Some(m: scala.collection.Map[_, _]) =>
+        m.asInstanceOf[scala.collection.Map[String, Any]].iterator.map { case (k, v) => k -> stringify(v) }.toMap
+      case _ => Map.empty
+    }
+
+    val loader: String => Any = pathStr => {
+      val p        = Paths.get(pathStr)
+      val resolved = if (p.isAbsolute) p else baseDir.resolve(p).normalize()
+      parseRaw(readUtf8(resolved))
+    }
+
+    // 2. resolve presets first with a context that has only env + load (presets refers to itself recursively
+    //    only via `$extends: ${{ presets.X }}`, which would loop — current model: presets resolve in declaration order)
+    val rawPresets = raw.get("presets") match {
+      case Some(m: java.util.Map[_, _])        => m.asInstanceOf[java.util.Map[String, Any]].asScala.toMap
+      case Some(m: scala.collection.Map[_, _]) => m.asInstanceOf[scala.collection.Map[String, Any]].toMap
+      case _                                   => Map.empty
+    }
+    val presetCtx = Expression.Context(env = env, loadYaml = loader)
+    val resolvedPresetsRaw = ExtendsResolver.resolve(rawPresets.asJava, presetCtx) match {
+      case jm: java.util.Map[_, _] => jm.asInstanceOf[java.util.Map[String, Any]].asScala.toMap
+      case _                       => Map.empty[String, Any]
+    }
+    val resolvedPresets: Map[String, Any] = Expression.resolveDeep(resolvedPresetsRaw.asJava, presetCtx) match {
+      case jm: java.util.Map[_, _] => jm.asInstanceOf[java.util.Map[String, Any]].asScala.toMap
+      case _                       => Map.empty
+    }
+
+    // 3. full context for the rest of the tree
+    val fullCtx = Expression.Context(env = env, presets = resolvedPresets, loadYaml = loader)
+
+    // 4. resolve $extends everywhere in the rest of the tree
+    val withoutPresets  = raw - "presets"
+    val extendsResolved = ExtendsResolver.resolve(withoutPresets.asJava, fullCtx)
+
+    // 5. resolve remaining expressions in string leaves; needs.* and `when:` operator forms are deferred to runtime
+    val resolved = Expression.resolveDeep(extendsResolved, fullCtx, lenient = true) match {
+      case jm: java.util.Map[_, _] => jm.asInstanceOf[java.util.Map[String, Any]].asScala.toMap
+      case other                   => sys.error(s"workflow root must be a map, got: $other")
+    }
+
+    // 6. attach the resolved presets back so the resulting Workflow case class carries them
+    val withPresets = resolved + ("presets" -> resolvedPresets)
+
+    mapper.convertValue(withPresets.asJava, classOf[Workflow])
+  }
+
+  private def stringify(value: Any): String = value match {
+    case null      => ""
+    case s: String => s
+    case other     => other.toString
+  }
+}
diff --git a/pipeline/src/test/resources/workflows/extends-via-load.yaml b/pipeline/src/test/resources/workflows/extends-via-load.yaml
new file mode 100644
index 000000000..1287c6969
--- /dev/null
+++ b/pipeline/src/test/resources/workflows/extends-via-load.yaml
@@ -0,0 +1,11 @@
+name: extends-via-load
+
+jobs:
+  pi:
+    kind: spark
+    artifact: "com.kakao.actionbase:pipeline:0.x"
+    mainClass: SparkPiJob
+    submit:
+      $extends: ${{ load('preset/spark-small.yaml') }}
+      conf:
+        spark.sql.shuffle.partitions: 8
diff --git a/pipeline/src/test/resources/workflows/preset/spark-small.yaml b/pipeline/src/test/resources/workflows/preset/spark-small.yaml
new file mode 100644
index 000000000..1a614818f
--- /dev/null
+++ b/pipeline/src/test/resources/workflows/preset/spark-small.yaml
@@ -0,0 +1,2 @@
+driver-memory: 1g
+executor-memory: 2g
diff --git a/pipeline/src/test/resources/workflows/spark-pi.yaml b/pipeline/src/test/resources/workflows/spark-pi.yaml
new file mode 100644
index 000000000..38d3d86c0
--- /dev/null
+++ b/pipeline/src/test/resources/workflows/spark-pi.yaml
@@ -0,0 +1,27 @@
+name: spark-pi
+
+env:
+  samples: "1000000"
+
+presets:
+  spark-small:
+    driver-memory: 1g
+    executor-memory: 2g
+
+jobs:
+  pi:
+    kind: spark
+    artifact: "com.kakao.actionbase:pipeline:0.x"
+    mainClass: SparkPiJob
+    args:
+      samples: "${{ env.samples }}"
+    submit:
+      $extends: ${{ presets.spark-small }}
+      conf:
+        spark.sql.shuffle.partitions: 8
+
+  report:
+    kind: bash
+    needs: [pi]
+    when: "${{ needs.pi.result == 'success' }}"
+    run: 'echo "pi computed"'
diff --git a/pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/ExpressionTest.scala b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/ExpressionTest.scala
new file mode 100644
index 000000000..f0c172cd2
--- /dev/null
+++ b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/ExpressionTest.scala
@@ -0,0 +1,67 @@
+package com.kakao.actionbase.pipeline.runner
+
+import com.kakao.actionbase.pipeline.runner.Expression.{Context, NeedsView}
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+
+import scala.collection.JavaConverters._
+
+class ExpressionTest {
+
+  @Test
+  def evaluatesEnv(): Unit = {
+    val ctx = Context(env = Map("samples" -> "1000"))
+    assertEquals("1000", Expression.evaluate("env.samples", ctx))
+  }
+
+  @Test
+  def evaluatesPresetReturningMap(): Unit = {
+    val preset = Map[String, Any]("driver-memory" -> "1g", "executor-memory" -> "2g")
+    val ctx    = Context(presets = Map("spark-small" -> preset))
+    assertEquals(preset, Expression.evaluate("presets.spark-small", ctx))
+  }
+
+  @Test
+  def evaluatesNeedsResultAndOutputs(): Unit = {
+    val ctx = Context(needs = Map("pi" -> NeedsView("success", Map("estimate" -> "3.14"))))
+    assertEquals("success", Expression.evaluate("needs.pi.result", ctx))
+    assertEquals("3.14", Expression.evaluate("needs.pi.outputs.estimate", ctx))
+  }
+
+  @Test
+  def loadCallInvokesContextLoader(): Unit = {
+    val loaded = Map[String, Any]("a" -> 1).asJava
+    val ctx    = Context(loadYaml = path => { assertEquals("preset.yaml", path); loaded })
+    assertEquals(loaded, Expression.evaluate("load('preset.yaml')", ctx))
+  }
+
+  @Test
+  def resolveDeepReplacesWholeTokenWithRawValue(): Unit = {
+    val preset = Map[String, Any]("driver-memory" -> "1g").asJava
+    val ctx    = Context(presets = Map("small" -> preset))
+    val tree   = Map("submit" -> "${{ presets.small }}").asJava
+
+    val out = Expression.resolveDeep(tree, ctx).asInstanceOf[java.util.Map[String, Any]].asScala
+    assertEquals(preset, out("submit"))
+  }
+
+  @Test
+  def resolveDeepDoesStringInterpolation(): Unit = {
+    val ctx  = Context(env = Map("date" -> "2026-05-10"))
+    val tree = Map("path" -> "/data/${{ env.date }}/file").asJava
+    val out  = Expression.resolveDeep(tree, ctx).asInstanceOf[java.util.Map[String, Any]].asScala
+    assertEquals("/data/2026-05-10/file", out("path"))
+  }
+
+  @Test
+  def rejectsUnknownEnvKey(): Unit = {
+    val ctx = Context()
+    assertThrows(classOf[NoSuchElementException], () => Expression.evaluate("env.missing", ctx))
+  }
+
+  @Test
+  def rejectsMalformedExpression(): Unit = {
+    val ctx = Context()
+    assertThrows(classOf[IllegalArgumentException], () => Expression.evaluate("not.a.known.prefix", ctx))
+  }
+}
diff --git a/pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/ExtendsResolverTest.scala b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/ExtendsResolverTest.scala
new file mode 100644
index 000000000..f476896b6
--- /dev/null
+++ b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/runner/ExtendsResolverTest.scala
@@ -0,0 +1,96 @@
+package com.kakao.actionbase.pipeline.runner
+
+import com.kakao.actionbase.pipeline.runner.Expression.Context
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+
+import scala.collection.JavaConverters._
+
+class ExtendsResolverTest {
+
+  @Test
+  def deepMergesExtendedWithSiblings(): Unit = {
+    val preset = Map[String, Any]("driver-memory" -> "1g", "executor-memory" -> "2g").asJava
+    val ctx    = Context(presets = Map("small" -> preset))
+
+    val tree = Map[String, Any](
+      "$extends" -> "${{ presets.small }}",
+      "conf"     -> Map[String, Any]("spark.shuffle.partitions" -> 8).asJava
+    ).asJava
+
+    val out = ExtendsResolver.resolve(tree, ctx).asInstanceOf[java.util.Map[String, Any]].asScala
+    assertEquals("1g", out("driver-memory"))
+    assertEquals("2g", out("executor-memory"))
+    assertEquals(8, out("conf").asInstanceOf[java.util.Map[String, Any]].get("spark.shuffle.partitions"))
+  }
+
+  @Test
+  def siblingOverridesExtendedAtSameKey(): Unit = {
+    val preset = Map[String, Any]("driver-memory" -> "1g").asJava
+    val ctx    = Context(presets = Map("small" -> preset))
+
+    val tree = Map[String, Any](
+      "$extends"      -> "${{ presets.small }}",
+      "driver-memory" -> "4g"
+    ).asJava
+
+    val out = ExtendsResolver.resolve(tree, ctx).asInstanceOf[java.util.Map[String, Any]].asScala
+    assertEquals("4g", out("driver-memory"))
+  }
+
+  @Test
+  def deepMergesNestedMaps(): Unit = {
+    val preset = Map[String, Any](
+      "conf" -> Map[String, Any]("a" -> 1, "b" -> 2).asJava
+    ).asJava
+    val ctx = Context(presets = Map("p" -> preset))
+
+    val tree = Map[String, Any](
+      "$extends" -> "${{ presets.p }}",
+      "conf"     -> Map[String, Any]("b" -> 99, "c" -> 3).asJava
+    ).asJava
+
+    val out  = ExtendsResolver.resolve(tree, ctx).asInstanceOf[java.util.Map[String, Any]].asScala
+    val conf = out("conf").asInstanceOf[java.util.Map[String, Any]].asScala
+    assertEquals(1, conf("a"))
+    assertEquals(99, conf("b"))
+    assertEquals(3, conf("c"))
+  }
+
+  @Test
+  def resolvesNestedExtendsInsideExtended(): Unit = {
+    val inner = Map[String, Any]("driver-memory" -> "1g").asJava
+    val outer = Map[String, Any](
+      "$extends"        -> "${{ presets.inner }}",
+      "executor-memory" -> "2g"
+    ).asJava
+    val ctx = Context(presets = Map("inner" -> inner, "outer" -> outer))
+
+    val tree = Map[String, Any]("$extends" -> "${{ presets.outer }}").asJava
+
+    val out = ExtendsResolver.resolve(tree, ctx).asInstanceOf[java.util.Map[String, Any]].asScala
+    assertEquals("1g", out("driver-memory"))
+    assertEquals("2g", out("executor-memory"))
+  }
+
+  @Test
+  def rejectsNonStringExtendsValue(): Unit = {
+    val ctx  = Context()
+    val tree = Map[String, Any]("$extends" -> 42).asJava
+    assertThrows(classOf[IllegalArgumentException], () => ExtendsResolver.resolve(tree, ctx))
+  }
+
+  @Test
+  def rejectsExtendsValueNotWrappedInExpression(): Unit = {
+    val ctx  = Context()
+    val tree = Map[String, Any]("$extends" -> "presets.small").asJava
+    assertThrows(classOf[IllegalArgumentException], () => ExtendsResolver.resolve(tree, ctx))
+  }
+
+  @Test
+  def rejectsExtendsValueResolvingToScalar(): Unit = {
+    val ctx  = Context(env = Map("x" -> "scalar"))
+    val tree = Map[String, Any]("$extends" -> "${{ env.x }}").asJava
+    assertThrows(classOf[IllegalArgumentException], () => ExtendsResolver.resolve(tree, ctx))
+  }
+}
diff --git a/pipeline/src/test/scala/com/kakao/actionbase/pipeline/workflow/WorkflowLoaderTest.scala b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/workflow/WorkflowLoaderTest.scala
new file mode 100644
index 000000000..44717cfce
--- /dev/null
+++ b/pipeline/src/test/scala/com/kakao/actionbase/pipeline/workflow/WorkflowLoaderTest.scala
@@ -0,0 +1,71 @@
+package com.kakao.actionbase.pipeline.workflow
+
+import org.junit.jupiter.api.Assertions._
+import org.junit.jupiter.api.Test
+
+import java.nio.file.{Path, Paths}
+import scala.collection.JavaConverters._
+
+class WorkflowLoaderTest {
+
+  private val ResourcesRoot: Path =
+    Paths.get(getClass.getResource("/workflows").toURI)
+
+  @Test
+  def loadsBasicWorkflow(): Unit = {
+    val wf = WorkflowLoader.load(ResourcesRoot.resolve("spark-pi.yaml"))
+
+    assertEquals("spark-pi", wf.name)
+    assertEquals("1000000", wf.env("samples"))
+    assertEquals(2, wf.jobs.size)
+    assertEquals("spark", wf.jobs("pi").kind)
+    assertEquals(Some("SparkPiJob"), wf.jobs("pi").mainClass)
+    assertEquals(Some("com.kakao.actionbase:pipeline:0.x"), wf.jobs("pi").artifact)
+  }
+
+  @Test
+  def resolvesEnvExpressionInArgs(): Unit = {
+    val wf = WorkflowLoader.load(ResourcesRoot.resolve("spark-pi.yaml"))
+    val pi = wf.jobs("pi")
+    assertEquals("1000000", pi.args("samples"))
+  }
+
+  @Test
+  def resolvesPresetExtendsInSubmit(): Unit = {
+    val wf     = WorkflowLoader.load(ResourcesRoot.resolve("spark-pi.yaml"))
+    val submit = wf.jobs("pi").submit
+    // From preset spark-small
+    assertEquals("1g", submit("driver-memory"))
+    assertEquals("2g", submit("executor-memory"))
+    // From sibling override
+    val conf = submit("conf").asInstanceOf[scala.collection.Map[String, Any]]
+    assertEquals(8, conf("spark.sql.shuffle.partitions"))
+  }
+
+  @Test
+  def resolvesLoadExtendsAgainstFile(): Unit = {
+    val wf     = WorkflowLoader.load(ResourcesRoot.resolve("extends-via-load.yaml"))
+    val submit = wf.jobs("pi").submit
+    assertEquals("1g", submit("driver-memory"))
+    assertEquals("2g", submit("executor-memory"))
+    val conf = submit("conf").asInstanceOf[scala.collection.Map[String, Any]]
+    assertEquals(8, conf("spark.sql.shuffle.partitions"))
+  }
+
+  @Test
+  def carriesNeedsAndWhen(): Unit = {
+    val wf     = WorkflowLoader.load(ResourcesRoot.resolve("spark-pi.yaml"))
+    val report = wf.jobs("report")
+    assertEquals(Seq("pi"), report.needs)
+    assertTrue(report.`when`.isDefined)
+    assertTrue(report.`when`.get.contains("needs.pi.result"))
+  }
+
+  @Test
+  def deferredNeedsExpressionsArePreservedAtLoadTime(): Unit = {
+    // `needs.*` is runtime-only — should remain unresolved in the loaded Workflow's `when` field.
+    val wf  = WorkflowLoader.load(ResourcesRoot.resolve("spark-pi.yaml"))
+    val raw = wf.jobs("report").`when`.get
+    assertTrue(raw.contains("${{"), s"expected unresolved expression, got: $raw")
+  }
+}