zhidongqu-db commented on code in PR #59615:
URL: https://github.com/apache/spark/pull/59615#discussion_r4365108888
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/vectorExpressions.scala:
##########
@@ -674,11 +676,74 @@ object VectorAvg {
newCount: Long): Unit = {
// Update running average: new_avg = old_avg + (new_value - old_avg) /
new_count
val invCount = 1.0f / newCount
+ if (dim < 4) {
+ updateRemainingElements(accBytes, inputArray, 0, dim, invCount)
+ } else {
+ updateElementsInBlocks(accBytes, inputArray, dim, invCount)
+ }
+ }
+
+ private def updateElementsInBlocks(
+ accBytes: Array[Byte],
+ inputArray: ArrayData,
+ dim: Int,
+ invCount: Float): Unit = {
var i = 0
+ while (i <= dim - 4) {
+ val off = Platform.BYTE_ARRAY_OFFSET + i.toLong * 4
+ val oldAvg0 = Platform.getFloat(accBytes, off)
+ val delta0 = inputArray.getFloat(i) - oldAvg0
+ val oldAvg1 = Platform.getFloat(accBytes, off + 4)
+ val delta1 = inputArray.getFloat(i + 1) - oldAvg1
+ val oldAvg2 = Platform.getFloat(accBytes, off + 8)
+ val delta2 = inputArray.getFloat(i + 2) - oldAvg2
+ val oldAvg3 = Platform.getFloat(accBytes, off + 12)
+ val delta3 = inputArray.getFloat(i + 3) - oldAvg3
+ // A finite sum guarantees finite deltas; sum overflow only triggers a
scalar fallback.
+ val deltaSum = (delta0 + delta1) + (delta2 + delta3)
+ if (!JFloat.isFinite(deltaSum)) {
+ updateRemainingElements(accBytes, inputArray, i, dim, invCount)
+ return
Review Comment:
Returning here sends every remaining element of the vector through the
scalar path. An Inf/NaN average never goes away, so one non-finite element near
the start of a long vector (e.g. element 0 of a 1536-dim embedding) puts every
later row of that group on the scalar path for the whole vector.
The check runs before any write in the block, and each element depends only
on its own old value and input. So it should be safe to handle just this block
on the scalar path and keep going in blocks, with bit-identical results:
```scala
if (!JFloat.isFinite(deltaSum)) {
updateRemainingElements(accBytes, inputArray, i, i + 4, invCount) // `dim`
param -> `end`
} else {
Platform.putFloat(accBytes, off, oldAvg0 + delta0 * invCount)
// ... other three puts unchanged
}
i += 4
```
In a rough standalone microbenchmark (Unsafe-backed arrays rather than
`ArrayData`, so indicative only) with an Inf at element 0 and dim 1536, this
was about 25-30% faster than the current code, and unchanged for all-finite
input.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/vectorExpressions.scala:
##########
@@ -674,11 +676,74 @@ object VectorAvg {
newCount: Long): Unit = {
// Update running average: new_avg = old_avg + (new_value - old_avg) /
new_count
val invCount = 1.0f / newCount
+ if (dim < 4) {
+ updateRemainingElements(accBytes, inputArray, 0, dim, invCount)
+ } else {
+ updateElementsInBlocks(accBytes, inputArray, dim, invCount)
+ }
+ }
+
+ private def updateElementsInBlocks(
+ accBytes: Array[Byte],
+ inputArray: ArrayData,
+ dim: Int,
+ invCount: Float): Unit = {
var i = 0
+ while (i <= dim - 4) {
+ val off = Platform.BYTE_ARRAY_OFFSET + i.toLong * 4
+ val oldAvg0 = Platform.getFloat(accBytes, off)
+ val delta0 = inputArray.getFloat(i) - oldAvg0
+ val oldAvg1 = Platform.getFloat(accBytes, off + 4)
+ val delta1 = inputArray.getFloat(i + 1) - oldAvg1
+ val oldAvg2 = Platform.getFloat(accBytes, off + 8)
+ val delta2 = inputArray.getFloat(i + 2) - oldAvg2
+ val oldAvg3 = Platform.getFloat(accBytes, off + 12)
+ val delta3 = inputArray.getFloat(i + 3) - oldAvg3
+ // A finite sum guarantees finite deltas; sum overflow only triggers a
scalar fallback.
+ val deltaSum = (delta0 + delta1) + (delta2 + delta3)
+ if (!JFloat.isFinite(deltaSum)) {
+ updateRemainingElements(accBytes, inputArray, i, dim, invCount)
+ return
+ }
+ Platform.putFloat(accBytes, off, oldAvg0 + delta0 * invCount)
+ Platform.putFloat(accBytes, off + 4, oldAvg1 + delta1 * invCount)
+ Platform.putFloat(accBytes, off + 8, oldAvg2 + delta2 * invCount)
+ Platform.putFloat(accBytes, off + 12, oldAvg3 + delta3 * invCount)
+ i += 4
+ }
+ if (i < dim) {
+ updateRemainingElements(accBytes, inputArray, i, dim, invCount)
+ }
+ }
+
+ private def updateRemainingElements(
+ accBytes: Array[Byte],
+ inputArray: ArrayData,
+ start: Int,
+ dim: Int,
+ invCount: Float): Unit = {
+ var i = start
while (i < dim) {
val off = Platform.BYTE_ARRAY_OFFSET + i.toLong * 4
val oldAvg = Platform.getFloat(accBytes, off)
- Platform.putFloat(accBytes, off, oldAvg + (inputArray.getFloat(i) -
oldAvg) * invCount)
+ val value = inputArray.getFloat(i)
+ val delta = value - oldAvg
+ val newAvg = if (JFloat.isFinite(delta)) {
+ oldAvg + delta * invCount
+ } else if (JFloat.isFinite(oldAvg) && JFloat.isFinite(value)) {
+ // Scale before subtracting when opposite finite values overflow the
delta.
+ // For oldAvg = Float.MaxValue, value = -Float.MaxValue and invCount
= 0.5f,
+ // the delta is -Infinity, but scaling first gives the correct
average of 0.0f.
+ oldAvg + (value * invCount - oldAvg * invCount)
Review Comment:
This isn't introduced by this PR, but it's related to the overflow claim in
the migration note: the unchanged `mergeElements` can still overflow on finite
inputs. `leftWeight` and `rightWeight` are rounded to float independently, so
they can sum to more than 1. For example, merging two buffers whose average is
`Float.MaxValue`, with counts 11 and 161043324, gives `leftWeight = 6.83046E-8`
and `rightWeight = 1.0`, and `MaxValue * leftWeight + MaxValue * rightWeight`
is `Infinity`. The update path keeps `MaxValue` for the same values.
The Final/PartialMerge stage of a regular aggregation goes through `merge`,
so partial-aggregation plans can still return Infinity for finite inputs. This
could be fixed here (e.g. `leftAvg + (rightAvg - leftAvg) * rightWeight` with
the same overflow handling as the update path) or tracked as a follow-up, with
the migration note scoped accordingly.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/vectorExpressions.scala:
##########
@@ -674,11 +676,74 @@ object VectorAvg {
newCount: Long): Unit = {
// Update running average: new_avg = old_avg + (new_value - old_avg) /
new_count
val invCount = 1.0f / newCount
+ if (dim < 4) {
+ updateRemainingElements(accBytes, inputArray, 0, dim, invCount)
Review Comment:
nit: The comment still says `/ new_count`, but the code multiplies by
`invCount`, and the formula now only describes the finite-delta case. Maybe
something like: `// Running average: new_avg = old_avg + (new_value - old_avg)
* invCount, with special handling for overflow and non-finite values in
updateRemainingElements.`
Also, `updateRemainingElements(accBytes, inputArray, 0, dim, invCount)`
covers the whole vector here, so a name like `updateElementsScalar` might read
better.
##########
docs/sql-migration-guide.md:
##########
@@ -28,6 +28,7 @@ license: |
## Upgrading from Spark SQL 4.3 to 4.4
+- Since Spark 4.4, `vector_avg` improves overflow and non-finite value
handling for `ARRAY<FLOAT>` inputs. Extreme finite values that previously
produced infinity may now produce a finite average. Infinity combined with
finite values or infinity of the same sign now remains infinity instead of
producing `NaN`. Opposite infinities still produce `NaN`, and `NaN` inputs
continue to propagate.
Review Comment:
A few things on this note:
- Master is 5.0.0-SNAPSHOT, and there is an "Upgrading from Spark SQL 4.4 to
5.0" section above this one. Please confirm the target release. If this is
backported to branch-4.2/branch-4.x (both have the same bug), "Since Spark 4.4"
would need to change as well.
- "now remains infinity" reads as if the result was already infinity, but
previously these produced `NaN`. Suggest: "Infinity combined with a finite
value, or with infinity of the same sign, now produces infinity instead of
`NaN`." Strictly speaking, this only changed when the infinity is already in
the running average; a finite value followed by infinity already produced
infinity.
- "Extreme finite values that previously produced infinity may now produce a
finite average" holds for the update path only. See the comment on
`mergeElements` overflow on the scale-first branch.
- This line currently conflicts with master, where a Variant bullet was
added at the same spot. Keeping both bullets resolves it.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/VectorAggSuite.scala:
##########
@@ -405,6 +405,151 @@ class VectorAggSuite extends SparkFunSuite {
assert(result(1) === 3.0f)
}
+ test("VectorAvg - opposite extreme values do not overflow") {
+ val (agg, buffer, _) = createVectorAvg()
+ agg.update(buffer, createInputRow(Array(Float.MaxValue, -Float.MaxValue)))
+ agg.update(buffer, createInputRow(Array(-Float.MaxValue, Float.MaxValue)))
+ assertResult(Array(0.0f, 0.0f))(evalAsFloatArray(agg, buffer))
+
+ val (_, buffer1, _) = createVectorAvg()
+ val (_, buffer2, _) = createVectorAvg()
+ agg.update(buffer1, createInputRow(Array(Float.MaxValue, -Float.MaxValue)))
+ agg.update(buffer2, createInputRow(Array(-Float.MaxValue, Float.MaxValue)))
+ agg.merge(buffer1, buffer2)
+ assertResult(Array(0.0f, 0.0f))(evalAsFloatArray(agg, buffer1))
+ }
+
+ test("VectorAvg - finite overflow preserves cumulative weights") {
+ val (agg, buffer, _) = createVectorAvg()
+ val first = Array(Float.MaxValue, -Float.MaxValue)
+ agg.update(buffer, createInputRow(first))
+ agg.update(buffer, createInputRow(first))
+ // The third update overflows the delta. An unweighted midpoint would
incorrectly return zero.
+ agg.update(buffer, createInputRow(first.map(-_)))
+ val expected = (Float.MaxValue.toDouble / 3).toFloat
+ assertFloatArrayEquals(
+ evalAsFloatArray(agg, buffer),
+ Array(expected, -expected),
+ tolerance = math.ulp(expected) * 2)
+ }
+
+ test("VectorAvg - non-finite results do not depend on input order") {
+ val first =
+ Array(Float.PositiveInfinity, 1.0f, Float.NegativeInfinity,
Float.PositiveInfinity)
+ val second = Array(1.0f, Float.PositiveInfinity, 1.0f,
Float.NegativeInfinity)
+ def checkResult(result: Array[Float]): Unit = {
+ assertResult(Float.PositiveInfinity)(result(0))
+ assertResult(Float.PositiveInfinity)(result(1))
+ assertResult(Float.NegativeInfinity)(result(2))
+ assert(result(3).isNaN)
+ }
+
+ Seq(Seq(first, second), Seq(second, first)).foreach { vectors =>
+ val (agg, buffer, _) = createVectorAvg()
+ vectors.foreach(v => agg.update(buffer, createInputRow(v)))
+ checkResult(evalAsFloatArray(agg, buffer))
+ agg.update(buffer, createInputRow(Array(2.0f, 2.0f, 2.0f, 2.0f)))
+ checkResult(evalAsFloatArray(agg, buffer))
+
+ val (_, buffer1, _) = createVectorAvg()
+ val (_, buffer2, _) = createVectorAvg()
+ agg.update(buffer1, createInputRow(vectors.head))
+ agg.update(buffer2, createInputRow(vectors(1)))
+ agg.merge(buffer1, buffer2)
+ checkResult(evalAsFloatArray(agg, buffer1))
+ }
+ }
+
+ test("VectorAvg - finite elements around non-finite deltas") {
+ val first = Array(1.0f, Float.MaxValue, 3.0f, Float.PositiveInfinity, 5.0f,
+ Float.NegativeInfinity, 7.0f, Float.PositiveInfinity, 9.0f, Float.NaN,
11.0f)
+ val second = Array(3.0f, -Float.MaxValue, 5.0f, 1.0f, 7.0f, 1.0f, 9.0f,
+ Float.NegativeInfinity, 11.0f, 2.0f, 13.0f)
+ val expected = Array(2.0f, 0.0f, 4.0f, Float.PositiveInfinity, 6.0f,
+ Float.NegativeInfinity, 8.0f, Float.NaN, 10.0f, Float.NaN, 12.0f)
+
+ Seq(false, true).foreach { useGenericArray =>
+ val (agg, buffer, _) = createVectorAvg()
+ Seq(first, second).foreach { values =>
+ val array = if (useGenericArray) {
+ new GenericArrayData(values)
+ } else {
+ ArrayData.toArrayData(values)
+ }
+ agg.update(buffer, InternalRow(array))
+ }
+ assertResult(expected.map(JFloat.floatToIntBits))(
+ evalAsFloatArray(agg, buffer).map(JFloat.floatToIntBits))
+ }
+ }
+
+ test("VectorAvg - finite updates preserve running-average rounding") {
+ Seq(false, true).foreach { includeInfinity =>
+ val (agg, buffer, _) = createVectorAvg()
+ val first = Array.tabulate(17)(i => if (i % 2 == 0) 1.0f else -1.0f)
+ val last = first.map(_ * 6)
+ val expected = Array.tabulate(17)(i => if (i % 2 == 0) 0x402aaaab else
0xc02aaaab)
+ if (includeInfinity) {
+ first(8) = Float.PositiveInfinity
+ last(8) = 1.0f
+ expected(8) = 0x7f800000
+ }
+ agg.update(buffer, createInputRow(first))
+ agg.update(buffer, createInputRow(first))
+ agg.update(buffer, createInputRow(last))
+ assertResult(expected)(
+ evalAsFloatArray(agg, buffer).map(JFloat.floatToRawIntBits))
Review Comment:
nit: This uses `floatToRawIntBits`, while the other new tests use
`floatToIntBits`. The first new test uses `assertResult(Array(0.0f, 0.0f))`,
i.e. `==`, which also accepts `-0.0f`. Using `floatToIntBits` throughout would
be consistent.
Also, the test name says "finite updates", but the `includeInfinity` variant
covers an infinite element. Maybe "... rounding with and without a non-finite
element".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/vectorExpressions.scala:
##########
@@ -674,11 +676,74 @@ object VectorAvg {
newCount: Long): Unit = {
// Update running average: new_avg = old_avg + (new_value - old_avg) /
new_count
val invCount = 1.0f / newCount
+ if (dim < 4) {
+ updateRemainingElements(accBytes, inputArray, 0, dim, invCount)
+ } else {
+ updateElementsInBlocks(accBytes, inputArray, dim, invCount)
+ }
+ }
+
+ private def updateElementsInBlocks(
+ accBytes: Array[Byte],
+ inputArray: ArrayData,
+ dim: Int,
+ invCount: Float): Unit = {
var i = 0
+ while (i <= dim - 4) {
+ val off = Platform.BYTE_ARRAY_OFFSET + i.toLong * 4
+ val oldAvg0 = Platform.getFloat(accBytes, off)
+ val delta0 = inputArray.getFloat(i) - oldAvg0
+ val oldAvg1 = Platform.getFloat(accBytes, off + 4)
+ val delta1 = inputArray.getFloat(i + 1) - oldAvg1
+ val oldAvg2 = Platform.getFloat(accBytes, off + 8)
+ val delta2 = inputArray.getFloat(i + 2) - oldAvg2
+ val oldAvg3 = Platform.getFloat(accBytes, off + 12)
+ val delta3 = inputArray.getFloat(i + 3) - oldAvg3
+ // A finite sum guarantees finite deltas; sum overflow only triggers a
scalar fallback.
+ val deltaSum = (delta0 + delta1) + (delta2 + delta3)
+ if (!JFloat.isFinite(deltaSum)) {
+ updateRemainingElements(accBytes, inputArray, i, dim, invCount)
+ return
+ }
+ Platform.putFloat(accBytes, off, oldAvg0 + delta0 * invCount)
+ Platform.putFloat(accBytes, off + 4, oldAvg1 + delta1 * invCount)
+ Platform.putFloat(accBytes, off + 8, oldAvg2 + delta2 * invCount)
+ Platform.putFloat(accBytes, off + 12, oldAvg3 + delta3 * invCount)
+ i += 4
+ }
+ if (i < dim) {
+ updateRemainingElements(accBytes, inputArray, i, dim, invCount)
+ }
+ }
+
+ private def updateRemainingElements(
+ accBytes: Array[Byte],
+ inputArray: ArrayData,
+ start: Int,
+ dim: Int,
+ invCount: Float): Unit = {
+ var i = start
while (i < dim) {
val off = Platform.BYTE_ARRAY_OFFSET + i.toLong * 4
val oldAvg = Platform.getFloat(accBytes, off)
- Platform.putFloat(accBytes, off, oldAvg + (inputArray.getFloat(i) -
oldAvg) * invCount)
+ val value = inputArray.getFloat(i)
+ val delta = value - oldAvg
+ val newAvg = if (JFloat.isFinite(delta)) {
+ oldAvg + delta * invCount
+ } else if (JFloat.isFinite(oldAvg) && JFloat.isFinite(value)) {
+ // Scale before subtracting when opposite finite values overflow the
delta.
+ // For oldAvg = Float.MaxValue, value = -Float.MaxValue and invCount
= 0.5f,
+ // the delta is -Infinity, but scaling first gives the correct
average of 0.0f.
+ oldAvg + (value * invCount - oldAvg * invCount)
+ } else {
+ // IEEE addition handles these cases in either operand order:
+ // Inf + Inf -> Inf; -Inf + (-Inf) -> -Inf.
+ // Inf + (-Inf) -> NaN.
+ // NaN + any value -> NaN.
+ // Inf + finite -> Inf; -Inf + finite -> -Inf.
Review Comment:
nit: "in either operand order" is a bit confusing because addition is
commutative. I think the point is that the result doesn't depend on which
operand is the running average. Maybe: "At least one operand is non-finite.
Plain addition gives the IEEE result whichever operand is the running average:
same-sign infinities keep their sign, opposite infinities and any NaN give NaN,
and an infinity absorbs a finite value."
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/VectorAggSuite.scala:
##########
@@ -405,6 +405,151 @@ class VectorAggSuite extends SparkFunSuite {
assert(result(1) === 3.0f)
}
+ test("VectorAvg - opposite extreme values do not overflow") {
+ val (agg, buffer, _) = createVectorAvg()
+ agg.update(buffer, createInputRow(Array(Float.MaxValue, -Float.MaxValue)))
+ agg.update(buffer, createInputRow(Array(-Float.MaxValue, Float.MaxValue)))
+ assertResult(Array(0.0f, 0.0f))(evalAsFloatArray(agg, buffer))
+
+ val (_, buffer1, _) = createVectorAvg()
+ val (_, buffer2, _) = createVectorAvg()
+ agg.update(buffer1, createInputRow(Array(Float.MaxValue, -Float.MaxValue)))
+ agg.update(buffer2, createInputRow(Array(-Float.MaxValue, Float.MaxValue)))
+ agg.merge(buffer1, buffer2)
+ assertResult(Array(0.0f, 0.0f))(evalAsFloatArray(agg, buffer1))
+ }
+
+ test("VectorAvg - finite overflow preserves cumulative weights") {
+ val (agg, buffer, _) = createVectorAvg()
+ val first = Array(Float.MaxValue, -Float.MaxValue)
+ agg.update(buffer, createInputRow(first))
+ agg.update(buffer, createInputRow(first))
+ // The third update overflows the delta. An unweighted midpoint would
incorrectly return zero.
+ agg.update(buffer, createInputRow(first.map(-_)))
+ val expected = (Float.MaxValue.toDouble / 3).toFloat
+ assertFloatArrayEquals(
+ evalAsFloatArray(agg, buffer),
+ Array(expected, -expected),
+ tolerance = math.ulp(expected) * 2)
+ }
+
+ test("VectorAvg - non-finite results do not depend on input order") {
+ val first =
+ Array(Float.PositiveInfinity, 1.0f, Float.NegativeInfinity,
Float.PositiveInfinity)
+ val second = Array(1.0f, Float.PositiveInfinity, 1.0f,
Float.NegativeInfinity)
+ def checkResult(result: Array[Float]): Unit = {
+ assertResult(Float.PositiveInfinity)(result(0))
+ assertResult(Float.PositiveInfinity)(result(1))
+ assertResult(Float.NegativeInfinity)(result(2))
+ assert(result(3).isNaN)
+ }
+
+ Seq(Seq(first, second), Seq(second, first)).foreach { vectors =>
+ val (agg, buffer, _) = createVectorAvg()
+ vectors.foreach(v => agg.update(buffer, createInputRow(v)))
+ checkResult(evalAsFloatArray(agg, buffer))
+ agg.update(buffer, createInputRow(Array(2.0f, 2.0f, 2.0f, 2.0f)))
+ checkResult(evalAsFloatArray(agg, buffer))
+
+ val (_, buffer1, _) = createVectorAvg()
+ val (_, buffer2, _) = createVectorAvg()
+ agg.update(buffer1, createInputRow(vectors.head))
+ agg.update(buffer2, createInputRow(vectors(1)))
+ agg.merge(buffer1, buffer2)
+ checkResult(evalAsFloatArray(agg, buffer1))
+ }
+ }
+
+ test("VectorAvg - finite elements around non-finite deltas") {
+ val first = Array(1.0f, Float.MaxValue, 3.0f, Float.PositiveInfinity, 5.0f,
+ Float.NegativeInfinity, 7.0f, Float.PositiveInfinity, 9.0f, Float.NaN,
11.0f)
+ val second = Array(3.0f, -Float.MaxValue, 5.0f, 1.0f, 7.0f, 1.0f, 9.0f,
+ Float.NegativeInfinity, 11.0f, 2.0f, 13.0f)
+ val expected = Array(2.0f, 0.0f, 4.0f, Float.PositiveInfinity, 6.0f,
+ Float.NegativeInfinity, 8.0f, Float.NaN, 10.0f, Float.NaN, 12.0f)
+
+ Seq(false, true).foreach { useGenericArray =>
+ val (agg, buffer, _) = createVectorAvg()
+ Seq(first, second).foreach { values =>
+ val array = if (useGenericArray) {
+ new GenericArrayData(values)
+ } else {
+ ArrayData.toArrayData(values)
+ }
Review Comment:
nit: This array-building block is duplicated in "non-finite deltas at
different positions" (lines 532-536). A small helper such as `def
toInputRow(values: Array[Float], useGenericArray: Boolean): InternalRow` would
remove both copies.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/VectorAggSuite.scala:
##########
@@ -86,7 +86,7 @@ class VectorAggSuite extends SparkFunSuite {
// Asserts that two floats are approximately equal within a tolerance.
def assertFloatEquals(actual: Float, expected: Float, tolerance: Float =
1e-5f): Unit = {
assert(
- java.lang.Float.isNaN(expected) && java.lang.Float.isNaN(actual) ||
+ JFloat.isNaN(expected) && JFloat.isNaN(actual) ||
Review Comment:
nit (pre-existing): This assertion fails when `expected` and `actual` are
the same infinity, because `math.abs(Inf - Inf)` is NaN. Nothing hits it today,
but it's easy to trip over when adding Inf cases with this helper. Adding
`actual == expected ||` would fix it.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/aggregate/VectorAggSuite.scala:
##########
@@ -405,6 +405,151 @@ class VectorAggSuite extends SparkFunSuite {
assert(result(1) === 3.0f)
}
+ test("VectorAvg - opposite extreme values do not overflow") {
+ val (agg, buffer, _) = createVectorAvg()
+ agg.update(buffer, createInputRow(Array(Float.MaxValue, -Float.MaxValue)))
+ agg.update(buffer, createInputRow(Array(-Float.MaxValue, Float.MaxValue)))
+ assertResult(Array(0.0f, 0.0f))(evalAsFloatArray(agg, buffer))
+
+ val (_, buffer1, _) = createVectorAvg()
+ val (_, buffer2, _) = createVectorAvg()
+ agg.update(buffer1, createInputRow(Array(Float.MaxValue, -Float.MaxValue)))
+ agg.update(buffer2, createInputRow(Array(-Float.MaxValue, Float.MaxValue)))
+ agg.merge(buffer1, buffer2)
+ assertResult(Array(0.0f, 0.0f))(evalAsFloatArray(agg, buffer1))
+ }
+
+ test("VectorAvg - finite overflow preserves cumulative weights") {
Review Comment:
This is the only test where the scale-first branch runs with `invCount !=
0.5f`. At count 2, scale-first and plain `oldAvg + value` both give 0 for
`MaxValue`/`-MaxValue`, so the other tests can't tell them apart. With dim 2,
it also only runs on the `dim < 4` path. Consider also covering block-sized
vectors, so the block fallback reaches the scale-first branch with a 1/3
weight, e.g.:
```scala
Seq(2, 4, 5, 8).foreach { dim =>
val (agg, buffer, _) = createVectorAvg()
val first = Array.tabulate(dim)(i => if (i % 2 == 0) Float.MaxValue else
-Float.MaxValue)
Seq(first, first, first.map(-_)).foreach(v => agg.update(buffer,
createInputRow(v)))
val expected = (Float.MaxValue.toDouble / 3).toFloat
assertFloatArrayEquals(
evalAsFloatArray(agg, buffer),
Array.tabulate(dim)(i => if (i % 2 == 0) expected else -expected),
tolerance = math.ulp(expected) * 2)
}
```
--
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]
