HyukjinKwon commented on code in PR #59466:
URL: https://github.com/apache/spark/pull/59466#discussion_r4284052667
##########
sql/core/src/test/scala/org/apache/spark/sql/MiscFunctionsSuite.scala:
##########
@@ -319,6 +319,27 @@ class MiscFunctionsSuite extends SharedSparkSession {
assert(math.abs(sum(0) - 4.0f) < 0.01f)
assert(math.abs(sum(1) - 6.0f) < 0.01f)
}
+
+ test("SPARK-60245: vector_sum preserves double precision in DataFrames") {
+ val df = Seq(Seq(16777216.0d, 1.0d), Seq(1.0d,
1e-8d)).toDF("v").repartition(2)
+ val result = df.select(vector_sum($"v"))
+ assertResult(ArrayType(DoubleType, containsNull =
false))(result.schema.head.dataType)
+ assertResult(Seq(16777217.0d,
1.00000001d))(result.head().getSeq[Double](0))
+ }
+
+ gridTest("SPARK-60245: observe vector aggregates")(Seq(
+ ("float", "vector_sum"),
+ ("double", "vector_sum"),
+ ("float", "vector_avg"))) { case (elementType, functionName) =>
+ val df = spark.range(0, 8, 1, 4).select(
+ array($"id".cast(elementType), lit(1).cast(elementType)).as("v"))
+ val aggregate = if (functionName == "vector_sum") vector_sum($"v") else
vector_avg($"v")
+ val expected = if (functionName == "vector_sum") Seq(28.0, 8.0) else
Seq(3.5, 1.0)
+ val observation = Observation(s"${functionName}_$elementType")
+
+ assertResult(8)(df.observe(observation,
aggregate.as("value")).collect().length)
+ assertResult(Map("value" -> expected))(observation.get)
Review Comment:
**Non-blocking (P2):** The `("float", "vector_avg")` row compares the
observed average exactly with `Seq(3.5, 1.0)`, but that value depends on the
order in which the driver merges the four task buffers.
`AggregatingAccumulator` folds each task's buffer into the driver buffer as
tasks complete, and `FloatAvg.mergeElements` uses single-precision weights such
as `4f/6` and `2f/6`. With partial averages 0.5, 2.5, 4.5 and 6.5, 4 of the 24
possible completion orders produce `3.5000002f`, for example p1, p2, p0, p3,
which is reachable under `local[2]`. So this assertion can fail intermittently.
The Float-vs-Double element type is not the problem here. Scala compares
boxed numbers cooperatively, so `Map("value" -> Seq(28.0, 8.0))` equals the
observed `Map("value" -> ArraySeq(28.0f, 8.0f))`, and the float `vector_sum`
row is fine. Switching to `.doubleValue()` or float literals would not fix the
`vector_avg` row. Comparing the `vector_avg` values with a small tolerance, as
the existing `vector functions` test does, would make it deterministic.
**Verification:**
- **Regression:** The vector_avg observe assertion accepts a merged first
element of 3.5000002f and still fails for a materially wrong average.
- **Behavior:** The observe grid passes for float vector_sum, double
vector_sum and float vector_avg.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
