fix: validate against the data contract

docs/plan/v1.0-readiness.md

@@ -33,11 +33,6 @@ 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.
 
-### `validate` does not validate
-`validate` returns `passed = true` with no violations on both the in-memory and the Spark engine, whatever
-the contract says. A caller cannot tell a conforming dataset from a non-conforming one. Either implement it
-against the contract or remove it from the algebra.
-
 ### 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
@@ -52,6 +47,11 @@ These are deliberate for v1.0 and recorded so nobody rediscovers them as bugs.
   Spark-backed dataset. They were not part of the sample bug and are not pushed into the frame.
   `sortBy` takes an arbitrary `Ordering[K]`, which a frame cannot express, so that one needs a different
   signature rather than a different implementation.
+- `validate` applies the contract to every record on both engines and reports an exact count of the records
+  that broke it, but the list of distinct violation messages is capped at 1000. A message built from a
+  record's own values is otherwise as large as the dataset. The cap cannot change `passed` or `score`, which
+  come from the count. Every violation is reported as `Critical`: a contract is the shape the caller declared
+  the data has, so the severity levels have no other source here.
 - `DataProfile.nullCount` is always 0 on both engines. The field has no agreed meaning - records with a null
   field, or null records - and inventing one in the Spark engine alone would make the two engines disagree.
 - When the Deequ module is on the classpath, `runQualityChecks` on Spark runs Deequ's `NotNull` constraints

modules/core/src/main/scala/com/flowforge/core/algebra/DataAlgebra.scala

@@ -238,8 +238,18 @@ trait DataAlgebra[F[_]] extends CDCOperations[F] with TableOperations[F] with Da
   /**
    * Validate dataset against external data contract service. Uses F[_] because it may involve external
    * validation service.
+   *
+   * The contract is applied to every record, so `DataDecoder[A]` is needed for the reason given on
+   * [[filter]]. The result is built by [[DataAlgebra.contractResult]] on every engine, so what `passed`,
+   * `score` and `recordsAffected` mean does not depend on which engine ran the contract.
+   *
+   * The returned dataset is the one that went in. Validating reports on the records; it does not drop the
+   * ones that failed.
    */
-  def validate[A](dataset: Dataset[A], contract: PDataContract[A]): F[QualityResult[Dataset[A]]]
+  def validate[A: DataDecoder](
+    dataset: Dataset[A],
+    contract: PDataContract[A],
+  ): F[QualityResult[Dataset[A]]]
 
   /**
    * Run quality checks that may involve external services. Uses F[_] because checks may involve external
@@ -388,6 +398,52 @@ object DataAlgebra {
     violations: List[QualityViolation],
     score: Double)
 
+  /**
+   * Build the result of validating a dataset against a contract, from counts taken over its records.
+   *
+   * This is the one definition of what a contract result means, and every engine calls it rather than
+   * assembling a `QualityResult` itself. The engines differ in how they reach the records - the driver holds
+   * them, or a frame does - and that is the only thing they should differ in. Two engines each building their
+   * own result is how they come to disagree about the same dataset.
+   *
+   * `score` is the share of records the contract accepted, so an empty dataset scores 1.0: a contract cannot
+   * be violated by a record that is not there. `recordsAffected` on a violation counts the records that
+   * reported that message, not the times it was reported, because one record can break a rule once.
+   *
+   * Every violation is `Critical`. A contract is the shape a caller declared the data has, so a record that
+   * breaks it is wrong rather than suspicious, and the severity levels have no other source here.
+   *
+   * @param totalRecords
+   *   records the contract was applied to
+   * @param violatingRecords
+   *   records that reported at least one violation
+   * @param messageCounts
+   *   how many records reported each distinct message
+   */
+  def contractResult[A](
+    dataset: A,
+    totalRecords: Long,
+    violatingRecords: Long,
+    messageCounts: List[(String, Long)],
+  ): QualityResult[A] = {
+    val violations = messageCounts.map {
+      case (message, affected) =>
+        QualityViolation(
+          rule = "contract",
+          message = message,
+          severity = ViolationSeverity.Critical,
+          recordsAffected = affected,
+        )
+    }
+    val score = if (totalRecords <= 0L) 1.0 else (totalRecords - violatingRecords).toDouble / totalRecords
+    QualityResult(
+      data = dataset,
+      passed = violatingRecords == 0L,
+      violations = violations,
+      score = score,
+    )
+  }
+
   /**
    * Quality check result.
    */

