[SYSTEMDS-3782] Bag-of-words encoder for Spark backend

Closes #2145.

Author: e-strauss

src/main/java/org/apache/sysds/runtime/instructions/spark/MultiReturnParameterizedBuiltinSPInstruction.java

@@ -62,6 +62,7 @@
 import org.apache.sysds.runtime.meta.DataCharacteristics;
 import org.apache.sysds.runtime.transform.TfUtils;
 import org.apache.sysds.runtime.transform.encode.ColumnEncoder;
+import org.apache.sysds.runtime.transform.encode.ColumnEncoderBagOfWords;
 import org.apache.sysds.runtime.transform.encode.ColumnEncoderBin;
 import org.apache.sysds.runtime.transform.encode.ColumnEncoderComposite;
 import org.apache.sysds.runtime.transform.encode.ColumnEncoderRecode;
@@ -263,6 +264,7 @@ public Iterator<Tuple2<Integer, Object>> call(Iterator<Tuple2<Long, FrameBlock>>
 			// encoder-specific outputs
 			List<ColumnEncoderRecode> raEncoders = _encoder.getColumnEncoders(ColumnEncoderRecode.class);
 			List<ColumnEncoderBin> baEncoders = _encoder.getColumnEncoders(ColumnEncoderBin.class);
+			List<ColumnEncoderBagOfWords> bowEncoders = _encoder.getColumnEncoders(ColumnEncoderBagOfWords.class);
 			ArrayList<Tuple2<Integer, Object>> ret = new ArrayList<>();
 
 			// output recode maps as columnID - token pairs
@@ -273,8 +275,14 @@ public Iterator<Tuple2<Integer, Object>> call(Iterator<Tuple2<Long, FrameBlock>>
 				for(Entry<Integer, HashSet<Object>> e1 : tmp.entrySet())
 					for(Object token : e1.getValue())
 						ret.add(new Tuple2<>(e1.getKey(), token));
-				if(!raEncoders.isEmpty())
-					raEncoders.forEach(columnEncoderRecode -> columnEncoderRecode.getCPRecodeMapsPartial().clear());
+				raEncoders.forEach(columnEncoderRecode -> columnEncoderRecode.getCPRecodeMapsPartial().clear());
+			}
+
+			if(!bowEncoders.isEmpty()){
+				for (ColumnEncoderBagOfWords bowEnc : bowEncoders)
+					for (Object token : bowEnc.getPartialTokenDictionary())
+						ret.add(new Tuple2<>(bowEnc.getColID(), token));
+				bowEncoders.forEach(enc -> enc.getPartialTokenDictionary().clear());
 			}
 
 			// output binning column min/max as columnID - min/max pairs
@@ -321,7 +329,8 @@ public Iterator<String> call(Tuple2<Integer, Iterable<Object>> arg0) throws Exce
 			StringBuilder sb = new StringBuilder();
 
 			// handle recode maps
-			if(_encoder.containsEncoderForID(colID, ColumnEncoderRecode.class)) {
+			if(_encoder.containsEncoderForID(colID, ColumnEncoderRecode.class) ||
+					_encoder.containsEncoderForID(colID, ColumnEncoderBagOfWords.class)) {
 				while(iter.hasNext()) {
 					String token = TfUtils.sanitizeSpaces(iter.next().toString());
 					sb.append(rowID).append(' ').append(scolID).append(' ');

src/main/java/org/apache/sysds/runtime/instructions/spark/ParameterizedBuiltinSPInstruction.java

@@ -88,6 +88,7 @@
 import org.apache.sysds.runtime.transform.TfUtils.TfMethod;
 import org.apache.sysds.runtime.transform.decode.Decoder;
 import org.apache.sysds.runtime.transform.decode.DecoderFactory;
+import org.apache.sysds.runtime.transform.encode.ColumnEncoderBagOfWords;
 import org.apache.sysds.runtime.transform.encode.EncoderFactory;
 import org.apache.sysds.runtime.transform.encode.MultiColumnEncoder;
 import org.apache.sysds.runtime.transform.meta.TfMetaUtils;
@@ -1056,6 +1057,13 @@ public Tuple2<Long, FrameBlock> call(Tuple2<Long, FrameBlock> in) throws Excepti
 
 			// execute block transform apply
 			MultiColumnEncoder encoder = _bencoder.getValue();
+			// we need to create a copy of the encoder since the bag of word encoder stores frameblock specific state
+			// which would be overwritten when multiple blocks are located on a executor
+			// to avoid this, we need to create a shallow copy of the MCEncoder, where we only instantiate new bow
+			// encoders objects with the frameblock specific fields and shallow copy the other fields (like meta)
+			// other encoders are reused and not newly instantiated
+			if(!encoder.getColumnEncoders(ColumnEncoderBagOfWords.class).isEmpty())
+				encoder = new MultiColumnEncoder(encoder); // create copy
 			MatrixBlock tmp = encoder.apply(blk);
 			// remap keys
 			if(_omap != null) {

src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoder.java

@@ -450,7 +450,7 @@ protected void setApplyRowBlocksPerColumn(int nPart) {
 	}
 
 	public enum EncoderType {
-		Recode, FeatureHash, PassThrough, Bin, Dummycode, Omit, MVImpute, Composite, WordEmbedding,
+		Recode, FeatureHash, PassThrough, Bin, Dummycode, Omit, MVImpute, Composite, WordEmbedding, BagOfWords
 	}
 
 	/*