From ea70afba3fade730a3002031476f6246bd975cd9 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sun, 4 Oct 2026 14:45:01 -0600 Subject: [PATCH 1/2] test: compare NaN, infinity sign and row order in tolerance checks `query tolerance=` in Comet SQL tests and the Scala `*WithTol*` helpers skipped the comparison when either side was NaN, accepted +Infinity against -Infinity, and paired rows in collect order. The tolerance check now shares `checkCometAnswer`: NaN matches only NaN, an infinity only the same infinity, and rows are sorted when the plan has no Sort. Floating-point fields sort by value, because a `toString` sort can separate two values that match within tolerance. Closes #6616. --- .github/workflows/pr_build_linux.yml | 1 + .github/workflows/pr_build_macos.yml | 1 + .../contributor-guide/sql-file-tests.md | 26 ++-- .../sql-tests/expressions/math/csc.sql | 2 +- .../sql-tests/expressions/math/pow.sql | 8 +- .../comet/CometAnswerToleranceSuite.scala | 98 ++++++++++++++ .../org/apache/spark/sql/CometTestBase.scala | 123 ++++++++++-------- 7 files changed, 190 insertions(+), 69 deletions(-) create mode 100644 spark/src/test/scala/org/apache/comet/CometAnswerToleranceSuite.scala diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index 2360d7a5a46..5b13788e271 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.github/workflows/pr_build_linux.yml @@ -585,6 +585,7 @@ jobs: org.apache.comet.GenerateDocsSuite org.apache.comet.CometSqlFileTestSuite org.apache.comet.SqlFileTestParserSuite + org.apache.comet.CometAnswerToleranceSuite org.apache.comet.CometExpressionCoverageSuite org.apache.comet.CometVariantTypeSuite org.apache.comet.CometHashExpressionSuite diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index 0a8254bff85..129682b364f 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.github/workflows/pr_build_macos.yml @@ -289,6 +289,7 @@ jobs: org.apache.comet.GenerateDocsSuite org.apache.comet.CometSqlFileTestSuite org.apache.comet.SqlFileTestParserSuite + org.apache.comet.CometAnswerToleranceSuite org.apache.comet.CometExpressionCoverageSuite org.apache.comet.CometVariantTypeSuite org.apache.comet.CometHashExpressionSuite diff --git a/docs/source/contributor-guide/sql-file-tests.md b/docs/source/contributor-guide/sql-file-tests.md index 57cc717547f..ba65c0cc575 100644 --- a/docs/source/contributor-guide/sql-file-tests.md +++ b/docs/source/contributor-guide/sql-file-tests.md @@ -176,20 +176,26 @@ SELECT some_expression(v) FROM test_table #### `query tolerance=` -Checks results with a numeric tolerance. Useful for floating-point functions where small -differences are acceptable. +Checks results like the default mode, except that top-level `float` and `double` values match +when they are within the given absolute tolerance. Useful for floating-point functions where small +differences are acceptable. The tolerance must be greater than zero and at most `1e-6`. -The comparison ignores the sign of zero (`+0.0` matches `-0.0`) and the sign of infinity -(`+Infinity` matches `-Infinity`). Keep tolerance for ordinary values and add a plain -`query` for cases whose expected result is signed zero or signed infinity. +NaN matches only NaN, and an infinity matches only the infinity of the same sign. The comparison +ignores the sign of zero (`+0.0` matches `-0.0`), so keep tolerance for ordinary values and add a +plain `query` for cases whose expected result is signed zero. Values nested inside arrays, structs +and maps are compared exactly. + +As in the default mode, both results are sorted before they are compared unless the query has an +`ORDER BY`. Floating-point values sort by value, so two results that match within the tolerance +end up in the same position. ```sql -query tolerance=0.0001 +query tolerance=1e-6 SELECT cos(v) FROM test_trig --- csc(-0.0) == -Infinity; a tolerance check would also accept +Infinity +-- tan(-0.0) == -0.0; a tolerance check would also accept +0.0 query -SELECT csc(double('-0.0')) +SELECT tan(double('-0.0')) ``` #### `query expect_fallback()` @@ -365,8 +371,8 @@ common ones include: equal, so `sort_array(...)` is not a unique projection when both signs are present (the SQL test comparator distinguishes the bits). Prefer a sign-aware form such as `sort_array(transform(arr, x -> cast(x AS string)))`. A `query tolerance=...` check - likewise treats the two zero signs as equal, and `+Infinity` / `-Infinity` as equal, - so signed-zero and signed-infinity results need a separate plain `query`. + likewise treats the two zero signs as equal, so signed-zero results need a separate plain + `query`. - **Special characters and multibyte UTF-8** -- for string functions (e.g. `'é'`, `'中文'`, `'\t'`) - **Empty arrays/maps** -- for collection functions diff --git a/spark/src/test/resources/sql-tests/expressions/math/csc.sql b/spark/src/test/resources/sql-tests/expressions/math/csc.sql index e5c0cdb02a4..f8ba96a583f 100644 --- a/spark/src/test/resources/sql-tests/expressions/math/csc.sql +++ b/spark/src/test/resources/sql-tests/expressions/math/csc.sql @@ -25,7 +25,7 @@ query tolerance=1e-6 SELECT csc(d) FROM test_csc -- signed-zero: exact comparison so csc(-0.0) == -Infinity is distinguished --- from +Infinity (tolerance only checks isInfinity). +-- from +Infinity. query SELECT csc(d) FROM test_csc WHERE d = 0.0 diff --git a/spark/src/test/resources/sql-tests/expressions/math/pow.sql b/spark/src/test/resources/sql-tests/expressions/math/pow.sql index f6baddffa91..3a2c9935f39 100644 --- a/spark/src/test/resources/sql-tests/expressions/math/pow.sql +++ b/spark/src/test/resources/sql-tests/expressions/math/pow.sql @@ -35,11 +35,9 @@ INSERT INTO test_pow VALUES (1.0, cast('NaN' as double)), (-1.0, cast('NaN' as double)), (cast('4.9E-324' as double), 2.0), (2.0, cast('4.9E-324' as double)) --- Every pair above yields an exact double, so use exact comparison (no tolerance) to assert the --- NaN and signed-Infinity edge cases rather than silently skipping them: a tolerance comparison --- makes no assertion at all for NaN and ignores the sign of Infinity. Exact comparison is --- payload-insensitive for NaN (the test framework canonicalizes it) but not for signed zero, so --- the -0.0 results are asserted exactly. +-- Every pair above yields an exact double, so use exact comparison (no tolerance). Unlike a +-- tolerance comparison it distinguishes signed zero, so the -0.0 results are asserted exactly. +-- It is payload-insensitive for NaN (the test framework canonicalizes it). query SELECT pow(base, exp) FROM test_pow diff --git a/spark/src/test/scala/org/apache/comet/CometAnswerToleranceSuite.scala b/spark/src/test/scala/org/apache/comet/CometAnswerToleranceSuite.scala new file mode 100644 index 00000000000..07682fb2965 --- /dev/null +++ b/spark/src/test/scala/org/apache/comet/CometAnswerToleranceSuite.scala @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.comet + +import org.scalatest.exceptions.TestFailedException + +import org.apache.spark.sql.{CometTestBase, DataFrame, Row} + +/** + * Tests the tolerance comparison behind `query tolerance=` in Comet SQL tests and the + * `checkSparkAnswer*WithTol*` helpers. The DataFrame stands in for the Comet result and the rows + * for the Spark answer. + */ +class CometAnswerToleranceSuite extends CometTestBase { + import testImplicits._ + + private def assertMatch(comet: DataFrame, spark: Row*): Unit = + checkCometAnswerWithTolerance(comet, spark, 1e-6) + + private def assertMismatch(comet: DataFrame, spark: Row*): Unit = { + val e = intercept[TestFailedException](checkCometAnswerWithTolerance(comet, spark, 1e-6)) + assert(e.getMessage.startsWith("Results do not match within tolerance"), e.getMessage) + } + + test("finite values match within the tolerance") { + assertMatch(Seq(1.0 + 1e-7).toDF("d"), Row(1.0)) + assertMismatch(Seq(1.0 + 1e-5).toDF("d"), Row(1.0)) + assertMatch(Seq(1.0f).toDF("f"), Row(1.0f)) + assertMismatch(Seq(1.001f).toDF("f"), Row(1.0f)) + } + + test("NaN matches only NaN") { + assertMatch(Seq(Double.NaN).toDF("d"), Row(Double.NaN)) + assertMismatch(Seq(Double.NaN).toDF("d"), Row(1.0)) + assertMismatch(Seq(1.0).toDF("d"), Row(Double.NaN)) + assertMismatch(Seq(Double.NaN).toDF("d"), Row(Double.PositiveInfinity)) + assertMatch(Seq(Float.NaN).toDF("f"), Row(Float.NaN)) + assertMismatch(Seq(Float.NaN).toDF("f"), Row(1.0f)) + assertMismatch(Seq(1.0f).toDF("f"), Row(Float.NaN)) + } + + test("infinities compare with their sign") { + assertMatch(Seq(Double.PositiveInfinity).toDF("d"), Row(Double.PositiveInfinity)) + assertMatch(Seq(Double.NegativeInfinity).toDF("d"), Row(Double.NegativeInfinity)) + assertMismatch(Seq(Double.PositiveInfinity).toDF("d"), Row(Double.NegativeInfinity)) + assertMismatch(Seq(Double.NegativeInfinity).toDF("d"), Row(Double.PositiveInfinity)) + assertMismatch(Seq(Double.MaxValue).toDF("d"), Row(Double.PositiveInfinity)) + assertMismatch(Seq(Float.PositiveInfinity).toDF("f"), Row(Float.NegativeInfinity)) + } + + test("signed zeros match each other") { + assertMatch(Seq(-0.0).toDF("d"), Row(0.0)) + assertMatch(Seq(0.0f).toDF("f"), Row(-0.0f)) + } + + test("other values compare exactly") { + assertMatch(Seq(("a", 1.0)).toDF("s", "d"), Row("a", 1.0 + 1e-7)) + assertMismatch(Seq(("a", 1.0)).toDF("s", "d"), Row("b", 1.0)) + // Only top-level floating-point values get the tolerance. + assertMismatch(Seq(Seq(1.0 + 1e-7)).toDF("a"), Row(Seq(1.0))) + } + + test("rows are sorted before they are paired") { + assertMatch( + Seq(Some(Double.NaN), Some(1.0), None, Some(Double.NegativeInfinity), Some(-0.0)).toDF("d"), + Row(0.0), + Row(Double.NegativeInfinity), + Row(null), + Row(1.0), + Row(Double.NaN)) + // Sorting by `toString` would put 1.2E-16 after 0.5 but 0.0 before it, and pair 0.5 with 0.0. + assertMatch(Seq(0.5, 1.2246467991473532e-16).toDF("d"), Row(0.0), Row(0.5)) + assertMismatch(Seq(1.0, 2.0).toDF("d"), Row(1.0), Row(Double.NaN)) + assertMismatch(Seq(1.0, 2.0).toDF("d"), Row(1.0)) + } + + test("rows of a sorted query are paired in order") { + assertMatch(Seq(1.0, 2.0).toDF("d").orderBy($"d".desc), Row(2.0), Row(1.0)) + assertMismatch(Seq(1.0, 2.0).toDF("d").orderBy($"d".desc), Row(1.0), Row(2.0)) + } +} diff --git a/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala b/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala index 8903f43769b..edaa41974b1 100644 --- a/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala +++ b/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala @@ -142,7 +142,7 @@ abstract class CometTestBase } val dfComet = datasetOfRows(spark, df.logicalPlan) if (withTol.isDefined) { - checkAnswerWithTolerance(dfComet, expected, withTol.get) + checkCometAnswerWithTolerance(dfComet, expected, withTol.get) } else { checkCometAnswer(dfComet, expected) } @@ -484,7 +484,27 @@ abstract class CometTestBase * correctly identify which side is Comet and which is Spark. This avoids the misleading "Spark * Answer" label that Spark's built-in `checkAnswer` would apply to the Comet result. */ - protected def checkCometAnswer(cometDf: DataFrame, sparkAnswer: Seq[Row]): Unit = { + protected def checkCometAnswer(cometDf: DataFrame, sparkAnswer: Seq[Row]): Unit = + checkCometAnswer(cometDf, sparkAnswer, absTol = None) + + /** + * [[checkCometAnswer]], but top-level `float` and `double` values match when they are within + * `absTol` of each other. NaN matches only NaN, and an infinity matches only the infinity of + * the same sign. `+0.0` matches `-0.0`. Values nested in arrays, structs and maps are compared + * exactly. + */ + protected def checkCometAnswerWithTolerance( + cometDf: DataFrame, + sparkAnswer: Seq[Row], + absTol: Double): Unit = { + require(absTol > 0 && absTol <= 1e-6, s"absTol $absTol is out of range (0, 1e-6]") + checkCometAnswer(cometDf, sparkAnswer, Some(absTol)) + } + + private def checkCometAnswer( + cometDf: DataFrame, + sparkAnswer: Seq[Row], + absTol: Option[Double]): Unit = { val isSorted = cometDf.logicalPlan.collect { case s: logical.Sort => s }.nonEmpty val cometAnswer = try cometDf.collect().toSeq @@ -497,14 +517,23 @@ abstract class CometTestBase |${org.apache.spark.sql.catalyst.util.stackTraceToString(e)} """.stripMargin) } - val preparedSpark = prepareCometAnswer(sparkAnswer, isSorted) - val preparedComet = prepareCometAnswer(cometAnswer, isSorted) - if (!QueryTest.compare(preparedSpark, preparedComet)) { + val ordering = + if (absTol.isDefined) toleranceOrdering else Ordering.by[Row, String](_.toString) + val preparedSpark = prepareCometAnswer(sparkAnswer, isSorted, ordering) + val preparedComet = prepareCometAnswer(cometAnswer, isSorted, ordering) + val matches = absTol match { + case Some(tol) => + preparedSpark.length == preparedComet.length && + preparedSpark.zip(preparedComet).forall { case (s, c) => rowsMatch(s, c, tol) } + case None => QueryTest.compare(preparedSpark, preparedComet) + } + if (!matches) { val getRowType: Option[Row] => String = row => row .map(r => if (r.schema == null) "struct<>" else r.schema.catalogString) .getOrElse("struct<>") - fail(s"""Results do not match for query: + val tolerance = absTol.map(t => s" within tolerance $t").getOrElse("") + fail(s"""Results do not match$tolerance for query: |Timezone: ${java.util.TimeZone.getDefault} |Timezone Env: ${sys.env.getOrElse("TZ", "")} | @@ -528,9 +557,12 @@ abstract class CometTestBase * makes the toString-based sort in `prepareAnswer` non-deterministic and causes spurious * mismatches between the two sides. */ - private def prepareCometAnswer(answer: Seq[Row], isSorted: Boolean): Seq[Row] = { + private def prepareCometAnswer( + answer: Seq[Row], + isSorted: Boolean, + ordering: Ordering[Row]): Seq[Row] = { val converted = answer.map(prepareCometRow) - if (isSorted) converted else converted.sortBy(_.toString()) + if (isSorted) converted else converted.sorted(ordering) } private def prepareCometRow(row: Row): Row = { @@ -564,56 +596,41 @@ abstract class CometTestBase } /** - * A helper function for comparing Comet DataFrame with Spark result using absolute tolerance. + * Orders rows for a tolerance comparison: field by field, with `float` and `double` fields + * compared by value and everything else by its string form. Sorting whole rows by `toString`, + * as the exact comparison does, can put two values that match within tolerance at different + * positions: `1.2246467991473532E-16` sorts after `0.5`, but `0.0` sorts before it. The two + * zeros are equal here because the comparison matches them, and NaN sorts last. */ - private def checkAnswerWithTolerance( - dataFrame: DataFrame, - expectedAnswer: Seq[Row], - absTol: Double): Unit = { - val actualAnswer = dataFrame.collect() - require( - actualAnswer.length == expectedAnswer.length, - s"actual num rows ${actualAnswer.length} != expected num of rows ${expectedAnswer.length}") - - actualAnswer.zip(expectedAnswer).foreach { case (actualRow, expectedRow) => - checkAnswerWithTolerance(actualRow, expectedRow, absTol) + private val toleranceOrdering: Ordering[Row] = new Ordering[Row] { + override def compare(x: Row, y: Row): Int = + x.toSeq.iterator + .zip(y.toSeq.iterator) + .map { case (l, r) => compareField(l, r) } + .find(_ != 0) + .getOrElse(0) + + private def compareField(left: Any, right: Any): Int = (left, right) match { + case (null, null) => 0 + case (null, _) => -1 + case (_, null) => 1 + case (l: Float, r: Float) => if (l == r) 0 else java.lang.Float.compare(l, r) + case (l: Double, r: Double) => if (l == r) 0 else java.lang.Double.compare(l, r) + case _ => left.toString.compareTo(right.toString) } } - /** - * Compares two answers and makes sure the answer is within absTol of the expected result. - */ - private def checkAnswerWithTolerance( - actualAnswer: Row, - expectedAnswer: Row, - absTol: Double): Unit = { - require( - actualAnswer.length == expectedAnswer.length, - s"actual answer length ${actualAnswer.length} != " + - s"expected answer length ${expectedAnswer.length}") - require(absTol > 0 && absTol <= 1e-6, s"absTol $absTol is out of range (0, 1e-6]") - - actualAnswer.toSeq.zip(expectedAnswer.toSeq).foreach { - case (actual: Float, expected: Float) => - if (actual.isInfinity || expected.isInfinity) { - assert(actual.isInfinity == expected.isInfinity, s"actual answer $actual != $expected") - } else if (!actual.isNaN && !expected.isNaN) { - assert( - math.abs(actual - expected) < absTol, - s"actual answer $actual not within $absTol of correct answer $expected") - } - case (actual: Double, expected: Double) => - if (actual.isInfinity || expected.isInfinity) { - assert(actual.isInfinity == expected.isInfinity, s"actual answer $actual != $expected") - } else if (!actual.isNaN && !expected.isNaN) { - assert( - math.abs(actual - expected) < absTol, - s"actual answer $actual not within $absTol of correct answer $expected") - } - case (actual, expected) => - assert(actual == expected, s"$actualAnswer did not equal $expectedAnswer") + private def rowsMatch(spark: Row, comet: Row, absTol: Double): Boolean = + spark.length == comet.length && spark.toSeq.zip(comet.toSeq).forall { + case (s: Float, c: Float) => valuesMatch(s, c, absTol) + case (s: Double, c: Double) => valuesMatch(s, c, absTol) + case (s, c) => QueryTest.compare(s, c) } - } + + private def valuesMatch(spark: Double, comet: Double, absTol: Double): Boolean = + if (spark.isNaN || comet.isNaN) spark.isNaN && comet.isNaN + else if (spark.isInfinite || comet.isInfinite) spark == comet + else math.abs(spark - comet) < absTol protected def checkCometOperators(plan: SparkPlan, excludedClasses: Class[_]*): Unit = { findFirstNonCometOperator(plan, excludedClasses: _*) match { From 054ae3978db489cba187e5453734a2836a78d526 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 13 Oct 2026 09:09:25 -0600 Subject: [PATCH 2/2] test: pair unordered rows one-to-one in tolerance checks Rows of an unordered answer now pair one-to-one: they are grouped by their fields other than the top-level floats, and a group with several float fields that does not pair in sorted order falls back to Kuhn's algorithm. Also widen Float to Double explicitly for the Spark 3.5 strict-warnings build. --- .../contributor-guide/sql-file-tests.md | 5 +- .../comet/CometAnswerToleranceSuite.scala | 29 ++++ .../org/apache/spark/sql/CometTestBase.scala | 145 ++++++++++++++++-- 3 files changed, 159 insertions(+), 20 deletions(-) diff --git a/docs/source/contributor-guide/sql-file-tests.md b/docs/source/contributor-guide/sql-file-tests.md index 389b54ef7ed..405c5b2cfa1 100644 --- a/docs/source/contributor-guide/sql-file-tests.md +++ b/docs/source/contributor-guide/sql-file-tests.md @@ -263,9 +263,8 @@ ignores the sign of zero (`+0.0` matches `-0.0`), so keep tolerance for ordinary plain `query` for cases whose expected result is signed zero. Values nested inside arrays, structs and maps are compared exactly. -As in the default mode, both results are sorted before they are compared unless the query has an -`ORDER BY`. Floating-point values sort by value, so two results that match within the tolerance -end up in the same position. +As in the default mode, the order of the rows only matters when the query has an `ORDER BY`. +Otherwise each row of the Spark result must match a different row of the Comet result. ```sql query tolerance=1e-6 diff --git a/spark/src/test/scala/org/apache/comet/CometAnswerToleranceSuite.scala b/spark/src/test/scala/org/apache/comet/CometAnswerToleranceSuite.scala index 07682fb2965..5c56dc2b634 100644 --- a/spark/src/test/scala/org/apache/comet/CometAnswerToleranceSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometAnswerToleranceSuite.scala @@ -75,6 +75,8 @@ class CometAnswerToleranceSuite extends CometTestBase { assertMismatch(Seq(("a", 1.0)).toDF("s", "d"), Row("b", 1.0)) // Only top-level floating-point values get the tolerance. assertMismatch(Seq(Seq(1.0 + 1e-7)).toDF("a"), Row(Seq(1.0))) + assertMismatch(Seq(Seq(-0.0)).toDF("a"), Row(Seq(0.0))) + assertMatch(Seq(Seq(Double.NaN)).toDF("a"), Row(Seq(Double.NaN))) } test("rows are sorted before they are paired") { @@ -91,6 +93,33 @@ class CometAnswerToleranceSuite extends CometTestBase { assertMismatch(Seq(1.0, 2.0).toDF("d"), Row(1.0)) } + test("rows of an unordered query pair one-to-one") { + // A leading value within tolerance must not pair rows whose other fields differ. + assertMatch(Seq((1e-7, "a"), (0.0, "b")).toDF("d", "s"), Row(0.0, "a"), Row(0.0, "b")) + assertMismatch(Seq((1e-7, "b"), (0.0, "b")).toDF("d", "s"), Row(0.0, "a"), Row(0.0, "b")) + // The same with a second floating-point field, which sorting alone cannot pair. + assertMatch(Seq((1e-7, 1.0), (0.0, 2.0)).toDF("x", "y"), Row(0.0, 1.0), Row(0.0, 2.0)) + // The first Spark row matches both Comet rows, the second only the first Comet row. Pairing + // in order, or giving each Spark row the first free Comet row it matches, pairs the first + // rows and leaves the second Spark row without a partner. + assertMatch(Seq((2e-7, 5e-7), (3e-7, -8e-7)).toDF("x", "y"), Row(0.0, 0.0), Row(1e-7, 9e-7)) + // Every row matches a row on the other side, but not one of its own. + assertMismatch( + Seq((0.0, 0.0), (1.0, 1.0), (1.0, 1.0)).toDF("x", "y"), + Row(0.0, 0.0), + Row(0.0, 0.0), + Row(1.0, 1.0)) + // Maps match whatever the order of their entries, which their string form does not ignore. + assertMatch( + Seq(Map("b" -> 2, "a" -> 1), Map("a" -> 2, "b" -> 1)).toDF("m"), + Row(Map("a" -> 1, "b" -> 2)), + Row(Map("b" -> 1, "a" -> 2))) + assertMismatch( + Seq(Map("b" -> 2, "a" -> 1), Map("a" -> 1, "b" -> 2)).toDF("m"), + Row(Map("a" -> 1, "b" -> 2)), + Row(Map("b" -> 1, "a" -> 2))) + } + test("rows of a sorted query are paired in order") { assertMatch(Seq(1.0, 2.0).toDF("d").orderBy($"d".desc), Row(2.0), Row(1.0)) assertMismatch(Seq(1.0, 2.0).toDF("d").orderBy($"d".desc), Row(1.0), Row(2.0)) diff --git a/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala b/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala index 7789560c700..98e064f1186 100644 --- a/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala +++ b/spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala @@ -21,6 +21,7 @@ package org.apache.spark.sql import java.util.concurrent.atomic.AtomicInteger +import scala.collection.mutable import scala.concurrent.duration._ import scala.reflect.ClassTag import scala.reflect.runtime.universe.TypeTag @@ -522,7 +523,8 @@ abstract class CometTestBase * [[checkCometAnswer]], but top-level `float` and `double` values match when they are within * `absTol` of each other. NaN matches only NaN, and an infinity matches only the infinity of * the same sign. `+0.0` matches `-0.0`. Values nested in arrays, structs and maps are compared - * exactly. + * exactly. Unless the query is sorted, each Spark row must match a Comet row of its own, in any + * order. */ protected def checkCometAnswerWithTolerance( cometDf: DataFrame, @@ -553,9 +555,7 @@ abstract class CometTestBase val preparedSpark = prepareCometAnswer(sparkAnswer, isSorted, ordering) val preparedComet = prepareCometAnswer(cometAnswer, isSorted, ordering) val matches = absTol match { - case Some(tol) => - preparedSpark.length == preparedComet.length && - preparedSpark.zip(preparedComet).forall { case (s, c) => rowsMatch(s, c, tol) } + case Some(tol) => answersMatchWithinTolerance(preparedSpark, preparedComet, isSorted, tol) case None => QueryTest.compare(preparedSpark, preparedComet) } if (!matches) { @@ -627,33 +627,144 @@ abstract class CometTestBase } /** - * Orders rows for a tolerance comparison: field by field, with `float` and `double` fields - * compared by value and everything else by its string form. Sorting whole rows by `toString`, - * as the exact comparison does, can put two values that match within tolerance at different - * positions: `1.2246467991473532E-16` sorts after `0.5`, but `0.0` sorts before it. The two - * zeros are equal here because the comparison matches them, and NaN sorts last. + * Orders rows for a tolerance comparison: by their top-level `float` and `double` fields, + * compared by value, and then by their string form. Sorting by the string form alone, as the + * exact comparison does, can put two values that match within tolerance far apart: + * `1.2246467991473532E-16` sorts after `0.5`, but `0.0` sorts before it. The two zeros are + * equal here because the comparison matches them. Null sorts first and NaN last. */ private val toleranceOrdering: Ordering[Row] = new Ordering[Row] { override def compare(x: Row, y: Row): Int = x.toSeq.iterator .zip(y.toSeq.iterator) - .map { case (l, r) => compareField(l, r) } + .map { case (l, r) => compareFloats(l, r) } .find(_ != 0) - .getOrElse(0) + .getOrElse(x.toString.compareTo(y.toString)) - private def compareField(left: Any, right: Any): Int = (left, right) match { - case (null, null) => 0 - case (null, _) => -1 - case (_, null) => 1 + private def compareFloats(left: Any, right: Any): Int = (left, right) match { case (l: Float, r: Float) => if (l == r) 0 else java.lang.Float.compare(l, r) case (l: Double, r: Double) => if (l == r) 0 else java.lang.Double.compare(l, r) - case _ => left.toString.compareTo(right.toString) + case (null, _: Float | _: Double) => -1 + case (_: Float | _: Double, null) => 1 + case _ => 0 } } + /** + * Whether two answers, sorted by [[toleranceOrdering]], match within `absTol`: each Spark row + * must match a Comet row of its own. A sorted query pairs the rows in order. Otherwise two rows + * can only pair when their fields other than the top-level `float` and `double` ones compare + * equal, so the rows are grouped by those fields and each group must pair up on its own, which + * also makes duplicates count. Within a group, the rows are in order of their floating-point + * values. With at most one such field, pairing the rows in that order works whenever any + * pairing does. With more, rows that match can sort differently on each side, so a group that + * does not pair in order falls back to [[pairsOneToOne]]. + */ + private def answersMatchWithinTolerance( + spark: Seq[Row], + comet: Seq[Row], + isSorted: Boolean, + absTol: Double): Boolean = { + def pairInOrder(sparkRows: Seq[Row], cometRows: Seq[Row]): Boolean = + sparkRows.length == cometRows.length && + sparkRows.zip(cometRows).forall { case (s, c) => rowsMatch(s, c, absTol) } + if (isSorted) { + pairInOrder(spark, comet) + } else { + val sparkGroups = spark.groupBy(toleranceGroupKey) + val cometGroups = comet.groupBy(toleranceGroupKey) + sparkGroups.keySet == cometGroups.keySet && sparkGroups.forall { case (key, sparkRows) => + val cometRows = cometGroups(key) + val floatFields = key.count(_.isEmpty) + pairInOrder(sparkRows, cometRows) || + (floatFields > 1 && sparkRows.length == cometRows.length && + pairsOneToOne(sparkRows.toIndexedSeq, cometRows.toIndexedSeq, absTol)) + } + } + } + + /** + * Keys a row by its fields other than the top-level `float` and `double` ones, which become + * `None`. Two rows get the same key when `QueryTest.compare` matches those other fields. + */ + private def toleranceGroupKey(row: Row): Seq[Option[Any]] = row.toSeq.map { + case _: Float | _: Double => None + case value => Some(comparisonKey(value)) + } + + /** + * A normalized value in a form that is equal for two values when `QueryTest.compare` matches + * them, so that it can key a hash map: a map becomes the set of its entries, and a `float` or + * `double` its bits. + */ + private def comparisonKey(value: Any): Any = value match { + case row: Row => row.toSeq.map(comparisonKey) + case map: scala.collection.Map[_, _] => + map.iterator.map { case (k, v) => (comparisonKey(k), comparisonKey(v)) }.toSet + case seq: scala.collection.Iterable[_] => seq.map(comparisonKey).toSeq + case d: Double => java.lang.Double.doubleToRawLongBits(d) + case f: Float => java.lang.Float.floatToRawIntBits(f) + case other => other + } + + /** + * Whether each Spark row can pair with a Comet row of its own that it matches within `absTol`, + * found with Kuhn's algorithm. The rows at the same position start out paired when they match. + * Each Spark row left without a partner then searches breadth-first for a path that ends at a + * Comet row without one: a Comet row that it matches, whose partner moves on to another Comet + * row that it matches, and so on. Shifting the pairs along that path pairs one more row. When a + * Spark row finds no such path, no pairing of all the rows exists. + */ + private def pairsOneToOne( + spark: IndexedSeq[Row], + comet: IndexedSeq[Row], + absTol: Double): Boolean = { + val sparkPartner = Array.fill(spark.length)(-1) + val cometPartner = Array.fill(comet.length)(-1) + spark.indices.filter(i => rowsMatch(spark(i), comet(i), absTol)).foreach { i => + sparkPartner(i) = i + cometPartner(i) = i + } + // The Comet rows that each Spark row matches, found when the search first needs them. + val candidates = mutable.Map.empty[Int, Seq[Int]] + def candidatesOf(s: Int): Seq[Int] = + candidates.getOrElseUpdate( + s, + comet.indices.filter(c => rowsMatch(spark(s), comet(c), absTol))) + + def pairUp(start: Int): Boolean = { + // The Spark row from which the search first reached each Comet row. + val reachedFrom = Array.fill(comet.length)(-1) + val queue = mutable.Queue(start) + var unpaired = -1 + while (unpaired < 0 && queue.nonEmpty) { + val s = queue.dequeue() + for (c <- candidatesOf(s) if unpaired < 0 && reachedFrom(c) < 0) { + reachedFrom(c) = s + if (cometPartner(c) < 0) { + unpaired = c + } else { + queue.enqueue(cometPartner(c)) + } + } + } + var c = unpaired + while (c >= 0) { + val s = reachedFrom(c) + val previous = sparkPartner(s) + sparkPartner(s) = c + cometPartner(c) = s + c = previous + } + unpaired >= 0 + } + + spark.indices.forall(s => sparkPartner(s) >= 0 || pairUp(s)) + } + private def rowsMatch(spark: Row, comet: Row, absTol: Double): Boolean = spark.length == comet.length && spark.toSeq.zip(comet.toSeq).forall { - case (s: Float, c: Float) => valuesMatch(s, c, absTol) + case (s: Float, c: Float) => valuesMatch(s.toDouble, c.toDouble, absTol) case (s: Double, c: Double) => valuesMatch(s, c, absTol) case (s, c) => QueryTest.compare(s, c) }