[BWARE] Handle hash columns in transform decoders and tighten decode metadata

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

@@ -607,17 +607,23 @@ public double getAsNaNDouble(int i) {
 
 	private static double getAsDouble(String s) {
 		try {
-
 			return DoubleArray.parseDouble(s);
 		}
 		catch(Exception e) {
-			String ls = s.toLowerCase();
-			if(ls.equals("true") || ls.equals("t"))
+			// fallback for boolean-like tokens, without allocating a lower-cased copy
+			final int len = s.length();
+			if(len == 1) {
+				final char c = s.charAt(0);
+				if(c == 't' || c == 'T')
+					return 1;
+				else if(c == 'f' || c == 'F')
+					return 0;
+			}
+			else if(len == 4 && s.compareToIgnoreCase("true") == 0)
 				return 1;
-			else if(ls.equals("false") || ls.equals("f"))
+			else if(len == 5 && s.compareToIgnoreCase("false") == 0)
 				return 0;
-			else
-				throw new DMLRuntimeException("Unable to change to double: " + s, e);
+			throw new DMLRuntimeException("Unable to change to double: " + s, e);
 		}
 	}
 

src/main/java/org/apache/sysds/runtime/transform/decode/Decoder.java

@@ -23,13 +23,21 @@
 import java.io.IOException;
 import java.io.ObjectInput;
 import java.io.ObjectOutput;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
 
+import org.apache.commons.lang3.ArrayUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.sysds.common.Types.ValueType;
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.frame.data.FrameBlock;
+import org.apache.sysds.runtime.frame.data.columns.ColumnMetadata;
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.util.CommonThreadPool;
+import org.apache.sysds.runtime.util.UtilFunctions;
 
 /**
  * Base class for all transform decoders providing both a row and block
@@ -43,11 +51,38 @@ public abstract class Decoder implements Externalizable{
 	protected ValueType[] _schema;
 	protected int[] _colList;
 	protected String[] _colnames = null;
+	// dummycoded columns that were feature-hashed: domain size K is read from the meta cell, not
+	// numDistinct. Only used during initMetaData (driver side), so not serialized.
+	protected transient int[] _dcHashCols = null;
+
 	protected Decoder(ValueType[] schema, int[] colList) {
 		_schema = schema;
 		_colList = colList;
 	}
 
+	protected boolean isHashCol(int colID) {
+		return ArrayUtils.contains(_dcHashCols, colID);
+	}
+
+	/**
+	 * Domain size of a dummycoded source column: the hash domain K from the meta cell for
+	 * feature-hashed columns, otherwise the column's {@code numDistinct} (0 when unset).
+	 *
+	 * @param meta   transform meta frame
+	 * @param colID  1-based column id of the dummycoded source column
+	 * @param isHash whether the column was feature-hashed
+	 * @return the domain size, never negative
+	 */
+	protected static int getNumDummycodeDistinct(FrameBlock meta, int colID, boolean isHash) {
+		if(isHash) {
+			Object o = meta.get(0, colID - 1);
+			return (o == null) ? 0 : (int) UtilFunctions.parseToLong(o.toString());
+		}
+		ColumnMetadata d = meta.getColumnMetadata()[colID - 1];
+		int ndist = d.isDefault() ? 0 : (int) d.getNumDistinct();
+		return Math.max(ndist, 0);
+	}
+
 	public ValueType[] getSchema() {
 		return _schema;
 	}
@@ -77,8 +112,31 @@ public String[] getColnames() {
 	 * @param k   Parallelization degree
 	 * @return returns the given output frame block for convenience
 	 */
-	public FrameBlock decode(MatrixBlock in, FrameBlock out, int k) {
-		return decode(in, out);
+	public FrameBlock decode(final MatrixBlock in, final FrameBlock out, final int k) {
+		if(k <= 1)
+			return decode(in, out);
+		final ExecutorService pool = CommonThreadPool.get(k);
+		out.ensureAllocatedColumns(in.getNumRows());
+		try {
+			final List<Future<?>> tasks = new ArrayList<>();
+			int blz = Math.max((in.getNumRows() + k) / k, 1000);
+
+			for(int i = 0; i < in.getNumRows(); i += blz){
+				final int start = i;
+				final int end = Math.min(in.getNumRows(), i + blz);
+				tasks.add(pool.submit(() -> decode(in, out, start, end)));
+			}
+
+			for(Future<?> f : tasks)
+				f.get();
+			return out;
+		}
+		catch(Exception e) {
+			throw new RuntimeException(e);
+		}
+		finally {
+			pool.shutdown();
+		}
 	}
 
 	/**

src/main/java/org/apache/sysds/runtime/transform/decode/DecoderBin.java

@@ -42,16 +42,23 @@ public class DecoderBin extends Decoder {
 	private static final long serialVersionUID = -3784249774608228805L;
 
 	// a) column bin boundaries
-	private int[] _numBins;
+	private int[] _dcCols = null;
+	private int[] _srcCols = null;
 	private double[][] _binMins = null;
 	private double[][] _binMaxs = null;
 
 	public DecoderBin() {
 		super(null, null);
 	}
 
-	protected DecoderBin(ValueType[] schema, int[] binCols) {
+	protected DecoderBin(ValueType[] schema, int[] binCols, int[] dcCols) {
+		this(schema, binCols, dcCols, null);
+	}
+
+	protected DecoderBin(ValueType[] schema, int[] binCols, int[] dcCols, int[] hashCols) {
 		super(schema, binCols);
+		_dcCols = dcCols;
+		_dcHashCols = hashCols;
 	}
 
 	@Override
@@ -66,14 +73,28 @@ public void decode(MatrixBlock in, FrameBlock out, int rl, int ru) {
 		for( int i=rl; i< ru; i++ ) {
 			for( int j=0; j<_colList.length; j++ ) {
 				final Array<?> a = out.getColumn(_colList[j] - 1);
-				final double val = in.get(i, _colList[j] - 1);
+				final double val = in.get(i, _srcCols[j] - 1);
 				if(!Double.isNaN(val)){
-					final int key = (int) Math.round(val);
-					double bmin = _binMins[j][key - 1];
-					double bmax = _binMaxs[j][key - 1];
-					double oval = bmin + (bmax - bmin) / 2 // bin center
-						+ (val - key) * (bmax - bmin); // bin fractions
-					a.set(i, oval);
+					try{
+
+						final int key = (int) Math.round(val);
+						if(key == 0){
+							a.set(i, _binMins[j][key]);
+						}
+						else{
+							double bmin = _binMins[j][key - 1];
+							double bmax = _binMaxs[j][key - 1];
+							double oval = bmin + (bmax - bmin) / 2 // bin center
+								+ (val - key) * (bmax - bmin); // bin fractions
+							a.set(i, oval);
+						}
+					}
+					catch(Exception e){
+						LOG.error(a);
+						LOG.error(in.slice(0, in.getNumRows()-1, _colList[j]-1,_colList[j]-1));
+						LOG.error( val);
+						throw e;
+					}
 				}
 				else 
 					a.set(i, val); // NaN
@@ -90,7 +111,6 @@ public Decoder subRangeDecoder(int colStart, int colEnd, int dummycodedOffset) {
 	@Override
 	public void initMetaData(FrameBlock meta) {
 		//initialize bin boundaries
-		_numBins = new int[_colList.length];
 		_binMins = new double[_colList.length][];
 		_binMaxs = new double[_colList.length][];
 
@@ -111,34 +131,72 @@ public void initMetaData(FrameBlock meta) {
 				_binMaxs[j][i] = Double.parseDouble(parts[1]);
 			}
 		}
+
+
+		if( _dcCols.length > 0 ) {
+			//prepare source column id mapping w/ dummy coding
+			_srcCols = new int[_colList.length];
+			int ix1 = 0, ix2 = 0, off = 0;
+			while( ix1<_colList.length ) {
+				if( ix2>=_dcCols.length || _colList[ix1] < _dcCols[ix2] ) {
+					_srcCols[ix1] = _colList[ix1] + off;
+					ix1 ++;
+				}
+				else { //_colList[ix1] > _dcCols[ix2]
+					int dcCol = _dcCols[ix2];
+					off += getNumDummycodeDistinct(meta, dcCol, isHashCol(dcCol)) - 1;
+					ix2 ++;
+				}
+			}
+		}
+		else {
+			//prepare direct source column mapping
+			_srcCols = _colList;
+		}
 	}
 
 	@Override
 	public void writeExternal(ObjectOutput out) throws IOException {
 		super.writeExternal(out);
+		// bin boundaries; the per-column bin count is the length of the boundary arrays
 		for( int i=0; i<_colList.length; i++ ) {
-			int len = _numBins[i];
+			int len = _binMins[i].length;
 			out.writeInt(len);
 			for(int j=0; j<len; j++) {
 				out.writeDouble(_binMins[i][j]);
 				out.writeDouble(_binMaxs[i][j]);
 			}
 		}
+		// source-column mapping (rebuilt in initMetaData, but persisted for Spark broadcast)
+		out.writeInt(_srcCols.length);
+		for(int i = 0; i < _srcCols.length; i++)
+			out.writeInt(_srcCols[i]);
+
+		out.writeInt(_dcCols == null ? 0 : _dcCols.length);
+		for(int i = 0; _dcCols != null && i < _dcCols.length; i++)
+			out.writeInt(_dcCols[i]);
 	}
 
 	@Override
 	public void readExternal(ObjectInput in) throws IOException {
 		super.readExternal(in);
-		_numBins = new int[_colList.length];
 		_binMins = new double[_colList.length][];
 		_binMaxs = new double[_colList.length][];
 		for( int i=0; i<_colList.length; i++ ) {
 			int len = in.readInt();
-			_numBins[i] = len;
+			_binMins[i] = new double[len];
+			_binMaxs[i] = new double[len];
 			for(int j=0; j<len; j++) {
 				_binMins[i][j] = in.readDouble();
 				_binMaxs[i][j] = in.readDouble();
 			}
 		}
+		_srcCols = new int[in.readInt()];
+		for(int i = 0; i < _srcCols.length; i++)
+			_srcCols[i] = in.readInt();
+
+		_dcCols = new int[in.readInt()];
+		for(int i = 0; i < _dcCols.length; i++)
+			_dcCols[i] = in.readInt();
 	}
 }

src/main/java/org/apache/sysds/runtime/transform/decode/DecoderComposite.java

@@ -25,13 +25,10 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
 
 import org.apache.sysds.common.Types.ValueType;
 import org.apache.sysds.runtime.frame.data.FrameBlock;
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
-import org.apache.sysds.runtime.util.CommonThreadPool;
 
 /**
  * Simple composite decoder that applies a list of decoders
@@ -59,33 +56,6 @@ public FrameBlock decode(MatrixBlock in, FrameBlock out) {
 		return out;
 	}
 
-
-	@Override
-	public FrameBlock decode(final MatrixBlock in, final FrameBlock out, final int k) {
-		final ExecutorService pool = CommonThreadPool.get(k);
-		out.ensureAllocatedColumns(in.getNumRows());
-		try {
-			final List<Future<?>> tasks = new ArrayList<>();
-			int blz = Math.max(in.getNumRows() / k, 1000);
-			for(Decoder decoder : _decoders){
-				for(int i = 0; i < in.getNumRows(); i += blz){
-					final int start = i;
-					final int end = Math.min(in.getNumRows(), i + blz);
-					tasks.add(pool.submit(() -> decoder.decode(in, out, start, end)));
-				}
-			}
-			for(Future<?> f : tasks)
-				f.get();
-			return out;
-		}
-		catch(Exception e) {
-			throw new RuntimeException(e);
-		}
-		finally {
-			pool.shutdown();
-		}
-	}
-
 	@Override
 	public void decode(MatrixBlock in, FrameBlock out, int rl, int ru){
 		for( Decoder decoder : _decoders )