diff --git a/docs/plan/v1.0-readiness.md b/docs/plan/v1.0-readiness.md index 9c1042c3..94c296db 100644 --- a/docs/plan/v1.0-readiness.md +++ b/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 diff --git a/modules/core/src/main/scala/com/flowforge/core/algebra/DataAlgebra.scala b/modules/core/src/main/scala/com/flowforge/core/algebra/DataAlgebra.scala index deeb8db0..953cd2b8 100644 --- a/modules/core/src/main/scala/com/flowforge/core/algebra/DataAlgebra.scala +++ b/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. */ diff --git a/modules/core/src/main/scala/com/flowforge/core/impl/InMemoryDataAlgebra.scala b/modules/core/src/main/scala/com/flowforge/core/impl/InMemoryDataAlgebra.scala index d4f9ed5f..bc1a6704 100644 --- a/modules/core/src/main/scala/com/flowforge/core/impl/InMemoryDataAlgebra.scala +++ b/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], diff --git a/modules/core/src/main/scala/com/flowforge/core/impl/InMemoryPipelineBuilder.scala b/modules/core/src/main/scala/com/flowforge/core/impl/InMemoryPipelineBuilder.scala index 29360d31..a784cc79 100644 --- a/modules/core/src/main/scala/com/flowforge/core/impl/InMemoryPipelineBuilder.scala +++ b/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 diff --git a/modules/core/src/main/scala/com/flowforge/core/syntax/PipelineSyntax.scala b/modules/core/src/main/scala/com/flowforge/core/syntax/PipelineSyntax.scala index 1bb4c4a5..cb812247 100644 --- a/modules/core/src/main/scala/com/flowforge/core/syntax/PipelineSyntax.scala +++ b/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) diff --git a/modules/core/src/test/scala/com/flowforge/core/impl/InMemoryDataAlgebraSpec.scala b/modules/core/src/test/scala/com/flowforge/core/impl/InMemoryDataAlgebraSpec.scala index bfe32b6d..3247fec5 100644 --- a/modules/core/src/test/scala/com/flowforge/core/impl/InMemoryDataAlgebraSpec.scala +++ b/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"), 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 88509e78..a8e4f825 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 @@ -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]]] = diff --git a/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkDataAlgebra.scala b/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkDataAlgebra.scala index 27ff21fd..70abaaa4 100644 --- a/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkDataAlgebra.scala +++ b/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], diff --git a/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkFrameOps.scala b/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkFrameOps.scala index 9dc540b1..dacea16a 100644 --- a/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkFrameOps.scala +++ b/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. * diff --git a/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkPipelineBuilder.scala b/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkPipelineBuilder.scala index fc07abbd..6f5383ae 100644 --- a/modules/engines-spark/src/main/scala/com/flowforge/engines/spark/SparkPipelineBuilder.scala +++ b/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]( diff --git a/modules/engines-spark/src/test/scala/com/flowforge/engines/spark/EngineParitySpec.scala b/modules/engines-spark/src/test/scala/com/flowforge/engines/spark/EngineParitySpec.scala index 12a7da54..9e3f6169 100644 --- a/modules/engines-spark/src/test/scala/com/flowforge/engines/spark/EngineParitySpec.scala +++ b/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) diff --git a/modules/engines-spark/src/test/scala/com/flowforge/engines/spark/SparkLocalBatchSpec.scala b/modules/engines-spark/src/test/scala/com/flowforge/engines/spark/SparkLocalBatchSpec.scala index 1b2df52a..17854e7e 100644 --- a/modules/engines-spark/src/test/scala/com/flowforge/engines/spark/SparkLocalBatchSpec.scala +++ b/modules/engines-spark/src/test/scala/com/flowforge/engines/spark/SparkLocalBatchSpec.scala @@ -15,7 +15,7 @@ import com.flowforge.core.algebra.{ UnsupportedFormat, } import com.flowforge.core.instances.EffectInstances -import com.flowforge.core.types.PipelineTypes.QualityCheck +import com.flowforge.core.types.PipelineTypes.{ DataContract, QualityCheck } import com.flowforge.core.types.RefinedTypes.BucketName import com.flowforge.core.types._ import org.apache.spark.sql.SparkSession @@ -232,6 +232,30 @@ class SparkLocalBatchSpec extends AnyFunSuite with Matchers with BeforeAndAfterA results.head.message should include("id is 150") } + test("validate sees a contract violation past the sample") { + val rows = (1 to 150).map(i => Person(i, s"p$i")).toList + val source = LocalDataSource(csvFixture(rows).toString, DataFormat.CSV) + val alg = SparkDataAlgebra.createSparkDataAlgebra[IO](spark).algebra + + val dataset = alg.read[Person](source).unsafeRunSync() + + assume(!dataset.data.exists(_.id > 100), "the sample must not hold the violating records") + + val idUnder101: DataContract[Person] = p => + if (p.id <= 100) ().validNel + else + ValidationError + .SchemaViolation("id", "at most 100", p.id.toString, message = "id is too large").invalidNel + + val result = alg.validate(dataset, idUnder101).unsafeRunSync() + + result.passed shouldBe false + // 50 of 150 records break the contract, and all 50 are outside the sample the driver holds. + result.violations.map(v => (v.message, v.recordsAffected)) shouldBe List(("id is too large", 50L)) + result.score shouldBe (100.0 / 150.0) + result.data shouldBe dataset + } + test("an empty local source reads as an empty dataset") { val source = LocalDataSource(csvFixture(Nil).toString, DataFormat.CSV) val alg = SparkDataAlgebra.createSparkDataAlgebra[IO](spark).algebra