From ae8a19433a063a941b8802b64a3f4cc290712982 Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 12:10:37 +0200
Subject: [PATCH 1/9] build: make engines-flink resolve on scala 2.13

---
 build.sbt                  | 32 +++++++++++++++++++++++---------
 project/Dependencies.scala | 12 +++++++-----
 2 files changed, 30 insertions(+), 14 deletions(-)

diff --git a/build.sbt b/build.sbt
index 7fe160b8..addbcbb5 100644
--- a/build.sbt
+++ b/build.sbt
@@ -13,10 +13,7 @@ ThisBuild / dynverSeparator := "-"
 ThisBuild / dynverSonatypeSnapshots := true
 // Default Scala stays 2.13 for most modules; Spark/Deequ modules are handled pragmatically via deps.
 ThisBuild / scalaVersion := Dependencies.Versions.scala213
-// Cross-compile defaults: build-level + commands iterate over Scala 2.13 and 3 for speed and stability.
-// Module-specific overrides (e.g., enginesFlink) can still target 2.12 explicitly.
-// Global cross-build: keep 2.13 only for stability and speed.
-// Module overrides handle 2.12 (Flink) and Scala 3 (experimental) explicitly.
+// Global cross-build: keep 2.13 only for stability and speed. `core` additionally builds on Scala 3.
 ThisBuild / crossScalaVersions := Seq(
   Dependencies.Versions.scala213
 )
@@ -171,6 +168,7 @@ lazy val root = (project in file("."))
     connectorsGcs,
     connectorsJdbc,
     enginesSpark,
+    enginesFlink,
     qualityDeequ, // Removed empty quality module per v1.0-2 plan
     examples,
     compileFailTests,
