From 968e63fb96879f14e877893e654cb6f4183cacbd 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/7] 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 101074eefb3bbf6217a9307db692cb27807dbf78 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/7] 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 70954fc785e0a7e953bdf5006a61487070174b33 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/7] 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 59da7b7e3bbc60448fa3a1596d70d86def83db15 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/7] 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 4a89ac72757557efc5cd93efa8a7ead8cf8c146c 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/7] 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); } From aaba9e797eb5fb0741650a284fac689b23297c99 Mon Sep 17 00:00:00 2001 From: Isso-W Date: Sat, 25 Jul 2026 03:36:22 +0200 Subject: [PATCH 6/7] [SYSTEMDS-3524] Address review: activate parallel unique and fix batched memory guards - AggregateUnaryCPInstruction now passes op.getNumThreads() into LibMatrixSketch.getUniqueValues, so the multi-threaded implementation is actually used instead of always falling back to the single-threaded path. - Fix int overflow in the batched task-range computation ((int) Math.min((long) pos + taskLen, len)). - Correct the memory guard for the row-wise and column-wise paths: their workers reuse one set that is cleared per row/column, so only one live set per thread is charged (numThreads * cellsPerIndex * 64 <= budget) instead of charging all indexes, which forced batched execution where it is safely avoidable. - For batched RowCol reserve the all-distinct worst case of the merged set up front and size taskLen against the remaining budget; fall back to sequential execution when even the merged set does not fit. - Add a getUniqueValues overload with an injectable maxLocalBytes budget, move the parallel test into functions.unique so CI actually runs it, and add cases that deterministically exercise all three batched paths and their sequential fallbacks (the heap-derived default budget never triggers them in CI). --- .../cp/AggregateUnaryCPInstruction.java | 2 +- .../runtime/matrix/data/LibMatrixSketch.java | 329 ++++++++++++------ .../LibMatrixSketchUniqueParallelTest.java | 171 ++++++++- 3 files changed, 372 insertions(+), 130 deletions(-) rename src/test/java/org/apache/sysds/{runtime/matrix/data => test/functions/unique}/LibMatrixSketchUniqueParallelTest.java (54%) diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/AggregateUnaryCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/AggregateUnaryCPInstruction.java index d0904b38fba..08fffdae0d3 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/cp/AggregateUnaryCPInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/AggregateUnaryCPInstruction.java @@ -232,7 +232,7 @@ else if( input1.getDataType().isMatrix() || input1.getDataType().isFrame() ) { } UnarySketchOperator op = (UnarySketchOperator) _optr; - MatrixBlock res = LibMatrixSketch.getUniqueValues(input, op.getDirection()); + MatrixBlock res = LibMatrixSketch.getUniqueValues(input, op.getDirection(), op.getNumThreads()); ec.releaseMatrixInput(input1.getName()); ec.setMatrixOutput(outputName, res); break; diff --git a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketch.java b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketch.java index f0afec847f8..92ecc450de8 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 @@ -35,14 +35,18 @@ public class LibMatrixSketch { private static final long PAR_UNIQUE_NUMCELL_THRESHOLD = 1024 * 16; private static final long PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION = 4; - private static final long PAR_UNIQUE_LOCAL_BYTES_OVERHEAD = 8; + /** + * Conservative footprint of one value retained in a HashSet<Double>: the boxed Double plus amortized hash-map + * node and backing-array overhead. + */ + private static final long PAR_UNIQUE_BYTES_PER_CELL = Double.BYTES * 8; /** - * Computes unique values with the original single-threaded behavior. - * The overload with a parallelism argument keeps this path as the k=1 baseline. + * 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 + * @param dir unique direction * @return matrix block containing unique values */ public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir) { @@ -50,45 +54,58 @@ public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir } /** - * Computes unique values. 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 + * @param dir unique direction + * @param k requested degree of parallelism * @return matrix block containing unique values */ public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir, int k) { + return getUniqueValues(blkIn, dir, k, getDefaultLocalBytesBudget()); + } + + /** + * Computes unique values with an explicit budget for the transient deduplication structures. The budget decides + * between the full parallel path, its batched variant, and the sequential fallback. This overload exists so tests + * can inject a small budget and deterministically exercise the batched path, which the heap-derived default would + * not trigger. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @param maxLocalBytes budget in bytes for transient deduplication structures + * @return matrix block containing unique values + */ + public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir, int k, long maxLocalBytes) { // Similar to R's unique, this operation computes unique values according // to the requested direction. - if( !satisfiesMultiThreadingConstraints(blkIn, dir, k) ) + if(!satisfiesMultiThreadingConstraints(blkIn, dir, k)) return getUniqueValuesSequential(blkIn, dir); - boolean localDedupMemorySafe = isLocalDedupMemoryBudgetSafe(blkIn, dir); + boolean localDedupMemorySafe = isLocalDedupMemoryBudgetSafe(blkIn, dir, k, maxLocalBytes); switch(dir) { case RowCol: - return localDedupMemorySafe ? - getUniqueValuesRowColParallel(blkIn, k) : - getUniqueValuesRowColBatchedParallel(blkIn, dir, k); + return localDedupMemorySafe ? getUniqueValuesRowColParallel(blkIn, + k) : getUniqueValuesRowColBatchedParallel(blkIn, dir, k, maxLocalBytes); case Row: - return localDedupMemorySafe ? - getUniqueRowValuesParallel(blkIn, k) : - getUniqueRowValuesBatchedParallel(blkIn, dir, k); + return localDedupMemorySafe ? getUniqueRowValuesParallel(blkIn, + k) : getUniqueRowValuesBatchedParallel(blkIn, dir, k, maxLocalBytes); case Col: - return localDedupMemorySafe ? - getUniqueColumnValuesParallel(blkIn, k) : - getUniqueColumnValuesBatchedParallel(blkIn, dir, k); + return localDedupMemorySafe ? getUniqueColumnValuesParallel(blkIn, + k) : getUniqueColumnValuesBatchedParallel(blkIn, dir, k, maxLocalBytes); default: throw new IllegalArgumentException("Unrecognized direction: " + dir); } } /** - * Single-threaded baseline implementation for all unique directions. - * This preserves the original row-wise and column-wise unique behavior. + * Single-threaded baseline implementation for all unique directions. This preserves the original row-wise and + * column-wise unique behavior. * * @param blkIn input matrix block - * @param dir unique direction + * @param dir unique direction * @return matrix block containing unique values */ private static MatrixBlock getUniqueValuesSequential(MatrixBlock blkIn, Types.Direction dir) { @@ -129,7 +146,7 @@ private static MatrixBlock getUniqueValuesSequential(MatrixBlock blkIn, Types.Di for( int j=0; j tasks = new ArrayList<>(); - for( int[] range : getBalancedRanges(blkIn.getNumRows(), numThreads) ) + for(int[] range : getBalancedRanges(blkIn.getNumRows(), numThreads)) tasks.add(new UniqueValueTask(blkIn, range[0], range[1])); // Merge local sets after the workers complete. HashSet hashSet = new HashSet<>(); List>> rtasks = pool.invokeAll(tasks); - for( int i = 0; i < rtasks.size(); i++ ) { + for(int i = 0; i < rtasks.size(); i++) { HashSet localSet = rtasks.get(i).get(); hashSet.addAll(localSet); localSet.clear(); @@ -203,11 +220,11 @@ private static MatrixBlock getUniqueValuesRowColParallel(MatrixBlock blkIn, int } /** - * Parallel row-wise unique values. A first pass computes the output width, - * and a second pass materializes the row-local unique values. + * Parallel row-wise unique values. A first pass computes the output width, and a second pass materializes the + * row-local unique values. * * @param blkIn input matrix block - * @param k requested degree of parallelism + * @param k requested degree of parallelism * @return matrix block containing row-wise unique values */ private static MatrixBlock getUniqueRowValuesParallel(MatrixBlock blkIn, int k) { @@ -232,11 +249,11 @@ private static MatrixBlock getUniqueRowValuesParallel(MatrixBlock blkIn, int k) } /** - * Parallel column-wise unique values. A first pass computes the output height, - * and a second pass materializes the column-local unique values. + * 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 + * @param k requested degree of parallelism * @return matrix block containing column-wise unique values */ private static MatrixBlock getUniqueColumnValuesParallel(MatrixBlock blkIn, int k) { @@ -264,28 +281,29 @@ private static MatrixBlock getUniqueColumnValuesParallel(MatrixBlock blkIn, int * Batched parallel unique for all matrix values. * * @param blkIn input matrix block - * @param dir unique direction - * @param k requested degree of parallelism + * @param dir unique direction + * @param k requested degree of parallelism * @return one-column matrix block containing the unique values */ - private static MatrixBlock getUniqueValuesRowColBatchedParallel(MatrixBlock blkIn, Types.Direction dir, int k) { - BatchConfig config = getBatchConfig(blkIn, dir, k); - if( config == null ) + private static MatrixBlock getUniqueValuesRowColBatchedParallel(MatrixBlock blkIn, Types.Direction dir, int k, + long maxLocalBytes) { + BatchConfig config = getRowColBatchConfig(blkIn, dir, k, maxLocalBytes); + if(config == null) return getUniqueValuesSequential(blkIn, dir); ExecutorService pool = CommonThreadPool.get(config._numThreads); try { HashSet hashSet = new HashSet<>(); - for( int pos = 0; pos < config._len; ) { + 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); + for(int i = 0; i < config._numThreads && pos < config._len; i++) { + int end = (int) Math.min((long) pos + config._taskLen, config._len); tasks.add(new UniqueValueTask(blkIn, pos, end)); pos = end; } List>> rtasks = pool.invokeAll(tasks); - for( int i = 0; i < rtasks.size(); i++ ) { + for(int i = 0; i < rtasks.size(); i++) { HashSet localSet = rtasks.get(i).get(); hashSet.addAll(localSet); localSet.clear(); @@ -307,13 +325,14 @@ private static MatrixBlock getUniqueValuesRowColBatchedParallel(MatrixBlock blkI * Batched parallel row-wise unique values. * * @param blkIn input matrix block - * @param dir unique direction - * @param k requested degree of parallelism + * @param dir unique direction + * @param k requested degree of parallelism * @return matrix block containing row-wise unique values */ - private static MatrixBlock getUniqueRowValuesBatchedParallel(MatrixBlock blkIn, Types.Direction dir, int k) { - BatchConfig config = getBatchConfig(blkIn, dir, k); - if( config == null ) + private static MatrixBlock getUniqueRowValuesBatchedParallel(MatrixBlock blkIn, Types.Direction dir, int k, + long maxLocalBytes) { + BatchConfig config = getBatchConfig(blkIn, dir, k, maxLocalBytes); + if(config == null) return getUniqueValuesSequential(blkIn, dir); ExecutorService pool = CommonThreadPool.get(config._numThreads); @@ -338,13 +357,14 @@ private static MatrixBlock getUniqueRowValuesBatchedParallel(MatrixBlock blkIn, * Batched parallel column-wise unique values. * * @param blkIn input matrix block - * @param dir unique direction - * @param k requested degree of parallelism + * @param dir unique direction + * @param k requested degree of parallelism * @return matrix block containing column-wise unique values */ - private static MatrixBlock getUniqueColumnValuesBatchedParallel(MatrixBlock blkIn, Types.Direction dir, int k) { - BatchConfig config = getBatchConfig(blkIn, dir, k); - if( config == null ) + private static MatrixBlock getUniqueColumnValuesBatchedParallel(MatrixBlock blkIn, Types.Direction dir, int k, + long maxLocalBytes) { + BatchConfig config = getBatchConfig(blkIn, dir, k, maxLocalBytes); + if(config == null) return getUniqueValuesSequential(blkIn, dir); ExecutorService pool = CommonThreadPool.get(config._numThreads); @@ -371,12 +391,12 @@ private static MatrixBlock getUniqueColumnValuesBatchedParallel(MatrixBlock blkI private static int getMaxUniqueValues(ExecutorService pool, MatrixBlock blkIn, Types.Direction dir, ArrayList ranges) throws Exception { ArrayList tasks = new ArrayList<>(); - for( int[] range : ranges ) + 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++ ) { + for(int i = 0; i < rtasks.size(); i++) { ret = Math.max(ret, rtasks.get(i).get()); rtasks.set(i, null); } @@ -389,10 +409,10 @@ private static int getMaxUniqueValues(ExecutorService pool, MatrixBlock blkIn, T 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 ) + 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++ ) { + for(int i = 0; i < rtasks.size(); i++) { rtasks.get(i).get(); rtasks.set(i, null); } @@ -404,16 +424,16 @@ private static void fillUniqueValues(ExecutorService pool, MatrixBlock blkIn, Ma 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; ) { + 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); + for(int i = 0; i < config._numThreads && pos < config._len; i++) { + int end = (int) Math.min((long) pos + config._taskLen, config._len); tasks.add(new UniqueCountTask(blkIn, dir, pos, end)); pos = end; } List> rtasks = pool.invokeAll(tasks); - for( int i = 0; i < rtasks.size(); i++ ) { + for(int i = 0; i < rtasks.size(); i++) { ret = Math.max(ret, rtasks.get(i).get()); rtasks.set(i, null); } @@ -426,16 +446,16 @@ private static int getMaxUniqueValuesBatched(ExecutorService pool, MatrixBlock b */ private static void fillUniqueValuesBatched(ExecutorService pool, MatrixBlock blkIn, MatrixBlock blkOut, Types.Direction dir, BatchConfig config) throws Exception { - for( int pos = 0; pos < config._len; ) { + 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); + for(int i = 0; i < config._numThreads && pos < config._len; i++) { + int end = (int) Math.min((long) pos + config._taskLen, config._len); tasks.add(new UniqueOutputTask(blkIn, blkOut, dir, pos, end)); pos = end; } List> rtasks = pool.invokeAll(tasks); - for( int i = 0; i < rtasks.size(); i++ ) { + for(int i = 0; i < rtasks.size(); i++) { rtasks.get(i).get(); rtasks.set(i, null); } @@ -446,12 +466,12 @@ private static void fillUniqueValuesBatched(ExecutorService pool, MatrixBlock bl * 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 + * @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 ) + if(k <= 1 || ((long) blkIn.getNumRows()) * blkIn.getNumColumns() < PAR_UNIQUE_NUMCELL_THRESHOLD) return false; switch(dir) { @@ -467,17 +487,16 @@ private static boolean satisfiesMultiThreadingConstraints(MatrixBlock blkIn, Typ } /** - * Creates balanced half-open ranges [start, end) using the same utility pattern as - * other SystemDS matrix libraries. + * 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 + * @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++ ) + 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; } @@ -485,7 +504,7 @@ private static ArrayList getBalancedRanges(int len, int k) { /** * Caps the number of workers by the number of row or column partitions available. * - * @param k requested degree of parallelism + * @param k requested degree of parallelism * @param len number of rows or columns to partition * @return effective number of worker threads */ @@ -497,29 +516,66 @@ private static int getNumThreads(int k, int len) { * Builds the execution plan for the batched parallel path. * * @param blkIn input matrix block - * @param dir unique direction - * @param k requested degree of parallelism + * @param dir unique direction + * @param k requested degree of parallelism * @return batch configuration, or null if batching would not use parallelism */ - private static BatchConfig getBatchConfig(MatrixBlock blkIn, Types.Direction dir, int k) { + private static BatchConfig getBatchConfig(MatrixBlock blkIn, Types.Direction dir, int k, long budget) { int len = getPartitionLength(blkIn, dir); - long maxBatchIndexes = getMaxLocalDedupIndexes(blkIn, dir); - if( maxBatchIndexes < 2 ) + long maxBatchIndexes = getMaxLocalDedupIndexes(blkIn, dir, budget); + if(maxBatchIndexes < 2) return null; int numThreads = getNumThreads(k, (int) Math.min(len, maxBatchIndexes)); - if( numThreads <= 1 ) + if(numThreads <= 1) return null; int taskLen = Math.max(1, (int) Math.min(Integer.MAX_VALUE, maxBatchIndexes / numThreads)); return new BatchConfig(numThreads, taskLen, len); } + /** + * Builds the batched execution plan for RowCol. Unlike the row-wise and column-wise workers, the RowCol workers + * accumulate over their entire range and the merged set grows toward the total distinct count, so the merged worst + * case is reserved up front and only the remaining budget sizes the concurrent local sets. If not even the merged + * set fits, batching cannot help and the caller falls back to sequential execution. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @param budget budget in bytes for transient deduplication structures + * @return batch configuration, or null if batching would not use parallelism safely + */ + private static BatchConfig getRowColBatchConfig(MatrixBlock blkIn, Types.Direction dir, int k, long budget) { + int len = getPartitionLength(blkIn, dir); + long cellsPerIndex = getCellsPerIndex(blkIn, dir); + if(cellsPerIndex <= 0 || cellsPerIndex > Long.MAX_VALUE / PAR_UNIQUE_BYTES_PER_CELL) + return null; + + // reserve the worst case (all cells distinct) for the merged set + long numCells = (long) len * cellsPerIndex; + if(!fits(numCells, PAR_UNIQUE_BYTES_PER_CELL, budget)) + return null; + long localBudget = budget - numCells * PAR_UNIQUE_BYTES_PER_CELL; + + long bytesPerIndex = cellsPerIndex * PAR_UNIQUE_BYTES_PER_CELL; + int numThreads = getNumThreads(k, (int) Math.min(len, Math.max(1, localBudget / bytesPerIndex))); + if(numThreads <= 1) + return null; + + long maxIdxPerTask = localBudget / (numThreads * bytesPerIndex); + if(maxIdxPerTask < 1) + return null; + + int taskLen = (int) Math.min(Integer.MAX_VALUE, maxIdxPerTask); + return new BatchConfig(numThreads, taskLen, len); + } + /** * Returns the number of rows or columns that define the partition direction. * * @param blkIn input matrix block - * @param dir unique direction + * @param dir unique direction * @return number of partition indexes */ private static int getPartitionLength(MatrixBlock blkIn, Types.Direction dir) { @@ -530,30 +586,81 @@ private static int getPartitionLength(MatrixBlock blkIn, Types.Direction dir) { * Estimates how many partition indexes can be processed in one batch. * * @param blkIn input matrix block - * @param dir unique direction + * @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 ) + private static long getMaxLocalDedupIndexes(MatrixBlock blkIn, Types.Direction dir, long budget) { + long cellsPerIndex = getCellsPerIndex(blkIn, dir); + if(cellsPerIndex <= 0 || cellsPerIndex > Long.MAX_VALUE / PAR_UNIQUE_BYTES_PER_CELL) return 0; - long bytesPerIndex = cellsPerIndex * Double.BYTES * PAR_UNIQUE_LOCAL_BYTES_OVERHEAD; - long maxLocalBytes = Runtime.getRuntime().maxMemory() / PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION; - return maxLocalBytes / bytesPerIndex; + long bytesPerIndex = cellsPerIndex * PAR_UNIQUE_BYTES_PER_CELL; + return budget / bytesPerIndex; + } + + /** + * Default per-operation budget for the transient deduplication structures. + * + * @return budget in bytes + */ + private static long getDefaultLocalBytesBudget() { + return Runtime.getRuntime().maxMemory() / PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION; } /** - * Conservative memory guard for full local deduplication. The estimate includes - * a small overhead factor for set objects. + * Returns the number of cells scanned per partition index (columns per row, or rows per column). * * @param blkIn input matrix block - * @param dir unique direction - * @return true if local deduplication is small enough for the parallel path + * @param dir unique direction + * @return number of cells per row or column + */ + private static long getCellsPerIndex(MatrixBlock blkIn, Types.Direction dir) { + return dir == Types.Direction.Col ? blkIn.getNumRows() : blkIn.getNumColumns(); + } + + /** + * Overflow-safe check for {@code count * bytesPerUnit <= budget}. + * + * @param count number of units + * @param bytesPerUnit bytes charged per unit + * @param budget budget in bytes + * @return true if the total fits the budget */ - private static boolean isLocalDedupMemoryBudgetSafe(MatrixBlock blkIn, Types.Direction dir) { - return getMaxLocalDedupIndexes(blkIn, dir) >= getPartitionLength(blkIn, dir); + private static boolean fits(long count, long bytesPerUnit, long budget) { + if(count <= 0) + return true; + return count <= budget / bytesPerUnit; + } + + /** + * Memory guard deciding whether the full (non-batched) parallel path fits the budget. + * + *

+ * The row-wise and column-wise workers reuse a single set that is cleared per row/column, so only one live set per + * thread is charged, i.e. {@code numThreads * cellsPerIndex * bytesPerCell + * <= budget}. Charging all indexes instead would force batched execution where it is safely avoidable. The RowCol + * workers instead accumulate over their whole range and are merged into a set that grows toward the total distinct + * count, so the all-distinct worst case is charged for the local sets and the merged set alike. + * + * @param blkIn input matrix block + * @param dir unique direction + * @param k requested degree of parallelism + * @param budget budget in bytes for transient deduplication structures + * @return true if the full parallel path is safe; false to use the batched path + */ + private static boolean isLocalDedupMemoryBudgetSafe(MatrixBlock blkIn, Types.Direction dir, int k, long budget) { + int len = getPartitionLength(blkIn, dir); + long cellsPerIndex = getCellsPerIndex(blkIn, dir); + if(len <= 0 || cellsPerIndex <= 0) + return true; + if(cellsPerIndex > Long.MAX_VALUE / PAR_UNIQUE_BYTES_PER_CELL) + return false; + + if(dir == Types.Direction.RowCol) + return fits((long) len * cellsPerIndex, 2 * PAR_UNIQUE_BYTES_PER_CELL, budget); + + int numThreads = getNumThreads(k, len); + return fits(numThreads * cellsPerIndex, PAR_UNIQUE_BYTES_PER_CELL, budget); } /** @@ -566,7 +673,7 @@ 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++ ) + for(int i = 0; i < rlen; i++) blkOut.set(i, 0, iter.next()); blkOut.recomputeNonZeros(); blkOut.examSparsity(); @@ -582,7 +689,7 @@ private static MatrixBlock createRowColOutput(HashSet values) { */ private static MatrixBlock allocateOutputBlock(int rlen, int clen) { MatrixBlock blkOut = new MatrixBlock(rlen, clen, false); - if( rlen > 0 && clen > 0 ) + if(rlen > 0 && clen > 0) blkOut.allocateBlock(); return blkOut; } @@ -619,8 +726,8 @@ private UniqueValueTask(MatrixBlock blkIn, int rl, int ru) { @Override public HashSet call() { HashSet ret = new HashSet<>(); - for( int i = _rl; i < _ru; i++ ) - for( int j = 0; j < _blkIn.getNumColumns(); j++ ) + for(int i = _rl; i < _ru; i++) + for(int j = 0; j < _blkIn.getNumColumns(); j++) ret.add(_blkIn.get(i, j)); return ret; } @@ -646,18 +753,18 @@ private UniqueCountTask(MatrixBlock blkIn, Types.Direction dir, int l, int u) { public Integer call() { HashSet hashSet = new HashSet<>(); int ret = 0; - if( _dir == Types.Direction.Row ) { - for( int i = _l; i < _u; i++ ) { + if(_dir == Types.Direction.Row) { + for(int i = _l; i < _u; i++) { hashSet.clear(); - for( int j = 0; j < _blkIn.getNumColumns(); j++ ) + 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++ ) { + for(int j = _l; j < _u; j++) { hashSet.clear(); - for( int i = 0; i < _blkIn.getNumRows(); i++ ) + for(int i = 0; i < _blkIn.getNumRows(); i++) hashSet.add(_blkIn.get(i, j)); ret = Math.max(ret, hashSet.size()); } @@ -687,23 +794,23 @@ private UniqueOutputTask(MatrixBlock blkIn, MatrixBlock blkOut, Types.Direction @Override public Void call() { HashSet hashSet = new HashSet<>(); - if( _dir == Types.Direction.Row ) { - for( int i = _l; i < _u; i++ ) { + if(_dir == Types.Direction.Row) { + for(int i = _l; i < _u; i++) { hashSet.clear(); - for( int j = 0; j < _blkIn.getNumColumns(); j++ ) + for(int j = 0; j < _blkIn.getNumColumns(); j++) hashSet.add(_blkIn.get(i, j)); int pos = 0; - for( Double val : hashSet ) + for(Double val : hashSet) _blkOut.set(i, pos++, val); } } else { - for( int j = _l; j < _u; j++ ) { + for(int j = _l; j < _u; j++) { hashSet.clear(); - for( int i = 0; i < _blkIn.getNumRows(); i++ ) + for(int i = 0; i < _blkIn.getNumRows(); i++) hashSet.add(_blkIn.get(i, j)); int pos = 0; - for( Double val : hashSet ) + for(Double val : hashSet) _blkOut.set(pos++, j, val); } } diff --git a/src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java b/src/test/java/org/apache/sysds/test/functions/unique/LibMatrixSketchUniqueParallelTest.java similarity index 54% rename from src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java rename to src/test/java/org/apache/sysds/test/functions/unique/LibMatrixSketchUniqueParallelTest.java index cf95b593dc3..5d488c9a0bc 100644 --- a/src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java +++ b/src/test/java/org/apache/sysds/test/functions/unique/LibMatrixSketchUniqueParallelTest.java @@ -17,23 +17,26 @@ * under the License. */ -package org.apache.sysds.runtime.matrix.data; +package org.apache.sysds.test.functions.unique; import static org.junit.Assert.assertEquals; import java.util.HashSet; import org.apache.sysds.common.Types; +import org.apache.sysds.runtime.matrix.data.LibMatrixSketch; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; import org.junit.Test; /** - * Tests LibMatrixSketch unique paths with k=1 and k>1. + * Tests LibMatrixSketch unique paths with k=1 and k>1. The batched paths are selected by injecting a small + * local-memory budget, since the heap-derived default budget never triggers them in CI. */ public class LibMatrixSketchUniqueParallelTest { @Test public void testRowColUniqueMatchesBaseline() { MatrixBlock input = new MatrixBlock(20000, 2, false).allocateBlock(); - for( int i = 0; i < input.getNumRows(); i++ ) { + for(int i = 0; i < input.getNumRows(); i++) { input.set(i, 0, i % 7); input.set(i, 1, (i + 3) % 11); } @@ -49,7 +52,7 @@ public void testRowColUniqueMatchesBaseline() { @Test public void testRowUniqueMatchesBaselineAndExpectedValues() { MatrixBlock input = new MatrixBlock(12000, 4, false).allocateBlock(); - for( int i = 0; i < input.getNumRows(); i++ ) { + for(int i = 0; i < input.getNumRows(); i++) { int pattern = i % 4; input.set(i, 0, pattern); input.set(i, 1, pattern); @@ -69,7 +72,7 @@ public void testRowUniqueMatchesBaselineAndExpectedValues() { @Test public void testColumnUniqueMatchesBaselineAndExpectedValues() { MatrixBlock input = new MatrixBlock(4, 5000, false).allocateBlock(); - for( int j = 0; j < input.getNumColumns(); j++ ) { + for(int j = 0; j < input.getNumColumns(); j++) { int pattern = j % 4; input.set(0, j, pattern); input.set(1, j, pattern); @@ -95,7 +98,7 @@ public void testLargerInputsMatchBaseline() { private static void testRowColLargeInput() { MatrixBlock input = new MatrixBlock(1200000, 1, false).allocateBlock(); - for( int i = 0; i < input.getNumRows(); i++ ) + for(int i = 0; i < input.getNumRows(); i++) input.set(i, 0, i % 7); input.recomputeNonZeros(); @@ -108,8 +111,8 @@ private static void testRowColLargeInput() { 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++ ) + for(int i = 0; i < input.getNumRows(); i++) + for(int j = 0; j < input.getNumColumns(); j++) input.set(i, j, j % 4 + 1); input.recomputeNonZeros(); @@ -123,8 +126,8 @@ private static void testRowLargeInput() { 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++ ) + for(int j = 0; j < input.getNumColumns(); j++) + for(int i = 0; i < input.getNumRows(); i++) input.set(i, j, i % 4 + 1); input.recomputeNonZeros(); @@ -136,15 +139,147 @@ private static void testColumnLargeInput() { assertSameColumnSet(parallel, 0, new double[] {1, 2, 3, 4}, "Col large input expected values"); } + /** + * Forces the batched RowCol path with a budget that reserves the all-distinct merged set but leaves too little for + * the full parallel path. + */ + @Test + public void testRowColBatchedPathMatchesBaseline() { + 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(); + + // numCells=40000; full path needs 2*40000*64 bytes, batching needs at least 40000*64 + long budget = 3000000; + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.RowCol); + MatrixBlock batched = LibMatrixSketch.getUniqueValues(input, Types.Direction.RowCol, 4, budget); + + assertDimensions(batched, 11, 1, "RowCol batched dimensions"); + assertSameScalarSet(baseline, batched, "RowCol batched baseline vs parallel"); + } + + /** + * A budget too small to hold even the merged set must fall back to sequential execution and still return the + * correct result. + */ + @Test + public void testRowColTinyBudgetFallsBackToSequential() { + 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 fallback = LibMatrixSketch.getUniqueValues(input, Types.Direction.RowCol, 4, 1024); + + assertDimensions(fallback, 11, 1, "RowCol tiny-budget dimensions"); + assertSameScalarSet(baseline, fallback, "RowCol tiny-budget baseline vs fallback"); + } + + /** + * Forces the batched row-wise path with a budget below one live row set per thread. + */ + @Test + public void testRowBatchedPathMatchesBaseline() { + MatrixBlock input = new MatrixBlock(12000, 4, false).allocateBlock(); + for(int i = 0; i < input.getNumRows(); i++) { + int pattern = i % 4; + input.set(i, 0, pattern); + input.set(i, 1, pattern); + input.set(i, 2, pattern + 10); + input.set(i, 3, pattern + 20); + } + input.recomputeNonZeros(); + + // full path needs numThreads*clen*64 = 4*4*64 = 1024 bytes + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row); + MatrixBlock batched = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row, 4, 512); + + assertDimensions(batched, input.getNumRows(), 3, "Row batched dimensions"); + assertBlockEquals(baseline, batched, "Row batched baseline vs parallel"); + assertSameRowSet(batched, 5, new double[] {1, 11, 21}, "Row batched expected values"); + } + + /** + * Row-wise unique under a budget too small for any batch must fall back to sequential execution. + */ + @Test + public void testRowTinyBudgetFallsBackToSequential() { + MatrixBlock input = new MatrixBlock(12000, 4, false).allocateBlock(); + for(int i = 0; i < input.getNumRows(); i++) { + int pattern = i % 4; + input.set(i, 0, pattern); + input.set(i, 1, pattern); + input.set(i, 2, pattern + 10); + input.set(i, 3, pattern + 20); + } + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row); + MatrixBlock fallback = LibMatrixSketch.getUniqueValues(input, Types.Direction.Row, 4, 64); + + assertBlockEquals(baseline, fallback, "Row tiny-budget baseline vs fallback"); + } + + /** + * Forces the batched column-wise path with a budget below one live column set per thread. + */ + @Test + public void testColumnBatchedPathMatchesBaseline() { + MatrixBlock input = new MatrixBlock(4, 5000, false).allocateBlock(); + for(int j = 0; j < input.getNumColumns(); j++) { + int pattern = j % 4; + input.set(0, j, pattern); + input.set(1, j, pattern); + input.set(2, j, pattern + 10); + input.set(3, j, pattern + 20); + } + input.recomputeNonZeros(); + + // full path needs numThreads*rlen*64 = 4*4*64 = 1024 bytes + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col); + MatrixBlock batched = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col, 4, 512); + + assertDimensions(batched, 3, input.getNumColumns(), "Col batched dimensions"); + assertBlockEquals(baseline, batched, "Col batched baseline vs parallel"); + assertSameColumnSet(batched, 5, new double[] {1, 11, 21}, "Col batched expected values"); + } + + /** + * Column-wise unique under a budget too small for any batch must fall back to sequential execution. + */ + @Test + public void testColumnTinyBudgetFallsBackToSequential() { + MatrixBlock input = new MatrixBlock(4, 5000, false).allocateBlock(); + for(int j = 0; j < input.getNumColumns(); j++) { + int pattern = j % 4; + input.set(0, j, pattern); + input.set(1, j, pattern); + input.set(2, j, pattern + 10); + input.set(3, j, pattern + 20); + } + input.recomputeNonZeros(); + + MatrixBlock baseline = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col); + MatrixBlock fallback = LibMatrixSketch.getUniqueValues(input, Types.Direction.Col, 4, 64); + + assertBlockEquals(baseline, fallback, "Col tiny-budget baseline vs fallback"); + } + /** * Compares two MatrixBlocks cell by cell with exact equality. */ private static void assertBlockEquals(MatrixBlock expected, MatrixBlock actual, String message) { assertDimensions(actual, expected.getNumRows(), expected.getNumColumns(), message); - for( int i = 0; i < expected.getNumRows(); i++ ) - for( int j = 0; j < expected.getNumColumns(); j++ ) - assertEquals(message + " mismatch at (" + i + ", " + j + ")", expected.get(i, j), - actual.get(i, j), 0); + for(int i = 0; i < expected.getNumRows(); i++) + for(int j = 0; j < expected.getNumColumns(); j++) + assertEquals(message + " mismatch at (" + i + ", " + j + ")", expected.get(i, j), actual.get(i, j), 0); } /** @@ -162,7 +297,7 @@ private static void assertSameScalarSet(MatrixBlock expected, MatrixBlock actual */ private static HashSet collectScalars(MatrixBlock block) { HashSet ret = new HashSet<>(); - for( int i = 0; i < block.getNumRows(); i++ ) + for(int i = 0; i < block.getNumRows(); i++) ret.add(block.get(i, 0)); return ret; } @@ -173,7 +308,7 @@ private static HashSet collectScalars(MatrixBlock block) { 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++ ) + for(int j = 0; j < block.getNumColumns(); j++) actual.add(block.get(row, j)); assertEquals(message + " mismatch", expected, actual); } @@ -184,7 +319,7 @@ private static void assertSameRowSet(MatrixBlock block, int row, double[] expect 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++ ) + for(int i = 0; i < block.getNumRows(); i++) actual.add(block.get(i, col)); assertEquals(message + " mismatch", expected, actual); } @@ -194,7 +329,7 @@ private static void assertSameColumnSet(MatrixBlock block, int col, double[] exp */ private static HashSet collectExpected(double[] values) { HashSet ret = new HashSet<>(); - for( double value : values ) + for(double value : values) ret.add(value); return ret; } From 36f48c9ddeec25e43290947391f0cd9fb28932ac Mon Sep 17 00:00:00 2001 From: Isso-W Date: Sat, 25 Jul 2026 03:58:05 +0200 Subject: [PATCH 7/7] [SYSTEMDS-3524] Use InfrastructureAnalyzer for the local memory budget Derive the transient deduplication budget from InfrastructureAnalyzer.getLocalMaxMemory() instead of Runtime.getRuntime().maxMemory(). The analyzer reflects the local JVM memory that parfor adjusts per worker, so the raw JVM maximum would over-estimate the budget available to a parfor worker. This also matches the convention used by the neighbouring matrix libraries (e.g. LibMatrixMult). Also expand the direction dispatch into explicit if/return branches so the formatter does not wrap the calls mid-argument. --- .../runtime/matrix/data/LibMatrixSketch.java | 21 ++++++++++++------- 1 file changed, 13 insertions(+), 8 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 92ecc450de8..3067caa2434 100644 --- a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketch.java +++ b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketch.java @@ -31,6 +31,7 @@ import org.apache.sysds.runtime.DMLRuntimeException; import org.apache.sysds.runtime.util.CommonThreadPool; import org.apache.sysds.runtime.util.UtilFunctions; +import org.apache.sysds.utils.stats.InfrastructureAnalyzer; public class LibMatrixSketch { private static final long PAR_UNIQUE_NUMCELL_THRESHOLD = 1024 * 16; @@ -87,14 +88,17 @@ public static MatrixBlock getUniqueValues(MatrixBlock blkIn, Types.Direction dir boolean localDedupMemorySafe = isLocalDedupMemoryBudgetSafe(blkIn, dir, k, maxLocalBytes); switch(dir) { case RowCol: - return localDedupMemorySafe ? getUniqueValuesRowColParallel(blkIn, - k) : getUniqueValuesRowColBatchedParallel(blkIn, dir, k, maxLocalBytes); + if(localDedupMemorySafe) + return getUniqueValuesRowColParallel(blkIn, k); + return getUniqueValuesRowColBatchedParallel(blkIn, dir, k, maxLocalBytes); case Row: - return localDedupMemorySafe ? getUniqueRowValuesParallel(blkIn, - k) : getUniqueRowValuesBatchedParallel(blkIn, dir, k, maxLocalBytes); + if(localDedupMemorySafe) + return getUniqueRowValuesParallel(blkIn, k); + return getUniqueRowValuesBatchedParallel(blkIn, dir, k, maxLocalBytes); case Col: - return localDedupMemorySafe ? getUniqueColumnValuesParallel(blkIn, - k) : getUniqueColumnValuesBatchedParallel(blkIn, dir, k, maxLocalBytes); + if(localDedupMemorySafe) + return getUniqueColumnValuesParallel(blkIn, k); + return getUniqueColumnValuesBatchedParallel(blkIn, dir, k, maxLocalBytes); default: throw new IllegalArgumentException("Unrecognized direction: " + dir); } @@ -599,12 +603,13 @@ private static long getMaxLocalDedupIndexes(MatrixBlock blkIn, Types.Direction d } /** - * Default per-operation budget for the transient deduplication structures. + * Default per-operation budget for the transient deduplication structures. This uses the infrastructure-analyzer + * view of the local JVM memory, which parfor adjusts per worker, rather than the raw JVM maximum. * * @return budget in bytes */ private static long getDefaultLocalBytesBudget() { - return Runtime.getRuntime().maxMemory() / PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION; + return InfrastructureAnalyzer.getLocalMaxMemory() / PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION; } /**