You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "dbtsai (via GitHub)" <gi...@apache.org> on 2026/10/15 01:05:57 UTC
[PR] [SPARK-60254][MLLIB] Replace Breeze delegation in vector and matrix primitives [spark]
dbtsai opened a new pull request, #59485:
URL: https://github.com/apache/spark/pull/59485
### What changes were proposed in this pull request?
Replace Breeze delegation in Spark's vector and matrix primitives:
- Dispatch vector element access to Spark's dense and sparse implementations.
- Convert sparse CSC/CSR layouts with counts, prefix sums, and scatter, preserving sorted indices and removing explicitly stored zeros without dense storage.
- Implement matrix equality and sparse hashing in Spark, retaining Breeze's existing floating-point and explicit-zero behavior.
- Add a shared matrix formatter that preserves output layout, truncation, transpose handling, and locale-sensitive formatting.
- Reuse the ML matrix implementations from legacy MLlib through the existing `asML` adapters.
- Add compatibility regressions and Breeze attribution in the NOTICE files.
This PR is limited to these primitives. The `asBreeze`/`fromBreeze` adapters and Breeze dependency declarations remain for existing algorithm consumers.
### Why are the changes needed?
These basic operations can work directly with Spark's dense and sparse storage. Moving them into Spark is a preparatory step toward reducing its dependence on Breeze and its transitive dependencies. This PR does not yet reduce the shipped dependency set.
### Does this PR introduce _any_ user-facing change?
No. Existing equality, sparse hash, and formatting behavior is preserved, including the compatibility edge cases for NaN, signed zeros, explicitly stored zeros, and transposed sparse matrices. There are no public API or dependency changes.
### How was this patch tested?
Added regression tests for sparse layout conversion, empty and very large sparse matrices, floating-point edge cases, equality and hashing parity, and formatting/truncation/locale parity. Legacy MLlib tests cover equality, sparse hashing, and formatting parity as well.
Local verification: **197 tests passed across 13 suites**, using a direct Scala compiler/ScalaTest fallback with cached dependencies:
- `mllib-local`: 107 tests across `MatricesSuite`, `VectorsSuite`, `BreezeMatrixConversionSuite`, `BreezeVectorConversionSuite`, `BLASSuite`, `MultivariateGaussianSuite`, `UtilsSuite`, and `TestingUtilsSuite`.
- Legacy `mllib`: 90 tests across `MatricesSuite`, `VectorsSuite`, `BreezeMatrixConversionSuite`, `BreezeVectorConversionSuite`, and `BLASSuite`.
The fallback compiled the current production and test sources without stubs or source/build-configuration changes. It used Java 21, Breeze 2.1.0, and cached netlib 3.0.4. The local-module run used Scala 2.13.17; the legacy run used Scala 2.13.18 and cached Spark 4.2.0-SNAPSHOT supporting binaries, including the real `SparkFunSuite`, with the newly compiled local/legacy linear-algebra classes first on the classpath.
**Build limitation:** the standard SBT invocation was attempted but could not download the sbt 1.13.0 launcher, so it never reached compilation or tests. These fallback results are not a full build against the current branch's pinned dependencies. Standard build and CI verification remain necessary.
Standard commands to reproduce the covered suites in a fully provisioned checkout:
```sh
build/sbt 'mllib-local/testOnly *MatricesSuite *VectorsSuite *BreezeMatrixConversionSuite *BreezeVectorConversionSuite *BLASSuite *MultivariateGaussianSuite *UtilsSuite *TestingUtilsSuite'
build/sbt 'mllib/testOnly org.apache.spark.mllib.linalg.MatricesSuite org.apache.spark.mllib.linalg.VectorsSuite org.apache.spark.mllib.linalg.BreezeMatrixConversionSuite org.apache.spark.mllib.linalg.BreezeVectorConversionSuite org.apache.spark.mllib.linalg.BLASSuite'
```
`git diff --check` and changed-source ASCII/line-length checks also passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60254][MLLIB] Replace Breeze delegation in vector and matrix primitives [spark]
Posted by "dbtsai (via GitHub)" <gi...@apache.org>.
dbtsai commented on PR #59485:
URL: https://github.com/apache/spark/pull/59485#issuecomment-6347208002
Thanks for the detailed review. Commit 653f04b5523 addresses the requested changes:
- Both sparse conversion passes now follow the same pointer ranges, including skipped backing entries and overlapping/nonmonotone ranges. The pointer array doubles as the scatter cursor, then has its starts restored.
- Adjacent duplicate coordinates are summed after scatter, retaining cancellation zeros. Summation uses stored order; exact floating-point roundoff from Breeze's unstable sort is not replicated for noncanonical duplicate input, and the PR description calls this out.
- Mixed dense/CSR equality normalizes the CSR operand before binary search.
- Out-of-range nonzero indices and negative dimensions now produce clear argument errors in the affected operations. Constructor validation is unchanged.
- Equality, sparse hashing, normalization, and an index-based sparse-entry cursor now live in a separate Breeze-headed `MatrixUtils` file. Formatting uses that cursor too, without per-column iterators or per-entry tuples.
- The dense `Int.MinValue` line-limit exception behavior is preserved. The locale test reaches the dense summary path; localized strings and hash values are pinned. Legacy tests check `asML` delegation directly. Explicit-zero equality is labeled as a known Breeze compatibility quirk.
- Breeze notices follow the ASF header, the unrelated EOF changes are reverted, and LICENSE inventories identify the two Breeze-headed helper files.
- The unused `Vector.apply` defaults now throw `UnsupportedOperationException`, matching `copy`; concrete accessors and the existing method ABI are retained. The PR description no longer presents that cleanup as a runtime dispatch change.
Validation: all 205 tests across the eight local and five legacy suites passed using freshly compiled sources and cached dependencies (no stubs). A randomized comparison of 3,000 canonical sparse matrices in both layouts against Breeze also passed. Scalastyle passed across 34 files, and `javap` confirmed the default vector methods remain concrete. Standard SBT and Apache RAT are blocked before validation by unavailable launcher/JAR downloads, so fresh CI remains necessary. The PR description details the fallback dependency differences and the malformed-input compatibility caveat.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60254][MLLIB] Replace Breeze delegation in vector and matrix primitives [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #59485:
URL: https://github.com/apache/spark/pull/59485#discussion_r4297407268
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/Matrices.scala:
##########
@@ -1329,6 +1384,57 @@ object Matrices {
}
}
+ private[ml] def equals(left: Matrix, right: Matrix): Boolean = {
+ if (left.numRows != right.numRows || left.numCols != right.numCols) {
+ return false
+ }
+ (left, right) match {
+ case (leftSparse: SparseMatrix, rightSparse: SparseMatrix) =>
+ // Adapted from Breeze's CSCMatrix.equals. Retain its treatment of explicit zeros:
+ // the last active entry is compared by coordinate even when its value is zero.
+ val leftEntries = sparseEntries(leftSparse)
+ val rightEntries = sparseEntries(rightSparse)
+ while (leftEntries.hasNext && rightEntries.hasNext) {
+ var leftEntry = leftEntries.next()
+ var rightEntry = rightEntries.next()
+ while (leftEntry._3 == 0.0 && leftEntries.hasNext) {
+ leftEntry = leftEntries.next()
+ }
+ while (rightEntry._3 == 0.0 && rightEntries.hasNext) {
+ rightEntry = rightEntries.next()
+ }
+ if (leftEntry._1 != rightEntry._1 || leftEntry._2 != rightEntry._2 ||
+ leftEntry._3 != rightEntry._3) {
+ return false
+ }
+ }
+ leftEntries.forall(_._3 == 0.0) && rightEntries.forall(_._3 == 0.0)
+ case _ =>
+ var j = 0
+ while (j < left.numCols) {
+ var i = 0
+ while (i < left.numRows) {
+ if (left(i, j) != right(i, j)) {
Review Comment:
**[Low] Dense vs transposed sparse equality now binary-searches the raw CSR arrays.**
The old code compared against `asBreeze` of the transposed matrix, which is the sorted and merged copy built by Breeze's transpose, so `CSCMatrix.apply` found every entry. Here `right(i, j)` goes to `SparseMatrix.index`, which runs `Arrays.binarySearch` over the stored CSR column indices (L709). A row whose column indices are unsorted or duplicated therefore gives false negatives. The sparse/sparse branch above normalizes through `toSparseColMajor`, which sorts, so the two branches now disagree and equality is not transitive.
For example:
- Take `t = new SparseMatrix(1, 3, Array(0, 2), Array(2, 0), Array(5.0, 7.0), isTransposed = true)` and `d = new DenseMatrix(1, 3, Array(7.0, 0.0, 5.0))`. `d == t` was true and is now false, because `t(0, 0)` misses in `[2, 0]` and returns 0.0. `t == t.toDense` also changes from true to false.
- Take `u = new SparseMatrix(1, 2, Array(0, 2), Array(1, 0), Array(1.0, 2.0), isTransposed = true)`, `s = new SparseMatrix(1, 2, Array(0, 1, 2), Array(0, 0), Array(2.0, 1.0))` and `d2 = new DenseMatrix(1, 2, Array(2.0, 1.0))`. Now `u == s` and `s == d2` are true, but `u == d2` is false. All three were true before.
Non-transposed CSC input already had the same miss before, so only transposed operands changed. If parity matters here, could a transposed sparse operand be normalized with `toSparseColMajor` before this element-wise loop? That restores the old result.
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/Matrices.scala:
##########
@@ -755,14 +779,45 @@ class SparseMatrix @Since("2.0.0") (
* order.
*/
private[ml] override def toSparseMatrix(colMajor: Boolean): SparseMatrix = {
- if (isColMajor && !colMajor) {
- // it is col major and we want row major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]].t
- Matrices.fromBreeze(breezeTransposed).transpose.asInstanceOf[SparseMatrix]
- } else if (isRowMajor && colMajor) {
- // it is row major and we want col major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]]
- Matrices.fromBreeze(breezeTransposed).asInstanceOf[SparseMatrix]
+ if (isColMajor != colMajor) {
+ val numPtrs = if (colMajor) numCols else numRows
+ val newColPtrs = new Array[Int](numPtrs + 1)
+ var i = 0
+ while (i < values.length) {
+ if (values(i) != 0.0) {
+ newColPtrs(rowIndices(i) + 1) += 1
Review Comment:
**[Low] An out-of-range index now throws `ArrayIndexOutOfBoundsException`, including from transposed `toString`, `hashCode` and `equals` that used to succeed.**
Nothing checks `rowIndices(i)` before `newColPtrs(rowIndices(i) + 1)`. The old Breeze builder checked the bound with a `require` (`IllegalArgumentException`), but not for the first sorted entry. So a transposed matrix whose only nonzero is out of range used to print and hash fine.
For example, take `t = new SparseMatrix(1, 2, Array(0, 1), Array(5), Array(1.0), isTransposed = true)`; the constructor does not check index ranges.
- `t.toString` was `1 x 2 CSCMatrix` and `t.hashCode` was 652125558. Both now throw `ArrayIndexOutOfBoundsException: Index 6 out of bounds for length 3`, and so does `t.equals` with another 1 x 2 `SparseMatrix`. `df.show()` on such a row now fails as well.
- `new SparseMatrix(2, 2, Array(0, 1, 2), Array(0, 5), Array(1.0, 2.0)).toSparseRowMajor` used to throw `IllegalArgumentException: requirement failed: Column index 5 is out of bounds for number of columns 2!`. It now throws `ArrayIndexOutOfBoundsException: Index 6 ...`, whose index is one more than the user's index.
- An index of -1 passes this line (`newColPtrs(0) += 1`) and fails later, at L810.
Could the count loop check the bound, e.g. `require(idx >= 0 && idx < numPtrs, ...)`, so that a malformed matrix fails with a clear message?
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/Matrices.scala:
##########
@@ -755,14 +779,45 @@ class SparseMatrix @Since("2.0.0") (
* order.
*/
private[ml] override def toSparseMatrix(colMajor: Boolean): SparseMatrix = {
- if (isColMajor && !colMajor) {
- // it is col major and we want row major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]].t
- Matrices.fromBreeze(breezeTransposed).transpose.asInstanceOf[SparseMatrix]
- } else if (isRowMajor && colMajor) {
- // it is row major and we want col major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]]
- Matrices.fromBreeze(breezeTransposed).asInstanceOf[SparseMatrix]
+ if (isColMajor != colMajor) {
+ val numPtrs = if (colMajor) numCols else numRows
+ val newColPtrs = new Array[Int](numPtrs + 1)
+ var i = 0
+ while (i < values.length) {
+ if (values(i) != 0.0) {
+ newColPtrs(rowIndices(i) + 1) += 1
+ }
+ i += 1
+ }
+ i = 0
+ while (i < numPtrs) {
+ newColPtrs(i + 1) += newColPtrs(i)
+ i += 1
+ }
+
+ val nnz = newColPtrs(numPtrs)
+ val newRowIndices = new Array[Int](nnz)
+ val newValues = new Array[Double](nnz)
+ val next = newColPtrs.clone()
+ var j = 0
+ // Visiting the original rows or columns in order keeps the new indices sorted.
+ while (j < colPtrs.length - 1) {
+ var idx = colPtrs(j)
+ val idxEnd = colPtrs(j + 1)
+ while (idx < idxEnd) {
+ if (values(idx) != 0.0) {
Review Comment:
**[Low] Duplicate coordinates are no longer summed.**
The old conversion went through Breeze's `canTranspose_CSC` and `CSCMatrix.Builder.result`, which sums entries with the same (row, column). This counting sort keeps each duplicate as a separate entry. The constructor does not reject duplicates (the Scaladoc only says that the indices "must be in strictly increasing order"). So for such input, the result now differs from Breeze, here and in the transposed `toString`, `hashCode` and `equals` that use this conversion.
For example, take `t = new SparseMatrix(1, 2, Array(0, 2), Array(1, 1), Array(1.0, 2.0), isTransposed = true)`.
- `t.toSparseColMajor` was `[0, 0, 1]` / `[0]` / `[3.0]`, and it is now `[0, 0, 2]` / `[0, 0]` / `[1.0, 2.0]`.
- `t.toString` printed `(0,1) 3.0`, and it now prints `(0,1) 1.0` and `(0,1) 2.0`.
- `t.hashCode` changes from -942016480 to -780899452.
- `t.equals(SparseMatrix.fromCOO(1, 2, Seq((0, 1, 3.0))))` changes from true to false.
The converted matrix can also disagree with its source in BLAS. Take `a = new SparseMatrix(1, 1, Array(0, 2), Array(0, 0), Array(1.0, 2.0))` and a one-element `SparseVector` `x`. `a.multiply(x)` sums both entries (3 * x0), but `a.toSparseRowMajor.multiply(x)` returns x0. The transposed sparse-sparse path in `BLAS.gemv` moves on after the first match.
Is keeping duplicates intended? If exact Breeze parity is the goal, the conversion could merge adjacent equal indices after the scatter. Otherwise, could the PR description say that only canonical input keeps the Breeze behavior?
##########
mllib-local/src/test/scala/org/apache/spark/ml/linalg/MatricesSuite.scala:
##########
@@ -105,6 +105,70 @@ class MatricesSuite extends SparkMLFunSuite {
assert(sm1 === dm2.transpose)
}
+ test("equals preserves Breeze floating-point and storage semantics") {
+ val values = Seq(0.0, -0.0, 1.0, Double.NaN,
+ Double.PositiveInfinity, Double.NegativeInfinity)
+ val matrices = values.flatMap { value =>
+ val dense = new DenseMatrix(2, 2, Array(0.0, value, 2.0, 0.0))
+ val sparse = new SparseMatrix(2, 2, Array(0, 1, 2), Array(1, 0), Array(value, 2.0))
+ Seq(dense, dense.copy, sparse, sparse.copy, dense.transpose, sparse.transpose)
+ } ++ Seq(
+ new DenseMatrix(0, 2, Array.emptyDoubleArray),
+ new DenseMatrix(2, 0, Array.emptyDoubleArray),
+ new SparseMatrix(0, 2, Array(0, 0, 0), Array.emptyIntArray, Array.emptyDoubleArray),
+ new SparseMatrix(2, 0, Array(0), Array.emptyIntArray, Array.emptyDoubleArray))
+
+ for (left <- matrices; right <- matrices) {
+ assert(left.equals(right) === left.asBreeze.equals(right.asBreeze))
Review Comment:
**[Low, tests] Most of the new coverage is parity with `asBreeze`, without pinned values.**
The new tests compare with `asBreeze` at 11 places, about 2,000 checks at runtime: here, L157, L1027, L1030 and L1048-1050, plus L149, L155, L552 and L554 in the mllib suite. Only 16 assertions pin literal results. In particular, every sparse hash value and the whole locale output are checked only against Breeze. That is a strong differential test while Breeze is on the classpath. However, the PR describes itself as a step toward reducing the Breeze dependency, and once `asBreeze` goes away these checks would have to be deleted or rewritten without the reference.
Could some expected values be pinned now, while Breeze 2.1.0 can still generate them? For example, the six hash values in "sparse hashCode preserves Breeze value hashing" are 1999056974 and -274380619 for `sparse` and its transpose, -1538214276 and 2094930921 for `ordinary`, and 43 and 43 for `empty`. The `ar-EG` sparse output could be a literal string. Also, mllib now just delegates to `asML`, so the mllib tests could assert the delegation directly, e.g. `m.toString(l, w) == m.asML.toString(l, w)` and `m.hashCode == m.asML.hashCode`.
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/MatrixFormatter.scala:
##########
@@ -0,0 +1,115 @@
+// scalastyle:off license
+/*
+ Copyright 2012 David Hall
+
+ Licensed under the Apache License, Version 2.0 (the "License")
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+// scalastyle:on license
+
+package org.apache.spark.ml.linalg
+
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Adapted from Matrix.scala and CSCMatrix.scala in Breeze 2.1.0 to use Spark matrix storage,
+ * preserving the original layout and truncation rules.
+ */
+private[linalg] object MatrixFormatter {
+
+ def format(matrix: Matrix): String = {
+ // Breeze uses a fixed 80 x 80 terminal and reserves three lines for sparse matrices.
+ val maxLines = if (matrix.isInstanceOf[SparseMatrix]) 77 else 80
+ format(matrix, maxLines, 80)
+ }
+
+ def format(matrix: Matrix, maxLines: Int, maxLineWidth: Int): String = matrix match {
+ case sparse: SparseMatrix => formatSparse(sparse, maxLines)
+ case dense: DenseMatrix => formatDense(dense, maxLines, maxLineWidth)
+ }
+
+ private def formatSparse(matrix: SparseMatrix, maxLines: Int): String = {
+ // The previous Breeze transpose also reordered entries and removed explicit zeros.
+ val columns = if (matrix.isTransposed) matrix.toSparseColMajor else matrix
+ val result = new StringBuilder("%d x %d CSCMatrix".format(matrix.numRows, matrix.numCols))
+ val limit = math.min(columns.values.length, maxLines - 1)
Review Comment:
**[Medium] `formatSparse` walks `values` by position instead of by column range.**
Breeze's `CSCMatrix.toString`, which the old `asBreeze.toString` used, prints `activeIterator.take(maxLines - 1)`, and `activeIterator` walks the `colPtrs(c) until colPtrs(c + 1)` ranges. This loop instead prints `values(0 until limit)` and derives each column from the pointers. That matches only when `colPtrs(0) == 0` and `colPtrs` is monotone, and the constructor checks neither.
For example:
- `new SparseMatrix(3, 0, Array(2), Array(0, 1), Array(1.0, 2.0)).toString` now throws `ArrayIndexOutOfBoundsException` at L48 (`columns.colPtrs(col + 1)`); the old output was `3 x 0 CSCMatrix`. `df.show()`, `CAST(m AS STRING)` and `HiveResult` reach this through `UserDefinedType.stringifyValue`, which `MatrixUDT` does not override, so a single such row fails the whole `show()`.
- `new SparseMatrix(1, 2, Array(1, 2, 3), Array(0, 0, 0), Array(9.0, 1.0, 2.0)).toString` now starts with an extra `(0,0) 9.0` line, for an entry that `apply` and `toArray` never see.
- `new SparseMatrix(1, 3, Array(0, 1, 0, 1), Array(0), Array(5.0)).toString` now omits `(0,2) 5.0`, although `apply(0, 2)` returns 5.0.
This PR already adds `Matrices.sparseEntries` (`Matrices.scala` L1428-1436), which walks the ranges exactly like `activeIterator`. Could `formatSparse` reuse it after widening it to `private[linalg]`, e.g. `Matrices.sparseEntries(matrix).take(maxLines - 1).foreach { case (r, c, v) => ... }`? For valid input the output stays the same, including `maxLines <= 0` and the `Int.MinValue - 1` wrap. The PR would also keep only one copy of the Breeze CSC traversal.
##########
NOTICE:
##########
@@ -1,6 +1,8 @@
Apache Spark
Copyright 2014 and onwards The Apache Software Foundation.
+Breeze is distributed under an Apache License V2.0 (See LICENSE)
Review Comment:
**[Low, licensing] This line splits the header block that must be at the top of NOTICE.**
The ASF source header policy (https://www.apache.org/legal/src-headers.html) says: "The top of each NOTICE file must include the following text". That text is the product name and copyright line, followed by "This product includes software developed at The Apache Software Foundation". Required third-party notices go after that block. That is where the existing Metrics and JSR-166 entries are at the end of this file, and where `NOTICE-binary` L1052 has this same Breeze sentence.
The sentence itself is right, since it is the first line of Breeze's own NOTICE at `releases/v2.1.0`. Could it move below the ASF block, e.g. next to the other third-party entries at the end? Also, the end-of-file newline changes in `NOTICE` and `NOTICE-binary` look unrelated to this PR.
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/Matrices.scala:
##########
@@ -755,14 +779,45 @@ class SparseMatrix @Since("2.0.0") (
* order.
*/
private[ml] override def toSparseMatrix(colMajor: Boolean): SparseMatrix = {
- if (isColMajor && !colMajor) {
- // it is col major and we want row major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]].t
- Matrices.fromBreeze(breezeTransposed).transpose.asInstanceOf[SparseMatrix]
- } else if (isRowMajor && colMajor) {
- // it is row major and we want col major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]]
- Matrices.fromBreeze(breezeTransposed).asInstanceOf[SparseMatrix]
+ if (isColMajor != colMajor) {
+ val numPtrs = if (colMajor) numCols else numRows
+ val newColPtrs = new Array[Int](numPtrs + 1)
+ var i = 0
+ while (i < values.length) {
Review Comment:
**[Medium] The count pass and the fill pass visit different entries.**
This count loop walks all of `values` (L786-791), but the fill loop below walks only the `colPtrs(j) until colPtrs(j + 1)` ranges (L804-818). The `SparseMatrix` constructor only checks the array lengths and `values.length == colPtrs.last`, so it accepts `colPtrs(0) > 0` and non-monotone `colPtrs`. A zero-copy view of the trailing columns, `colPtrs.slice(n - k, n + 1)` over the shared `rowIndices` and `values`, is one example. `apply`, `foreachActive`, `colIter` and BLAS all honor the ranges for such a matrix, and so did the old Breeze transpose (`canTranspose_CSC`). Here, an entry outside the ranges is counted but never written, so its slot keeps the default `(index 0, 0.0)`.
For example, `a = new SparseMatrix(1, 2, Array(1, 2, 3), Array(0, 0, 0), Array(9.0, 1.0, 2.0))` has `a.toArray == Array(1.0, 2.0)`.
- The new `a.toSparseRowMajor` has rowPtrs `[0, 3]`, colIndices `[0, 1, 0]` and values `[1.0, 2.0, 0.0]`. Its `toArray` is `[0.0, 2.0]`, because the phantom `(0, 0) = 0.0` overwrites `1.0`. The old result was `[0, 2]` / `[0, 1]` / `[1.0, 2.0]`.
- For `new SparseMatrix(2, 1, Array(1, 3), Array(0, 0, 1), Array(1.0, 2.0, 3.0))`, `toSparseRowMajor.toDense` is `[0.0, 3.0]` instead of `[2.0, 3.0]`, and the result's `colIter` throws `IllegalArgumentException` on the first `next()`.
- With non-monotone `colPtrs`, `new SparseMatrix(1, 3, Array(0, 1, 0, 1), Array(0), Array(5.0)).toSparseRowMajor` throws `ArrayIndexOutOfBoundsException` at L811, where the old code returned a CSR matrix.
Transposed matrices now go through this conversion in `toString` (`MatrixFormatter.scala` L42), `hashCode` (L671) and `equals` (L1430), so they inherit it. For example, `a.transpose.toString` prints an extra `(0,0) 0.0` line, and with non-monotone `colPtrs` all three throw.
Could the count be done inside the same `colPtrs` range loop as the fill, so that both passes see exactly the same entries?
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/Matrices.scala:
##########
@@ -755,14 +779,45 @@ class SparseMatrix @Since("2.0.0") (
* order.
*/
private[ml] override def toSparseMatrix(colMajor: Boolean): SparseMatrix = {
- if (isColMajor && !colMajor) {
- // it is col major and we want row major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]].t
- Matrices.fromBreeze(breezeTransposed).transpose.asInstanceOf[SparseMatrix]
- } else if (isRowMajor && colMajor) {
- // it is row major and we want col major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]]
- Matrices.fromBreeze(breezeTransposed).asInstanceOf[SparseMatrix]
+ if (isColMajor != colMajor) {
+ val numPtrs = if (colMajor) numCols else numRows
+ val newColPtrs = new Array[Int](numPtrs + 1)
Review Comment:
**[Low] Negative dimensions now behave differently.**
The constructors accept negative dimensions. `DenseMatrix` only checks `values.length == numRows * numCols`, and a transposed `SparseMatrix` does not check `numCols`. Breeze's constructors rejected them, and its transpose builder inferred the size instead. As a result, the old code threw or returned a resized result. The new code is silently lenient for dense matrices and throws here for sparse ones.
For example:
- For `new DenseMatrix(-2, -3, new Array[Double](6))` (or `DenseMatrix.zeros(-2, -3)`), `toString` and `equals` used to throw `IndexOutOfBoundsException: Rows must be larger than zero. It was -2`. Now `toString` returns `""`. `equals` compares no cells, so even `new DenseMatrix(-1, -1, Array(1.0)) == new DenseMatrix(-1, -1, Array(2.0))` is true.
- For `SparseMatrix.fromCOO(-3, 1, Seq.empty).transpose`, `toString`, `hashCode` and `equals` used to return `1 x 1 CSCMatrix`, 43 and true. Now all three throw `NegativeArraySizeException` from `new Array[Int](numPtrs + 1)` here.
This only affects degenerate input, so a note may be enough. A `require(numRows >= 0 && numCols >= 0, ...)` in the constructors would make both cases fail early and consistently, though that could be a separate JIRA.
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/MatrixFormatter.scala:
##########
@@ -0,0 +1,115 @@
+// scalastyle:off license
+/*
+ Copyright 2012 David Hall
+
+ Licensed under the Apache License, Version 2.0 (the "License")
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+// scalastyle:on license
+
+package org.apache.spark.ml.linalg
+
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Adapted from Matrix.scala and CSCMatrix.scala in Breeze 2.1.0 to use Spark matrix storage,
+ * preserving the original layout and truncation rules.
+ */
+private[linalg] object MatrixFormatter {
+
+ def format(matrix: Matrix): String = {
+ // Breeze uses a fixed 80 x 80 terminal and reserves three lines for sparse matrices.
+ val maxLines = if (matrix.isInstanceOf[SparseMatrix]) 77 else 80
+ format(matrix, maxLines, 80)
+ }
+
+ def format(matrix: Matrix, maxLines: Int, maxLineWidth: Int): String = matrix match {
+ case sparse: SparseMatrix => formatSparse(sparse, maxLines)
+ case dense: DenseMatrix => formatDense(dense, maxLines, maxLineWidth)
+ }
+
+ private def formatSparse(matrix: SparseMatrix, maxLines: Int): String = {
+ // The previous Breeze transpose also reordered entries and removed explicit zeros.
+ val columns = if (matrix.isTransposed) matrix.toSparseColMajor else matrix
+ val result = new StringBuilder("%d x %d CSCMatrix".format(matrix.numRows, matrix.numCols))
+ val limit = math.min(columns.values.length, maxLines - 1)
+ var col = 0
+ var index = 0
+ while (index < limit) {
+ while (index >= columns.colPtrs(col + 1)) {
+ col += 1
+ }
+ result.append('\n').append("(%d,%d) ".format(columns.rowIndices(index), col))
+ .append(columns.values(index))
+ index += 1
+ }
+ result.toString()
+ }
+
+ private def formatDense(matrix: DenseMatrix, maxLines: Int, maxLineWidth: Int): String = {
+ val rows = matrix.numRows
+ val cols = matrix.numCols
+ val showRows = if (rows > maxLines) maxLines - 1 else rows
+ val widths = new ArrayBuffer[Int]
+ var totalWidth = 0
+ var col = 0
+ while (col < cols && totalWidth < maxLineWidth) {
+ var width = 0
+ var row = 0
+ while (row < showRows) {
+ width = math.max(width, matrix(row, col).toString.length + 2)
Review Comment:
**[Nit] `toString(Int.MinValue, width)` now throws a different exception.**
With `maxLines = Int.MinValue`, `showRows` wraps to `Int.MaxValue` at L61, exactly as in Breeze, so this loop reads past the last row. Breeze's `DenseMatrix.apply` threw `IndexOutOfBoundsException`. Spark's `DenseMatrix.apply` throws `IllegalArgumentException` from the `require` in `index`.
For example, `Matrices.rand(5, 10, rng).toString(Int.MinValue, 80)` used to throw `IndexOutOfBoundsException: (5,0) not in [-5,5) x [-10,10)`. It now throws `IllegalArgumentException: requirement failed: Expected 0 <= i < 5, got i = 5.`
The call already failed before, so this only affects the exact-parity claim. The existing "toString" test calls `toString(Int.MinValue, Int.MinValue)`, which returns `5 x 10 matrix` before it reads any cell, so this path is not covered. Feel free to ignore it if exception parity is not a goal.
##########
mllib-local/src/test/scala/org/apache/spark/ml/linalg/MatricesSuite.scala:
##########
@@ -870,6 +1009,57 @@ class MatricesSuite extends SparkMLFunSuite {
assert(lines.length == 5 && lines.forall(_.length <= 100))
}
+ test("matrix formatting preserves Breeze layout and truncation") {
+ val dense = new DenseMatrix(3, 4,
+ Array(0.0, -0.0, 1.0, 12.0, -123.0, 1e100, Double.NaN,
+ Double.PositiveInfinity, Double.NegativeInfinity, 0.01, 0.0, 3.0))
+ val sparse = new SparseMatrix(3, 4, Array(0, 2, 2, 4, 5),
+ Array(0, 2, 0, 1, 2), Array(-0.0, 1.0, Double.NaN, 0.0, Double.PositiveInfinity))
+ val tall = new DenseMatrix(85, 1, Array.tabulate(85)(_.toDouble))
+ val wide = DenseMatrix.ones(2, 30)
+ val matrices = Seq(dense, sparse, dense.transpose, sparse.transpose,
+ tall, tall.toSparse, wide, DenseMatrix.zeros(0, 0), DenseMatrix.zeros(0, 3),
+ DenseMatrix.zeros(3, 0), SparseMatrix.fromCOO(0, 3, Seq.empty),
+ SparseMatrix.fromCOO(3, 0, Seq.empty))
+ val limits = Seq((-1, -5), (0, 0), (1, 1), (2, 15), (3, 25), (5, 50),
+ (80, 80), (Int.MaxValue, Int.MaxValue))
+ matrices.foreach { matrix =>
+ assert(matrix.toString == matrix.asBreeze.toString())
+ limits.foreach { case (lines, width) =>
+ withClue(s"${matrix.numRows} x ${matrix.numCols}, limits ($lines, $width): ") {
+ assert(matrix.toString(lines, width) == matrix.asBreeze.toString(lines, width))
+ }
+ }
+ }
+ val expectedWideRow = "1.0 " * 13
+ assert(wide.toString == s"$expectedWideRow... (30 total)\n$expectedWideRow...")
+ assert(new DenseMatrix(2, 2, Array(1.0, 20.0, 3.0, 4.0)).toString(10, 80) ==
+ "1.0 3.0 \n20.0 4.0 ")
+ assert(new SparseMatrix(2, 2, Array(0, 1, 2), Array(1, 0), Array(2.0, 3.0))
+ .toString(2, 1) == "2 x 2 CSCMatrix\n(1,0) 2.0")
+ }
+
+ test("matrix formatting preserves the default formatting locale") {
+ val originalLocale = Locale.getDefault(Locale.Category.FORMAT)
+ try {
+ Locale.setDefault(Locale.Category.FORMAT, Locale.forLanguageTag("ar-EG"))
+ val dense = new DenseMatrix(2, 2, Array(1.0, 2.0, 3.0, 4.0))
+ val sparse = dense.toSparse
+ assert(dense.toString(1, 1) == dense.asBreeze.toString(1, 1))
Review Comment:
**[Low, tests] The dense assertion of this locale test never reaches the locale-sensitive code.**
For this 2 x 2 matrix, `toString(1, 1)` sets `showRows = 0`, so every column width is 0 and all columns fit. Both implementations return `"\n... (2 total)"`, which is built by interpolation and appends. The only locale-sensitive dense path, `"%d x %d matrix".format(rows, cols)` at `MatrixFormatter.scala` L82, never runs. If L82 were changed to `s"$rows x $cols matrix"`, this test would still pass. So would every other test, because CI uses a default locale with Latin digits.
`dense.toString(5, 1)` (or `toString(2, 5)`) reaches L82 and returns `2 x 2 matrix` with Arabic-Indic digits under `ar-EG`, in Breeze and here. Could the dense assertion use one of those limits? The two sparse assertions do go through `.format`.
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/Vectors.scala:
##########
@@ -104,7 +104,10 @@ sealed trait Vector extends Serializable {
* @param i index
*/
@Since("2.0.0")
- def apply(i: Int): Double = asBreeze(i)
+ def apply(i: Int): Double = this match {
Review Comment:
**[Low, cleanup] This trait body cannot run, and it would recurse if an override were removed.**
`Vector` is sealed, and both `DenseVector` (L539) and `SparseVector` (L680) override `apply`, so this body is never executed. The old `asBreeze(i)` body was never executed either. The same holds for the mllib `Vector` (mllib `Vectors.scala` L115-118, with overrides at L681 and L833). So "Dispatch vector element access to Spark's dense and sparse implementations" in the PR description is not a runtime change.
Each case calls the same virtual `apply` again. Suppose a later refactor removes one of the overrides, e.g. while removing Breeze. Then this still compiles without a warning, and `v(i)` throws `StackOverflowError`. By contrast, the old body fell back to Breeze, and an abstract method would fail to compile.
Could the body avoid re-dispatching? For example, it could throw `UnsupportedOperationException` like `copy` in the same trait, or keep `asBreeze(i)` for now. Making `apply` abstract would also work, but it needs a MiMa exclusion (`DirectMissingMethodProblem` for `Vector.apply` in both packages).
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/Matrices.scala:
##########
@@ -755,14 +779,45 @@ class SparseMatrix @Since("2.0.0") (
* order.
*/
private[ml] override def toSparseMatrix(colMajor: Boolean): SparseMatrix = {
- if (isColMajor && !colMajor) {
- // it is col major and we want row major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]].t
- Matrices.fromBreeze(breezeTransposed).transpose.asInstanceOf[SparseMatrix]
- } else if (isRowMajor && colMajor) {
- // it is row major and we want col major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]]
- Matrices.fromBreeze(breezeTransposed).asInstanceOf[SparseMatrix]
+ if (isColMajor != colMajor) {
+ val numPtrs = if (colMajor) numCols else numRows
+ val newColPtrs = new Array[Int](numPtrs + 1)
+ var i = 0
+ while (i < values.length) {
+ if (values(i) != 0.0) {
+ newColPtrs(rowIndices(i) + 1) += 1
+ }
+ i += 1
+ }
+ i = 0
+ while (i < numPtrs) {
+ newColPtrs(i + 1) += newColPtrs(i)
+ i += 1
+ }
+
+ val nnz = newColPtrs(numPtrs)
+ val newRowIndices = new Array[Int](nnz)
+ val newValues = new Array[Double](nnz)
+ val next = newColPtrs.clone()
Review Comment:
**[Low, performance] `next` doubles the pointer-array allocation for wide, very sparse matrices.**
This clone allocates a second `Int` array of `numPtrs + 1` elements. The old Breeze transpose allocated one array of that size (`outCols` in `Builder.result`) plus O(nnz) arrays. So when `numPtrs` is much larger than nnz, this conversion allocates about twice as much, and so do the transposed `toString`, `hashCode` and `equals` that use it. For example:
- The new 100000 x 100000 test matrix with 2 entries: about 400 KB before, 800 KB now.
- A 3 x 50,000,000 CSR matrix with 2 entries: about 200 MB before, 400 MB now.
The ratio approaches 2 once `numPtrs` is above about 1000 x nnz. Below about 70 x nnz, this code already allocates less than Breeze. The realistic case is a wide CSR coefficient matrix from `compressed`, e.g. a multinomial `LogisticRegression` model trained with L1.
Could the cursor reuse `newColPtrs` without a copy, similar to scipy's `csr_tocsc`? Keep the inclusive prefix sums, use `newColPtrs(newCol)` as the write cursor and increment it. Then shift the array right by one and set `newColPtrs(0) = 0`. I checked that this produces the same arrays as Breeze's transpose on 30,000 random valid matrices.
##########
mllib-local/src/test/scala/org/apache/spark/ml/linalg/MatricesSuite.scala:
##########
@@ -105,6 +105,70 @@ class MatricesSuite extends SparkMLFunSuite {
assert(sm1 === dm2.transpose)
}
+ test("equals preserves Breeze floating-point and storage semantics") {
+ val values = Seq(0.0, -0.0, 1.0, Double.NaN,
+ Double.PositiveInfinity, Double.NegativeInfinity)
+ val matrices = values.flatMap { value =>
+ val dense = new DenseMatrix(2, 2, Array(0.0, value, 2.0, 0.0))
+ val sparse = new SparseMatrix(2, 2, Array(0, 1, 2), Array(1, 0), Array(value, 2.0))
+ Seq(dense, dense.copy, sparse, sparse.copy, dense.transpose, sparse.transpose)
+ } ++ Seq(
+ new DenseMatrix(0, 2, Array.emptyDoubleArray),
+ new DenseMatrix(2, 0, Array.emptyDoubleArray),
+ new SparseMatrix(0, 2, Array(0, 0, 0), Array.emptyIntArray, Array.emptyDoubleArray),
+ new SparseMatrix(2, 0, Array(0), Array.emptyIntArray, Array.emptyDoubleArray))
+
+ for (left <- matrices; right <- matrices) {
+ assert(left.equals(right) === left.asBreeze.equals(right.asBreeze))
+ }
+ matrices.foreach { matrix =>
+ assert(!matrix.equals(null))
+ assert(!matrix.equals(matrix.toArray))
+ }
+ }
+
+ test("sparse equality preserves explicit-zero compatibility") {
+ val empty = new SparseMatrix(3, 1, Array(0, 0), Array.emptyIntArray, Array.emptyDoubleArray)
+ val zeroAtFirst = new SparseMatrix(3, 1, Array(0, 1), Array(0), Array(0.0))
+ val zeroAtLast = new SparseMatrix(3, 1, Array(0, 1), Array(2), Array(-0.0))
+ assert(zeroAtFirst.equals(empty))
+ assert(zeroAtLast.equals(empty))
+ assert(!zeroAtFirst.equals(zeroAtLast))
Review Comment:
**[Low, tests] This pins a non-transitive equality as Spark behavior.**
`zeroAtFirst == empty` and `zeroAtLast == empty` hold (L134-135), but `zeroAtFirst != zeroAtLast` here, so `equals` is not transitive.
- The same two matrices are equal after `transpose`, `toSparse` or `toDense`.
- Their single `colIter` columns are equal as vectors, because `Vectors.equals` skips zeros.
- They have the same `hashCode` (43).
Valid matrices get such trailing explicit zeros too, e.g. after `update(i, j, 0.0)` or `map`. So the result of `distinct` or a `HashSet` can depend on where explicit zeros are stored.
I understand that the PR keeps Breeze's behavior on purpose, as the comment at `Matrices.scala` L1393-1394 says, and that changing it would change behavior. However, once these assertions (here and L144) pin it, a later fix has to invert them. Could the test and the comment call it a known Breeze quirk kept for compatibility, rather than the intended semantics? A separate JIRA could then make sparse equality skip zeros like `Vectors.equals`.
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/MatrixFormatter.scala:
##########
@@ -0,0 +1,115 @@
+// scalastyle:off license
+/*
+ Copyright 2012 David Hall
+
+ Licensed under the Apache License, Version 2.0 (the "License")
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+// scalastyle:on license
+
+package org.apache.spark.ml.linalg
+
+import scala.collection.mutable.ArrayBuffer
+
+/**
+ * Adapted from Matrix.scala and CSCMatrix.scala in Breeze 2.1.0 to use Spark matrix storage,
+ * preserving the original layout and truncation rules.
+ */
+private[linalg] object MatrixFormatter {
+
+ def format(matrix: Matrix): String = {
+ // Breeze uses a fixed 80 x 80 terminal and reserves three lines for sparse matrices.
+ val maxLines = if (matrix.isInstanceOf[SparseMatrix]) 77 else 80
+ format(matrix, maxLines, 80)
+ }
+
+ def format(matrix: Matrix, maxLines: Int, maxLineWidth: Int): String = matrix match {
+ case sparse: SparseMatrix => formatSparse(sparse, maxLines)
+ case dense: DenseMatrix => formatDense(dense, maxLines, maxLineWidth)
+ }
+
+ private def formatSparse(matrix: SparseMatrix, maxLines: Int): String = {
+ // The previous Breeze transpose also reordered entries and removed explicit zeros.
+ val columns = if (matrix.isTransposed) matrix.toSparseColMajor else matrix
Review Comment:
**[Low, performance] A transposed matrix is fully converted on every `toString`, `hashCode` and `equals` call.**
For a CSR matrix, `toString` converts the whole matrix to CSC here, in O(nnz + numRows + numCols) time with four new arrays, only to print at most `maxLines - 1` entries (76 by default).
- `SparseMatrix.hashCode` (`Matrices.scala` L671) builds a full `SparseMatrix` only to read `values` in column-major order.
- `equals` (L1430) converts each transposed operand, then allocates an iterator per column and a boxed `Tuple3` per entry in `sparseEntries`.
A realistic case is printing a wide L1 multinomial `LogisticRegression` coefficient matrix, which `compressed` stores as CSR, or calling `df.show()` on it. The bundled examples print `coefficientMatrix`.
Breeze also transposed everything on each call, so this is a missed improvement rather than a regression. Two smaller steps could still help:
- A single helper for the "convert if transposed" view. It is now repeated in three places with three different comments: here, L671 and L1430.
- An index-based loop instead of the `Iterator.range(...).flatMap(...).map(...)` chain in `sparseEntries`.
##########
LICENSE:
##########
@@ -217,6 +217,8 @@ core/src/main/resources/org/apache/spark/ui/static/vis*
connector/spark-ganglia-lgpl/src/main/java/com/codahale/metrics/ganglia/GangliaReporter.java
core/src/main/resources/org/apache/spark/ui/static/d3-flamegraph.min.js
core/src/main/resources/org/apache/spark/ui/static/d3-flamegraph.css
+mllib-local/src/main/scala/org/apache/spark/ml/linalg/Matrices.scala (portions)
Review Comment:
**[Low, licensing] `Matrices.scala (portions)` introduces a new, hybrid attribution model.**
This is the only partial-file entry in `LICENSE` and `LICENSE-binary`. It also makes `Matrices.scala` the only source file that carries both the ASF header and a third-party copyright block (L18-25).
Spark's existing conventions are different:
- Within the same PR, `MatrixFormatter.scala` follows the usual per-file convention, the one HyukjinKwon's earlier review pointed to: the upstream header plus a `LICENSE` entry, like `TimSort.java` and `scala/collection/compat/package.scala`.
- For small snippets inside ASF files, Spark otherwise uses only an inline "based on" comment, e.g. `Murmur3_x86_32.java` L25 and `XXH64.java` L29.
The adapted lines (L671-681 and L1393-1411) do have inline markers. However, it is unclear whether the dense `case _` branch (L1412-1424, which follows Breeze's `Matrix.equals`) is covered. When this code is rewritten later, the header block and the two `LICENSE` lines can easily go stale.
Could `Matrices.equals`, `sparseEntries` and the sparse hash loop move into a Breeze-headed `private[linalg]` object like `MatrixFormatter`? Then the attribution stays per file, and `DenseMatrix.equals`, `SparseMatrix.equals` and `SparseMatrix.hashCode` become one-line delegations.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60254][MLLIB] Replace Breeze delegation in vector and matrix primitives [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #59485:
URL: https://github.com/apache/spark/pull/59485#discussion_r4286389631
##########
mllib-local/src/test/scala/org/apache/spark/ml/linalg/MatricesSuite.scala:
##########
@@ -870,6 +1009,54 @@ class MatricesSuite extends SparkMLFunSuite {
assert(lines.length == 5 && lines.forall(_.length <= 100))
}
+ test("matrix formatting preserves Breeze layout and truncation") {
+ val dense = new DenseMatrix(3, 4,
+ Array(0.0, -0.0, 1.0, 12.0, -123.0, 1e100, Double.NaN,
+ Double.PositiveInfinity, Double.NegativeInfinity, 0.01, 0.0, 3.0))
+ val sparse = new SparseMatrix(3, 4, Array(0, 2, 2, 4, 5),
+ Array(0, 2, 0, 1, 2), Array(-0.0, 1.0, Double.NaN, 0.0, Double.PositiveInfinity))
+ val tall = new DenseMatrix(85, 1, Array.tabulate(85)(_.toDouble))
+ val matrices = Seq(dense, sparse, dense.transpose, sparse.transpose,
Review Comment:
**Nit (P3):** Every dense matrix in this default-`toString` parity list renders far below the 80-character width. The widest row is about 32 characters (the 3x4 fixture), and the mllib and locale tests add no wide dense case either. So the Spark-owned default in `MatrixFormatter.format(matrix)` (`format(matrix, maxLines, 80)`) is never exercised. If that `80` changed, every assertion here would still pass while the default rendering of wide matrices changed. Adding a wide dense fixture such as `DenseMatrix.ones(2, 30)` to this list would pin the default column truncation and the `... (30 total)` marker.
**Verification:**
- **Regression:** Default toString of a dense matrix wider than 80 characters matches the Breeze rendering, including column truncation and the '... (N total)' marker, and would differ if the default width were changed.
- **Behavior:** The existing mllib-local matrix formatting, equality, hashing, and sparse layout conversion parity tests still pass.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-60254][MLLIB] Replace Breeze delegation in vector and matrix primitives [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #59485:
URL: https://github.com/apache/spark/pull/59485#discussion_r4354110360
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/Matrices.scala:
##########
@@ -755,14 +750,90 @@ class SparseMatrix @Since("2.0.0") (
* order.
*/
private[ml] override def toSparseMatrix(colMajor: Boolean): SparseMatrix = {
- if (isColMajor && !colMajor) {
- // it is col major and we want row major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]].t
- Matrices.fromBreeze(breezeTransposed).transpose.asInstanceOf[SparseMatrix]
- } else if (isRowMajor && colMajor) {
- // it is row major and we want col major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]]
- Matrices.fromBreeze(breezeTransposed).asInstanceOf[SparseMatrix]
+ MatrixUtils.checkDimensions(this)
+ if (isColMajor != colMajor) {
+ val numPtrs = if (colMajor) numCols else numRows
+ val newColPtrs = new Array[Int](numPtrs + 1)
+ var j = 0
+ // Count the same pointer ranges used below, including shared or skipped backing entries.
+ while (j < colPtrs.length - 1) {
+ var idx = colPtrs(j)
+ val idxEnd = colPtrs(j + 1)
+ while (idx < idxEnd) {
+ if (values(idx) != 0.0) {
+ val newCol = rowIndices(idx)
+ require(newCol >= 0 && newCol < numPtrs,
Review Comment:
**[Low, compatibility] This `require` changes the default `serialVersionUID` of `SparseMatrix`.**
`SparseMatrix` is `Serializable` through `Matrix` and has no `@SerialVersionUID`. Scala 2.13 compiles the by-name message of `require` into a lambda whose body becomes a new `public static final` method of the class, here `$anonfun$toSparseMatrix$1(int, int)`. `ObjectStreamClass` includes every non-private method when it computes the default `serialVersionUID`, so this line changes it. Spark's build does not enable the Scala inliner, so `require` is not inlined away. The existing messages in the constructor and in `index` already compile to such methods (`$anonfun$new$2` to `$anonfun$index$4` in `javap -p`), so the value only stays stable while no new one is added.
`serialver` gives `3621969306166067103` for `org.apache.spark.ml.linalg.SparseMatrix` in the released 4.0.0 jar, and the value is the same through 4.2.0 and on today's master. So, for example, an `RDD[SparseMatrix]` written by 4.2.0 with `saveAsObjectFile` and read with `sc.objectFile[SparseMatrix]` on a build with this PR would fail with `java.io.InvalidClassException: org.apache.spark.ml.linalg.SparseMatrix; local class incompatible: stream classdesc serialVersionUID = 3621969306166067103, ...`, and the same happens in the other direction. The other classes this PR touches keep their values, because only method bodies change there.
Spark does not promise cross-version Java serialization, and this is the `require` I suggested in [round 1](https://github.com/apache/spark/pull/59485#discussion_r4297407281), so I missed this side effect. Could it be an explicit check instead? That keeps the value and also avoids allocating a `Function0` per checked entry:
```scala
if (newCol < 0 || newCol >= numPtrs) {
throw new IllegalArgumentException(
s"Expected 0 <= inner index < $numPtrs, got $newCol.")
}
```
`require(cond, "literal")` would not help, because the literal is still passed by name. The test at L560 uses `contains`, so it still passes without the `requirement failed: ` prefix.
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/Matrices.scala:
##########
@@ -755,14 +750,90 @@ class SparseMatrix @Since("2.0.0") (
* order.
*/
private[ml] override def toSparseMatrix(colMajor: Boolean): SparseMatrix = {
- if (isColMajor && !colMajor) {
- // it is col major and we want row major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]].t
- Matrices.fromBreeze(breezeTransposed).transpose.asInstanceOf[SparseMatrix]
- } else if (isRowMajor && colMajor) {
- // it is row major and we want col major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]]
- Matrices.fromBreeze(breezeTransposed).asInstanceOf[SparseMatrix]
+ MatrixUtils.checkDimensions(this)
+ if (isColMajor != colMajor) {
+ val numPtrs = if (colMajor) numCols else numRows
+ val newColPtrs = new Array[Int](numPtrs + 1)
+ var j = 0
+ // Count the same pointer ranges used below, including shared or skipped backing entries.
+ while (j < colPtrs.length - 1) {
+ var idx = colPtrs(j)
+ val idxEnd = colPtrs(j + 1)
+ while (idx < idxEnd) {
+ if (values(idx) != 0.0) {
+ val newCol = rowIndices(idx)
+ require(newCol >= 0 && newCol < numPtrs,
+ s"Expected 0 <= inner index < $numPtrs, got $newCol.")
+ newColPtrs(newCol + 1) += 1
+ }
+ idx += 1
+ }
+ j += 1
+ }
+ var i = 0
+ while (i < numPtrs) {
+ newColPtrs(i + 1) += newColPtrs(i)
+ i += 1
+ }
+
+ val nnz = newColPtrs(numPtrs)
+ val newRowIndices = new Array[Int](nnz)
+ val newValues = new Array[Double](nnz)
+ j = 0
+ // Visiting the original rows or columns in order keeps the new indices sorted.
+ while (j < colPtrs.length - 1) {
+ var idx = colPtrs(j)
+ val idxEnd = colPtrs(j + 1)
+ while (idx < idxEnd) {
+ if (values(idx) != 0.0) {
+ val newCol = rowIndices(idx)
+ val newIdx = newColPtrs(newCol)
+ newRowIndices(newIdx) = j
+ newValues(newIdx) = values(idx)
+ newColPtrs(newCol) += 1
+ }
+ idx += 1
+ }
+ j += 1
+ }
+ // Restore the starts after using the pointers as insertion cursors.
+ i = numPtrs
+ while (i > 0) {
+ newColPtrs(i) = newColPtrs(i - 1)
+ i -= 1
+ }
+ newColPtrs(0) = 0
+
+ // Sum duplicate coordinates in stored order, retaining sums that become zero.
Review Comment:
**[Low, performance] For a wide row-major matrix, this conversion is now several times slower than master's Breeze transpose.**
With this round, the conversion makes three read-modify-write passes over the `numPtrs + 1` pointers: the prefix sum (L773-777), the restore shift (L799-805, which I suggested in [round 1](https://github.com/apache/spark/pull/59485#discussion_r4297407329)), and this duplicate merge (L807-828), which runs on every call. Breeze's `CSCMatrix.Builder.result` allocates the zeroed pointer array and writes each pointer once. So when `numCols` is much larger than `nnz`, the pointer passes dominate. Transposed `hashCode`, `toString` and `equals` all go through this conversion via `MatrixUtils.columnMajor`.
I measured an independent model of L752-836 (not the PR code itself) against Breeze 2.1.0 on master's `asBreeze` path. The model reproduces Breeze's arrays, hash and string. JDK 21, medians of 41 runs:
| CSR shape | nnz | Operation | master (Breeze) | this PR (model) |
|---|---|---|---|---|
| 3 x 2^24 | 3000 | `hashCode` | 1.56 ms | 16.8-19.4 ms |
| 3 x 2^24 | 3000 | `toString` | 4.3-4.5 ms | 16.9-19.5 ms |
| 3 x 2^18 | 1000 | `hashCode` | 0.05 ms | 0.34 ms |
| 3 x 2^18 | 0 | `hashCode` | 0.006 ms | 0.22 ms |
| 3 x 2^18 | 10000 | `hashCode` | 0.48 ms | 0.42 ms |
2^18 is the `HashingTF` default width, and the all-zero case is the constant-label `LogisticRegression` coefficient matrix (`LogisticRegression.scala` L585-587). The crossover is around `numCols / nnz` = 30-50. Round-1 point 12 was about converting a transposed operand at all, which Breeze did too. This one is about the constant factor of the new conversion, and the merge pass is new in this round.
Some options. Each gave identical arrays, or the same exception, on 400,000 random inputs, canonical and malformed:
- Skip the merge unless it can find something. A duplicate needs a repeated inner index within one source range, so the count loop can set a flag when `newCol <= prev`, with `prev` reset per range. This alone saved 21-58%.
- Fold the restore shift into the merge loop. After the scatter, `newColPtrs(c)` already holds the end of column `c`, so the merge can read `end = newColPtrs(j)`, write `newColPtrs(j) = newNnz` and keep the start in a local. This saved another 2-21%.
- Return early when the count pass finds no nonzero, as Breeze does.
Together with a register-carried prefix sum and a reverse scatter with decrementing cursors (no restore pass), the model's time dropped by 70-75%, to about 3x Breeze at 2^24. I found no Spark caller that runs this in a loop (`df.show()` formats each shown row once), so this is low priority.
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/MatrixUtils.scala:
##########
@@ -0,0 +1,142 @@
+// scalastyle:off license
+/*
+ Copyright 2012 David Hall
+
+ Licensed under the Apache License, Version 2.0 (the "License")
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+// scalastyle:on license
+
+package org.apache.spark.ml.linalg
+
+import scala.util.hashing.MurmurHash3
+
+/**
+ * Equality, sparse hashing, and traversal adapted from Matrix.scala, CSCMatrix.scala, and
+ * Tensor.scala in Breeze 2.1.0 to use Spark matrix storage.
+ */
+private[linalg] object MatrixUtils {
+
+ def checkDimensions(matrix: Matrix): Unit = {
+ require(matrix.numRows >= 0 && matrix.numCols >= 0,
+ s"Matrix dimensions must be nonnegative: ${matrix.numRows} x ${matrix.numCols}")
+ }
+
+ private def columnMajor(matrix: SparseMatrix): SparseMatrix = {
+ checkDimensions(matrix)
+ // Breeze's transpose normalizes the storage, unlike its ordinary CSC wrapper.
+ if (matrix.isTransposed) matrix.toSparseColMajor else matrix
Review Comment:
**[Nit, description] Could the description mention that this also applies to formatting, hashing and equality of row-major matrices?**
The user-facing section says "Layout-changing sparse conversion also rejects out-of-range nonzero inner indices". `toString`, `toString(maxLines, maxLineWidth)`, `hashCode` and `equals` of a row-major matrix normalize through this line, so they throw that `IllegalArgumentException` too.
For example, take `t = new SparseMatrix(1, 2, Array(0, 1), Array(5), Array(1.0), isTransposed = true)`. On master, `t.toString` is `1 x 2 CSCMatrix`, `t.hashCode` is `652125558` and `t == DenseMatrix.zeros(1, 2)` is true, because Breeze's builder skips the bound check for the first sorted entry. With this PR, all three throw `Expected 0 <= inner index < 2, got 5.`, so `df.show()` or `CAST(m AS STRING)` on such a row now fails.
That is the behavior I asked for in [round 1](https://github.com/apache/spark/pull/59485#discussion_r4297407281), so only the wording is open, e.g. "Layout-changing sparse conversion, and therefore formatting, hashing and equality of row-major sparse matrices, which normalize through it, also rejects ...". The index test at L553-563 only calls `toSparseMatrix`. An `intercept` around `matrix.transpose.toString` there would pin the rest.
##########
mllib-local/src/test/scala/org/apache/spark/ml/linalg/MatricesSuite.scala:
##########
@@ -105,6 +105,120 @@ class MatricesSuite extends SparkMLFunSuite {
assert(sm1 === dm2.transpose)
}
+ test("equals preserves Breeze floating-point and storage semantics") {
+ val values = Seq(0.0, -0.0, 1.0, Double.NaN,
+ Double.PositiveInfinity, Double.NegativeInfinity)
+ val matrices = values.flatMap { value =>
+ val dense = new DenseMatrix(2, 2, Array(0.0, value, 2.0, 0.0))
+ val sparse = new SparseMatrix(2, 2, Array(0, 1, 2), Array(1, 0), Array(value, 2.0))
+ Seq(dense, dense.copy, sparse, sparse.copy, dense.transpose, sparse.transpose)
+ } ++ Seq(
+ new DenseMatrix(0, 2, Array.emptyDoubleArray),
+ new DenseMatrix(2, 0, Array.emptyDoubleArray),
+ new SparseMatrix(0, 2, Array(0, 0, 0), Array.emptyIntArray, Array.emptyDoubleArray),
+ new SparseMatrix(2, 0, Array(0), Array.emptyIntArray, Array.emptyDoubleArray))
+
+ for (left <- matrices; right <- matrices) {
+ assert(left.equals(right) === left.asBreeze.equals(right.asBreeze))
+ }
+ matrices.foreach { matrix =>
+ assert(!matrix.equals(null))
+ assert(!matrix.equals(matrix.toArray))
+ }
+ }
+
+ test("sparse equality preserves Breeze's known non-transitive explicit-zero quirk") {
+ val empty = new SparseMatrix(3, 1, Array(0, 0), Array.emptyIntArray, Array.emptyDoubleArray)
+ val zeroAtFirst = new SparseMatrix(3, 1, Array(0, 1), Array(0), Array(0.0))
+ val zeroAtLast = new SparseMatrix(3, 1, Array(0, 1), Array(2), Array(-0.0))
+ assert(zeroAtFirst.equals(empty))
+ assert(zeroAtLast.equals(empty))
+ assert(!zeroAtFirst.equals(zeroAtLast))
+ assert(zeroAtFirst.transpose.equals(zeroAtLast.transpose))
+
+ val nonzero = new SparseMatrix(3, 1, Array(0, 1), Array(0), Array(2.0))
+ val middleZero = new SparseMatrix(3, 1, Array(0, 2), Array(0, 1), Array(2.0, 0.0))
+ val lastZero = new SparseMatrix(3, 1, Array(0, 2), Array(0, 2), Array(2.0, -0.0))
+ assert(middleZero.equals(nonzero))
+ assert(lastZero.equals(nonzero))
+ assert(!middleZero.equals(lastZero))
+ assert(middleZero.transpose.equals(lastZero.transpose))
+ }
+
+ test("dense equality normalizes unsorted row-major sparse indices") {
+ val dense = new DenseMatrix(2, 3, Array(1.0, 0.0, 0.0, 5.0, 3.0, 6.0))
+ val sparse = new SparseMatrix(2, 3, Array(0, 2, 4), Array(2, 0, 2, 1),
+ Array(3.0, 1.0, 6.0, 5.0), isTransposed = true)
+ assert(dense.equals(sparse))
+ assert(sparse.equals(dense))
+ val different = dense.copy
+ different.values(0) = 2.0
+ assert(!different.equals(sparse))
+ assert(!sparse.equals(different))
+ }
+
+ test("row-major sparse duplicates are normalized for equality, hashing and formatting") {
+ val matrix = new SparseMatrix(1, 2, Array(0, 2), Array(1, 1), Array(1.0, 2.0),
+ isTransposed = true)
+ val dense = new DenseMatrix(1, 2, Array(0.0, 3.0))
+ val sparse = SparseMatrix.fromCOO(1, 2, Seq((0, 1, 3.0)))
+ assert(matrix.equals(dense))
+ assert(dense.equals(matrix))
+ assert(matrix.equals(sparse))
+ assert(sparse.equals(matrix))
+ assert(matrix.hashCode() === -942016480)
+ assert(matrix.toString == "1 x 2 CSCMatrix\n(0,1) 3.0")
+ }
+
+ test("matrix operations reject invalid negative dimensions") {
+ val dense = new DenseMatrix(-1, -1, Array(1.0))
+ val sparse = new SparseMatrix(-1, 1, Array(0, 0),
+ Array.emptyIntArray, Array.emptyDoubleArray)
+ Seq(dense, sparse, sparse.transpose).foreach { matrix =>
+ intercept[IllegalArgumentException] {
+ matrix.equals(matrix.copy)
+ }
+ intercept[IllegalArgumentException] {
+ matrix.toString
+ }
+ }
+ Seq(sparse, sparse.transpose).foreach { matrix =>
+ intercept[IllegalArgumentException] {
+ matrix.hashCode()
+ }
+ Seq(true, false).foreach { colMajor =>
+ intercept[IllegalArgumentException] {
+ matrix.toSparseMatrix(colMajor)
+ }
+ }
+ }
+ }
+
+ test("sparse hashCode preserves Breeze value hashing") {
+ val sparse = new SparseMatrix(3, 2, Array(0, 3, 6), Array(0, 1, 2, 0, 1, 2),
+ Array(0.0, -0.0, Double.NaN, Double.PositiveInfinity, 4294967297.0, 2.5))
+ val ordinary = new SparseMatrix(3, 2, Array(0, 2, 4), Array(0, 2, 0, 1),
+ Array(1.0, 2.0, 3.0, 4.0))
+ val empty = new SparseMatrix(2, 3, Array(0, 0, 0, 0),
+ Array.emptyIntArray, Array.emptyDoubleArray)
+ val matrices = Seq(sparse, ordinary, empty).flatMap(m => Seq(m, m.transpose))
+ val expectedHashes = Seq(1999056974, -274380619, -1538214276, 2094930921, 43, 43)
+ matrices.zip(expectedHashes).foreach { case (matrix, expectedHash) =>
Review Comment:
**[Nit, tests] `zip` silently drops a fixture without an expected hash.**
`zip` stops at the shorter sequence. If a fixture is appended to `Seq(sparse, ordinary, empty)` without two more hashes, it and its transpose get neither the pinned check nor the `asBreeze` check, and the test still passes. The counts match today, so nothing is unchecked now. Pairs keep each hash next to its matrix, so an omission stands out, and a bare matrix in the list would not compile because of `matrix.asBreeze`:
```scala
Seq(
sparse -> 1999056974, sparse.transpose -> -274380619,
ordinary -> -1538214276, ordinary.transpose -> 2094930921,
empty -> 43, empty.transpose -> 43
).foreach { case (matrix, expectedHash) =>
```
Alternatively, `assert(matrices.length === expectedHashes.length)` would do.
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/Matrices.scala:
##########
@@ -755,14 +750,90 @@ class SparseMatrix @Since("2.0.0") (
* order.
*/
private[ml] override def toSparseMatrix(colMajor: Boolean): SparseMatrix = {
- if (isColMajor && !colMajor) {
- // it is col major and we want row major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]].t
- Matrices.fromBreeze(breezeTransposed).transpose.asInstanceOf[SparseMatrix]
- } else if (isRowMajor && colMajor) {
- // it is row major and we want col major, use breeze to remove explicit zeros
- val breezeTransposed = asBreeze.asInstanceOf[BSM[Double]]
- Matrices.fromBreeze(breezeTransposed).asInstanceOf[SparseMatrix]
+ MatrixUtils.checkDimensions(this)
+ if (isColMajor != colMajor) {
+ val numPtrs = if (colMajor) numCols else numRows
+ val newColPtrs = new Array[Int](numPtrs + 1)
+ var j = 0
+ // Count the same pointer ranges used below, including shared or skipped backing entries.
Review Comment:
**[Nit] This comment reads as if skipped entries are counted.**
"including shared or skipped backing entries" most naturally says that skipped entries are part of the count. That is what the round-1 code did, for example with the `99.0` in the new `view` fixture. The loop counts only entries inside the ranges: a shared entry once per range, and an entry outside every range not at all. Maybe:
```suggestion
// Count only entries inside the pointer ranges, exactly as the scatter below visits them.
```
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/MatrixUtils.scala:
##########
@@ -0,0 +1,142 @@
+// scalastyle:off license
+/*
+ Copyright 2012 David Hall
+
+ Licensed under the Apache License, Version 2.0 (the "License")
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+// scalastyle:on license
+
+package org.apache.spark.ml.linalg
+
+import scala.util.hashing.MurmurHash3
+
+/**
+ * Equality, sparse hashing, and traversal adapted from Matrix.scala, CSCMatrix.scala, and
+ * Tensor.scala in Breeze 2.1.0 to use Spark matrix storage.
+ */
+private[linalg] object MatrixUtils {
+
+ def checkDimensions(matrix: Matrix): Unit = {
+ require(matrix.numRows >= 0 && matrix.numCols >= 0,
+ s"Matrix dimensions must be nonnegative: ${matrix.numRows} x ${matrix.numCols}")
+ }
+
+ private def columnMajor(matrix: SparseMatrix): SparseMatrix = {
+ checkDimensions(matrix)
+ // Breeze's transpose normalizes the storage, unlike its ordinary CSC wrapper.
+ if (matrix.isTransposed) matrix.toSparseColMajor else matrix
+ }
+
+ def equals(left: Matrix, other: Any): Boolean = other match {
+ case right: Matrix =>
+ checkDimensions(left)
+ checkDimensions(right)
+ if (left.numRows != right.numRows || left.numCols != right.numCols) {
+ return false
+ }
+ (left, right) match {
+ case (leftSparse: SparseMatrix, rightSparse: SparseMatrix) =>
+ val leftEntries = new SparseEntries(leftSparse)
+ val rightEntries = new SparseEntries(rightSparse)
+ while (leftEntries.hasNext && rightEntries.hasNext) {
+ leftEntries.next()
+ rightEntries.next()
+ // Known Breeze quirk retained for compatibility: a trailing explicit zero is
+ // compared by coordinate. This can make sparse equality non-transitive.
+ while (leftEntries.value == 0.0 && leftEntries.hasNext) {
+ leftEntries.next()
+ }
+ while (rightEntries.value == 0.0 && rightEntries.hasNext) {
+ rightEntries.next()
+ }
+ if (leftEntries.row != rightEntries.row || leftEntries.col != rightEntries.col ||
+ leftEntries.value != rightEntries.value) {
+ return false
+ }
+ }
+ remainingZeros(leftEntries) && remainingZeros(rightEntries)
+ case (leftSparse: SparseMatrix, _) => equalsByElement(columnMajor(leftSparse), right)
+ case (_, rightSparse: SparseMatrix) => equalsByElement(left, columnMajor(rightSparse))
+ case _ => equalsByElement(left, right)
+ }
+ case _ => false
+ }
+
+ private def equalsByElement(left: Matrix, right: Matrix): Boolean = {
Review Comment:
**[Nit, performance] Optional: two dense operands could be compared in storage order.**
This loop calls the bounds-checked `apply` for every cell in column-major order. So two row-major dense matrices are read with a stride of `numCols`, and a dense vs CSC pair runs a binary search per cell. It is already much faster than master's Breeze `Matrix.equals`, so this is not a regression. In an independent model, two equal row-major 20000 x 2000 matrices took 223 ms here vs 10 ms for a linear scan of `values`. A 4000 x 4000 dense matrix vs an equal 10%-dense CSC took 190 ms vs 21 ms with a per-column cursor.
If you want it, two `DenseMatrix` with the same `isTransposed` can compare `values` linearly with `!=`, which keeps the NaN and signed-zero semantics. A `numRows.toLong * numCols == values.length` guard keeps the current `ArrayIndexOutOfBoundsException` for matrices whose size check overflowed in the constructor, e.g. `new DenseMatrix(65536, 65536, Array.emptyDoubleArray)`. Nothing in Spark compares matrices on a hot path, so feel free to skip this.
##########
mllib-local/src/main/scala/org/apache/spark/ml/linalg/MatrixUtils.scala:
##########
@@ -0,0 +1,142 @@
+// scalastyle:off license
+/*
+ Copyright 2012 David Hall
+
+ Licensed under the Apache License, Version 2.0 (the "License")
+ you may not use this file except in compliance with the License.
+ You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+ */
+// scalastyle:on license
+
+package org.apache.spark.ml.linalg
+
+import scala.util.hashing.MurmurHash3
+
+/**
+ * Equality, sparse hashing, and traversal adapted from Matrix.scala, CSCMatrix.scala, and
+ * Tensor.scala in Breeze 2.1.0 to use Spark matrix storage.
+ */
+private[linalg] object MatrixUtils {
+
+ def checkDimensions(matrix: Matrix): Unit = {
+ require(matrix.numRows >= 0 && matrix.numCols >= 0,
+ s"Matrix dimensions must be nonnegative: ${matrix.numRows} x ${matrix.numCols}")
+ }
+
+ private def columnMajor(matrix: SparseMatrix): SparseMatrix = {
+ checkDimensions(matrix)
+ // Breeze's transpose normalizes the storage, unlike its ordinary CSC wrapper.
+ if (matrix.isTransposed) matrix.toSparseColMajor else matrix
+ }
+
+ def equals(left: Matrix, other: Any): Boolean = other match {
+ case right: Matrix =>
+ checkDimensions(left)
+ checkDimensions(right)
+ if (left.numRows != right.numRows || left.numCols != right.numCols) {
+ return false
+ }
+ (left, right) match {
+ case (leftSparse: SparseMatrix, rightSparse: SparseMatrix) =>
+ val leftEntries = new SparseEntries(leftSparse)
+ val rightEntries = new SparseEntries(rightSparse)
+ while (leftEntries.hasNext && rightEntries.hasNext) {
+ leftEntries.next()
+ rightEntries.next()
+ // Known Breeze quirk retained for compatibility: a trailing explicit zero is
+ // compared by coordinate. This can make sparse equality non-transitive.
+ while (leftEntries.value == 0.0 && leftEntries.hasNext) {
+ leftEntries.next()
+ }
+ while (rightEntries.value == 0.0 && rightEntries.hasNext) {
+ rightEntries.next()
+ }
+ if (leftEntries.row != rightEntries.row || leftEntries.col != rightEntries.col ||
+ leftEntries.value != rightEntries.value) {
+ return false
+ }
+ }
+ remainingZeros(leftEntries) && remainingZeros(rightEntries)
+ case (leftSparse: SparseMatrix, _) => equalsByElement(columnMajor(leftSparse), right)
+ case (_, rightSparse: SparseMatrix) => equalsByElement(left, columnMajor(rightSparse))
+ case _ => equalsByElement(left, right)
+ }
+ case _ => false
+ }
+
+ private def equalsByElement(left: Matrix, right: Matrix): Boolean = {
+ var col = 0
+ while (col < left.numCols) {
+ var row = 0
+ while (row < left.numRows) {
+ if (left(row, col) != right(row, col)) {
+ return false
+ }
+ row += 1
+ }
+ col += 1
+ }
+ true
+ }
+
+ private def remainingZeros(entries: SparseEntries): Boolean = {
+ while (entries.hasNext) {
+ entries.next()
+ if (entries.value != 0.0) {
+ return false
+ }
+ }
+ true
+ }
+
+ def hashCode(matrix: SparseMatrix): Int = {
Review Comment:
**[Low] After a rebase on master, this hash loop will exist twice.**
My SPARK-60296 landed on master on Oct 16, after the base of this PR. It changed `DenseMatrix.hashCode` to the same Breeze `QuasiTensor` scheme, so that equal dense and sparse matrices get the same hash ([Matrices.scala L335-L350 on master](https://github.com/apache/spark/blob/979bb1094a2fab9cfa6616bca8286b7e44b3a7ee/mllib-local/src/main/scala/org/apache/spark/ml/linalg/Matrices.scala#L335-L350)). This PR merges cleanly with master, and CI builds the merged tree. So the result has two copies of this loop: this one, and the one in `DenseMatrix.hashCode` in the ASF-headed `Matrices.scala`, with the `MurmurHash3` import back. They differ only in where `data` comes from.
That has two costs:
- The dense/sparse hash contract depends on the two copies staying identical, and only the SPARK-60296 tests check that, for a few fixtures.
- It partly undoes the per-file attribution this round set up. Master's copy does name Breeze's `QuasiTensor.hashCode` in a comment, so this is about consistency rather than licensing.
Neither master nor this PR alone has two Spark-side copies, so this PR is the natural place to merge them once it is rebased. A small separate change right after it would also work. For example, a `MatrixUtils.hashCode(matrix: Matrix)` could hash `columnMajor(sparse).values`, or `if (isTransposed) toArray else values` for a dense matrix, which keeps master's column-major order. Then `DenseMatrix.hashCode` becomes `MatrixUtils.hashCode(this)`, and every hash value stays the same. The legacy test `equals and sparse hashCode delegate to ML matrices` could then drop its dense skip (`case _ =>`), since mllib `DenseMatrix.hashCode` is `asML.hashCode` on master as well.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org