spark-test commented on code in PR #59590:
URL: https://github.com/apache/spark/pull/59590#discussion_r4496869277
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/VectorAggSuite.scala:
##########
@@ -821,4 +858,200 @@ class VectorAggSuite extends SparkFunSuite {
condition = "VECTOR_DIMENSION_MISMATCH",
parameters = Map("functionName" -> "`vector_sum`", "leftDim" -> "2",
"rightDim" -> "3"))
}
+
+ test("VectorAvg - double: empty buffer returns null") {
+ val (agg, buffer) = createDoubleVectorAvg()
+ assertResult(null)(agg.eval(buffer))
+ }
+
+ test("VectorAvg - double: multiple vectors") {
+ val (agg, buffer) = createDoubleVectorAvg()
+ agg.update(buffer, createDoubleInputRow(Array(1.0, 2.0)))
+ agg.update(buffer, createDoubleInputRow(Array(2.0, 4.0)))
+ agg.update(buffer, createDoubleInputRow(Array(3.0, 6.0)))
+ assertResult(Array(2.0, 4.0))(evalAsDoubleArray(agg, buffer))
+ }
+
+ test("VectorAvg - double: result is independent of later buffer updates") {
+ val (agg, buffer) = createDoubleVectorAvg()
+ agg.update(buffer, createDoubleInputRow(Array(1.0, 2.0, 3.0)))
+ val result = agg.eval(buffer).asInstanceOf[ArrayData]
+
+ agg.update(buffer, createDoubleInputRow(Array(3.0, 4.0, 5.0)))
+ assertResult(Array(2.0, 3.0, 4.0))(evalAsDoubleArray(agg, buffer))
+ assertResult(Array(1.0, 2.0, 3.0))(result.toDoubleArray())
+ }
+
+ test("VectorAvg - double: keeps precision lost by float") {
+ val (agg, buffer) = createDoubleVectorAvg()
+ agg.update(buffer, createDoubleInputRow(Array(16777216.0, 1.0e300)))
+ agg.update(buffer, createDoubleInputRow(Array(16777218.0, 1.0e300)))
+ assertResult(Array(16777217.0, 1.0e300))(evalAsDoubleArray(agg, buffer))
+ }
+
+ test("VectorAvg - double: merge buffers with different counts") {
+ val (agg, buffer1) = createDoubleVectorAvg()
+ val (_, buffer2) = createDoubleVectorAvg()
+ agg.update(buffer1, createDoubleInputRow(Array(16777216.0, 10.0)))
+ agg.update(buffer2, createDoubleInputRow(Array(16777218.0, 1.0)))
+ agg.update(buffer2, createDoubleInputRow(Array(16777218.0, 2.0)))
+ agg.update(buffer2, createDoubleInputRow(Array(16777218.0, 3.0)))
+ agg.merge(buffer1, buffer2)
+ assertResult(Array(16777217.5, 4.0))(evalAsDoubleArray(agg, buffer1))
+ }
+
+ test("VectorAvg - double: merge with empty buffers") {
+ val (agg, buffer1) = createDoubleVectorAvg()
+ val (_, buffer2) = createDoubleVectorAvg()
+ agg.merge(buffer1, buffer2)
+ assertResult(null)(agg.eval(buffer1))
+
+ agg.update(buffer2, createDoubleInputRow(Array(1.5, 2.5)))
+ agg.merge(buffer1, buffer2)
+ assertResult(Array(1.5, 2.5))(evalAsDoubleArray(agg, buffer1))
+ }
+
+ test("VectorAvg - double: null vectors and vectors with null elements are
skipped") {
+ val (agg, buffer) = createDoubleVectorAvg()
+ agg.update(buffer, createDoubleInputRow(Array(1.0, 2.0)))
+ agg.update(buffer, createNullInputRow())
+ agg.update(buffer, createDoubleInputRowWithNullElement(Array(null, 10.0)))
+ agg.update(buffer, createDoubleInputRow(Array(3.0, 4.0)))
+ assertResult(Array(2.0, 3.0))(evalAsDoubleArray(agg, buffer))
+ }
+
+ test("VectorAvg - double: empty vectors") {
+ val (agg, buffer) = createDoubleVectorAvg()
+ agg.update(buffer, createDoubleInputRow(Array.empty[Double]))
+ agg.update(buffer, createDoubleInputRow(Array.empty[Double]))
+ assertResult(Array.empty[Double])(evalAsDoubleArray(agg, buffer))
+ }
+
+ test("VectorAvg - double: special values") {
+ val (agg, buffer) = createDoubleVectorAvg()
+ agg.update(buffer, createDoubleInputRow(Array(1.0, 2.0, Double.MaxValue)))
+ agg.update(buffer,
+ createDoubleInputRow(Array(Double.NaN, Double.NegativeInfinity,
Double.MaxValue)))
+ val result = evalAsDoubleArray(agg, buffer)
+ assert(result(0).isNaN)
+ assertResult(Double.NegativeInfinity)(result(1))
+ assertResult(Double.MaxValue)(result(2))
+ }
+
+ test("VectorAvg - double: opposite extreme values do not overflow") {
+ val (agg, buffer) = createDoubleVectorAvg()
+ agg.update(buffer, createDoubleInputRow(Array(Double.MaxValue,
-Double.MaxValue)))
+ agg.update(buffer, createDoubleInputRow(Array(-Double.MaxValue,
Double.MaxValue)))
+ assertResult(Array(0.0, 0.0))(evalAsDoubleArray(agg, buffer))
+
+ val (_, buffer1) = createDoubleVectorAvg()
+ val (_, buffer2) = createDoubleVectorAvg()
+ agg.update(buffer1, createDoubleInputRow(Array(Double.MaxValue,
-Double.MaxValue)))
+ agg.update(buffer2, createDoubleInputRow(Array(-Double.MaxValue,
Double.MaxValue)))
+ agg.merge(buffer1, buffer2)
Review Comment:
**Nit (P3):** This merge combines two single-row buffers with opposite
extremes, a case where the common sum-then-divide form `(leftAvg * currentCount
+ rightAvg * inputCount) / newCount` also returns `0.0`. A bit-exact binary64
evaluation shows that this rewrite passes every double merge assertion in this
PR, including the `repartition(2)`, `observe` and non-finite cases. Yet it
returns `Infinity` instead of `Double.MaxValue` when merging two
`[Double.MaxValue]` partials. More generally, it overflows whenever a partial
average times its row count exceeds `Double.MaxValue`. So no test guards the
overflow safety that weighting each partial average gives multi-partition
aggregation, `observe` and streaming state merges.
The float tests have the same gap, but the float merge can itself overflow
once counts exceed 2^24, while the double merge stays finite. So the property
is worth pinning here. Some count splits, such as one-to-two or three-to-five,
legitimately round one ulp below `Double.MaxValue`, so exact assertions need
splits that land on it.
**Verification:**
- **Regression:** Merge two double vector_avg buffers whose partial averages
hold Double.MaxValue and -Double.MaxValue in matching positions, with equal
counts and with a one-to-three count split. The result must be exactly
Double.MaxValue and -Double.MaxValue in those positions. A merge that
multiplies each average by its count before dividing would produce infinities
and fail.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/vectorExpressions.scala:
##########
@@ -778,7 +782,55 @@ object VectorAvg {
}
}
- private val kernels: Seq[VectorAggregateKernel] = Seq(FloatAvg)
+ private object DoubleAvg extends VectorAggregateKernel.DoubleKernel {
+ override def updateElements(
+ accBytes: Array[Byte],
+ inputArray: ArrayData,
+ dim: Int,
+ newCount: Long): Unit = {
+ // Update running average: new_avg = old_avg + (new_value - old_avg) /
new_count
Review Comment:
**Nit (P3):** This states the plain running-average formula, but the code
below departs from it in exactly the cases it is there to handle. For `oldAvg =
+Infinity` and `value = 1.0` at count 2, the formula gives `NaN` while the code
returns `+Infinity`. For `Double.MaxValue` and `-Double.MaxValue`, it gives
`-Infinity` while the code returns `0.0`. The comment predates the overflow and
non-finite branches and was not updated with them; `FloatAvg` documents the
same cases. Suggested wording:
```suggestion
// Finite deltas use: new_avg = old_avg + (new_value - old_avg) *
invCount.
// When subtracting finite operands overflows, both are scaled by
invCount first.
// When either operand is non-finite, addition gives the IEEE result
in either order:
// infinities absorb finite values and same-sign infinities; Inf +
(-Inf) and NaN give NaN.
```
**Verification:**
- **Inspection:** The comment on the double vector_avg update loop states
the finite-delta formula with the precomputed reciprocal of the count, and
names the overflowing-subtraction and non-finite cases with their results. It
must be consistent with the float kernel's documentation and with the code
beneath it.
--
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]
