Suggested constraint code must work on data that constraints were suggested from
- Dominant language
- Scala
- Stars
- 3.6k
- Forks
- 586
- Avg merge
- 13d 13h
- Merged PRs (30d)
- 1
Description
The `Check` Scala code that is associated with suggested constraints must always be able to pass the data that the constraint was suggested from. This occurs most of the time, but I found a counter example on a very trivial dataset.
Given the following CSV data:
```
Zip_Code,Total_Population,Median_Age,Total_Males,Total_Females,Total_Households,Average_Household_Size
91371,1,73.5,0,1,1,1
90001,57110,26.6,28468,28642,12971,4.4
90002,51223,25.5,24876,26347,11731,4.36
90003,66266,26.3,32631,33635,15642,4.22
90004,62180,34.8,31302,30878,22547,2.73
90005,37681,33.9,19299,18382,15044,2.5
90006,59185,32.4,30254,28931,18617,3.13
90007,40920,24,20915,20005,11944,3
90008,32327,39.7,14477,17850,13841,2.33
90010,3800,37.8,1874,1926,2014,1.87
90011,103892,26.2,52794,51098,22168,4.67
90012,31103,36.3,19493,11610,10327,2.12
90013,11772,44.6,7629,4143,6416,1.26
90014,7005,44.8,4471,2534,4109,1.34
90015,18986,31.3,9833,9153,7420,2.45
90016,47596,33.9,22778,24818,16145,2.93
90017,23768,29.4,12818,10950,9338,2.53
90018,49310,33.2,23770,25540,15493,3.12
90019,64458,35.8,31442,33016,23344,2.7
90020,38967,34.6,19381,19586,16514,2.35
90021,3951,44.3,2790,1161,1561,1.57
90022,67179,29.8,33216,33963,17023,3.94
90023,45903,28.4,23037,22866,10727,4.26
90024,47452,23.6,22248,25204,17903,2.03
90025,42147,34.7,20859,21288,21228,1.97
90026,67869,34,34515,33354,24956,2.68
90027,45151,38.3,22362,22789,21929,1.99
90028,28714,34,16056,12658,14964,1.78
90029,38617,34.6,19575,19042,13883,2.7
```
Running `ConstraintSuggestionRunner()` on a `DataFrame` created from it, with `Rules.DEFAULT`, will produce a few suggested constraints whose code will _not_ verify the _exact_ data it was built from.
The following is a program that demonstrates the failure:
```scala
import java.io.{File, PrintWriter}
import com.amazon.deequ.VerificationResult
import com.amazon.deequ.checks.CheckStatus
import com.amazon.deequ.suggestions.{ConstraintSuggestionResult, ConstraintSuggestionRunner, Rules}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, SparkSession}
import tools.reflect.ToolBox
import scala.reflect.runtime.currentMirror
import scala.util.Try
object FailureOfSuggestedConstraintCode {
def main(args: Array[String]): Unit = {
val sesh: SparkSession = {
Logger.getLogger("org").setLevel(Level.OFF)
Logger.getLogger("akka").setLevel(Level.OFF)
SparkSession
.builder()
.config("spark.master", "local")
.getOrCreate()
}
val data: DataFrame = {
val f = writeToTempFile(csvData)
sesh.read.option("header", true.toString).csv(f.getCanonicalPath)
}
println(s"Data has ${data.rdd.count()} rows")
val constraintSuggestions = suggest(data)
println(
s"Found ${constraintSuggestions.constraintSuggestions.view.map { _._2.size }.sum} suggested constraints"
)
val verifier = generateVerifier(constraintSuggestions).get
val result = verifier(data)
println("----------------------------------")
println(s"Verification result status: ${result.status}")
result.checkResults.filter { case (_, cr) => cr.status != CheckStatus.Success }.foreach {
case (check, _) => println(s"Failed check: ${check.description}")
}
}
//
//
//
// Helpers
//
//
//
def suggest(data: DataFrame): ConstraintSuggestionResult = {
ConstraintSuggestionRunner()
.onData(data)
.addConstraintRules(Rules.DEFAULT)
.run()
}
type Verifier = DataFrame => VerificationResult
def generateVerifier(cs: ConstraintSuggestionResult): Try[Verifier] = {
val constraintCheckCodes: Seq[String] = cs.constraintSuggestions.flatMap {
case (_, suggestions) => suggestions.map { _.codeForConstraint }
}.toSeq
def checkSrcCode(checkCodeMethod: String, id: Int): String =
s"""com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "$id")$checkCodeMethod"""
val verifierSrcCode = s"""{
|import com.amazon.deequ.constraints.ConstrainableDataTypes
|import com.amazon.deequ.{VerificationResult, VerificationSuite}
|import org.apache.spark.sql.DataFrame
|
|val checks = Seq(
| ${constraintCheckCodes.zipWithIndex
.map { (checkSrcCode _).tupled }
.mkString(",\n ")}
|)
|
|(data: DataFrame) => VerificationSuite().onData(data).addChecks(checks).run()
|}
""".stripMargin.trim
println(s"Verification function source code:\n$verifierSrcCode\n")
compile[Verifier](verifierSrcCode)
}
/** Compiles the scala source code that, when evaluated, produces a value of type T. */
def compile[T](source: String): Try[T] =
Try {
val toolbox = currentMirror.mkToolBox()
val tree = toolbox.parse(source)
val compiledCode = toolbox.compile(tree)
compiledCode().asInstanceOf[T]
}
/** Creates a temporary file, writes the input string to the file, and the file handle.
*
* NOTE: This funciton uses the createTempFile function from the File class. The prefix and
* suffix must be at least 3 characters long, otherwise this function throws an
* IllegalArgumentException.
*/
def writeToTempFile(contents: String,
prefix: Option[String] = None,
suffix: Option[String] = None): File = {
val tempFi = File.createTempFile(prefix.getOrElse("prefix-"), suffix.getOrElse("-suffix"))
tempFi.deleteOnExit()
new PrintWriter(tempFi) {
// Any statements inside the body of a class in scala are executed on construction.
// Therefore, the following try-finally block is executed immediately as we're creating
// a standard PrinterWriter (with its implementation) and then using it.
// Alternatively, we could have created the PrintWriter, assigned it a name,
// then called .write() and .close() on it. Here, we're simply opting for a terser representation.
try {
write(contents)
} finally {
close()
}
}
tempFi
}
//
// Data
//
lazy val csvData: String =
"""Zip_Code,Total_Population,Median_Age,Total_Males,Total_Females,Total_Households,Average_Household_Size
|91371,1,73.5,0,1,1,1
|90001,57110,26.6,28468,28642,12971,4.4
|90002,51223,25.5,24876,26347,11731,4.36
|90003,66266,26.3,32631,33635,15642,4.22
|90004,62180,34.8,31302,30878,22547,2.73
|90005,37681,33.9,19299,18382,15044,2.5
|90006,59185,32.4,30254,28931,18617,3.13
|90007,40920,24,20915,20005,11944,3
|90008,32327,39.7,14477,17850,13841,2.33
|90010,3800,37.8,1874,1926,2014,1.87
|90011,103892,26.2,52794,51098,22168,4.67
|90012,31103,36.3,19493,11610,10327,2.12
|90013,11772,44.6,7629,4143,6416,1.26
|90014,7005,44.8,4471,2534,4109,1.34
|90015,18986,31.3,9833,9153,7420,2.45
|90016,47596,33.9,22778,24818,16145,2.93
|90017,23768,29.4,12818,10950,9338,2.53
|90018,49310,33.2,23770,25540,15493,3.12
|90019,64458,35.8,31442,33016,23344,2.7
|90020,38967,34.6,19381,19586,16514,2.35
|90021,3951,44.3,2790,1161,1561,1.57
|90022,67179,29.8,33216,33963,17023,3.94
|90023,45903,28.4,23037,22866,10727,4.26
|90024,47452,23.6,22248,25204,17903,2.03
|90025,42147,34.7,20859,21288,21228,1.97
|90026,67869,34,34515,33354,24956,2.68
|90027,45151,38.3,22362,22789,21929,1.99
|90028,28714,34,16056,12658,14964,1.78
|90029,38617,34.6,19575,19042,13883,2.7""".stripMargin
}
```
Running this program generates the following output:
```
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Data has 29 rows
Found 21 suggested constraints
Verification function source code:
{
import com.amazon.deequ.constraints.ConstrainableDataTypes
import com.amazon.deequ.{VerificationResult, VerificationSuite}
import org.apache.spark.sql.DataFrame
val checks = Seq(
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "0").isComplete("Average_Household_Size"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "1").hasDataType("Average_Household_Size", ConstrainableDataTypes.Fractional),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "2").isNonNegative("Average_Household_Size"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "3").isComplete("Zip_Code"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "4").hasDataType("Zip_Code", ConstrainableDataTypes.Integral),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "5").isNonNegative("Zip_Code"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "6").isComplete("Total_Males"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "7").hasDataType("Total_Males", ConstrainableDataTypes.Integral),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "8").isNonNegative("Total_Males"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "9").isComplete("Total_Females"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "10").hasDataType("Total_Females", ConstrainableDataTypes.Integral),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "11").isNonNegative("Total_Females"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "12").isComplete("Total_Households"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "13").hasDataType("Total_Households", ConstrainableDataTypes.Integral),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "14").isNonNegative("Total_Households"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "15").isComplete("Median_Age"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "16").hasDataType("Median_Age", ConstrainableDataTypes.Fractional),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "17").isNonNegative("Median_Age"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "18").isComplete("Total_Population"),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "19").hasDataType("Total_Population", ConstrainableDataTypes.Integral),
com.amazon.deequ.checks.Check(com.amazon.deequ.checks.CheckLevel.Error, "20").isNonNegative("Total_Population")
)
(data: DataFrame) => VerificationSuite().onData(data).addChecks(checks).run()
}
----------------------------------
Verification result status: Error
Failed check: 16
Failed check: 1
```
Notably the `VerificationResult`'s `status` is `Error`: 2 suggested constraints fail to apply to the very same data that they were suggested from! :-(
Contributor guide
Research direction
Run the supplied Scala reproducer, starting with ConstraintSuggestionRunner().onData(...).addConstraintRules(Rules.DEFAULT).run(), then inspect the generated constraint code and the VerificationSuite result. Compare the suggested constraints with the exact source DataFrame; done means every generated constraint verifies successfully on that same data.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala, spark
- Domain
- data, testing-qa
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 55/100