Workshop context

This is week 1 homework for the DET (Data Engineer Things) Technical writing workshop led by Yaakov Bressler .

Workshop site: dataengineerthings.org

Workshop mental model

The DET workshop uses a 7 Writer Objectives × 6 Reader Objectives matrix to align writer intent with reader needs.

%%{init: {'theme': 'base', 'themeVariables': { 'fontSize': '14px'}}}%%
flowchart TD
    W["Writer Objectives(7)"] --> M["Objectives Matrix 7×6 = 42 alignments"]
    R["Reader Objectives(6)"] --> M
    M --> A["Alignment Find best match"]
    A --> T["Transform Adjust content"]

    W --> EX["Examples:• Teach | Resource | Convince"]
    R --> EX2["Examples:• Learn | Reference | Decide"]

    style M fill:#e1f5ff,stroke:#0066cc,color:#000
    style A fill:#d4edda,stroke:#28a745,color:#000
    style T fill:#fff3cd,stroke:#ffc107,color:#000

Assignment context

Homework 1: Redline the compile-time data contracts article (original here ) to show transformation from Teach → Resource objective.

Reviewer instructions

This redlined document uses:

• Strike-through for content to REMOVE
• Admonition boxes explaining WHY each decision
• Plain text for content to KEEP
• Visual emphasis on the 70% reduction when objectives align

The goal: demonstrate objectives-driven editing through transparent redlining.

Original vs target objectives

Original article objective: Teach

Writer intent: Build understanding from scratch

Reader need: Learn compile-time contracts conceptually

Content structure:

• Personal stories - “Once, an upstream team silently renamed..”
• Mental models - “Here’s how to think about macros..”
• Step-by-step walkthroughs - “Let’s break this down line by line..”
• Philosophy and motivation - “Why this matters”

Target objective: Resource

Writer intent: Provide quick reference for experts

Reader need: Look up API, patterns, gotchas

Content structure:

• Prerequisites (what you need to know first)
• Quick start (runnable code immediately)
• API reference (shapes, policies, macros)
• Production patterns (what works in the real world)
• Gotchas (common mistakes)

Transformation strategy

This redlining systematically removes Teaching content (stories, walkthroughs, philosophy) while preserving Resource content (code, patterns, references, gotchas).

Expected reduction: ~70% (from 1,751 lines to ~500-600 lines of Resource content + homework annotations).

Redlining method

Using Talmudic commentary style:

1. Circle (keep): Code examples, production patterns, gotchas, references
2. Cross (remove): Stories, mental models, step-by-step explanations, “What’s happening” sections
3. Annotate (explain): WHY each decision using admonition boxes

EVERY line from the original is preserved below - either struck-through with explanation or kept with explanation.

Complete original article (redlined)

What if broken pipelines never launched - because the compiler stopped them?

> “If it compiles, contracts align."

I’ve been working on this problem for a while now. You know how it goes - you’re running a data pipeline, everything’s fine, then someone changes a field name upstream and boom, your job crashes at 2 AM. The on-call engineer (probably you) gets paged, and you spend the next hour figuring out what went wrong.

Once, an upstream team silently renamed a column amount → amt. Our pipeline didn’t crash immediately - it happily wrote nulls for weeks. By the time we caught it, we had to backfill millions of records. That’s the day I realized: runtime validation catches problems late; compile-time contracts catch them earlier.

This post shows you how to catch these issues at compile time. Not at runtime. Not even at test time. At compile time. Your code literally won’t build if schemas drift.

We’ll build this from scratch - first understand how Scala 3 macros work, then write a minimal contract system that proves schema compatibility before your code even runs. I’ll also show the Scala 2 version because, let’s be honest, many teams are still on Scala 2.

All code here runs in a vanilla sbt project. You can grab the working code in github from https://github.com/vim89/compile-time-data-contracts and run it yourself.

## Why this matters

Look, I get it. Writing tests is important. But tests are samples. They check what you think might break. The compiler, though? It’s exhaustive for structure. It can prove your schemas match - every field, every type, every time.

Here’s the thing - schemas change all the time in real projects. And when they do: - Your build should fail early, not your production job - Keep transformations pure and effects at the edges (so retries don’t duplicate writes) - Let the compiler do the heavy lifting

Important upfront: This focuses on preventing schema drift within your controlled boundaries - the systems you build, deploy, and version. It doesn’t replace runtime validation for external data sources or third-party APIs. Think of it as catching a large class of schema mismatches during development and deployment, not eliminating all drift everywhere. You’ll still need runtime checks for data coming from systems you don’t control.

## What we’ll build

We’re going to build a small compile-time validation system: 1. A way to describe case class structure at compile time (we’ll call it “shape”) 2. A policy system - Exact, Backward, Forward compatibility modes 3. A macro that proves “these two shapes conform under policy P” - or refuses to compile 4. A phantom-typed pipeline builder that enforces correct construction order

First, Scala 3 (using inline + quotes), then Scala 2 (using blackbox macros). I’ll point out the differences as we go.

### Run the code in github

1
2
3
4
5

git clone https://github.com/vim89/compile-time-data-contracts.git
cd compile-time-data-contracts

# Scala 3
sbt "runMain ctdc.CtdcPoc"

## Part 1 - Scala 3: the mental model (inline + quotes)

Scala 3 macros are different from Scala 2. Better, actually. They’re safer and easier to reason about.

Two main ideas: - inline def: tells the compiler “expand this at compile time” - quotes + Expr[T]: gives you a typed AST that you can inspect

sequenceDiagram
    actor Dev
    participant Compiler
    participant Macro as "Conforms.impl"
    participant Reflect as "TypeRepr/quotes"
    Dev->>Compiler: compile
    Compiler->>Macro: splice ${ ... }
    Macro->>Reflect: inspect types
    Reflect-->>Macro: shapes for Out and Contract
    alt conforming
        Macro-->>Compiler: emit Conforms[Out,Contract,P]
        Compiler-->>Dev: build succeeds
    else drift
        Macro-->>Compiler: report.errorAndAbort(msg)
        Compiler-->>Dev: compile error with drift details
    end

