[SYSTEMDS-3644] Compressed transform encode

Author: Baunsgaard

src/main/java/org/apache/sysds/runtime/frame/data/FrameBlock.java

@@ -1267,8 +1267,8 @@ public void copy(int rl, int ru, int cl, int cu, FrameBlock src) {
 	 * @param col is the column # from frame data which contains Recode map generated earlier.
 	 * @return map of token and code for every element in the input column of a frame containing Recode map
 	 */
-	public Map<Object, Long> getRecodeMap(int col) {
-		return _coldata[col].getRecodeMap();
+	public Map<Object, Integer> getRecodeMap(int col) {
+		return _coldata[col].getRecodeMap(4);
 	}
 
 	@Override

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

@@ -19,8 +19,12 @@
 
 package org.apache.sysds.runtime.frame.data.columns;
 
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 
 public abstract class ABooleanArray extends Array<Boolean> {
 
@@ -55,13 +59,25 @@ public boolean possiblyContainsNaN() {
 	 * @param value The string array to set from.
 	 */
 	public abstract void setNullsFromString(int rl, int ru, Array<String> value);
-	
+
+	@Override
+	protected void mergeRecodeMaps(Map<Boolean, Integer> target, Map<Boolean, Integer> from) {
+		final List<Boolean> fromEntriesOrdered = new ArrayList<>(Collections.nCopies(from.size(), null));
+		for(Map.Entry<Boolean, Integer> e : from.entrySet())
+			fromEntriesOrdered.set(e.getValue() - 1, e.getKey());
+		int id = target.size();
+		for(Boolean e : fromEntriesOrdered) {
+			if(target.putIfAbsent(e, id) == null)
+				id++;
+		}
+	}
+
 	@Override
-	protected Map<Boolean, Long> createRecodeMap() {
-		Map<Boolean, Long> map = new HashMap<>();
-		long id = 1;
+	protected Map<Boolean, Integer> createRecodeMap(int estimate, ExecutorService pool) {
+		Map<Boolean, Integer> map = new HashMap<>();
+		int id = 1;
 		for(int i = 0; i < size() && id <= 2; i++) {
-			Long v = map.putIfAbsent(get(i), id);
+			Integer v = map.putIfAbsent(get(i), id);
 			if(v == null)
 				id++;
 		}

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

@@ -19,8 +19,12 @@
 
 package org.apache.sysds.runtime.frame.data.columns;
 
+import java.util.Map;
+
+import org.apache.commons.lang3.NotImplementedException;
 import org.apache.sysds.common.Types.ValueType;
 import org.apache.sysds.runtime.compress.DMLCompressionException;
+import org.apache.sysds.runtime.compress.colgroup.mapping.AMapToData;
 import org.apache.sysds.runtime.frame.data.compress.ArrayCompressionStatistics;
 
 /**
@@ -59,11 +63,6 @@ public void setFromOtherType(int rl, int ru, Array<?> value) {
 		throw new DMLCompressionException("Invalid to set value in CompressedArray");
 	}
 
-	@Override
-	public void set(int rl, int ru, Array<T> value, int rlSrc) {
-		throw new DMLCompressionException("Invalid to set value in CompressedArray");
-	}
-
 	@Override
 	public void setNz(int rl, int ru, Array<T> value) {
 		throw new DMLCompressionException("Invalid to set value in CompressedArray");
@@ -154,4 +153,15 @@ protected Array<Object> changeTypeHash64(Array<Object> retA, int l, int u) {
 	protected Array<Object> changeTypeHash32(Array<Object> ret, int l, int u) {
 		throw new DMLCompressionException("Invalid to change sub compressed array");
 	}
+
+	@Override
+	public void setM(Map<T, Integer> map, AMapToData m, int i) {
+		throw new NotImplementedException();
+	}
+
+	@Override
+	protected void mergeRecodeMaps(Map<T, Integer> target, Map<T, Integer> from) {
+		throw new DMLCompressionException("Invalid to change sub compressed array");
+	}
+
 }

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

@@ -20,20 +20,26 @@
 package org.apache.sysds.runtime.frame.data.columns;
 
 import java.lang.ref.SoftReference;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 
 import org.apache.commons.logging.Log;
 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;
 import org.apache.sysds.runtime.frame.data.columns.ArrayFactory.FrameArrayType;
 import org.apache.sysds.runtime.frame.data.compress.ArrayCompressionStatistics;
 import org.apache.sysds.runtime.matrix.data.Pair;
+import org.apache.sysds.utils.stats.Timing;
 
 /**
  * Generic, resizable native arrays for the internal representation of the columns in the FrameBlock. We use this custom
@@ -43,7 +49,7 @@ public abstract class Array<T> implements Writable {
 	protected static final Log LOG = LogFactory.getLog(Array.class.getName());
 
 	/** A soft reference to a memorization of this arrays mapping, used in transformEncode */
-	protected SoftReference<Map<T, Long>> _rcdMapCache = null;
+	protected SoftReference<Map<T, Integer>> _rcdMapCache = null;
 
 	/** The current allocated number of elements in this Array */
 	protected int _size;
@@ -63,7 +69,7 @@ protected int newSize() {
 	 * 
 	 * @return The cached recode map
 	 */
-	public final SoftReference<Map<T, Long>> getCache() {
+	public final SoftReference<Map<T, Integer>> getCache() {
 		return _rcdMapCache;
 	}
 
@@ -72,7 +78,7 @@ public final SoftReference<Map<T, Long>> getCache() {
 	 * 
 	 * @param m The element to cache.
 	 */
-	public final void setCache(SoftReference<Map<T, Long>> m) {
+	public final void setCache(SoftReference<Map<T, Integer>> m) {
 		_rcdMapCache = m;
 	}
 
@@ -83,43 +89,109 @@ public final void setCache(SoftReference<Map<T, Long>> m) {
 	 * 
 	 * @return A recode map
 	 */
-	public synchronized final Map<T, Long> getRecodeMap() {
+	public synchronized final Map<T, Integer> getRecodeMap() {
+		return getRecodeMap(4);
+	}
+
+	public synchronized final Map<T, Integer> getRecodeMap(int estimate) {
+		return getRecodeMap(estimate, null);
+	}
+
+	/**
+	 * Get a recode map that maps each unique value in the array, to a long ID. Null values are ignored, and not included
+	 * in the mapping. The resulting recode map in stored in a soft reference to speed up repeated calls to the same
+	 * column.
+	 * 
+	 * @param estimate the estimated number of unique.
+	 * @param pool An executor pool to be used for parallel execution
+	 * @return A recode map
+	 */
+	public synchronized final Map<T, Integer> getRecodeMap(int estimate, ExecutorService pool) {
 		// probe cache for existing map
-		Map<T, Long> map;
-		SoftReference<Map<T, Long>> tmp = getCache();
+		Map<T, Integer> map;
+		SoftReference<Map<T, Integer>> tmp = getCache();
 		map = (tmp != null) ? tmp.get() : null;
 		if(map != null)
 			return map;
 
 		// construct recode map
-		map = createRecodeMap();
+		map = createRecodeMap(estimate, pool);
 
 		// put created map into cache
 		setCache(new SoftReference<>(map));
 
 		return map;
 	}
 
-	/**
-	 * Recreate the recode map from what is inside array. This is an internal method for arrays, and the result is cached
-	 * in the main class of the arrays.
-	 * 
-	 * @return The recode map
-	 */
-	protected Map<T, Long> createRecodeMap() {
-		Map<T, Long> map = new HashMap<>();
-		long id = 1;
-		for(int i = 0; i < size(); i++) {
-			T val = get(i);
-			if(val != null) {
-				Long v = map.putIfAbsent(val, id);
-				if(v == null)
-					id++;
+	protected Map<T, Integer> createRecodeMap(int estimate, ExecutorService pool) {
+		Timing t = new Timing();
+		final int s = size();
+		int k = OptimizerUtils.getTransformNumThreads();
+		Map<T, Integer> ret;
+		if(pool == null || s < 10000 || estimate < 1024)
+			ret = createRecodeMap(estimate, 0, s);
+		else
+			ret = parallelCreateRecodeMap(estimate, pool, s, k);
+
+		if(LOG.isDebugEnabled()) {
+			String base = "CreateRecodeMap estimate: %10d actual %10d time: %10.5f";
+			LOG.debug(String.format(base, estimate, ret.size(), t.stop()));
+		}
+		return ret;
+	}
+
+	private Map<T, Integer> parallelCreateRecodeMap(int estimate, ExecutorService pool, final int s, int k) {
+
+		try {
+			final int blk = Math.max(10000, (s + k) / k);
+			final List<Future<Map<T, Integer>>> tasks = new ArrayList<>();
+			for(int i = blk; i < s; i += blk) { // start at blk for the other threads
+				final int start = i;
+				final int end = Math.min(i + blk, s);
+				tasks.add(pool.submit(() -> createRecodeMap(estimate, start, end)));
 			}
+			// make the initial map thread local allocation.
+			final Map<T, Integer> map = new HashMap<>((int) (estimate * 1.3));
+			createRecodeMap(map, 0, blk);
+			for(int i = 0; i < tasks.size(); i++) { // merge with other threads work.
+				final Map<T, Integer> map2 = tasks.get(i).get();
+				mergeRecodeMaps(map, map2);
+			}
+			return map;
+		}
+		catch(Exception e) {
+			throw new RuntimeException(e);
 		}
+		finally {
+			pool.shutdown();
+		}
+	}
+
+	protected abstract void mergeRecodeMaps(Map<T, Integer> target, Map<T, Integer> from);
+
+	private Map<T, Integer> createRecodeMap(final int estimate, final int s, final int e) {
+		// * 1.3 because we hashMap has a load factor of 1.75
+		final Map<T, Integer> map = new HashMap<>((int) (Math.min((long) estimate, (e - s)) * 1.3));
+		return createRecodeMap(map, s, e);
+	}
+
+	private Map<T, Integer> createRecodeMap(Map<T, Integer> map, final int s, final int e) {
+		int id = 1;
+		for(int i = s; i < e; i++)
+			id = addValRecodeMap(map, id, i);
 		return map;
 	}
 
+	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++;
+		}
+		return id;
+	}
+
 	/**
 	 * Get the number of elements in the array, this does not necessarily reflect the current allocated size.
 	 * 
@@ -224,15 +296,10 @@ public double getAsNaNDouble(int i) {
 	 * 
 	 * @param rl    row lower
 	 * @param ru    row upper (inclusive)
-	 * @param value value array to take values from (same type)
+	 * @param value value array to take values from (same type) offset by rl.
 	 */
 	public abstract void set(int rl, int ru, Array<T> value);
 
-	// {
-	// for(int i = rl; i <= ru; i++)
-	// set(i, value.get(i));
-	// }
-
 	/**
 	 * Set range to given arrays value with an offset into other array
 	 * 
@@ -243,7 +310,7 @@ public double getAsNaNDouble(int i) {
 	 */
 	public void set(int rl, int ru, Array<T> value, int rlSrc) {
 		for(int i = rl, off = rlSrc; i <= ru; i++, off++)
-			set(i, value.get(off));
+			set(i, value.getInternal(off));
 	}
 
 	/**
@@ -918,4 +985,22 @@ public double[] minMax(int l, int u) {
 		}
 		return new double[] {min, max};
 	}
+
+	public void setM(Map<T, Integer> map, AMapToData m, int i) {
+		m.set(i, map.get(getInternal(i)).intValue() - 1);
+	}
+
+	public void setM(Map<T, Integer> map, int si, AMapToData m, int i) {
+		try {
+			final T v = getInternal(i);
+			if(v != null)
+				m.set(i, map.get(v).intValue() - 1);
+			else
+				m.set(i, si);
+		}
+		catch(Exception e) {
+			String error = "expected: " + getInternal(i) + " to be in map: " + map;
+			throw new RuntimeException(error, e);
+		}
+	}
 }

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

@@ -356,7 +356,7 @@ else if(target.getFrameArrayType() != FrameArrayType.OPTIONAL //
 
 		Array<C> targetC = (Array<C>) (ta != tc ? target.changeType(tc) : target);
 		Array<C> srcC = (Array<C>) (tb != tc ? src.changeType(tc) : src);
-		targetC.set(rl, ru, srcC);
+		targetC.set(rl, ru, srcC, 0);
 		return targetC;
 
 	}

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

@@ -23,7 +23,11 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
 
 import org.apache.sysds.common.Types.ValueType;
 import org.apache.sysds.runtime.DMLRuntimeException;
@@ -377,6 +381,19 @@ public boolean possiblyContainsNaN() {
 		return false;
 	}
 
+
+	@Override
+	protected void mergeRecodeMaps(Map<Character, Integer> target, Map<Character, Integer> from) {
+		final List<Character> fromEntriesOrdered = new ArrayList<>(Collections.nCopies(from.size(), null));
+		for(Map.Entry<Character, Integer> e : from.entrySet())
+			fromEntriesOrdered.set(e.getValue() - 1, e.getKey());
+		int id = target.size();
+		for(Character e : fromEntriesOrdered) {
+			if(target.putIfAbsent(e, id) == null)
+				id++;
+		}
+	}
+
 	@Override
 	public String toString() {
 		StringBuilder sb = new StringBuilder(_size * 2 + 15);