diff --git a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketch.java b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketch.java index 8fdc276d660..f0afec847f8 100644 --- a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketch.java +++ b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketch.java @@ -19,89 +19,695 @@ 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; public class LibMatrixSketch { + private static final long PAR_UNIQUE_NUMCELL_THRESHOLD = 1024 * 16; + private static final long PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION = 4; + private static final long PAR_UNIQUE_LOCAL_BYTES_OVERHEAD = 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) { + // 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); + switch(dir) { + case RowCol: + return localDedupMemorySafe ? + getUniqueValuesRowColParallel(blkIn, k) : + getUniqueValuesRowColBatchedParallel(blkIn, dir, k); + case Row: + return localDedupMemorySafe ? + getUniqueRowValuesParallel(blkIn, k) : + getUniqueRowValuesBatchedParallel(blkIn, dir, k); + case Col: + return localDedupMemorySafe ? + getUniqueColumnValuesParallel(blkIn, k) : + getUniqueColumnValuesBatchedParallel(blkIn, dir, k); + default: + throw new IllegalArgumentException("Unrecognized direction: " + dir); + } + } + + /** + * Single-threaded baseline implementation for all unique directions. + * This preserves the original row-wise and column-wise unique behavior. + * + * @param blkIn input matrix block + * @param dir unique direction + * @return matrix block containing unique values + */ + private static MatrixBlock getUniqueValuesSequential(MatrixBlock blkIn, Types.Direction dir) { int rlen = blkIn.getNumRows(); int clen = blkIn.getNumColumns(); MatrixBlock blkOut = null; - // TODO optimize for dense/sparse/compressed (once multi-column support added) - switch (dir) { - case RowCol: { - // obtain set of unique items (dense input vector) + case RowCol: + // TODO optimize for dense/sparse/compressed (once multi-column support added) + + // obtain set of unique items HashSet hashSet = new HashSet<>(); for( int i=0; i hashSet = new HashSet<>(); + + case Row: + // 2-pass algorithm to avoid unnecessarily large mem requirements + HashSet rowSet = new HashSet<>(); int clen2 = 0; for( int i=0; i hashSet = new HashSet<>(); + + case Col: + // 2-pass algorithm to avoid unnecessarily large mem requirements + HashSet colSet = new HashSet<>(); int rlen2 = 0; for( int j=0; j tasks = new ArrayList<>(); + for( int[] range : getBalancedRanges(blkIn.getNumRows(), numThreads) ) + tasks.add(new UniqueValueTask(blkIn, range[0], range[1])); + + // Merge local sets after the workers complete. + HashSet hashSet = new HashSet<>(); + List>> rtasks = pool.invokeAll(tasks); + for( int i = 0; i < rtasks.size(); i++ ) { + HashSet localSet = rtasks.get(i).get(); + hashSet.addAll(localSet); + localSet.clear(); + rtasks.set(i, null); } + + return createRowColOutput(hashSet); + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + pool.shutdown(); + } + } + + /** + * Parallel row-wise unique values. A first pass computes the output width, + * and a second pass materializes the row-local unique values. + * + * @param blkIn input matrix block + * @param k requested degree of parallelism + * @return matrix block containing row-wise unique values + */ + private static MatrixBlock getUniqueRowValuesParallel(MatrixBlock blkIn, int k) { + int numThreads = getNumThreads(k, blkIn.getNumRows()); + ExecutorService pool = CommonThreadPool.get(numThreads); + try { + ArrayList ranges = getBalancedRanges(blkIn.getNumRows(), numThreads); + int clen2 = getMaxUniqueValues(pool, blkIn, Types.Direction.Row, ranges); + MatrixBlock blkOut = allocateOutputBlock(blkIn.getNumRows(), clen2); + fillUniqueValues(pool, blkIn, blkOut, Types.Direction.Row, ranges); + + blkOut.recomputeNonZeros(); + blkOut.examSparsity(); + return blkOut; + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + pool.shutdown(); + } + } + + /** + * Parallel column-wise unique values. A first pass computes the output height, + * and a second pass materializes the column-local unique values. + * + * @param blkIn input matrix block + * @param k requested degree of parallelism + * @return matrix block containing column-wise unique values + */ + private static MatrixBlock getUniqueColumnValuesParallel(MatrixBlock blkIn, int k) { + int numThreads = getNumThreads(k, blkIn.getNumColumns()); + ExecutorService pool = CommonThreadPool.get(numThreads); + try { + ArrayList ranges = getBalancedRanges(blkIn.getNumColumns(), numThreads); + int rlen2 = getMaxUniqueValues(pool, blkIn, Types.Direction.Col, ranges); + MatrixBlock blkOut = allocateOutputBlock(rlen2, blkIn.getNumColumns()); + fillUniqueValues(pool, blkIn, blkOut, Types.Direction.Col, ranges); + + blkOut.recomputeNonZeros(); + blkOut.examSparsity(); + return blkOut; + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + pool.shutdown(); + } + } + + /** + * Batched parallel unique for all matrix values. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @return one-column matrix block containing the unique values + */ + private static MatrixBlock getUniqueValuesRowColBatchedParallel(MatrixBlock blkIn, Types.Direction dir, int k) { + BatchConfig config = getBatchConfig(blkIn, dir, k); + if( config == null ) + return getUniqueValuesSequential(blkIn, dir); + + ExecutorService pool = CommonThreadPool.get(config._numThreads); + try { + HashSet hashSet = new HashSet<>(); + for( int pos = 0; pos < config._len; ) { + ArrayList tasks = new ArrayList<>(); + for( int i = 0; i < config._numThreads && pos < config._len; i++ ) { + int end = Math.min(pos + config._taskLen, config._len); + tasks.add(new UniqueValueTask(blkIn, pos, end)); + pos = end; + } + + List>> rtasks = pool.invokeAll(tasks); + for( int i = 0; i < rtasks.size(); i++ ) { + HashSet localSet = rtasks.get(i).get(); + hashSet.addAll(localSet); + localSet.clear(); + rtasks.set(i, null); + } + } + + return createRowColOutput(hashSet); + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + pool.shutdown(); + } + } + + /** + * Batched parallel row-wise unique values. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @return matrix block containing row-wise unique values + */ + private static MatrixBlock getUniqueRowValuesBatchedParallel(MatrixBlock blkIn, Types.Direction dir, int k) { + BatchConfig config = getBatchConfig(blkIn, dir, k); + if( config == null ) + return getUniqueValuesSequential(blkIn, dir); + + ExecutorService pool = CommonThreadPool.get(config._numThreads); + try { + int clen2 = getMaxUniqueValuesBatched(pool, blkIn, dir, config); + MatrixBlock blkOut = allocateOutputBlock(blkIn.getNumRows(), clen2); + fillUniqueValuesBatched(pool, blkIn, blkOut, dir, config); + + blkOut.recomputeNonZeros(); + blkOut.examSparsity(); + return blkOut; + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + pool.shutdown(); + } + } + + /** + * Batched parallel column-wise unique values. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @return matrix block containing column-wise unique values + */ + private static MatrixBlock getUniqueColumnValuesBatchedParallel(MatrixBlock blkIn, Types.Direction dir, int k) { + BatchConfig config = getBatchConfig(blkIn, dir, k); + if( config == null ) + return getUniqueValuesSequential(blkIn, dir); + + ExecutorService pool = CommonThreadPool.get(config._numThreads); + try { + int rlen2 = getMaxUniqueValuesBatched(pool, blkIn, dir, config); + MatrixBlock blkOut = allocateOutputBlock(rlen2, blkIn.getNumColumns()); + fillUniqueValuesBatched(pool, blkIn, blkOut, dir, config); + + blkOut.recomputeNonZeros(); + blkOut.examSparsity(); + return blkOut; + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + pool.shutdown(); + } + } + + /** + * Computes the maximum row-wise or column-wise unique count over balanced ranges. + */ + private static int getMaxUniqueValues(ExecutorService pool, MatrixBlock blkIn, Types.Direction dir, + ArrayList ranges) throws Exception { + ArrayList tasks = new ArrayList<>(); + for( int[] range : ranges ) + tasks.add(new UniqueCountTask(blkIn, dir, range[0], range[1])); + + int ret = 0; + List> rtasks = pool.invokeAll(tasks); + for( int i = 0; i < rtasks.size(); i++ ) { + ret = Math.max(ret, rtasks.get(i).get()); + rtasks.set(i, null); + } + return ret; + } + + /** + * Fills row-wise or column-wise unique values over balanced ranges. + */ + private static void fillUniqueValues(ExecutorService pool, MatrixBlock blkIn, MatrixBlock blkOut, + Types.Direction dir, ArrayList ranges) throws Exception { + ArrayList tasks = new ArrayList<>(); + for( int[] range : ranges ) + tasks.add(new UniqueOutputTask(blkIn, blkOut, dir, range[0], range[1])); + List> rtasks = pool.invokeAll(tasks); + for( int i = 0; i < rtasks.size(); i++ ) { + rtasks.get(i).get(); + rtasks.set(i, null); + } + } + + /** + * Batched variant of getMaxUniqueValues. + */ + private static int getMaxUniqueValuesBatched(ExecutorService pool, MatrixBlock blkIn, Types.Direction dir, + BatchConfig config) throws Exception { + int ret = 0; + for( int pos = 0; pos < config._len; ) { + ArrayList tasks = new ArrayList<>(); + for( int i = 0; i < config._numThreads && pos < config._len; i++ ) { + int end = Math.min(pos + config._taskLen, config._len); + tasks.add(new UniqueCountTask(blkIn, dir, pos, end)); + pos = end; + } + + List> rtasks = pool.invokeAll(tasks); + for( int i = 0; i < rtasks.size(); i++ ) { + ret = Math.max(ret, rtasks.get(i).get()); + rtasks.set(i, null); + } + } + return ret; + } + + /** + * Batched variant of fillUniqueValues. + */ + private static void fillUniqueValuesBatched(ExecutorService pool, MatrixBlock blkIn, MatrixBlock blkOut, + Types.Direction dir, BatchConfig config) throws Exception { + for( int pos = 0; pos < config._len; ) { + ArrayList tasks = new ArrayList<>(); + for( int i = 0; i < config._numThreads && pos < config._len; i++ ) { + int end = Math.min(pos + config._taskLen, config._len); + tasks.add(new UniqueOutputTask(blkIn, blkOut, dir, pos, end)); + pos = end; + } + + List> rtasks = pool.invokeAll(tasks); + for( int i = 0; i < rtasks.size(); i++ ) { + rtasks.get(i).get(); + rtasks.set(i, null); + } + } + } + + /** + * Decides whether the input is large enough to justify local deduplication tasks. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @return true if the parallel path should be used + */ + private static boolean satisfiesMultiThreadingConstraints(MatrixBlock blkIn, Types.Direction dir, int k) { + if( k <= 1 || ((long) blkIn.getNumRows()) * blkIn.getNumColumns() < PAR_UNIQUE_NUMCELL_THRESHOLD ) + return false; + + switch(dir) { + case RowCol: + return blkIn.getNumRows() > 1; + case Row: + return blkIn.getNumRows() > 1; + case Col: + return blkIn.getNumColumns() > 1; default: throw new IllegalArgumentException("Unrecognized direction: " + dir); } + } + + /** + * Creates balanced half-open ranges [start, end) using the same utility pattern as + * other SystemDS matrix libraries. + * + * @param len number of rows or columns to partition + * @param k requested degree of parallelism + * @return list of balanced index ranges + */ + private static ArrayList getBalancedRanges(int len, int k) { + ArrayList ranges = new ArrayList<>(); + ArrayList blklens = UtilFunctions.getBalancedBlockSizesDefault(len, getNumThreads(k, len), false); + for( int i = 0, lb = 0; i < blklens.size(); lb += blklens.get(i), i++ ) + ranges.add(new int[] {lb, lb + blklens.get(i)}); + return ranges; + } + + /** + * Caps the number of workers by the number of row or column partitions available. + * + * @param k requested degree of parallelism + * @param len number of rows or columns to partition + * @return effective number of worker threads + */ + private static int getNumThreads(int k, int len) { + return Math.max(1, Math.min(k, len)); + } + + /** + * Builds the execution plan for the batched parallel path. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @return batch configuration, or null if batching would not use parallelism + */ + private static BatchConfig getBatchConfig(MatrixBlock blkIn, Types.Direction dir, int k) { + int len = getPartitionLength(blkIn, dir); + long maxBatchIndexes = getMaxLocalDedupIndexes(blkIn, dir); + if( maxBatchIndexes < 2 ) + return null; + + int numThreads = getNumThreads(k, (int) Math.min(len, maxBatchIndexes)); + if( numThreads <= 1 ) + return null; + + int taskLen = Math.max(1, (int) Math.min(Integer.MAX_VALUE, maxBatchIndexes / numThreads)); + return new BatchConfig(numThreads, taskLen, len); + } + + /** + * Returns the number of rows or columns that define the partition direction. + * + * @param blkIn input matrix block + * @param dir unique direction + * @return number of partition indexes + */ + private static int getPartitionLength(MatrixBlock blkIn, Types.Direction dir) { + return dir == Types.Direction.Col ? blkIn.getNumColumns() : blkIn.getNumRows(); + } + + /** + * Estimates how many partition indexes can be processed in one batch. + * + * @param blkIn input matrix block + * @param dir unique direction + * @return maximum number of rows or columns per batch + */ + private static long getMaxLocalDedupIndexes(MatrixBlock blkIn, Types.Direction dir) { + long cellsPerIndex = dir == Types.Direction.Col ? blkIn.getNumRows() : blkIn.getNumColumns(); + if( cellsPerIndex <= 0 || + cellsPerIndex > Long.MAX_VALUE / Double.BYTES / PAR_UNIQUE_LOCAL_BYTES_OVERHEAD ) + return 0; + + long bytesPerIndex = cellsPerIndex * Double.BYTES * PAR_UNIQUE_LOCAL_BYTES_OVERHEAD; + long maxLocalBytes = Runtime.getRuntime().maxMemory() / PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION; + return maxLocalBytes / bytesPerIndex; + } + + /** + * Conservative memory guard for full local deduplication. The estimate includes + * a small overhead factor for set objects. + * + * @param blkIn input matrix block + * @param dir unique direction + * @return true if local deduplication is small enough for the parallel path + */ + private static boolean isLocalDedupMemoryBudgetSafe(MatrixBlock blkIn, Types.Direction dir) { + return getMaxLocalDedupIndexes(blkIn, dir) >= getPartitionLength(blkIn, dir); + } + + /** + * Allocates and fills a one-column MatrixBlock from a set of unique scalar values. + * + * @param values unique scalar values + * @return one-column matrix block + */ + private static MatrixBlock createRowColOutput(HashSet values) { + int rlen = values.size(); + MatrixBlock blkOut = allocateOutputBlock(rlen, 1); + Iterator iter = values.iterator(); + for( int i = 0; i < rlen; i++ ) + blkOut.set(i, 0, iter.next()); + blkOut.recomputeNonZeros(); + blkOut.examSparsity(); + return blkOut; + } + /** + * Creates an output block and allocates storage only when at least one cell exists. + * + * @param rlen number of rows + * @param clen number of columns + * @return matrix block ready for writes when non-empty + */ + private static MatrixBlock allocateOutputBlock(int rlen, int clen) { + MatrixBlock blkOut = new MatrixBlock(rlen, clen, false); + if( rlen > 0 && clen > 0 ) + blkOut.allocateBlock(); return blkOut; } + + /** + * Configuration for batched execution. + */ + private static class BatchConfig { + private final int _numThreads; + private final int _taskLen; + private final int _len; + + private BatchConfig(int numThreads, int taskLen, int len) { + _numThreads = numThreads; + _taskLen = taskLen; + _len = len; + } + } + + /** + * Worker that deduplicates all scalar values in a row range locally. + */ + private static class UniqueValueTask implements Callable> { + private final MatrixBlock _blkIn; + private final int _rl; + private final int _ru; + + private UniqueValueTask(MatrixBlock blkIn, int rl, int ru) { + _blkIn = blkIn; + _rl = rl; + _ru = ru; + } + + @Override + public HashSet call() { + HashSet ret = new HashSet<>(); + for( int i = _rl; i < _ru; i++ ) + for( int j = 0; j < _blkIn.getNumColumns(); j++ ) + ret.add(_blkIn.get(i, j)); + return ret; + } + } + + /** + * Worker that computes the largest row-wise or column-wise unique count in a range. + */ + private static class UniqueCountTask implements Callable { + private final MatrixBlock _blkIn; + private final Types.Direction _dir; + private final int _l; + private final int _u; + + private UniqueCountTask(MatrixBlock blkIn, Types.Direction dir, int l, int u) { + _blkIn = blkIn; + _dir = dir; + _l = l; + _u = u; + } + + @Override + public Integer call() { + HashSet hashSet = new HashSet<>(); + int ret = 0; + if( _dir == Types.Direction.Row ) { + for( int i = _l; i < _u; i++ ) { + hashSet.clear(); + for( int j = 0; j < _blkIn.getNumColumns(); j++ ) + hashSet.add(_blkIn.get(i, j)); + ret = Math.max(ret, hashSet.size()); + } + } + else { + for( int j = _l; j < _u; j++ ) { + hashSet.clear(); + for( int i = 0; i < _blkIn.getNumRows(); i++ ) + hashSet.add(_blkIn.get(i, j)); + ret = Math.max(ret, hashSet.size()); + } + } + return ret; + } + } + + /** + * Worker that writes row-wise or column-wise unique values for a range. + */ + private static class UniqueOutputTask implements Callable { + private final MatrixBlock _blkIn; + private final MatrixBlock _blkOut; + private final Types.Direction _dir; + private final int _l; + private final int _u; + + private UniqueOutputTask(MatrixBlock blkIn, MatrixBlock blkOut, Types.Direction dir, int l, int u) { + _blkIn = blkIn; + _blkOut = blkOut; + _dir = dir; + _l = l; + _u = u; + } + + @Override + public Void call() { + HashSet hashSet = new HashSet<>(); + if( _dir == Types.Direction.Row ) { + for( int i = _l; i < _u; i++ ) { + hashSet.clear(); + for( int j = 0; j < _blkIn.getNumColumns(); j++ ) + hashSet.add(_blkIn.get(i, j)); + int pos = 0; + for( Double val : hashSet ) + _blkOut.set(i, pos++, val); + } + } + else { + for( int j = _l; j < _u; j++ ) { + hashSet.clear(); + for( int i = 0; i < _blkIn.getNumRows(); i++ ) + hashSet.add(_blkIn.get(i, j)); + int pos = 0; + for( Double val : hashSet ) + _blkOut.set(pos++, j, val); + } + } + return null; + } + } } diff --git a/src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java b/src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java new file mode 100644 index 00000000000..cf95b593dc3 --- /dev/null +++ b/src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java @@ -0,0 +1,209 @@ +/* + * 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.runtime.matrix.data; + +import static org.junit.Assert.assertEquals; + +import java.util.HashSet; + +import org.apache.sysds.common.Types; +import org.junit.Test; + +/** + * Tests LibMatrixSketch unique paths with k=1 and k>1. + */ +public class LibMatrixSketchUniqueParallelTest { + @Test + public void testRowColUniqueMatchesBaseline() { + MatrixBlock input = new MatrixBlock(20000, 2, false).allocateBlock(); + for( int i = 0; i < input.getNumRows(); i++ ) { + input.set(i, 0, i % 7); + input.set(i, 1, (i + 3) % 11); + } + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.RowCol); + MatrixBlock parallel = LibMatrixSketch.getUniqueValues(input, Types.Direction.RowCol, 4); + + assertDimensions(parallel, 11, 1, "RowCol parallel dimensions"); + assertSameScalarSet(baseline, parallel, "RowCol baseline vs parallel"); + } + + @Test + public void testRowUniqueMatchesBaselineAndExpectedValues() { + MatrixBlock input = new MatrixBlock(12000, 4, false).allocateBlock(); + for( int i = 0; i < input.getNumRows(); i++ ) { + int pattern = i % 4; + input.set(i, 0, pattern); + input.set(i, 1, pattern); + input.set(i, 2, pattern + 10); + input.set(i, 3, pattern + 20); + } + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row); + MatrixBlock parallel = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row, 4); + + assertDimensions(parallel, input.getNumRows(), 3, "Row parallel dimensions"); + assertBlockEquals(baseline, parallel, "Row baseline vs parallel"); + assertSameRowSet(parallel, 5, new double[] {1, 11, 21}, "Row expected values"); + } + + @Test + public void testColumnUniqueMatchesBaselineAndExpectedValues() { + MatrixBlock input = new MatrixBlock(4, 5000, false).allocateBlock(); + for( int j = 0; j < input.getNumColumns(); j++ ) { + int pattern = j % 4; + input.set(0, j, pattern); + input.set(1, j, pattern); + input.set(2, j, pattern + 10); + input.set(3, j, pattern + 20); + } + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col); + MatrixBlock parallel = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col, 4); + + assertDimensions(parallel, 3, input.getNumColumns(), "Col parallel dimensions"); + assertBlockEquals(baseline, parallel, "Col baseline vs parallel"); + assertSameColumnSet(parallel, 5, new double[] {1, 11, 21}, "Col expected values"); + } + + @Test + public void testLargerInputsMatchBaseline() { + testRowColLargeInput(); + testRowLargeInput(); + testColumnLargeInput(); + } + + private static void testRowColLargeInput() { + MatrixBlock input = new MatrixBlock(1200000, 1, false).allocateBlock(); + for( int i = 0; i < input.getNumRows(); i++ ) + input.set(i, 0, i % 7); + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.RowCol); + MatrixBlock parallel = LibMatrixSketch.getUniqueValues(input, Types.Direction.RowCol, 4); + + assertDimensions(parallel, 7, 1, "RowCol large input dimensions"); + assertSameScalarSet(baseline, parallel, "RowCol large input baseline vs parallel"); + } + + private static void testRowLargeInput() { + MatrixBlock input = new MatrixBlock(80000, 16, false).allocateBlock(); + for( int i = 0; i < input.getNumRows(); i++ ) + for( int j = 0; j < input.getNumColumns(); j++ ) + input.set(i, j, j % 4 + 1); + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row); + MatrixBlock parallel = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row, 4); + + assertDimensions(parallel, input.getNumRows(), 4, "Row large input dimensions"); + assertBlockEquals(baseline, parallel, "Row large input baseline vs parallel"); + assertSameRowSet(parallel, 0, new double[] {1, 2, 3, 4}, "Row large input expected values"); + } + + private static void testColumnLargeInput() { + MatrixBlock input = new MatrixBlock(16, 80000, false).allocateBlock(); + for( int j = 0; j < input.getNumColumns(); j++ ) + for( int i = 0; i < input.getNumRows(); i++ ) + input.set(i, j, i % 4 + 1); + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col); + MatrixBlock parallel = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col, 4); + + assertDimensions(parallel, 4, input.getNumColumns(), "Col large input dimensions"); + assertBlockEquals(baseline, parallel, "Col large input baseline vs parallel"); + assertSameColumnSet(parallel, 0, new double[] {1, 2, 3, 4}, "Col large input expected values"); + } + + /** + * Compares two MatrixBlocks cell by cell with exact equality. + */ + private static void assertBlockEquals(MatrixBlock expected, MatrixBlock actual, String message) { + assertDimensions(actual, expected.getNumRows(), expected.getNumColumns(), message); + for( int i = 0; i < expected.getNumRows(); i++ ) + for( int j = 0; j < expected.getNumColumns(); j++ ) + assertEquals(message + " mismatch at (" + i + ", " + j + ")", expected.get(i, j), + actual.get(i, j), 0); + } + + /** + * Compares RowCol output as a set. + */ + private static void assertSameScalarSet(MatrixBlock expected, MatrixBlock actual, String message) { + assertDimensions(actual, expected.getNumRows(), expected.getNumColumns(), message); + HashSet expectedValues = collectScalars(expected); + HashSet actualValues = collectScalars(actual); + assertEquals(message + " mismatch", expectedValues, actualValues); + } + + /** + * Collects all values from a one-column MatrixBlock into a set. + */ + private static HashSet collectScalars(MatrixBlock block) { + HashSet ret = new HashSet<>(); + for( int i = 0; i < block.getNumRows(); i++ ) + ret.add(block.get(i, 0)); + return ret; + } + + /** + * Compares one output row as a set of scalar values. + */ + private static void assertSameRowSet(MatrixBlock block, int row, double[] expectedValues, String message) { + HashSet expected = collectExpected(expectedValues); + HashSet actual = new HashSet<>(); + for( int j = 0; j < block.getNumColumns(); j++ ) + actual.add(block.get(row, j)); + assertEquals(message + " mismatch", expected, actual); + } + + /** + * Compares one output column as a set of scalar values. + */ + private static void assertSameColumnSet(MatrixBlock block, int col, double[] expectedValues, String message) { + HashSet expected = collectExpected(expectedValues); + HashSet actual = new HashSet<>(); + for( int i = 0; i < block.getNumRows(); i++ ) + actual.add(block.get(i, col)); + assertEquals(message + " mismatch", expected, actual); + } + + /** + * Collects expected scalar values into a set. + */ + private static HashSet collectExpected(double[] values) { + HashSet ret = new HashSet<>(); + for( double value : values ) + ret.add(value); + return ret; + } + + /** + * Checks MatrixBlock dimensions and reports a readable failure. + */ + private static void assertDimensions(MatrixBlock block, int rows, int cols, String message) { + assertEquals(message + " row dimension", rows, block.getNumRows()); + assertEquals(message + " column dimension", cols, block.getNumColumns()); + } +}