fix hash nnz

Author: Baunsgaard

src/main/java/org/apache/sysds/runtime/frame/data/columns/ABooleanArray.java

@@ -59,7 +59,7 @@ public boolean possiblyContainsNaN() {
 
 
 	@Override
-	protected Map<Boolean, Integer> createRecodeMap(int estimate, ExecutorService pool) {
+	protected Map<Boolean, Integer> createRecodeMap(int estimate, ExecutorService pool, int k) {
 		Map<Boolean, Integer> map = new HashMap<>();
 		int id = 1;
 		for(int i = 0; i < size() && id <= 2; i++) {

src/main/java/org/apache/sysds/runtime/frame/data/columns/Array.java

@@ -34,7 +34,6 @@
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.io.Writable;
 import org.apache.sysds.common.Types.ValueType;
-import org.apache.sysds.hops.OptimizerUtils;
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.compress.colgroup.mapping.AMapToData;
 import org.apache.sysds.runtime.compress.estim.sample.SampleEstimatorFactory;
@@ -137,7 +136,7 @@ public synchronized final Map<T, Integer> getRecodeMap(int estimate, ExecutorSer
 			return map;
 
 		// construct recode map
-		map = createRecodeMap(estimate, pool);
+		map = createRecodeMap(estimate, pool, k);
 
 		// put created map into cache
 		setCache(new SoftReference<>(map));
@@ -153,22 +152,23 @@ public synchronized final Map<T, Integer> getRecodeMap(int estimate, ExecutorSer
 	 * @param estimate The estimate number of unique values inside this array.
 	 * @param pool     The thread pool to use for parallel creation of recode map (can be null). (Note this method does
 	 *                 not shutdown the pool)
+	 * @param k	       The allowed degree of parallelism
 	 * @return The recode map created.
 	 * @throws ExecutionException   if the parallel execution fails
 	 * @throws InterruptedException if the parallel execution fails
 	 */
-	protected Map<T, Integer> createRecodeMap(int estimate, ExecutorService pool)
+	protected Map<T, Integer> createRecodeMap(int estimate, ExecutorService pool, int k)
 		throws InterruptedException, ExecutionException {
-		Timing t = new Timing();
+		final boolean debug = LOG.isDebugEnabled();
+		final Timing t = debug ? new Timing() : null;
 		final int s = size();
-		int k = OptimizerUtils.getTransformNumThreads();
-		Map<T, Integer> ret;
+		final Map<T, Integer> ret;
 		if(k <= 1 || pool == null || s < ROW_PARALLELIZATION_THRESHOLD)
 			ret = createRecodeMap(estimate, 0, s);
 		else
 			ret = parallelCreateRecodeMap(estimate, pool, s, k);
 
-		if(LOG.isDebugEnabled()) {
+		if(debug) {
 			String base = "CreateRecodeMap estimate: %10d actual %10d time: %10.5f";
 			LOG.debug(String.format(base, estimate, ret.size(), t.stop()));
 		}
@@ -230,12 +230,9 @@ private Map<T, Integer> createRecodeMap(Map<T, Integer> map, final int s, final
 	}
 
 	protected int addValRecodeMap(Map<T, Integer> map, int id, int i) {
-		T val = getInternal(i);
-		if(val != null) {
-			Integer v = map.putIfAbsent(val, id);
-			if(v == null)
-				id++;
-		}
+		final T val = getInternal(i);
+		if(val != null && map.putIfAbsent(val, id) == null) 
+			id++;
 		return id;
 	}
 

src/main/java/org/apache/sysds/runtime/frame/data/columns/DDCArray.java

@@ -178,9 +178,9 @@ public static <T> Array<T> compressToDDC(Array<T> arr, int estimateUnique) {
 	}
 
 	@Override
-	protected Map<T, Integer> createRecodeMap(int estimate, ExecutorService pool)
+	protected Map<T, Integer> createRecodeMap(int estimate, ExecutorService pool, int k)
 		throws InterruptedException, ExecutionException {
-		return dict.createRecodeMap(estimate, pool);
+		return dict.createRecodeMap(estimate, pool, k);
 	}
 
 	@Override

src/main/java/org/apache/sysds/runtime/frame/data/columns/StringArray.java

@@ -669,7 +669,7 @@ public final boolean isNotEmpty(int i) {
 	}
 
 	@Override
-	protected Map<String, Integer> createRecodeMap(int estimate, ExecutorService pool) throws InterruptedException, ExecutionException {
+	protected Map<String, Integer> createRecodeMap(int estimate, ExecutorService pool, int k) throws InterruptedException, ExecutionException {
 		try {
 			Map<String, Integer> map = new HashMap<>((int) Math.min((long) estimate * 2, size()));
 			for(int i = 0; i < size(); i++) {
@@ -682,7 +682,7 @@ protected Map<String, Integer> createRecodeMap(int estimate, ExecutorService poo
 			return map;
 		}
 		catch(Exception e) {
-			return super.createRecodeMap(estimate, pool);
+			return super.createRecodeMap(estimate, pool, k);
 		}
 	}
 

src/main/java/org/apache/sysds/runtime/transform/encode/CompressedEncode.java

@@ -531,6 +531,7 @@ private AColGroup hash(ColumnEncoderComposite c) {
 
 		AMapToData m = createHashMappingAMapToData(a, domain, nulls);
 		AColGroup ret = ColGroupDDC.create(colIndexes, d, m, null);
+		nnz.addAndGet(ret.getNumberNonZeros(in.getNumRows()));
 		return ret;
 	}
 
@@ -545,7 +546,9 @@ private AColGroup hashToDummy(ColumnEncoderComposite c) {
 			return ColGroupConst.create(colIndexes, new double[] {1});
 		ADictionary d = new IdentityDictionary(colIndexes.size(), nulls);
 		AMapToData m = createHashMappingAMapToData(a, domain, nulls);
-		return ColGroupDDC.create(colIndexes, d, m, null);
+		AColGroup ret = ColGroupDDC.create(colIndexes, d, m, null);
+		nnz.addAndGet(ret.getNumberNonZeros(in.getNumRows()));
+		return ret;
 	}
 
 	@SuppressWarnings("unchecked")

src/test/java/org/apache/sysds/performance/frame/Transform.java

@@ -113,11 +113,11 @@ public static void main(String[] args) throws Exception {
 					ValueType.UINT4, ValueType.UINT4, ValueType.UINT4, ValueType.UINT4, ValueType.UINT4, ValueType.UINT4},
 				32);
 
-			// System.out.println("10 col without null");
-			// run10(k, in);
-			System.out.println("10 col compressed without null");
-			in = FrameLibCompress.compress(in, k);
+			System.out.println("10 col without null");
 			run10(k, in);
+			// System.out.println("10 col compressed without null");
+			// in = FrameLibCompress.compress(in, k);
+			// run10(k, in);
 
 			in = TestUtils.generateRandomFrameBlock(
 				100000 * i, new ValueType[] {ValueType.UINT4, ValueType.UINT4, ValueType.UINT4, ValueType.UINT4,