From ea72223cecca7abc23d5e0e53381ecf164d98c1c Mon Sep 17 00:00:00 2001 From: Xuanyan Wang <54443787+shieru1214@users.noreply.github.com> Date: Mon, 8 Jun 2026 16:31:25 +0200 Subject: [PATCH 1/5] Add multi-threaded unique in LibMatrixSketch --- .../runtime/matrix/data/LibMatrixSketch.java | 583 ++++++++++++++++-- .../LibMatrixSketchUniqueParallelTest.java | 167 +++++ 2 files changed, 691 insertions(+), 59 deletions(-) create mode 100644 src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java 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..c1e4aeb65ce 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,554 @@ package org.apache.sysds.runtime.matrix.data; -import org.apache.sysds.common.Types; - +import java.util.ArrayList; +import java.util.Collection; import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; + +import org.apache.commons.lang3.NotImplementedException; +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; + /** + * Computes unique values, rows, or columns 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, rows, or columns + */ 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, rows, or columns. Parallel execution is used only for + * sufficiently large inputs with k > 1; otherwise the existing sequential path is used. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @return matrix block containing unique values, rows, or columns + */ + public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir, int k) { + // similar to R's unique, this operation takes a matrix and computes the unique values + // (or rows in case of multiple column inputs) + if( !satisfiesMultiThreadingConstraints(blkIn, dir, k) ) + return getUniqueValuesSequential(blkIn, dir); + + switch(dir) { + case RowCol: + return getUniqueValuesRowColParallel(blkIn, k); + case Row: + return getUniqueRowsParallel(blkIn, k); + case Col: + return getUniqueColumnsParallel(blkIn, k); + default: + throw new IllegalArgumentException("Unrecognized direction: " + dir); + } + } + + /** + * Single-threaded baseline implementation for all unique directions. + * This preserves the original RowCol and Row behavior and adds the sequential Col path. + * + * @param blkIn input matrix block + * @param dir unique direction + * @return matrix block containing unique values, rows, or columns + */ + 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: { + case RowCol: + if( clen != 1 ) + throw new NotImplementedException("Unique only support single-column vectors yet"); + // TODO optimize for dense/sparse/compressed (once multi-column support added) + // obtain set of unique items (dense input vector) HashSet hashSet = new HashSet<>(); for( int i=0; i hashSet = new HashSet<>(); - int clen2 = 0; - for( int i=0; i retainedRows = new ArrayList<>(); + + for (int i=0; i hashSet = new HashSet<>(); - int rlen2 = 0; - for( int j=0; j retainedColumns = new LinkedHashMap<>(); + for( int j = 0; j < clen; j++ ) { + double[] currentColumn = copyColumn(blkIn, j); + retainedColumns.putIfAbsent(new ColKey(currentColumn), currentColumn); } + blkOut = createColumnOutput(retainedColumns.values(), rlen); break; - } + default: throw new IllegalArgumentException("Unrecognized direction: " + dir); } return blkOut; } + + /** + * Parallel unique for single-column vectors. Rows are split into balanced partitions, + * each task builds a local set, and the caller merges all local sets afterwards. + * + * @param blkIn input single-column matrix block + * @param k requested degree of parallelism + * @return one-column matrix block containing the unique values + */ + private static MatrixBlock getUniqueValuesRowColParallel(MatrixBlock blkIn, int k) { + if( blkIn.getNumColumns() != 1 ) + throw new NotImplementedException("Unique only support single-column vectors yet"); + + ExecutorService pool = CommonThreadPool.get(getNumThreads(k, blkIn.getNumRows())); + try { + ArrayList tasks = new ArrayList<>(); + for( int[] range : getBalancedRanges(blkIn.getNumRows(), k) ) + tasks.add(new UniqueValueTask(blkIn, range[0], range[1])); + + // Merge after local deduplication to avoid a shared synchronized set in the workers. + HashSet hashSet = new HashSet<>(); + List>> rtasks = pool.invokeAll(tasks); + for( Future> task : rtasks ) + hashSet.addAll(task.get()); + + return createRowColOutput(hashSet); + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + pool.shutdown(); + } + } + + /** + * Parallel unique rows. Each worker deduplicates its row partition locally, and the + * final merge scans the partition results in input order to keep the first occurrence. + * + * @param blkIn input matrix block + * @param k requested degree of parallelism + * @return matrix block containing exact unique rows + */ + private static MatrixBlock getUniqueRowsParallel(MatrixBlock blkIn, int k) { + ExecutorService pool = CommonThreadPool.get(getNumThreads(k, blkIn.getNumRows())); + try { + ArrayList tasks = new ArrayList<>(); + for( int[] range : getBalancedRanges(blkIn.getNumRows(), k) ) + tasks.add(new UniqueRowTask(blkIn, range[0], range[1])); + + // Global merge is intentionally single-threaded and ordered for correctness. + LinkedHashMap retainedRows = new LinkedHashMap<>(); + List>> rtasks = pool.invokeAll(tasks); + for( Future> task : rtasks ) + for( java.util.Map.Entry entry : task.get().entrySet() ) + retainedRows.putIfAbsent(entry.getKey(), entry.getValue()); + + return createRowOutput(retainedRows.values(), blkIn.getNumColumns()); + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + pool.shutdown(); + } + } + + /** + * Parallel unique columns. Each worker deduplicates a column partition locally, and + * the final merge scans partitions from left to right to keep the first occurrence. + * + * @param blkIn input matrix block + * @param k requested degree of parallelism + * @return matrix block containing exact unique columns + */ + private static MatrixBlock getUniqueColumnsParallel(MatrixBlock blkIn, int k) { + ExecutorService pool = CommonThreadPool.get(getNumThreads(k, blkIn.getNumColumns())); + try { + ArrayList tasks = new ArrayList<>(); + for( int[] range : getBalancedRanges(blkIn.getNumColumns(), k) ) + tasks.add(new UniqueColumnTask(blkIn, range[0], range[1])); + + // Global merge is intentionally single-threaded and ordered for correctness. + LinkedHashMap retainedColumns = new LinkedHashMap<>(); + List>> rtasks = pool.invokeAll(tasks); + for( Future> task : rtasks ) + for( java.util.Map.Entry entry : task.get().entrySet() ) + retainedColumns.putIfAbsent(entry.getKey(), entry.getValue()); + + return createColumnOutput(retainedColumns.values(), blkIn.getNumRows()); + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + pool.shutdown(); + } + } + + /** + * 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: + 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)); + } + + /** + * Copies one row into an immutable key/value array for safe HashMap storage. + * + * @param blkIn input matrix block + * @param row row index to copy + * @return copied row values + */ + private static double[] copyRow(MatrixBlock blkIn, int row) { + int clen = blkIn.getNumColumns(); + double[] ret = new double[clen]; + for( int j = 0; j < clen; j++ ) + ret[j] = blkIn.get(row, j); + return ret; + } + + /** + * Copies one column into an immutable key/value array for safe HashMap storage. + * + * @param blkIn input matrix block + * @param col column index to copy + * @return copied column values + */ + private static double[] copyColumn(MatrixBlock blkIn, int col) { + int rlen = blkIn.getNumRows(); + double[] ret = new double[rlen]; + for( int i = 0; i < rlen; i++ ) + ret[i] = blkIn.get(i, col); + return ret; + } + + /** + * 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; + } + + /** + * Allocates and fills a MatrixBlock from retained row copies. + * + * @param rows unique row copies + * @param clen number of columns in the output + * @return matrix block containing the unique rows + */ + private static MatrixBlock createRowOutput(Collection rows, int clen) { + MatrixBlock blkOut = allocateOutputBlock(rows.size(), clen); + int i = 0; + for( double[] row : rows ) { + for( int j = 0; j < clen; j++ ) + blkOut.set(i, j, row[j]); + i++; + } + blkOut.recomputeNonZeros(); + blkOut.examSparsity(); + return blkOut; + } + + /** + * Allocates and fills a MatrixBlock from retained column copies. + * + * @param columns unique column copies + * @param rlen number of rows in the output + * @return matrix block containing the unique columns + */ + private static MatrixBlock createColumnOutput(Collection columns, int rlen) { + MatrixBlock blkOut = allocateOutputBlock(rlen, columns.size()); + int j = 0; + for( double[] column : columns ) { + for( int i = 0; i < rlen; i++ ) + blkOut.set(i, j, column[i]); + j++; + } + 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; + } + + /** + * Computes a stable content hash for row and column keys using numeric equality. + * + * @param values copied row or column values + * @return content hash code + */ + private static int hashValues(double[] values) { + int ret = 1; + for( double value : values ) + ret = 31 * ret + (value == 0 ? 0 : Double.hashCode(value)); + return ret; + } + + /** + * Compares copied row or column values with exact numeric equality. + * + * @param left first copied value array + * @param right second copied value array + * @return true if the arrays represent the same row or column + */ + private static boolean equalValues(double[] left, double[] right) { + if( left.length != right.length ) + return false; + for( int i = 0; i < left.length; i++ ) + if( left[i] != right[i] && Double.doubleToLongBits(left[i]) != Double.doubleToLongBits(right[i]) ) + return false; + return true; + } + + /** + * Worker that deduplicates a row partition of a single-column vector 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++ ) + ret.add(_blkIn.get(i, 0)); + return ret; + } + } + + /** + * Worker that deduplicates copied rows within a row partition before global merge. + */ + private static class UniqueRowTask implements Callable> { + private final MatrixBlock _blkIn; + private final int _rl; + private final int _ru; + + private UniqueRowTask(MatrixBlock blkIn, int rl, int ru) { + _blkIn = blkIn; + _rl = rl; + _ru = ru; + } + + @Override + public LinkedHashMap call() { + LinkedHashMap ret = new LinkedHashMap<>(); + for( int i = _rl; i < _ru; i++ ) { + double[] row = copyRow(_blkIn, i); + ret.putIfAbsent(new RowKey(row), row); + } + return ret; + } + } + + /** + * Worker that deduplicates copied columns within a column partition before global merge. + */ + private static class UniqueColumnTask implements Callable> { + private final MatrixBlock _blkIn; + private final int _cl; + private final int _cu; + + private UniqueColumnTask(MatrixBlock blkIn, int cl, int cu) { + _blkIn = blkIn; + _cl = cl; + _cu = cu; + } + + @Override + public LinkedHashMap call() { + LinkedHashMap ret = new LinkedHashMap<>(); + for( int j = _cl; j < _cu; j++ ) { + double[] column = copyColumn(_blkIn, j); + ret.putIfAbsent(new ColKey(column), column); + } + return ret; + } + } + + /** + * Content-based key for copied rows. The referenced array is never mutated after + * construction, so it is safe to reuse the copy as both key contents and output data. + */ + private static class RowKey { + private final double[] _values; + + private RowKey(double[] values) { + _values = values; + } + + @Override + public int hashCode() { + return hashValues(_values); + } + + @Override + public boolean equals(Object obj) { + return obj instanceof RowKey && equalValues(_values, ((RowKey) obj)._values); + } + } + + /** + * Content-based key for copied columns. The referenced array is never mutated after + * construction, so it is safe to reuse the copy as both key contents and output data. + */ + private static class ColKey { + private final double[] _values; + + private ColKey(double[] values) { + _values = values; + } + + @Override + public int hashCode() { + return hashValues(_values); + } + + @Override + public boolean equals(Object obj) { + return obj instanceof ColKey && equalValues(_values, ((ColKey) obj)._values); + } + } } 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..f032d7604b9 --- /dev/null +++ b/src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java @@ -0,0 +1,167 @@ +/* + * 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 java.util.HashSet; + +import org.apache.sysds.common.Types; + +/** + * Small standalone parallel unique test for LibMatrixSketch with k=1 and k>1. + * + * This class intentionally avoids JUnit so it can be run directly from an IDE + * or from a full SystemDS checkout with the normal project classpath. + */ +public class LibMatrixSketchUniqueParallelTest { + public static void main(String[] args) { + testRowColUniqueMatchesBaseline(); + testRowUniqueMatchesBaselineAndExpectedRows(); + testColumnUniqueMatchesBaselineAndExpectedColumns(); + System.out.println("LibMatrixSketch unique parallel tests passed."); + } + + /** + * Checks RowCol unique on a large single-column vector so the k=4 path is used. + */ + private static void testRowColUniqueMatchesBaseline() { + MatrixBlock input = new MatrixBlock(20000, 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 parallel dimensions"); + assertSameScalarSet(baseline, parallel, "RowCol baseline vs parallel"); + } + + /** + * Checks Row unique with repeated rows. The expected output is ordered by first + * occurrence, which also verifies the ordered global merge across row partitions. + */ + private static void testRowUniqueMatchesBaselineAndExpectedRows() { + MatrixBlock input = new MatrixBlock(12000, 2, false).allocateBlock(); + for( int i = 0; i < input.getNumRows(); i++ ) { + int pattern = i % 4; + input.set(i, 0, pattern); + input.set(i, 1, pattern * 10); + } + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row); + MatrixBlock parallel = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row, 4); + MatrixBlock expected = matrix(new double[][] { + {0, 0}, + {1, 10}, + {2, 20}, + {3, 30} + }); + + assertBlockEquals(expected, baseline, "Row expected vs baseline"); + assertBlockEquals(expected, parallel, "Row expected vs parallel"); + } + + /** + * Checks Col unique with repeated columns. The expected output shape is + * original_num_rows x number_of_unique_columns. + */ + private static void testColumnUniqueMatchesBaselineAndExpectedColumns() { + MatrixBlock input = new MatrixBlock(4, 5000, false).allocateBlock(); + double[][] uniqueColumns = new double[][] { + {1, 2, 3, 4}, + {5, 6, 7, 8}, + {9, 10, 11, 12} + }; + + for( int j = 0; j < input.getNumColumns(); j++ ) { + double[] column = uniqueColumns[j % uniqueColumns.length]; + for( int i = 0; i < input.getNumRows(); i++ ) + input.set(i, j, column[i]); + } + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col); + MatrixBlock parallel = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col, 4); + MatrixBlock expected = matrix(new double[][] { + {1, 5, 9}, + {2, 6, 10}, + {3, 7, 11}, + {4, 8, 12} + }); + + assertBlockEquals(expected, baseline, "Col expected vs baseline"); + assertBlockEquals(expected, parallel, "Col expected vs parallel"); + } + + /** + * Builds a dense MatrixBlock from a plain two-dimensional Java array. + */ + private static MatrixBlock matrix(double[][] values) { + MatrixBlock ret = new MatrixBlock(values.length, values[0].length, false).allocateBlock(); + for( int i = 0; i < values.length; i++ ) + for( int j = 0; j < values[i].length; j++ ) + ret.set(i, j, values[i][j]); + ret.recomputeNonZeros(); + return ret; + } + + /** + * 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++ ) + if( expected.get(i, j) != actual.get(i, j) ) + throw new AssertionError(message + " mismatch at (" + i + ", " + j + "): expected " + + expected.get(i, j) + " but found " + actual.get(i, j)); + } + + /** + * Compares RowCol output as a set because the scalar unique path is hash-set based. + */ + 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); + if( !expectedValues.equals(actualValues) ) + throw new AssertionError(message + " mismatch: expected " + expectedValues + " but found " + 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; + } + + /** + * Checks MatrixBlock dimensions and reports a readable failure. + */ + private static void assertDimensions(MatrixBlock block, int rows, int cols, String message) { + if( block.getNumRows() != rows || block.getNumColumns() != cols ) + throw new AssertionError(message + " dimensions mismatch: expected " + rows + "x" + cols + + " but found " + block.getNumRows() + "x" + block.getNumColumns()); + } +} From e9fef4da10a60d418f9de1df1f4c50be3b76d698 Mon Sep 17 00:00:00 2001 From: Xuanyan Wang <54443787+shieru1214@users.noreply.github.com> Date: Mon, 8 Jun 2026 17:01:57 +0200 Subject: [PATCH 2/5] Refine multi-threaded unique memory handling --- .../runtime/matrix/data/LibMatrixSketch.java | 67 ++++++++++++++----- .../test/functions/unique/UniqueBase.java | 2 +- .../test/functions/unique/UniqueRow.java | 13 ++-- 3 files changed, 59 insertions(+), 23 deletions(-) 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 c1e4aeb65ce..be99496b201 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 @@ -37,6 +37,7 @@ public class LibMatrixSketch { private static final long PAR_UNIQUE_NUMCELL_THRESHOLD = 1024 * 16; + private static final long PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION = 4; /** * Computes unique values, rows, or columns with the original single-threaded behavior. @@ -187,17 +188,22 @@ private static MatrixBlock getUniqueValuesRowColParallel(MatrixBlock blkIn, int if( blkIn.getNumColumns() != 1 ) throw new NotImplementedException("Unique only support single-column vectors yet"); - ExecutorService pool = CommonThreadPool.get(getNumThreads(k, blkIn.getNumRows())); + int numThreads = getNumThreads(k, blkIn.getNumRows()); + ExecutorService pool = CommonThreadPool.get(numThreads); try { ArrayList tasks = new ArrayList<>(); - for( int[] range : getBalancedRanges(blkIn.getNumRows(), k) ) + for( int[] range : getBalancedRanges(blkIn.getNumRows(), numThreads) ) tasks.add(new UniqueValueTask(blkIn, range[0], range[1])); // Merge after local deduplication to avoid a shared synchronized set in the workers. HashSet hashSet = new HashSet<>(); List>> rtasks = pool.invokeAll(tasks); - for( Future> task : rtasks ) - hashSet.addAll(task.get()); + 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); } @@ -218,18 +224,23 @@ private static MatrixBlock getUniqueValuesRowColParallel(MatrixBlock blkIn, int * @return matrix block containing exact unique rows */ private static MatrixBlock getUniqueRowsParallel(MatrixBlock blkIn, int k) { - ExecutorService pool = CommonThreadPool.get(getNumThreads(k, blkIn.getNumRows())); + int numThreads = getNumThreads(k, blkIn.getNumRows()); + ExecutorService pool = CommonThreadPool.get(numThreads); try { ArrayList tasks = new ArrayList<>(); - for( int[] range : getBalancedRanges(blkIn.getNumRows(), k) ) + for( int[] range : getBalancedRanges(blkIn.getNumRows(), numThreads) ) tasks.add(new UniqueRowTask(blkIn, range[0], range[1])); // Global merge is intentionally single-threaded and ordered for correctness. LinkedHashMap retainedRows = new LinkedHashMap<>(); List>> rtasks = pool.invokeAll(tasks); - for( Future> task : rtasks ) - for( java.util.Map.Entry entry : task.get().entrySet() ) + for( int i = 0; i < rtasks.size(); i++ ) { + LinkedHashMap localRows = rtasks.get(i).get(); + for( java.util.Map.Entry entry : localRows.entrySet() ) retainedRows.putIfAbsent(entry.getKey(), entry.getValue()); + localRows.clear(); + rtasks.set(i, null); + } return createRowOutput(retainedRows.values(), blkIn.getNumColumns()); } @@ -250,18 +261,23 @@ private static MatrixBlock getUniqueRowsParallel(MatrixBlock blkIn, int k) { * @return matrix block containing exact unique columns */ private static MatrixBlock getUniqueColumnsParallel(MatrixBlock blkIn, int k) { - ExecutorService pool = CommonThreadPool.get(getNumThreads(k, blkIn.getNumColumns())); + int numThreads = getNumThreads(k, blkIn.getNumColumns()); + ExecutorService pool = CommonThreadPool.get(numThreads); try { ArrayList tasks = new ArrayList<>(); - for( int[] range : getBalancedRanges(blkIn.getNumColumns(), k) ) + for( int[] range : getBalancedRanges(blkIn.getNumColumns(), numThreads) ) tasks.add(new UniqueColumnTask(blkIn, range[0], range[1])); // Global merge is intentionally single-threaded and ordered for correctness. LinkedHashMap retainedColumns = new LinkedHashMap<>(); List>> rtasks = pool.invokeAll(tasks); - for( Future> task : rtasks ) - for( java.util.Map.Entry entry : task.get().entrySet() ) + for( int i = 0; i < rtasks.size(); i++ ) { + LinkedHashMap localColumns = rtasks.get(i).get(); + for( java.util.Map.Entry entry : localColumns.entrySet() ) retainedColumns.putIfAbsent(entry.getKey(), entry.getValue()); + localColumns.clear(); + rtasks.set(i, null); + } return createColumnOutput(retainedColumns.values(), blkIn.getNumRows()); } @@ -287,10 +303,11 @@ private static boolean satisfiesMultiThreadingConstraints(MatrixBlock blkIn, Typ switch(dir) { case RowCol: + return blkIn.getNumRows() > 1 && isLocalDedupMemoryBudgetSafe(blkIn); case Row: - return blkIn.getNumRows() > 1; + return blkIn.getNumRows() > 1 && isLocalDedupMemoryBudgetSafe(blkIn); case Col: - return blkIn.getNumColumns() > 1; + return blkIn.getNumColumns() > 1 && isLocalDedupMemoryBudgetSafe(blkIn); default: throw new IllegalArgumentException("Unrecognized direction: " + dir); } @@ -323,6 +340,20 @@ private static int getNumThreads(int k, int len) { return Math.max(1, Math.min(k, len)); } + /** + * Conservative memory guard for parallel paths with thread-local sets or maps. + * This does not try to predict object overhead; it simply avoids starting local + * deduplication when the raw cell values already take a large fraction of the heap. + * + * @param blkIn input matrix block + * @return true if local deduplication is small enough for the parallel path + */ + private static boolean isLocalDedupMemoryBudgetSafe(MatrixBlock blkIn) { + long copiedValueBytes = ((long) blkIn.getNumRows()) * blkIn.getNumColumns() * Double.BYTES; + long maxLocalBytes = Runtime.getRuntime().maxMemory() / PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION; + return copiedValueBytes <= maxLocalBytes; + } + /** * Copies one row into an immutable key/value array for safe HashMap storage. * @@ -532,14 +563,16 @@ public LinkedHashMap call() { */ private static class RowKey { private final double[] _values; + private final int _hash; private RowKey(double[] values) { _values = values; + _hash = hashValues(values); } @Override public int hashCode() { - return hashValues(_values); + return _hash; } @Override @@ -554,14 +587,16 @@ public boolean equals(Object obj) { */ private static class ColKey { private final double[] _values; + private final int _hash; private ColKey(double[] values) { _values = values; + _hash = hashValues(values); } @Override public int hashCode() { - return hashValues(_values); + return _hash; } @Override 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..d834fe45aea 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 @@ -45,7 +45,7 @@ protected void uniqueTest(double[][] inputMatrix, double[][] expectedMatrix, loadTestConfiguration(getTestConfiguration(getTestName())); String HOME = SCRIPT_DIR + getTestDir(); fullDMLScriptName = HOME + getTestName() + ".dml"; - programArgs = new String[]{"-args", input("I"), output("A")}; + programArgs = new String[]{ "-args", input("I"), output("A")}; writeInputMatrixWithMTD("I", inputMatrix, true); 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..ee8c664efaa 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 @@ -27,6 +27,7 @@ public class UniqueRow extends UniqueBase { private final static String TEST_DIR = "functions/unique/"; private static final String TEST_CLASS_DIR = TEST_DIR + UniqueRow.class.getSimpleName() + "/"; + @Override protected String getTestName() { return TEST_NAME; @@ -51,22 +52,22 @@ public void testBaseCaseCP() { @Test public void testSkinnyCP() { - double[][] inputMatrix = {{1,1,6,9,4,2,0,9,0,0,4,4}}; - double[][] expectedMatrix = {{1,6,9,4,2,0}}; + 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 testSquareCP() { - double[][] inputMatrix = {{1, 4, 1}, {2, 5, 2}, {3, 6, 3}}; - double[][] expectedMatrix = {{1, 4},{2, 5},{3, 6}}; + double[][] inputMatrix = {{1, 2, 3}, {4, 5, 6}, {1, 2, 3}}; + double[][] expectedMatrix = {{1, 2, 3},{4, 5, 6}}; uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0); } @Test public void testWideCP() { - double[][] inputMatrix = {{1,7,1},{2,8,2},{3,9,3},{4,10,4},{5,11,5},{6,12,6}}; - double[][] expectedMatrix = {{1,7},{2,8},{3,9},{4,10},{5,11},{6,12}}; + double[][] inputMatrix = {{1, 2, 3, 4, 5, 6}, {7, 8, 9, 10, 11, 12}, {1, 2, 3, 4, 5, 6}}; + double[][] expectedMatrix = {{1, 2, 3, 4, 5, 6}, {7, 8, 9, 10, 11, 12}}; uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0); } From 46882fcb54fbadb54ca2851f37b8455ad6f9901e Mon Sep 17 00:00:00 2001 From: Xuanyan Wang <54443787+shieru1214@users.noreply.github.com> Date: Mon, 29 Jun 2026 14:36:53 +0200 Subject: [PATCH 3/5] Account for object overhead in parallel unique memory guard --- .../sysds/runtime/matrix/data/LibMatrixSketch.java | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) 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 be99496b201..160e6552feb 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 @@ -38,6 +38,7 @@ 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 = 2; /** * Computes unique values, rows, or columns with the original single-threaded behavior. @@ -342,14 +343,18 @@ private static int getNumThreads(int k, int len) { /** * Conservative memory guard for parallel paths with thread-local sets or maps. - * This does not try to predict object overhead; it simply avoids starting local - * deduplication when the raw cell values already take a large fraction of the heap. + * The estimate includes a small overhead factor for key and map objects, so the + * parallel path is avoided before local deduplication becomes too memory-heavy. * * @param blkIn input matrix block * @return true if local deduplication is small enough for the parallel path */ private static boolean isLocalDedupMemoryBudgetSafe(MatrixBlock blkIn) { - long copiedValueBytes = ((long) blkIn.getNumRows()) * blkIn.getNumColumns() * Double.BYTES; + long numCells = ((long) blkIn.getNumRows()) * blkIn.getNumColumns(); + if( numCells > Long.MAX_VALUE / Double.BYTES / PAR_UNIQUE_LOCAL_BYTES_OVERHEAD ) + return false; + + long copiedValueBytes = numCells * Double.BYTES * PAR_UNIQUE_LOCAL_BYTES_OVERHEAD; long maxLocalBytes = Runtime.getRuntime().maxMemory() / PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION; return copiedValueBytes <= maxLocalBytes; } From 199a2476ef8cd719b027f43eb655a953c32138b4 Mon Sep 17 00:00:00 2001 From: Xuanyan Wang <54443787+shieru1214@users.noreply.github.com> Date: Sun, 12 Jul 2026 00:29:34 +0200 Subject: [PATCH 4/5] Add batched parallel unique handling --- .../runtime/matrix/data/LibMatrixSketch.java | 253 ++++++++++++++++-- .../LibMatrixSketchUniqueParallelTest.java | 101 ++++++- 2 files changed, 317 insertions(+), 37 deletions(-) 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 160e6552feb..a931fe0b2a3 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 @@ -53,8 +53,8 @@ public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir } /** - * Computes unique values, rows, or columns. Parallel execution is used only for - * sufficiently large inputs with k > 1; otherwise the existing sequential path is used. + * Computes unique values, rows, or columns. 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 @@ -67,13 +67,20 @@ public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir if( !satisfiesMultiThreadingConstraints(blkIn, dir, k) ) return getUniqueValuesSequential(blkIn, dir); + boolean localDedupMemorySafe = isLocalDedupMemoryBudgetSafe(blkIn, dir); switch(dir) { case RowCol: - return getUniqueValuesRowColParallel(blkIn, k); + return localDedupMemorySafe ? + getUniqueValuesRowColParallel(blkIn, k) : + getUniqueValuesRowColBatchedParallel(blkIn, dir, k); case Row: - return getUniqueRowsParallel(blkIn, k); + return localDedupMemorySafe ? + getUniqueRowsParallel(blkIn, k) : + getUniqueRowsBatchedParallel(blkIn, dir, k); case Col: - return getUniqueColumnsParallel(blkIn, k); + return localDedupMemorySafe ? + getUniqueColumnsParallel(blkIn, k) : + getUniqueColumnsBatchedParallel(blkIn, dir, k); default: throw new IllegalArgumentException("Unrecognized direction: " + dir); } @@ -196,7 +203,7 @@ private static MatrixBlock getUniqueValuesRowColParallel(MatrixBlock blkIn, int for( int[] range : getBalancedRanges(blkIn.getNumRows(), numThreads) ) tasks.add(new UniqueValueTask(blkIn, range[0], range[1])); - // Merge after local deduplication to avoid a shared synchronized set in the workers. + // Merge local sets after the workers complete. HashSet hashSet = new HashSet<>(); List>> rtasks = pool.invokeAll(tasks); for( int i = 0; i < rtasks.size(); i++ ) { @@ -232,7 +239,7 @@ private static MatrixBlock getUniqueRowsParallel(MatrixBlock blkIn, int k) { for( int[] range : getBalancedRanges(blkIn.getNumRows(), numThreads) ) tasks.add(new UniqueRowTask(blkIn, range[0], range[1])); - // Global merge is intentionally single-threaded and ordered for correctness. + // Keep the merge ordered to preserve first occurrences. LinkedHashMap retainedRows = new LinkedHashMap<>(); List>> rtasks = pool.invokeAll(tasks); for( int i = 0; i < rtasks.size(); i++ ) { @@ -269,7 +276,7 @@ private static MatrixBlock getUniqueColumnsParallel(MatrixBlock blkIn, int k) { for( int[] range : getBalancedRanges(blkIn.getNumColumns(), numThreads) ) tasks.add(new UniqueColumnTask(blkIn, range[0], range[1])); - // Global merge is intentionally single-threaded and ordered for correctness. + // Keep the merge ordered to preserve first occurrences. LinkedHashMap retainedColumns = new LinkedHashMap<>(); List>> rtasks = pool.invokeAll(tasks); for( int i = 0; i < rtasks.size(); i++ ) { @@ -290,6 +297,140 @@ private static MatrixBlock getUniqueColumnsParallel(MatrixBlock blkIn, int k) { } } + /** + * Batched parallel unique for single-column vectors. + * + * @param blkIn input single-column 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) { + if( blkIn.getNumColumns() != 1 ) + throw new NotImplementedException("Unique only support single-column vectors yet"); + + 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 unique rows. Partitions are merged in row order. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @return matrix block containing exact unique rows + */ + private static MatrixBlock getUniqueRowsBatchedParallel(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 { + LinkedHashMap retainedRows = new LinkedHashMap<>(); + 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 UniqueRowTask(blkIn, pos, end)); + pos = end; + } + + List>> rtasks = pool.invokeAll(tasks); + for( int i = 0; i < rtasks.size(); i++ ) { + LinkedHashMap localRows = rtasks.get(i).get(); + for( java.util.Map.Entry entry : localRows.entrySet() ) + retainedRows.putIfAbsent(entry.getKey(), entry.getValue()); + localRows.clear(); + rtasks.set(i, null); + } + } + + return createRowOutput(retainedRows.values(), blkIn.getNumColumns()); + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + pool.shutdown(); + } + } + + /** + * Batched parallel unique columns. Column ranges are merged from left to right. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @return matrix block containing exact unique columns + */ + private static MatrixBlock getUniqueColumnsBatchedParallel(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 { + LinkedHashMap retainedColumns = new LinkedHashMap<>(); + 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 UniqueColumnTask(blkIn, pos, end)); + pos = end; + } + + List>> rtasks = pool.invokeAll(tasks); + for( int i = 0; i < rtasks.size(); i++ ) { + LinkedHashMap localColumns = rtasks.get(i).get(); + for( java.util.Map.Entry entry : localColumns.entrySet() ) + retainedColumns.putIfAbsent(entry.getKey(), entry.getValue()); + localColumns.clear(); + rtasks.set(i, null); + } + } + + return createColumnOutput(retainedColumns.values(), blkIn.getNumRows()); + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + pool.shutdown(); + } + } + /** * Decides whether the input is large enough to justify local deduplication tasks. * @@ -304,11 +445,11 @@ private static boolean satisfiesMultiThreadingConstraints(MatrixBlock blkIn, Typ switch(dir) { case RowCol: - return blkIn.getNumRows() > 1 && isLocalDedupMemoryBudgetSafe(blkIn); + return blkIn.getNumRows() > 1; case Row: - return blkIn.getNumRows() > 1 && isLocalDedupMemoryBudgetSafe(blkIn); + return blkIn.getNumRows() > 1; case Col: - return blkIn.getNumColumns() > 1 && isLocalDedupMemoryBudgetSafe(blkIn); + return blkIn.getNumColumns() > 1; default: throw new IllegalArgumentException("Unrecognized direction: " + dir); } @@ -342,25 +483,70 @@ private static int getNumThreads(int k, int len) { } /** - * Conservative memory guard for parallel paths with thread-local sets or maps. - * The estimate includes a small overhead factor for key and map objects, so the - * parallel path is avoided before local deduplication becomes too memory-heavy. + * Builds the execution plan for the batched parallel path. * * @param blkIn input matrix block - * @return true if local deduplication is small enough for the parallel path + * @param dir unique direction + * @param k requested degree of parallelism + * @return batch configuration, or null if batching would not use parallelism */ - private static boolean isLocalDedupMemoryBudgetSafe(MatrixBlock blkIn) { - long numCells = ((long) blkIn.getNumRows()) * blkIn.getNumColumns(); - if( numCells > Long.MAX_VALUE / Double.BYTES / PAR_UNIQUE_LOCAL_BYTES_OVERHEAD ) - return false; + 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 copiedValueBytes = numCells * Double.BYTES * PAR_UNIQUE_LOCAL_BYTES_OVERHEAD; + long bytesPerIndex = cellsPerIndex * Double.BYTES * PAR_UNIQUE_LOCAL_BYTES_OVERHEAD; long maxLocalBytes = Runtime.getRuntime().maxMemory() / PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION; - return copiedValueBytes <= maxLocalBytes; + return maxLocalBytes / bytesPerIndex; + } + + /** + * Conservative memory guard for full local deduplication. The estimate includes + * a small overhead factor for key and map 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); } /** - * Copies one row into an immutable key/value array for safe HashMap storage. + * Copies one row for key/value storage. * * @param blkIn input matrix block * @param row row index to copy @@ -375,7 +561,7 @@ private static double[] copyRow(MatrixBlock blkIn, int row) { } /** - * Copies one column into an immutable key/value array for safe HashMap storage. + * Copies one column for key/value storage. * * @param blkIn input matrix block * @param col column index to copy @@ -489,6 +675,21 @@ private static boolean equalValues(double[] left, double[] right) { return true; } + /** + * 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 a row partition of a single-column vector locally. */ @@ -563,8 +764,7 @@ public LinkedHashMap call() { } /** - * Content-based key for copied rows. The referenced array is never mutated after - * construction, so it is safe to reuse the copy as both key contents and output data. + * Content-based key for copied rows. */ private static class RowKey { private final double[] _values; @@ -587,8 +787,7 @@ public boolean equals(Object obj) { } /** - * Content-based key for copied columns. The referenced array is never mutated after - * construction, so it is safe to reuse the copy as both key contents and output data. + * Content-based key for copied columns. */ private static class ColKey { private final double[] _values; 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 index f032d7604b9..dad9dfe41fe 100644 --- a/src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java +++ b/src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java @@ -24,21 +24,19 @@ import org.apache.sysds.common.Types; /** - * Small standalone parallel unique test for LibMatrixSketch with k=1 and k>1. - * - * This class intentionally avoids JUnit so it can be run directly from an IDE - * or from a full SystemDS checkout with the normal project classpath. + * Standalone checks for LibMatrixSketch unique paths with k=1 and k>1. */ public class LibMatrixSketchUniqueParallelTest { public static void main(String[] args) { testRowColUniqueMatchesBaseline(); testRowUniqueMatchesBaselineAndExpectedRows(); testColumnUniqueMatchesBaselineAndExpectedColumns(); + testBatchedPathInputsMatchBaseline(); System.out.println("LibMatrixSketch unique parallel tests passed."); } /** - * Checks RowCol unique on a large single-column vector so the k=4 path is used. + * Checks RowCol unique against the single-threaded baseline. */ private static void testRowColUniqueMatchesBaseline() { MatrixBlock input = new MatrixBlock(20000, 1, false).allocateBlock(); @@ -54,8 +52,7 @@ private static void testRowColUniqueMatchesBaseline() { } /** - * Checks Row unique with repeated rows. The expected output is ordered by first - * occurrence, which also verifies the ordered global merge across row partitions. + * Checks Row unique with repeated rows. */ private static void testRowUniqueMatchesBaselineAndExpectedRows() { MatrixBlock input = new MatrixBlock(12000, 2, false).allocateBlock(); @@ -80,8 +77,7 @@ private static void testRowUniqueMatchesBaselineAndExpectedRows() { } /** - * Checks Col unique with repeated columns. The expected output shape is - * original_num_rows x number_of_unique_columns. + * Checks Col unique with repeated columns. */ private static void testColumnUniqueMatchesBaselineAndExpectedColumns() { MatrixBlock input = new MatrixBlock(4, 5000, false).allocateBlock(); @@ -111,6 +107,67 @@ private static void testColumnUniqueMatchesBaselineAndExpectedColumns() { assertBlockEquals(expected, parallel, "Col expected vs parallel"); } + /** + * Uses larger inputs which take the batched path with a small heap. + */ + private static void testBatchedPathInputsMatchBaseline() { + testRowColBatchedPathInput(); + testRowBatchedPathInput(); + testColumnBatchedPathInput(); + } + + /** + * Checks RowCol on a larger input against the single-threaded baseline. + */ + private static void testRowColBatchedPathInput() { + 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 batched dimensions"); + assertSameScalarSet(baseline, parallel, "RowCol batched baseline vs parallel"); + } + + /** + * Checks Row on a larger input with a small number of unique row patterns. + */ + private static void testRowBatchedPathInput() { + 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, (i % 4) * 100 + j); + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row); + MatrixBlock parallel = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row, 4); + MatrixBlock expected = rowPatterns(4, 16); + + assertBlockEquals(expected, baseline, "Row batched expected vs baseline"); + assertBlockEquals(expected, parallel, "Row batched expected vs parallel"); + } + + /** + * Checks Col on a larger input with repeated column patterns. + */ + private static void testColumnBatchedPathInput() { + 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, (j % 4) * 100 + i); + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col); + MatrixBlock parallel = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col, 4); + MatrixBlock expected = columnPatterns(16, 4); + + assertBlockEquals(expected, baseline, "Col batched expected vs baseline"); + assertBlockEquals(expected, parallel, "Col batched expected vs parallel"); + } + /** * Builds a dense MatrixBlock from a plain two-dimensional Java array. */ @@ -123,6 +180,30 @@ private static MatrixBlock matrix(double[][] values) { return ret; } + /** + * Builds repeated row patterns. + */ + private static MatrixBlock rowPatterns(int patterns, int cols) { + MatrixBlock ret = new MatrixBlock(patterns, cols, false).allocateBlock(); + for( int i = 0; i < patterns; i++ ) + for( int j = 0; j < cols; j++ ) + ret.set(i, j, i * 100 + j); + ret.recomputeNonZeros(); + return ret; + } + + /** + * Builds repeated column patterns. + */ + private static MatrixBlock columnPatterns(int rows, int patterns) { + MatrixBlock ret = new MatrixBlock(rows, patterns, false).allocateBlock(); + for( int j = 0; j < patterns; j++ ) + for( int i = 0; i < rows; i++ ) + ret.set(i, j, j * 100 + i); + ret.recomputeNonZeros(); + return ret; + } + /** * Compares two MatrixBlocks cell by cell with exact equality. */ @@ -136,7 +217,7 @@ private static void assertBlockEquals(MatrixBlock expected, MatrixBlock actual, } /** - * Compares RowCol output as a set because the scalar unique path is hash-set based. + * Compares RowCol output as a set. */ private static void assertSameScalarSet(MatrixBlock expected, MatrixBlock actual, String message) { assertDimensions(actual, expected.getNumRows(), expected.getNumColumns(), message); From bcf2d756a74e4ecfd051f186436b7f8c1d90d2b7 Mon Sep 17 00:00:00 2001 From: Xuanyan Wang <54443787+shieru1214@users.noreply.github.com> Date: Sun, 12 Jul 2026 03:02:52 +0200 Subject: [PATCH 5/5] Align unique parallel path with SystemDS semantics --- .../runtime/matrix/data/LibMatrixSketch.java | 580 ++++++++---------- .../LibMatrixSketchUniqueParallelTest.java | 213 +++---- .../test/functions/unique/UniqueBase.java | 2 +- .../test/functions/unique/UniqueRow.java | 13 +- 4 files changed, 335 insertions(+), 473 deletions(-) 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 a931fe0b2a3..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 @@ -20,16 +20,13 @@ package org.apache.sysds.runtime.matrix.data; import java.util.ArrayList; -import java.util.Collection; import java.util.HashSet; import java.util.Iterator; -import java.util.LinkedHashMap; import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; -import org.apache.commons.lang3.NotImplementedException; import org.apache.sysds.common.Types; import org.apache.sysds.runtime.DMLRuntimeException; import org.apache.sysds.runtime.util.CommonThreadPool; @@ -38,32 +35,32 @@ 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 = 2; + private static final long PAR_UNIQUE_LOCAL_BYTES_OVERHEAD = 8; /** - * Computes unique values, rows, or columns with the original single-threaded behavior. + * 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, rows, or columns + * @return matrix block containing unique values */ public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir) { return getUniqueValues(blkIn, dir, 1); } /** - * Computes unique values, rows, or columns. For sufficiently large inputs and - * k > 1, this uses parallel local deduplication or its batched variant. + * 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, rows, or columns + * @return matrix block containing unique values */ public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir, int k) { - // similar to R's unique, this operation takes a matrix and computes the unique values - // (or rows in case of multiple column inputs) + // Similar to R's unique, this operation computes unique values according + // to the requested direction. if( !satisfiesMultiThreadingConstraints(blkIn, dir, k) ) return getUniqueValuesSequential(blkIn, dir); @@ -75,12 +72,12 @@ public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir getUniqueValuesRowColBatchedParallel(blkIn, dir, k); case Row: return localDedupMemorySafe ? - getUniqueRowsParallel(blkIn, k) : - getUniqueRowsBatchedParallel(blkIn, dir, k); + getUniqueRowValuesParallel(blkIn, k) : + getUniqueRowValuesBatchedParallel(blkIn, dir, k); case Col: return localDedupMemorySafe ? - getUniqueColumnsParallel(blkIn, k) : - getUniqueColumnsBatchedParallel(blkIn, dir, k); + getUniqueColumnValuesParallel(blkIn, k) : + getUniqueColumnValuesBatchedParallel(blkIn, dir, k); default: throw new IllegalArgumentException("Unrecognized direction: " + dir); } @@ -88,11 +85,11 @@ public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir /** * Single-threaded baseline implementation for all unique directions. - * This preserves the original RowCol and Row behavior and adds the sequential Col path. + * 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, rows, or columns + * @return matrix block containing unique values */ private static MatrixBlock getUniqueValuesSequential(MatrixBlock blkIn, Types.Direction dir) { int rlen = blkIn.getNumRows(); @@ -101,14 +98,13 @@ private static MatrixBlock getUniqueValuesSequential(MatrixBlock blkIn, Types.Di MatrixBlock blkOut = null; switch (dir) { case RowCol: - if( clen != 1 ) - throw new NotImplementedException("Unique only support single-column vectors yet"); // TODO optimize for dense/sparse/compressed (once multi-column support added) - // obtain set of unique items (dense input vector) + // obtain set of unique items HashSet hashSet = new HashSet<>(); for( int i=0; i retainedRows = new ArrayList<>(); - - for (int i=0; i rowSet = new HashSet<>(); + int clen2 = 0; + for( int i=0; i retainedColumns = new LinkedHashMap<>(); - for( int j = 0; j < clen; j++ ) { - double[] currentColumn = copyColumn(blkIn, j); - retainedColumns.putIfAbsent(new ColKey(currentColumn), currentColumn); + // 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 UniqueRowTask(blkIn, range[0], range[1])); - - // Keep the merge ordered to preserve first occurrences. - LinkedHashMap retainedRows = new LinkedHashMap<>(); - List>> rtasks = pool.invokeAll(tasks); - for( int i = 0; i < rtasks.size(); i++ ) { - LinkedHashMap localRows = rtasks.get(i).get(); - for( java.util.Map.Entry entry : localRows.entrySet() ) - retainedRows.putIfAbsent(entry.getKey(), entry.getValue()); - localRows.clear(); - rtasks.set(i, null); - } + 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); - return createRowOutput(retainedRows.values(), blkIn.getNumColumns()); + blkOut.recomputeNonZeros(); + blkOut.examSparsity(); + return blkOut; } catch(Exception ex) { throw new DMLRuntimeException(ex); @@ -261,33 +232,25 @@ private static MatrixBlock getUniqueRowsParallel(MatrixBlock blkIn, int k) { } /** - * Parallel unique columns. Each worker deduplicates a column partition locally, and - * the final merge scans partitions from left to right to keep the first occurrence. + * 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 exact unique columns + * @return matrix block containing column-wise unique values */ - private static MatrixBlock getUniqueColumnsParallel(MatrixBlock blkIn, int k) { + private static MatrixBlock getUniqueColumnValuesParallel(MatrixBlock blkIn, int k) { int numThreads = getNumThreads(k, blkIn.getNumColumns()); ExecutorService pool = CommonThreadPool.get(numThreads); try { - ArrayList tasks = new ArrayList<>(); - for( int[] range : getBalancedRanges(blkIn.getNumColumns(), numThreads) ) - tasks.add(new UniqueColumnTask(blkIn, range[0], range[1])); + 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); - // Keep the merge ordered to preserve first occurrences. - LinkedHashMap retainedColumns = new LinkedHashMap<>(); - List>> rtasks = pool.invokeAll(tasks); - for( int i = 0; i < rtasks.size(); i++ ) { - LinkedHashMap localColumns = rtasks.get(i).get(); - for( java.util.Map.Entry entry : localColumns.entrySet() ) - retainedColumns.putIfAbsent(entry.getKey(), entry.getValue()); - localColumns.clear(); - rtasks.set(i, null); - } - - return createColumnOutput(retainedColumns.values(), blkIn.getNumRows()); + blkOut.recomputeNonZeros(); + blkOut.examSparsity(); + return blkOut; } catch(Exception ex) { throw new DMLRuntimeException(ex); @@ -298,17 +261,14 @@ private static MatrixBlock getUniqueColumnsParallel(MatrixBlock blkIn, int k) { } /** - * Batched parallel unique for single-column vectors. + * Batched parallel unique for all matrix values. * - * @param blkIn input single-column matrix block + * @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) { - if( blkIn.getNumColumns() != 1 ) - throw new NotImplementedException("Unique only support single-column vectors yet"); - BatchConfig config = getBatchConfig(blkIn, dir, k); if( config == null ) return getUniqueValuesSequential(blkIn, dir); @@ -344,40 +304,27 @@ private static MatrixBlock getUniqueValuesRowColBatchedParallel(MatrixBlock blkI } /** - * Batched parallel unique rows. Partitions are merged in row order. + * 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 exact unique rows + * @return matrix block containing row-wise unique values */ - private static MatrixBlock getUniqueRowsBatchedParallel(MatrixBlock blkIn, Types.Direction dir, int k) { + 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 { - LinkedHashMap retainedRows = new LinkedHashMap<>(); - 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 UniqueRowTask(blkIn, pos, end)); - pos = end; - } - - List>> rtasks = pool.invokeAll(tasks); - for( int i = 0; i < rtasks.size(); i++ ) { - LinkedHashMap localRows = rtasks.get(i).get(); - for( java.util.Map.Entry entry : localRows.entrySet() ) - retainedRows.putIfAbsent(entry.getKey(), entry.getValue()); - localRows.clear(); - rtasks.set(i, null); - } - } + int clen2 = getMaxUniqueValuesBatched(pool, blkIn, dir, config); + MatrixBlock blkOut = allocateOutputBlock(blkIn.getNumRows(), clen2); + fillUniqueValuesBatched(pool, blkIn, blkOut, dir, config); - return createRowOutput(retainedRows.values(), blkIn.getNumColumns()); + blkOut.recomputeNonZeros(); + blkOut.examSparsity(); + return blkOut; } catch(Exception ex) { throw new DMLRuntimeException(ex); @@ -388,40 +335,27 @@ private static MatrixBlock getUniqueRowsBatchedParallel(MatrixBlock blkIn, Types } /** - * Batched parallel unique columns. Column ranges are merged from left to right. + * 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 exact unique columns + * @return matrix block containing column-wise unique values */ - private static MatrixBlock getUniqueColumnsBatchedParallel(MatrixBlock blkIn, Types.Direction dir, int k) { + 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 { - LinkedHashMap retainedColumns = new LinkedHashMap<>(); - 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 UniqueColumnTask(blkIn, pos, end)); - pos = end; - } + int rlen2 = getMaxUniqueValuesBatched(pool, blkIn, dir, config); + MatrixBlock blkOut = allocateOutputBlock(rlen2, blkIn.getNumColumns()); + fillUniqueValuesBatched(pool, blkIn, blkOut, dir, config); - List>> rtasks = pool.invokeAll(tasks); - for( int i = 0; i < rtasks.size(); i++ ) { - LinkedHashMap localColumns = rtasks.get(i).get(); - for( java.util.Map.Entry entry : localColumns.entrySet() ) - retainedColumns.putIfAbsent(entry.getKey(), entry.getValue()); - localColumns.clear(); - rtasks.set(i, null); - } - } - - return createColumnOutput(retainedColumns.values(), blkIn.getNumRows()); + blkOut.recomputeNonZeros(); + blkOut.examSparsity(); + return blkOut; } catch(Exception ex) { throw new DMLRuntimeException(ex); @@ -431,6 +365,83 @@ private static MatrixBlock getUniqueColumnsBatchedParallel(MatrixBlock blkIn, Ty } } + /** + * 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. * @@ -535,7 +546,7 @@ private static long getMaxLocalDedupIndexes(MatrixBlock blkIn, Types.Direction d /** * Conservative memory guard for full local deduplication. The estimate includes - * a small overhead factor for key and map objects. + * a small overhead factor for set objects. * * @param blkIn input matrix block * @param dir unique direction @@ -545,36 +556,6 @@ private static boolean isLocalDedupMemoryBudgetSafe(MatrixBlock blkIn, Types.Dir return getMaxLocalDedupIndexes(blkIn, dir) >= getPartitionLength(blkIn, dir); } - /** - * Copies one row for key/value storage. - * - * @param blkIn input matrix block - * @param row row index to copy - * @return copied row values - */ - private static double[] copyRow(MatrixBlock blkIn, int row) { - int clen = blkIn.getNumColumns(); - double[] ret = new double[clen]; - for( int j = 0; j < clen; j++ ) - ret[j] = blkIn.get(row, j); - return ret; - } - - /** - * Copies one column for key/value storage. - * - * @param blkIn input matrix block - * @param col column index to copy - * @return copied column values - */ - private static double[] copyColumn(MatrixBlock blkIn, int col) { - int rlen = blkIn.getNumRows(); - double[] ret = new double[rlen]; - for( int i = 0; i < rlen; i++ ) - ret[i] = blkIn.get(i, col); - return ret; - } - /** * Allocates and fills a one-column MatrixBlock from a set of unique scalar values. * @@ -592,46 +573,6 @@ private static MatrixBlock createRowColOutput(HashSet values) { return blkOut; } - /** - * Allocates and fills a MatrixBlock from retained row copies. - * - * @param rows unique row copies - * @param clen number of columns in the output - * @return matrix block containing the unique rows - */ - private static MatrixBlock createRowOutput(Collection rows, int clen) { - MatrixBlock blkOut = allocateOutputBlock(rows.size(), clen); - int i = 0; - for( double[] row : rows ) { - for( int j = 0; j < clen; j++ ) - blkOut.set(i, j, row[j]); - i++; - } - blkOut.recomputeNonZeros(); - blkOut.examSparsity(); - return blkOut; - } - - /** - * Allocates and fills a MatrixBlock from retained column copies. - * - * @param columns unique column copies - * @param rlen number of rows in the output - * @return matrix block containing the unique columns - */ - private static MatrixBlock createColumnOutput(Collection columns, int rlen) { - MatrixBlock blkOut = allocateOutputBlock(rlen, columns.size()); - int j = 0; - for( double[] column : columns ) { - for( int i = 0; i < rlen; i++ ) - blkOut.set(i, j, column[i]); - j++; - } - blkOut.recomputeNonZeros(); - blkOut.examSparsity(); - return blkOut; - } - /** * Creates an output block and allocates storage only when at least one cell exists. * @@ -646,35 +587,6 @@ private static MatrixBlock allocateOutputBlock(int rlen, int clen) { return blkOut; } - /** - * Computes a stable content hash for row and column keys using numeric equality. - * - * @param values copied row or column values - * @return content hash code - */ - private static int hashValues(double[] values) { - int ret = 1; - for( double value : values ) - ret = 31 * ret + (value == 0 ? 0 : Double.hashCode(value)); - return ret; - } - - /** - * Compares copied row or column values with exact numeric equality. - * - * @param left first copied value array - * @param right second copied value array - * @return true if the arrays represent the same row or column - */ - private static boolean equalValues(double[] left, double[] right) { - if( left.length != right.length ) - return false; - for( int i = 0; i < left.length; i++ ) - if( left[i] != right[i] && Double.doubleToLongBits(left[i]) != Double.doubleToLongBits(right[i]) ) - return false; - return true; - } - /** * Configuration for batched execution. */ @@ -691,7 +603,7 @@ private BatchConfig(int numThreads, int taskLen, int len) { } /** - * Worker that deduplicates a row partition of a single-column vector locally. + * Worker that deduplicates all scalar values in a row range locally. */ private static class UniqueValueTask implements Callable> { private final MatrixBlock _blkIn; @@ -708,104 +620,94 @@ private UniqueValueTask(MatrixBlock blkIn, int rl, int ru) { public HashSet call() { HashSet ret = new HashSet<>(); for( int i = _rl; i < _ru; i++ ) - ret.add(_blkIn.get(i, 0)); + for( int j = 0; j < _blkIn.getNumColumns(); j++ ) + ret.add(_blkIn.get(i, j)); return ret; } } /** - * Worker that deduplicates copied rows within a row partition before global merge. + * Worker that computes the largest row-wise or column-wise unique count in a range. */ - private static class UniqueRowTask implements Callable> { + private static class UniqueCountTask implements Callable { private final MatrixBlock _blkIn; - private final int _rl; - private final int _ru; + private final Types.Direction _dir; + private final int _l; + private final int _u; - private UniqueRowTask(MatrixBlock blkIn, int rl, int ru) { + private UniqueCountTask(MatrixBlock blkIn, Types.Direction dir, int l, int u) { _blkIn = blkIn; - _rl = rl; - _ru = ru; + _dir = dir; + _l = l; + _u = u; } @Override - public LinkedHashMap call() { - LinkedHashMap ret = new LinkedHashMap<>(); - for( int i = _rl; i < _ru; i++ ) { - double[] row = copyRow(_blkIn, i); - ret.putIfAbsent(new RowKey(row), row); + 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 deduplicates copied columns within a column partition before global merge. + * Worker that writes row-wise or column-wise unique values for a range. */ - private static class UniqueColumnTask implements Callable> { + private static class UniqueOutputTask implements Callable { private final MatrixBlock _blkIn; - private final int _cl; - private final int _cu; + private final MatrixBlock _blkOut; + private final Types.Direction _dir; + private final int _l; + private final int _u; - private UniqueColumnTask(MatrixBlock blkIn, int cl, int cu) { + private UniqueOutputTask(MatrixBlock blkIn, MatrixBlock blkOut, Types.Direction dir, int l, int u) { _blkIn = blkIn; - _cl = cl; - _cu = cu; + _blkOut = blkOut; + _dir = dir; + _l = l; + _u = u; } @Override - public LinkedHashMap call() { - LinkedHashMap ret = new LinkedHashMap<>(); - for( int j = _cl; j < _cu; j++ ) { - double[] column = copyColumn(_blkIn, j); - ret.putIfAbsent(new ColKey(column), column); + 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); + } } - return ret; - } - } - - /** - * Content-based key for copied rows. - */ - private static class RowKey { - private final double[] _values; - private final int _hash; - - private RowKey(double[] values) { - _values = values; - _hash = hashValues(values); - } - - @Override - public int hashCode() { - return _hash; - } - - @Override - public boolean equals(Object obj) { - return obj instanceof RowKey && equalValues(_values, ((RowKey) obj)._values); - } - } - - /** - * Content-based key for copied columns. - */ - private static class ColKey { - private final double[] _values; - private final int _hash; - - private ColKey(double[] values) { - _values = values; - _hash = hashValues(values); - } - - @Override - public int hashCode() { - return _hash; - } - - @Override - public boolean equals(Object obj) { - return obj instanceof ColKey && equalValues(_values, ((ColKey) obj)._values); + 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 index dad9dfe41fe..cf95b593dc3 100644 --- a/src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java +++ b/src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java @@ -19,107 +19,81 @@ 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; /** - * Standalone checks for LibMatrixSketch unique paths with k=1 and k>1. + * Tests LibMatrixSketch unique paths with k=1 and k>1. */ public class LibMatrixSketchUniqueParallelTest { - public static void main(String[] args) { - testRowColUniqueMatchesBaseline(); - testRowUniqueMatchesBaselineAndExpectedRows(); - testColumnUniqueMatchesBaselineAndExpectedColumns(); - testBatchedPathInputsMatchBaseline(); - System.out.println("LibMatrixSketch unique parallel tests passed."); - } - - /** - * Checks RowCol unique against the single-threaded baseline. - */ - private static void testRowColUniqueMatchesBaseline() { - MatrixBlock input = new MatrixBlock(20000, 1, false).allocateBlock(); - for( int i = 0; i < input.getNumRows(); i++ ) + @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, 7, 1, "RowCol parallel dimensions"); + assertDimensions(parallel, 11, 1, "RowCol parallel dimensions"); assertSameScalarSet(baseline, parallel, "RowCol baseline vs parallel"); } - /** - * Checks Row unique with repeated rows. - */ - private static void testRowUniqueMatchesBaselineAndExpectedRows() { - MatrixBlock input = new MatrixBlock(12000, 2, false).allocateBlock(); + @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 * 10); + 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); - MatrixBlock expected = matrix(new double[][] { - {0, 0}, - {1, 10}, - {2, 20}, - {3, 30} - }); - - assertBlockEquals(expected, baseline, "Row expected vs baseline"); - assertBlockEquals(expected, parallel, "Row expected vs parallel"); + + 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"); } - /** - * Checks Col unique with repeated columns. - */ - private static void testColumnUniqueMatchesBaselineAndExpectedColumns() { + @Test + public void testColumnUniqueMatchesBaselineAndExpectedValues() { MatrixBlock input = new MatrixBlock(4, 5000, false).allocateBlock(); - double[][] uniqueColumns = new double[][] { - {1, 2, 3, 4}, - {5, 6, 7, 8}, - {9, 10, 11, 12} - }; - for( int j = 0; j < input.getNumColumns(); j++ ) { - double[] column = uniqueColumns[j % uniqueColumns.length]; - for( int i = 0; i < input.getNumRows(); i++ ) - input.set(i, j, column[i]); + 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); - MatrixBlock expected = matrix(new double[][] { - {1, 5, 9}, - {2, 6, 10}, - {3, 7, 11}, - {4, 8, 12} - }); - - assertBlockEquals(expected, baseline, "Col expected vs baseline"); - assertBlockEquals(expected, parallel, "Col expected vs parallel"); + + 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"); } - /** - * Uses larger inputs which take the batched path with a small heap. - */ - private static void testBatchedPathInputsMatchBaseline() { - testRowColBatchedPathInput(); - testRowBatchedPathInput(); - testColumnBatchedPathInput(); + @Test + public void testLargerInputsMatchBaseline() { + testRowColLargeInput(); + testRowLargeInput(); + testColumnLargeInput(); } - /** - * Checks RowCol on a larger input against the single-threaded baseline. - */ - private static void testRowColBatchedPathInput() { + 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); @@ -128,80 +102,38 @@ private static void testRowColBatchedPathInput() { MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.RowCol); MatrixBlock parallel = LibMatrixSketch.getUniqueValues(input, Types.Direction.RowCol, 4); - assertDimensions(parallel, 7, 1, "RowCol batched dimensions"); - assertSameScalarSet(baseline, parallel, "RowCol batched baseline vs parallel"); + assertDimensions(parallel, 7, 1, "RowCol large input dimensions"); + assertSameScalarSet(baseline, parallel, "RowCol large input baseline vs parallel"); } - /** - * Checks Row on a larger input with a small number of unique row patterns. - */ - private static void testRowBatchedPathInput() { + 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, (i % 4) * 100 + 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); - MatrixBlock expected = rowPatterns(4, 16); - assertBlockEquals(expected, baseline, "Row batched expected vs baseline"); - assertBlockEquals(expected, parallel, "Row batched expected vs parallel"); + 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"); } - /** - * Checks Col on a larger input with repeated column patterns. - */ - private static void testColumnBatchedPathInput() { + 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, (j % 4) * 100 + 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); - MatrixBlock expected = columnPatterns(16, 4); - - assertBlockEquals(expected, baseline, "Col batched expected vs baseline"); - assertBlockEquals(expected, parallel, "Col batched expected vs parallel"); - } - - /** - * Builds a dense MatrixBlock from a plain two-dimensional Java array. - */ - private static MatrixBlock matrix(double[][] values) { - MatrixBlock ret = new MatrixBlock(values.length, values[0].length, false).allocateBlock(); - for( int i = 0; i < values.length; i++ ) - for( int j = 0; j < values[i].length; j++ ) - ret.set(i, j, values[i][j]); - ret.recomputeNonZeros(); - return ret; - } - - /** - * Builds repeated row patterns. - */ - private static MatrixBlock rowPatterns(int patterns, int cols) { - MatrixBlock ret = new MatrixBlock(patterns, cols, false).allocateBlock(); - for( int i = 0; i < patterns; i++ ) - for( int j = 0; j < cols; j++ ) - ret.set(i, j, i * 100 + j); - ret.recomputeNonZeros(); - return ret; - } - /** - * Builds repeated column patterns. - */ - private static MatrixBlock columnPatterns(int rows, int patterns) { - MatrixBlock ret = new MatrixBlock(rows, patterns, false).allocateBlock(); - for( int j = 0; j < patterns; j++ ) - for( int i = 0; i < rows; i++ ) - ret.set(i, j, j * 100 + i); - ret.recomputeNonZeros(); - return ret; + 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"); } /** @@ -211,9 +143,8 @@ private static void assertBlockEquals(MatrixBlock expected, MatrixBlock actual, assertDimensions(actual, expected.getNumRows(), expected.getNumColumns(), message); for( int i = 0; i < expected.getNumRows(); i++ ) for( int j = 0; j < expected.getNumColumns(); j++ ) - if( expected.get(i, j) != actual.get(i, j) ) - throw new AssertionError(message + " mismatch at (" + i + ", " + j + "): expected " - + expected.get(i, j) + " but found " + actual.get(i, j)); + assertEquals(message + " mismatch at (" + i + ", " + j + ")", expected.get(i, j), + actual.get(i, j), 0); } /** @@ -223,8 +154,7 @@ private static void assertSameScalarSet(MatrixBlock expected, MatrixBlock actual assertDimensions(actual, expected.getNumRows(), expected.getNumColumns(), message); HashSet expectedValues = collectScalars(expected); HashSet actualValues = collectScalars(actual); - if( !expectedValues.equals(actualValues) ) - throw new AssertionError(message + " mismatch: expected " + expectedValues + " but found " + actualValues); + assertEquals(message + " mismatch", expectedValues, actualValues); } /** @@ -237,12 +167,43 @@ private static HashSet collectScalars(MatrixBlock block) { 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) { - if( block.getNumRows() != rows || block.getNumColumns() != cols ) - throw new AssertionError(message + " dimensions mismatch: expected " + rows + "x" + cols - + " but found " + block.getNumRows() + "x" + block.getNumColumns()); + assertEquals(message + " row dimension", rows, block.getNumRows()); + assertEquals(message + " column dimension", cols, block.getNumColumns()); } } 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 d834fe45aea..6e65c01f7c9 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 @@ -45,7 +45,7 @@ protected void uniqueTest(double[][] inputMatrix, double[][] expectedMatrix, loadTestConfiguration(getTestConfiguration(getTestName())); String HOME = SCRIPT_DIR + getTestDir(); fullDMLScriptName = HOME + getTestName() + ".dml"; - programArgs = new String[]{ "-args", input("I"), output("A")}; + programArgs = new String[]{"-args", input("I"), output("A")}; writeInputMatrixWithMTD("I", inputMatrix, true); 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 ee8c664efaa..fda9aa4a3c0 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 @@ -27,7 +27,6 @@ public class UniqueRow extends UniqueBase { private final static String TEST_DIR = "functions/unique/"; private static final String TEST_CLASS_DIR = TEST_DIR + UniqueRow.class.getSimpleName() + "/"; - @Override protected String getTestName() { return TEST_NAME; @@ -52,22 +51,22 @@ public void testBaseCaseCP() { @Test public void testSkinnyCP() { - double[][] inputMatrix = {{1},{1},{6},{9},{4},{2},{0},{9},{0},{0},{4},{4}}; - double[][] expectedMatrix = {{1},{6},{9},{4},{2},{0}}; + 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 testSquareCP() { - double[][] inputMatrix = {{1, 2, 3}, {4, 5, 6}, {1, 2, 3}}; - double[][] expectedMatrix = {{1, 2, 3},{4, 5, 6}}; + double[][] inputMatrix = {{1, 4, 1}, {2, 5, 2}, {3, 6, 3}}; + double[][] expectedMatrix = {{1, 4},{2, 5},{3, 6}}; uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0); } @Test public void testWideCP() { - double[][] inputMatrix = {{1, 2, 3, 4, 5, 6}, {7, 8, 9, 10, 11, 12}, {1, 2, 3, 4, 5, 6}}; - double[][] expectedMatrix = {{1, 2, 3, 4, 5, 6}, {7, 8, 9, 10, 11, 12}}; + double[][] inputMatrix = {{1,7,1},{2,8,2},{3,9,3},{4,10,4},{5,11,5},{6,12,6}}; + double[][] expectedMatrix = {{1,7},{2,8},{3,9},{4,10},{5,11},{6,12}}; uniqueTest(inputMatrix, expectedMatrix, Types.ExecType.CP, 0.0); }