gaturchenko commented on code in PR #2542:
URL: https://github.com/apache/systemds/pull/2542#discussion_r3664225866
##########
src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketch.java:
##########
@@ -19,89 +19,807 @@
package org.apache.sysds.runtime.matrix.data;
-import org.apache.sysds.common.Types;
-
+import java.util.ArrayList;
import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.sysds.common.Types;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.util.CommonThreadPool;
+import org.apache.sysds.runtime.util.UtilFunctions;
+import org.apache.sysds.utils.stats.InfrastructureAnalyzer;
public class LibMatrixSketch {
+ private static final long PAR_UNIQUE_NUMCELL_THRESHOLD = 1024 * 16;
+ private static final long PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION = 4;
+ /**
+ * Conservative footprint of one value retained in a
HashSet<Double>: the boxed Double plus amortized hash-map
+ * node and backing-array overhead.
+ */
+ private static final long PAR_UNIQUE_BYTES_PER_CELL = Double.BYTES * 8;
+ /**
+ * Computes unique values with the original single-threaded behavior.
The overload with a parallelism argument keeps
+ * this path as the k=1 baseline.
+ *
+ * @param blkIn input matrix block
+ * @param dir unique direction
+ * @return matrix block containing unique values
+ */
public static MatrixBlock getUniqueValues(MatrixBlock blkIn,
Types.Direction dir) {
- //similar to R's unique, this operation takes a matrix and
computes the
- //unique values (or rows in case of multiple column inputs)
-
+ return getUniqueValues(blkIn, dir, 1);
+ }
+
+ /**
+ * Computes unique values. For sufficiently large inputs and k > 1,
this uses parallel local deduplication or its
+ * batched variant.
+ *
+ * @param blkIn input matrix block
+ * @param dir unique direction
+ * @param k requested degree of parallelism
+ * @return matrix block containing unique values
+ */
+ public static MatrixBlock getUniqueValues(MatrixBlock blkIn,
Types.Direction dir, int k) {
+ return getUniqueValues(blkIn, dir, k,
getDefaultLocalBytesBudget());
+ }
+
+ /**
+ * Computes unique values with an explicit budget for the transient
deduplication structures. The budget decides
+ * between the full parallel path, its batched variant, and the
sequential fallback. This overload exists so tests
+ * can inject a small budget and deterministically exercise the batched
path, which the heap-derived default would
+ * not trigger.
+ *
+ * @param blkIn input matrix block
+ * @param dir unique direction
+ * @param k requested degree of parallelism
+ * @param maxLocalBytes budget in bytes for transient deduplication
structures
+ * @return matrix block containing unique values
+ */
+ public static MatrixBlock getUniqueValues(MatrixBlock blkIn,
Types.Direction dir, int k, long maxLocalBytes) {
+ // Similar to R's unique, this operation computes unique values
according
+ // to the requested direction.
+ if(!satisfiesMultiThreadingConstraints(blkIn, dir, k))
+ return getUniqueValuesSequential(blkIn, dir);
+
+ boolean localDedupMemorySafe =
isLocalDedupMemoryBudgetSafe(blkIn, dir, k, maxLocalBytes);
+ switch(dir) {
+ case RowCol:
Review Comment:
Sorry, I missed this one in my prior review. Based on your benchmarking
results, it seems that multi-threading yields a slowdown for `RowCol`. Can we
then just fall back to single-threaded here?
##########
src/test/java/org/apache/sysds/test/functions/unique/UniqueBase.java:
##########
@@ -40,8 +41,39 @@ public void setUp() {
protected void uniqueTest(double[][] inputMatrix, double[][]
expectedMatrix,
Types.ExecType
instType, double epsilon) {
+ uniqueTest(inputMatrix, expectedMatrix, instType, epsilon, -1,
false);
+ }
+
+ /**
+ * Runs the unique script and compares the result row by row, in order.
Use this where the expected output is
+ * unambiguous, i.e. where every row or column of the result holds a
single value and the iteration order of the
+ * internal hash sets cannot affect the outcome.
+ */
+ protected void uniqueTestOrdered(double[][] inputMatrix, double[][]
expectedMatrix, Types.ExecType instType,
Review Comment:
Why do you need this separate function? The existing `uniqueTest` function
compares the results out-of-order that works both when the output is
deterministic or random. Am I missing something here?
##########
src/test/java/org/apache/sysds/test/functions/unique/UniqueBatchedPathTest.java:
##########
Review Comment:
Why is this still a unit test? You have limited the memory with
`InfrastructureAnalyzer` but you can limit concurrency as well by calling
`InfrastructureAnalyzer.setLocalPar()`, so I don't quite see what holds you
back from exercising the batched path for all 3 cases end-to-end
##########
src/test/java/org/apache/sysds/test/functions/unique/UniqueBase.java:
##########
@@ -52,9 +84,13 @@ protected void uniqueTest(double[][] inputMatrix, double[][]
expectedMatrix,
runTest(true, false, null, -1);
writeExpectedMatrix("A", expectedMatrix);
- compareResultsRowsOutOfOrder(epsilon);
+ if(orderedComparison)
Review Comment:
Again as per my comment above, I don't really understand the purpose of
`orderedComparison`
##########
src/test/java/org/apache/sysds/test/functions/unique/UniqueRow.java:
##########
@@ -76,4 +76,58 @@ public void testNoDuplicatesCP() {
double[][] expectedMatrix = {{1, 2, 3}, {4, 5, 6}, {7, 8, 9}};
uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0);
}
+
+ /**
+ * Large enough to take the multi-threaded path. Every row holds a
single distinct value, so the expected result is
+ * one column and independent of any hash set iteration order.
+ */
+ @Test
+ public void testMultiThreadedCP() {
Review Comment:
Same comment as for `Col`
##########
src/test/java/org/apache/sysds/test/functions/unique/UniqueRowCol.java:
##########
Review Comment:
If multi-threading is not beneficial for `RowCol`, we don't need any new
tests here
##########
src/test/java/org/apache/sysds/test/functions/unique/UniqueCol.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.sysds.test.functions.unique;
+
+import org.apache.sysds.common.Types;
+import org.junit.Test;
+
+public class UniqueCol extends UniqueBase {
+ private final static String TEST_NAME = "uniqueCol";
+ private final static String TEST_DIR = "functions/unique/";
+ private static final String TEST_CLASS_DIR = TEST_DIR +
UniqueCol.class.getSimpleName() + "/";
+
+ @Override
+ protected String getTestName() {
+ return TEST_NAME;
+ }
+
+ @Override
+ protected String getTestDir() {
+ return TEST_DIR;
+ }
+
+ @Override
+ protected String getTestClassDir() {
+ return TEST_CLASS_DIR;
+ }
+
+ @Test
+ public void testBaseCaseCP() {
+ double[][] inputMatrix = {{0}};
+ double[][] expectedMatrix = {{0}};
+ uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0);
+ }
+
+ @Test
+ public void testSingleColumnCP() {
+ double[][] inputMatrix = {{1}, {1}, {6}, {9}, {4}, {2}, {0},
{9}, {0}, {0}, {4}, {4}};
+ double[][] expectedMatrix = {{1}, {6}, {9}, {4}, {2}, {0}};
+ uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0);
+ }
+
+ @Test
+ public void testConstantColumnsCP() {
+ // every column holds a single distinct value, so the result
has one row
+ double[][] inputMatrix = {{1, 2, 3}, {1, 2, 3}, {1, 2, 3}};
+ double[][] expectedMatrix = {{1, 2, 3}};
+ uniqueTestOrdered(inputMatrix, expectedMatrix,
Types.ExecType.CP, 0.0);
+ }
+
+ @Test
+ public void testNoDuplicatesCP() {
+ double[][] inputMatrix = {{1, 2, 3}, {4, 5, 6}, {7, 8, 9}};
+ double[][] expectedMatrix = {{1, 2, 3}, {4, 5, 6}, {7, 8, 9}};
+ uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0);
+ }
+
+ /**
+ * Large enough to take the multi-threaded path. Every column holds a
single distinct value, so the expected result
+ * is one row and independent of any hash set iteration order.
+ */
+ @Test
+ public void testMultiThreadedCP() {
Review Comment:
Thank you for covering the `Col` case as well. However, what do you mean by
`large enough`? The multi-threaded execution should be picked up by default
unless the constraints are violated. Why are the 4 existing tests insufficient?
##########
src/test/java/org/apache/sysds/test/functions/unique/UniqueRow.java:
##########
@@ -76,4 +76,58 @@ public void testNoDuplicatesCP() {
double[][] expectedMatrix = {{1, 2, 3}, {4, 5, 6}, {7, 8, 9}};
uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0);
}
+
+ /**
+ * Large enough to take the multi-threaded path. Every row holds a
single distinct value, so the expected result is
+ * one column and independent of any hash set iteration order.
+ */
+ @Test
+ public void testMultiThreadedCP() {
+ uniqueTestOrdered(constantRows(400, 64),
expectedConstantRows(400), Types.ExecType.CP, 0.0);
+ }
+
+ /**
+ * Same input under a heavily reduced local memory budget. Row-wise
workers reuse a single set that is cleared per
+ * row, so only one live set per thread is charged and the parallel
path stays applicable; this guards against
+ * needlessly falling back to batched or sequential execution.
+ */
+ @Test
+ public void testReducedMemoryBudgetCP() {
Review Comment:
Same comment as for `Col`
##########
src/test/java/org/apache/sysds/test/functions/unique/UniqueCol.java:
##########
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.sysds.test.functions.unique;
+
+import org.apache.sysds.common.Types;
+import org.junit.Test;
+
+public class UniqueCol extends UniqueBase {
+ private final static String TEST_NAME = "uniqueCol";
+ private final static String TEST_DIR = "functions/unique/";
+ private static final String TEST_CLASS_DIR = TEST_DIR +
UniqueCol.class.getSimpleName() + "/";
+
+ @Override
+ protected String getTestName() {
+ return TEST_NAME;
+ }
+
+ @Override
+ protected String getTestDir() {
+ return TEST_DIR;
+ }
+
+ @Override
+ protected String getTestClassDir() {
+ return TEST_CLASS_DIR;
+ }
+
+ @Test
+ public void testBaseCaseCP() {
+ double[][] inputMatrix = {{0}};
+ double[][] expectedMatrix = {{0}};
+ uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0);
+ }
+
+ @Test
+ public void testSingleColumnCP() {
+ double[][] inputMatrix = {{1}, {1}, {6}, {9}, {4}, {2}, {0},
{9}, {0}, {0}, {4}, {4}};
+ double[][] expectedMatrix = {{1}, {6}, {9}, {4}, {2}, {0}};
+ uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0);
+ }
+
+ @Test
+ public void testConstantColumnsCP() {
+ // every column holds a single distinct value, so the result
has one row
+ double[][] inputMatrix = {{1, 2, 3}, {1, 2, 3}, {1, 2, 3}};
+ double[][] expectedMatrix = {{1, 2, 3}};
+ uniqueTestOrdered(inputMatrix, expectedMatrix,
Types.ExecType.CP, 0.0);
+ }
+
+ @Test
+ public void testNoDuplicatesCP() {
+ double[][] inputMatrix = {{1, 2, 3}, {4, 5, 6}, {7, 8, 9}};
+ double[][] expectedMatrix = {{1, 2, 3}, {4, 5, 6}, {7, 8, 9}};
+ uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0);
+ }
+
+ /**
+ * Large enough to take the multi-threaded path. Every column holds a
single distinct value, so the expected result
+ * is one row and independent of any hash set iteration order.
+ */
+ @Test
+ public void testMultiThreadedCP() {
+ int rlen = 64, clen = 400; // 25,600 cells, above the
multi-threading threshold
+ double[][] inputMatrix = new double[rlen][clen];
+ double[][] expectedMatrix = new double[1][clen];
+ for(int j = 0; j < clen; j++) {
+ for(int i = 0; i < rlen; i++)
+ inputMatrix[i][j] = j;
+ expectedMatrix[0][j] = j;
+ }
+ uniqueTestOrdered(inputMatrix, expectedMatrix,
Types.ExecType.CP, 0.0);
+ }
+
+ /**
+ * Same input under a heavily reduced local memory budget. Column-wise
workers reuse a single set that is cleared
+ * per column, so only one live set per thread is charged and the
parallel path stays applicable; this guards
+ * against needlessly falling back to batched or sequential execution.
+ */
+ @Test
+ public void testReducedMemoryBudgetCP() {
Review Comment:
This should rather be in the batched test, although in principle you compare
the results here, but do not actually check that the batched path was selected.
So maybe if there is no way to verify it we don't need this test at all
##########
src/test/java/org/apache/sysds/test/functions/unique/UniqueRow.java:
##########
@@ -76,4 +76,58 @@ public void testNoDuplicatesCP() {
double[][] expectedMatrix = {{1, 2, 3}, {4, 5, 6}, {7, 8, 9}};
uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0);
}
+
+ /**
+ * Large enough to take the multi-threaded path. Every row holds a
single distinct value, so the expected result is
+ * one column and independent of any hash set iteration order.
+ */
+ @Test
+ public void testMultiThreadedCP() {
+ uniqueTestOrdered(constantRows(400, 64),
expectedConstantRows(400), Types.ExecType.CP, 0.0);
+ }
+
+ /**
+ * Same input under a heavily reduced local memory budget. Row-wise
workers reuse a single set that is cleared per
+ * row, so only one live set per thread is charged and the parallel
path stays applicable; this guards against
+ * needlessly falling back to batched or sequential execution.
+ */
+ @Test
+ public void testReducedMemoryBudgetCP() {
+ uniqueTestConstrainedMemory(constantRows(400, 64),
expectedConstantRows(400), Types.ExecType.CP, 0.0,
+ 16 * 1024 * 1024);
+ }
+
+ /**
+ * Sparse counterpart of the multi-threaded case: only every eighth row
is populated, so the input is read in sparse
+ * format. Every row still holds a single distinct value, either its
filler or zero, so the expected result stays
+ * one column.
+ */
+ @Test
+ public void testSparseMultiThreadedCP() {
Review Comment:
What do you verify with this test? Why is this case different from the other
tests?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
