You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemds.apache.org by ar...@apache.org on 2021/09/08 17:59:15 UTC
[systemds] branch master updated: [SYSTEMDS-2972] Transformencode
sparse improvements
This is an automated email from the ASF dual-hosted git repository.
arnabp20 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/systemds.git
The following commit(s) were added to refs/heads/master by this push:
new cdff113 [SYSTEMDS-2972] Transformencode sparse improvements
cdff113 is described below
commit cdff113648371b2dace5905dc0eb81ebf486f094
Author: Lukas Erlbacher <lu...@gmail.com>
AuthorDate: Wed Sep 8 19:58:50 2021 +0200
[SYSTEMDS-2972] Transformencode sparse improvements
This PR introduces the sparse implementations for the transform encoders.
Furthermore, this adds support for row partitioning, statistics, and
debug logging. Now we can enable multithreaded transorm via a flag
in the config file, "parallel.encode".
Closes #1388, closes #1383
---
.../apache/sysds/conf/ConfigurationManager.java | 4 +
src/main/java/org/apache/sysds/conf/DMLConfig.java | 4 +-
.../java/org/apache/sysds/hops/OptimizerUtils.java | 18 +++
.../runtime/compress/CompressedMatrixBlock.java | 5 -
...ltiReturnParameterizedBuiltinCPInstruction.java | 4 +-
.../sysds/runtime/matrix/data/MatrixBlock.java | 35 +----
.../runtime/transform/encode/ColumnEncoder.java | 141 +++++++++++++-----
.../runtime/transform/encode/ColumnEncoderBin.java | 90 ++++++++++--
.../transform/encode/ColumnEncoderComposite.java | 26 ++--
.../transform/encode/ColumnEncoderDummycode.java | 115 +++++++--------
.../transform/encode/ColumnEncoderFeatureHash.java | 64 +++++++--
.../transform/encode/ColumnEncoderPassThrough.java | 90 ++++++++++--
.../transform/encode/ColumnEncoderRecode.java | 81 +++++++++--
.../runtime/transform/encode/EncoderFactory.java | 26 ++--
.../runtime/transform/encode/EncoderMVImpute.java | 36 +++--
.../runtime/transform/encode/EncoderOmit.java | 22 +--
.../transform/encode/MultiColumnEncoder.java | 91 ++++++++----
.../apache/sysds/runtime/util/DependencyTask.java | 24 +++-
.../sysds/runtime/util/DependencyThreadPool.java | 12 +-
.../apache/sysds/runtime/util/UtilFunctions.java | 9 ++
.../java/org/apache/sysds/utils/Statistics.java | 159 +++++++++++++++++++--
.../mt/TransformFrameBuildMultithreadedTest.java | 2 +
.../mt/TransformFrameEncodeMultithreadedTest.java | 4 +-
.../datasets/homes3/homes.tfspec_dummy_all.json | 1 -
.../datasets/homes3/homes.tfspec_dummy_sparse.json | 1 +
25 files changed, 805 insertions(+), 259 deletions(-)
diff --git a/src/main/java/org/apache/sysds/conf/ConfigurationManager.java b/src/main/java/org/apache/sysds/conf/ConfigurationManager.java
index 5a3667c..93654b7 100644
--- a/src/main/java/org/apache/sysds/conf/ConfigurationManager.java
+++ b/src/main/java/org/apache/sysds/conf/ConfigurationManager.java
@@ -171,6 +171,10 @@ public class ConfigurationManager
return getCompilerConfigFlag(ConfigType.PARALLEL_CP_MATRIX_OPERATIONS);
}
+ public static boolean isParallelTransform() {
+ return getDMLConfig().getBooleanValue(DMLConfig.PARALLEL_ENCODE);
+ }
+
public static boolean isParallelParFor() {
return getCompilerConfigFlag(ConfigType.PARALLEL_LOCAL_OR_REMOTE_PARFOR);
}
diff --git a/src/main/java/org/apache/sysds/conf/DMLConfig.java b/src/main/java/org/apache/sysds/conf/DMLConfig.java
index 0b7692b..db59505 100644
--- a/src/main/java/org/apache/sysds/conf/DMLConfig.java
+++ b/src/main/java/org/apache/sysds/conf/DMLConfig.java
@@ -67,6 +67,7 @@ public class DMLConfig
public static final String DEFAULT_BLOCK_SIZE = "sysds.defaultblocksize";
public static final String CP_PARALLEL_OPS = "sysds.cp.parallel.ops";
public static final String CP_PARALLEL_IO = "sysds.cp.parallel.io";
+ public static final String PARALLEL_ENCODE = "sysds.parallel.encode"; // boolean: enable multi-threaded transformencode and apply
public static final String COMPRESSED_LINALG = "sysds.compressed.linalg";
public static final String COMPRESSED_LOSSY = "sysds.compressed.lossy";
public static final String COMPRESSED_VALID_COMPRESSIONS = "sysds.compressed.valid.compressions";
@@ -125,6 +126,7 @@ public class DMLConfig
_defaultVals.put(DEFAULT_BLOCK_SIZE, String.valueOf(OptimizerUtils.DEFAULT_BLOCKSIZE) );
_defaultVals.put(CP_PARALLEL_OPS, "true" );
_defaultVals.put(CP_PARALLEL_IO, "true" );
+ _defaultVals.put(PARALLEL_ENCODE, "false" );
_defaultVals.put(COMPRESSED_LINALG, Compression.CompressConfig.FALSE.name() );
_defaultVals.put(COMPRESSED_LOSSY, "false" );
_defaultVals.put(COMPRESSED_VALID_COMPRESSIONS, "SDC,DDC");
@@ -398,7 +400,7 @@ public class DMLConfig
public String getConfigInfo() {
String[] tmpConfig = new String[] {
LOCAL_TMP_DIR,SCRATCH_SPACE,OPTIMIZATION_LEVEL, DEFAULT_BLOCK_SIZE,
- CP_PARALLEL_OPS, CP_PARALLEL_IO, NATIVE_BLAS, NATIVE_BLAS_DIR,
+ CP_PARALLEL_OPS, CP_PARALLEL_IO, PARALLEL_ENCODE, NATIVE_BLAS, NATIVE_BLAS_DIR,
COMPRESSED_LINALG, COMPRESSED_LOSSY, COMPRESSED_VALID_COMPRESSIONS, COMPRESSED_OVERLAPPING,
COMPRESSED_SAMPLING_RATIO, COMPRESSED_COCODE, COMPRESSED_TRANSPOSE,
CODEGEN, CODEGEN_API, CODEGEN_COMPILER, CODEGEN_OPTIMIZER, CODEGEN_PLANCACHE, CODEGEN_LITERALS,
diff --git a/src/main/java/org/apache/sysds/hops/OptimizerUtils.java b/src/main/java/org/apache/sysds/hops/OptimizerUtils.java
index 01769c7..d63b2cb 100644
--- a/src/main/java/org/apache/sysds/hops/OptimizerUtils.java
+++ b/src/main/java/org/apache/sysds/hops/OptimizerUtils.java
@@ -1008,6 +1008,24 @@ public class OptimizerUtils
return ret;
}
+
+ public static int getTransformNumThreads(int maxNumThreads)
+ {
+ //by default max local parallelism (vcores)
+ int ret = InfrastructureAnalyzer.getLocalParallelism();
+
+ //apply external max constraint (e.g., set by parfor or other rewrites)
+ if( maxNumThreads > 0 ) {
+ ret = Math.min(ret, maxNumThreads);
+ }
+
+ //check if enabled in config.xml
+ if( !ConfigurationManager.isParallelTransform() ) {
+ ret = 1;
+ }
+
+ return ret;
+ }
public static Level getDefaultLogLevel() {
Level log = Logger.getRootLogger().getLevel();
diff --git a/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java b/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
index 2c205f3..d374d3c 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
@@ -1345,11 +1345,6 @@ public class CompressedMatrixBlock extends MatrixBlock {
}
@Override
- public void quickSetValueThreadSafe(int r, int c, double v) {
- throw new DMLCompressionException("Thread safe execution does not work on Compressed Matrix");
- }
-
- @Override
public double quickGetValueThreadSafe(int r, int c) {
throw new DMLCompressionException("Thread safe execution does not work on Compressed Matrix");
}
diff --git a/src/main/java/org/apache/sysds/runtime/instructions/cp/MultiReturnParameterizedBuiltinCPInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/cp/MultiReturnParameterizedBuiltinCPInstruction.java
index b6e0d97..800aa51 100644
--- a/src/main/java/org/apache/sysds/runtime/instructions/cp/MultiReturnParameterizedBuiltinCPInstruction.java
+++ b/src/main/java/org/apache/sysds/runtime/instructions/cp/MultiReturnParameterizedBuiltinCPInstruction.java
@@ -25,6 +25,7 @@ import java.util.List;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.sysds.common.Types.DataType;
import org.apache.sysds.common.Types.ValueType;
+import org.apache.sysds.hops.OptimizerUtils;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysds.runtime.instructions.InstructionUtils;
@@ -85,7 +86,8 @@ public class MultiReturnParameterizedBuiltinCPInstruction extends ComputationCPI
// execute block transform encode
MultiColumnEncoder encoder = EncoderFactory.createEncoder(spec, colnames, fin.getNumColumns(), null);
- MatrixBlock data = encoder.encode(fin); // build and apply
+ // TODO: Assign #threads in compiler and pass via the instruction string
+ MatrixBlock data = encoder.encode(fin, OptimizerUtils.getTransformNumThreads(-1)); // build and apply
FrameBlock meta = encoder.getMetaData(new FrameBlock(fin.getNumColumns(), ValueType.STRING));
meta.setColumnNames(colnames);
diff --git a/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java b/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
index 160f8e9..a86a878 100644
--- a/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/matrix/data/MatrixBlock.java
@@ -117,7 +117,7 @@ import org.apache.sysds.utils.NativeHelper;
public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizable {
// private static final Log LOG = LogFactory.getLog(MatrixBlock.class.getName());
-
+
private static final long serialVersionUID = 7319972089143154056L;
//sparsity nnz threshold, based on practical experiments on space consumption and performance
@@ -654,27 +654,6 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
}
}
- /**
- * Thread save set.
- * Blocks need to be allocated, and in case of MCSR sparse, all rows
- * that are going to be accessed need to be allocated as well.
- *
- * @param r row
- * @param c column
- * @param v value
- */
- public void quickSetValueThreadSafe(int r, int c, double v) {
- if(sparse) {
- if(!(sparseBlock instanceof SparseBlockMCSR))
- throw new RuntimeException("Only MCSR Blocks are supported for Multithreaded sparse set.");
- synchronized (sparseBlock.get(r)) {
- sparseBlock.set(r,c,v);
- }
- }
- else
- denseBlock.set(r,c,v);
- }
-
public double quickGetValueThreadSafe(int r, int c) {
if(sparse) {
if(!(sparseBlock instanceof SparseBlockMCSR))
@@ -976,7 +955,7 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
/**
* Wrapper method for single threaded reduceall-colSum of a matrix.
- *
+ *
* @return A new MatrixBlock containing the column sums of this matrix.
*/
public MatrixBlock colSum() {
@@ -986,7 +965,7 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
/**
* Wrapper method for single threaded reduceall-rowSum of a matrix.
- *
+ *
* @return A new MatrixBlock containing the row sums of this matrix.
*/
public MatrixBlock rowSum(){
@@ -1422,7 +1401,7 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
throw new RuntimeException( "Copy must not overwrite itself!" );
if(that instanceof CompressedMatrixBlock)
that = CompressedMatrixBlock.getUncompressed(that, "Copy not effecient into a MatrixBlock");
-
+
rlen=that.rlen;
clen=that.clen;
sparse=sp;
@@ -2935,7 +2914,7 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
LibMatrixBincell.isValidDimensionsBinary(this, that);
if(thatValue instanceof CompressedMatrixBlock)
return ((CompressedMatrixBlock) thatValue).binaryOperationsLeft(op, this, result);
-
+
//compute output dimensions
boolean outer = (LibMatrixBincell.getBinaryAccessType(this, that)
== BinaryAccessType.OUTER_VECTOR_VECTOR);
@@ -2980,7 +2959,7 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
m2 = ((CompressedMatrixBlock) m2).getUncompressed("Ternay Operator arg2 " + op.fn.getClass().getSimpleName());
if(m3 instanceof CompressedMatrixBlock)
m3 = ((CompressedMatrixBlock) m3).getUncompressed("Ternay Operator arg3 " + op.fn.getClass().getSimpleName());
-
+
//prepare inputs
final boolean s1 = (rlen==1 && clen==1);
final boolean s2 = (m2.rlen==1 && m2.clen==1);
@@ -3674,7 +3653,7 @@ public class MatrixBlock extends MatrixValue implements CacheBlock, Externalizab
"Invalid nCol dimension for append rbind: was " + in[i].clen + " should be: " + clen);
}
}
-
+
public static MatrixBlock naryOperations(Operator op, MatrixBlock[] matrices, ScalarObject[] scalars, MatrixBlock ret) {
//note: currently only min max, plus supported and hence specialized implementation
//prepare operator
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoder.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoder.java
index 33bf452..e04eeff 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoder.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoder.java
@@ -20,6 +20,7 @@
package org.apache.sysds.runtime.transform.encode;
import static org.apache.sysds.runtime.transform.encode.EncoderFactory.getEncoderType;
+import static org.apache.sysds.runtime.util.UtilFunctions.getBlockSizes;
import java.io.Externalizable;
import java.io.IOException;
@@ -45,6 +46,8 @@ import org.apache.sysds.runtime.util.DependencyThreadPool;
*/
public abstract class ColumnEncoder implements Externalizable, Encoder, Comparable<ColumnEncoder> {
protected static final Log LOG = LogFactory.getLog(ColumnEncoder.class.getName());
+ protected static final int APPLY_ROW_BLOCKS_PER_COLUMN = 1;
+ public static int BUILD_ROW_BLOCKS_PER_COLUMN = 1;
private static final long serialVersionUID = 2299156350718979064L;
protected int _colID;
@@ -52,7 +55,23 @@ public abstract class ColumnEncoder implements Externalizable, Encoder, Comparab
_colID = colID;
}
- public abstract MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol);
+ /**
+ * Apply Functions are only used in Single Threaded or Multi-Threaded Dense context.
+ * That's why there is no regard for MT sparse!
+ *
+ * @param in Input Block
+ * @param out Output Matrix
+ * @param outputCol The output column for the given column
+ * @return same as out
+ *
+ */
+ public MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol){
+ return apply(in, out, outputCol, 0, -1);
+ }
+
+ public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol){
+ return apply(in, out, outputCol, 0, -1);
+ }
public abstract MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol, int rowStart, int blk);
@@ -172,18 +191,18 @@ public abstract class ColumnEncoder implements Externalizable, Encoder, Comparab
* complete if all previous tasks are done. This is so that we can use the last task as a dependency for the whole
* build, reducing unnecessary dependencies.
*/
- public List<DependencyTask<?>> getBuildTasks(FrameBlock in, int blockSize) {
+ public List<DependencyTask<?>> getBuildTasks(FrameBlock in) {
List<Callable<Object>> tasks = new ArrayList<>();
List<List<? extends Callable<?>>> dep = null;
- if(blockSize <= 0 || blockSize >= in.getNumRows()) {
+ int nRows = in.getNumRows();
+ int[] blockSizes = getBlockSizes(nRows, getNumBuildRowPartitions());
+ if(blockSizes.length == 1) {
tasks.add(getBuildTask(in));
}
else {
HashMap<Integer, Object> ret = new HashMap<>();
- for(int i = 0; i < in.getNumRows(); i = i + blockSize)
- tasks.add(getPartialBuildTask(in, i, blockSize, ret));
- if(in.getNumRows() % blockSize != 0)
- tasks.add(getPartialBuildTask(in, in.getNumRows() - in.getNumRows() % blockSize, -1, ret));
+ for(int startRow = 0, i = 0; i < blockSizes.length; startRow+=blockSizes[i], i++)
+ tasks.add(getPartialBuildTask(in, startRow, blockSizes[i], ret));
tasks.add(getPartialMergeBuildTask(ret));
dep = new ArrayList<>(Collections.nCopies(tasks.size() - 1, null));
dep.add(tasks.subList(0, tasks.size() - 1));
@@ -198,24 +217,63 @@ public abstract class ColumnEncoder implements Externalizable, Encoder, Comparab
public Callable<Object> getPartialBuildTask(FrameBlock in, int startRow, int blockSize,
HashMap<Integer, Object> ret) {
throw new DMLRuntimeException(
- "Trying to get the PartialBuild task of an Encoder which does not support " + "partial building");
+ "Trying to get the PartialBuild task of an Encoder which does not support partial building");
}
public Callable<Object> getPartialMergeBuildTask(HashMap<Integer, ?> ret) {
throw new DMLRuntimeException(
- "Trying to get the BuildMergeTask task of an Encoder which does not support " + "partial building");
+ "Trying to get the BuildMergeTask task of an Encoder which does not support partial building");
}
public List<DependencyTask<?>> getApplyTasks(FrameBlock in, MatrixBlock out, int outputCol) {
- List<Callable<Object>> tasks = new ArrayList<>();
- tasks.add(new ColumnApplyTask(this, in, out, outputCol));
- return DependencyThreadPool.createDependencyTasks(tasks, null);
+ return getApplyTasks(in, null, out, outputCol);
}
public List<DependencyTask<?>> getApplyTasks(MatrixBlock in, MatrixBlock out, int outputCol) {
+ return getApplyTasks(null, in, out, outputCol);
+ }
+
+ private List<DependencyTask<?>> getApplyTasks(FrameBlock inF, MatrixBlock inM, MatrixBlock out, int outputCol){
List<Callable<Object>> tasks = new ArrayList<>();
- tasks.add(new ColumnApplyTask(this, in, out, outputCol));
- return DependencyThreadPool.createDependencyTasks(tasks, null);
+ List<List<? extends Callable<?>>> dep = null;
+ if ((inF != null && inM != null) || (inF == null && inM == null))
+ throw new DMLRuntimeException("getApplyTasks needs to be called with either FrameBlock input " +
+ "or MatrixBlock input");
+ int nRows = inF == null ? inM.getNumRows() : inF.getNumRows();
+ int[] blockSizes = getBlockSizes(nRows, getNumApplyRowPartitions());
+ for(int startRow = 0, i = 0; i < blockSizes.length; startRow+=blockSizes[i], i++){
+ if(inF != null)
+ if(out.isInSparseFormat())
+ tasks.add(getSparseTask(inF, out, outputCol, startRow, blockSizes[i]));
+ else
+ tasks.add(new ColumnApplyTask<>(this, inF, out, outputCol, startRow, blockSizes[i]));
+ else
+ if(out.isInSparseFormat())
+ tasks.add(getSparseTask(inM, out, outputCol, startRow, blockSizes[i]));
+ else
+ tasks.add(new ColumnApplyTask<>(this, inM, out, outputCol, startRow, blockSizes[i]));
+ }
+ if(tasks.size() > 1){
+ dep = new ArrayList<>(Collections.nCopies(tasks.size(), null));
+ tasks.add(() -> null); // Empty task as barrier
+ dep.add(tasks.subList(0, tasks.size()-1));
+ }
+
+ return DependencyThreadPool.createDependencyTasks(tasks, dep);
+ }
+
+ protected abstract ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(FrameBlock in, MatrixBlock out, int outputCol, int startRow, int blk);
+
+ protected abstract ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(MatrixBlock in, MatrixBlock out, int outputCol, int startRow, int blk);
+
+ protected int getNumApplyRowPartitions(){
+ return APPLY_ROW_BLOCKS_PER_COLUMN;
+ }
+
+ protected int getNumBuildRowPartitions(){
+ return BUILD_ROW_BLOCKS_PER_COLUMN;
}
public enum EncoderType {
@@ -226,39 +284,54 @@ public abstract class ColumnEncoder implements Externalizable, Encoder, Comparab
* This is the base Task for each column apply. If no custom "getApplyTasks" is implemented in an Encoder this task
* will be used.
*/
- private static class ColumnApplyTask implements Callable<Object> {
+ protected static class ColumnApplyTask<T extends ColumnEncoder> implements Callable<Object> {
+
+ protected final T _encoder;
+ protected final FrameBlock _inputF;
+ protected final MatrixBlock _inputM;
+ protected final MatrixBlock _out;
+ protected final int _outputCol;
+ protected final int _startRow;
+ protected final int _blk;
+
+ protected ColumnApplyTask(T encoder, FrameBlock input, MatrixBlock out, int outputCol){
+ this(encoder, input, out, outputCol, 0, -1);
+ }
- private final ColumnEncoder _encoder;
- private final FrameBlock _inputF;
- private final MatrixBlock _inputM;
- private final MatrixBlock _out;
- private final int _outputCol;
+ protected ColumnApplyTask(T encoder, MatrixBlock input, MatrixBlock out, int outputCol){
+ this(encoder, input, out, outputCol, 0, -1);
+ }
- protected ColumnApplyTask(ColumnEncoder encoder, FrameBlock input, MatrixBlock out, int outputCol) {
- _encoder = encoder;
- _inputF = input;
- _inputM = null;
- _out = out;
- _outputCol = outputCol;
+ protected ColumnApplyTask(T encoder, FrameBlock input, MatrixBlock out, int outputCol, int startRow, int blk) {
+ this(encoder, input, null, out, outputCol, startRow, blk);
}
- protected ColumnApplyTask(ColumnEncoder encoder, MatrixBlock input, MatrixBlock out, int outputCol) {
+ protected ColumnApplyTask(T encoder, MatrixBlock input, MatrixBlock out, int outputCol, int startRow, int blk) {
+ this(encoder, null, input, out, outputCol, startRow, blk);
+ }
+ private ColumnApplyTask(T encoder, FrameBlock inputF, MatrixBlock inputM, MatrixBlock out, int outputCol,
+ int startRow, int blk){
_encoder = encoder;
- _inputM = input;
- _inputF = null;
+ _inputM = inputM;
+ _inputF = inputF;
_out = out;
_outputCol = outputCol;
+ _startRow = startRow;
+ _blk = blk;
}
@Override
- public Void call() throws Exception {
+ public Object call() throws Exception {
assert _outputCol >= 0;
- int _rowStart = 0;
- int _blk = -1;
+ if(_out.isInSparseFormat()){
+ // this is an issue since most sparse Tasks modify the sparse structure so normal get and set calls are
+ // not possible.
+ throw new DMLRuntimeException("ColumnApplyTask called although output is in sparse format.");
+ }
if(_inputF == null)
- _encoder.apply(_inputM, _out, _outputCol, _rowStart, _blk);
+ _encoder.apply(_inputM, _out, _outputCol, _startRow, _blk);
else
- _encoder.apply(_inputF, _out, _outputCol, _rowStart, _blk);
+ _encoder.apply(_inputF, _out, _outputCol, _startRow, _blk);
return null;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderBin.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderBin.java
index b4d4800..5736c1e 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderBin.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderBin.java
@@ -28,11 +28,15 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.concurrent.Callable;
+import org.apache.commons.lang3.NotImplementedException;
import org.apache.commons.lang3.tuple.MutableTriple;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.lops.Lop;
+import org.apache.sysds.runtime.data.SparseRowVector;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.UtilFunctions;
+import org.apache.sysds.utils.Statistics;
public class ColumnEncoderBin extends ColumnEncoder {
public static final String MIN_PREFIX = "min";
@@ -84,10 +88,13 @@ public class ColumnEncoderBin extends ColumnEncoder {
@Override
public void build(FrameBlock in) {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
if(!isApplicable())
return;
double[] pairMinMax = getMinMaxOfCol(in, _colID, 0, -1);
computeBins(pairMinMax[0], pairMinMax[1]);
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformBinningBuildTime(System.nanoTime()-t0);
}
private static double[] getMinMaxOfCol(FrameBlock in, int colID, int startRow, int blockSize) {
@@ -118,6 +125,8 @@ public class ColumnEncoderBin extends ColumnEncoder {
return new BinMergePartialBuildTask(this, ret);
}
+
+
public void computeBins(double min, double max) {
// ensure allocated internal transformation metadata
if(_binMins == null || _binMaxs == null) {
@@ -146,39 +155,47 @@ public class ColumnEncoderBin extends ColumnEncoder {
}
@Override
- public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol) {
- return apply(in, out, outputCol, 0, -1);
- }
-
- @Override
- public MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol) {
- return apply(in, out, outputCol, 0, -1);
- }
-
- @Override
public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol, int rowStart, int blk) {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
for(int i = rowStart; i < getEndIndex(in.getNumRows(), rowStart, blk); i++) {
double inVal = UtilFunctions.objectToDouble(in.getSchema()[_colID - 1], in.get(i, _colID - 1));
int ix = Arrays.binarySearch(_binMaxs, inVal);
int binID = ((ix < 0) ? Math.abs(ix + 1) : ix) + 1;
- out.quickSetValueThreadSafe(i, outputCol, binID);
+ out.quickSetValue(i, outputCol, binID);
}
+ if (DMLScript.STATISTICS)
+ Statistics.incTransformBinningApplyTime(System.nanoTime()-t0);
return out;
}
@Override
public MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol, int rowStart, int blk) {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
int end = getEndIndex(in.getNumRows(), rowStart, blk);
for(int i = rowStart; i < end; i++) {
double inVal = in.quickGetValueThreadSafe(i, _colID - 1);
int ix = Arrays.binarySearch(_binMaxs, inVal);
int binID = ((ix < 0) ? Math.abs(ix + 1) : ix) + 1;
- out.quickSetValueThreadSafe(i, outputCol, binID);
+ out.quickSetValue(i, outputCol, binID);
}
+ if (DMLScript.STATISTICS)
+ Statistics.incTransformBinningApplyTime(System.nanoTime()-t0);
return out;
}
@Override
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(FrameBlock in, MatrixBlock out, int outputCol, int startRow, int blk) {
+ return new BinSparseApplyTask(this, in, out, outputCol);
+ }
+
+ @Override
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(MatrixBlock in, MatrixBlock out, int outputCol, int startRow, int blk) {
+ throw new NotImplementedException("Sparse Binning for MatrixBlocks not jet implemented");
+ }
+
+ @Override
public void mergeAt(ColumnEncoder other) {
if(other instanceof ColumnEncoderBin) {
ColumnEncoderBin otherBin = (ColumnEncoderBin) other;
@@ -264,6 +281,43 @@ public class ColumnEncoderBin extends ColumnEncoder {
}
}
+ private static class BinSparseApplyTask extends ColumnApplyTask<ColumnEncoderBin> {
+
+ public BinSparseApplyTask(ColumnEncoderBin encoder, FrameBlock input,
+ MatrixBlock out, int outputCol, int startRow, int blk) {
+ super(encoder, input, out, outputCol, startRow, blk);
+ }
+
+ private BinSparseApplyTask(ColumnEncoderBin encoder, FrameBlock input, MatrixBlock out, int outputCol) {
+ super(encoder, input, out, outputCol);
+ }
+
+ public Object call() throws Exception {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+ int index = _encoder._colID - 1;
+ if(_out.getSparseBlock() == null)
+ return null;
+ assert _inputF != null;
+ for(int r = _startRow; r < getEndIndex(_inputF.getNumRows(), _startRow, _blk); r++) {
+ SparseRowVector row = (SparseRowVector) _out.getSparseBlock().get(r);
+ double inVal = UtilFunctions.objectToDouble(_inputF.getSchema()[index], _inputF.get(r, index));
+ int ix = Arrays.binarySearch(_encoder._binMaxs, inVal);
+ int binID = ((ix < 0) ? Math.abs(ix + 1) : ix) + 1;
+ row.values()[index] = binID;
+ row.indexes()[index] = _outputCol;
+ }
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformBinningApplyTime(System.nanoTime()-t0);
+ return null;
+ }
+
+ @Override
+ public String toString() {
+ return getClass().getSimpleName() + "<ColId: " + _encoder._colID + ">";
+ }
+
+ }
+
private static class BinPartialBuildTask implements Callable<Object> {
private final FrameBlock _input;
@@ -284,7 +338,13 @@ public class ColumnEncoderBin extends ColumnEncoder {
@Override
public double[] call() throws Exception {
- _partialMinMax.put(_startRow, getMinMaxOfCol(_input, _colID, _startRow, _blockSize));
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+ double[] minMax = getMinMaxOfCol(_input, _colID, _startRow, _blockSize);
+ synchronized (_partialMinMax){
+ _partialMinMax.put(_startRow, minMax);
+ }
+ if (DMLScript.STATISTICS)
+ Statistics.incTransformBinningBuildTime(System.nanoTime()-t0);
return null;
}
@@ -306,6 +366,7 @@ public class ColumnEncoderBin extends ColumnEncoder {
@Override
public Object call() throws Exception {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
double min = Double.POSITIVE_INFINITY;
double max = Double.NEGATIVE_INFINITY;
for(Object minMax : _partialMaps.values()) {
@@ -313,6 +374,9 @@ public class ColumnEncoderBin extends ColumnEncoder {
max = Math.max(max, ((double[]) minMax)[1]);
}
_encoder.computeBins(min, max);
+
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformBinningBuildTime(System.nanoTime()-t0);
return null;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderComposite.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderComposite.java
index 54b8795..f7611c3 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderComposite.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderComposite.java
@@ -139,11 +139,23 @@ public class ColumnEncoderComposite extends ColumnEncoder {
}
@Override
- public List<DependencyTask<?>> getBuildTasks(FrameBlock in, int blockSize) {
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(MatrixBlock in, MatrixBlock out, int outputCol, int startRow, int blk) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(FrameBlock in, MatrixBlock out, int outputCol, int startRow, int blk) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public List<DependencyTask<?>> getBuildTasks(FrameBlock in) {
List<DependencyTask<?>> tasks = new ArrayList<>();
Map<Integer[], Integer[]> depMap = null;
for(ColumnEncoder columnEncoder : _columnEncoders) {
- List<DependencyTask<?>> t = columnEncoder.getBuildTasks(in, blockSize);
+ List<DependencyTask<?>> t = columnEncoder.getBuildTasks(in);
if(t == null)
continue;
// Linear execution between encoders so they can't be built in parallel
@@ -179,16 +191,6 @@ public class ColumnEncoderComposite extends ColumnEncoder {
}
@Override
- public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol) {
- return apply(in, out, outputCol, 0, -1);
- }
-
- @Override
- public MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol) {
- return apply(in, out, outputCol, 0, -1);
- }
-
- @Override
public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol, int rowStart, int blk) {
try {
for(int i = 0; i < _columnEncoders.size(); i++) {
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderDummycode.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderDummycode.java
index 25b2eb9..1047f54 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderDummycode.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderDummycode.java
@@ -24,17 +24,15 @@ import static org.apache.sysds.runtime.util.UtilFunctions.getEndIndex;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
-import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
-import java.util.concurrent.Callable;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.runtime.DMLRuntimeException;
-import org.apache.sysds.runtime.data.SparseRowVector;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.DependencyTask;
-import org.apache.sysds.runtime.util.DependencyThreadPool;
+import org.apache.sysds.utils.Statistics;
public class ColumnEncoderDummycode extends ColumnEncoder {
private static final long serialVersionUID = 5832130477659116489L;
@@ -60,26 +58,18 @@ public class ColumnEncoderDummycode extends ColumnEncoder {
}
@Override
- public List<DependencyTask<?>> getBuildTasks(FrameBlock in, int blockSize) {
+ public List<DependencyTask<?>> getBuildTasks(FrameBlock in) {
return null;
}
@Override
- public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol) {
- return apply(in, out, outputCol, 0, -1);
- }
-
- public MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol) {
- return apply(in, out, outputCol, 0, -1);
- }
-
- @Override
public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol, int rowStart, int blk) {
throw new DMLRuntimeException("Called DummyCoder with FrameBlock");
}
@Override
public MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol, int rowStart, int blk) {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
// Out Matrix should already be correct size!
// append dummy coded or unchanged values to output
for(int i = rowStart; i < getEndIndex(in.getNumRows(), rowStart, blk); i++) {
@@ -89,20 +79,25 @@ public class ColumnEncoderDummycode extends ColumnEncoder {
int nCol = outputCol + (int) val - 1;
// Setting value to 0 first in case of sparse so the row vector does not need to be resized
if(nCol != outputCol)
- out.quickSetValueThreadSafe(i, outputCol, 0);
- out.quickSetValueThreadSafe(i, nCol, 1);
+ out.quickSetValue(i, outputCol, 0);
+ out.quickSetValue(i, nCol, 1);
}
+ if (DMLScript.STATISTICS)
+ Statistics.incTransformDummyCodeApplyTime(System.nanoTime()-t0);
return out;
}
+
+ @Override
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(MatrixBlock in, MatrixBlock out, int outputCol, int startRow, int blk) {
+ return new DummycodeSparseApplyTask(this, in, out, outputCol, startRow, blk);
+ }
+
@Override
- public List<DependencyTask<?>> getApplyTasks(MatrixBlock in, MatrixBlock out, int outputCol) {
- List<Callable<Object>> tasks = new ArrayList<>();
- if(out.isInSparseFormat())
- tasks.add(new DummycodeSparseApplyTask(this, in, out, outputCol));
- else
- return super.getApplyTasks(in, out, outputCol);
- return DependencyThreadPool.createDependencyTasks(tasks, null);
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(FrameBlock in, MatrixBlock out, int outputCol, int startRow, int blk) {
+ throw new DMLRuntimeException("Called DummyCoder with FrameBlock");
}
@Override
@@ -139,6 +134,7 @@ public class ColumnEncoderDummycode extends ColumnEncoder {
if(distinct != -1) {
_domainSize = distinct;
+ LOG.debug("DummyCoder for column: " + _colID + " has domain size: " + _domainSize);
}
}
}
@@ -188,48 +184,47 @@ public class ColumnEncoderDummycode extends ColumnEncoder {
return _domainSize;
}
- private static class DummycodeSparseApplyTask implements Callable<Object> {
- private final ColumnEncoderDummycode _encoder;
- private final MatrixBlock _input;
- private final MatrixBlock _out;
- private final int _outputCol;
+ private static class DummycodeSparseApplyTask extends ColumnApplyTask<ColumnEncoderDummycode> {
+
+ protected DummycodeSparseApplyTask(ColumnEncoderDummycode encoder, MatrixBlock input,
+ MatrixBlock out, int outputCol) {
+ super(encoder, input, out, outputCol);
+ }
- private DummycodeSparseApplyTask(ColumnEncoderDummycode encoder, MatrixBlock input, MatrixBlock out,
- int outputCol) {
- _encoder = encoder;
- _input = input;
- _out = out;
- _outputCol = outputCol;
+ protected DummycodeSparseApplyTask(ColumnEncoderDummycode encoder, MatrixBlock input,
+ MatrixBlock out, int outputCol, int startRow, int blk) {
+ super(encoder, input, out, outputCol, startRow, blk);
}
public Object call() throws Exception {
- for(int r = 0; r < _input.getNumRows(); r++) {
- if(_out.getSparseBlock() == null)
- return null;
- synchronized(_out.getSparseBlock().get(r)) {
- // Since the recoded values are already offset in the output matrix (same as input at this point)
- // the dummycoding only needs to offset them within their column domain. Which means that the
- // indexes in the SparseRowVector do not need to be sorted anymore and can be updated directly.
- //
- // Input: Output:
- //
- // 1 | 0 | 2 | 0 1 | 0 | 0 | 1
- // 2 | 0 | 1 | 0 ===> 0 | 1 | 1 | 0
- // 1 | 0 | 2 | 0 1 | 0 | 0 | 1
- // 1 | 0 | 1 | 0 1 | 0 | 1 | 0
- //
- // Example SparseRowVector Internals (1. row):
- //
- // indexes = [0,2] ===> indexes = [0,3]
- // values = [1,2] values = [1,1]
- int index = ((SparseRowVector) _out.getSparseBlock().get(r)).getIndex(_outputCol);
- double val = _out.getSparseBlock().get(r).values()[index];
- int nCol = _outputCol + (int) val - 1;
-
- _out.getSparseBlock().get(r).indexes()[index] = nCol;
- _out.getSparseBlock().get(r).values()[index] = 1;
- }
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+ assert _inputM != null;
+ if(_out.getSparseBlock() == null)
+ return null;
+ for(int r = _startRow; r < getEndIndex(_inputM.getNumRows(), _startRow, _blk); r++) {
+ // Since the recoded values are already offset in the output matrix (same as input at this point)
+ // the dummycoding only needs to offset them within their column domain. Which means that the
+ // indexes in the SparseRowVector do not need to be sorted anymore and can be updated directly.
+ //
+ // Input: Output:
+ //
+ // 1 | 0 | 2 | 0 1 | 0 | 0 | 1
+ // 2 | 0 | 1 | 0 ===> 0 | 1 | 1 | 0
+ // 1 | 0 | 2 | 0 1 | 0 | 0 | 1
+ // 1 | 0 | 1 | 0 1 | 0 | 1 | 0
+ //
+ // Example SparseRowVector Internals (1. row):
+ //
+ // indexes = [0,2] ===> indexes = [0,3]
+ // values = [1,2] values = [1,1]
+ int index = _encoder._colID - 1;
+ double val = _out.getSparseBlock().get(r).values()[index];
+ int nCol = _outputCol + (int) val - 1;
+ _out.getSparseBlock().get(r).indexes()[index] = nCol;
+ _out.getSparseBlock().get(r).values()[index] = 1;
}
+ if (DMLScript.STATISTICS)
+ Statistics.incTransformDummyCodeApplyTime(System.nanoTime()-t0);
return null;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderFeatureHash.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderFeatureHash.java
index 1c74ae5..d30d8dc 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderFeatureHash.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderFeatureHash.java
@@ -26,11 +26,15 @@ import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.util.List;
+import org.apache.commons.lang3.NotImplementedException;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.data.SparseRowVector;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.DependencyTask;
import org.apache.sysds.runtime.util.UtilFunctions;
+import org.apache.sysds.utils.Statistics;
/**
* Class used for feature hashing transformation of frames.
@@ -56,7 +60,7 @@ public class ColumnEncoderFeatureHash extends ColumnEncoder {
}
private long getCode(String key) {
- return key.hashCode() % _K;
+ return (key.hashCode() % _K) + 1;
}
@Override
@@ -65,22 +69,25 @@ public class ColumnEncoderFeatureHash extends ColumnEncoder {
}
@Override
- public List<DependencyTask<?>> getBuildTasks(FrameBlock in, int blockSize) {
+ public List<DependencyTask<?>> getBuildTasks(FrameBlock in) {
return null;
}
@Override
- public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol) {
- return apply(in, out, outputCol, 0, -1);
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(FrameBlock in, MatrixBlock out, int outputCol, int startRow, int blk) {
+ return new FeatureHashSparseApplyTask(this, in, out, outputCol, startRow, blk);
}
@Override
- public MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol) {
- return apply(in, out, outputCol, 0, -1);
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(MatrixBlock in, MatrixBlock out, int outputCol, int startRow, int blk) {
+ throw new NotImplementedException("Sparse FeatureHashing for MatrixBlocks not jet implemented");
}
@Override
public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol, int rowStart, int blk) {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
// apply feature hashing column wise
for(int i = rowStart; i < getEndIndex(in.getNumRows(), rowStart, blk); i++) {
Object okey = in.get(i, _colID - 1);
@@ -88,21 +95,26 @@ public class ColumnEncoderFeatureHash extends ColumnEncoder {
if(key == null)
throw new DMLRuntimeException("Missing Value encountered in input Frame for FeatureHash");
long code = getCode(key);
- out.quickSetValueThreadSafe(i, outputCol, (code >= 0) ? code : Double.NaN);
+ out.quickSetValue(i, outputCol, (code >= 0) ? code : Double.NaN);
}
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformFeatureHashingApplyTime(System.nanoTime()-t0);
return out;
}
@Override
public MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol, int rowStart, int blk) {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
int end = getEndIndex(in.getNumRows(), rowStart, blk);
// apply feature hashing column wise
for(int i = rowStart; i < end; i++) {
Object okey = in.quickGetValueThreadSafe(i, _colID - 1);
String key = okey.toString();
long code = getCode(key);
- out.quickSetValueThreadSafe(i, outputCol, (code >= 0) ? code : Double.NaN);
+ out.quickSetValue(i, outputCol, (code >= 0) ? code : Double.NaN);
}
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformFeatureHashingApplyTime(System.nanoTime()-t0);
return out;
}
@@ -145,4 +157,40 @@ public class ColumnEncoderFeatureHash extends ColumnEncoder {
super.readExternal(in);
_K = in.readLong();
}
+
+ public static class FeatureHashSparseApplyTask extends ColumnApplyTask<ColumnEncoderFeatureHash>{
+
+ public FeatureHashSparseApplyTask(ColumnEncoderFeatureHash encoder, FrameBlock input,
+ MatrixBlock out, int outputCol, int startRow, int blk) {
+ super(encoder, input, out, outputCol, startRow, blk);
+ }
+
+ public FeatureHashSparseApplyTask(ColumnEncoderFeatureHash encoder, FrameBlock input,
+ MatrixBlock out, int outputCol) {
+ super(encoder, input, out, outputCol);
+ }
+
+ @Override
+ public Object call() throws Exception {
+ if(_out.getSparseBlock() == null)
+ return null;
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+ int index = _encoder._colID - 1;
+ assert _inputF != null;
+ for(int r = _startRow; r < getEndIndex(_inputF.getNumRows(), _startRow, _blk); r++){
+ SparseRowVector row = (SparseRowVector) _out.getSparseBlock().get(r);
+ Object okey = _inputF.get(r, index);
+ String key = (okey != null) ? okey.toString() : null;
+ if(key == null)
+ throw new DMLRuntimeException("Missing Value encountered in input Frame for FeatureHash");
+ long code = _encoder.getCode(key);
+ row.values()[index] = code;
+ row.indexes()[index] = _outputCol;
+ }
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformFeatureHashingApplyTime(System.nanoTime()-t0);
+ return null;
+ }
+ }
+
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderPassThrough.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderPassThrough.java
index 7a8df24..5c7392c 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderPassThrough.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderPassThrough.java
@@ -21,16 +21,22 @@ package org.apache.sysds.runtime.transform.encode;
import static org.apache.sysds.runtime.util.UtilFunctions.getEndIndex;
+import java.util.ArrayList;
import java.util.List;
+import org.apache.commons.lang3.NotImplementedException;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.common.Types.ValueType;
+import org.apache.sysds.runtime.data.SparseRowVector;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.DependencyTask;
import org.apache.sysds.runtime.util.UtilFunctions;
+import org.apache.sysds.utils.Statistics;
public class ColumnEncoderPassThrough extends ColumnEncoder {
private static final long serialVersionUID = -8473768154646831882L;
+ private List<Integer> sparseRowsWZeros = null;
protected ColumnEncoderPassThrough(int ptCols) {
super(ptCols); // 1-based
@@ -40,37 +46,46 @@ public class ColumnEncoderPassThrough extends ColumnEncoder {
this(-1);
}
+ public List<Integer> getSparseRowsWZeros(){
+ return sparseRowsWZeros;
+ }
+
@Override
public void build(FrameBlock in) {
// do nothing
}
@Override
- public List<DependencyTask<?>> getBuildTasks(FrameBlock in, int blockSize) {
+ public List<DependencyTask<?>> getBuildTasks(FrameBlock in) {
return null;
}
@Override
- public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol) {
- return apply(in, out, outputCol, 0, -1);
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(FrameBlock in, MatrixBlock out, int outputCol, int startRow, int blk) {
+ return new PassThroughSparseApplyTask(this, in, out, outputCol, startRow, blk);
}
@Override
- public MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol) {
- return apply(in, out, outputCol, 0, -1);
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(MatrixBlock in, MatrixBlock out, int outputCol, int startRow, int blk) {
+ throw new NotImplementedException("Sparse PassThrough for MatrixBlocks not jet implemented");
}
@Override
public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol, int rowStart, int blk) {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
int col = _colID - 1; // 1-based
ValueType vt = in.getSchema()[col];
for(int i = rowStart; i < getEndIndex(in.getNumRows(), rowStart, blk); i++) {
Object val = in.get(i, col);
double v = (val == null ||
- (vt == ValueType.STRING && val.toString().isEmpty())) ? Double.NaN : UtilFunctions.objectToDouble(vt,
- val);
- out.quickSetValueThreadSafe(i, outputCol, v);
+ (vt == ValueType.STRING && val.toString().isEmpty()))
+ ? Double.NaN : UtilFunctions.objectToDouble(vt, val);
+ out.quickSetValue(i, outputCol, v);
}
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformPassThroughApplyTime(System.nanoTime()-t0);
return out;
}
@@ -79,12 +94,15 @@ public class ColumnEncoderPassThrough extends ColumnEncoder {
// only transfer from in to out
if(in == out)
return out;
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
int col = _colID - 1; // 1-based
int end = getEndIndex(in.getNumRows(), rowStart, blk);
for(int i = rowStart; i < end; i++) {
double val = in.quickGetValueThreadSafe(i, col);
- out.quickSetValueThreadSafe(i, outputCol, val);
+ out.quickSetValue(i, outputCol, val);
}
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformPassThroughApplyTime(System.nanoTime()-t0);
return out;
}
@@ -106,4 +124,58 @@ public class ColumnEncoderPassThrough extends ColumnEncoder {
public void initMetaData(FrameBlock meta) {
// do nothing
}
+
+ public static class PassThroughSparseApplyTask extends ColumnApplyTask<ColumnEncoderPassThrough>{
+
+
+ protected PassThroughSparseApplyTask(ColumnEncoderPassThrough encoder, FrameBlock input,
+ MatrixBlock out, int outputCol) {
+ super(encoder, input, out, outputCol);
+ }
+
+ protected PassThroughSparseApplyTask(ColumnEncoderPassThrough encoder, FrameBlock input, MatrixBlock out,
+ int outputCol, int startRow, int blk) {
+ super(encoder, input, out, outputCol, startRow, blk);
+ }
+
+ @Override
+ public Object call() throws Exception {
+ if(_out.getSparseBlock() == null)
+ return null;
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+ int index = _encoder._colID - 1;
+ assert _inputF != null;
+ List<Integer> sparseRowsWZeros = null;
+ ValueType vt = _inputF.getSchema()[index];
+ for(int r = _startRow; r < getEndIndex(_inputF.getNumRows(), _startRow, _blk); r++) {
+ Object val = _inputF.get(r, index);
+ double v = (val == null || (vt == ValueType.STRING && val.toString().isEmpty())) ?
+ Double.NaN : UtilFunctions.objectToDouble(vt, val);
+ SparseRowVector row = (SparseRowVector) _out.getSparseBlock().get(r);
+ if(v == 0) {
+ if(sparseRowsWZeros == null)
+ sparseRowsWZeros = new ArrayList<>();
+ sparseRowsWZeros.add(r);
+ }
+ row.values()[index] = v;
+ row.indexes()[index] = _outputCol;
+ }
+ if(sparseRowsWZeros != null){
+ synchronized (_encoder){
+ if(_encoder.sparseRowsWZeros == null)
+ _encoder.sparseRowsWZeros = new ArrayList<>();
+ _encoder.sparseRowsWZeros.addAll(sparseRowsWZeros);
+ }
+ }
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformPassThroughApplyTime(System.nanoTime()-t0);
+ return null;
+ }
+
+ public String toString() {
+ return getClass().getSimpleName() + "<ColId: " + _encoder._colID + ">";
+ }
+
+ }
+
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderRecode.java b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderRecode.java
index fd18d86..e190d74 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderRecode.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/ColumnEncoderRecode.java
@@ -33,10 +33,13 @@ import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.Callable;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.lops.Lop;
import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.data.SparseRowVector;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.utils.Statistics;
public class ColumnEncoderRecode extends ColumnEncoder {
private static final long serialVersionUID = 8213163881283341874L;
@@ -135,7 +138,11 @@ public class ColumnEncoderRecode extends ColumnEncoder {
public void build(FrameBlock in) {
if(!isApplicable())
return;
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
makeRcdMap(in, _rcdMap, _colID, 0, in.getNumRows());
+ if(DMLScript.STATISTICS){
+ Statistics.incTransformRecodeBuildTime(System.nanoTime() - t0);
+ }
}
@Override
@@ -186,18 +193,17 @@ public class ColumnEncoderRecode extends ColumnEncoder {
}
@Override
- public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol) {
- return apply(in, out, outputCol, 0, -1);
- }
-
- @Override
public MatrixBlock apply(FrameBlock in, MatrixBlock out, int outputCol, int rowStart, int blk) {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
// FrameBlock is column Major and MatrixBlock row Major this results in cache inefficiencies :(
for(int i = rowStart; i < getEndIndex(in.getNumRows(), rowStart, blk); i++) {
Object okey = in.get(i, _colID - 1);
String key = (okey != null) ? okey.toString() : null;
long code = lookupRCDMap(key);
- out.quickSetValueThreadSafe(i, outputCol, (code >= 0) ? code : Double.NaN);
+ out.quickSetValue(i, outputCol, (code >= 0) ? code : Double.NaN);
+ }
+ if(DMLScript.STATISTICS){
+ Statistics.incTransformRecodeApplyTime(System.nanoTime() - t0);
}
return out;
}
@@ -209,9 +215,16 @@ public class ColumnEncoderRecode extends ColumnEncoder {
}
@Override
- public MatrixBlock apply(MatrixBlock in, MatrixBlock out, int outputCol) {
- throw new DMLRuntimeException(
- "Recode called with MatrixBlock. Should not happen since Recode is the first " + "encoder in the Stack");
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(FrameBlock in, MatrixBlock out, int outputCol, int startRow, int blk){
+ return new RecodeSparseApplyTask(this, in ,out, outputCol, startRow, blk);
+ }
+
+ @Override
+ protected ColumnApplyTask<? extends ColumnEncoder>
+ getSparseTask(MatrixBlock in, MatrixBlock out, int outputCol, int startRow, int blk) {
+ throw new DMLRuntimeException("Recode called with MatrixBlock. Should not happen since Recode is the first " +
+ "encoder in the Stack");
}
@Override
@@ -313,6 +326,48 @@ public class ColumnEncoderRecode extends ColumnEncoder {
return _rcdMap;
}
+ private static class RecodeSparseApplyTask extends ColumnApplyTask<ColumnEncoderRecode>{
+
+ public RecodeSparseApplyTask(ColumnEncoderRecode encoder, FrameBlock input, MatrixBlock out, int outputCol) {
+ super(encoder, input, out, outputCol);
+ }
+
+ protected RecodeSparseApplyTask(ColumnEncoderRecode encoder, FrameBlock input, MatrixBlock out,
+ int outputCol, int startRow, int blk) {
+ super(encoder, input, out, outputCol, startRow, blk);
+ }
+
+ public Object call() throws Exception {
+ int index = _encoder._colID - 1;
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+ if(_out.getSparseBlock() == null)
+ return null;
+ assert _inputF != null;
+ for(int r = _startRow; r < getEndIndex(_inputF.getNumRows(), _startRow, _blk); r++) {
+ SparseRowVector row = (SparseRowVector) _out.getSparseBlock().get(r);
+ Object okey = _inputF.get(r, index);
+ String key = (okey != null) ? okey.toString() : null;
+ long code = _encoder.lookupRCDMap(key);
+ double val = (code < 0) ? Double.NaN : code;
+ row.values()[index] = val;
+ row.indexes()[index] = _outputCol;
+ }
+ if(DMLScript.STATISTICS){
+ Statistics.incTransformRecodeApplyTime(System.nanoTime() - t0);
+ }
+ return null;
+ }
+
+ @Override
+ public String toString() {
+ String str = getClass().getSimpleName() + "<ColId: " + _encoder._colID + ">";
+ if(_blk != -1)
+ str+= "<Sr: " + _startRow + ">";
+ return str;
+ }
+
+ }
+
private static class RecodePartialBuildTask implements Callable<Object> {
private final FrameBlock _input;
@@ -332,11 +387,15 @@ public class ColumnEncoderRecode extends ColumnEncoder {
@Override
public HashMap<String, Long> call() throws Exception {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
HashMap<String, Long> partialMap = new HashMap<>();
makeRcdMap(_input, partialMap, _colID, _startRow, _blockSize);
synchronized(_partialMaps) {
_partialMaps.put(_startRow, partialMap);
}
+ if(DMLScript.STATISTICS){
+ Statistics.incTransformRecodeBuildTime(System.nanoTime() - t0);
+ }
return null;
}
@@ -358,6 +417,7 @@ public class ColumnEncoderRecode extends ColumnEncoder {
@Override
public Object call() throws Exception {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
HashMap<String, Long> rcdMap = _encoder.getRcdMap();
_partialMaps.forEach((start_row, map) -> {
((HashMap<?, ?>) map).forEach((k, v) -> {
@@ -366,6 +426,9 @@ public class ColumnEncoderRecode extends ColumnEncoder {
});
});
_encoder._rcdMap = rcdMap;
+ if(DMLScript.STATISTICS){
+ Statistics.incTransformRecodeBuildTime(System.nanoTime() - t0);
+ }
return null;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderFactory.java b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderFactory.java
index 4b48d2a..012379a 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderFactory.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderFactory.java
@@ -19,16 +19,8 @@
package org.apache.sysds.runtime.transform.encode;
-import static org.apache.sysds.runtime.util.CollectionUtils.except;
-import static org.apache.sysds.runtime.util.CollectionUtils.unionDistinct;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map.Entry;
-
import org.apache.commons.lang.ArrayUtils;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.common.Types.ValueType;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
@@ -36,9 +28,19 @@ import org.apache.sysds.runtime.transform.TfUtils.TfMethod;
import org.apache.sysds.runtime.transform.encode.ColumnEncoder.EncoderType;
import org.apache.sysds.runtime.transform.meta.TfMetaUtils;
import org.apache.sysds.runtime.util.UtilFunctions;
+import org.apache.sysds.utils.Statistics;
import org.apache.wink.json4j.JSONArray;
import org.apache.wink.json4j.JSONObject;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map.Entry;
+
+import static org.apache.sysds.runtime.util.CollectionUtils.except;
+import static org.apache.sysds.runtime.util.CollectionUtils.unionDistinct;
+
public class EncoderFactory {
public static MultiColumnEncoder createEncoder(String spec, String[] colnames, int clen, FrameBlock meta) {
@@ -125,16 +127,22 @@ public class EncoderFactory {
}
// create composite decoder of all created encoders
for(Entry<Integer, List<ColumnEncoder>> listEntry : colEncoders.entrySet()) {
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformEncoderCount(listEntry.getValue().size());
lencoders.add(new ColumnEncoderComposite(listEntry.getValue()));
}
encoder = new MultiColumnEncoder(lencoders);
if(!oIDs.isEmpty()) {
encoder.addReplaceLegacyEncoder(new EncoderOmit(jSpec, colnames, schema.length, minCol, maxCol));
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformEncoderCount(1);
}
if(!mvIDs.isEmpty()) {
EncoderMVImpute ma = new EncoderMVImpute(jSpec, colnames, schema.length, minCol, maxCol);
ma.initRecodeIDList(rcIDs);
encoder.addReplaceLegacyEncoder(ma);
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformEncoderCount(1);
}
// initialize meta data w/ robustness for superset of cols
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderMVImpute.java b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderMVImpute.java
index cda6b2a..f77e690 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderMVImpute.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderMVImpute.java
@@ -19,21 +19,8 @@
package org.apache.sysds.runtime.transform.encode;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.stream.Collectors;
-
import org.apache.commons.lang.ArrayUtils;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.functionobjects.KahanPlus;
import org.apache.sysds.runtime.functionobjects.Mean;
@@ -44,10 +31,25 @@ import org.apache.sysds.runtime.transform.TfUtils.TfMethod;
import org.apache.sysds.runtime.transform.meta.TfMetaUtils;
import org.apache.sysds.runtime.util.IndexRange;
import org.apache.sysds.runtime.util.UtilFunctions;
+import org.apache.sysds.utils.Statistics;
import org.apache.wink.json4j.JSONArray;
import org.apache.wink.json4j.JSONException;
import org.apache.wink.json4j.JSONObject;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.stream.Collectors;
+
public class EncoderMVImpute extends LegacyEncoder {
private static final long serialVersionUID = 9057868620144662194L;
// objects required to compute mean and variance of all non-missing entries
@@ -173,6 +175,7 @@ public class EncoderMVImpute extends LegacyEncoder {
@Override
public void build(FrameBlock in) {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
try {
for(int j = 0; j < _colList.length; j++) {
int colID = _colList[j];
@@ -215,10 +218,13 @@ public class EncoderMVImpute extends LegacyEncoder {
catch(Exception ex) {
throw new RuntimeException(ex);
}
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformImputeBuildTime(System.nanoTime()-t0);
}
@Override
public MatrixBlock apply(FrameBlock in, MatrixBlock out) {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
for(int i = 0; i < in.getNumRows(); i++) {
for(int j = 0; j < _colList.length; j++) {
int colID = _colList[j];
@@ -226,6 +232,8 @@ public class EncoderMVImpute extends LegacyEncoder {
out.quickSetValue(i, colID - 1, Double.parseDouble(_replacementList[j]));
}
}
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformImputeApplyTime(System.nanoTime()-t0);
return out;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderOmit.java b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderOmit.java
index db61fc1..c2f3b68 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderOmit.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/EncoderOmit.java
@@ -19,16 +19,9 @@
package org.apache.sysds.runtime.transform.encode;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Objects;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.common.Types.ValueType;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
@@ -37,9 +30,18 @@ import org.apache.sysds.runtime.transform.TfUtils.TfMethod;
import org.apache.sysds.runtime.transform.meta.TfMetaUtils;
import org.apache.sysds.runtime.util.IndexRange;
import org.apache.sysds.runtime.util.UtilFunctions;
+import org.apache.sysds.utils.Statistics;
import org.apache.wink.json4j.JSONException;
import org.apache.wink.json4j.JSONObject;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+
public class EncoderOmit extends LegacyEncoder {
/*
* THIS CLASS IS ONLY FOR LEGACY SUPPORT!!! and will be fazed out slowly.
@@ -126,6 +128,7 @@ public class EncoderOmit extends LegacyEncoder {
public MatrixBlock apply(FrameBlock in, MatrixBlock out) {
// local rmRows for broadcasting encoder in spark
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
boolean[] rmRows;
if(_federated)
rmRows = _rmRows;
@@ -148,7 +151,8 @@ public class EncoderOmit extends LegacyEncoder {
}
_rmRows = rmRows;
-
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformOmitApplyTime(System.nanoTime()-t0);
return ret;
}
diff --git a/src/main/java/org/apache/sysds/runtime/transform/encode/MultiColumnEncoder.java b/src/main/java/org/apache/sysds/runtime/transform/encode/MultiColumnEncoder.java
index 6db63f5..73d33a9 100644
--- a/src/main/java/org/apache/sysds/runtime/transform/encode/MultiColumnEncoder.java
+++ b/src/main/java/org/apache/sysds/runtime/transform/encode/MultiColumnEncoder.java
@@ -28,6 +28,8 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
@@ -36,22 +38,31 @@ import java.util.stream.Collectors;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.api.DMLScript;
import org.apache.sysds.common.Types;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.data.SparseBlock;
import org.apache.sysds.runtime.data.SparseBlockMCSR;
+import org.apache.sysds.runtime.data.SparseRowVector;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.DependencyTask;
import org.apache.sysds.runtime.util.DependencyThreadPool;
import org.apache.sysds.runtime.util.DependencyWrapperTask;
import org.apache.sysds.runtime.util.IndexRange;
+import org.apache.sysds.utils.Statistics;
public class MultiColumnEncoder implements Encoder {
protected static final Log LOG = LogFactory.getLog(MultiColumnEncoder.class.getName());
private static final boolean MULTI_THREADED = true;
- public static boolean MULTI_THREADED_STAGES = true;
+ // If true build and apply separately by placing a synchronization barrier
+ public static boolean MULTI_THREADED_STAGES = false;
+
+ // Only affects if MULTI_THREADED_STAGES is true
+ // if true apply tasks for each column will complete
+ // before the next will start.
+ public static boolean APPLY_ENCODER_SEPARATE_STAGES = false;
private List<ColumnEncoderComposite> _columnEncoders;
// These encoders are deprecated and will be phased out soon.
@@ -60,18 +71,6 @@ public class MultiColumnEncoder implements Encoder {
private int _colOffset = 0; // offset for federated Workers who are using subrange encoders
private FrameBlock _meta = null;
- // TEMP CONSTANTS for testing only
- //private int APPLY_BLOCKSIZE = 0; // temp only for testing until automatic calculation of block size
- public static int BUILD_BLOCKSIZE = 0;
-
- /*public void setApplyBlockSize(int blk) {
- APPLY_BLOCKSIZE = blk;
- }*/
-
- public void setBuildBlockSize(int blk) {
- BUILD_BLOCKSIZE = blk;
- }
-
public MultiColumnEncoder(List<ColumnEncoderComposite> columnEncoders) {
_columnEncoders = columnEncoders;
}
@@ -90,6 +89,7 @@ public class MultiColumnEncoder implements Encoder {
if(MULTI_THREADED && k > 1 && !MULTI_THREADED_STAGES && !hasLegacyEncoder()) {
out = new MatrixBlock();
DependencyThreadPool pool = new DependencyThreadPool(k);
+ LOG.debug("Encoding with full DAG on " + k + " Threads");
try {
pool.submitAllAndWait(getEncodeTasks(in, out, pool));
}
@@ -98,10 +98,10 @@ public class MultiColumnEncoder implements Encoder {
e.printStackTrace();
}
pool.shutdown();
- out.recomputeNonZeros();
- return out;
+ outputMatrixPostProcessing(out);
}
else {
+ LOG.debug("Encoding with staged approach on: " + k + " Threads");
build(in, k);
if(_legacyMVImpute != null) {
// These operations are redundant for every encoder excluding the legacyMVImpute, the workaround to
@@ -127,9 +127,10 @@ public class MultiColumnEncoder implements Encoder {
List<DependencyTask<?>> applyTAgg = null;
Map<Integer[], Integer[]> depMap = new HashMap<>();
boolean hasDC = getColumnEncoders(ColumnEncoderDummycode.class).size() > 0;
+ boolean applyOffsetDep = false;
tasks.add(DependencyThreadPool.createDependencyTask(new InitOutputMatrixTask(this, in, out)));
for(ColumnEncoderComposite e : _columnEncoders) {
- List<DependencyTask<?>> buildTasks = e.getBuildTasks(in, BUILD_BLOCKSIZE);
+ List<DependencyTask<?>> buildTasks = e.getBuildTasks(in);
tasks.addAll(buildTasks);
if(buildTasks.size() > 0) {
@@ -152,11 +153,14 @@ public class MultiColumnEncoder implements Encoder {
// colUpdateTask can start when all domain sizes, because it can now calculate the offsets for
// each column
depMap.put(new Integer[] {-2, -1}, new Integer[] {tasks.size() - 1, tasks.size()});
+ buildTasks.forEach(t -> t.setPriority(5));
+ applyOffsetDep = true;
}
- if(hasDC) {
+ if(hasDC && applyOffsetDep) {
// Apply Task dependency to output col update task (is last in list)
- // All ApplyTasks need to wait for this task so they all have the correct offsets.
+ // All ApplyTasks need to wait for this task, so they all have the correct offsets.
+ // But only for the columns that come after the first DC coder since they don't have an offset
depMap.put(new Integer[] {tasks.size(), tasks.size() + 1}, new Integer[] {-2, -1});
applyTAgg = applyTAgg == null ? new ArrayList<>() : applyTAgg;
@@ -195,7 +199,7 @@ public class MultiColumnEncoder implements Encoder {
private List<DependencyTask<?>> getBuildTasks(FrameBlock in) {
List<DependencyTask<?>> tasks = new ArrayList<>();
for(ColumnEncoderComposite columnEncoder : _columnEncoders) {
- tasks.addAll(columnEncoder.getBuildTasks(in, BUILD_BLOCKSIZE));
+ tasks.addAll(columnEncoder.getBuildTasks(in));
}
return tasks;
}
@@ -241,23 +245,23 @@ public class MultiColumnEncoder implements Encoder {
if(in.getNumColumns() != numEncoders)
throw new DMLRuntimeException("Not every column in has a CompositeEncoder. Please make sure every column "
+ "has a encoder or slice the input accordingly");
- // Block allocation for MT access
- outputMatrixPreProcessing(out, in);
// TODO smart checks
if(MULTI_THREADED && k > 1) {
+ // Block allocation for MT access
+ outputMatrixPreProcessing(out, in);
applyMT(in, out, outputCol, k);
}
else {
int offset = outputCol;
for(ColumnEncoderComposite columnEncoder : _columnEncoders) {
columnEncoder.apply(in, out, columnEncoder._colID - 1 + offset);
- if(columnEncoder.hasEncoder(ColumnEncoderDummycode.class))
+ if (columnEncoder.hasEncoder(ColumnEncoderDummycode.class))
offset += columnEncoder.getEncoder(ColumnEncoderDummycode.class)._domainSize - 1;
}
}
// Recomputing NNZ since we access the block directly
// TODO set NNZ explicit count them in the encoders
- out.recomputeNonZeros();
+ outputMatrixPostProcessing(out);
if(_legacyOmit != null)
out = _legacyOmit.apply(in, out);
if(_legacyMVImpute != null)
@@ -280,16 +284,26 @@ public class MultiColumnEncoder implements Encoder {
private void applyMT(FrameBlock in, MatrixBlock out, int outputCol, int k) {
DependencyThreadPool pool = new DependencyThreadPool(k);
try {
- pool.submitAllAndWait(getApplyTasks(in, out, outputCol));
+ if(APPLY_ENCODER_SEPARATE_STAGES){
+ int offset = outputCol;
+ for (ColumnEncoderComposite e : _columnEncoders) {
+ pool.submitAllAndWait(e.getApplyTasks(in, out, e._colID - 1 + offset));
+ if (e.hasEncoder(ColumnEncoderDummycode.class))
+ offset += e.getEncoder(ColumnEncoderDummycode.class)._domainSize - 1;
+ }
+ }else{
+ pool.submitAllAndWait(getApplyTasks(in, out, outputCol));
+ }
}
catch(ExecutionException | InterruptedException e) {
- LOG.error("MT Column encode failed");
+ LOG.error("MT Column apply failed");
e.printStackTrace();
}
pool.shutdown();
}
private static void outputMatrixPreProcessing(MatrixBlock output, FrameBlock input) {
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
output.allocateBlock();
if(output.isInSparseFormat()) {
SparseBlock block = output.getSparseBlock();
@@ -300,8 +314,33 @@ public class MultiColumnEncoder implements Encoder {
// allocate all sparse rows so MT sync can be done.
// should be rare that rows have only 0
block.allocate(r, input.getNumColumns());
+ // Setting the size here makes it possible to run all sparse apply tasks without any sync
+ // could become problematic if the input is very sparse since we allocate the same size as the input
+ // should be fine in theory ;)
+ ((SparseRowVector)block.get(r)).setSize(input.getNumColumns());
+ }
+ }
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformOutMatrixPreProcessingTime(System.nanoTime()-t0);
+ }
+
+ private void outputMatrixPostProcessing(MatrixBlock output){
+ long t0 = DMLScript.STATISTICS ? System.nanoTime() : 0;
+ Set<Integer> indexSet = getColumnEncoders(ColumnEncoderPassThrough.class).stream()
+ .map(ColumnEncoderPassThrough::getSparseRowsWZeros).flatMap(l -> {
+ if(l == null)
+ return null;
+ return l.stream();
+ }).collect(Collectors.toSet());
+ if(!indexSet.stream().allMatch(Objects::isNull)){
+ for(Integer row : indexSet){
+ // TODO: Maybe MT in special cases when the number of rows is large
+ output.getSparseBlock().get(row).compact();
}
}
+ output.recomputeNonZeros();
+ if(DMLScript.STATISTICS)
+ Statistics.incTransformOutMatrixPostProcessingTime(System.nanoTime()-t0);
}
@Override
@@ -660,7 +699,7 @@ public class MultiColumnEncoder implements Encoder {
}
/*
- * Currently not in use will be integrated in the future
+ * Currently, not in use will be integrated in the future
*/
@SuppressWarnings("unused")
private static class MultiColumnLegacyBuildTask implements Callable<Object> {
diff --git a/src/main/java/org/apache/sysds/runtime/util/DependencyTask.java b/src/main/java/org/apache/sysds/runtime/util/DependencyTask.java
index 5aff39b..17351a6 100644
--- a/src/main/java/org/apache/sysds/runtime/util/DependencyTask.java
+++ b/src/main/java/org/apache/sysds/runtime/util/DependencyTask.java
@@ -25,16 +25,20 @@ import java.util.concurrent.CompletableFuture;
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.runtime.DMLRuntimeException;
-public class DependencyTask<E> implements Callable<E> {
+public class DependencyTask<E> implements Comparable<DependencyTask<?>>, Callable<E> {
public static final boolean ENABLE_DEBUG_DATA = false;
+ protected static final Log LOG = LogFactory.getLog(DependencyTask.class.getName());
private final Callable<E> _task;
protected final List<DependencyTask<?>> _dependantTasks;
public List<DependencyTask<?>> _dependencyTasks = null; // only for debugging
private CompletableFuture<Future<?>> _future;
private int _rdy = 0;
+ private Integer _priority = 0;
private ExecutorService _pool;
public DependencyTask(Callable<E> task, List<DependencyTask<?>> dependantTasks) {
@@ -54,6 +58,10 @@ public class DependencyTask<E> implements Callable<E> {
return _rdy == 0;
}
+ public void setPriority(int priority) {
+ _priority = priority;
+ }
+
private boolean decrease() {
synchronized(this) {
_rdy -= 1;
@@ -68,7 +76,11 @@ public class DependencyTask<E> implements Callable<E> {
@Override
public E call() throws Exception {
+ LOG.debug("Executing Task: " + this);
+ long t0 = System.nanoTime();
E ret = _task.call();
+ LOG.debug("Finished Task: " + this + " in: " +
+ (String.format("%.3f", (System.nanoTime()-t0)*1e-9)) + "sec.");
_dependantTasks.forEach(t -> {
if(t.decrease()) {
if(_pool == null)
@@ -79,4 +91,14 @@ public class DependencyTask<E> implements Callable<E> {
return ret;
}
+
+ @Override
+ public String toString(){
+ return _task.toString() + "<Prio: " + _priority + ">" + "<Waiting: " + _dependantTasks.size() + ">";
+ }
+
+ @Override
+ public int compareTo(DependencyTask<?> task) {
+ return -1 * this._priority.compareTo(task._priority);
+ }
}
diff --git a/src/main/java/org/apache/sysds/runtime/util/DependencyThreadPool.java b/src/main/java/org/apache/sysds/runtime/util/DependencyThreadPool.java
index 4fdd63a..50675d6 100644
--- a/src/main/java/org/apache/sysds/runtime/util/DependencyThreadPool.java
+++ b/src/main/java/org/apache/sysds/runtime/util/DependencyThreadPool.java
@@ -19,7 +19,12 @@
package org.apache.sysds.runtime.util;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.runtime.DMLRuntimeException;
+
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -31,11 +36,10 @@ import java.util.concurrent.Future;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import org.apache.sysds.runtime.DMLRuntimeException;
-
public class DependencyThreadPool {
+ protected static final Log LOG = LogFactory.getLog(DependencyThreadPool.class.getName());
private final ExecutorService _pool;
public DependencyThreadPool(int k) {
@@ -50,6 +54,8 @@ public class DependencyThreadPool {
List<Future<Future<?>>> futures = new ArrayList<>();
List<Integer> rdyTasks = new ArrayList<>();
int i = 0;
+ // sort by priority
+ Collections.sort(dtasks);
for(DependencyTask<?> t : dtasks) {
CompletableFuture<Future<?>> f = new CompletableFuture<>();
t.addPool(_pool);
@@ -63,6 +69,8 @@ public class DependencyThreadPool {
futures.add(f);
i++;
}
+ LOG.debug("Initial Starting tasks: \n\t" +
+ rdyTasks.stream().map(index -> dtasks.get(index).toString()).collect(Collectors.joining("\n\t")));
// Two stages to avoid race condition!
for(Integer index : rdyTasks) {
synchronized(_pool) {
diff --git a/src/main/java/org/apache/sysds/runtime/util/UtilFunctions.java b/src/main/java/org/apache/sysds/runtime/util/UtilFunctions.java
index de6d7d6..7431a82 100644
--- a/src/main/java/org/apache/sysds/runtime/util/UtilFunctions.java
+++ b/src/main/java/org/apache/sysds/runtime/util/UtilFunctions.java
@@ -989,6 +989,15 @@ public class UtilFunctions {
return (blockSize <= 0)? arrayLength: Math.min(arrayLength, startIndex + blockSize);
}
+ public static int[] getBlockSizes(int num, int numBlocks){
+ int[] blockSizes = new int[numBlocks];
+ Arrays.fill(blockSizes, num/numBlocks);
+ for (int i = 0; i < num%numBlocks; i++){
+ blockSizes[i]++;
+ }
+ return blockSizes;
+ }
+
public static String[] splitRecodeEntry(String s) {
//forward to column encoder, as UtilFunctions available in map context
return ColumnEncoderRecode.splitRecodeMapEntry(s);
diff --git a/src/main/java/org/apache/sysds/utils/Statistics.java b/src/main/java/org/apache/sysds/utils/Statistics.java
index cacd2f7..d91d9c5 100644
--- a/src/main/java/org/apache/sysds/utils/Statistics.java
+++ b/src/main/java/org/apache/sysds/utils/Statistics.java
@@ -19,20 +19,6 @@
package org.apache.sysds.utils;
-import java.lang.management.CompilationMXBean;
-import java.lang.management.GarbageCollectorMXBean;
-import java.lang.management.ManagementFactory;
-import java.text.DecimalFormat;
-import java.util.Arrays;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.DoubleAdder;
-import java.util.concurrent.atomic.LongAdder;
-
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.sysds.api.DMLScript;
@@ -51,6 +37,20 @@ import org.apache.sysds.runtime.lineage.LineageCacheConfig.ReuseCacheType;
import org.apache.sysds.runtime.lineage.LineageCacheStatistics;
import org.apache.sysds.runtime.privacy.CheckedConstraintsLog;
+import java.lang.management.CompilationMXBean;
+import java.lang.management.GarbageCollectorMXBean;
+import java.lang.management.ManagementFactory;
+import java.text.DecimalFormat;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.DoubleAdder;
+import java.util.concurrent.atomic.LongAdder;
+
/**
* This class captures all statistics.
*/
@@ -149,7 +149,28 @@ public class Statistics
private static final LongAdder lTotalUIPVar = new LongAdder();
private static final LongAdder lTotalLix = new LongAdder();
private static final LongAdder lTotalLixUIP = new LongAdder();
-
+
+ // Transformencode stats
+ private static final LongAdder transformEncoderCount = new LongAdder();
+
+ //private static final LongAdder transformBuildTime = new LongAdder();
+ private static final LongAdder transformRecodeBuildTime = new LongAdder();
+ private static final LongAdder transformBinningBuildTime = new LongAdder();
+ private static final LongAdder transformImputeBuildTime = new LongAdder();
+
+ //private static final LongAdder transformApplyTime = new LongAdder();
+ private static final LongAdder transformRecodeApplyTime = new LongAdder();
+ private static final LongAdder transformDummyCodeApplyTime = new LongAdder();
+ private static final LongAdder transformPassThroughApplyTime = new LongAdder();
+ private static final LongAdder transformFeatureHashingApplyTime = new LongAdder();
+ private static final LongAdder transformBinningApplyTime = new LongAdder();
+ private static final LongAdder transformOmitApplyTime = new LongAdder();
+ private static final LongAdder transformImputeApplyTime = new LongAdder();
+
+
+ private static final LongAdder transformOutMatrixPreProcessingTime = new LongAdder();
+ private static final LongAdder transformOutMatrixPostProcessingTime = new LongAdder();
+
// Federated stats
private static final LongAdder federatedReadCount = new LongAdder();
private static final LongAdder federatedPutCount = new LongAdder();
@@ -649,6 +670,70 @@ public class Statistics
public static void accFedPSCommunicationTime(long t) { fedPSCommunicationTime.add(t);}
+ public static void incTransformEncoderCount(long encoders){
+ transformEncoderCount.add(encoders);
+ }
+
+ public static void incTransformRecodeApplyTime(long t){
+ transformRecodeApplyTime.add(t);
+ }
+
+ public static void incTransformDummyCodeApplyTime(long t){
+ transformDummyCodeApplyTime.add(t);
+ }
+
+ public static void incTransformBinningApplyTime(long t){
+ transformBinningApplyTime.add(t);
+ }
+
+ public static void incTransformPassThroughApplyTime(long t){
+ transformPassThroughApplyTime.add(t);
+ }
+
+ public static void incTransformFeatureHashingApplyTime(long t){
+ transformFeatureHashingApplyTime.add(t);
+ }
+
+ public static void incTransformOmitApplyTime(long t) {
+ transformOmitApplyTime.add(t);
+ }
+
+ public static void incTransformImputeApplyTime(long t) {
+ transformImputeApplyTime.add(t);
+ }
+
+ public static void incTransformRecodeBuildTime(long t){
+ transformRecodeBuildTime.add(t);
+ }
+
+ public static void incTransformBinningBuildTime(long t){
+ transformBinningBuildTime.add(t);
+ }
+
+ public static void incTransformImputeBuildTime(long t) {
+ transformImputeBuildTime.add(t);
+ }
+
+ public static void incTransformOutMatrixPreProcessingTime(long t){
+ transformOutMatrixPreProcessingTime.add(t);
+ }
+
+ public static void incTransformOutMatrixPostProcessingTime(long t){
+ transformOutMatrixPostProcessingTime.add(t);
+ }
+
+ public static long getTransformEncodeBuildTime(){
+ return transformBinningBuildTime.longValue() + transformImputeBuildTime.longValue() +
+ transformRecodeBuildTime.longValue();
+ }
+
+ public static long getTransformEncodeApplyTime(){
+ return transformDummyCodeApplyTime.longValue() + transformBinningApplyTime.longValue() +
+ transformFeatureHashingApplyTime.longValue() + transformPassThroughApplyTime.longValue() +
+ transformRecodeApplyTime.longValue() + transformOmitApplyTime.longValue() +
+ transformImputeApplyTime.longValue();
+ }
+
public static String getCPHeavyHitterCode( Instruction inst )
{
String opcode = null;
@@ -1129,6 +1214,50 @@ public class Statistics
federatedExecuteInstructionCount.longValue() + "/" +
federatedExecuteUDFCount.longValue() + ".\n");
}
+ if( transformEncoderCount.longValue() > 0) {
+ //TODO: Cleanup and condense
+ sb.append("TransformEncode num. encoders:\t").append(transformEncoderCount.longValue()).append("\n");
+ sb.append("TransformEncode build time:\t").append(String.format("%.3f",
+ getTransformEncodeBuildTime()*1e-9)).append(" sec.\n");
+ if(transformRecodeBuildTime.longValue() > 0)
+ sb.append("\tRecode build time:\t").append(String.format("%.3f",
+ transformRecodeBuildTime.longValue()*1e-9)).append(" sec.\n");
+ if(transformBinningBuildTime.longValue() > 0)
+ sb.append("\tBinning build time:\t").append(String.format("%.3f",
+ transformBinningBuildTime.longValue()*1e-9)).append(" sec.\n");
+ if(transformImputeBuildTime.longValue() > 0)
+ sb.append("\tImpute build time:\t").append(String.format("%.3f",
+ transformImputeBuildTime.longValue()*1e-9)).append(" sec.\n");
+
+ sb.append("TransformEncode apply time:\t").append(String.format("%.3f",
+ getTransformEncodeApplyTime()*1e-9)).append(" sec.\n");
+ if(transformRecodeApplyTime.longValue() > 0)
+ sb.append("\tRecode apply time:\t").append(String.format("%.3f",
+ transformRecodeApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(transformBinningApplyTime.longValue() > 0)
+ sb.append("\tBinning apply time:\t").append(String.format("%.3f",
+ transformBinningApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(transformDummyCodeApplyTime.longValue() > 0)
+ sb.append("\tDummyCode apply time:\t").append(String.format("%.3f",
+ transformDummyCodeApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(transformFeatureHashingApplyTime.longValue() > 0)
+ sb.append("\tHashing apply time:\t").append(String.format("%.3f",
+ transformFeatureHashingApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(transformPassThroughApplyTime.longValue() > 0)
+ sb.append("\tPassThrough apply time:\t").append(String.format("%.3f",
+ transformPassThroughApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(transformOmitApplyTime.longValue() > 0)
+ sb.append("\tOmit apply time:\t").append(String.format("%.3f",
+ transformOmitApplyTime.longValue()*1e-9)).append(" sec.\n");
+ if(transformImputeApplyTime.longValue() > 0)
+ sb.append("\tImpute apply time:\t").append(String.format("%.3f",
+ transformImputeApplyTime.longValue()*1e-9)).append(" sec.\n");
+
+ sb.append("TransformEncode PreProc. time:\t").append(String.format("%.3f",
+ transformOutMatrixPreProcessingTime.longValue()*1e-9)).append(" sec.\n");
+ sb.append("TransformEncode PostProc. time:\t").append(String.format("%.3f",
+ transformOutMatrixPostProcessingTime.longValue()*1e-9)).append(" sec.\n");
+ }
if(ConfigurationManager.isCompressionEnabled()){
DMLCompressionStatistics.display(sb);
diff --git a/src/test/java/org/apache/sysds/test/functions/transform/mt/TransformFrameBuildMultithreadedTest.java b/src/test/java/org/apache/sysds/test/functions/transform/mt/TransformFrameBuildMultithreadedTest.java
index 8824b9d..b70571b 100644
--- a/src/test/java/org/apache/sysds/test/functions/transform/mt/TransformFrameBuildMultithreadedTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/transform/mt/TransformFrameBuildMultithreadedTest.java
@@ -30,6 +30,7 @@ import org.apache.sysds.common.Types;
import org.apache.sysds.runtime.io.FileFormatPropertiesCSV;
import org.apache.sysds.runtime.io.FrameReaderFactory;
import org.apache.sysds.runtime.matrix.data.FrameBlock;
+import org.apache.sysds.runtime.transform.encode.ColumnEncoder;
import org.apache.sysds.runtime.transform.encode.ColumnEncoderBin;
import org.apache.sysds.runtime.transform.encode.ColumnEncoderRecode;
import org.apache.sysds.runtime.transform.encode.EncoderFactory;
@@ -173,6 +174,7 @@ public class TransformFrameBuildMultithreadedTest extends AutomatedTestBase {
.readFrameFromHDFS(DATASET, -1L, -1L);
StringBuilder specSb = new StringBuilder();
Files.readAllLines(Paths.get(SPEC)).forEach(s -> specSb.append(s).append("\n"));
+ ColumnEncoder.BUILD_ROW_BLOCKS_PER_COLUMN = Math.max(blockSize, 1);
MultiColumnEncoder encoderS = EncoderFactory.createEncoder(specSb.toString(), input.getColumnNames(),
input.getNumColumns(), null);
MultiColumnEncoder encoderM = EncoderFactory.createEncoder(specSb.toString(), input.getColumnNames(),
diff --git a/src/test/java/org/apache/sysds/test/functions/transform/mt/TransformFrameEncodeMultithreadedTest.java b/src/test/java/org/apache/sysds/test/functions/transform/mt/TransformFrameEncodeMultithreadedTest.java
index 2156250..6679f36 100644
--- a/src/test/java/org/apache/sysds/test/functions/transform/mt/TransformFrameEncodeMultithreadedTest.java
+++ b/src/test/java/org/apache/sysds/test/functions/transform/mt/TransformFrameEncodeMultithreadedTest.java
@@ -48,7 +48,7 @@ public class TransformFrameEncodeMultithreadedTest extends AutomatedTestBase {
private final static String DATASET1 = "homes3/homes.csv";
private final static String SPEC1 = "homes3/homes.tfspec_recode.json";
private final static String SPEC2 = "homes3/homes.tfspec_dummy.json";
- private final static String SPEC2all = "homes3/homes.tfspec_dummy_all.json";
+ private final static String SPEC2sparse = "homes3/homes.tfspec_dummy_sparse.json";
private final static String SPEC3 = "homes3/homes.tfspec_bin.json"; // recode
private final static String SPEC6 = "homes3/homes.tfspec_recode_dummy.json";
private final static String SPEC7 = "homes3/homes.tfspec_binDummy.json"; // recode+dummy
@@ -164,7 +164,7 @@ public class TransformFrameEncodeMultithreadedTest extends AutomatedTestBase {
DATASET = DATASET1;
break;
case DUMMY_ALL:
- SPEC = SPEC2all;
+ SPEC = SPEC2sparse;
DATASET = DATASET1;
break;
case BIN:
diff --git a/src/test/resources/datasets/homes3/homes.tfspec_dummy_all.json b/src/test/resources/datasets/homes3/homes.tfspec_dummy_all.json
deleted file mode 100644
index 65b8fee..0000000
--- a/src/test/resources/datasets/homes3/homes.tfspec_dummy_all.json
+++ /dev/null
@@ -1 +0,0 @@
-{"ids": true, "dummycode": [ 2, 7, 1, 3, 4, 5, 6, 8, 9 ] }
\ No newline at end of file
diff --git a/src/test/resources/datasets/homes3/homes.tfspec_dummy_sparse.json b/src/test/resources/datasets/homes3/homes.tfspec_dummy_sparse.json
new file mode 100644
index 0000000..ed48308
--- /dev/null
+++ b/src/test/resources/datasets/homes3/homes.tfspec_dummy_sparse.json
@@ -0,0 +1 @@
+{"ids": true, "dummycode": [ 2, 7, 1, 3, 4, 6, 8, 9 ] }
\ No newline at end of file