@@ -199,7 +197,7 @@ lazy val infrastructure = moduleProject("infrastructure")
 lazy val core = moduleProject("core")
   .settings(
     description := "Core abstractions and custom type system",
-    // Core is the only module published for both. Flink-specific modules handle 2.12 separately.
+    // Core is the only module published for both.
     crossScalaVersions := Seq(Dependencies.Versions.scala213, Dependencies.Versions.scala3),
     libraryDependencies ++= Dependencies.forModule("core"),
     // Minimal, justified excludes only
@@ -317,10 +315,25 @@ lazy val enginesSpark = moduleProject("engines-spark")
 lazy val enginesFlink = moduleProject("engines-flink")
   .dependsOn(core, connectors, enginesSpark % "test->compile")
   .settings(
-    description        := "Apache Flink execution engine",
-    // Flink Scala API is 2.12-only; avoid invalid 2.13 cross build
-    crossScalaVersions := Seq(Dependencies.Versions.scala212),
+    description := "Apache Flink execution engine",
     libraryDependencies ++= Dependencies.forModule("engines-flink"),
+    // Flink serializes records with Kryo, which reflects into `java.base`. On Java 17 and later that is
+    // closed by default, and the failure arrives as an `InaccessibleObjectException` from inside a running
+    // job rather than at startup. These are the opens Flink's own scripts set for the same reason.
+    Test / javaOptions ++= Seq(
+      "--add-opens=java.base/java.lang=ALL-UNNAMED",
+      "--add-opens=java.base/java.lang.invoke=ALL-UNNAMED",
+      "--add-opens=java.base/java.io=ALL-UNNAMED",
+      "--add-opens=java.base/java.net=ALL-UNNAMED",
+      "--add-opens=java.base/java.nio=ALL-UNNAMED",
+      "--add-opens=java.base/java.text=ALL-UNNAMED",
+      "--add-opens=java.base/java.time=ALL-UNNAMED",
+      "--add-opens=java.base/java.util=ALL-UNNAMED",
+      "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED",
+      "--add-opens=java.base/java.util.concurrent.atomic=ALL-UNNAMED",
+      "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED",
+    ),
+    Test / fork := true,
   )
   .settings(mimaSettings("engines-flink"): _*)
 
@@ -537,7 +550,7 @@ lazy val experimental = moduleProject("experimental")
 import sbtunidoc.ScalaUnidocPlugin
 import sbtunidoc.ScalaUnidocPlugin.autoImport._
 
-// Only aggregate Scala 2.13 modules (Flink is 2.12-only). This keeps unidoc stable.
+// Only aggregate the Scala 2.13 modules. This keeps unidoc stable.
 lazy val unidocProjects = Seq(
   core,
   contracts,
@@ -545,6 +558,7 @@ lazy val unidocProjects = Seq(
   connectorsGcs,
   connectorsJdbc,
   enginesSpark,
+  enginesFlink,
   qualityDeequ,
   infrastructure,
   examples,
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index 82f8d2d4..4297b349 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -6,7 +6,6 @@ object Dependencies {
   // ===== VERSION CATALOG =====
   object Versions {
     // Scala ecosystem
-    val scala212 = "2.12.20"
     val scala213 = "2.13.16"
     val scala3   = "3.3.3"
 
@@ -135,11 +134,14 @@ object Dependencies {
       "io.delta"         %% "delta-spark" % Versions.delta,
     )
 
+    // Flink's Java artifacts carry no Scala suffix, so they resolve on whatever Scala version the module is
+    // built for. The Scala API (`flink-scala`, `flink-streaming-scala`) is published for 2.12 only and is
+    // deprecated upstream; depending on it pinned this module to 2.12, which `core` does not publish for, so
+    // the module could not resolve at all.
     val flink = Seq(
-      // NOTE: Flink Scala API only available for 2.12, documented constraint
-      "org.apache.flink" % "flink-scala_2.12"           % Versions.flink % "provided",
-      "org.apache.flink" % "flink-streaming-scala_2.12" % Versions.flink % "provided",
-      "org.apache.flink" % "flink-table-runtime"        % Versions.flink % "provided",
+      "org.apache.flink" % "flink-streaming-java" % Versions.flink % "provided",
+      // The local MiniCluster `env.execute` runs on. Only the tests start one; a cluster supplies it.
+      "org.apache.flink" % "flink-clients" % Versions.flink % Test,
     )
 
     val all: Seq[ModuleID] = spark ++ flink

From 630f5c57a897fe7bb864d56c77a2a6fcb44a8a44 Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 12:10:38 +0200
Subject: [PATCH 2/9] feat(flink): run data operations as flink batch jobs

---
 .../engines/flink/FlinkDataAlgebra.scala      | 277 ++++++++++++++++--
 .../engines/flink/FlinkDataset.scala          |  96 ++++++
 .../flowforge/engines/flink/FlinkRows.scala   | 108 +++++++
 .../engines/flink/FlinkStreamOps.scala        | 206 +++++++++++++
 .../com/flowforge/engines/flink/package.scala |  24 ++
 5 files changed, 687 insertions(+), 24 deletions(-)
 create mode 100644 modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataset.scala
 create mode 100644 modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala
 create mode 100644 modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala
 create mode 100644 modules/engines-flink/src/main/scala/com/flowforge/engines/flink/package.scala

diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataAlgebra.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataAlgebra.scala
index a8e4f825..426f90e9 100644
--- a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataAlgebra.scala
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataAlgebra.scala
@@ -1,6 +1,6 @@
 package com.flowforge.engines.flink
 
-import cats.data.{ NonEmptyList, ValidatedNel }
+import cats.data.{ NonEmptyList, Validated, ValidatedNel }
 import com.flowforge.core.algebra.DataAlgebra._
 import com.flowforge.core.algebra._
 import com.flowforge.core.types.PipelineTypes.{ DataContract => PDataContract, QualityCheck }
@@ -10,11 +10,20 @@ import com.flowforge.core.types._
 import java.time.Instant
 
 /**
- * `DataAlgebra` for the Flink module. It runs no Flink: every method forwards to `InMemoryDataAlgebra`, so
- * the work happens on the driver heap and nothing is submitted to a Flink cluster.
+ * `DataAlgebra` backed by Flink batch jobs.
  *
- * It exists so a pipeline written against the algebra can be named for Flink and run, which is a placeholder
- * for an implementation rather than one. See `docs/plan/v1.0-readiness.md` for where that stands.
+ * A dataset this engine produces is a [[FlinkDataset]]: a plan for the records rather than the records, so
+ * nothing is held on the driver beyond a bounded sample. `read`, `write`, `filter`, `map`, `flatMap`,
+ * `union`, `validate` and `runQualityChecks` run in Flink and see every record.
+ *
+ * The rest of the algebra still forwards to `InMemoryDataAlgebra`, which reads `dataset.data`. On a
+ * Flink-backed dataset that is the sample, so `groupBy`, `join`, `sortBy`, `take`, `drop`, `profile`, the CDC
+ * operations and the table operations answer for at most [[FlinkDataset.sampleSize]] records. They are listed
+ * in `docs/plan/v1.0-readiness.md` rather than hidden here. A dataset from some other engine is held by the
+ * driver in full, so forwarding it is exact.
+ *
+ * The one thing this engine does not do is stream. Every source it reads is a bounded file and every job runs
+ * in batch mode, for the reason given on `FlinkStreamOps.batchEnv`.
  */
 final class FlinkDataAlgebra[F[_]](implicit F: EffectSystem[F]) extends DataAlgebra[F] {
 
@@ -24,24 +33,117 @@ final class FlinkDataAlgebra[F[_]](implicit F: EffectSystem[F]) extends DataAlge
     Set(Capability.Read, Capability.Write, Capability.QualityChecks)
 
   // ---------- External IO ----------
-  override def read[A: DataDecoder](source: DataSource): F[Dataset[A]] =
-    delegate.read(source)
+
+  /**
+   * The plan for a local file, or None for a format this engine cannot read.
+   *
+   * Reading the CSV header is an action, so this is one too, and the caller runs it inside `F`. Parquet and
+   * Delta are absent because reading them needs a connector this module does not depend on; saying so beats
+   * reading them wrong.
+   */
+  private def planFor(local: LocalDataSource): Option[FlinkPlan] = local.format match {
+    case DataFormat.CSV =>
+      val header = {
+        val lines = scala.io.Source.fromFile(local.location, "UTF-8")
+        try
+          lines
+            .getLines().find(_.trim.nonEmpty).fold(List.empty[String])(
+              _.trim.split(",", -1).toList.map(_.trim),
+            )
+        finally lines.close()
+      }
+      Some(FlinkStreamOps.readCsv(local.location, header))
+    case DataFormat.JSON | DataFormat.JSONL => Some(FlinkStreamOps.readJsonLines(local.location))
+    case _                                  => None
+  }
+
+  override def read[A: DataDecoder](source: DataSource): F[Dataset[A]] = {
+    val result: F[Dataset[A]] = source match {
+      case local: LocalDataSource =>
+        F.flatMap(F.blocking(planFor(local))) {
+          case None =>
+            F.raiseError[Dataset[A]](
+              new UnsupportedOperationException(s"Format ${local.format} not supported"),
+            )
+          case Some(plan) =>
+            F.blocking(FlinkDataset.fromPlan[A](plan, Some(source)))
+        }
+
+      case other =>
+        F.raiseError[Dataset[A]](
+          new UnsupportedOperationException(
+            s"DataSource type ${other.getClass.getSimpleName} not supported",
+          ),
+        )
+    }
+
+    // The Spark engine reports a read failure as the framework's error type. Reporting it the same way here
+    // means a caller that handles one engine's read failure handles both.
+    F.handleErrorWith(result) { error =>
+      F.raiseError(
+        DataProcessingError.ProcessingFailure(
+          stepName = "flink-read",
+          reason = error.getMessage,
+          message = s"Failed to read from ${source.getClass.getSimpleName}",
+          cause = Some(error),
+        ),
+      )
+    }
+  }
 
   override def readWithSchema[A: DataDecoder](
     source: DataSource,
     expectedSchema: DataSchema,
   ): F[ValidatedNel[FlowForgeError, Dataset[A]]] =
-    delegate.readWithSchema(source, expectedSchema)
+    F.map(read[A](source))(dataset => Validated.validNel(dataset))
 
+  /**
+   * One batch read, handed back as a stream of a single chunk. Nothing here is a Flink streaming job: see the
+   * class comment.
+   */
   override def stream[A: DataDecoder](source: DataSource): F[DataStream[F, A]] =
-    delegate.stream(source)
-
+    F.pure(new DataStream[F, A] {
+      def chunks: F[List[Dataset[A]]] = F.map(read[A](source))(dataset => List(dataset))
+    })
+
+  /**
+   * Write every record, one per line, through a Flink sink.
+   *
+   * Only a local JSON sink is supported. The records are re-encoded with the caller's encoder rather than
+   * passed through as the plan's own JSON, so what lands on disk is what the encoder says a record looks
+   * like. CSV is not supported: a dataflow sink cannot order a header line ahead of the rows it writes, and
+   * writing the rows without the header would produce a file this engine could not read back.
+   *
+   * The record count is the one the dataset already established, so reporting it starts no extra job.
+   */
   override def write[A: DataEncoder](
     dataset: Dataset[A],
     sink: DataSink,
     options: WriteOptions = WriteOptions.default,
-  ): F[WriteResult] =
-    delegate.write(dataset, sink, options)
+  ): F[WriteResult] = (dataset, sink) match {
+    case (fd: FlinkDataset[A], local: LocalDataSink)
+        if local.format == DataFormat.JSON || local.format == DataFormat.JSONL =>
+      F.blocking {
+        val encoded = FlinkStreamOps.encodeRows[A](fd.plan, local.format)(fd.decoder, DataEncoder[A])
+        FlinkStreamOps.writeLines(encoded, local.location)
+        val bytes = java.nio.file.Files.size(java.nio.file.Paths.get(local.location))
+        WriteResult(
+          recordsWritten = fd.metadata.recordCount,
+          partitionsWritten = 1,
+          bytesWritten = bytes,
+          success = true,
+        )
+      }
+
+    case (_: FlinkDataset[A], local: LocalDataSink) =>
+      F.raiseError[WriteResult](
+        new UnsupportedOperationException(s"Format ${local.format} not supported by the Flink sink"),
+      )
+
+    case _ =>
+      // Not a dataset this engine produced, so the driver holds it in full and the in-memory writer is exact.
+      delegate.write(dataset, sink, options)
+  }
 
   override def writeWithValidation[A: DataEncoder](
     dataset: Dataset[A],
@@ -52,17 +154,33 @@ final class FlinkDataAlgebra[F[_]](implicit F: EffectSystem[F]) extends DataAlge
     delegate.writeWithValidation(dataset, sink, contract, options)
 
   // ---------- Pure transformations ----------
+
+  // These three append an operator to the plan and rebuild the dataset around it. Appending is a calculation;
+  // the job that follows is what `FlinkDataset.fromPlan` runs to resample and count. Reading `dataset.data`
+  // instead, as the forwarding version did, narrowed the sample and left the records alone, so a filter
+  // followed by a write wrote every record.
   override def filter[A: DataDecoder](dataset: Dataset[A], predicate: A => Boolean): Dataset[A] =
-    delegate.filter(dataset, predicate)
+    dataset match {
+      case fd: FlinkDataset[A] => FlinkDataset.withPlan[A](fd, FlinkStreamOps.filter[A](fd.plan, predicate))
+      case other               => delegate.filter(other, predicate)
+    }
 
   override def map[A: DataDecoder, B: DataEncoder: DataDecoder](dataset: Dataset[A], f: A => B): Dataset[B] =
-    delegate.map(dataset, f)
+    dataset match {
+      case fd: FlinkDataset[A] =>
+        FlinkDataset.fromPlan[B](FlinkStreamOps.mapRows[A, B](fd.plan, a => List(f(a))), fd.metadata.source)
+      case other => delegate.map(other, f)
+    }
 
   override def flatMap[A: DataDecoder, B: DataEncoder: DataDecoder](
     dataset: Dataset[A],
     f: A => Dataset[B],
   ): Dataset[B] =
-    delegate.flatMap(dataset, f)
+    dataset match {
+      case fd: FlinkDataset[A] =>
+        FlinkDataset.fromPlan[B](FlinkStreamOps.mapRows[A, B](fd.plan, a => f(a).data), fd.metadata.source)
+      case other => delegate.flatMap(other, f)
+    }
 
   override def groupBy[A, K, V: DataEncoder](
     dataset: Dataset[A],
@@ -80,8 +198,13 @@ final class FlinkDataAlgebra[F[_]](implicit F: EffectSystem[F]) extends DataAlge
   ): Dataset[C] =
     delegate.join(left, right, leftKey, rightKey, combiner)
 
-  override def union[A](left: Dataset[A], right: Dataset[A]): Dataset[A] =
-    delegate.union(left, right)
+  // `union` carries no `DataDecoder[A]`, so the one the left dataset was read with is what rebuilds the
+  // result. This is why a `FlinkDataset` keeps its decoder.
+  override def union[A](left: Dataset[A], right: Dataset[A]): Dataset[A] = (left, right) match {
+    case (lf: FlinkDataset[A], rf: FlinkDataset[A]) =>
+      FlinkDataset.fromPlan[A](FlinkStreamOps.union(lf.plan, rf.plan), lf.metadata.source)(lf.decoder)
+    case _ => delegate.union(left, right)
+  }
 
   override def sortBy[A, K: Ordering](dataset: Dataset[A], keyExtractor: A => K): Dataset[A] =
     delegate.sortBy(dataset, keyExtractor)
@@ -129,17 +252,90 @@ final class FlinkDataAlgebra[F[_]](implicit F: EffectSystem[F]) extends DataAlge
   override def queryLineage(query: LineageQuery): F[List[LineageRecord]] =
     delegate.queryLineage(query)
 
+  /**
+   * Apply the contract to every record.
+   *
+   * The result is built by `DataAlgebra.contractResult`, the same function the other engines call, so what
+   * `passed`, `score` and `recordsAffected` mean here is what they mean there.
+   *
+   * Two jobs run, for the two halves of the answer. The count of violating records is exact, because that is
+   * what decides `passed` and `score`. The list of distinct messages is capped, because a message built from
+   * a record's own values is otherwise as numerous as the dataset. Capping messages cannot turn a failing
+   * contract into a passing one.
+   */
   override def validate[A: DataDecoder](
     dataset: Dataset[A],
     contract: PDataContract[A],
-  ): F[QualityResult[Dataset[A]]] =
-    delegate.validate(dataset, contract)
-
+  ): F[QualityResult[Dataset[A]]] = {
+
+    /** The messages for the violations the contract reports on one record, empty when it accepts it. */
+    val messagesOf: A => List[String] = a => contract(a).fold(_.toList.map(_.message), _ => Nil)
+
+    dataset match {
+      case fd: FlinkDataset[A] =>
+        F.blocking {
+          val violating =
+            FlinkStreamOps.count(FlinkStreamOps.filter[A](fd.plan, a => messagesOf(a).nonEmpty))
+          val messageCounts = FlinkStreamOps.countKeys[A](fd.plan, messagesOf)
+          DataAlgebra.contractResult(dataset, fd.metadata.recordCount, violating, messageCounts)
+        }
+
+      case other => delegate.validate(other, contract)
+    }
+  }
+
+  /**
+   * Apply every check to every record.
+   *
+   * Two jobs again, and the split is the same one for the same reason. The first counts how many records each
+   * check rejected, keyed by the check's index: that key set is as large as `checks`, so it fits and the set
+   * of failed checks is exact. The second collects messages, capped, so a dataset of violations cannot fill
+   * the driver. A check can therefore be reported as failed with no message to show.
+   */
   override def runQualityChecks[A: DataDecoder](
     dataset: Dataset[A],
     checks: NonEmptyList[QualityCheck[A]],
-  ): F[List[QualityCheckResult]] =
-    delegate.runQualityChecks(dataset, checks)
+  ): F[List[QualityCheckResult]] = {
+    val messagesPerCheck: List[A => List[String]] =
+      checks.toList.map(chk => (a: A) => chk(a).fold(_.toList.map(_.message), _ => Nil))
+
+    /** The indexes of the checks that rejected one record. */
+    val failedIndexesOf: A => List[String] = a =>
+      messagesPerCheck.zipWithIndex.collect { case (check, idx) if check(a).nonEmpty => idx.toString }
+
+    /** One key per violation, carrying both the check it came from and its message. */
+    val indexedMessagesOf: A => List[String] = a =>
+      messagesPerCheck.zipWithIndex.flatMap {
+        case (check, idx) => check(a).map(message => CheckKey.of(idx, message))
+      }
+
+    dataset match {
+      case fd: FlinkDataset[A] =>
+        F.blocking {
+          val failed =
+            FlinkStreamOps.countKeys[A](fd.plan, failedIndexesOf, messagesPerCheck.size).map(_._1).toSet
+          val messagesByIndex = FlinkStreamOps
+            .countKeys[A](fd.plan, indexedMessagesOf)
+            .map(_._1)
+            .groupBy(CheckKey.indexOf)
+
+          messagesPerCheck.indices.toList.map { idx =>
+            if (!failed.contains(idx.toString))
+              QualityCheckResult(s"check_$idx", passed = true, message = "ok", score = 1.0)
+            else {
+              val reported = messagesByIndex.getOrElse(idx.toString, Nil).map(CheckKey.messageOf)
+              val message = reported.take(3) match {
+                case Nil   => "violations found, messages not collected"
+                case shown => shown.mkString("; ")
+              }
+              QualityCheckResult(s"check_$idx", passed = false, message = message, score = 0.0)
+            }
+          }
+        }
+
+      case other => delegate.runQualityChecks(other, checks)
+    }
+  }
 
   override def profile[A](dataset: Dataset[A]): F[DataProfile[A]] =
     delegate.profile(dataset)
@@ -199,10 +395,43 @@ final class FlinkDataAlgebra[F[_]](implicit F: EffectSystem[F]) extends DataAlge
     delegate.vacuumTable(table, retentionHours, dryRun)
 
   // ---------- Utilities ----------
-  override def count[A](dataset: Dataset[A]): Long      = delegate.count(dataset)
-  override def isEmpty[A](dataset: Dataset[A]): Boolean = delegate.isEmpty(dataset)
+
+  // A `FlinkDataset` established its record count when it was built, so this is the whole dataset's count and
+  // not the sample's, and asking for it starts no job. The frame is not re-counted here because `count` is
+  // declared outside `F`: running a job from it would make a method the interface calls pure an action.
+  //
+  // `size` is not used, because it is an `Int` and this returns a `Long`, so a dataset of more than
+  // `Int.MaxValue` records would come back negative.
+  override def count[A](dataset: Dataset[A]): Long = dataset match {
+    case fd: FlinkDataset[A] => fd.metadata.recordCount
+    case other               => other.size.toLong
+  }
+  override def isEmpty[A](dataset: Dataset[A]): Boolean = dataset.isEmpty
   override def cache[A](dataset: Dataset[A], strategy: CacheStrategy): F[Dataset[A]] =
     delegate.cache(dataset, strategy)
   override def partition[A](dataset: Dataset[A], partitioner: Partitioner[A]): List[Dataset[A]] =
     delegate.partition(dataset, partitioner)
 }
+
+/**
+ * One key carrying both the check a violation came from and the violation's message.
+ *
+ * `runQualityChecks` needs per-check counts and per-message counts, and a single counting job can only count
+ * one kind of key, so the two facts are joined into one key and taken apart again on the driver. The
+ * separator is a control character because a violation message is human-readable text, so one cannot appear
+ * inside a message and split a key in the wrong place.
+ *
+ * This is a top-level object rather than a field of the algebra on purpose: the function that builds these
+ * keys travels to a Flink task, and a field would make it carry the whole algebra along, which is not
+ * serializable.
+ */
+private[flink] object CheckKey {
+
+  private val separator: Char = 1.toChar
+
+  def of(checkIndex: Int, message: String): String = s"$checkIndex$separator$message"
+
+  def indexOf(key: String): String = key.takeWhile(_ != separator)
+
+  def messageOf(key: String): String = key.dropWhile(_ != separator).drop(1)
+}
diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataset.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataset.scala
new file mode 100644
index 00000000..5e7be475
--- /dev/null
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataset.scala
@@ -0,0 +1,96 @@
+package com.flowforge.engines.flink
+
+import com.flowforge.core.algebra.{ DataAlgebra, DataDecoder }
+import com.flowforge.core.types.{ DataSchema, DataSource }
+
+import java.time.Instant
+
+/**
+ * A dataset this engine produced: a plan for the records, plus a sample of them.
+ *
+ * Nothing here holds the dataset. `DataAlgebra.Dataset` requires `data: List[A]`, so a Flink-backed dataset
+ * has to answer that question with something, and the honest answer is a bounded sample. `size`, `isEmpty`
+ * and the algebra's `count` answer from the plan instead, so a caller that wants a fact about the whole
+ * dataset gets one.
+ *
+ * The decoder the records were read with is carried along, because the plan holds JSON rows and every
+ * operation that has to see a record has to decode one. `DataAlgebra.write` is handed an encoder and no
+ * decoder, so without this field the engine could not re-encode a row for a sink.
+ *
+ * @param sampleData
+ *   at most [[FlinkDataset.sampleSize]] records, for the `data` the algebra's interface asks for
+ * @param plan
+ *   how to rebuild the stream of every record, in a fresh environment
+ */
+final case class FlinkDataset[A](
+  sampleData: List[A],
+  plan: FlinkPlan,
+  decoder: DataDecoder[A],
+  schema: DataSchema,
+  metadata: DataAlgebra.DatasetMetadata)
+    extends DataAlgebra.Dataset[A] {
+
+  /** A sample, not the dataset. See the class comment. */
+  override def data: List[A] = sampleData
+
+  /**
+   * The whole dataset's record count, not the sample's.
+   *
+   * It is established once, when the dataset is built, so asking again starts no job. `Int` is the
+   * interface's type, so a dataset of more than `Int.MaxValue` records overflows it; the algebra's `count`
+   * returns the same number as a `Long`.
+   */
+  override def size: Int = metadata.recordCount.toInt
+
+  override def isEmpty: Boolean = metadata.recordCount == 0L
+}
+
+object FlinkDataset {
+
+  /** How many records the driver keeps, matching what the Spark engine samples. */
+  val sampleSize: Int = 100
+
+  /**
+   * Build a dataset around a plan.
+   *
+   * One job runs, for the sample. The record count comes from the sample when the sample turned out to be the
+   * whole dataset: collecting fewer rows than the limit means there were no more to collect, so a second job
+   * would only confirm a number already in hand. A dataset larger than the sample pays for that second job.
+   *
+   * @param source
+   *   where the records came from, which a plan does not carry
+   */
+  def fromPlan[A](plan: FlinkPlan, source: Option[DataSource] = None)(implicit
+    decoder: DataDecoder[A],
+  ): FlinkDataset[A] = {
+    val sampleRows = FlinkStreamOps.collect(plan, sampleSize)
+    val sample     = sampleRows.flatMap(row => FlinkRows.decode(row, decoder))
+    val recordCount =
+      if (sampleRows.sizeIs < sampleSize) sampleRows.size.toLong else FlinkStreamOps.count(plan)
+    val schema = FlinkRows.schemaOf(sampleRows)
+
+    FlinkDataset(
+      sampleData = sample,
+      plan = plan,
+      decoder = decoder,
+      schema = schema,
+      metadata = DataAlgebra.DatasetMetadata(
+        recordCount = recordCount,
+        schema = schema,
+        partitions = 1,
+        createdAt = Instant.now(),
+        source = source,
+      ),
+    )
+  }
+
+  /**
+   * Rebuild a dataset around the plan an operation produced.
+   *
+   * The sample, the schema and the record count all describe what the dataset now holds rather than what it
+   * held before the operation. `metadata.source` is carried over, because it names where the records came
+   * from and an operation does not change that.
+   */
+  def withPlan[A: DataDecoder](base: FlinkDataset[A], plan: FlinkPlan): FlinkDataset[A] =
+    fromPlan[A](plan, base.metadata.source)
+}
diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala
new file mode 100644
index 00000000..9cf17253
--- /dev/null
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala
@@ -0,0 +1,108 @@
+package com.flowforge.engines.flink
+
+import com.flowforge.core.algebra.{ DataDecoder, DataEncoder, EncodedData }
+import com.flowforge.core.types.RefinedTypes.{ FieldName, SchemaVersion }
+import com.flowforge.core.types.{ DataFormat, DataSchema, DataType, StructField }
+import io.circe.Json
+
+import java.nio.charset.StandardCharsets
+import java.time.Instant
+
+/**
+ * Everything this engine does to a single JSON row, with no environment and no job involved.
+ *
+ * These are all calculations, so they can be tested without starting Flink, and the Flink function classes in
+ * [[FlinkStreamOps]] can call them from inside a task. They are reached through this object's static instance
+ * rather than captured, which is what keeps those function classes serializable.
+ */
+private[flink] object FlinkRows {
+
+  /** Decode one JSON row. `None` when the decoder rejects it. */
+  def decode[A](json: String, decoder: DataDecoder[A]): Option[A] =
+    decoder
+      .decode(EncodedData(json.getBytes(StandardCharsets.UTF_8), DataFormat.JSON), DataFormat.JSON)
+      .toOption
+
+  /** Encode one record in `format`. `None` when the encoder rejects it. */
+  def encode[A](value: A, encoder: DataEncoder[A], format: DataFormat): Option[String] =
+    encoder.encode(value, format).toOption.map(ed => new String(ed.data, StandardCharsets.UTF_8))
+
+  /**
+   * Read one CSV line as a JSON object, under the column names the file's header line gave.
+   *
+   * `None` for a blank line and for the header line itself, because the source hands every line of the file
+   * to the same function and the header is not a record.
+   *
+   * A line with fewer cells than the header leaves the remaining fields null. Extra cells are dropped: the
+   * header is what names a column, so a cell with no name has nowhere to go.
+   */
+  def csvLineToJson(header: List[String], line: String): Option[String] = {
+    val trimmed = line.trim
+    if (trimmed.isEmpty || trimmed == header.mkString(",")) None
+    else {
+      val cells  = trimmed.split(",", -1).toList
+      val fields = header.zipAll(cells, "", "").collect { case (name, cell) if name.nonEmpty => name -> cell }
+      Some(Json.obj(fields.map { case (name, cell) => name -> cellToJson(cell) }: _*).noSpaces)
+    }
+  }
+
+  /**
+   * A CSV cell carries no type, so it is read the way a CSV reader with schema inference reads it.
+   *
+   * This matters for decoding rather than for display: a decoder for a record with an `Int` field rejects
+   * `{"id":"1"}` and accepts `{"id":1}`, so quoting every cell would make every typed record undecodable.
+   */
+  private def cellToJson(cell: String): Json = {
+    val value = cell.trim
+    if (value.isEmpty) Json.Null
+    else if (value == "true") Json.True
+    else if (value == "false") Json.False
+    else
+      value.toLongOption
+        .map(Json.fromLong)
+        .orElse(value.toDoubleOption.flatMap(Json.fromDouble))
+        .getOrElse(Json.fromString(cell))
+  }
+
+  /**
+   * The schema of a dataset, read off the rows it starts with.
+   *
+   * A stream of JSON rows carries no schema of its own, so the only thing that can describe one is the rows
+   * themselves. The sample is what the caller already paid for, so this costs no extra job. A field absent
+   * from every sampled row is absent from the schema, which is the same limit any schema-on-read inference
+   * has.
+   */
+  def schemaOf(rows: List[String]): DataSchema = {
+    val fields = rows.headOption
+      .flatMap(row => io.circe.parser.parse(row).toOption)
+      .flatMap(_.asObject)
+      .toList
+      .flatMap(_.toList)
+      .map {
+        case (name, value) =>
+          StructField(FieldName.unsafeFrom(name), typeOf(value), nullable = value.isNull)
+      }
+    DataSchema(
+      fields = fields,
+      version = SchemaVersion.unsafeFrom(1),
+      metadata = Map("inferred_from" -> "flink_json_sample"),
+      createdAt = Instant.now(),
+    )
+  }
+
+  /** The FlowForge type of one JSON value. A null carries no type, so it is reported as a string. */
+  private def typeOf(value: Json): DataType =
+    value.fold(
+      jsonNull = DataType.String,
+      jsonBoolean = _ => DataType.Boolean,
+      jsonNumber = number => if (number.toLong.isDefined) DataType.Long else DataType.Double,
+      jsonString = _ => DataType.String,
+      jsonArray = elements => DataType.Array(elements.headOption.fold[DataType](DataType.String)(typeOf)),
+      jsonObject = obj =>
+        DataType.Struct(
+          obj.toList.map {
+            case (name, field) => StructField(FieldName.unsafeFrom(name), typeOf(field), nullable = field.isNull)
+          },
+        ),
+    )
+}
diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala
new file mode 100644
index 00000000..007043b9
--- /dev/null
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala
@@ -0,0 +1,206 @@
+package com.flowforge.engines.flink
+
+import com.flowforge.core.algebra.{ DataDecoder, DataEncoder }
+import com.flowforge.core.types.DataFormat
+import org.apache.flink.api.common.RuntimeExecutionMode
+import org.apache.flink.api.common.functions.{ FilterFunction, FlatMapFunction }
+import org.apache.flink.api.common.typeinfo.Types
+import org.apache.flink.api.java.functions.KeySelector
+import org.apache.flink.api.java.tuple.Tuple2
+import org.apache.flink.api.java.typeutils.TupleTypeInfo
+import org.apache.flink.core.fs.FileSystem
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
+import org.apache.flink.util.Collector
+
+import scala.annotation.nowarn
+import scala.jdk.CollectionConverters._
+
+/**
+ * Record-level operations run inside a Flink job rather than on the driver, and the plans that describe them.
+ *
+ * The plan combinators are calculations: a [[FlinkPlan]] in, a plan out, no job submitted. The four methods
+ * below them are the only actions in this module, and each one builds its own environment, applies the plan
+ * and runs exactly one job.
+ *
+ * The caller's function and the codecs travel to the task that runs them, which is why `DataDecoder` and
+ * `DataEncoder` are `Serializable` and why the function classes here are named rather than anonymous: a named
+ * top-level class holds only the fields it declares, so there is no enclosing instance to serialize.
+ *
+ * A record the decoder rejects is dropped, for the same reason it is in `SparkFrameOps`: a pure operation that
+ * returns a `Dataset` has nowhere to report the failure.
+ */
+private[flink] object FlinkStreamOps {
+
+  /** How many distinct keys [[countKeys]] brings back. See [[violationMessageLimit]]. */
+  val violationMessageLimit: Int = 1000
+
+  /** The type of the `(key, count)` pair the counting job carries. Flink needs it stated, not inferred. */
+  private val pairType: TupleTypeInfo[Tuple2[String, java.lang.Long]] =
+    new TupleTypeInfo[Tuple2[String, java.lang.Long]](Types.STRING, Types.LONG)
+
+  // ---------- Plans: calculations ----------
+
+  /** Read a file of JSON rows, one record per line. */
+  @nowarn("cat=deprecation")
+  def readJsonLines(path: String): FlinkPlan = env => env.readTextFile(path)
+
+  /**
+   * Read a CSV file as JSON rows, under the given column names.
+   *
+   * The header is a driver-side fact: it has to be known before the job is built, because it names the fields
+   * every row will carry. The job itself still reads the whole file.
+   */
+  @nowarn("cat=deprecation")
+  def readCsv(path: String, header: List[String]): FlinkPlan =
+    env => env.readTextFile(path).flatMap(new CsvLineToJson(header), Types.STRING)
+
+  /** Keep the rows whose decoded record satisfies the predicate. */
+  def filter[A](plan: FlinkPlan, predicate: A => Boolean)(implicit decoder: DataDecoder[A]): FlinkPlan = {
+    val function = new DecodedFilter[A](decoder, predicate)
+    env => plan(env).filter(function)
+  }
+
+  /** Apply `f` to every decoded record and carry the results on as JSON rows. */
+  def mapRows[A, B](plan: FlinkPlan, f: A => List[B])(implicit
+    decoder: DataDecoder[A],
+    encoder: DataEncoder[B],
+  ): FlinkPlan = {
+    val function = new DecodedFlatMap[A, B](decoder, encoder, f)
+    env => plan(env).flatMap(function, Types.STRING)
+  }
+
+  /** Both plans build in the one environment, so their streams can be joined into a single stream. */
+  def union(left: FlinkPlan, right: FlinkPlan): FlinkPlan = env => left(env).union(right(env))
+
+  /** Re-encode every record in `format`, for a sink that wants something other than the plan's JSON. */
+  def encodeRows[A](plan: FlinkPlan, format: DataFormat)(implicit
+    decoder: DataDecoder[A],
+    encoder: DataEncoder[A],
+  ): FlinkPlan = {
+    val function = new ReEncode[A](decoder, encoder, format)
+    env => plan(env).flatMap(function, Types.STRING)
+  }
+
+  // ---------- Jobs: actions ----------
+
+  /**
+   * A fresh environment in batch mode.
+   *
+   * Every source this engine reads is a bounded file, and batch mode is what makes the aggregation in
+   * [[countKeys]] report one final count per key instead of a running count per element. The trade is that
+   * nothing here can consume an unbounded source; this engine runs Flink batch jobs, not Flink streaming.
+   */
+  private def batchEnv(): StreamExecutionEnvironment = {
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    env.setRuntimeMode(RuntimeExecutionMode.BATCH)
+    env
+  }
+
+  /** Run the plan and bring back at most `limit` rows. */
+  def collect(plan: FlinkPlan, limit: Int): List[String] =
+    plan(batchEnv()).executeAndCollect(limit).asScala.toList
+
+  /**
+   * Run the plan and count its rows.
+   *
+   * The rows are counted as they arrive and never held, so this is bounded whatever the dataset's size.
+   */
+  def count(plan: FlinkPlan): Long = {
+    val rows = plan(batchEnv()).executeAndCollect()
+    try rows.asScala.length.toLong
+    finally rows.close()
+  }
+
+  /**
+   * Count the records behind each distinct key the caller's function reports.
+   *
+   * A key is counted once per record: a record that reports the same key twice has still broken that rule
+   * once, which is what the count claims to say.
+   *
+   * At most `limit` keys come back, because a key built from a record's own values is otherwise as numerous as
+   * the dataset. Which keys survive the cap is not specified. The Spark engine orders by key before capping,
+   * so it truncates deterministically; a `DataStream` has no cheap global sort, so this does not. A caller
+   * that needs an exact answer over an unbounded key set needs a count, not a key list, and [[count]] gives
+   * one.
+   */
+  def countKeys[A](
+    plan: FlinkPlan,
+    keysOf: A => List[String],
+    limit: Int = violationMessageLimit,
+  )(implicit decoder: DataDecoder[A],
+  ): List[(String, Long)] = {
+    val perRecord = new KeysOfRecord[A](decoder, keysOf)
+    val counted = plan(batchEnv())
+      .flatMap(perRecord, pairType)
+      .keyBy(new FirstField, Types.STRING)
+      .sum(1)
+    counted
+      .executeAndCollect(limit)
+      .asScala
+      .toList
+      .map(pair => (pair.f0, pair.f1.longValue()))
+      .sortBy(_._1)
+  }
+
+  /**
+   * Run the plan and write every row to `path` as one line of text.
+   *
+   * The sink runs at parallelism one, so `path` is the file the caller named rather than a directory of part
+   * files. That serializes the write; a sink that fans out cannot also promise a single named file.
+   */
+  @nowarn("cat=deprecation")
+  def writeLines(plan: FlinkPlan, path: String): Unit = {
+    val env = batchEnv()
+    plan(env).writeAsText(path, FileSystem.WriteMode.OVERWRITE).setParallelism(1)
+    env.execute(s"flowforge-write-$path")
+    ()
+  }
+}
+
+/** Keeps the rows whose decoded record satisfies `predicate`. */
+private[flink] final class DecodedFilter[A](decoder: DataDecoder[A], predicate: A => Boolean)
+    extends FilterFunction[String] {
+  override def filter(json: String): Boolean = FlinkRows.decode(json, decoder).exists(predicate)
+}
+
+/** Applies `f` to each decoded record and emits the results as JSON rows. */
+private[flink] final class DecodedFlatMap[A, B](
+  decoder: DataDecoder[A],
+  encoder: DataEncoder[B],
+  f: A => List[B])
+    extends FlatMapFunction[String, String] {
+
+  override def flatMap(json: String, out: Collector[String]): Unit =
+    FlinkRows.decode(json, decoder).foreach { a =>
+      f(a).foreach(b => FlinkRows.encode(b, encoder, DataFormat.JSON).foreach(out.collect))
+    }
+}
+
+/** Re-encodes each row in `format`, leaving the record itself alone. */
+private[flink] final class ReEncode[A](decoder: DataDecoder[A], encoder: DataEncoder[A], format: DataFormat)
+    extends FlatMapFunction[String, String] {
+
+  override def flatMap(json: String, out: Collector[String]): Unit =
+    FlinkRows.decode(json, decoder).foreach(a => FlinkRows.encode(a, encoder, format).foreach(out.collect))
+}
+
+/** Turns one CSV line into one JSON row. Emits nothing for the header line and for a blank line. */
+private[flink] final class CsvLineToJson(header: List[String]) extends FlatMapFunction[String, String] {
+  override def flatMap(line: String, out: Collector[String]): Unit =
+    FlinkRows.csvLineToJson(header, line).foreach(out.collect)
+}
+
+/** Emits each distinct key one record reports, paired with a count of one for the sum downstream. */
+private[flink] final class KeysOfRecord[A](decoder: DataDecoder[A], keysOf: A => List[String])
+    extends FlatMapFunction[String, Tuple2[String, java.lang.Long]] {
+
+  override def flatMap(json: String, out: Collector[Tuple2[String, java.lang.Long]]): Unit =
+    FlinkRows.decode(json, decoder).foreach { a =>
+      keysOf(a).distinct.foreach(key => out.collect(Tuple2.of(key, java.lang.Long.valueOf(1L))))
+    }
+}
+
+/** The key of a `(key, count)` pair. */
+private[flink] final class FirstField extends KeySelector[Tuple2[String, java.lang.Long], String] {
+  override def getKey(pair: Tuple2[String, java.lang.Long]): String = pair.f0
+}
diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/package.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/package.scala
new file mode 100644
index 00000000..e50cf99c
--- /dev/null
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/package.scala
@@ -0,0 +1,24 @@
+package com.flowforge.engines
+
+import org.apache.flink.streaming.api.datastream.DataStream
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
+
+package object flink {
+
+  /**
+   * How to build a stream of JSON rows, given an environment to build it in.
+   *
+   * A Flink job graph belongs to one environment and is consumed when that environment executes, so a dataset
+   * cannot hold a built `DataStream` and stay usable: the second operation on it would re-run every sink the
+   * first one added. A plan holds no environment, so each terminal operation builds the graph fresh in its
+   * own and the dataset stays a value that can be used more than once.
+   *
+   * Building a plan is a calculation. Running one is an action, and every action in this module lives in
+   * [[FlinkStreamOps]].
+   *
+   * The rows are JSON because JSON is the one representation both sides already speak: `DataDecoder` and
+   * `DataEncoder` are defined over it, so a FlowForge operation written against `A` can be applied to a row
+   * inside a Flink task. This is the same choice the Spark engine makes in `SparkFrameOps`.
+   */
+  type FlinkPlan = StreamExecutionEnvironment => DataStream[String]
+}

From 24abf4fba13fbae9428341f1cd8d8971819184cd Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 12:10:38 +0200
Subject: [PATCH 3/9] fix(flink): move the demo off the deprecated scala api

---
 .../engines/flink/FlinkStreamingDemo.scala    | 108 ++++++++----------
 1 file changed, 48 insertions(+), 60 deletions(-)

diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamingDemo.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamingDemo.scala
index 9fdfd36a..f8347524 100644
--- a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamingDemo.scala
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamingDemo.scala
@@ -1,18 +1,24 @@
 package com.flowforge.engines.flink
 
+import org.apache.flink.api.common.functions.{ FilterFunction, MapFunction }
 import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.core.fs.FileSystem
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
 import org.apache.flink.streaming.api.functions.ProcessFunction
-import org.apache.flink.streaming.api.scala._
 import org.apache.flink.util.Collector
 
+import scala.annotation.nowarn
+import scala.jdk.CollectionConverters._
+
 /**
  * Minimal Flink runner demo proving "same business logic, different runner".
  *
- * This demo shows that FlowForge's pure domain transforms can be reused across different execution engines
- * (Spark, Flink, etc.) without modification.
+ * The transforms below are pure functions of a `User`, so the same ones a Spark pipeline applies are the ones
+ * this job applies. Only the wrappers that hand a record to Flink are engine-specific.
  *
- * The business logic (data transformations) remains pure and engine-agnostic, while only the execution
- * framework changes.
+ * It uses the Java `DataStream` API, like the rest of this module: the Scala API is published for 2.12 only
+ * and is deprecated upstream, so depending on it would pin this module to a Scala version `core` does not
+ * publish for.
  */
 object FlinkStreamingDemo {
 
@@ -27,11 +33,11 @@ object FlinkStreamingDemo {
     email: String,
     processed: Boolean)
 
-  // Flink requires explicit TypeInformation for case classes
-  implicit val userTypeInfo: TypeInformation[User] = TypeInformation.of(classOf[User])
-  implicit val processedUserTypeInfo: TypeInformation[ProcessedUser] =
+  // The Java API infers no types from Scala signatures, so each one is stated.
+  private val userTypeInfo: TypeInformation[User] = TypeInformation.of(classOf[User])
+  private val processedUserTypeInfo: TypeInformation[ProcessedUser] =
     TypeInformation.of(classOf[ProcessedUser])
-  implicit val stringTypeInfo: TypeInformation[String] = TypeInformation.of(classOf[String])
+  private val stringTypeInfo: TypeInformation[String] = TypeInformation.of(classOf[String])
 
   /**
    * PURE DOMAIN TRANSFORM - Engine Agnostic
@@ -57,65 +63,49 @@ object FlinkStreamingDemo {
       user.name.nonEmpty &&
       user.email.contains("@")
 
+  /** The records the demo runs on. A bounded collection, so the job finishes. */
+  private val users: List[User] = List(
+    User(1, "john doe", "JOHN@EXAMPLE.COM"),
+    User(2, "jane smith", "JANE@TEST.ORG"),
+    User(0, "", "invalid"), // Invalid user - will be filtered
+    User(3, "bob wilson", "BOB@COMPANY.NET"),
+  )
+
+  /** Where the demo writes its results. */
+  val outputPath: String = "/tmp/flowforge-flink-output.txt"
+
+  @nowarn("cat=deprecation")
   def main(args: Array[String]): Unit = {
-    // Set up Flink execution environment
     val env = StreamExecutionEnvironment.getExecutionEnvironment
     env.setParallelism(2)
 
-    // Option 1: Socket text stream (as specified in plan)
-    // Uncomment to read from socket: nc -l 9999
-    // val socketStream = env.socketTextStream("localhost", 9999)
-    //   .map(line => {
-    //     val parts = line.split(",")
-    //     if (parts.length >= 3) User(parts(0).toLong, parts(1), parts(2))
-    //     else User(0, "", "")
-    //   })
-
-    // Option 2: Test data stream for demo
-    val users = Seq(
-      User(1, "john doe", "JOHN@EXAMPLE.COM"),
-      User(2, "jane smith", "JANE@TEST.ORG"),
-      User(0, "", "invalid"), // Invalid user - will be filtered
-      User(3, "bob wilson", "BOB@COMPANY.NET"),
-    )
-
-    val dataStream = env.fromCollection(users)
-
-    // Apply the same business logic as Spark pipelines
-    val processedStream = dataStream
-      .filter(isValidUser)              // Pure domain filter
-      .process(new UserProcessFunction) // Flink wrapper around pure transform
+    val processedStream = env
+      .fromCollection(users.asJava, userTypeInfo)
+      .filter(ValidUser)                                       // Pure domain filter
+      .process(new UserProcessFunction, processedUserTypeInfo) // Flink wrapper around pure transform
       .name("Process Users")
 
-    // Output results - both console and file sink (as specified in plan)
-    processedStream.print("Processed Users")
+    // `DataStream.print` is a Flink sink, not Scala's `print`: it adds an operator to the job graph that
+    // writes each record as the job runs. The linter matches on the name, so the rule is waived here.
+    processedStream.print("Processed Users") // scalafix:ok DisableSyntax.noPrintln
 
-    // File sink as required by plan
     processedStream
-      .map((user: ProcessedUser) => user.toString)
-      .writeAsText("/tmp/flowforge-flink-output.txt")
+      .map(RenderUser, stringTypeInfo)
+      .writeAsText(outputPath, FileSystem.WriteMode.OVERWRITE)
       .setParallelism(1)
 
-    println("""
-=== FlowForge Flink Demo ===
-
-This demo proves FlowForge's engine abstraction:
-- Same pure domain transforms (processUser, isValidUser)
-- Different execution engine (Flink vs Spark)
-- Business logic remains unchanged
-- Only the engine-specific wrappers change
-
-Expected Output:
-- Processed User: ProcessedUser(1,JOHN DOE,john@example.com,true)
-- Processed User: ProcessedUser(2,JANE SMITH,jane@test.org,true) 
-- Processed User: ProcessedUser(3,BOB WILSON,bob@company.net,true)
-- Invalid user (id=0) filtered out
+    env.execute("FlowForge Flink Demo")
+    ()
+  }
 
-Run this with: sbt "engines-flink/run"
-    """)
+  /** Flink's wrapper around [[isValidUser]]. */
+  private object ValidUser extends FilterFunction[User] {
+    override def filter(user: User): Boolean = isValidUser(user)
+  }
 
-    // Execute the Flink job
-    env.execute("FlowForge Flink Demo")
+  /** How a result reaches a text sink. The sink writes lines, so a record has to become one. */
+  private object RenderUser extends MapFunction[ProcessedUser, String] {
+    override def map(user: ProcessedUser): String = user.toString
   }
 
   /**
@@ -129,10 +119,8 @@ Run this with: sbt "engines-flink/run"
       user: User,
       ctx: ProcessFunction[User, ProcessedUser]#Context,
       out: Collector[ProcessedUser],
-    ): Unit = {
+    ): Unit =
       // Call the pure domain transform (same as used in Spark)
-      val processed = processUser(user)
-      out.collect(processed)
-    }
+      out.collect(processUser(user))
   }
 }

From 8fa1d15f6fe4bfed020c98ddba521f8315dc8097 Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 12:10:38 +0200
Subject: [PATCH 4/9] test(flink): compare the engine with the in-memory one

---
 .../engines/flink/CrossEngineParitySpec.scala | 249 ++++++++++++++----
 1 file changed, 194 insertions(+), 55 deletions(-)

diff --git a/modules/engines-flink/src/test/scala/com/flowforge/engines/flink/CrossEngineParitySpec.scala b/modules/engines-flink/src/test/scala/com/flowforge/engines/flink/CrossEngineParitySpec.scala
index f7d6def1..da4d6077 100644
--- a/modules/engines-flink/src/test/scala/com/flowforge/engines/flink/CrossEngineParitySpec.scala
+++ b/modules/engines-flink/src/test/scala/com/flowforge/engines/flink/CrossEngineParitySpec.scala
@@ -1,13 +1,14 @@
-// scalafix:off DisableSyntax.var DisableSyntax.throw DisableSyntax.null DisableSyntax.noUnsafeRunSync
+// scalafix:off DisableSyntax.throw DisableSyntax.noUnsafeRunSync
 package com.flowforge.engines.flink
 
 import cats.data.NonEmptyList
 import cats.effect.IO
 import cats.effect.unsafe.implicits.global
-import cats.implicits._
+import cats.syntax.validated._
 import com.flowforge.core.algebra._
-import com.flowforge.core.instances.EffectInstances.catsEffectSystemInstance
-import com.flowforge.core.types.PipelineTypes.QualityCheck
+import com.flowforge.core.impl.InMemoryDataAlgebra
+import com.flowforge.core.instances.EffectInstances
+import com.flowforge.core.types.PipelineTypes.{ DataContract => PDataContract, QualityCheck }
 import com.flowforge.core.types._
 import org.scalatest.funsuite.AnyFunSuite
 import org.scalatest.matchers.should.Matchers
@@ -16,39 +17,50 @@ import java.nio.charset.StandardCharsets
 import java.nio.file.Files
 
 /**
- * What this spec can check is delegation, not parity. `FlinkDataAlgebra` holds an `InMemoryDataAlgebra` and
- * forwards every method to it, so comparing the two compares an object with itself and passes whatever they
- * both do. Parity between two implementations is checked in
- * `modules/engines-spark/src/test/scala/com/flowforge/engines/spark/EngineParitySpec.scala`, against Spark,
- * which is the only other algebra with an implementation of its own.
+ * The codecs, the checks and the contracts live here rather than in the suite.
+ *
+ * Every one of them travels to a Flink task, so every one of them is serialized. A value declared in the
+ * suite's body captures the suite, which is not serializable, so the job would fail on the way out rather
+ * than on anything the test is about.
  */
-class CrossEngineParitySpec extends AnyFunSuite with Matchers {
+object CrossEngineParitySpec {
 
-  case class User(id: Int, name: String)
+  case class Person(id: Int, name: String)
 
-  implicit val userEncoder: DataEncoder[User] = new DataEncoder[User] {
-    def encode(u: User, format: DataFormat) = format match {
+  implicit val personEncoder: DataEncoder[Person] = new DataEncoder[Person] {
+    def encode(p: Person, format: DataFormat) = format match {
       case DataFormat.CSV =>
-        val line = s"${u.id},${u.name}"
-        Right(EncodedData(line.getBytes("UTF-8"), format))
-      case other => Left(UnsupportedFormat(other, "User"))
+        Right(EncodedData(s"${p.id},${p.name}".getBytes(StandardCharsets.UTF_8), format))
+      case DataFormat.JSON | DataFormat.JSONL =>
+        val json = s"""{"id":${p.id},"name":"${p.name}"}"""
+        Right(EncodedData(json.getBytes(StandardCharsets.UTF_8), format))
+      case other => Left(UnsupportedFormat(other, "Person"))
     }
     def schema(format: DataFormat): DataSchema =
-      DataSchema.builder
-        .addField("id", DataType.Integer)
-        .addField("name", DataType.String)
-        .build
-    def estimateSize(u: User, format: DataFormat): Long               = 0L
-    def supportsFormat(format: DataFormat): Boolean                   = format == DataFormat.CSV
-    def optimizationHints(u: User, format: DataFormat): EncodingHints = EncodingHints.default
+      DataSchema.builder.addField("id", DataType.Integer).addField("name", DataType.String).build
+    def estimateSize(p: Person, format: DataFormat): Long = 0L
+    def supportsFormat(format: DataFormat): Boolean =
+      format == DataFormat.CSV || format == DataFormat.JSON || format == DataFormat.JSONL
+    def optimizationHints(p: Person, format: DataFormat): EncodingHints = EncodingHints.default
   }
 
-  implicit val userDecoder: DataDecoder[User] = new DataDecoder[User] {
-    def decode(encoded: EncodedData, format: DataFormat) = format match {
-      case DataFormat.CSV =>
-        val parts = new String(encoded.data, "UTF-8").split(',')
-        Right(User(parts(0).toInt, parts(1)))
-      case other => Left(CorruptedData(s"Unsupported format: $other"))
+  implicit val personDecoder: DataDecoder[Person] = new DataDecoder[Person] {
+    def decode(encoded: EncodedData, format: DataFormat) = {
+      val text = new String(encoded.data, StandardCharsets.UTF_8).trim
+      format match {
+        case DataFormat.CSV =>
+          text.split(',') match {
+            case Array(id, name) =>
+              id.toIntOption.map(i => Person(i, name)).toRight(CorruptedData(s"Not a person record: $text"))
+            case _ => Left(CorruptedData(s"Expected two columns: $text"))
+          }
+        case DataFormat.JSON | DataFormat.JSONL =>
+          val id   = """"id"\s*:\s*(-?\d+)""".r.findFirstMatchIn(text).flatMap(_.group(1).toIntOption)
+          val name = """"name"\s*:\s*"([^"]*)"""".r.findFirstMatchIn(text).map(_.group(1))
+          id.zip(name).map { case (i, n) => Person(i, n) }
+            .toRight(CorruptedData(s"Not a person record: $text"))
+        case other => Left(CorruptedData(s"Unsupported format: $other"))
+      }
     }
     def validateSchema(encoded: EncodedData, expected: DataSchema) = Right(())
     def decodeWithEvolution(
@@ -57,41 +69,168 @@ class CrossEngineParitySpec extends AnyFunSuite with Matchers {
       target: DataSchema,
     ) =
       decode(encoded, format)
-    override def supportsFormat(format: DataFormat): Boolean = format == DataFormat.CSV
+    override def supportsFormat(format: DataFormat): Boolean =
+      format == DataFormat.CSV || format == DataFormat.JSON || format == DataFormat.JSONL
+  }
+
+  val nameNotBlank: QualityCheck[Person] = p =>
+    if (p.name.trim.nonEmpty) ().validNel
+    else ValidationError.SchemaViolation("name", "non-blank", "blank", message = "name is blank").invalidNel
+
+  val idIsNegative: QualityCheck[Person] = p =>
+    if (p.id < 0) ().validNel
+    else
+      ValidationError
+        .SchemaViolation("id", "negative", p.id.toString, message = "id is not negative").invalidNel
+
+  /** Rejects Bob and nobody else, so a passing result means the contract was not applied. */
+  val nameIsNotBob: PDataContract[Person] = p =>
+    if (p.name != "Bob") ().validNel
+    else ValidationError.SchemaViolation("name", "not Bob", p.name, message = "name is Bob").invalidNel
+
+  val isEven: Person => Boolean = _.id % 2 == 0
+
+  val shout: Person => Person = p => p.copy(name = p.name.toUpperCase)
+}
+
+/**
+ * Parity between the Flink algebra and the in-memory one, on the same fixture.
+ *
+ * This suite used to compare `FlinkDataAlgebra` with the `InMemoryDataAlgebra` it forwarded every call to,
+ * which compared an object with itself and passed whatever the two did. The Flink algebra now runs `read`,
+ * `write`, `filter`, `map`, `validate` and `runQualityChecks` as Flink jobs, so there are two implementations
+ * to compare and a disagreement between them can fail a test.
+ *
+ * Each test runs its own Flink job on a local MiniCluster, so these are slow by the standards of the rest of
+ * the build. That is the price of checking that the engine runs rather than that it delegates.
+ */
+class CrossEngineParitySpec extends AnyFunSuite with Matchers {
+
+  import CrossEngineParitySpec._
+
+  implicit private val F: EffectSystem[IO] = EffectInstances.catsEffectSystemInstance
+
+  private val flink: DataAlgebra[IO]  = new FlinkDataAlgebra[IO]()
+  private val memory: DataAlgebra[IO] = new InMemoryDataAlgebra[IO]()
+
+  private val rows = List(Person(1, "Alice"), Person(2, "Bob"), Person(3, "Carol"))
+
+  /** Small enough that the Flink sample holds every record, so the two `data` lists are comparable. */
+  private def csvFixture(people: List[Person] = rows): LocalDataSource = {
+    val file  = Files.createTempFile("ff-flink-parity", ".csv")
+    val lines = "id,name" :: people.map(p => s"${p.id},${p.name}")
+    Files.write(file, lines.mkString("\n").getBytes(StandardCharsets.UTF_8))
+    LocalDataSource(file.toString, DataFormat.CSV)
+  }
+
+  test("both engines read the same records and the same count from one csv") {
+    val source = csvFixture()
+
+    val fromFlink  = flink.read[Person](source).unsafeRunSync()
+    val fromMemory = memory.read[Person](source).unsafeRunSync()
+
+    fromFlink.data.sortBy(_.id) shouldBe rows
+    fromMemory.data.sortBy(_.id) shouldBe rows
+    flink.count(fromFlink) shouldBe memory.count(fromMemory)
+  }
+
+  test("the record count comes from the job, not from the sample") {
+    val people = (1 to 250).toList.map(i => Person(i, s"person-$i"))
+    val source = csvFixture(people)
+
+    val dataset = flink.read[Person](source).unsafeRunSync()
+
+    // `data` is capped at the sample size and says so. `count` answers for the whole file.
+    dataset.data.size shouldBe FlinkDataset.sampleSize
+    flink.count(dataset) shouldBe 250L
+    flink.count(flink.filter[Person](dataset, isEven)) shouldBe 125L
   }
 
-  private val check: QualityCheck[User] = _ => ().validNel
-  private val checks                    = NonEmptyList.one(check)
+  test("both engines filter and map to the same records") {
+    val source = csvFixture()
+
+    val flinkOut = flink.map[Person, Person](
+      flink.filter[Person](flink.read[Person](source).unsafeRunSync(), isEven),
+      shout,
+    )
+    val memoryOut = memory.map[Person, Person](
+      memory.filter[Person](memory.read[Person](source).unsafeRunSync(), isEven),
+      shout,
+    )
 
-  private def writeSourceFile(): java.nio.file.Path = {
-    val tmp     = Files.createTempFile("users", ".csv")
-    val content = "id,name\n1,Alice\n2,Bob\n"
-    Files.write(tmp, content.getBytes(StandardCharsets.UTF_8))
-    tmp
+    flinkOut.data.sortBy(_.id) shouldBe List(Person(2, "BOB"))
+    memoryOut.data.sortBy(_.id) shouldBe flinkOut.data.sortBy(_.id)
   }
 
-  test("the Flink algebra forwards read, write and quality calls to the in-memory engine") {
-    val sourcePath = writeSourceFile()
-    val source     = LocalDataSource(sourcePath.toString, DataFormat.CSV)
+  test("both engines report the same quality check outcome") {
+    val source = csvFixture()
+    val checks = NonEmptyList.of(nameNotBlank, idIsNegative)
+
+    val flinkResults =
+      flink.runQualityChecks(flink.read[Person](source).unsafeRunSync(), checks).unsafeRunSync()
+    val memoryResults =
+      memory.runQualityChecks(memory.read[Person](source).unsafeRunSync(), checks).unsafeRunSync()
+
+    // The second check fails on every record. An engine that reports it as passing is not running the check.
+    flinkResults.map(r => r.checkName -> r.passed) shouldBe List("check_0" -> true, "check_1" -> false)
+    memoryResults.map(r => r.checkName -> r.passed) shouldBe flinkResults.map(r => r.checkName -> r.passed)
+  }
+
+  test("both engines report the same contract result") {
+    val source = csvFixture()
+
+    val fromFlink = flink.validate(flink.read[Person](source).unsafeRunSync(), nameIsNotBob).unsafeRunSync()
+    val fromMemory =
+      memory.validate(memory.read[Person](source).unsafeRunSync(), nameIsNotBob).unsafeRunSync()
 
-    val inMemoryDA = new com.flowforge.core.impl.InMemoryDataAlgebra[IO]()
-    val flinkDA    = new FlinkDataAlgebra[IO]()
+    fromFlink.passed shouldBe false
+    fromFlink.score shouldBe (2.0 / 3.0)
+    fromFlink.violations.map(v => (v.message, v.recordsAffected)) shouldBe List(("name is Bob", 1L))
 
-    val memOut   = Files.createTempFile("mem-out", ".csv").toString
-    val flinkOut = Files.createTempFile("flink-out", ".csv").toString
+    // `data` differs - each engine returns its own dataset - so compare what the contract said about it.
+    fromMemory.passed shouldBe fromFlink.passed
+    fromMemory.score shouldBe fromFlink.score
+    fromMemory.violations shouldBe fromFlink.violations
+  }
+
+  test("the flink sink writes every record, and the records read back") {
+    val source = csvFixture()
+    val out    = Files.createTempFile("ff-flink-out", ".jsonl")
+    Files.delete(out) // the sink writes the file itself
+
+    val dataset = flink.read[Person](source).unsafeRunSync()
+    val result  = flink.write(dataset, LocalDataSink(out.toString, DataFormat.JSONL)).unsafeRunSync()
+
+    result.success shouldBe true
+    result.recordsWritten shouldBe 3L
+
+    val readBack = flink.read[Person](LocalDataSource(out.toString, DataFormat.JSONL)).unsafeRunSync()
+    readBack.data.sortBy(_.id) shouldBe rows
+  }
+
+  test("a sink format the flink engine does not write fails rather than writing a sample") {
+    val source = csvFixture()
+    val out    = Files.createTempFile("ff-flink-csv-out", ".csv").toString
+
+    val dataset = flink.read[Person](source).unsafeRunSync()
+    val failure = intercept[UnsupportedOperationException](
+      flink.write(dataset, LocalDataSink(out, DataFormat.CSV)).unsafeRunSync(),
+    )
+
+    failure.getMessage should include("CSV")
+  }
 
-    val memRes = (for {
-      ds <- inMemoryDA.read[User](source)
-      _  <- inMemoryDA.write(ds, LocalDataSink(memOut, DataFormat.CSV))
-      qc <- inMemoryDA.runQualityChecks(ds, checks)
-    } yield qc).unsafeRunSync()
+  test("a source format neither engine reads fails on both, with the error type each one raises") {
+    val source = LocalDataSource(csvFixture().location, DataFormat.Avro)
 
-    val flinkRes = (for {
-      ds <- flinkDA.read[User](source)
-      _  <- flinkDA.write(ds, LocalDataSink(flinkOut, DataFormat.CSV))
-      qc <- flinkDA.runQualityChecks(ds, checks)
-    } yield qc).unsafeRunSync()
+    // A known divergence, pinned rather than hidden: the Flink engine wraps the failure in the framework's
+    // error type, like Spark does, and the in-memory algebra raises the plain exception.
+    val flinkFailure =
+      intercept[DataProcessingError.ProcessingFailure](flink.read[Person](source).unsafeRunSync())
+    val memoryFailure =
+      intercept[UnsupportedOperationException](memory.read[Person](source).unsafeRunSync())
 
-    memRes.map(_.passed) shouldEqual flinkRes.map(_.passed)
+    flinkFailure.reason should include("Avro")
+    memoryFailure.getMessage should include("Avro")
   }
 }

From d4d02c80cf576789880a7e94de74d8ec4073ba64 Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 12:10:38 +0200
Subject: [PATCH 5/9] ci: run the engines-flink tests

---
 .github/workflows/ci.yml | 24 ++++++++++++++++++++++++
 1 file changed, 24 insertions(+)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index e02371b8..a6c342f1 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -76,6 +76,7 @@ jobs:
           connectors-gcs/mimaReportBinaryIssues
           connectors-jdbc/mimaReportBinaryIssues
           engines-spark/mimaReportBinaryIssues
+          engines-flink/mimaReportBinaryIssues
           quality-deequ/mimaReportBinaryIssues
           infrastructure/mimaReportBinaryIssues
           contracts-sdk/mimaReportBinaryIssues
@@ -228,6 +229,29 @@ jobs:
       - name: Run engines-spark tests
         run: sbt "-Dsbt.log.noformat=true" engines-spark/test
 
+  flink-engine:
+    name: Flink engine tests
+    needs: quick-checks
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    if: "!contains(github.event.head_commit.message, '[skip ci]') && !contains(github.event.head_commit.message, '[skip release]')"
+    steps:
+      - uses: actions/checkout@v6
+
+      - uses: sbt/setup-sbt@v1
+
+      - uses: actions/setup-java@v5
+        with:
+          distribution: temurin
+          java-version: 17
+          cache: sbt
+
+      # These tests start a local Flink MiniCluster, so a failure here is a Flink failure rather than a
+      # compile error. The root `testQuick` runs them too; this job is what names them.
+      - name: Run engines-flink tests
+        run: sbt "-Dsbt.log.noformat=true" engines-flink/test
+
   spark-it:
     name: Spark integration tests
     needs: [test]

From 3581eda6fa3617de5b4030f9f967249f6f4c63ca Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 12:10:38 +0200
Subject: [PATCH 6/9] docs: record what the flink engine does and does not do

---
 docs/plan/release-criteria.md |  2 +-
 docs/plan/v1.0-readiness.md   | 36 ++++++++++++++++++++++-------------
 2 files changed, 24 insertions(+), 14 deletions(-)

diff --git a/docs/plan/release-criteria.md b/docs/plan/release-criteria.md
index c8ded94a..6e517ca5 100644
--- a/docs/plan/release-criteria.md
+++ b/docs/plan/release-criteria.md
@@ -7,7 +7,7 @@ one place a readiness claim belongs.
 ## Build & Platform
 - CI runs on JDK 17 (Ubuntu 22.04), sbt 1.9+.
 - Scala 2.13 primary; Scala 3 sources compile in `core` (no Spark dependencies).
-- `engines-flink`: builds and resolves on a single axis, or the module is removed.
+- `engines-flink`: builds and resolves on Scala 2.13, and its tests run a Flink job rather than a delegation.
  - Spark IT job pinned to Java 17; println banned in production sources via CI guard.
 
 ## Contracts & Compile‑Time Safety
diff --git a/docs/plan/v1.0-readiness.md b/docs/plan/v1.0-readiness.md
index 94c296db..a65e1dd1 100644
--- a/docs/plan/v1.0-readiness.md
+++ b/docs/plan/v1.0-readiness.md
@@ -19,20 +19,17 @@ change before the version number means anything.
   filter followed by a write wrote unfiltered rows and a violation in record 101 went unreported. The
   operations that read every record now carry a `DataDecoder` bound, which is what lets the engine push them
   into the frame.
-- `engines-flink` does not build. `build.sbt` pins it to Scala 2.12 for the Flink Scala API, but `core` and
-  `connectors` publish 2.13 and 3 only, so dependency resolution fails before compilation starts. No CI job
-  builds the module.
+- `engines-flink` builds and tests on Scala 2.13 against Flink 1.18. It used to be pinned to 2.12 for the
+  Flink Scala API, which `core` does not publish for, so resolution failed before compilation started; the
+  module now uses Flink's Java artifacts, which carry no Scala suffix. It was also missing from the root
+  aggregate, so no build or test command reached it. `engines-flink/test` runs on every CI build and starts a
+  local MiniCluster.
+- `read`, `write`, `filter`, `map`, `flatMap`, `union`, `validate` and `runQualityChecks` on Flink run as
+  batch jobs over every record. A Flink-backed dataset holds a plan rather than the records, so the driver
+  keeps at most 100 of them.
 
 ## Blocking gaps
 
-### The Flink engine runs no Flink
-`FlinkDataAlgebra` forwards every method to `InMemoryDataAlgebra`. Work happens on the driver heap and
-nothing is submitted to a cluster. The module also cannot be compiled (see above), so this is not covered by
-any test.
-
-Before v1.0, either implement the engine on the Flink DataStream or Table API and give it a 2.12 build that
-resolves, or remove the module and say Spark is the only engine.
-
 ### Connector surfaces that are not implemented
 Several connector paths exist as types without behaviour: the S3 read and write path on Spark raises a
 failure, and the S3 connector body is commented out. The capability set advertises `Read` and `Write` without
@@ -64,8 +61,21 @@ These are deliberate for v1.0 and recorded so nobody rediscovers them as bugs.
   pinned there: on an unsupported format Spark raises `DataProcessingError.ProcessingFailure` and the
   in-memory engine raises a plain `UnsupportedOperationException`. A caller that handles one does not handle
   the other.
-- The Flink spec named for parity checks delegation, which is all it can check while the engine forwards to
-  the in-memory one.
+- Flink parity is tested against the in-memory engine in
+  `modules/engines-flink/src/test/scala/com/flowforge/engines/flink/CrossEngineParitySpec.scala`, with the
+  same unsupported-format divergence pinned there.
+- `groupBy`, `join`, `sortBy`, `take`, `drop`, `profile`, the CDC operations and the table operations still
+  forward to the in-memory engine on Flink, so on a Flink-backed dataset they answer for at most the 100
+  sampled records. They are the same operations that run on the driver for Spark, for the same reason.
+- The Flink sink writes JSON and JSONL only. A dataflow sink cannot order a CSV header line ahead of the rows
+  it writes, and a headerless CSV is a file this engine cannot read back, so a CSV sink raises rather than
+  writing one. The sink runs at parallelism one so the output is the single file the caller named.
+- The 1000-message cap on `validate` truncates deterministically on Spark, which orders by message before
+  capping, and not on Flink, which has no cheap global sort over a `DataStream`. Which messages survive is
+  unspecified there. On both engines the cap cannot change `passed` or `score`.
+- The Flink engine runs batch jobs, not streaming ones. Every job sets `RuntimeExecutionMode.BATCH`, which is
+  what makes a keyed count report one final value per key, and which means no unbounded source can be
+  consumed. Every source it reads is a bounded file.
 - Four `TODO` markers remain in `core`: record-level validation and the Reader-based pipeline in
   `syntax/PipelineSyntax.scala`, and the environment fallback and per-source loading in
   `algebra/ConfigurationAlgebra.scala`. Each one returns something rather than failing, so a caller gets a

From 7e4d7e9335c4ad5ebf138e068fbf4524af41b005 Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 12:19:23 +0200
Subject: [PATCH 7/9] style: format the new flink sources

---
 .../engines/flink/FlinkDataset.scala          |  5 +++-
 .../flowforge/engines/flink/FlinkRows.scala   |  9 ++++--
 .../engines/flink/FlinkStreamOps.scala        | 29 ++++++++++++-------
 3 files changed, 30 insertions(+), 13 deletions(-)

diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataset.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataset.scala
index 5e7be475..669133ce 100644
--- a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataset.scala
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataset.scala
@@ -60,7 +60,10 @@ object FlinkDataset {
    * @param source
    *   where the records came from, which a plan does not carry
    */
-  def fromPlan[A](plan: FlinkPlan, source: Option[DataSource] = None)(implicit
+  def fromPlan[A](
+    plan: FlinkPlan,
+    source: Option[DataSource] = None,
+  )(implicit
     decoder: DataDecoder[A],
   ): FlinkDataset[A] = {
     val sampleRows = FlinkStreamOps.collect(plan, sampleSize)
diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala
index 9cf17253..6f81d6f4 100644
--- a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala
@@ -24,7 +24,11 @@ private[flink] object FlinkRows {
       .toOption
 
   /** Encode one record in `format`. `None` when the encoder rejects it. */
-  def encode[A](value: A, encoder: DataEncoder[A], format: DataFormat): Option[String] =
+  def encode[A](
+    value: A,
+    encoder: DataEncoder[A],
+    format: DataFormat,
+  ): Option[String] =
     encoder.encode(value, format).toOption.map(ed => new String(ed.data, StandardCharsets.UTF_8))
 
   /**
@@ -101,7 +105,8 @@ private[flink] object FlinkRows {
       jsonObject = obj =>
         DataType.Struct(
           obj.toList.map {
-            case (name, field) => StructField(FieldName.unsafeFrom(name), typeOf(field), nullable = field.isNull)
+            case (name, field) =>
+              StructField(FieldName.unsafeFrom(name), typeOf(field), nullable = field.isNull)
           },
         ),
     )
diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala
index 007043b9..3a69324b 100644
--- a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala
@@ -26,8 +26,8 @@ import scala.jdk.CollectionConverters._
  * `DataEncoder` are `Serializable` and why the function classes here are named rather than anonymous: a named
  * top-level class holds only the fields it declares, so there is no enclosing instance to serialize.
  *
- * A record the decoder rejects is dropped, for the same reason it is in `SparkFrameOps`: a pure operation that
- * returns a `Dataset` has nowhere to report the failure.
+ * A record the decoder rejects is dropped, for the same reason it is in `SparkFrameOps`: a pure operation
+ * that returns a `Dataset` has nowhere to report the failure.
  */
 private[flink] object FlinkStreamOps {
 
@@ -61,7 +61,10 @@ private[flink] object FlinkStreamOps {
   }
 
   /** Apply `f` to every decoded record and carry the results on as JSON rows. */
-  def mapRows[A, B](plan: FlinkPlan, f: A => List[B])(implicit
+  def mapRows[A, B](
+    plan: FlinkPlan,
+    f: A => List[B],
+  )(implicit
     decoder: DataDecoder[A],
     encoder: DataEncoder[B],
   ): FlinkPlan = {
@@ -73,7 +76,10 @@ private[flink] object FlinkStreamOps {
   def union(left: FlinkPlan, right: FlinkPlan): FlinkPlan = env => left(env).union(right(env))
 
   /** Re-encode every record in `format`, for a sink that wants something other than the plan's JSON. */
-  def encodeRows[A](plan: FlinkPlan, format: DataFormat)(implicit
+  def encodeRows[A](
+    plan: FlinkPlan,
+    format: DataFormat,
+  )(implicit
     decoder: DataDecoder[A],
     encoder: DataEncoder[A],
   ): FlinkPlan = {
@@ -117,11 +123,11 @@ private[flink] object FlinkStreamOps {
    * A key is counted once per record: a record that reports the same key twice has still broken that rule
    * once, which is what the count claims to say.
    *
-   * At most `limit` keys come back, because a key built from a record's own values is otherwise as numerous as
-   * the dataset. Which keys survive the cap is not specified. The Spark engine orders by key before capping,
-   * so it truncates deterministically; a `DataStream` has no cheap global sort, so this does not. A caller
-   * that needs an exact answer over an unbounded key set needs a count, not a key list, and [[count]] gives
-   * one.
+   * At most `limit` keys come back, because a key built from a record's own values is otherwise as numerous
+   * as the dataset. Which keys survive the cap is not specified. The Spark engine orders by key before
+   * capping, so it truncates deterministically; a `DataStream` has no cheap global sort, so this does not. A
+   * caller that needs an exact answer over an unbounded key set needs a count, not a key list, and [[count]]
+   * gives one.
    */
   def countKeys[A](
     plan: FlinkPlan,
@@ -177,7 +183,10 @@ private[flink] final class DecodedFlatMap[A, B](
 }
 
 /** Re-encodes each row in `format`, leaving the record itself alone. */
-private[flink] final class ReEncode[A](decoder: DataDecoder[A], encoder: DataEncoder[A], format: DataFormat)
+private[flink] final class ReEncode[A](
+  decoder: DataDecoder[A],
+  encoder: DataEncoder[A],
+  format: DataFormat)
     extends FlatMapFunction[String, String] {
 
   override def flatMap(json: String, out: Collector[String]): Unit =

From 310502b28fc8daf10ac66f86610b2fbe5668299d Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 12:34:23 +0200
Subject: [PATCH 8/9] fix(flink): count rows in the job and skip the header
 line the file holds

---
 .../engines/flink/FlinkDataAlgebra.scala      | 10 ++----
 .../flowforge/engines/flink/FlinkRows.scala   | 30 +++++++++++++---
 .../engines/flink/FlinkStreamOps.scala        | 34 ++++++++++++++-----
 3 files changed, 54 insertions(+), 20 deletions(-)

diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataAlgebra.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataAlgebra.scala
index 426f90e9..b3190e5a 100644
--- a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataAlgebra.scala
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataAlgebra.scala
@@ -43,16 +43,12 @@ final class FlinkDataAlgebra[F[_]](implicit F: EffectSystem[F]) extends DataAlge
    */
   private def planFor(local: LocalDataSource): Option[FlinkPlan] = local.format match {
     case DataFormat.CSV =>
-      val header = {
+      val headerLine = {
         val lines = scala.io.Source.fromFile(local.location, "UTF-8")
-        try
-          lines
-            .getLines().find(_.trim.nonEmpty).fold(List.empty[String])(
-              _.trim.split(",", -1).toList.map(_.trim),
-            )
+        try lines.getLines().find(_.trim.nonEmpty).getOrElse("")
         finally lines.close()
       }
-      Some(FlinkStreamOps.readCsv(local.location, header))
+      Some(FlinkStreamOps.readCsv(local.location, headerLine))
     case DataFormat.JSON | DataFormat.JSONL => Some(FlinkStreamOps.readJsonLines(local.location))
     case _                                  => None
   }
diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala
index 6f81d6f4..3f1ed936 100644
--- a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkRows.scala
@@ -31,21 +31,41 @@ private[flink] object FlinkRows {
   ): Option[String] =
     encoder.encode(value, format).toOption.map(ed => new String(ed.data, StandardCharsets.UTF_8))
 
+  /**
+   * A CSV file's header: the line as the file holds it, and the column names read off it.
+   *
+   * Both are needed and neither is derivable from the other. The names are what a row's cells are named
+   * under, and they are trimmed, so rebuilding a line from them does not give back the line - a header
+   * written `id, name` rebuilds as `id,name`. The line is what [[csvLineToJson]] compares against to leave
+   * the header out of the records.
+   */
+  final case class CsvHeader(line: String, names: List[String])
+
+  /** Read the column names off a file's header line. */
+  def header(line: String): CsvHeader =
+    CsvHeader(line, line.trim.split(",", -1).toList.map(_.trim))
+
   /**
    * Read one CSV line as a JSON object, under the column names the file's header line gave.
    *
    * `None` for a blank line and for the header line itself, because the source hands every line of the file
-   * to the same function and the header is not a record.
+   * to the same function and the header is not a record. The comparison is against the header line as the
+   * file holds it, so a header the reader had to trim is still recognised as the header.
+   *
+   * A record whose text is the header line is dropped with it. A line-oriented source reports no line number,
+   * so there is nothing to tell the two apart by; the alternative is reading the header as a record, which is
+   * worse for every file that has one.
    *
    * A line with fewer cells than the header leaves the remaining fields null. Extra cells are dropped: the
    * header is what names a column, so a cell with no name has nowhere to go.
    */
-  def csvLineToJson(header: List[String], line: String): Option[String] = {
+  def csvLineToJson(header: CsvHeader, line: String): Option[String] = {
     val trimmed = line.trim
-    if (trimmed.isEmpty || trimmed == header.mkString(",")) None
+    if (trimmed.isEmpty || trimmed == header.line.trim) None
     else {
-      val cells  = trimmed.split(",", -1).toList
-      val fields = header.zipAll(cells, "", "").collect { case (name, cell) if name.nonEmpty => name -> cell }
+      val cells = trimmed.split(",", -1).toList
+      val fields =
+        header.names.zipAll(cells, "", "").collect { case (name, cell) if name.nonEmpty => name -> cell }
       Some(Json.obj(fields.map { case (name, cell) => name -> cellToJson(cell) }: _*).noSpaces)
     }
   }
diff --git a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala
index 3a69324b..6bcf5410 100644
--- a/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala
+++ b/modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkStreamOps.scala
@@ -3,7 +3,7 @@ package com.flowforge.engines.flink
 import com.flowforge.core.algebra.{ DataDecoder, DataEncoder }
 import com.flowforge.core.types.DataFormat
 import org.apache.flink.api.common.RuntimeExecutionMode
-import org.apache.flink.api.common.functions.{ FilterFunction, FlatMapFunction }
+import org.apache.flink.api.common.functions.{ FilterFunction, FlatMapFunction, MapFunction }
 import org.apache.flink.api.common.typeinfo.Types
 import org.apache.flink.api.java.functions.KeySelector
 import org.apache.flink.api.java.tuple.Tuple2
@@ -45,14 +45,16 @@ private[flink] object FlinkStreamOps {
   def readJsonLines(path: String): FlinkPlan = env => env.readTextFile(path)
 
   /**
-   * Read a CSV file as JSON rows, under the given column names.
+   * Read a CSV file as JSON rows, under the column names its header line gives.
    *
    * The header is a driver-side fact: it has to be known before the job is built, because it names the fields
    * every row will carry. The job itself still reads the whole file.
    */
   @nowarn("cat=deprecation")
-  def readCsv(path: String, header: List[String]): FlinkPlan =
+  def readCsv(path: String, headerLine: String): FlinkPlan = {
+    val header = FlinkRows.header(headerLine)
     env => env.readTextFile(path).flatMap(new CsvLineToJson(header), Types.STRING)
+  }
 
   /** Keep the rows whose decoded record satisfies the predicate. */
   def filter[A](plan: FlinkPlan, predicate: A => Boolean)(implicit decoder: DataDecoder[A]): FlinkPlan = {
@@ -109,12 +111,21 @@ private[flink] object FlinkStreamOps {
   /**
    * Run the plan and count its rows.
    *
-   * The rows are counted as they arrive and never held, so this is bounded whatever the dataset's size.
+   * The job counts; the driver reads one number. Counting on the driver instead would send every row through
+   * `executeAndCollect`, which carries each one to this process over the REST API, so a count of a large
+   * dataset would cost a transfer of the whole dataset. It would also be counted as an `Int` and so could
+   * overflow, which a `Long` sum inside the job cannot.
+   *
+   * Every row carries the same key, so the sum runs in one place. That is what a single total means, and a
+   * count is one record out however many went in.
    */
   def count(plan: FlinkPlan): Long = {
-    val rows = plan(batchEnv()).executeAndCollect()
-    try rows.asScala.length.toLong
-    finally rows.close()
+    val counted = plan(batchEnv())
+      .map(OnePerRow, pairType)
+      .keyBy(new FirstField, Types.STRING)
+      .sum(1)
+    // An empty plan reports no total rather than a total of zero, because there is no key to report under.
+    counted.executeAndCollect(1).asScala.toList.headOption.fold(0L)(_.f1.longValue())
   }
 
   /**
@@ -193,8 +204,15 @@ private[flink] final class ReEncode[A](
     FlinkRows.decode(json, decoder).foreach(a => FlinkRows.encode(a, encoder, format).foreach(out.collect))
 }
 
+/** Pairs every row with the one key a total is reported under, and a count of one to sum. */
+private[flink] object OnePerRow extends MapFunction[String, Tuple2[String, java.lang.Long]] {
+  override def map(row: String): Tuple2[String, java.lang.Long] =
+    Tuple2.of("", java.lang.Long.valueOf(1L))
+}
+
 /** Turns one CSV line into one JSON row. Emits nothing for the header line and for a blank line. */
-private[flink] final class CsvLineToJson(header: List[String]) extends FlatMapFunction[String, String] {
+private[flink] final class CsvLineToJson(header: FlinkRows.CsvHeader)
+    extends FlatMapFunction[String, String] {
   override def flatMap(line: String, out: Collector[String]): Unit =
     FlinkRows.csvLineToJson(header, line).foreach(out.collect)
 }

From bcc819b03e49ee0611ea14e6dcad9fa1b4270424 Mon Sep 17 00:00:00 2001
From: Vitthal Mirji <vitthalmirji@gmail.com>
Date: Tue, 6 Oct 2026 12:34:23 +0200
Subject: [PATCH 9/9] test(flink): pin one count per key and a padded header

---
 docs/plan/v1.0-readiness.md                   |  8 +++++++
 .../engines/flink/CrossEngineParitySpec.scala | 23 +++++++++++++++++++
 2 files changed, 31 insertions(+)

diff --git a/docs/plan/v1.0-readiness.md b/docs/plan/v1.0-readiness.md
index a65e1dd1..6adac667 100644
--- a/docs/plan/v1.0-readiness.md
+++ b/docs/plan/v1.0-readiness.md
@@ -70,6 +70,14 @@ These are deliberate for v1.0 and recorded so nobody rediscovers them as bugs.
 - The Flink sink writes JSON and JSONL only. A dataflow sink cannot order a CSV header line ahead of the rows
   it writes, and a headerless CSV is a file this engine cannot read back, so a CSV sink raises rather than
   writing one. The sink runs at parallelism one so the output is the single file the caller named.
+- Both engines hand every row to the caller's `DataDecoder` as JSON, including rows read from a CSV file,
+  because JSON is the one representation the codecs and the engines both speak. A decoder that supports only
+  one other format therefore decodes nothing, and the dataset's `data` is empty while its count is not. The
+  decoder's `supportsFormat` is not consulted, so the caller gets no signal.
+- A record count on both engines counts the rows the source holds, not the records that decoded. A file with
+  rows the decoder rejects reports a count larger than the records anything can see, which inflates the
+  denominator `validate` scores against. The in-memory engine counts decoded records instead, so the three
+  engines agree only on a source every row of which decodes.
 - The 1000-message cap on `validate` truncates deterministically on Spark, which orders by message before
   capping, and not on Flink, which has no cheap global sort over a `DataStream`. Which messages survive is
   unspecified there. On both engines the cap cannot change `passed` or `score`.
diff --git a/modules/engines-flink/src/test/scala/com/flowforge/engines/flink/CrossEngineParitySpec.scala b/modules/engines-flink/src/test/scala/com/flowforge/engines/flink/CrossEngineParitySpec.scala
index da4d6077..9c903e20 100644
--- a/modules/engines-flink/src/test/scala/com/flowforge/engines/flink/CrossEngineParitySpec.scala
+++ b/modules/engines-flink/src/test/scala/com/flowforge/engines/flink/CrossEngineParitySpec.scala
@@ -193,6 +193,29 @@ class CrossEngineParitySpec extends AnyFunSuite with Matchers {
     fromMemory.violations shouldBe fromFlink.violations
   }
 
+  test("a violation many records share is reported once, with the number of records behind it") {
+    // Two Bobs. A keyed sum that reported running totals would answer `(name is Bob, 1)` and
+    // `(name is Bob, 2)`, so this is what pins the counting job to one final total per key.
+    val source = csvFixture(List(Person(1, "Bob"), Person(2, "Bob"), Person(3, "Carol")))
+
+    val fromFlink = flink.validate(flink.read[Person](source).unsafeRunSync(), nameIsNotBob).unsafeRunSync()
+
+    fromFlink.violations.map(v => (v.message, v.recordsAffected)) shouldBe List(("name is Bob", 2L))
+    fromFlink.score shouldBe (1.0 / 3.0)
+  }
+
+  test("a header the file pads with spaces is still a header") {
+    val file = Files.createTempFile("ff-flink-padded", ".csv")
+    Files.write(file, "id, name\n1,Alice\n2,Bob".getBytes(StandardCharsets.UTF_8))
+    val source = LocalDataSource(file.toString, DataFormat.CSV)
+
+    val dataset = flink.read[Person](source).unsafeRunSync()
+
+    // Reading the header as a record would make this 3, and would hand the decoder `{"id":"id"}`.
+    flink.count(dataset) shouldBe 2L
+    dataset.data.sortBy(_.id) shouldBe List(Person(1, "Alice"), Person(2, "Bob"))
+  }
+
   test("the flink sink writes every record, and the records read back") {
     val source = csvFixture()
     val out    = Files.createTempFile("ff-flink-out", ".jsonl")