[BWARE] Add getCategoricalMask DML builtin

src/main/java/org/apache/sysds/common/Builtins.java

@@ -154,6 +154,7 @@ public enum Builtins {
 	GARCH("garch", true),
 	GAUSSIAN_CLASSIFIER("gaussianClassifier", true),
 	GET_ACCURACY("getAccuracy", true),
+	GET_CATEGORICAL_MASK("getCategoricalMask", false),
 	GLM("glm", true),
 	GLM_PREDICT("glmPredict", true),
 	GLOVE("glove", true),

src/main/java/org/apache/sysds/common/Opcodes.java

@@ -215,6 +215,8 @@ public enum Opcodes {
 	TRANSFORMMETA("transformmeta", InstructionType.ParameterizedBuiltin),
 	TRANSFORMENCODE("transformencode", InstructionType.MultiReturnParameterizedBuiltin,  InstructionType.MultiReturnBuiltin),
 
+	GET_CATEGORICAL_MASK("get_categorical_mask", InstructionType.Binary),
+
 	//Ternary instruction opcodes
 	PM("+*", InstructionType.Ternary),
 	MINUSMULT("-*", InstructionType.Ternary),

src/main/java/org/apache/sysds/common/Types.java

@@ -639,6 +639,7 @@ public enum OpOp2 {
 		MINUS_NZ(false), //sparse-safe minus: X-(mean*ppred(X,0,!=))
 		LOG_NZ(false), //sparse-safe log; ppred(X,0,"!=")*log(X,0.5)
 		MINUS1_MULT(false), //1-X*Y
+		GET_CATEGORICAL_MASK(false), // get transformation mask
 		QUANTIZE_COMPRESS(false), //quantization-fused compression
 		UNION_DISTINCT(false);
 

src/main/java/org/apache/sysds/hops/BinaryOp.java

@@ -853,7 +853,10 @@ else if( (op == OpOp2.CBIND && getDataType().isList())
 				|| (op == OpOp2.RBIND && getDataType().isList())) {
 			_etype = ExecType.CP;
 		}
-
+
+		if( op == OpOp2.GET_CATEGORICAL_MASK)
+			_etype = ExecType.CP;
+
 		//mark for recompile (forever)
 		setRequiresRecompileIfNecessary();
 

src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java

@@ -2018,6 +2018,15 @@ else if(this.getOpCode() == Builtins.MAX_POOL || this.getOpCode() == Builtins.AV
 			else
 				raiseValidateError("The compress or decompress instruction is not allowed in dml scripts");
 			break;
+		case GET_CATEGORICAL_MASK: 
+			checkNumParameters(2);
+			checkFrameParam(getFirstExpr());
+			checkScalarParam(getSecondExpr());
+			output.setDataType(DataType.MATRIX);
+			output.setDimensions(1, -1);
+			output.setBlocksize( id.getBlocksize());
+			output.setValueType(ValueType.FP64);
+			break;
 		case QUANTIZE_COMPRESS:
 			if(OptimizerUtils.ALLOW_SCRIPT_LEVEL_QUANTIZE_COMPRESS_COMMAND) {
 				checkNumParameters(2);
@@ -2383,6 +2392,13 @@ protected void checkMatrixFrameParam(Expression e) { //always unconditional
 			raiseValidateError("Expecting matrix or frame parameter for function "+ getOpCode(), false, LanguageErrorCodes.UNSUPPORTED_PARAMETERS);
 		}
 	}
+
+	protected void checkFrameParam(Expression e) {
+		if(e.getOutput().getDataType() != DataType.FRAME) {
+			raiseValidateError("Expecting frame parameter for function " + getOpCode(), false,
+				LanguageErrorCodes.UNSUPPORTED_PARAMETERS);
+		}
+	}
 
 	protected void checkMatrixScalarParam(Expression e) { //always unconditional
 		if (e.getOutput().getDataType() != DataType.MATRIX && e.getOutput().getDataType() != DataType.SCALAR) {

src/main/java/org/apache/sysds/parser/DMLTranslator.java

@@ -2821,6 +2821,9 @@ else if ( in.length == 2 )
 				DataType.MATRIX, target.getValueType(), AggOp.COUNT_DISTINCT, Direction.Col, expr);
 			break;
 
+		case GET_CATEGORICAL_MASK:
+			currBuiltinOp = new BinaryOp(target.getName(), DataType.MATRIX, ValueType.FP64, OpOp2.GET_CATEGORICAL_MASK,  expr, expr2);
+			break;
 		default:
 			throw new ParseException("Unsupported builtin function type: "+source.getOpCode());
 		}

src/main/java/org/apache/sysds/runtime/functionobjects/Builtin.java

@@ -54,7 +54,7 @@ public enum BuiltinCode { AUTODIFF, SIN, COS, TAN, SINH, COSH, TANH, ASIN, ACOS,
 		MAX, ABS, SIGN, SQRT, EXP, PLOGP, PRINT, PRINTF, NROW, NCOL, LENGTH, LINEAGE, ROUND, MAXINDEX, MININDEX,
 		STOP, CEIL, FLOOR, CUMSUM, ROWCUMSUM, CUMPROD, CUMMIN, CUMMAX, CUMSUMPROD, INVERSE, SPROP, SIGMOID, EVAL, LIST,
 		TYPEOF, APPLY_SCHEMA, DETECTSCHEMA, ISNA, ISNAN, ISINF, DROP_INVALID_TYPE, 
-		DROP_INVALID_LENGTH, VALUE_SWAP, FRAME_ROW_REPLICATE,
+		DROP_INVALID_LENGTH, VALUE_SWAP, FRAME_ROW_REPLICATE, GET_CATEGORICAL_MASK,
 		MAP, COUNT_DISTINCT, COUNT_DISTINCT_APPROX, UNIQUE}
 
 	private static final VectorSpecies<Double> SPECIES = DoubleVector.SPECIES_PREFERRED;
@@ -120,6 +120,7 @@ public enum BuiltinCode { AUTODIFF, SIN, COS, TAN, SINH, COSH, TANH, ASIN, ACOS,
 		String2BuiltinCode.put( "_map", BuiltinCode.MAP);
 		String2BuiltinCode.put( "valueSwap", BuiltinCode.VALUE_SWAP);
 		String2BuiltinCode.put( "applySchema", BuiltinCode.APPLY_SCHEMA);
+		String2BuiltinCode.put( "get_categorical_mask", BuiltinCode.GET_CATEGORICAL_MASK);
 	}
 
 	protected Builtin(BuiltinCode bf) {

src/main/java/org/apache/sysds/runtime/instructions/cp/BinaryCPInstruction.java

@@ -59,6 +59,8 @@ else if (in1.getDataType() == DataType.TENSOR && in2.getDataType() == DataType.T
 			return new BinaryTensorTensorCPInstruction(operator, in1, in2, out, opcode, str);
 		else if (in1.getDataType() == DataType.FRAME && in2.getDataType() == DataType.FRAME)
 			return new BinaryFrameFrameCPInstruction(operator, in1, in2, out, opcode, str);
+		else if (in1.getDataType() == DataType.FRAME && in2.getDataType() == DataType.SCALAR)
+			return new BinaryFrameScalarCPInstruction(operator, in1, in2, out, opcode, str);
 		else if (in1.getDataType() == DataType.FRAME && in2.getDataType() == DataType.MATRIX)
 			return new BinaryFrameMatrixCPInstruction(operator, in1, in2, out, opcode, str);
 		else

src/main/java/org/apache/sysds/runtime/instructions/cp/BinaryFrameScalarCPInstruction.java

@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.sysds.runtime.instructions.cp;
+
+import java.util.Arrays;
+
+import org.apache.sysds.common.Builtins;
+import org.apache.sysds.common.Types.ValueType;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
+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.matrix.operators.MultiThreadedOperator;
+import org.apache.sysds.runtime.transform.TfUtils.TfMethod;
+import org.apache.sysds.runtime.util.UtilFunctions;
+import org.apache.wink.json4j.JSONArray;
+import org.apache.wink.json4j.JSONObject;
+
+public class BinaryFrameScalarCPInstruction extends BinaryCPInstruction {
+	// private static final Log LOG = LogFactory.getLog(BinaryFrameFrameCPInstruction.class.getName());
+
+	protected BinaryFrameScalarCPInstruction(MultiThreadedOperator op, CPOperand in1, CPOperand in2, CPOperand out,
+		String opcode, String istr) {
+		super(CPType.Binary, op, in1, in2, out, opcode, istr);
+	}
+
+	@Override
+	public void processInstruction(ExecutionContext ec) {
+		// get input frames
+		FrameBlock inBlock1 = ec.getFrameInput(input1.getName());
+		ScalarObject spec = ec.getScalarInput(input2.getName(), ValueType.STRING, true);
+		if(getOpcode().equals(Builtins.GET_CATEGORICAL_MASK.toString().toLowerCase())) {
+			processGetCategorical(ec, inBlock1, spec);
+		}
+		else {
+			throw new DMLRuntimeException("Unsupported operation");
+		}
+
+		// Release the memory occupied by input frames
+		ec.releaseFrameInput(input1.getName());
+	}
+
+	public void processGetCategorical(ExecutionContext ec, FrameBlock f, ScalarObject spec) {
+		try {
+
+			// MatrixBlock ret = new MatrixBlock();
+			int nCol = f.getNumColumns();
+
+			JSONObject jSpec = new JSONObject(spec.getStringValue());
+
+			if(!jSpec.containsKey("ids") || !jSpec.getBoolean("ids")) {
+				throw new DMLRuntimeException("not supported non ID based spec for get_categorical_mask");
+			}
+
+			String recode = TfMethod.RECODE.toString();
+			String dummycode = TfMethod.DUMMYCODE.toString();
+			String hash = TfMethod.HASH.toString();
+
+			int[] lengths = new int[nCol];
+			// assume all columns encode to at least one column.
+			Arrays.fill(lengths, 1);
+			boolean[] categorical = new boolean[nCol];
+
+			// feature-hashed columns map to K buckets; a plain hashed column
+			// produces a single (categorical) bucket-id column, while a hashed
+			// column that is additionally dummycoded expands to K columns.
+			boolean[] hashed = new boolean[nCol];
+			int K = 0;
+			if(jSpec.containsKey(hash)) {
+				K = jSpec.getInt("K");
+				JSONArray a = jSpec.getJSONArray(hash);
+				for(Object aa : a) {
+					int av = (Integer) aa - 1;
+					hashed[av] = true;
+					categorical[av] = true;
+				}
+			}
+
+			if(jSpec.containsKey(recode)) {
+				JSONArray a = jSpec.getJSONArray(recode);
+				for(Object aa : a) {
+					int av = (Integer) aa - 1;
+					categorical[av] = true;
+				}
+			}
+
+			if(jSpec.containsKey(dummycode)) {
+				JSONArray a = jSpec.getJSONArray(dummycode);
+				for(Object aa : a) {
+					int av = (Integer) aa - 1;
+					int ndist;
+					if(hashed[av]) {
+						// feature hashing followed by dummycoding yields K columns
+						ndist = K;
+					}
+					else {
+						ColumnMetadata d = f.getColumnMetadata()[av];
+						String v = f.getString(0, av);
+						if(v.length() > 1 && v.charAt(0) == '¿') {
+							ndist = UtilFunctions.parseToInt(v.substring(1));
+						}
+						else {
+							ndist = d.isDefault() ? 0 : (int) d.getNumDistinct();
+						}
+					}
+					lengths[av] = ndist;
+					categorical[av] = true;
+				}
+			}
+
+			// get total size after mapping
+
+			int sumLengths = 0;
+			for(int i : lengths) {
+				sumLengths += i;
+			}
+
+			MatrixBlock ret = new MatrixBlock(1, sumLengths, false);
+			ret.allocateDenseBlock();
+			int off = 0;
+			for(int i = 0; i < lengths.length; i++) {
+				for(int j = 0; j < lengths[i]; j++) {
+					ret.set(0, off++, categorical[i] ? 1 : 0);
+				}
+			}
+
+			ec.setMatrixOutput(output.getName(), ret);
+
+		}
+		catch(Exception e) {
+			throw new DMLRuntimeException(e);
+		}
+	}
+}

src/test/java/org/apache/sysds/test/TestUtils.java

@@ -2941,6 +2941,25 @@ public static void writeTestScalar(String file, double value) {
 		}
 	}
 
+
+	/**
+	 * Write scalar to file
+	 * 
+	 * @param file  File to write to
+	 * @param value Value to write
+	 */
+	public static void writeTestScalar(String file, String value) {
+		try {
+			DataOutputStream out = new DataOutputStream(new FileOutputStream(file));
+			try(PrintWriter pw = new PrintWriter(out)) {
+				pw.println(value);
+			}
+		}
+		catch(IOException e) {
+			fail("unable to write test scalar (" + file + "): " + e.getMessage());
+		}
+	}
+
 	/**
 	 * Write scalar to file
 	 *