You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by yi...@apache.org on 2026/08/05 07:01:56 UTC
(spark) branch branch-4.3 updated: [SPARK-58177][GRAPHX][TEST][FOLLOWUP] Add regression test for SVD++ aggregateMessages combiner
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. <mg...@uci.edu>
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. <mg...@uci.edu>
Signed-off-by: Yicong Huang <17...@users.noreply.github.com>
(cherry picked from commit 15b4cdf0b9e148d6bc5205e45edd67785ae64a74)
Signed-off-by: Yicong Huang <17...@users.noreply.github.com>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org