[SYSTEMDS-3524] Add multi-threaded unique in LibMatrixSketch

[shieru1214]

Summary

This PR adds a multi-threaded path for unique in LibMatrixSketch.

The existing single-threaded implementation is kept and is still used as the baseline/fallback. I added a new overload: getUniqueValues(MatrixBlock blkIn, Types.Direction dir, int k), where k controls the requested number of threads.

The implementation follows the existing SystemDS unique direction semantics:

• RowCol: unique scalar values over the input matrix, returned as one column
• Row: row-wise unique scalar values, preserving the number of rows
• Col: column-wise unique scalar values, preserving the number of columns

Implementation

For k > 1 and sufficiently large inputs, the code uses balanced row or column ranges and CommonThreadPool.

The parallel logic is direction-specific:

• RowCol: each worker builds a local HashSet<Double> over its row range, and the local sets are merged afterwards.
• Row: rows are partitioned across workers. The code first computes the maximum number of unique values per row, allocates the output, and then fills the row-wise unique values in parallel.
• Col: columns are partitioned across workers. The code first computes the maximum number of unique values per column, allocates the output, and then fills the column-wise unique values in parallel.

There are two parallel paths:

• regular parallel deduplication, used when the estimated local memory usage is safe
• batched parallel deduplication, used when full thread-local temporary data may be too large

The batched path processes a limited number of ranges at a time, clears temporary local data earlier, and then continues with the next batch. If the input is small, k <= 1, or batching is not useful, the code falls back to the original single-threaded path.

Testing

I added a focused JUnit test:

• src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.java

It checks RowCol, Row, and Col, and compares the multi-threaded result against the single-threaded baseline.

I also ran the existing unique tests:

• mvn -Dtest=UniqueRow,UniqueRowCol test

I ran the focused test locally as well, including one run with a smaller JVM heap to exercise the memory-aware path.

Runtime experiments

I ran local runtime experiments with fixed k=4, comparing against k=1.

The experiment varies:

• direction: RowCol, Row, Col
• input size: small and large
• unique share: 1%, 10%, 50%, 100%
• Unique Values means:
  • for RowCol: distinct scalar values over the input
  • for Row: distinct scalar values per row
  • for Col: distinct scalar values per column
• Speedup is computed as k=1 time / k=4 time, so values above 1 mean that the multi-threaded path was faster.

The results are mostly in line with the expected trade-off. The row-wise and column-wise cases benefit from the parallel path because the work can be partitioned by independent rows or columns. In contrast, RowCol shows limited benefit in these local runs, since the thread-local sets still need to be merged into one global set and the merge overhead can dominate, especially when the number of unique scalar values is high.

The table below shows the local runtime results. These numbers are from local runs and are meant as an initial comparison, not as a full benchmark.

Direction	Size	Input	Unique Share	Unique Values	k=1 (ms)	k=4 (ms)	Speedup
RowCol	small	100000x1	1%	1000	4.728	17.376	0.27
RowCol	small	100000x1	10%	10000	4.645	26.974	0.17
RowCol	small	100000x1	50%	50000	8.205	23.694	0.35
RowCol	small	100000x1	100%	100000	4.026	8.222	0.49
RowCol	large	2000000x1	1%	20000	21.942	25.846	0.85
RowCol	large	2000000x1	10%	200000	40.109	104.492	0.38
RowCol	large	2000000x1	50%	1000000	78.907	129.610	0.61
RowCol	large	2000000x1	100%	2000000	82.564	89.095	0.93
Row	small	10000x64	1%	1	17.506	3.557	4.92
Row	small	10000x64	10%	6	12.252	4.836	2.53
Row	small	10000x64	50%	32	15.368	6.129	2.51
Row	small	10000x64	100%	64	20.363	8.750	2.33
Row	large	200000x64	1%	1	298.942	197.641	1.51
Row	large	200000x64	10%	6	250.292	78.065	3.21
Row	large	200000x64	50%	32	316.849	113.942	2.78
Row	large	200000x64	100%	64	407.338	166.244	2.45
Col	small	64x10000	1%	1	14.698	4.441	3.31
Col	small	64x10000	10%	6	16.239	5.173	3.14
Col	small	64x10000	50%	32	21.762	7.588	2.87
Col	small	64x10000	100%	64	23.307	10.271	2.27
Col	large	64x200000	1%	1	305.964	94.816	3.23
Col	large	64x200000	10%	6	349.636	108.726	3.22
Col	large	64x200000	50%	32	469.790	159.105	2.95
Col	large	64x200000	100%	64	557.950	235.712	2.37