Here’s a minimal macro to see how it works:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// build.sbt: use Scala 3
// scalaVersion := "3.3.3"

// src/main/scala/com/example/IntroMacro.scala
package com.example
import scala.quoted.*

object IntroMacro:
  inline def show(inline x: Any): String = ${ showImpl('x) }
  private def showImpl(x: Expr[Any])(using Quotes): Expr[String] =
    Expr(s"You passed: ${x.show}")

What’s happening here? - The inline method is what users call - The ${ ... } splice calls the implementation - The implementation gets 'x (quoted code) as Expr[Any] - We return Expr[String] - the AST for what the compiler will embed

Try it:

1
2
3
4

// src/main/scala/com/example/Demo.scala
package com.example
@main def run(): Unit =
  println(IntroMacro.show(1 + 2)) // prints: You passed: 1.+(2)

See how it printed the AST of 1 + 2? That’s the power here - we can inspect code structure at compile time.

Check out the Scala 3 macros overview and best practices for more details.

Part 2 - Describe shapes at compile time (Scala 3)

Now we need to describe case class structure. In real projects, you might use Shapeless, Magnolia, or Scala 3’s Mirror. I’ll show a compact Mirror-based version here because it’s self-contained.

The idea: a typeclass Shape[A] that knows the fields of A.

classDiagram
    class Field {
      +name: String
      +tpe: String
      +hasDefault: Boolean
      +isOptional: Boolean
    }
    class Shape {
      +fields(): List~Field~
    }
    Shape "1" --> "*" Field : returns

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

// src/main/scala/com/example/Shape.scala
package com.example

import scala.deriving.Mirror

final case class Field(name: String, tpe: String, hasDefault: Boolean = false, isOptional: Boolean = false)
trait Shape[A]:
  def fields: List[Field]

object Shape:
  given Shape[String] with { def fields = Nil }
  given Shape[Int]     with { def fields = Nil }
  given Shape[Long]    with { def fields = Nil }
  given [A]: Shape[Option[A]] with { def fields = Nil } // element optionality handled elsewhere

  inline given derived[A](using m: Mirror.Of[A]): Shape[A] =
    inline m match
      case p: Mirror.ProductOf[A] =>
        val labels = constValueTuple[p.MirroredElemLabels]
        val types  = typeNames[p.MirroredElemTypes]
        val zipped = zip(labels, types)
        new Shape[A] {
          def fields: List[Field] =
            zipped.map { (name, tpe) => Field(name, tpe, hasDefault = false, isOptional = tpe.startsWith("Option[")) }.toList
        }
      case _ => new Shape[A] { def fields = Nil }

  import scala.compiletime.{erasedValue, constValueTuple}
  private inline def typeNames[T <: Tuple]: List[String] = inline erasedValue[T] match
    case _: EmptyTuple => Nil
    case _: (h *: t)   => summonTypeName[h] :: typeNames[t]

  private inline def summonTypeName[T]: String = constValue["" + T]

  private inline def zip[L <: Tuple, R <: List[String]](labels: L, types: List[String]): List[(String, String)] =
    inline labels match
      case _: EmptyTuple => Nil
      case _: (h *: t)   => constValue[h].asInstanceOf[String] -> types.head :: zip[t, List[String]](erasedValue, types.tail)

Yeah, it’s a bit dense. But here’s what it does: - Uses Scala 3’s Mirror to reflect on case classes at compile time - Extracts field names and types - Marks optional fields (those wrapped in Option) - Returns a List[Field] describing the structure

flowchart LR
    A["Case class A(...)"] --> M["Mirror.Of[A]"]
    M --> L["MirroredElemLabels"]
    M --> T["MirroredElemTypes"]
    L --> Z["zip(labels, types)"]
    T --> Z
    Z --> S["Shape[A].fields: List[Field]"]

Usage:

1
2
3
4
5
6
7
8

// src/test/scala/com/example/ShapeSpec.scala
package com.example

final case class User(id: Long, email: String, note: Option[String])

@main def checkShape(): Unit =
  val s = summon[Shape[User]]
  println(s.fields) // List(Field("id","Long"), Field("email","String"), Field("note","Option[String]",false,true))

Now we can ask the compiler “what fields does this case class have?” at compile time. That’s the foundation.

If you prefer libraries, check out Magnolia - it works great for both Scala 2 and 3.

Part 3 - Policies: Exact / Backward / Forward

Here’s where we define compatibility rules. Think of these as your migration strategies:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// src/main/scala/com/example/Policy.scala
package com.example

sealed trait Policy
object Policy:
  sealed trait Exact     extends Policy
  sealed trait Backward  extends Policy // producer can add optional/defaulted fields
  sealed trait Forward   extends Policy // consumer tolerates extra fields
  case object Exact    extends Exact
  case object Backward extends Backward
  case object Forward  extends Forward

Why three policies?

• Exact: Schemas must match exactly. No surprises. Use this for critical paths.
• Backward: Producer can add optional fields. Useful when rolling out new features without breaking consumers.
• Forward: Consumer ignores extra fields. Useful when the producer might send more than you need.

flowchart LR
    EX["Policy.Exact"]
    BW["Policy.Backward"]
    FW["Policy.Forward"]

    EX --- BW
    EX --- FW

    EX:::tight -->|no diffs allowed| OK1["✔ match only"]
    BW:::relax -->|allow missing optional in producer\nallow extra fields in producer| OK2["✔ migration adds"]
    FW:::relax -->|allow extra fields in consumer side| OK3["✔ tolerant reader"]

    classDef tight fill:#eef,stroke:#446;
    classDef relax fill:#efe,stroke:#484;

Visual (save/share):

In practice: start with Exact, relax to Backward/Forward during migrations, then tighten back to Exact once stable.

## Part 4 - How this fits with existing tools

Before we dive deeper, you might be thinking: “We already have Avro, Protobuf, schema registries, and Great Expectations. Why add compile-time contracts?"

Fair question. Here’s how they compare:

Schema Registry (Confluent, AWS Glue) - What it does: Centralized schema storage with version control - When it helps: Runtime validation across services, schema evolution tracking - The gap: You find mismatches when the job runs, not when you compile - Use together: Schema registry for cross-service contracts; compile-time for intra-repo checks

Avro / Protobuf - What they do: Binary serialization with embedded schemas - When they help: Network efficiency, strong contracts between services - The gap: Schema compatibility is checked at runtime or via external tooling - Use together: Avro/Protobuf for wire format; compile-time checks to ensure your case classes match those schemas

Great Expectations / Deequ - What they do: Runtime data quality validation (nulls, ranges, distributions) - When they help: Catching bad data values, statistical anomalies - The gap: They validate data content, not schema structure at compile time - Use together: Compile-time for structure; Great Expectations for data quality

The compile-time advantage: Catches drift in your controlled codebase before deployment. If your transform expects field X but the source produces field Y, your build fails - not your production job. Think of it as an additional safety layer for systems you build and version together.

Real scenario: You have Avro schemas in a registry, but your Scala pipeline reads them into case classes. Compile-time contracts verify those case classes match the expected contract before you deploy. If upstream changes the Avro schema, your build breaks - you don’t wait for the job to crash.

## Part 5 - Understanding TypeRepr: The compiler’s view of types

Before we dive into the macro, you need to understand TypeRepr - Scala 3’s way of representing types during compilation.

Think of it like this: When you write List[String], the compiler doesn’t just see text. It sees a structured tree:

1
2
3
4

AppliedType(List, [String])
   │               │
   │               └─ TypeRepr for String
   └─ Type constructor

flowchart TB
    A["List[String]"] --> B["AppliedType(List,[String])"]
    B --> C["tycon = List"]
    B --> D["args = [String]"]

Key TypeRepr operations:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

val t: TypeRepr = TypeRepr.of[List[String]]

// 1. Subtype checking: Is List[String] a subtype of Seq[?]?
t <:< TypeRepr.of[Seq[?]]  // true

// 2. Type equality: Is this exactly String?
t =:= TypeRepr.of[String]  // false

// 3. Pattern matching: Break it apart
t match
  case AppliedType(tycon, args) =>
    // tycon = List
    // args = [String]

Why we need this: To compare schemas, we need to ask questions like: - “Is this field an Option?" - “What’s inside this List?" - “Does this case class have a primary constructor?"

TypeRepr lets us answer these at compile time.

## Part 5 - The TypeInspector pattern: One question at a time

Now let’s build the utilities. Each function answers ONE question about a type. Start simple:

1
2
3

def isCaseClass(t: TypeRepr): Boolean =
  val s = t.typeSymbol  // Get the symbol (the "name" of the type)
  s.isClassDef && s.flags.is(Flags.Case)

1
2
3

case class User(id: Long)  // ✅ isClassDef=true, has Case flag
trait UserTrait            // ❌ isClassDef=false
object UserObject          // ❌ not a class

1
2
3

def appliedArgs(t: TypeRepr): List[TypeRepr] = t match
  case AppliedType(_, args) => args  // Extract the arguments
  case _                    => Nil   // No arguments

1
2
3
4
5
6
7
8

List[String]        // AppliedType(List, [String])
                    // appliedArgs returns [String]

Map[String, Int]    // AppliedType(Map, [String, Int])
                    // appliedArgs returns [String, Int]

String              // Not an AppliedType
                    // appliedArgs returns Nil

1
2
3
4

def optionArg(t: TypeRepr): Option[TypeRepr] =
  if t <:< TypeRepr.of[Option[?]] then  // Is it a subtype of Option[Something]?
    appliedArgs(t).headOption           // Yes! Extract the Something
  else None                             // Nope

1
2
3
4
5
6
7
8
9

Option[String]  // ✅ Is subtype of Option[?]
                // appliedArgs returns [String]
                // headOption returns Some(String)

Some[Int]       // ✅ Some is subtype of Option
                // Returns Some(Int)

String          // ❌ Not an Option
                // Returns None

1
2
3
4
5
6
7
8

def seqArg(t: TypeRepr): Option[TypeRepr] =
  val isSeqLike =
    t <:< TypeRepr.of[List[?]] ||
    t <:< TypeRepr.of[Seq[?]] ||
    t <:< TypeRepr.of[Vector[?]] ||
    t <:< TypeRepr.of[Array[?]] ||
    t <:< TypeRepr.of[Set[?]]
  if isSeqLike then appliedArgs(t).headOption else None

1
2
3
4
5

List[String]    // ✅ Is List[?]
Seq[Int]        // ✅ Is Seq[?]
Vector[Long]    // ✅ Is Vector[?]
Array[Byte]     // ✅ Is Array[?]
Set[String]     // ✅ Is Set[?]

1
2
3
4
5
6

def mapArgs(t: TypeRepr): Option[(TypeRepr, TypeRepr)] =
  if t <:< TypeRepr.of[Map[?, ?]] then
    appliedArgs(t) match
      case k :: v :: Nil => Some((k, v))  // Got key and value
      case _ => report.errorAndAbort(s"Map requires two type args: ${t.show}")
  else None

1
2
3

Map[String, Int]  // ✅ Returns Some((String, Int))
Map[Long, User]   // ✅ Returns Some((Long, User))
String            // ❌ Returns None

1
2
3
4
5
6
7

def isAtomicKey(t: TypeRepr): Boolean =
  t =:= TypeRepr.of[String] ||
  t =:= TypeRepr.of[Int] ||
  t =:= TypeRepr.of[Long] ||
  t =:= TypeRepr.of[Short] ||
  t =:= TypeRepr.of[Byte] ||
  t =:= TypeRepr.of[Boolean]

1
2
3

Map[String, User]  // ✅ String is atomic
Map[Int, Data]     // ✅ Int is atomic
Map[User, String]  // ❌ User is NOT atomic - compile error!

## Part 6 - Building shapes recursively: The interpreter pattern

Now we use those inspectors to build a TypeShape - our internal representation of a type’s structure.

This is recursive. For List[Option[User]], we build:

1
2
3
4
5

SequenceShape(
  OptionalShape(
    StructShape([field1, field2, ...])
  )
)

Let me show you the algorithm step by step.

flowchart TD
    T["typeShapeOf(t)"] --> O{"Option?"}
    O -- "Yes" --> OI["inner = optionArg(t)"] --> R1["typeShapeOf(inner)"]
    O -- "No" --> S{"Sequence?"}
    S -- "Yes" --> SE["elem = seqArg(t)"] --> R2["SequenceShape(typeShapeOf(elem))"]
    S -- "No" --> M{"Map?"}
    M -- "Yes" --> MKV["(k,v) = mapArgs(t)"]
    MKV --> K{"isAtomicKey(k)?"}
    K -- "No" --> ERR["abort: unsupported key"]
    K -- "Yes" --> MV["MapShape(PrimitiveShape(k), typeShapeOf(v))"]
    M -- "No" --> CC{"Case class?"}
    CC -- "Yes" --> FS["StructShape(collect FieldShape...)"]
    CC -- "No" --> P["PrimitiveShape(t.show)"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

def typeShapeOf(t: TypeRepr): TypeShape =
  // Step 1: Is it an Option?
  optionArg(t).map(typeShapeOf).getOrElse {

    // Step 2: Is it a sequence?
    seqArg(t).map(a => SequenceShape(typeShapeOf(a))).getOrElse {

      // Step 3: Is it a Map?
      mapArgs(t).map { case (k, v) =>
        if !isAtomicKey(k) then
          report.errorAndAbort(s"Unsupported Map key: ${k.show}")
        MapShape(PrimitiveShape(k.show), typeShapeOf(v))
      }.getOrElse {

        // Step 4: Is it a case class?
        if isCaseClass(t) then
          // ... handle case class
        else
          // Step 5: Fallback - primitive
          PrimitiveShape(t.show)
      }
    }
  }

flowchart TD
    A["List[Option[String]]"] -->|typeShapeOf| B{"Option?"}
    B -- "No" --> C{"Sequence?"}
    C -- "Yes" --> D["elem = Option[String]"]
    D --> E["typeShapeOf(Option[String])"]
    E --> F{"Option?"}
    F -- "Yes" --> G["inner = String"]
    G --> H["typeShapeOf(String)"]
    H --> I{"Case class?"}
    I -- "No" --> J["PrimitiveShape(String)"]
    J --> K["SequenceShape(PrimitiveShape(String))"]

1
2
3

PrimitiveShape("String")
  → (from Option recursion)
  → SequenceShape(PrimitiveShape("String"))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

if isCaseClass(t) then
  val sym    = t.typeSymbol                      // Get the class symbol
  val params = sym.primaryConstructor.paramSymss.flatten  // Get constructor params

  val fields = params.map { p =>
    val name       = p.name                      // Field name
    val ptpe       = t.memberType(p)             // Field type
    val hasDefault = p.flags.is(Flags.HasDefault) // Has default value?
    val (uT, isOpt) = optionArg(ptpe).fold(ptpe -> false)(a => a -> true)

    FieldShape(name, typeShapeOf(uT), hasDefault, isOpt)
  }

  StructShape(fields)

flowchart TD
    U["User(id: Long, email: String, age: Option[Int] = None)"]
    U --> F1["Field 'id' : Long"]
    F1 --> S1["PrimitiveShape(Long)"]
    U --> F2["Field 'email' : String"]
    F2 --> S2["PrimitiveShape(String)"]
    U --> F3["Field 'age' : Option[Int] (default None)"]
    F3 --> O1{"Option?"}
    O1 -- "Yes" --> A1["inner = Int"]
    A1 --> S3["PrimitiveShape(Int)"]
    S1 --> COLLECT["Collect fields"]
    S2 --> COLLECT
    S3 --> COLLECT
    COLLECT --> FINAL["StructShape([ id:PrimitiveShape(Long), email:PrimitiveShape(String), age:PrimitiveShape(Int) (optional, default) ])"]

1
2
3
4
5

case class Order(
  items: List[LineItem],           // Recurses into List, then into LineItem
  metadata: Map[String, String],   // Recurses into Map, then String (twice)
  address: Option[Address]         // Recurses into Option, then Address
)

Part 7 - Field vs Element optionality: A critical distinction

Here’s something subtle that most guides miss. The code in github handles two kinds of optionality:

The code in github handles two kinds of optionality:

Field-level optionality (Option on the field itself)

1
2

case class User(name: String, age: Option[Int])
//                                   ^^^^^^^^^^^ field-level Option

The macro detects this and sets isOptional = true on the FieldShape:

1
2

val (uT, isOpt) = optionArg(ptpe).fold(ptpe -> false)(a => a -> true)
FieldShape(name, typeShapeOf(uT), hasDefault, isOpt)

Element-level optionality (Option inside a collection)

1
2

case class Data(items: List[Option[String]])
//                          ^^^^^^^^^^^^^^ element-level Option

This creates a nested shape: SequenceShape(OptionShape(PrimitiveShape("String")))

Wait, that’s not in the code in github! Let me check… Actually, the code in github’s typeShapeOf handles this by recursing:

1
2
3

optionArg(t).map(typeShapeOf).getOrElse {
  seqArg(t).map(a => SequenceShape(typeShapeOf(a))).getOrElse { ... }
}

So List[Option[String]] becomes:

1. seqArg detects List → extract Option[String]
2. Recurse on Option[String] → typeShapeOf(Option[String])
3. optionArg detects Option → extract String
4. Base case: PrimitiveShape("String")

But the final shape is SequenceShape(PrimitiveShape("String")) - the intermediate Option is consumed!

Why this matters for contracts: When you write:

1
2

case class Contract(items: List[String])
case class Producer(items: List[Option[String]])

These should probably NOT conform under Exact. The producer allows null elements, the contract doesn’t. But the current code in github might allow it because the Option gets unwrapped.

This is the kind of edge case you discover when shipping to production. The fix would be to add an OptionalShape wrapper in the model.

Visual (field vs element optionality):

flowchart LR
    A["List[Option[String]]"] --> B{"Element optionality?"}
    B -- "intended shape" --> C["SequenceShape(OptionalShape(PrimitiveShape(String)))"]
    B -- "current code in github\n(consumes Option)" --> D["SequenceShape(PrimitiveShape(String))"]

Part 8 - Compile‑time “conforms” evidence (the full picture)

Now let’s put it all together. Here’s the complete macro from the code in github:

Here’s the complete macro from the code in github:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

// src/main/scala/com/example/Conforms.scala
package com.example

import scala.quoted.*

final class Conforms[Out, Contract, P <: Policy]

object Conforms:
  inline given materialize[Out, Contract, P <: Policy](using Shape[Out], Shape[Contract]): Conforms[Out, Contract, P] =
    ${ conformsImpl[Out, Contract, P] }

  private def conformsImpl[Out: Type, Contract: Type, P: Type](using Quotes): Expr[Conforms[Out, Contract, P]] =
    import quotes.reflect.*

    def shapeOf[A: Type]: List[(String, String, Boolean, Boolean)] =
      val sh = Expr.summon[Shape[A]].getOrElse(report.errorAndAbort("No Shape available"))
      '{
        val fs = $sh.fields
        fs.map(f => (f.name, f.tpe, f.hasDefault, f.isOptional))
      }.valueOrAbort

    val out      = shapeOf[Out]
    val contract = shapeOf[Contract]

    // Compare fields by name (unordered, case-sensitive for now)
    val outMap      = out.map { case (n,t,_,opt) => n -> (t,opt) }.toMap
    val contractMap = contract.map { case (n,t,_,opt) => n -> (t,opt) }.toMap

    val missing = contract.collect { case (n,t,_,opt) if !outMap.contains(n) => s"$n:$t${if opt then " (optional)" else ""}" }
    val extra   = out.collect      { case (n,_,_,_) if !contractMap.contains(n) => n }
    val mism    = contract.collect {
      case (n,t,_,opt) if outMap.get(n).exists(_._1 != t)  => s"$n expected $t, found ${outMap(n)._1}"
      case (n,_,_,opt) if outMap.get(n).exists(_._2 != opt) => s"$n optionality mismatch"
    }

    val (okMissing, okExtra, okMism) = Type.of[P] match
      case '[Policy.Exact]    => (missing, extra, mism)
      case '[Policy.Backward] => (missing.filterNot(_.contains("(optional)")), Nil, mism)
      case '[Policy.Forward]  => (Nil, extra, mism)
      case _                  => (missing, extra, mism)

    if okMissing.nonEmpty || okExtra.nonEmpty || okMism.nonEmpty then
      val msg = s"""
        |Compile‑time contract drift (policy: ${Type.show[P]}).
        |Out: ${Type.show[Out]} vs Contract: ${Type.show[Contract]}
        |Missing: ${okMissing.mkString(", ")}
        |Extra: ${okExtra.mkString(", ")}
        |Mismatched: ${okMism.mkString("; ")}
        |""".stripMargin
      quotes.reflect.report.errorAndAbort(msg)
    else '{ new Conforms[Out, Contract, P] }

Let me break down what’s happening:

1. Extract shapes: We summon Shape[Out] and Shape[Contract] and get their field lists. Note the .valueOrAbort - this forces compile-time evaluation.

2. Compare: We build maps of field names → (type, optionality) and find differences.

3. Apply policy: - Exact: all differences are errors - Backward: missing optional fields are OK, extras are OK - Forward: extras are OK (but missing required fields are not)

4. Abort or succeed: If there are violations, we call report.errorAndAbort with a detailed message showing exactly what’s wrong. Otherwise, we return the evidence.

The beauty here: this runs during compilation. If schemas don’t match, your code won’t compile. Period.

Use it like this:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// src/test/scala/com/example/ConformsSpec.scala
package com.example

final case class Contract(id: Long, email: String)
final case class OutOk(id: Long, email: String)
final case class OutMissing(id: Long)

@main def checkConformance(): Unit =
  summon[Conforms[OutOk,      Contract, Policy.Exact]]   // ✅ compiles
  // summon[Conforms[OutMissing, Contract, Policy.Exact]] // ❌ compile‑time error (Missing: email:String)
  summon[Conforms[OutMissing, Contract, Policy.Backward]] // ✅ compiles (migration mode)

That’s it. The entire core. In production, you’d want to handle the element optionality issue I mentioned, but the structure stays the same.

## Part 9 - Developer ergonomics: What using this actually feels like

1
2
3
4
5

[error] Compile-time contract drift (policy: Policy.Exact).
[error] Out: CustomerProducer vs Contract: CustomerContract
[error] Extra: segment
[error] Missing:
[error] Mismatched:

1
2
3
4
5
6
7
8

// Your Spark job
val df = spark.read.parquet(path)
val typed = df.as[CustomerProducer]  // Spark encoder

// Compile-time check before transformation
summon[Conforms[CustomerProducer, CustomerContract, Policy.Exact]]

val output = typed.transform(dropSegment).as[CustomerContract]

1
2
3
4

// Avro schema in registry → case class → contract check
case class CustomerAvro(id: Long, email: String, amt: Double)
summon[Conforms[CustomerAvro, CustomerContract, Policy.Exact]]
// Build fails if Avro schema drift!

Part 10 - Nested types: Maps, Lists, and deep structures

The code in github handles nested types beautifully. Look at this example from CtdcPoc.scala:

Look at this example from CtdcPoc.scala:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

// Deep / Nested structures
final case class LineItem(sku: String, qty: Int, attrs: Map[String, String])
final case class Address(street: String, zip: String)

final case class OrderOut(
  id: Long,
  items: List[LineItem],
  shipTo: Option[Address],
  tags: Set[String]
)

final case class OrderContract(
  id: Long,
  items: Seq[LineItem],        // List vs Seq - both are sequences
  shipTo: Option[Address],      // Nested case class
  tags: Seq[String] = Nil       // Set vs Seq, has default
)

val evDeepOk: SchemaConforms[OrderOut, OrderContract, SchemaPolicy.Backward.type] = summon

What’s happening here:

1. Nested case classes - Address inside OrderOut. The macro recurses: when it sees Option[Address], it unwraps to Address, then checks if Address is a case class, and recurses again to get its fields.

2. Collections - List[LineItem] vs Seq[LineItem]. Both match because seqArg treats them equivalently. The macro sees “sequence of LineItem” and recurses on LineItem.

3. Maps - Map[String, String] in attrs. The macro checks key type (must be atomic), then recurses on value type.

4. Default values - tags: Seq[String] = Nil has a default. Under Backward policy, if OrderOut is missing tags, it’s OK because the contract has a default.

This is why the TypeInspector pattern matters - it handles arbitrary nesting without special cases.

Part 10 - Policy modifiers in two minutes (Ordered, CI, By‑Position)

Sometimes you need stricter comparison. Or different rules. Here are three quick policy extensions:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

// src/main/scala/com/example/PolicyMods.scala
package com.example

sealed trait PolicyMod
object PolicyMod:
  sealed trait ExactOrdered   extends PolicyMod // names + order + types must match
  sealed trait ExactCI        extends PolicyMod // case‑insensitive field names
  sealed trait ExactByPosition extends PolicyMod // ignore names; types by index
  case object ExactOrdered    extends ExactOrdered
  case object ExactCI         extends ExactCI
  case object ExactByPosition extends ExactByPosition

object ConformsMod:
  import scala.quoted.*
  inline def apply[Out, Contract, M <: PolicyMod](using Shape[Out], Shape[Contract]): Unit = ${ impl[Out, Contract, M] }

  private def impl[Out: Type, Contract: Type, M: Type](using Quotes): Expr[Unit] =
    import quotes.reflect.*
    def fields[A: Type]: List[(String,String)] =
      val sh = Expr.summon[Shape[A]].getOrElse(report.errorAndAbort("No Shape available"))
      '{ $sh.fields.map(f => (f.name, f.tpe)) }.valueOrAbort

    val out      = fields[Out]
    val contract = fields[Contract]

    val mismatches: List[String] = Type.of[M] match
      case '[PolicyMod.ExactCI] =>
        val n = (s: String) => s.toLowerCase
        val om = out.map{ case (n0,t) => n(n0) -> t }.toMap
        val cm = contract.map{ case (n0,t) => n(n0) -> t }.toMap
        (cm.keySet union om.keySet).toList.flatMap { k =>
          (cm.get(k), om.get(k)) match
            case (Some(t1), Some(t2)) if t1 != t2 => List(s"$k expected $t1, found $t2")
            case (Some(_), None) => List(s"missing ${k}")
            case (None, Some(_)) => List(s"extra ${k}")
            case _ => Nil
        }
      case '[PolicyMod.ExactByPosition] =>
        if out.length != contract.length then List(s"length mismatch: ${out.length} vs ${contract.length}")
        else
          out.zip(contract).zipWithIndex.collect {
            case (((_,tO),(_,tC)), i) if tO != tC => s"@$i expected $tC, found $tO"
          }
      case '[PolicyMod.ExactOrdered] =>
        if out.length != contract.length then List(s"length mismatch: ${out.length} vs ${contract.length}")
        else
          out.zip(contract).zipWithIndex.flatMap {
            case (((nO,tO),(nC,tC)), i) =>
              val nameOk = if nO == nC then Nil else List(s"@$i(name) expected $nC, found $nO")
              val typeOk = if tO == tC then Nil else List(s"@$i(type) expected $tC, found $tO")
              nameOk ++ typeOk
          }
      case _ => report.errorAndAbort("Unknown policy mod")

    if mismatches.nonEmpty then
      report.errorAndAbort(s"Modifier drift: ${mismatches.mkString("; ")}")
    else '{ () }

Try these:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

import com.example.*

final case class A(id: Long, name: String)
final case class B(name: String, id: Long)

ConformsMod[A, B, PolicyMod.ExactByPosition] // ✅ same types by index
// ConformsMod[A, B, PolicyMod.ExactOrdered] // ❌ order matters → name mismatch

final case class CIa(ID: Long)
final case class CIb(id: Long)
ConformsMod[CIa, CIb, PolicyMod.ExactCI] // ✅ case‑insensitive names match

Why these three? Because real systems need them: - ExactOrdered: For formats where position matters (some CSV parsers, protobuf with field numbers) - ExactCI: Because someone always uses UserId when the contract says userId - ExactByPosition: For legacy systems that ignore field names entirely

Part 11 - Phantom types: Building impossible-to-misuse APIs

Here’s where things get interesting. The code, also uses phantom types, type-indexing / typestate builder patterns to build a pipeline builder that’s impossible to misuse. What are phantom types? They’re type parameters that don’t appear in the class body but control what you can do with an instance. Think of them as compile-time state machines.

stateDiagram-v2
    [*] --> Empty
    Empty --> WithSource: addSource
    WithSource --> WithTransform: transformAs[Next]
    WithTransform --> Complete: addSink[Contract,P] + conforms
    Complete --> [*]: build

    note right of WithTransform
      Evidence constraints:
      - S <: WithSource to transform
      - S <: WithTransform to sink
      - S = Complete to build
    end note

From the codebase SparkCore.scala:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

// Phantom type states
sealed trait BuilderState
sealed trait Empty         extends BuilderState
sealed trait WithSource    extends BuilderState
sealed trait WithTransform extends BuilderState
sealed trait Complete      extends BuilderState

final case class PipelineBuilder[S <: BuilderState, CurContract] private (
  name: String,
  steps: List[PipelineStep]
):
  // Can only add source when Empty
  def addSource[C](src: TypedSource[C])(using SparkSchema[C]): PipelineBuilder[WithSource, C] =
    PipelineBuilder[WithSource, C](name, steps :+ ...)

  // Can only transform after adding source
  def transformAs[Next](f: DataFrame => DataFrame)(using
    ev: S <:< WithSource,  // This constraint enforces order!
    sch: SparkSchema[Next]
  ): PipelineBuilder[WithTransform, Next] =
    PipelineBuilder[WithTransform, Next](name, steps :+ ...)

  // Can only add sink after transform
  def addSink[R, P <: SchemaPolicy](sink: TypedSink[R])(using
    ev0: S <:< WithTransform,  // Must have transform
    ev1: SchemaConforms[CurContract, R, P],  // Compile-time contract check!
    sch: SparkSchema[R]
  ): PipelineBuilder[Complete, CurContract] =
    PipelineBuilder[Complete, CurContract](name, steps :+ ...)

  // Can only build when Complete
  def build(using ev: S =:= Complete): SparkSession => DataFrame =
    (spark: SparkSession) => ...

object PipelineBuilder:
  def apply[CurContract](name: String): PipelineBuilder[Empty, CurContract] =
    PipelineBuilder[Empty, CurContract](name, Nil)

### How this works:

1. State transitions as types - Each method returns a new type parameter S. The compiler tracks which state you’re in.

2. Evidence constraints - ev: S <:< WithSource means “S must be a subtype of WithSource”. If it’s not, this method doesn’t exist.

3. Compile-time state machine - You literally cannot call methods in the wrong order:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// ✅ This compiles
PipelineBuilder[Contract]("good")
  .addSource(src)
  .transformAs[Next](transform)
  .addSink[Contract, SchemaPolicy.Exact](sink)
  .build

// ❌ This doesn't compile - can't transform before adding source
PipelineBuilder[Contract]("bad")
  .transformAs[Next](transform)  // ERROR: No implicit evidence S <:< WithSource

4. Contract checking embedded - Notice ev1: SchemaConforms[CurContract, R, P] in addSink? That’s where the compile-time contract check happens. The pipeline literally won’t build if schemas drift.

This pattern is called the Phantom Type Builder Pattern. For more details, see:

• Compile-safe builder pattern using phantom types (Xebia)
• Phantom Types in Scala (Rhetorical Musings)

Why this matters: In production data pipelines, the order of operations matters. Read → Transform → Validate → Write. With phantom types, the compiler enforces this order. You can’t accidentally write before transforming. You can’t transform without a source.

This is the difference between “here’s how macros work” and “here’s how to build production contract systems."

Part 12 - Schema evolution and versioning: Handling change over time

Here’s the reality: Schemas evolve. Fields get added, deprecated, renamed. Teams roll out changes incrementally. The question isn’t whether schemas will change - it’s how you manage those changes without breaking production.

The schema evolution problem

Scenario: You have CustomerV1 running in production. Marketing wants to add a segment field for targeting. But you have 10 pipelines reading CustomerV1. How do you evolve without breaking everything?

Wrong approach: Add segment to CustomerV1, deploy, hope for the best. Result: Some pipelines break because they don’t handle the new field.

Right approach: Version your contracts and use policies to manage transitions.

Versioning strategy

1
2
3
4
5
6
7

// contracts/CustomerV1.scala - Current production
package contracts
final case class CustomerV1(id: Long, email: String)

// contracts/CustomerV2.scala - New version with segment
package contracts
final case class CustomerV2(id: Long, email: String, segment: Option[String] = None)

Notice segment is: 1. Optional (Option[String]) 2. Has a default (= None)

This makes it backward-compatible - old code can work with new data.

Migration phases

Phase 1: Add the field (Backward policy)

Deploy new producers writing CustomerV2:

1
2
3
4
5

val producer = PipelineBuilder[CustomerV2]("write-v2")
  .addSource(rawData)
  .transformAs[CustomerV2](addSegment)
  .addSink[CustomerV2, Policy.Backward](sink)  // Backward allows optional fields
  .build

Old consumers still read CustomerV1. They ignore the segment field. No breakage.

Phase 2: Migrate consumers

Update each consumer one by one:

1
2
3
4
5

val consumer = PipelineBuilder[CustomerV2]("read-v2")
  .addSource[CustomerV2](source)
  .transformAs[EnrichedCustomer](useSegment)  // Now uses segment field
  .addSink[EnrichedCustomer, Policy.Exact](sink)
  .build

The compile-time contract ensures: “Does this consumer handle the new schema?” If you forget to update the case class, it won’t compile.

Phase 3: Deprecate V1

Once all pipelines use CustomerV2:

1
2
3

// Mark V1 as deprecated
@deprecated("Use CustomerV2", "2025-10-01")
final case class CustomerV1(id: Long, email: String)

The compiler warns any remaining V1 usage. After a grace period, delete V1 entirely.

Handling breaking changes

What if you need to rename a field? Say, email → emailAddress?

Non-breaking approach:

1
2
3
4
5
6
7

// Step 1: Add the new field, keep the old
final case class CustomerV3(
  id: Long,
  email: String,              // Old field
  emailAddress: String,       // New field
  segment: Option[String] = None
)

Wait, that’s duplication. Better:

1
2
3
4
5
6
7
8
9

// Step 1: Add alias constructor
final case class CustomerV3(
  id: Long,
  emailAddress: String,       // New canonical name
  segment: Option[String] = None
)
object CustomerV3:
  def fromV2(v2: CustomerV2): CustomerV3 =
    CustomerV3(v2.id, v2.email, v2.segment)

Step 2: Migrate producers to write emailAddress Step 3: Migrate consumers to read emailAddress Step 4: Remove the fromV2 constructor

At each step, compile-time contracts verify: “Does this transformation produce the expected schema?"

Coexistence during migration

During migration, you have both V2 and V3 running. How to handle?

1
2
3
4
5
6
7
8

sealed trait CustomerSchema
case class V2(id: Long, email: String, segment: Option[String] = None) extends CustomerSchema
case class V3(id: Long, emailAddress: String, segment: Option[String] = None) extends CustomerSchema

// Transformation handles both
def normalize(schema: CustomerSchema): V3 = schema match
  case V2(id, email, segment) => V3(id, email, segment)
  case v3: V3 => v3

The contract checks both paths:

1
2

summon[Conforms[V2, V3Contract, Policy.Backward]]  // V2 → V3 allowed
summon[Conforms[V3, V3Contract, Policy.Exact]]     // V3 → V3 exact match

Real-world migration: Backward → Exact

Let’s look at a complete migration scenario from the code in github. This is exactly how you’d roll out schema changes in production.

From CtdcPoc.scala:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

/** Sink contract: the target schema we promise to write. */
final case class CustomerContract(id: Long, email: String, age: Option[Int] = None)

/** Producer: imagine an upstream source adds an extra field `segment`. */
final case class CustomerProducer(id: Long, email: String, age: Option[Int], segment: String)

/** Declared "Next" schema after a transform (dropping `segment`). */
final case class CustomerNext(id: Long, email: String, age: Option[Int])

// Step 1: Migration phase - use Backward policy
// Producer has extra field, but contract allows it during migration
val srcB = TypedSource[CustomerProducer]("csv", inPath, Map("header" -> "true"))
val sinkB = TypedSink[CustomerContract](tmpOutB)

val dropExtras: DataFrame => DataFrame = _.select($"id", $"email", $"age")

val planB =
  PipelineBuilder[CustomerContract]("CSV -> Parquet B: transformAs[CustomerNext], Exact")
    .addSource(srcB)
    .transformAs[CustomerNext]("drop segment")(dropExtras)
    .addSink[CustomerContract, SchemaPolicy.Exact.type](sinkB)
    .build

### What’s happening here:

Phase 1: Producer adds field - Upstream adds segment field to CustomerProducer - Our contract is still CustomerContract (no segment) - We use transformAs[CustomerNext] to explicitly drop the extra field - Then check CustomerNext conforms to CustomerContract under Exact

Why this works: 1. The transform explicitly declares the output schema (CustomerNext) 2. The compiler checks: Does CustomerNext match CustomerContract? Yes (both have id, email, age) 3. If we later change CustomerNext by accident, compilation fails

Phase 2: Stabilization Once the migration is done:

1
2
3
4
5
6
7

// Later: Everyone uses CustomerContract, no transform needed
val planStable =
  PipelineBuilder[CustomerContract]("stable")
    .addSource(contractSource)
    .noTransform  // Direct pass-through
    .addSink[CustomerContract, SchemaPolicy.Exact.type](sink)
    .build

This is real-world stuff. The code shows you exactly how to migrate schemas safely.

flowchart TD
    P1["Producer adds field 'segment'"] --> T1["transformAs[CustomerNext]\n(drop extras)"]
    T1 --> V1["summon Conforms[CustomerNext,CustomerContract,Policy.Exact]"]
    V1 --> W1["write sink"]

    subgraph Phase 1 ["Migration window"]
      direction TB
      P1 --> T1 --> V1 --> W1
    end

    W1 --> P2["Stabilize: sources match contract"]
    P2 --> V2["Conforms[CustomerContract,CustomerContract,Policy.Exact]"]
    V2 --> W2["noTransform + write"]

Part 13 - Testing compile‑time, for real (copy/paste tests)

How do you test that code fails to compile? With assertDoesNotCompile:

sequenceDiagram
    participant Dev
    participant CI as "GitHub Actions"
    participant SBT as "sbt testOnly *CompileTimeSpec"
    Dev->>CI: open PR
    CI->>SBT: run compile-fail suite
    SBT->>SBT: assertCompiles / assertDoesNotCompile
    alt all pass
        SBT-->>CI: success
        CI-->>Dev: green check
    else drift detected
        SBT-->>CI: failure with errorAndAbort msg
        CI-->>Dev: red X + compiler diff
    end

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

// src/test/scala/com/example/CompileTimeSpec.scala
package com.example
import org.scalatest.funsuite.AnyFunSuite

class CompileTimeSpec extends AnyFunSuite:
  test("Exact should fail when a required field is missing") {
    assertDoesNotCompile("summon[Conforms[OutMissing, Contract, Policy.Exact]]")
  }
  test("Backward should pass for the same mismatch") {
    assertCompiles("summon[Conforms[OutMissing, Contract, Policy.Backward]]")
  }
  test("By‑Position should accept re‑ordered names with same types") {
    assertCompiles("ConformsMod[A, B, PolicyMod.ExactByPosition]")
  }
  test("Ordered should reject the same re‑ordering") {
    assertDoesNotCompile("ConformsMod[A, B, PolicyMod.ExactOrdered]")
  }
  test("Pipeline builder enforces correct order") {
    assertDoesNotCompile("""
      PipelineBuilder[Contract]("bad")
        .transformAs[Next](identity)  // Can't transform without source
    """)
  }
  test("Nested types conform correctly") {
    assertCompiles("summon[Conforms[OrderOut, OrderContract, Policy.Backward]]")
  }

You can wire this into CI:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# .github/workflows/compile-fail.yml (concept)
name: compile-fail
on: [pull_request]
jobs:
  cf:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-java@v4
        with: { distribution: temurin, java-version: '17' }
      - name: Compile‑fail suite
        run: sbt -v "testOnly *CompileTimeSpec"

This is gold. Your CI literally proves that broken schemas can’t be deployed.

Part 14 - Scala 2: how to do the same (and what’s different)

Scala 2 uses def macros with a Context (blackbox/whitebox). The idea is identical, but the implementation uses c.universe trees instead of Expr/quotes.