[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,25 @@ 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. Dispatch on length first so non-boolean strings are
+			// rejected immediately, and avoid allocating a lower-cased copy by comparing case-insensitively
+			// (single char compare for the 1-char tokens).
+			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,18 @@
 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.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.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.util.CommonThreadPool;
 
 /**
  * Base class for all transform decoders providing both a row and block
@@ -77,8 +82,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

@@ -28,6 +28,7 @@
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.frame.data.FrameBlock;
 import org.apache.sysds.runtime.frame.data.columns.Array;
+import org.apache.sysds.runtime.frame.data.columns.ColumnMetadata;
 import org.apache.sysds.runtime.matrix.data.MatrixBlock;
 import org.apache.sysds.runtime.util.UtilFunctions;
 
@@ -43,15 +44,18 @@ public class DecoderBin extends Decoder {
 
 	// 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) {
 		super(schema, binCols);
+		_dcCols = dcCols;
 	}
 
 	@Override
@@ -66,14 +70,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
@@ -111,6 +129,34 @@ 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]
+					ColumnMetadata d =meta.getColumnMetadata()[_dcCols[ix2]-1];
+					String v = meta.getString(0, _dcCols[ix2]-1);
+					if(v.length() > 1 && v.charAt(0) == '¿'){
+						off += UtilFunctions.parseToLong(v.substring(1)) -1;
+					}
+					else {
+						off += d.isDefault() ? -1 : d.getNumDistinct() - 1;
+					}
+					ix2 ++;
+				}
+			}
+		}
+		else {
+			//prepare direct source column mapping
+			_srcCols = _colList;
+		}
 	}
 
 	@Override

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 )