PR6 fixes and docs

Author: GeekSheikh

src/main/scala/com/databricks/labs/validation/QuickTest.scala

@@ -0,0 +1,37 @@
+package com.databricks.labs.validation
+
+import com.databricks.labs.validation.utils.Structures.{Bounds, MinMaxRuleDef}
+import com.databricks.labs.validation.utils.SparkSessionWrapper
+import org.apache.spark.sql.functions._
+
+object QuickTest extends App with SparkSessionWrapper {
+
+  import spark.implicits._
+
+  val testDF = Seq(
+    (1, 2, 3),
+    (4, 5, 6),
+    (7, 8, 9)
+  ).toDF("retail_price", "scan_price", "cost")
+
+  Rule("Reasonable_sku_counts", count(col("sku")), Bounds(lower = 20.0, upper = 200.0))
+
+  val minMaxPriceDefs = Array(
+    MinMaxRuleDef("MinMax_Retail_Price_Minus_Scan_Price", col("retail_price")-col("scan_price"), Bounds(0.0, 29.99)),
+    MinMaxRuleDef("MinMax_Scan_Price_Minus_Retail_Price", col("scan_price")-col("retail_price"), Bounds(0.0, 29.99))
+  )
+
+  val someRuleSet = RuleSet(testDF)
+    .add(Rule("retail_pass", col("retail_price"), Bounds(lower = 1.0, upper = 7.0)))
+    .add(Rule("retail_agg_pass_high", max(col("retail_price")), Bounds(lower = 0.0, upper = 7.1)))
+    .add(Rule("retail_agg_pass_low", min(col("retail_price")), Bounds(lower = 0.0, upper = 7.0)))
+    .add(Rule("retail_fail_low", col("retail_price"), Bounds(lower = 1.1, upper = 7.0)))
+    .add(Rule("retail_fail_high", col("retail_price"), Bounds(lower = 0.0, upper = 6.9)))
+    .add(Rule("retail_agg_fail_high", max(col("retail_price")), Bounds(lower = 0.0, upper = 6.9)))
+    .add(Rule("retail_agg_fail_low", min(col("retail_price")), Bounds(lower = 1.1, upper = 7.0)))
+    .addMinMaxRules(minMaxPriceDefs: _*)
+  val (rulesReport, passed) = someRuleSet.validate()
+
+  testDF.show(20, false)
+  rulesReport.show(20, false)
+}

src/main/scala/com/databricks/labs/validation/Validator.scala

@@ -102,19 +102,33 @@ class Validator(ruleSet: RuleSet, detailLvl: Int) extends SparkSessionWrapper {
       // Results must have Invalid_Count & Failed
       rule.ruleType match {
         case RuleType.ValidateBounds =>
+          // Rule evaluation for NON-AGG RULES ONLY
           val invalid = rule.inputColumn < rule.boundaries.lower || rule.inputColumn > rule.boundaries.upper
+          // This is the first select it must come before subsequent selects as it aliases the original column name
+          // to that of the rule name. ADDITIONALLY, this evaluates the boundary rule WHEN the input col is not an Agg.
+          // This can be confusing because for Non-agg columns it renames the column to the rule_name AND returns a 0
+          // or 1 (not the original value)
+          // IF the rule is NOT an AGG then the column is simply aliased to the rule name and no evaluation takes place
+          // here.
+          val first = if (!rule.isAgg) { // Not Agg
+            sum(when(invalid, 1).otherwise(0)).alias(rule.ruleName)
+          } else { // Is Agg
+            rule.inputColumn.alias(rule.ruleName)
+          }
+          // WHEN RULE IS AGG -- this is where the evaluation happens. The input column was renamed to the name of the
+          // rule in the required previous select.
+          // IMPORTANT: REMEMBER - that agg expressions evaluate to a single output value thus the invalid_count in
+          // cases where agg is used cannot be > 1 since the sum of a single value cannot exceed 1.
+
+          // WHEN RULE NOT AGG - determine if the result of "first" select (0 or 1) is > 0, if it is, the rule has
+          // failed since the sum(1 or more 1s) means that 1 or more rows have failed thus the rule has failed
           val failed = if (rule.isAgg) {
             when(
               col(rule.ruleName) < rule.boundaries.lower || col(rule.ruleName) > rule.boundaries.upper, true)
               .otherwise(false).alias("Failed")
           } else{
             when(col(rule.ruleName) > 0,true).otherwise(false).alias("Failed")
           }
-          val first = if (!rule.isAgg) { // Not Agg
-            sum(when(invalid, 1).otherwise(0)).alias(rule.ruleName)
-          } else { // Is Agg
-            rule.inputColumn.alias(rule.ruleName)
-          }
           val results = if (rule.isAgg) {
             Seq(when(failed, 1).otherwise(0).cast(LongType).alias("Invalid_Count"), failed)
           } else {

src/test/scala/com/databricks/labs/validation/ValidatorTestSuite.scala

@@ -50,10 +50,10 @@ class ValidatorTestSuite extends org.scalatest.FunSuite with SparkSessionFixture
     assert(rulesReport.count() == 10)
   }
 
-  test("The input rule should have 3 invalid count for MinMax_Scan_Price_Minus_Retail_Price_min for failing complex type.") {
+  test("The input rule should have 1 invalid count for MinMax_Scan_Price_Minus_Retail_Price_min and max for failing complex type.") {
     val expectedDF = Seq(
-      ("MinMax_Retail_Price_Minus_Scan_Price_max","bounds",ValidationValue(null,null,Array(0.0, 29.99),null),0,false),
-      ("MinMax_Retail_Price_Minus_Scan_Price_min","bounds",ValidationValue(null,null,Array(0.0, 29.99),null),3,true),
+      ("MinMax_Retail_Price_Minus_Scan_Price_max","bounds",ValidationValue(null,null,Array(0.0, 29.99),null),1,true),
+      ("MinMax_Retail_Price_Minus_Scan_Price_min","bounds",ValidationValue(null,null,Array(0.0, 29.99),null),1,true),
       ("MinMax_Scan_Price_Minus_Retail_Price_max","bounds",ValidationValue(null,null,Array(0.0, 29.99),null),0,false),
       ("MinMax_Scan_Price_Minus_Retail_Price_min","bounds",ValidationValue(null,null,Array(0.0, 29.99),null),0,false)
     ).toDF("Rule_Name","Rule_Type","Validation_Values","Invalid_Count","Failed")