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 7b282ec6d24ee..beb15b0ccbc56 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 da139cec5c3d5..62f257d2cce3c 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)) + } }