[MINOR] CompressedMatrixBlock parallel nonzero count

Author: Baunsgaard

src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java

@@ -28,6 +28,7 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 
 import org.apache.commons.lang3.NotImplementedException;
@@ -88,6 +89,7 @@
 import org.apache.sysds.runtime.matrix.operators.ScalarOperator;
 import org.apache.sysds.runtime.matrix.operators.TernaryOperator;
 import org.apache.sysds.runtime.matrix.operators.UnaryOperator;
+import org.apache.sysds.runtime.util.CommonThreadPool;
 import org.apache.sysds.runtime.util.IndexRange;
 import org.apache.sysds.utils.DMLCompressionStatistics;
 import org.apache.sysds.utils.stats.InfrastructureAnalyzer;
@@ -319,6 +321,35 @@ public long recomputeNonZeros() {
 		return nonZeros;
 	}
 
+	@Override
+	public long recomputeNonZeros(int k) {
+		if(k <= 1 || isOverlapping() || _colGroups.size() <= 1)
+			return recomputeNonZeros();
+
+		final ExecutorService pool = CommonThreadPool.get(k);
+		try {
+			List<Future<Long>> tasks = new ArrayList<>();
+			for(AColGroup g : _colGroups)
+				tasks.add(pool.submit(() -> g.getNumberNonZeros(rlen)));
+
+			long nnz = 0;
+			for(Future<Long> t : tasks)
+				nnz += t.get();
+			nonZeros = nnz;
+		}
+		catch(Exception e) {
+			throw new DMLRuntimeException("Failed to count non zeros", e);
+		}
+		finally {
+			pool.shutdown();
+		}
+
+		if(nonZeros == 0) // If there is no nonzeros then reallocate into single empty column group.
+			allocateColGroup(ColGroupEmpty.create(getNumColumns()));
+
+		return nonZeros;
+	}
+
 	@Override
 	public long recomputeNonZeros(int rl, int ru) {
 		throw new NotImplementedException();