add variable substitution for minNumRows, fixes #58

Author: phpisciuneri

src/main/scala/com/target/data_validator/validator/ColumnBased.scala

@@ -35,26 +35,40 @@ abstract class ColumnBased(column: String, condTest: Expression) extends CheapCh
   }
 }
 
-case class MinNumRows(minNumRows: Long) extends ColumnBased("", ValidatorBase.L0) {
+case class MinNumRows(minNumRows: Json) extends ColumnBased("", ValidatorBase.L0) {
   override def name: String = "MinNumRows"
 
-  override def substituteVariables(dict: VarSubstitution): ValidatorBase = this
+  override def substituteVariables(dict: VarSubstitution): ValidatorBase = {
+    val ret = MinNumRows(getVarSubJson(minNumRows, "minNumRows", dict))
+    getEvents.foreach(ret.addEvent)
+    ret
+  }
 
   override def configCheck(df: DataFrame): Boolean = {
-    if (minNumRows <= 0) {
-      val msg = s"MinNumRows: $minNumRows <= 0"
+
+    def notNaturalNumber(): Unit = {
+      val msg = "minNumRows must be a natural number"
       logger.error(msg)
       addEvent(ValidatorError(msg))
-      failed = true
-      true
-    } else {
-      false
     }
+
+    minNumRows.asNumber match {
+      case Some(jsonNumber) => jsonNumber.toLong match {
+        case Some(x) if x > 0 =>
+        case _ => notNaturalNumber()
+      }
+      case _ => notNaturalNumber()
+    }
+    failed
   }
 
   override def quickCheck(row: Row, count: Long, idx: Int): Boolean = {
-    failed = count < minNumRows
-    val pctError = if (failed) calculatePctError(minNumRows, count) else "0.00%"
+    // Convert to `JsonNumber` then to `Long`
+    // safe because already handled in `configCheck`
+    val minNumRowsLong = minNumRows.asNumber.get.toLong.get
+
+    failed = count < minNumRowsLong
+    val pctError = if (failed) calculatePctError(minNumRowsLong, count) else "0.00%"
     addEvent(ValidatorCounter("rowCount", count))
     val msg = s"MinNumRowsCheck Expected: $minNumRows Actual: $count Relative Error: $pctError"
     val data = ListMap("expected" -> minNumRows.toString, "actual" -> count.toString, "relative_error" -> pctError)
@@ -64,7 +78,7 @@ case class MinNumRows(minNumRows: Long) extends ColumnBased("", ValidatorBase.L0
 
   override def toJson: Json = Json.obj(
     ("type", Json.fromString("rowCount")),
-    ("minNumRows", Json.fromLong(minNumRows)),
+    ("minNumRows", minNumRows),
     ("failed", Json.fromBoolean(failed)),
     ("events", this.getEvents.asJson)
   )

src/test/scala/com/target/data_validator/ConfigParserSpec.scala

@@ -28,7 +28,7 @@ class ConfigParserSpec extends FunSpec with BeforeAndAfterAll {
         "bar",
         Some(List("one", "two")),
         None,
-        List(MinNumRows(10294), NullCheck("mdse_item_i", None)) // scalastyle:ignore magic.number
+        List(MinNumRows(Json.fromInt(10294)), NullCheck("mdse_item_i", None)) // scalastyle:ignore magic.number
       ),
       ValidatorOrcFile("LocalFile.orc", None, Some("foo < 10"), List(NullCheck("start_d", None))),
       ValidatorParquetFile("LocFile.parquet", None, Some("bar < 10"), List(NullCheck("end_d", None)))

src/test/scala/com/target/data_validator/ConfigVarSubSpec.scala

@@ -62,9 +62,9 @@ class ConfigVarSubSpec extends FunSpec with Matchers with TestingSparkSession {
 
         describe("MinNumRows") {
 
-          it("MinNumRows doesn't support substitutions so it should be equal, no changes.") {
-            val sut = MinNumRows(100) // scalastyle:ignore
-            assert(sut.substituteVariables(dict) == sut)
+          it("should substitute variables properly") {
+            val sut = MinNumRows(Json.fromString("$one"))
+            assert(sut.substituteVariables(dict) == MinNumRows(Json.fromInt(1)))
           }
 
         }

src/test/scala/com/target/data_validator/ValidatorBaseSpec.scala

@@ -171,19 +171,27 @@ class ValidatorBaseSpec extends FunSpec with Matchers with TestingSparkSession {
 
   describe("ValidatorMinNumRows") {
 
+    val df = spark.createDataFrame(sc.parallelize(List(Row("Doug", 50), Row("Collin", 32))), schema) //scalastyle:ignore
+
+    it("configCheck() should fail for minNumRows as non-numeric") {
+      val dict = new VarSubstitution
+      val config = mkConfig(df, List(MinNumRows(Json.fromString("badinput"))))
+      assert(config.configCheck(spark, dict))
+      assert(config.failed)
+      assert(config.tables.head.failed)
+    }
+
     it("configCheck() should fail for negative minNumRows") {
       val dict = new VarSubstitution
-      val df = spark.createDataFrame(sc.parallelize(List(Row("Doug", 50), Row("Collin", 32))), schema) //scalastyle:ignore
-      val config = mkConfig(df, List(MinNumRows(-10))) //scalastyle:ignore
+      val config = mkConfig(df, List(MinNumRows(Json.fromLong(-10)))) // scalastyle:ignore magic.number
       assert(config.configCheck(spark, dict))
       assert(config.failed)
       assert(config.tables.head.failed)
     }
 
     it("quickCheck() should fail when rowCount < minNumRows") {
       val dict = new VarSubstitution
-      val df = spark.createDataFrame(sc.parallelize(List(Row("Doug", 50), Row("Collin", 32))), schema) //scalastyle:ignore
-      val minNumRowsCheck = MinNumRows(10) // scalastyle:ignore magic.number
+      val minNumRowsCheck = MinNumRows(Json.fromLong(10)) // scalastyle:ignore magic.number
       val config = mkConfig(df, List(minNumRowsCheck))
       assert(config.quickChecks(spark, dict))
       assert(config.failed)
@@ -197,8 +205,7 @@ class ValidatorBaseSpec extends FunSpec with Matchers with TestingSparkSession {
 
     it("quickCheck() should succeed when rowCount > minNumRows") {
       val dict = new VarSubstitution
-      val df = spark.createDataFrame(sc.parallelize(List(Row("Doug", 50), Row("Collin", 32))), schema) //scalastyle:ignore
-      val minNumRowsCheck = MinNumRows(1)
+      val minNumRowsCheck = MinNumRows(Json.fromInt(1))
       val config = mkConfig(df, List(minNumRowsCheck))
       assert(!config.configCheck(spark, dict))
       assert(!config.quickChecks(spark, dict))