Q:

How to compare Datasets?

I have a Spark application that runs custom well-formedness queries against a dataset. Each of these operate only a subset of the whole dataset called "groups" which are really just a filters on the dataset and can be defined by the programmer.
type Group = DataFrame => DataFrame
val groupA = _.filter($"column1" > 0)
val groupB = _.filter($"column2" > 0 && $"column3" === 0)

val constraint1 = constraint(groupA, _.count == 0)
val constraint2 = constraint(groupA, _.dropDuplicates($"column3").count == 1)
// and so on
val constraint3 = constraint(groupB, _.count == 0)
...

framework.add(constraint1, constraint2, constraint3)
framework.execute()

There will be a lot of constraints for each group, so in order to speed it up I want to collect the constraints by their groups, cache the group and run their constraints after each other (or in parallel).
So in order to determine whether two constraints belong to them same group I need some way to compare the datasets for equality.
My idea was to compare them using the semanticHash of the logical plans of the Datasets, however there are couple of logical plans associated to a single Dataset and I wonder which one to choose.
What is the best way to do this?

A:

So I did some experimenting and found out the following on Spark 2.4.0
def equal(a: Dataset[Row], b: Dataset[Row], expected: Boolean) = {
  println(s"by logical hashCode ${a.queryExecution.logical.semanticHash == b.queryExecution.logical.semanticHash}")
  println(s"by logical sameResult ${a.queryExecution.logical.sameResult(b.queryExecution.logical)}")
  println(s"by optimized hashCode ${a.queryExecution.optimizedPlan.semanticHash == b.queryExecution.optimizedPlan.semanticHash}")
  println(s"by optimized sameResult ${a.queryExecution.optimizedPlan.sameResult(b.queryExecution.optimizedPlan)}")
  println(s"expected: $expected")
  println("\n")
}

val a = spark.createDataset(Seq(1, 2)).filter($"value" > 1).filter($"value" > 1).toDF
val b = spark.createDataset(Seq(1, 2)).filter($"value" > 1).toDF
val c = spark.createDataset(Seq(2, 3)).filter($"value" > 1).toDF
val d = spark.createDataset(Seq(2, 3)).filter($"value" < 1).toDF
val e = spark.read.parquet("/test_1")
val f = spark.read.parquet("/test_1")
val g = spark.read.parquet("/test_2")
val h = spark.read.parquet("/test_1").filter($"value" < 1)
val i = spark.read.parquet("/test_1").filter($"value" > 1)

equal(a, b, true)
// by logical hashCode false 
// by logical sameResult false 
// by optimized hashCode true 
// by optimized sameResult true 
// expected: true 

equal(b, c, false)
// by logical hashCode false 
// by logical sameResult false 
// by optimized hashCode false 
// by optimized sameResult false 
// expected: false 

equal(c, d, false)
// by logical hashCode true 
// by logical sameResult false 
// by optimized hashCode false 
// by optimized sameResult false 
// expected: false 

equal(e, f, true)
// by logical hashCode true 
// by logical sameResult true 
// by optimized hashCode true 
// by optimized sameResult true 
// expected: true 

equal(e, g, false)
// by logical hashCode false 
// by logical sameResult false 
// by optimized hashCode false 
// by optimized sameResult false 
// expected: false

equal(h, i, false)
// by logical hashCode true 
// by logical sameResult false
// by optimized hashCode true
// by optimized sameResult false
// expected: false

So I guess I want to choose sameResults on the optimized plan.

