diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/AggregateUnaryCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/AggregateUnaryCPInstruction.java index d0904b38fba..08fffdae0d3 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/cp/AggregateUnaryCPInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/AggregateUnaryCPInstruction.java @@ -232,7 +232,7 @@ else if( input1.getDataType().isMatrix() || input1.getDataType().isFrame() ) { } UnarySketchOperator op = (UnarySketchOperator) _optr; - MatrixBlock res = LibMatrixSketch.getUniqueValues(input, op.getDirection()); + MatrixBlock res = LibMatrixSketch.getUniqueValues(input, op.getDirection(), op.getNumThreads()); ec.releaseMatrixInput(input1.getName()); ec.setMatrixOutput(outputName, res); break; 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..3067caa2434 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,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: + if(localDedupMemorySafe) + return getUniqueValuesRowColParallel(blkIn, k); + return getUniqueValuesRowColBatchedParallel(blkIn, dir, k, maxLocalBytes); + case Row: + if(localDedupMemorySafe) + return getUniqueRowValuesParallel(blkIn, k); + return getUniqueRowValuesBatchedParallel(blkIn, dir, k, maxLocalBytes); + case Col: + if(localDedupMemorySafe) + return getUniqueColumnValuesParallel(blkIn, k); + return getUniqueColumnValuesBatchedParallel(blkIn, dir, k, maxLocalBytes); + 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, + long maxLocalBytes) { + BatchConfig config = getRowColBatchConfig(blkIn, dir, k, maxLocalBytes); + 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 = (int) Math.min((long) 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, + long maxLocalBytes) { + BatchConfig config = getBatchConfig(blkIn, dir, k, maxLocalBytes); + 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, + long maxLocalBytes) { + BatchConfig config = getBatchConfig(blkIn, dir, k, maxLocalBytes); + 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 = (int) Math.min((long) 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 = (int) Math.min((long) 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, long budget) { + int len = getPartitionLength(blkIn, dir); + long maxBatchIndexes = getMaxLocalDedupIndexes(blkIn, dir, budget); + 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); + } + + /** + * Builds the batched execution plan for RowCol. Unlike the row-wise and column-wise workers, the RowCol workers + * accumulate over their entire range and the merged set grows toward the total distinct count, so the merged worst + * case is reserved up front and only the remaining budget sizes the concurrent local sets. If not even the merged + * set fits, batching cannot help and the caller falls back to sequential execution. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @param budget budget in bytes for transient deduplication structures + * @return batch configuration, or null if batching would not use parallelism safely + */ + private static BatchConfig getRowColBatchConfig(MatrixBlock blkIn, Types.Direction dir, int k, long budget) { + int len = getPartitionLength(blkIn, dir); + long cellsPerIndex = getCellsPerIndex(blkIn, dir); + if(cellsPerIndex <= 0 || cellsPerIndex > Long.MAX_VALUE / PAR_UNIQUE_BYTES_PER_CELL) + return null; + + // reserve the worst case (all cells distinct) for the merged set + long numCells = (long) len * cellsPerIndex; + if(!fits(numCells, PAR_UNIQUE_BYTES_PER_CELL, budget)) + return null; + long localBudget = budget - numCells * PAR_UNIQUE_BYTES_PER_CELL; + + long bytesPerIndex = cellsPerIndex * PAR_UNIQUE_BYTES_PER_CELL; + int numThreads = getNumThreads(k, (int) Math.min(len, Math.max(1, localBudget / bytesPerIndex))); + if(numThreads <= 1) + return null; + + long maxIdxPerTask = localBudget / (numThreads * bytesPerIndex); + if(maxIdxPerTask < 1) + return null; + + int taskLen = (int) Math.min(Integer.MAX_VALUE, maxIdxPerTask); + 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 budget) { + long cellsPerIndex = getCellsPerIndex(blkIn, dir); + if(cellsPerIndex <= 0 || cellsPerIndex > Long.MAX_VALUE / PAR_UNIQUE_BYTES_PER_CELL) + return 0; + + long bytesPerIndex = cellsPerIndex * PAR_UNIQUE_BYTES_PER_CELL; + return budget / bytesPerIndex; + } + + /** + * Default per-operation budget for the transient deduplication structures. This uses the infrastructure-analyzer + * view of the local JVM memory, which parfor adjusts per worker, rather than the raw JVM maximum. + * + * @return budget in bytes + */ + private static long getDefaultLocalBytesBudget() { + return InfrastructureAnalyzer.getLocalMaxMemory() / PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION; + } + + /** + * Returns the number of cells scanned per partition index (columns per row, or rows per column). + * + * @param blkIn input matrix block + * @param dir unique direction + * @return number of cells per row or column + */ + private static long getCellsPerIndex(MatrixBlock blkIn, Types.Direction dir) { + return dir == Types.Direction.Col ? blkIn.getNumRows() : blkIn.getNumColumns(); + } + + /** + * Overflow-safe check for {@code count * bytesPerUnit <= budget}. + * + * @param count number of units + * @param bytesPerUnit bytes charged per unit + * @param budget budget in bytes + * @return true if the total fits the budget + */ + private static boolean fits(long count, long bytesPerUnit, long budget) { + if(count <= 0) + return true; + return count <= budget / bytesPerUnit; + } + /** + * Memory guard deciding whether the full (non-batched) parallel path fits the budget. + * + *

+ * The row-wise and column-wise workers reuse a single set that is cleared per row/column, so only one live set per + * thread is charged, i.e. {@code numThreads * cellsPerIndex * bytesPerCell + * <= budget}. Charging all indexes instead would force batched execution where it is safely avoidable. The RowCol + * workers instead accumulate over their whole range and are merged into a set that grows toward the total distinct + * count, so the all-distinct worst case is charged for the local sets and the merged set alike. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @param budget budget in bytes for transient deduplication structures + * @return true if the full parallel path is safe; false to use the batched path + */ + private static boolean isLocalDedupMemoryBudgetSafe(MatrixBlock blkIn, Types.Direction dir, int k, long budget) { + int len = getPartitionLength(blkIn, dir); + long cellsPerIndex = getCellsPerIndex(blkIn, dir); + if(len <= 0 || cellsPerIndex <= 0) + return true; + if(cellsPerIndex > Long.MAX_VALUE / PAR_UNIQUE_BYTES_PER_CELL) + return false; + + if(dir == Types.Direction.RowCol) + return fits((long) len * cellsPerIndex, 2 * PAR_UNIQUE_BYTES_PER_CELL, budget); + + int numThreads = getNumThreads(k, len); + return fits(numThreads * cellsPerIndex, PAR_UNIQUE_BYTES_PER_CELL, budget); + } + + /** + * 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/performance/matrix/UniquePerf.java b/src/test/java/org/apache/sysds/performance/matrix/UniquePerf.java new file mode 100644 index 00000000000..32775c664b7 --- /dev/null +++ b/src/test/java/org/apache/sysds/performance/matrix/UniquePerf.java @@ -0,0 +1,175 @@ +/* + * 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.performance.matrix; + +import java.util.Arrays; + +import org.apache.sysds.common.Types; +import org.apache.sysds.runtime.matrix.data.LibMatrixSketch; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; + +/** + * Runtime comparison of unique() against the k=1 baseline, across different shares of unique values, for all three + * directions (RowCol, Row, Col) and a small and a large input, at a fixed degree of parallelism. + * + *

+ * The share of unique values is defined per direction, matching what that direction deduplicates: for RowCol it is the + * number of distinct scalar values in the whole input, for Row the number of distinct values within each row, and for + * Col the number within each column. + * + *

+ * Each configuration is warmed up and then measured several times; the median is reported and the order of the k=1 and + * k=N runs is alternated between iterations to reduce ordering effects. Input generation is excluded from the + * measurements. + * + *

+ * Usage: {@code UniquePerf [numThreads] [reps]} + */ +public class UniquePerf { + private static final int WARMUP = 3; + private static final double[] SHARES = {0.01, 0.10, 0.50, 1.00}; + + /** Per-direction input shapes, chosen so that each direction has a meaningful small and large case. */ + private static final Case[] CASES = {new Case(Types.Direction.RowCol, "small", 100000, 1), + new Case(Types.Direction.RowCol, "large", 2000000, 1), new Case(Types.Direction.Row, "small", 10000, 64), + new Case(Types.Direction.Row, "large", 200000, 64), new Case(Types.Direction.Col, "small", 64, 10000), + new Case(Types.Direction.Col, "large", 64, 200000)}; + + public static void main(String[] args) { + int k = args.length > 0 ? Integer.parseInt(args[0]) : 4; + int reps = args.length > 1 ? Integer.parseInt(args[1]) : 5; + + System.out.println("# unique() runtime vs. share of unique values"); + System.out.printf( + "# threads k=%d vs. k=1 baseline, %d warmup + %d measured runs (median), JVM maxMemory=%d MiB%n", k, WARMUP, + reps, Runtime.getRuntime().maxMemory() / 1024 / 1024); + + for(Types.Direction dir : new Types.Direction[] {Types.Direction.RowCol, Types.Direction.Row, + Types.Direction.Col}) { + System.out.printf("%n## %s%n%n", name(dir)); + System.out.printf("| size | input | unique share | %s | k=1 [ms] | k=%d [ms] | speedup |%n", + distinctLabel(dir), k); + System.out.println("|---|---|---|---|---|---|---|"); + + for(Case c : CASES) { + if(c._dir != dir) + continue; + for(double share : SHARES) { + int distinct = distinctCount(dir, c._rlen, c._clen, share); + MatrixBlock in = generate(dir, c._rlen, c._clen, distinct); + + double[] res = measure(in, dir, k, reps); + System.out.printf("| %s | %dx%d | %.0f%% | %d | %.3f | %.3f | %.2f |%n", c._label, c._rlen, c._clen, + share * 100, distinct, res[0], res[1], res[0] / res[1]); + } + } + } + } + + /** + * Warms up and then measures one configuration, alternating the order of the sequential and parallel runs. + * + * @return {median sequential time, median parallel time} in milliseconds + */ + private static double[] measure(MatrixBlock in, Types.Direction dir, int k, int reps) { + for(int w = 0; w < WARMUP; w++) { + LibMatrixSketch.getUniqueValues(in, dir, 1); + LibMatrixSketch.getUniqueValues(in, dir, k); + } + + double[] seq = new double[reps]; + double[] par = new double[reps]; + for(int i = 0; i < reps; i++) { + if(i % 2 == 0) { // alternate order to reduce ordering effects + seq[i] = time(in, dir, 1); + par[i] = time(in, dir, k); + } + else { + par[i] = time(in, dir, k); + seq[i] = time(in, dir, 1); + } + } + return new double[] {median(seq), median(par)}; + } + + private static double time(MatrixBlock in, Types.Direction dir, int k) { + long t0 = System.nanoTime(); + LibMatrixSketch.getUniqueValues(in, dir, k); + return (System.nanoTime() - t0) / 1e6; + } + + private static String name(Types.Direction dir) { + return dir == Types.Direction.RowCol ? "RowCol" : dir == Types.Direction.Row ? "Row" : "Col"; + } + + private static String distinctLabel(Types.Direction dir) { + return dir == Types.Direction.RowCol ? "unique values" : dir == Types.Direction.Row ? "unique values per row" : "unique values per column"; + } + + /** + * Number of distinct values the given share corresponds to, per deduplicated unit. + */ + private static int distinctCount(Types.Direction dir, int rlen, int clen, double share) { + long unit = dir == Types.Direction.RowCol ? (long) rlen * clen : dir == Types.Direction.Row ? clen : rlen; + return (int) Math.max(1, Math.round(share * unit)); + } + + /** + * Builds an input whose per-unit distinct count matches the requested share. + */ + private static MatrixBlock generate(Types.Direction dir, int rlen, int clen, int distinct) { + MatrixBlock ret = new MatrixBlock(rlen, clen, false).allocateBlock(); + for(int i = 0; i < rlen; i++) { + for(int j = 0; j < clen; j++) { + double val; + if(dir == Types.Direction.RowCol) + val = ((long) i * clen + j) % distinct; + else if(dir == Types.Direction.Row) + val = (i % 7) * (long) distinct + (j % distinct); + else + val = (j % 7) * (long) distinct + (i % distinct); + ret.set(i, j, val); + } + } + ret.recomputeNonZeros(); + return ret; + } + + private static double median(double[] values) { + double[] sorted = values.clone(); + Arrays.sort(sorted); + return sorted[sorted.length / 2]; + } + + /** One benchmarked input shape for a given direction. */ + private static class Case { + private final Types.Direction _dir; + private final String _label; + private final int _rlen; + private final int _clen; + + private Case(Types.Direction dir, String label, int rlen, int clen) { + _dir = dir; + _label = label; + _rlen = rlen; + _clen = clen; + } + } +} diff --git a/src/test/java/org/apache/sysds/test/functions/unique/UniqueBase.java b/src/test/java/org/apache/sysds/test/functions/unique/UniqueBase.java index 6e65c01f7c9..68aed76012c 100644 --- a/src/test/java/org/apache/sysds/test/functions/unique/UniqueBase.java +++ b/src/test/java/org/apache/sysds/test/functions/unique/UniqueBase.java @@ -23,6 +23,7 @@ import org.apache.sysds.test.AutomatedTestBase; import org.apache.sysds.test.TestConfiguration; import org.apache.sysds.test.TestUtils; +import org.apache.sysds.utils.stats.InfrastructureAnalyzer; public abstract class UniqueBase extends AutomatedTestBase { @@ -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, + double epsilon) { + uniqueTest(inputMatrix, expectedMatrix, instType, epsilon, -1, true); + } + + /** + * Runs the unique script with a temporarily reduced local memory budget. The multi-threaded unique implementation + * derives its budget for thread-local deduplication from the local memory, so this makes the memory-aware batched + * path reachable from an end-to-end script run. + * + * @param localMaxMemory local memory budget in bytes, or -1 to keep the current one + */ + protected void uniqueTestConstrainedMemory(double[][] inputMatrix, double[][] expectedMatrix, + Types.ExecType instType, double epsilon, long localMaxMemory) { + uniqueTest(inputMatrix, expectedMatrix, instType, epsilon, localMaxMemory, false); + } + + private void uniqueTest(double[][] inputMatrix, double[][] expectedMatrix, Types.ExecType instType, double epsilon, + long localMaxMemory, boolean orderedComparison) { Types.ExecMode platformOld = setExecMode(instType); + long localMaxMemoryOld = InfrastructureAnalyzer.getLocalMaxMemory(); try { + if(localMaxMemory > 0) + InfrastructureAnalyzer.setLocalMaxMemory(localMaxMemory); + loadTestConfiguration(getTestConfiguration(getTestName())); String HOME = SCRIPT_DIR + getTestDir(); fullDMLScriptName = HOME + getTestName() + ".dml"; @@ -52,9 +84,13 @@ protected void uniqueTest(double[][] inputMatrix, double[][] expectedMatrix, runTest(true, false, null, -1); writeExpectedMatrix("A", expectedMatrix); - compareResultsRowsOutOfOrder(epsilon); + if(orderedComparison) + compareResults(epsilon); + else + compareResultsRowsOutOfOrder(epsilon); } finally { + InfrastructureAnalyzer.setLocalMaxMemory(localMaxMemoryOld); rtplatform = platformOld; } } diff --git a/src/test/java/org/apache/sysds/test/functions/unique/UniqueBatchedPathTest.java b/src/test/java/org/apache/sysds/test/functions/unique/UniqueBatchedPathTest.java new file mode 100644 index 00000000000..d0528081ec9 --- /dev/null +++ b/src/test/java/org/apache/sysds/test/functions/unique/UniqueBatchedPathTest.java @@ -0,0 +1,77 @@ +/* + * 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 static org.junit.Assert.assertEquals; + +import org.apache.sysds.common.Types; +import org.apache.sysds.runtime.matrix.data.LibMatrixSketch; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.junit.Test; + +/** + * Covers the batched row-wise and column-wise unique paths, which the script-level tests cannot reach. + * + *

+ * The row-wise and column-wise workers charge only one live set per thread, so batching is selected only when + * {@code numThreads * cellsPerIndex * 64} exceeds the budget. Reaching that from an end-to-end run would require the + * budget to fall into a narrow window that depends on the number of threads of the executing machine, which is not + * deterministic across CI runners. The batched RowCol path does not have this problem and is covered end-to-end by + * {@link UniqueRowCol}. + */ +public class UniqueBatchedPathTest { + + @Test + public void testBatchedRowMatchesBaseline() { + MatrixBlock in = new MatrixBlock(400, 64, false).allocateBlock(); + for(int i = 0; i < in.getNumRows(); i++) + for(int j = 0; j < in.getNumColumns(); j++) + in.set(i, j, (i + j) % 8); + in.recomputeNonZeros(); + + // the full parallel path needs numThreads * clen * 64 bytes, so a budget below that batches + assertEquivalent(in, Types.Direction.Row, 4, 4 * 64 * 64 / 2); + } + + @Test + public void testBatchedColMatchesBaseline() { + MatrixBlock in = new MatrixBlock(64, 400, false).allocateBlock(); + for(int j = 0; j < in.getNumColumns(); j++) + for(int i = 0; i < in.getNumRows(); i++) + in.set(i, j, (i + j) % 8); + in.recomputeNonZeros(); + + assertEquivalent(in, Types.Direction.Col, 4, 4 * 64 * 64 / 2); + } + + /** + * Checks that the batched result is identical to the single-threaded baseline. + */ + private static void assertEquivalent(MatrixBlock in, Types.Direction dir, int k, long maxLocalBytes) { + MatrixBlock expected = LibMatrixSketch.getUniqueValues(in, dir); + MatrixBlock actual = LibMatrixSketch.getUniqueValues(in, dir, k, maxLocalBytes); + + assertEquals("number of rows", expected.getNumRows(), actual.getNumRows()); + assertEquals("number of columns", expected.getNumColumns(), actual.getNumColumns()); + for(int i = 0; i < expected.getNumRows(); i++) + for(int j = 0; j < expected.getNumColumns(); j++) + assertEquals("value at (" + i + ", " + j + ")", expected.get(i, j), actual.get(i, j), 0); + } +} diff --git a/src/test/java/org/apache/sysds/test/functions/unique/UniqueCol.java b/src/test/java/org/apache/sysds/test/functions/unique/UniqueCol.java new file mode 100644 index 00000000000..d2baef1d60e --- /dev/null +++ b/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() { + int rlen = 64, clen = 400; + 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; + } + uniqueTestConstrainedMemory(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0, 16 * 1024 * 1024); + } + + /** + * Sparse counterpart of the multi-threaded case: only every eighth column is populated, so the input is read in + * sparse format. Every column still holds a single distinct value, either its filler or zero, so the expected + * result stays one row. + */ + @Test + public void testSparseMultiThreadedCP() { + int rlen = 64, clen = 400; + double[][] inputMatrix = new double[rlen][clen]; + double[][] expectedMatrix = new double[1][clen]; + for(int j = 0; j < clen; j++) { + double value = (j % 8 == 0) ? j + 1 : 0; + for(int i = 0; i < rlen; i++) + inputMatrix[i][j] = value; + expectedMatrix[0][j] = value; + } + uniqueTestOrdered(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0); + } +} diff --git a/src/test/java/org/apache/sysds/test/functions/unique/UniqueRow.java b/src/test/java/org/apache/sysds/test/functions/unique/UniqueRow.java index fda9aa4a3c0..93150f1fcda 100644 --- a/src/test/java/org/apache/sysds/test/functions/unique/UniqueRow.java +++ b/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() { + int rlen = 400, clen = 64; + double[][] inputMatrix = new double[rlen][clen]; + double[][] expectedMatrix = new double[rlen][1]; + for(int i = 0; i < rlen; i++) { + double value = (i % 8 == 0) ? i + 1 : 0; + for(int j = 0; j < clen; j++) + inputMatrix[i][j] = value; + expectedMatrix[i][0] = value; + } + uniqueTestOrdered(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0); + } + + private static double[][] constantRows(int rlen, int clen) { + double[][] ret = new double[rlen][clen]; + for(int i = 0; i < rlen; i++) + for(int j = 0; j < clen; j++) + ret[i][j] = i; + return ret; + } + + private static double[][] expectedConstantRows(int rlen) { + double[][] ret = new double[rlen][1]; + for(int i = 0; i < rlen; i++) + ret[i][0] = i; + return ret; + } } diff --git a/src/test/java/org/apache/sysds/test/functions/unique/UniqueRowCol.java b/src/test/java/org/apache/sysds/test/functions/unique/UniqueRowCol.java index 7f6ad0ff56e..dc5a3bd73d7 100644 --- a/src/test/java/org/apache/sysds/test/functions/unique/UniqueRowCol.java +++ b/src/test/java/org/apache/sysds/test/functions/unique/UniqueRowCol.java @@ -70,4 +70,60 @@ public void testWideSmallCP() { double[][] expectedMatrix = {{1,6,9,4,2,0}}; uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0); } + + /** + * Large enough to take the multi-threaded path. The result is a single column, so the comparison is unaffected by + * the order in which the merged hash set is iterated. + */ + @Test + public void testMultiThreadedCP() { + uniqueTest(cyclicValues(500, 40, 37), expectedCyclicValues(37), Types.ExecType.CP, 0.0); + } + + /** + * Same input, but with a local memory budget that is too small to hold the thread-local sets and the merged set at + * once. This selects the batched path from an end-to-end script run. + */ + @Test + public void testBatchedCP() { + uniqueTestConstrainedMemory(cyclicValues(500, 40, 37), expectedCyclicValues(37), Types.ExecType.CP, 0.0, + 8 * 1024 * 1024); + } + + /** + * Sparse counterpart of the multi-threaded case: only every tenth cell is populated, so the input is read in sparse + * format. The result is a single column holding zero plus the fillers. + */ + @Test + public void testSparseMultiThreadedCP() { + int rlen = 500, clen = 40, distinct = 36; + double[][] inputMatrix = new double[rlen][clen]; + for(int i = 0; i < rlen; i++) + for(int j = 0; j < clen; j++) { + long pos = (long) i * clen + j; + inputMatrix[i][j] = (pos % 10 == 0) ? (pos / 10) % distinct + 1 : 0; + } + + // zero plus the fillers 1..distinct + double[][] expectedMatrix = new double[distinct + 1][1]; + for(int i = 0; i <= distinct; i++) + expectedMatrix[i][0] = i; + + uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0); + } + + private static double[][] cyclicValues(int rlen, int clen, int distinct) { + double[][] ret = new double[rlen][clen]; + for(int i = 0; i < rlen; i++) + for(int j = 0; j < clen; j++) + ret[i][j] = ((long) i * clen + j) % distinct; + return ret; + } + + private static double[][] expectedCyclicValues(int distinct) { + double[][] ret = new double[distinct][1]; + for(int i = 0; i < distinct; i++) + ret[i][0] = i; + return ret; + } } diff --git a/src/test/scripts/functions/unique/uniqueCol.dml b/src/test/scripts/functions/unique/uniqueCol.dml new file mode 100644 index 00000000000..3ea61e27127 --- /dev/null +++ b/src/test/scripts/functions/unique/uniqueCol.dml @@ -0,0 +1,24 @@ +#------------------------------------------------------------- +# +# 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. +# +#------------------------------------------------------------- + +input = read($1); +res = unique(input, dir="c"); +write(res, $2, format="text");