[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,206 @@
+/*
+ * 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.JSONException;
+import org.apache.wink.json4j.JSONObject;
+
+public class BinaryFrameScalarCPInstruction extends BinaryCPInstruction {
+	// private static final Log LOG = LogFactory.getLog(BinaryFrameFrameCPInstruction.class.getName());
+
+	private static final TfMethod[] UNSUPPORTED_MASK_METHODS = new TfMethod[] {TfMethod.BIN,
+		TfMethod.WORD_EMBEDDING, TfMethod.BAG_OF_WORDS, TfMethod.UDF};
+
+	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());
+	}
+
+	private static void validate(JSONObject jSpec) {
+		try {
+			if(!jSpec.containsKey("ids") || !jSpec.getBoolean("ids"))
+				throw new DMLRuntimeException("not supported non ID based spec for get_categorical_mask");
+
+			for(TfMethod m : UNSUPPORTED_MASK_METHODS)
+				if(jSpec.containsKey(m.toString()))
+					throw new DMLRuntimeException("unsupported transform method '" + m + "' for get_categorical_mask");
+		}
+		catch(JSONException e) {
+			throw new DMLRuntimeException(e);
+		}
+	}
+
+	public void processGetCategorical(ExecutionContext ec, FrameBlock f, ScalarObject spec) {
+		try {
+			// 1. extract the spec, 2. validate it
+			JSONObject jSpec = new JSONObject(spec.getStringValue());
+			validate(jSpec);
+
+			// 3.-5. fold each supported transform method into the per-column mask state
+			CategoricalMask mask = new CategoricalMask(f, jSpec);
+			mask.hash();
+			mask.recode();
+			mask.dummycode();
+
+			// 6.-7. size and materialize the output mask
+			ec.setMatrixOutput(output.getName(), mask.toMatrixBlock());
+		}
+		catch(Exception e) {
+			throw new DMLRuntimeException(e);
+		}
+	}
+
+	/**
+	 * Accumulates, per input column, how many output columns it expands to (lengths) and whether those
+	 * output columns are categorical (categorical). The arrays are allocated lazily: a column that no
+	 * method touches keeps the implicit default of a single, non-categorical output column.
+	 */
+	private static final class CategoricalMask {
+		private final FrameBlock f;
+		private final JSONObject jSpec;
+		private final int nCol;
+
+		private int[] lengths = null;
+		private boolean[] categorical = null;
+
+		// 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.
+		private boolean[] hashed = null;
+		private int K = 0;
+
+		private CategoricalMask(FrameBlock f, JSONObject jSpec) {
+			this.f = f;
+			this.jSpec = jSpec;
+			this.nCol = f.getNumColumns();
+		}
+
+		private void hash() throws JSONException {
+			String hash = TfMethod.HASH.toString();
+			if(!jSpec.containsKey(hash))
+				return;
+			K = jSpec.getInt("K");
+			hashed = new boolean[nCol];
+			ensureCategorical();
+			for(Object aa : jSpec.getJSONArray(hash)) {
+				int av = (Integer) aa - 1;
+				hashed[av] = true;
+				categorical[av] = true;
+			}
+		}
+
+		private void recode() throws JSONException {
+			String recode = TfMethod.RECODE.toString();
+			if(!jSpec.containsKey(recode))
+				return;
+			ensureCategorical();
+			for(Object aa : jSpec.getJSONArray(recode)) {
+				int av = (Integer) aa - 1;
+				categorical[av] = true;
+			}
+		}
+
+		private void dummycode() throws JSONException {
+			String dummycode = TfMethod.DUMMYCODE.toString();
+			if(!jSpec.containsKey(dummycode))
+				return;
+			ensureCategorical();
+			ensureLengths();
+			for(Object aa : jSpec.getJSONArray(dummycode)) {
+				int av = (Integer) aa - 1;
+				lengths[av] = distinctCount(av);
+				categorical[av] = true;
+			}
+		}
+
+		private int distinctCount(int av) {
+			if(hashed != null && hashed[av])
+				// feature hashing followed by dummycoding yields K columns
+				return K;
+			ColumnMetadata d = f.getColumnMetadata()[av];
+			String v = f.getString(0, av);
+			if(v.length() > 1 && v.charAt(0) == '¿')
+				return UtilFunctions.parseToInt(v.substring(1));
+			return d.isDefault() ? 0 : (int) d.getNumDistinct();
+		}
+
+		private int sumLengths() {
+			if(lengths == null)
+				return nCol;
+			int sum = 0;
+			for(int i = 0; i < nCol; i++)
+				sum += lengths[i];
+			return sum;
+		}
+
+		private MatrixBlock toMatrixBlock() {
+			MatrixBlock ret = new MatrixBlock(1, sumLengths(), false);
+			ret.allocateDenseBlock();
+			int off = 0;
+			for(int i = 0; i < nCol; i++) {
+				int len = (lengths == null) ? 1 : lengths[i];
+				double val = (categorical != null && categorical[i]) ? 1 : 0;
+				for(int j = 0; j < len; j++)
+					ret.set(0, off++, val);
+			}
+			return ret;
+		}
+
+		private void ensureCategorical() {
+			if(categorical == null)
+				categorical = new boolean[nCol];
+		}
+
+		private void ensureLengths() {
+			if(lengths == null) {
+				lengths = new int[nCol];
+				Arrays.fill(lengths, 1);
+			}
+		}
+	}
+}

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
 	 *