[SPARK-20687] mllib.Matrices.fromBreeze may cause crash when converting breeze CSCMatrix

In an operation of two A, B CSCMatrices the resulting C matrix may have some extra 0s
in rowIndices and data which are created for performance improvement by Breeze.
This causes problems on converting back to mllib.Matrix because it relies on
rowIndices and data being coherent with colPtrs. Therefore it is necessary to truncate
rowIndices and data to the active number of elements hold by the C matrix, before
creating a Spark's SparseMatrix.

Author: ghoto

mllib/src/main/scala/org/apache/spark/mllib/linalg/Matrices.scala

@@ -992,7 +992,20 @@ object Matrices {
         new DenseMatrix(dm.rows, dm.cols, dm.data, dm.isTranspose)
       case sm: BSM[Double] =>
         // There is no isTranspose flag for sparse matrices in Breeze
-        new SparseMatrix(sm.rows, sm.cols, sm.colPtrs, sm.rowIndices, sm.data)
+
+        // Some Breeze CSCMatrices may have extra trailing zeros in
+        // .rowIndices and .data, which are added after some matrix
+        // operations for efficiency.
+        //
+        // Therefore the last element of sm.colPtrs would no longer be
+        // coherent with the size of sm.rowIndices and sm.data
+        // despite sm being a valid CSCMatrix.
+        // We need to truncate both arrays (rowIndices, data)
+        // to the real size of the vector sm.activeSize to allow valid conversion
+
+        val truncRowIndices = sm.rowIndices.slice(0, sm.activeSize)
+        val truncData = sm.data.slice(0, sm.activeSize)
+        new SparseMatrix(sm.rows, sm.cols, sm.colPtrs, truncRowIndices, truncData)
       case _ =>
         throw new UnsupportedOperationException(
           s"Do not support conversion from type ${breeze.getClass.getName}.")

mllib/src/test/scala/org/apache/spark/mllib/linalg/MatricesSuite.scala

@@ -46,25 +46,21 @@ class MatricesSuite extends SparkFunSuite {
     }
   }
 
-  test("breeze conversion bug") {
+  test("Test Breeze Conversion Bug - SPARK-20687") {
     // (2, 0, 0)
     // (2, 0, 0)
     val mat1Brz = Matrices.sparse(2, 3, Array(0, 2, 2, 2), Array(0, 1), Array(2, 2)).asBreeze
     // (2, 1E-15, 1E-15)
-    // (2, 1E-15, 1E-15
+    // (2, 1E-15, 1E-15)
     val mat2Brz = Matrices.sparse(2, 3, Array(0, 2, 4, 6), Array(0, 0, 0, 1, 1, 1), Array(2, 1E-15, 1E-15, 2, 1E-15, 1E-15)).asBreeze
-    // The following shouldn't break
-    val t01 = mat1Brz - mat1Brz
-    val t02 = mat2Brz - mat2Brz
-    val t02Brz = Matrices.fromBreeze(t02)
-    val t01Brz = Matrices.fromBreeze(t01)
-
     val t1Brz = mat1Brz - mat2Brz
     val t2Brz = mat2Brz - mat1Brz
-    // The following ones should break
+    // The following operations raise exceptions on un-patch Matrices.fromBreeze
     val t1 = Matrices.fromBreeze(t1Brz)
     val t2 = Matrices.fromBreeze(t2Brz)
-
+    // t1 == t1Brz && t2 == t2Brz
+    assert((t1.asBreeze - t1Brz).iterator.map((x) => math.abs(x._2)).sum < 1E-15)
+    assert((t2.asBreeze - t2Brz).iterator.map((x) => math.abs(x._2)).sum < 1E-15)
   }
 
   test("sparse matrix construction") {