This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new b4cbf6b17c4f [SPARK-58177][GRAPHX][TEST][FOLLOWUP] Add regression test
for SVD++ aggregateMessages combiner
b4cbf6b17c4f is described below
commit b4cbf6b17c4faa64201d45f7c21fff86d4b407d2
Author: Matthew B. <[email protected]>
AuthorDate: Wed Aug 5 00:01:25 2026 -0700
[SPARK-58177][GRAPHX][TEST][FOLLOWUP] Add regression test for SVD++
aggregateMessages combiner
### What changes were proposed in this pull request?
Follow-up to
[SPARK-58177](https://issues.apache.org/jira/browse/SPARK-58177) / #57314,
which fixed a wrong operand in the SVD++ Phase 2 `aggregateMessages` combiner
but merged without a test. This adds the regression test:
- Extract the Phase 2 training-message merge logic into a package-private
`SVDPlusPlus.combineTrainMessages`, and use it in place of the inline
`aggregateMessages` combiner.
- Add a unit test in `SVDPlusPlusSuite` that asserts both vectors and the
scalar are summed element-wise, and that the input messages are not mutated.
### Why are the changes needed?
The fixed bug (the second vector was based on `g2._2`, which dropped
`g1._2` and doubled `g2._2`) only manifests when a vertex merges two or more
differing messages, and the combiner was an inline lambda that could not be
tested directly. Extracting it makes the merge logic unit-testable and guards
against a future regression.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added a unit test in `SVDPlusPlusSuite`; the full suite passes locally. The
test was confirmed to fail when the pre-fix operand (`g2._2`) is restored and
to pass with the fix in place.
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #57392 from Ma77Ball/SPARK-58231-svdpp-combiner-regression-test.
Authored-by: Matthew B. <[email protected]>
Signed-off-by: Yicong Huang <[email protected]>
(cherry picked from commit 15b4cdf0b9e148d6bc5205e45edd67785ae64a74)
Signed-off-by: Yicong Huang <[email protected]>
---
.../org/apache/spark/graphx/lib/SVDPlusPlus.scala | 20 ++++++++++++--------
.../apache/spark/graphx/lib/SVDPlusPlusSuite.scala | 15 +++++++++++++++
2 files changed, 27 insertions(+), 8 deletions(-)
diff --git
a/graphx/src/main/scala/org/apache/spark/graphx/lib/SVDPlusPlus.scala
b/graphx/src/main/scala/org/apache/spark/graphx/lib/SVDPlusPlus.scala
index 7b282ec6d24e..beb15b0ccbc5 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/lib/SVDPlusPlus.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/lib/SVDPlusPlus.scala
@@ -38,6 +38,17 @@ object SVDPlusPlus {
var gamma7: Double)
extends Serializable
+ /** Sums two Phase 2 training messages component-wise: both vectors and the
scalar. */
+ private[graphx] def combineTrainMessages(
+ g1: (Array[Double], Array[Double], Double),
+ g2: (Array[Double], Array[Double], Double)): (Array[Double],
Array[Double], Double) = {
+ val out1 = g1._1.clone()
+ BLAS.nativeBLAS.daxpy(out1.length, 1.0, g2._1, 1, out1, 1)
+ val out2 = g1._2.clone()
+ BLAS.nativeBLAS.daxpy(out2.length, 1.0, g2._2, 1, out2, 1)
+ (out1, out2, g1._3 + g2._3)
+ }
+
// scalastyle:off line.size.limit
/**
* Implement SVD++ based on "Factorization Meets the Neighborhood: a
Multifaceted
@@ -152,14 +163,7 @@ object SVDPlusPlus {
g.cache()
val t2 = g.aggregateMessages(
sendMsgTrainF(conf, u),
- (g1: (Array[Double], Array[Double], Double), g2: (Array[Double],
Array[Double], Double)) =>
- {
- val out1 = g1._1.clone()
- BLAS.nativeBLAS.daxpy(out1.length, 1.0, g2._1, 1, out1, 1)
- val out2 = g1._2.clone()
- BLAS.nativeBLAS.daxpy(out2.length, 1.0, g2._2, 1, out2, 1)
- (out1, out2, g1._3 + g2._3)
- })
+ combineTrainMessages)
val gJoinT2 = g.outerJoinVertices(t2) {
(vid: VertexId,
vd: (Array[Double], Array[Double], Double, Double),
diff --git
a/graphx/src/test/scala/org/apache/spark/graphx/lib/SVDPlusPlusSuite.scala
b/graphx/src/test/scala/org/apache/spark/graphx/lib/SVDPlusPlusSuite.scala
index da139cec5c3d..62f257d2cce3 100644
--- a/graphx/src/test/scala/org/apache/spark/graphx/lib/SVDPlusPlusSuite.scala
+++ b/graphx/src/test/scala/org/apache/spark/graphx/lib/SVDPlusPlusSuite.scala
@@ -49,4 +49,19 @@ class SVDPlusPlusSuite extends SparkFunSuite with
LocalSparkContext {
assert(graph.edges.count() == 0)
}
}
+
+ test("SPARK-58177: training-phase message combiner sums both vectors
element-wise") {
+ val g1 = (Array(1.0, 2.0), Array(10.0, 20.0), 100.0)
+ val g2 = (Array(3.0, 4.0), Array(30.0, 40.0), 200.0)
+ val (out1, out2, out3) = SVDPlusPlus.combineTrainMessages(g1, g2)
+ // Each component is the element-wise sum of the two messages.
+ assert(out1.toSeq === Seq(4.0, 6.0))
+ assert(out2.toSeq === Seq(40.0, 60.0))
+ assert(out3 === 300.0)
+ // Inputs are cloned, not mutated.
+ assert(g1._1.toSeq === Seq(1.0, 2.0))
+ assert(g1._2.toSeq === Seq(10.0, 20.0))
+ assert(g2._1.toSeq === Seq(3.0, 4.0))
+ assert(g2._2.toSeq === Seq(30.0, 40.0))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