modules/core/src/main/scala/com/flowforge/core/impl/InMemoryDataAlgebra.scala

@@ -283,11 +283,22 @@ final class InMemoryDataAlgebra[F[_]](implicit F: EffectSystem[F]) extends DataA
   override def queryLineage(query: LineageQuery): F[List[LineageRecord]] = F.pure(Nil)
 
   // ---------- Quality ----------
-  override def validate[A](
+  // The contract used to be ignored, so `passed` was true whatever the records were and a caller could not
+  // tell a conforming dataset from a non-conforming one. It is applied to every record now.
+  override def validate[A: DataDecoder](
     dataset: Dataset[A],
     contract: PDataContract[A],
-  ): F[QualityResult[Dataset[A]]] =
-    F.pure(QualityResult(dataset, passed = true, violations = Nil, score = 1.0))
+  ): F[QualityResult[Dataset[A]]] = F.pure {
+    val perRecord        = dataset.data.map(a => contract(a).fold(_.toList.map(_.message), _ => Nil))
+    val violatingRecords = perRecord.count(_.nonEmpty).toLong
+    val messageCounts = perRecord.flatMap(_.distinct).groupBy(identity).view.mapValues(_.size.toLong).toList
+    DataAlgebra.contractResult(
+      dataset,
+      dataset.data.size.toLong,
+      violatingRecords,
+      messageCounts.sortBy(_._1),
+    )
+  }
 
   override def runQualityChecks[A: DataDecoder](
     dataset: Dataset[A],

modules/core/src/main/scala/com/flowforge/core/impl/InMemoryPipelineBuilder.scala

@@ -2,7 +2,7 @@
 package com.flowforge.core.impl
 
 import cats.data.Kleisli
-import com.flowforge.core.algebra.{ DataAlgebra, DataEncoder, EffectSystem }
+import com.flowforge.core.algebra.{ DataAlgebra, DataDecoder, DataEncoder, EffectSystem }
 import com.flowforge.core.exec.{ ExecutableStage, StageComposer }
 import com.flowforge.core.types._
 import com.flowforge.framework.{ Pipeline, PipelineMetadata }
@@ -170,10 +170,13 @@ object InMemoryTypedBuilder {
     }
 
     /**
-     * Add data quality validation
+     * Add data quality validation.
+     *
+     * The decoder is what lets the engine apply the contract to every record. See `DataAlgebra.validate`.
      */
     def addQualityCheck(
       contract: com.flowforge.core.types.PipelineTypes.DataContract[E],
+    )(implicit decoder: DataDecoder[E],
     ): InMemoryTypedBuilder[F, In, DataAlgebra.Dataset[E]] = {
       val ef      = builder.ef
       val algebra = builder.dataAlgebra

modules/core/src/main/scala/com/flowforge/core/syntax/PipelineSyntax.scala

@@ -359,10 +359,13 @@ object PipelineSyntax {
     }
 
     /**
-     * Apply quality check
+     * Apply quality check.
+     *
+     * The decoder is what lets the engine apply the contract to every record. See `DataAlgebra.validate`.
      */
     def quality(
       contract: DataContract[A],
+    )(implicit decoder: DataDecoder[A],
     ): F[DatasetPipelineBuilder[F, A]] =
       dataAlgebra.validate(dataset, contract).flatMap { result =>
         if (result.passed) F.pure(this)

modules/core/src/test/scala/com/flowforge/core/impl/InMemoryDataAlgebraSpec.scala

@@ -892,6 +892,47 @@ test2"""
     result.data shouldBe dataset
   }
 
+  test("validate reports the records the contract rejects") {
+    val rows = List("ok", "bad", "bad", "ok")
+    val dataset = SimpleDataset(
+      rows,
+      DataSchema.builder.build,
+      DatasetMetadata(rows.size.toLong, DataSchema.builder.build, 1, Instant.now()),
+    )
+
+    // The contract used to be ignored, so this passed. A rejected record has to show up now.
+    val rejectBad: PDataContract[String] = s =>
+      if (s != "bad") ().validNel
+      else ValidationError.SchemaViolation("value", "not bad", s, message = "value is bad").invalidNel
+
+    val result = algebra.validate(dataset, rejectBad).unsafeRunSync()
+
+    result.passed shouldBe false
+    result.score shouldBe 0.5
+    // One message, counted once per record that reported it, not once per report.
+    result.violations.map(v => (v.message, v.recordsAffected)) shouldBe List(("value is bad", 2L))
+    result.violations.map(_.severity) shouldBe List(ViolationSeverity.Critical)
+    // Validating reports on the records; it does not drop the ones that failed.
+    result.data shouldBe dataset
+  }
+
+  test("validate passes on an empty dataset") {
+    val dataset = SimpleDataset(
+      List.empty[String],
+      DataSchema.builder.build,
+      DatasetMetadata(0, DataSchema.builder.build, 1, Instant.now()),
+    )
+    val rejectEverything: PDataContract[String] =
+      s => ValidationError.SchemaViolation("value", "nothing", s, message = "rejected").invalidNel
+
+    val result = algebra.validate(dataset, rejectEverything).unsafeRunSync()
+
+    // A contract cannot be violated by a record that is not there.
+    result.passed shouldBe true
+    result.score shouldBe 1.0
+    result.violations shouldBe List.empty
+  }
+
   test("runQualityChecks returns results for all checks") {
     val dataset = SimpleDataset(
       List("test"),

modules/engines-flink/src/main/scala/com/flowforge/engines/flink/FlinkDataAlgebra.scala

@@ -129,7 +129,7 @@ final class FlinkDataAlgebra[F[_]](implicit F: EffectSystem[F]) extends DataAlge
   override def queryLineage(query: LineageQuery): F[List[LineageRecord]] =
     delegate.queryLineage(query)
 
-  override def validate[A](
+  override def validate[A: DataDecoder](
     dataset: Dataset[A],
     contract: PDataContract[A],
   ): F[QualityResult[Dataset[A]]] =

modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkDataAlgebra.scala

@@ -890,18 +890,40 @@ object SparkDataAlgebra {
       // DATA QUALITY (F[_] Required)
       // ========================================
 
-      override def validate[A](
+      // The contract used to be ignored, so `passed` was true whatever the records were. It is applied to
+      // every record now, over the frame rather than over the driver's sample of it.
+      override def validate[A: DataDecoder](
         dataset: DataAlgebra.Dataset[A],
         contract: PipelineDataContract[A],
-      ): F[DataAlgebra.QualityResult[DataAlgebra.Dataset[A]]] =
-        F.pure(
-          DataAlgebra.QualityResult(
-            data = dataset,
-            passed = true,
-            violations = List.empty,
-            score = 1.0,
-          ),
-        )
+      ): F[DataAlgebra.QualityResult[DataAlgebra.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 pds: ProductionSparkDataset[A] =>
+            F.blocking {
+              val (violatingRecords, messageCounts) =
+                SparkFrameOps.countViolations(spark, pds.sparkDataFrame, messagesOf)
+              DataAlgebra.contractResult(
+                dataset,
+                pds.sparkDataFrame.count(),
+                violatingRecords,
+                messageCounts,
+              )
+            }
+
+          case other =>
+            // A dataset this engine did not produce, so the driver holds it in full.
+            F.pure {
+              val perRecord        = other.data.map(a => messagesOf(a).distinct)
+              val violatingRecords = perRecord.count(_.nonEmpty).toLong
+              val messageCounts =
+                perRecord.flatten.groupBy(identity).view.mapValues(_.size.toLong).toList.sortBy(_._1)
+              DataAlgebra.contractResult(dataset, other.data.size.toLong, violatingRecords, messageCounts)
+            }
+        }
+      }
 
       override def runQualityChecks[A: DataDecoder](
         dataset: DataAlgebra.Dataset[A],

modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkFrameOps.scala

@@ -81,6 +81,50 @@ private[spark] object SparkFrameOps {
    */
   val violationMessageLimit: Int = 1000
 
+  /**
+   * Count the records that report a violation, and the records behind each distinct message.
+   *
+   * `violationsOf` maps a record to the messages for the violations it has, empty when it has none.
+   *
+   * The record count is exact and the message list is capped, for the reason given on [[collectViolations]]:
+   * a dataset where every record violates the contract would otherwise bring one message per record back to
+   * the driver, while whether the contract passed only needs the count.
+   *
+   * A message is counted once per record. A record that reports the same message twice has still broken that
+   * rule once, which is what `recordsAffected` claims to say.
+   *
+   * This walks the frame twice, once for each half of the answer. One pass would mean holding the messages to
+   * count them, which is the cost the cap exists to avoid.
+   *
+   * @return
+   *   the number of records that reported at least one violation, and the record count per message
+   */
+  def countViolations[A](
+    spark: SparkSession,
+    frame: DataFrame,
+    violationsOf: A => List[String],
+    messageLimit: Int = violationMessageLimit,
+  )(implicit decoder: DataDecoder[A],
+  ): (Long, List[(String, Long)]) = {
+    import spark.implicits._
+    val rowDecoder = decoder
+    val perRecord = frame.toJSON.map { json =>
+      decodeRow(json, rowDecoder).toList.flatMap(violationsOf).distinct
+    }
+    val violatingRecords = perRecord.filter((messages: Seq[String]) => messages.nonEmpty).count()
+    val messageCounts = perRecord
+      .flatMap(identity)
+      .groupByKey(identity)
+      .count()
+      // `key` is the message. Ordering before the cap keeps which messages survive it deterministic, and
+      // leaves the result in the same order the in-memory algebra produces.
+      .orderBy("key")
+      .limit(messageLimit)
+      .collect()
+      .toList
+    (violatingRecords, messageCounts.map { case (message, count) => (message, count) })
+  }
+
   /**
    * Apply every check to every decoded record.
    *

modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkPipelineBuilder.scala

@@ -4,7 +4,7 @@ import cats.data.Kleisli
 import cats.effect.Resource
 import cats.implicits._
 import com.flowforge.core.algebra.DataAlgebra.WriteOptions
-import com.flowforge.core.algebra.{ DataAlgebra, EffectSystem }
+import com.flowforge.core.algebra.{ DataAlgebra, DataDecoder, EffectSystem }
 import com.flowforge.core.exec.{ ExecutableStage, StageComposer }
 import com.flowforge.core.types._
 import com.flowforge.framework.{ Pipeline, PipelineMetadata }
@@ -103,10 +103,13 @@ class SparkTypedBuilder[F[_]: EffectSystem, In, Out] private[spark] (
   }
 
   /**
-   * Add a data quality check using Spark's distributed validation
+   * Add a data quality check using Spark's distributed validation.
+   *
+   * The decoder is what lets the engine apply the contract to every record. See `DataAlgebra.validate`.
    */
   def addQualityCheck(
     contract: com.flowforge.core.types.PipelineTypes.DataContract[Out],
+  )(implicit decoder: DataDecoder[Out],
   ): SparkTypedBuilder[F, In, Out] = {
     EffectSystem[F]
     val stage = SparkStage.Quality[F, Out](

modules/engines-spark/src/test/scala/com/flowforge/engines/spark/EngineParitySpec.scala

@@ -8,7 +8,7 @@ import cats.syntax.validated._
 import com.flowforge.core.algebra.{ DataAlgebra, EffectSystem }
 import com.flowforge.core.impl.InMemoryDataAlgebra
 import com.flowforge.core.instances.EffectInstances
-import com.flowforge.core.types.PipelineTypes.QualityCheck
+import com.flowforge.core.types.PipelineTypes.{ DataContract => PDataContract, QualityCheck }
 import com.flowforge.core.types._
 import org.apache.spark.sql.SparkSession
 import org.scalatest.BeforeAndAfterAll
@@ -97,6 +97,29 @@ class EngineParitySpec extends AnyFunSuite with Matchers with BeforeAndAfterAll
     memoryResults.map(r => r.checkName -> r.passed) shouldBe sparkResults.map(r => r.checkName -> r.passed)
   }
 
+  test("both engines report the same contract result") {
+    val source = fixture
+
+    // 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 fromSpark =
+      sparkAlgebra.validate(sparkAlgebra.read[Person](source).unsafeRunSync(), nameIsNotBob).unsafeRunSync()
+    val fromMemory =
+      memoryAlgebra.validate(memoryAlgebra.read[Person](source).unsafeRunSync(), nameIsNotBob).unsafeRunSync()
+
+    fromSpark.passed shouldBe false
+    fromSpark.score shouldBe (2.0 / 3.0)
+    fromSpark.violations.map(v => (v.message, v.recordsAffected)) shouldBe List(("name is Bob", 1L))
+
+    // `data` differs - each engine returns its own dataset - so compare what the contract said about it.
+    fromMemory.passed shouldBe fromSpark.passed
+    fromMemory.score shouldBe fromSpark.score
+    fromMemory.violations shouldBe fromSpark.violations
+  }
+
   test("a format neither engine supports fails on both, with the error type each one raises") {
     val source = LocalDataSource(fixture.location, DataFormat.Avro)