This is an automated email from the ASF dual-hosted git repository.
baunsgaard pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/systemds.git
The following commit(s) were added to refs/heads/main by this push:
new eb2ca62487 [SYSTEMDS-3815] Fused table sequence
eb2ca62487 is described below
commit eb2ca6248794c87d4b071b6ad06e9925ad4ebea0
Author: Sebastian Baunsgaard <[email protected]>
AuthorDate: Tue Jan 21 11:57:34 2025 +0100
[SYSTEMDS-3815] Fused table sequence
This commit contains a new fused operator for:
table(seq(1, nrow(A)), A, w)
That removes the need to generate a vector of incrementing integers in the
size of A. Previously, we already had support for this operator and called it
rexpand. However, that implementation still allocated the seq vector.
We see a 1.4x improvement in the rexpand operator, and with the addition of
removing the seq allocation, it further improves to 1.72x.
The change is not fully integrated into the Federated Instructions and
needs additional work. The current workaround tries to compile the previous
instruction for federated use cases.
Closes #2181
---
src/main/java/org/apache/sysds/hops/TernaryOp.java | 15 +-
.../RewriteAlgebraicSimplificationDynamic.java | 23 +++
src/main/java/org/apache/sysds/lops/Ctable.java | 12 +-
.../sysds/parser/BuiltinFunctionExpression.java | 4 +-
.../apache/sysds/resource/cost/CostEstimator.java | 2 +-
.../runtime/compress/CompressedMatrixBlock.java | 6 +
.../runtime/compress/colgroup/ColGroupDDC.java | 36 ++---
.../sysds/runtime/compress/lib/CLALibReshape.java | 169 +++++++++++++++++++
.../sysds/runtime/compress/lib/CLALibRexpand.java | 149 +++++++++++++++++
.../controlprogram/federated/FederationMap.java | 1 -
.../instructions/cp/CtableCPInstruction.java | 31 ++--
.../instructions/cp/ReshapeCPInstruction.java | 7 +-
.../instructions/fed/CtableFEDInstruction.java | 7 +-
.../sysds/runtime/matrix/data/LibMatrixReorg.java | 180 +++++++++++++++++++++
.../sysds/runtime/matrix/data/MatrixBlock.java | 42 +----
.../component/compress/lib/CLALibReshapeTests.java | 113 +++++++++++++
.../test/component/compress/lib/SeqTableTest.java | 91 +++++++++++
.../sysds/test/component/matrix/SeqTableTest.java | 106 ++++++++++++
.../test/component/resource/RecompilationTest.java | 31 ++--
.../wordembedding/WordEmbeddingUseCase.java | 179 ++++++++++++++++++++
.../primitives/part5/FederatedCtableTest.java | 6 +-
.../table/CompressedTableOverwriteTest/01.dml | 53 ++++++
.../functions/compress/wordembedding/01.dml | 36 +++++
23 files changed, 1199 insertions(+), 100 deletions(-)
diff --git a/src/main/java/org/apache/sysds/hops/TernaryOp.java
b/src/main/java/org/apache/sysds/hops/TernaryOp.java
index 87c99fc5c0..0334dbbb2f 100644
--- a/src/main/java/org/apache/sysds/hops/TernaryOp.java
+++ b/src/main/java/org/apache/sysds/hops/TernaryOp.java
@@ -307,7 +307,7 @@ public class TernaryOp extends MultiThreadedHop
}
Ctable ternary = new Ctable(inputLops, ternaryOp,
- getDataType(), getValueType(), ignoreZeros,
outputEmptyBlocks, et);
+ getDataType(), getValueType(), ignoreZeros,
outputEmptyBlocks, et,
OptimizerUtils.getConstrainedNumThreads(getMaxNumThreads()));
ternary.getOutputParameters().setDimensions(getDim1(),
getDim2(), getBlocksize(), -1);
setLineNumbers(ternary);
@@ -480,6 +480,10 @@ public class TernaryOp extends MultiThreadedHop
}
+ public ExecType findExecTypeTernaryOp(){
+ return _etype == null ?
optFindExecType(OptimizerUtils.ALLOW_TRANSITIVE_SPARK_EXEC_TYPE) : _etype;
+ }
+
@Override
protected ExecType optFindExecType(boolean transitive)
{
@@ -637,7 +641,7 @@ public class TernaryOp extends MultiThreadedHop
return ret;
}
- private boolean isSequenceRewriteApplicable( boolean left )
+ public boolean isSequenceRewriteApplicable( boolean left )
{
boolean ret = false;
@@ -651,7 +655,9 @@ public class TernaryOp extends MultiThreadedHop
{
Hop input1 = getInput().get(0);
Hop input2 = getInput().get(1);
- if( input1.getDataType() == DataType.MATRIX &&
input2.getDataType() == DataType.MATRIX )
+ if( (input1.getDataType() == DataType.MATRIX
+ || input1.getDataType() ==
DataType.SCALAR )
+ && input2.getDataType() ==
DataType.MATRIX )
{
//probe rewrite on left input
if( left && input1 instanceof DataGenOp
)
@@ -663,6 +669,9 @@ public class TernaryOp extends MultiThreadedHop
||
dgop.getIncrementValue()==1.0; //set by recompiler
}
}
+ if( left && input1 instanceof LiteralOp
&& ((LiteralOp)input1).getStringValue().contains("seq(")){
+ ret = true;
+ }
//probe rewrite on right input
if( !left && input2 instanceof
DataGenOp )
{
diff --git
a/src/main/java/org/apache/sysds/hops/rewrite/RewriteAlgebraicSimplificationDynamic.java
b/src/main/java/org/apache/sysds/hops/rewrite/RewriteAlgebraicSimplificationDynamic.java
index 508499c725..eb51348a8e 100644
---
a/src/main/java/org/apache/sysds/hops/rewrite/RewriteAlgebraicSimplificationDynamic.java
+++
b/src/main/java/org/apache/sysds/hops/rewrite/RewriteAlgebraicSimplificationDynamic.java
@@ -30,6 +30,7 @@ import java.util.List;
import org.apache.sysds.common.Types.AggOp;
import org.apache.sysds.common.Types.DataType;
import org.apache.sysds.common.Types.Direction;
+import org.apache.sysds.common.Types.ExecType;
import org.apache.sysds.common.Types.OpOp1;
import org.apache.sysds.common.Types.OpOp2;
import org.apache.sysds.common.Types.OpOp3;
@@ -209,6 +210,8 @@ public class RewriteAlgebraicSimplificationDynamic extends
HopRewriteRule
//process childs recursively after rewrites (to
investigate pattern newly created by rewrites)
if( !descendFirst )
rule_AlgebraicSimplification(hi, descendFirst);
+
+ hi = fuseSeqAndTableExpand(hi);
}
hop.setVisited();
@@ -2913,4 +2916,24 @@ public class RewriteAlgebraicSimplificationDynamic
extends HopRewriteRule
}
return hi;
}
+
+
+ private static Hop fuseSeqAndTableExpand(Hop hi) {
+
+ if(TernaryOp.ALLOW_CTABLE_SEQUENCE_REWRITES && hi instanceof
TernaryOp ) {
+ TernaryOp thop = (TernaryOp) hi;
+ thop.getOp();
+
+ if(thop.isSequenceRewriteApplicable(true) &&
thop.findExecTypeTernaryOp() == ExecType.CP &&
+ thop.getInput(1).getForcedExecType() !=
Types.ExecType.FED) {
+ Hop input1 = thop.getInput(0);
+ if(input1 instanceof DataGenOp){
+ Hop literal = new LiteralOp("seq(1,
"+input1.getDim1() +")");
+
HopRewriteUtils.replaceChildReference(hi, input1, literal);
+ }
+ }
+
+ }
+ return hi;
+ }
}
diff --git a/src/main/java/org/apache/sysds/lops/Ctable.java
b/src/main/java/org/apache/sysds/lops/Ctable.java
index 3384119ed2..912519e41e 100644
--- a/src/main/java/org/apache/sysds/lops/Ctable.java
+++ b/src/main/java/org/apache/sysds/lops/Ctable.java
@@ -36,6 +36,7 @@ public class Ctable extends Lop
{
private final boolean _ignoreZeros;
private final boolean _outputEmptyBlocks;
+ private final int _numThreads;
public enum OperationTypes {
CTABLE_TRANSFORM,
@@ -58,15 +59,16 @@ public class Ctable extends Lop
OperationTypes operation;
- public Ctable(Lop[] inputLops, OperationTypes op, DataType dt,
ValueType vt, ExecType et) {
- this(inputLops, op, dt, vt, false, true, et);
+ public Ctable(Lop[] inputLops, OperationTypes op, DataType dt,
ValueType vt, ExecType et, int k) {
+ this(inputLops, op, dt, vt, false, true, et, k);
}
- public Ctable(Lop[] inputLops, OperationTypes op, DataType dt,
ValueType vt, boolean ignoreZeros, boolean outputEmptyBlocks, ExecType et) {
+ public Ctable(Lop[] inputLops, OperationTypes op, DataType dt,
ValueType vt, boolean ignoreZeros, boolean outputEmptyBlocks, ExecType et, int
k) {
super(Lop.Type.Ctable, dt, vt);
init(inputLops, op, et);
_ignoreZeros = ignoreZeros;
_outputEmptyBlocks = outputEmptyBlocks;
+ _numThreads = k;
}
private void init(Lop[] inputLops, OperationTypes op, ExecType et) {
@@ -175,6 +177,10 @@ public class Ctable extends Lop
sb.append( OPERAND_DELIMITOR );
sb.append( _outputEmptyBlocks );
}
+ else {
+ sb.append( OPERAND_DELIMITOR );
+ sb.append(_numThreads);
+ }
return sb.toString();
}
diff --git
a/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java
b/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java
index c12e4c4705..2fd1afd4a3 100644
--- a/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java
+++ b/src/main/java/org/apache/sysds/parser/BuiltinFunctionExpression.java
@@ -1987,8 +1987,8 @@ public class BuiltinFunctionExpression extends
DataIdentifier {
case DECOMPRESS:
if(OptimizerUtils.ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND){
checkNumParameters(1);
- checkMatrixParam(getFirstExpr());
- output.setDataType(DataType.MATRIX);
+ checkMatrixFrameParam(getFirstExpr());
+
output.setDataType(getFirstExpr().getOutput().getDataType());
output.setDimensions(id.getDim1(),
id.getDim2());
output.setBlocksize (id.getBlocksize());
output.setValueType(id.getValueType());
diff --git a/src/main/java/org/apache/sysds/resource/cost/CostEstimator.java
b/src/main/java/org/apache/sysds/resource/cost/CostEstimator.java
index bf6afbe9d2..09c7dadac8 100644
--- a/src/main/java/org/apache/sysds/resource/cost/CostEstimator.java
+++ b/src/main/java/org/apache/sysds/resource/cost/CostEstimator.java
@@ -822,7 +822,7 @@ public class CostEstimator
SparkCostUtils.getMatMulChainInstTime(mmchaininst, input1, input2, input3,
output, driverMetrics, executorMetrics);
} else if (inst instanceof CtableSPInstruction) {
CtableSPInstruction tableInst = (CtableSPInstruction)
inst;
- VarStats input1 = getStats(tableInst.input1.getName());
+ VarStats input1 =
getStatsWithDefaultScalar(tableInst.input1.getName());
VarStats input2 =
getStatsWithDefaultScalar(tableInst.input2.getName());
VarStats input3 =
getStatsWithDefaultScalar(tableInst.input3.getName());
double loadTime = loadRDDStatsAndEstimateTime(input1) +
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 e74e6c12f7..001e11dcd4 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java
@@ -53,6 +53,7 @@ import org.apache.sysds.runtime.compress.lib.CLALibDecompress;
import org.apache.sysds.runtime.compress.lib.CLALibMMChain;
import org.apache.sysds.runtime.compress.lib.CLALibMatrixMult;
import org.apache.sysds.runtime.compress.lib.CLALibMerge;
+import org.apache.sysds.runtime.compress.lib.CLALibReshape;
import org.apache.sysds.runtime.compress.lib.CLALibRexpand;
import org.apache.sysds.runtime.compress.lib.CLALibScalar;
import org.apache.sysds.runtime.compress.lib.CLALibSlice;
@@ -1281,6 +1282,11 @@ public class CompressedMatrixBlock extends MatrixBlock {
return getUncompressed().transpose(k);
}
+ @Override
+ public MatrixBlock reshape(int rows,int cols, boolean byRow){
+ return CLALibReshape.reshape(this, rows, cols, byRow);
+ }
+
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
diff --git
a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC.java
b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC.java
index 7763fef993..86ebb4400e 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/colgroup/ColGroupDDC.java
@@ -595,10 +595,8 @@ public class ColGroupDDC extends APreAgg implements
IMapToDataGroup {
public void leftMMIdentityPreAggregateDense(MatrixBlock that,
MatrixBlock ret, int rl, int ru, int cl, int cu) {
DenseBlock db = that.getDenseBlock();
DenseBlock retDB = ret.getDenseBlock();
- if(rl == ru - 1)
- leftMMIdentityPreAggregateDenseSingleRow(db.values(rl),
db.pos(rl), retDB.values(rl), retDB.pos(rl), cl, cu);
- else
- throw new NotImplementedException();
+ for(int i = rl; i < ru; i++)
+ leftMMIdentityPreAggregateDenseSingleRow(db.values(i),
db.pos(i), retDB.values(i), retDB.pos(i), cl, cu);
}
@Override
@@ -632,7 +630,8 @@ public class ColGroupDDC extends APreAgg implements
IMapToDataGroup {
}
}
- final void vectMM(double aa, double[] b, double[] c, int endT, int jd,
int crl, int cru, int offOut, int k, int vLen) {
+ final void vectMM(double aa, double[] b, double[] c, int endT, int jd,
int crl, int cru, int offOut, int k,
+ int vLen) {
// vVec = vVec.broadcast(aa);
final int offj = k * jd;
final int end = endT + offj;
@@ -919,16 +918,16 @@ public class ColGroupDDC extends APreAgg implements
IMapToDataGroup {
@Override
protected void denseSelection(MatrixBlock selection, P[] points,
MatrixBlock ret, int rl, int ru) {
- // morph(CompressionType.UNCOMPRESSED,
_data.size()).sparseSelection(selection, ret, rl, ru);;
- final SparseBlock sb = selection.getSparseBlock();
- final DenseBlock retB = ret.getDenseBlock();
- for(int r = rl; r < ru; r++) {
- if(sb.isEmpty(r))
- continue;
- final int sPos = sb.pos(r);
- final int rowCompressed = sb.indexes(r)[sPos];
// column index with 1
- decompressToDenseBlock(retB, rowCompressed,
rowCompressed + 1, r - rowCompressed, 0);
- }
+ // morph(CompressionType.UNCOMPRESSED,
_data.size()).sparseSelection(selection, ret, rl, ru);;
+ final SparseBlock sb = selection.getSparseBlock();
+ final DenseBlock retB = ret.getDenseBlock();
+ for(int r = rl; r < ru; r++) {
+ if(sb.isEmpty(r))
+ continue;
+ final int sPos = sb.pos(r);
+ final int rowCompressed = sb.indexes(r)[sPos]; //
column index with 1
+ decompressToDenseBlock(retB, rowCompressed,
rowCompressed + 1, r - rowCompressed, 0);
+ }
}
@@ -946,22 +945,21 @@ public class ColGroupDDC extends APreAgg implements
IMapToDataGroup {
for(int rc = cl; rc < cu; rc++, pos++) {
final int idx = _data.getIndex(rc);
if(idx != nVal)
- values2[_colIndexes.get(idx)]
+= values[pos];
+ values2[pos2 +
_colIndexes.get(idx)] += values[pos];
}
}
else {
for(int rc = cl; rc < cu; rc++, pos++)
-
values2[_colIndexes.get(_data.getIndex(rc))] += values[pos];
+ values2[pos2 +
_colIndexes.get(_data.getIndex(rc))] += values[pos];
}
}
}
-
private void
leftMMIdentityPreAggregateDenseSingleRowRangeIndex(double[] values, int pos,
double[] values2, int pos2,
int cl, int cu) {
IdentityDictionary a = (IdentityDictionary) _dict;
- final int firstCol = _colIndexes.get(0);
+ final int firstCol = pos2 + _colIndexes.get(0);
pos += cl; // left side matrix position offset.
if(a.withEmpty()) {
final int nVal =
_dict.getNumberOfValues(_colIndexes.size()) - 1;
diff --git
a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibReshape.java
b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibReshape.java
new file mode 100644
index 0000000000..f91779385d
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibReshape.java
@@ -0,0 +1,169 @@
+/*
+ * 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.compress.lib;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
+import org.apache.sysds.runtime.compress.DMLCompressionException;
+import org.apache.sysds.runtime.compress.colgroup.AColGroup;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.util.CommonThreadPool;
+import org.apache.sysds.utils.stats.InfrastructureAnalyzer;
+
+public class CLALibReshape {
+
+ protected static final Log LOG =
LogFactory.getLog(CLALibReshape.class.getName());
+
+ /** The minimum number of rows threshold for returning a compressed
output */
+ public static int COMPRESSED_RESHAPE_THRESHOLD = 1000;
+
+ final CompressedMatrixBlock in;
+
+ final int clen;
+ final int rlen;
+ final int rows;
+ final int cols;
+
+ final boolean rowwise;
+
+ final ExecutorService pool;
+
+ private CLALibReshape(CompressedMatrixBlock in, int rows, int cols,
boolean rowwise, int k) {
+ this.in = in;
+ this.rlen = in.getNumRows();
+ this.clen = in.getNumColumns();
+ this.rows = rows;
+ this.cols = cols;
+ this.rowwise = rowwise;
+ this.pool = k > 1 ? CommonThreadPool.get(k) : null;
+ }
+
+ public static MatrixBlock reshape(CompressedMatrixBlock in, int rows,
int cols, boolean rowwise) {
+ return new CLALibReshape(in, rows, cols, rowwise,
InfrastructureAnalyzer.getLocalParallelism()).apply();
+ }
+
+ public static MatrixBlock reshape(CompressedMatrixBlock in, int rows,
int cols, boolean rowwise, int k) {
+ return new CLALibReshape(in, rows, cols, rowwise, k).apply();
+ }
+
+ private MatrixBlock apply() {
+ try {
+ checkValidity();
+ if(shouldItBeCompressedOutputs())
+ return applyCompressed();
+ else
+ return in.decompress().reshape(rows, cols,
rowwise);
+ }
+ catch(Exception e) {
+ throw new DMLCompressionException("Failed reshaping of
compressed matrix", e);
+ }
+ finally {
+ if(pool != null)
+ pool.shutdown();
+ }
+ }
+
+ private MatrixBlock applyCompressed() throws Exception {
+ final int multiplier = rlen / rows;
+ final List<AColGroup> retGroups;
+ if(pool == null)
+ retGroups = applySingleThread(multiplier);
+ else if (in.getColGroups().size() == 1)
+ retGroups = applyParallelPushDown(multiplier);
+ else
+ retGroups = applyParallel(multiplier);
+
+ CompressedMatrixBlock ret = new CompressedMatrixBlock(rows,
cols);
+ ret.allocateColGroupList(retGroups);
+ ret.setNonZeros(in.getNonZeros());
+ return ret;
+ }
+
+ private List<AColGroup> applySingleThread(int multiplier) {
+ List<AColGroup> groups = in.getColGroups();
+ List<AColGroup> retGroups = new ArrayList<>(groups.size() *
multiplier);
+
+ for(AColGroup g : groups) {
+ final AColGroup[] tg = g.splitReshape(multiplier, rlen,
clen);
+ for(int i = 0; i < tg.length; i++)
+ retGroups.add(tg[i]);
+ }
+
+ return retGroups;
+
+ }
+
+
+ private List<AColGroup> applyParallelPushDown(int multiplier) throws
Exception {
+ List<AColGroup> groups = in.getColGroups();
+
+ List<AColGroup> retGroups = new ArrayList<>(groups.size() *
multiplier);
+ for(AColGroup g : groups){
+ final AColGroup[] tg =
g.splitReshapePushDown(multiplier, rlen, clen, pool);
+
+ for(int i = 0; i < tg.length; i++)
+ retGroups.add(tg[i]);
+ }
+
+ return retGroups;
+ }
+
+ private List<AColGroup> applyParallel(int multiplier) throws Exception {
+ List<AColGroup> groups = in.getColGroups();
+ List<Future<AColGroup[]>> tasks = new
ArrayList<>(groups.size());
+
+ for(AColGroup g : groups)
+ tasks.add(pool.submit(() -> g.splitReshape(multiplier,
rlen, clen)));
+
+ List<AColGroup> retGroups = new ArrayList<>(groups.size() *
multiplier);
+
+ for(Future<AColGroup[]> f : tasks) {
+ final AColGroup[] tg = f.get();
+ for(int i = 0; i < tg.length; i++)
+ retGroups.add(tg[i]);
+ }
+
+ return retGroups;
+ }
+
+ private void checkValidity() {
+
+ // check validity
+ if(((long) rlen) * clen != ((long) rows) * cols)
+ throw new DMLRuntimeException("Reshape matrix requires
consistent numbers of input/output cells (" + rlen + ":"
+ + clen + ", " + rows + ":" + cols + ").");
+
+ }
+
+ private boolean shouldItBeCompressedOutputs() {
+ // The number of rows in the reshaped allocations is fairly
large.
+ return rlen > COMPRESSED_RESHAPE_THRESHOLD && rowwise &&
+ // the reshape is a clean multiplier of number of rows,
meaning each column group cleanly reshape into x others
+ (double) rlen / rows % 1.0 == 0.0;
+ }
+
+}
diff --git
a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibRexpand.java
b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibRexpand.java
index 01b0ce14db..10c4ee1ab3 100644
--- a/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibRexpand.java
+++ b/src/main/java/org/apache/sysds/runtime/compress/lib/CLALibRexpand.java
@@ -19,14 +19,31 @@
package org.apache.sysds.runtime.compress.lib;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.conf.ConfigurationManager;
+import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
+import org.apache.sysds.runtime.compress.colgroup.AColGroup;
+import org.apache.sysds.runtime.compress.colgroup.ColGroupDDC;
+import org.apache.sysds.runtime.compress.colgroup.dictionary.ADictionary;
+import
org.apache.sysds.runtime.compress.colgroup.dictionary.IdentityDictionary;
+import org.apache.sysds.runtime.compress.colgroup.indexes.ColIndexFactory;
+import org.apache.sysds.runtime.compress.colgroup.indexes.IColIndex;
+import org.apache.sysds.runtime.compress.colgroup.mapping.AMapToData;
+import org.apache.sysds.runtime.compress.colgroup.mapping.MapToFactory;
import org.apache.sysds.runtime.matrix.data.LibMatrixReorg;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.util.CommonThreadPool;
import org.apache.sysds.runtime.util.UtilFunctions;
public final class CLALibRexpand {
+ public static boolean ALLOW_COMPRESSED_TABLE_SEQ = false;
protected static final Log LOG =
LogFactory.getLog(CLALibRexpand.class.getName());
private CLALibRexpand(){
@@ -42,6 +59,38 @@ public final class CLALibRexpand {
return rexpandCols(in, max, cast, ignore, k);
}
+ public static MatrixBlock rexpand(int seqHeight, MatrixBlock A) {
+ return rexpand(seqHeight, A, -1);
+ }
+
+ public static MatrixBlock rexpand(int seqHeight, MatrixBlock A, int
nColOut) {
+ return rexpand(seqHeight, A, nColOut, 1);
+ }
+
+ public static MatrixBlock rexpand(int seqHeight, MatrixBlock A, int
nColOut, int k) {
+
+ try {
+ final int[] map = new int[seqHeight];
+ int maxCol = constructInitialMapping(map, A, k);
+ boolean containsNull = maxCol < 0;
+ maxCol = Math.abs(maxCol);
+
+ if(nColOut == -1)
+ nColOut = maxCol;
+ else if(nColOut < maxCol)
+ throw new DMLRuntimeException("invalid nColOut,
requested: " + nColOut + " but have to be : " + maxCol);
+
+ final int nNulls = containsNull ? correctNulls(map,
nColOut) : 0;
+ if(nColOut == 0) // edge case of empty zero dimension
block.
+ return new MatrixBlock(seqHeight, 0, 0.0);
+ return createCompressedReturn(map, nColOut, seqHeight,
nNulls, containsNull, k);
+ }
+ catch(Exception e) {
+ throw new RuntimeException("Failed table seq operator",
e);
+ }
+ }
+
+
private static MatrixBlock rexpandCols(CompressedMatrixBlock in, double
max, boolean cast, boolean ignore, int k) {
return rexpandCols(in, UtilFunctions.toInt(max), cast, ignore,
k);
}
@@ -62,4 +111,104 @@ public final class CLALibRexpand {
return retC;
}
}
+
+
+
+ private static CompressedMatrixBlock createCompressedReturn(int[] map,
int nColOut, int seqHeight, int nNulls,
+ boolean containsNull, int k) throws Exception {
+ // create a single DDC Column group.
+ final IColIndex i = ColIndexFactory.create(0, nColOut);
+ final ADictionary d = new IdentityDictionary(nColOut,
containsNull);
+ final AMapToData m = MapToFactory.create(seqHeight, map,
nColOut + (containsNull ? 1 : 0), k);
+ final AColGroup g = ColGroupDDC.create(i, d, m, null);
+
+ final CompressedMatrixBlock cmb = new
CompressedMatrixBlock(seqHeight, nColOut);
+ cmb.allocateColGroup(g);
+ cmb.setNonZeros(seqHeight - nNulls);
+ return cmb;
+ }
+
+ private static int correctNulls(int[] map, int nColOut) {
+ int nNulls = 0;
+ for(int i = 0; i < map.length; i++) {
+ if(map[i] == -1) {
+ map[i] = nColOut;
+ nNulls++;
+ }
+ }
+ return nNulls;
+ }
+
+ private static int constructInitialMapping(int[] map, MatrixBlock A,
int k) {
+ if(A.isEmpty() || A.isInSparseFormat())
+ throw new DMLRuntimeException("not supported empty or
sparse construction of seq table");
+ final MatrixBlock Ac;
+ if(A instanceof CompressedMatrixBlock) {
+ // throw new NotImplementedException();
+ LOG.warn("Decompression of right side input to
CLALibTable, please implement alternative.");
+ Ac = ((CompressedMatrixBlock)
A).getUncompressed("rexpand", k);
+ }
+ else
+ Ac = A;
+
+ ExecutorService pool = CommonThreadPool.get(k);
+ try {
+
+ int blkz = Math.max((map.length / k), 1000);
+ List<Future<Integer>> tasks = new ArrayList<>();
+ for(int i = 0; i < map.length; i += blkz) {
+ final int start = i;
+ final int end = Math.min(i + blkz, map.length);
+ tasks.add(pool.submit(() -> partialMapping(map,
Ac, start, end)));
+ }
+
+ int maxCol = 0;
+ for(Future<Integer> f : tasks) {
+ int tmp = f.get();
+ if(Math.abs(tmp) > Math.abs(maxCol))
+ maxCol = tmp;
+ }
+ return maxCol;
+ }
+ catch(Exception e) {
+ throw new DMLRuntimeException(e);
+ }
+ finally {
+ pool.shutdown();
+ }
+
+ }
+
+ private static int partialMapping(int[] map, MatrixBlock A, int start,
int end) {
+
+ int maxCol = 0;
+ boolean containsNull = false;
+
+ final double[] aVals = A.getDenseBlockValues();
+
+ for(int i = start; i < end; i++) {
+ final double v2 = aVals[i];
+ if(Double.isNaN(v2)) {
+ map[i] = -1; // assign temporarily to -1
+ containsNull = true;
+ }
+ else {
+ // safe casts to long for consistent behavior
with indexing
+ int col = UtilFunctions.toInt(v2);
+ if(col <= 0)
+ throw new DMLRuntimeException(
+ "Erroneous input while
computing the contingency table (value <= zero): " + v2);
+
+ map[i] = col - 1;
+ // maintain max seen col
+ maxCol = Math.max(col, maxCol);
+ }
+ }
+
+ return containsNull ? maxCol * -1 : maxCol;
+ }
+
+ public static boolean compressedTableSeq() {
+ return ALLOW_COMPRESSED_TABLE_SEQ ||
ConfigurationManager.isCompressionEnabled();
+ }
}
diff --git
a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationMap.java
b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationMap.java
index a6ae6d5542..b0cccce171 100644
---
a/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationMap.java
+++
b/src/main/java/org/apache/sysds/runtime/controlprogram/federated/FederationMap.java
@@ -48,7 +48,6 @@ import org.apache.sysds.runtime.instructions.cp.CPOperand;
import org.apache.sysds.runtime.instructions.cp.ScalarObject;
import org.apache.sysds.runtime.instructions.cp.VariableCPInstruction;
import org.apache.sysds.runtime.lineage.LineageItem;
-import org.apache.sysds.runtime.util.CommonThreadPool;
import org.apache.sysds.runtime.util.IndexRange;
public class FederationMap {
diff --git
a/src/main/java/org/apache/sysds/runtime/instructions/cp/CtableCPInstruction.java
b/src/main/java/org/apache/sysds/runtime/instructions/cp/CtableCPInstruction.java
index 5530ca5aae..69b24ebc2b 100644
---
a/src/main/java/org/apache/sysds/runtime/instructions/cp/CtableCPInstruction.java
+++
b/src/main/java/org/apache/sysds/runtime/instructions/cp/CtableCPInstruction.java
@@ -19,10 +19,10 @@
package org.apache.sysds.runtime.instructions.cp;
-import org.apache.sysds.lops.Ctable;
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.lops.Ctable;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;
import org.apache.sysds.runtime.instructions.Instruction;
@@ -30,6 +30,7 @@ import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.lineage.LineageItemUtils;
import org.apache.sysds.runtime.matrix.data.CTableMap;
+import org.apache.sysds.runtime.matrix.data.LibMatrixReorg;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.util.DataConverter;
import org.apache.sysds.runtime.util.LongLongDoubleHashMap.EntryType;
@@ -39,21 +40,23 @@ public class CtableCPInstruction extends
ComputationCPInstruction {
private final CPOperand _outDim2;
private final boolean _isExpand;
private final boolean _ignoreZeros;
+ private final int _k;
private CtableCPInstruction(CPOperand in1, CPOperand in2, CPOperand
in3, CPOperand out,
String outputDim1, boolean dim1Literal, String
outputDim2, boolean dim2Literal, boolean isExpand,
- boolean ignoreZeros, String opcode, String istr) {
+ boolean ignoreZeros, String opcode, String istr, int k)
{
super(CPType.Ctable, null, in1, in2, in3, out, opcode, istr);
_outDim1 = new CPOperand(outputDim1, ValueType.FP64,
DataType.SCALAR, dim1Literal);
_outDim2 = new CPOperand(outputDim2, ValueType.FP64,
DataType.SCALAR, dim2Literal);
_isExpand = isExpand;
_ignoreZeros = ignoreZeros;
+ _k = k;
}
public static CtableCPInstruction parseInstruction(String inst)
{
String[] parts =
InstructionUtils.getInstructionPartsWithValueType(inst);
- InstructionUtils.checkNumFields ( parts, 7 );
+ InstructionUtils.checkNumFields ( parts, 8 );
String opcode = parts[0];
@@ -75,8 +78,10 @@ public class CtableCPInstruction extends
ComputationCPInstruction {
CPOperand out = new CPOperand(parts[6]);
boolean ignoreZeros = Boolean.parseBoolean(parts[7]);
+ int k = Integer.parseInt(parts[8]);
+
// ctable does not require any operator, so we simply pass-in a
dummy operator with null functionobject
- return new CtableCPInstruction(in1, in2, in3, out,
dim1Fields[0], Boolean.parseBoolean(dim1Fields[1]), dim2Fields[0],
Boolean.parseBoolean(dim2Fields[1]), isExpand, ignoreZeros, opcode, inst);
+ return new CtableCPInstruction(in1, in2, in3, out,
dim1Fields[0], Boolean.parseBoolean(dim1Fields[1]), dim2Fields[0],
Boolean.parseBoolean(dim2Fields[1]), isExpand, ignoreZeros, opcode, inst, k);
}
private Ctable.OperationTypes findCtableOperation() {
@@ -88,8 +93,8 @@ public class CtableCPInstruction extends
ComputationCPInstruction {
@Override
public void processInstruction(ExecutionContext ec) {
- MatrixBlock matBlock1 = ec.getMatrixInput(input1.getName());
- MatrixBlock matBlock2=null, wtBlock=null;
+ MatrixBlock matBlock1 =! _isExpand ? ec.getMatrixInput(input1):
null;
+ MatrixBlock matBlock2 = null, wtBlock=null;
double cst1, cst2;
CTableMap resultMap = new CTableMap(EntryType.INT);
@@ -110,10 +115,7 @@ public class CtableCPInstruction extends
ComputationCPInstruction {
if( !sparse )
resultBlock = new MatrixBlock((int)outputDim1,
(int)outputDim2, false);
}
- if( _isExpand ){
- resultBlock = new MatrixBlock( matBlock1.getNumRows(),
Integer.MAX_VALUE, true );
- }
-
+
switch(ctableOp) {
case CTABLE_TRANSFORM: //(VECTOR)
// F=ctable(A,B,W)
@@ -129,10 +131,13 @@ public class CtableCPInstruction extends
ComputationCPInstruction {
break;
case CTABLE_EXPAND_SCALAR_WEIGHT: //(VECTOR)
// F = ctable(seq,A) or F = ctable(seq,B,1)
+ // ignore first argument
+ if(input1.getDataType() == DataType.MATRIX){
+ LOG.warn("rewrite for table expand not
activated please fix");
+ }
matBlock2 = ec.getMatrixInput(input2.getName());
cst1 =
ec.getScalarInput(input3).getDoubleValue();
- // only resultBlock.rlen known,
resultBlock.clen set in operation
- matBlock1.ctableSeqOperations(matBlock2, cst1,
resultBlock);
+ resultBlock =
LibMatrixReorg.fusedSeqRexpand(matBlock2.getNumRows(), matBlock2, cst1,
resultBlock, true, _k);
break;
case CTABLE_TRANSFORM_HISTOGRAM: //(VECTOR)
// F=ctable(A,1) or F = ctable(A,1,1)
@@ -151,7 +156,7 @@ public class CtableCPInstruction extends
ComputationCPInstruction {
throw new DMLRuntimeException("Encountered an
invalid ctable operation ("+ctableOp+") while executing instruction: " +
this.toString());
}
- if(input1.getDataType() == DataType.MATRIX)
+ if(input1.getDataType() == DataType.MATRIX && ctableOp !=
Ctable.OperationTypes.CTABLE_EXPAND_SCALAR_WEIGHT )
ec.releaseMatrixInput(input1.getName());
if(input2.getDataType() == DataType.MATRIX)
ec.releaseMatrixInput(input2.getName());
diff --git
a/src/main/java/org/apache/sysds/runtime/instructions/cp/ReshapeCPInstruction.java
b/src/main/java/org/apache/sysds/runtime/instructions/cp/ReshapeCPInstruction.java
index 96fcc20a3f..caab05b603 100644
---
a/src/main/java/org/apache/sysds/runtime/instructions/cp/ReshapeCPInstruction.java
+++
b/src/main/java/org/apache/sysds/runtime/instructions/cp/ReshapeCPInstruction.java
@@ -29,7 +29,6 @@ import org.apache.sysds.runtime.data.TensorBlock;
import org.apache.sysds.runtime.instructions.InstructionUtils;
import org.apache.sysds.runtime.lineage.LineageItem;
import org.apache.sysds.runtime.lineage.LineageItemUtils;
-import org.apache.sysds.runtime.matrix.data.LibMatrixReorg;
import org.apache.sysds.runtime.matrix.data.MatrixBlock;
import org.apache.sysds.runtime.matrix.operators.Operator;
import org.apache.sysds.runtime.util.DataConverter;
@@ -97,11 +96,9 @@ public class ReshapeCPInstruction extends UnaryCPInstruction
{
int rows = (int)
ec.getScalarInput(_opRows).getLongValue(); //save cast
int cols = (int)
ec.getScalarInput(_opCols).getLongValue(); //save cast
BooleanObject byRow = (BooleanObject)
ec.getScalarInput(_opByRow.getName(), ValueType.BOOLEAN, _opByRow.isLiteral());
-
//execute operations
- MatrixBlock out = new MatrixBlock();
- LibMatrixReorg.reshape(in, out, rows, cols,
byRow.getBooleanValue(), -1);
-
+ MatrixBlock out = in.reshape(rows, cols,
byRow.getBooleanValue());
+
//set output and release inputs
ec.releaseMatrixInput(input1.getName());
ec.setMatrixOutput(output.getName(), out);
diff --git
a/src/main/java/org/apache/sysds/runtime/instructions/fed/CtableFEDInstruction.java
b/src/main/java/org/apache/sysds/runtime/instructions/fed/CtableFEDInstruction.java
index e953aa543a..e91af4f49c 100644
---
a/src/main/java/org/apache/sysds/runtime/instructions/fed/CtableFEDInstruction.java
+++
b/src/main/java/org/apache/sysds/runtime/instructions/fed/CtableFEDInstruction.java
@@ -66,10 +66,11 @@ public class CtableFEDInstruction extends
ComputationFEDInstruction {
}
public static CtableFEDInstruction parseInstruction(CtableCPInstruction
inst, ExecutionContext ec) {
- if((inst.getOpcode().equalsIgnoreCase("ctable") ||
inst.getOpcode().equalsIgnoreCase("ctableexpand")) &&
-
(ec.getCacheableData(inst.input1).isFederated(FType.ROW) ||
+ // TODO: add support for new tableexpand intruction.
+ if((inst.getOpcode().equalsIgnoreCase("ctable") ) &&
+ ((inst.input1.isMatrix() &&
ec.getCacheableData(inst.input1).isFederated(FType.ROW) ||
(inst.input2.isMatrix() &&
ec.getCacheableData(inst.input2).isFederated(FType.ROW)) ||
- (inst.input3.isMatrix() &&
ec.getCacheableData(inst.input3).isFederated(FType.ROW))))
+ (inst.input3.isMatrix() &&
ec.getCacheableData(inst.input3).isFederated(FType.ROW)))))
return CtableFEDInstruction.parseInstruction(inst);
return null;
}
diff --git
a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixReorg.java
b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixReorg.java
index 132154907c..486ec40694 100644
--- a/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixReorg.java
+++ b/src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixReorg.java
@@ -40,6 +40,7 @@ import org.apache.commons.logging.LogFactory;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
import org.apache.sysds.runtime.compress.DMLCompressionException;
+import org.apache.sysds.runtime.compress.lib.CLALibRexpand;
import org.apache.sysds.runtime.controlprogram.caching.MatrixObject.UpdateType;
import org.apache.sysds.runtime.data.DenseBlock;
import org.apache.sysds.runtime.data.DenseBlockFactory;
@@ -950,6 +951,185 @@ public class LibMatrixReorg {
return rexpandColumns(in, ret, max, cast, ignore, k);
}
+
+ /**
+ * The DML code to activate this function:
+ * <p>
+ *
+ * ret = table(seq(1, nrow(A)), A, w)
+ *
+ * @param seqHeight A sequence vector height.
+ * @param A The MatrixBlock vector to encode.
+ * @param w The weight matrix to multiply on output cells.
+ * @return A new MatrixBlock with the table result.
+ */
+ public static MatrixBlock fusedSeqRexpand(int seqHeight, MatrixBlock A,
double w) {
+ return fusedSeqRexpand(seqHeight, A, w, null, true, 1);
+ }
+
+ /**
+ * The DML code to activate this function:
+ * <p>
+ *
+ * ret = table(seq(1, nrow(A)), A, w)
+ *
+ * @param seqHeight A sequence vector height.
+ * @param A The MatrixBlock vector to encode.
+ * @param w The weight scalar to multiply on output cells.
+ * @param ret The output MatrixBlock, does not have to be used,
but depending on updateClen determine the
+ * output size.
+ * @param updateClen Update clen, if set to true, ignore dimensions of
ret, otherwise use the column dimension of
+ * ret.
+ * @return A new MatrixBlock or ret.
+ */
+ public static MatrixBlock fusedSeqRexpand(int seqHeight, MatrixBlock A,
double w, MatrixBlock ret,
+ boolean updateClen) {
+ return fusedSeqRexpand(seqHeight, A, w, ret, updateClen, 1);
+ }
+
+ /**
+ * The DML code to activate this function:
+ * <p>
+ *
+ * ret = table(seq(1, nrow(A)), A, w)
+ *
+ * @param seqHeight A sequence vector height.
+ * @param A The MatrixBlock vector to encode.
+ * @param w The weight matrix to multiply on output cells.
+ * @param ret The output MatrixBlock, does not have to be used,
but depending on updateClen determine the
+ * output size.
+ * @param updateClen Update clen, if set to true, ignore dimensions of
ret, otherwise use the column dimension of
+ * ret.
+ * @param k Parallelization degree
+ * @return A new MatrixBlock or ret.
+ */
+ public static MatrixBlock fusedSeqRexpand(int seqHeight, MatrixBlock A,
double w, MatrixBlock ret,
+ boolean updateClen, int k) {
+
+ if(A.getNumRows() != seqHeight)
+ throw new DMLRuntimeException(
+ "Invalid input sizes for table \"table(seq(1,
nrow(A)), A, w)\" : sequence height is: " + seqHeight
+ + " while A is: " + A.getNumRows());
+
+ if(A.getNumColumns() > 1)
+ throw new DMLRuntimeException(
+ "Invalid input A in table(seq(1, nrow(A)), A,
w): A should only have one column but has: "
+ + A.getNumColumns());
+
+ if(!Double.isNaN(w) && w != 0) {
+ if((CLALibRexpand.compressedTableSeq() || A instanceof
CompressedMatrixBlock) && w == 1)
+ return CLALibRexpand.rexpand(seqHeight, A,
updateClen ? -1 : ret.getNumColumns(), k);
+ else{
+ return fusedSeqRexpandSparse(seqHeight, A, w,
ret, updateClen);
+ }
+ }
+ else {
+ if(ret == null) {
+ ret = new MatrixBlock();
+ updateClen = true;
+ }
+
+ ret.rlen = seqHeight;
+ // empty output.
+ ret.denseBlock = null;
+ ret.sparseBlock = null;
+ ret.sparse = true;
+ ret.nonZeros = 0;
+ updateClenRexpand(ret, 0, updateClen);
+ return ret;
+ }
+
+ }
+
+ private static MatrixBlock fusedSeqRexpandSparse(int seqHeight,
MatrixBlock A, double w, MatrixBlock ret, boolean updateClen) {
+ if(ret == null) {
+ ret = new MatrixBlock();
+ updateClen = true;
+ }
+ final int rlen = seqHeight;
+ // prepare allocation of CSR sparse block
+ final int[] rowPointers = new int[rlen + 1];
+ final int[] indexes = new int[rlen];
+ final double[] values = new double[rlen];
+
+ ret.rlen = rlen;
+ // assign the output
+ ret.sparse = true;
+ ret.denseBlock = null;
+ // construct sparse CSR block from filled arrays
+ SparseBlockCSR csr = new SparseBlockCSR(rowPointers, indexes,
values, rlen);
+ ret.sparseBlock = csr;
+ int blkz = Math.min(1024, rlen);
+ int maxcol = 0;
+ boolean containsNull = false;
+ for(int i = 0; i < rlen; i += blkz) {
+ // blocked execution for earlier JIT compilation
+ int t = fusedSeqRexpandSparseBlock(csr, A, w, i,
Math.min(i + blkz, rlen));
+ if(t < 0) {
+ t = Math.abs(t);
+ containsNull = true;
+ }
+ maxcol = Math.max(t, maxcol);
+ }
+
+ if(containsNull)
+ csr.compact();
+
+ rowPointers[rlen] = rlen;
+ ret.setNonZeros(ret.sparseBlock.size());
+ if(updateClen)
+ ret.setNumColumns(maxcol);
+ return ret;
+ }
+
+ private static int fusedSeqRexpandSparseBlock(final SparseBlockCSR csr,
final MatrixBlock A, final double w, int rl, int ru) {
+
+ // prepare allocation of CSR sparse block
+ final int[] rowPointers = csr.rowPointers();
+ final int[] indexes = csr.indexes();
+ final double[] values = csr.values();
+
+ boolean containsNull = false;
+ int maxCol = 0;
+
+ for(int i = rl; i < ru; i++) {
+ int c = rexpandSingleRow(i, A.get(i, 0), w, indexes,
values);
+ if(c < 0)
+ containsNull = true;
+ else
+ maxCol = Math.max(c, maxCol);
+ rowPointers[i] = i;
+ }
+
+ return containsNull ? -maxCol: maxCol;
+ }
+
+ private static void updateClenRexpand(MatrixBlock ret, int maxCol,
boolean updateClen) {
+ // update meta data (initially unknown number of columns)
+ // Only allowed if we enable the update flag.
+ if(updateClen)
+ ret.clen = maxCol;
+ }
+
+ public static int rexpandSingleRow(int row, double v2, double w, int[]
retIx, double[] retVals) {
+ // If any of the values are NaN (i.e., missing) then
+ // we skip this tuple, proceed to the next tuple
+ if(Double.isNaN(v2))
+ return -1;
+
+ // safe casts to long for consistent behavior with indexing
+ int col = UtilFunctions.toInt(v2);
+ if(col <= 0)
+ throw new DMLRuntimeException("Erroneous input while
computing the contingency table (value <= zero): " + v2);
+
+ // set weight as value (expand is guaranteed to address
different cells)
+ retIx[row] = col - 1;
+ retVals[row] = w;
+
+ // maintain max seen col
+ return col;
+ }
+
/**
* Quick check if the input is valid for rexpand, this check does not
guarantee that the input is valid for rexpand
*
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 b5e4ae21d3..22fa5e43e7 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
@@ -5351,46 +5351,20 @@ public class MatrixBlock extends MatrixValue implements
CacheBlock<MatrixBlock>,
}
/**
+ * D = ctable(seq,A,w)
+ * <p>
+ * this = seq; thatMatrix = A; thatScalar = w; ret = D
+ *
* @param thatMatrix matrix value
* @param thatScalar scalar double
- * @param ret result matrix block
+ * @param ret result matrix block that is the weight to multiply
into the table output
* @param updateClen when this matrix already has the desired number of
columns updateClen can be set to false
* @return result matrix block
*/
- public MatrixBlock ctableSeqOperations(MatrixValue thatMatrix, double
thatScalar, MatrixBlock ret, boolean updateClen) {
+ public MatrixBlock ctableSeqOperations(MatrixValue thatMatrix, double
thatScalar, MatrixBlock ret,
+ boolean updateClen) {
MatrixBlock that = checkType(thatMatrix);
- CTable ctable = CTable.getCTableFnObject();
- double w = thatScalar;
-
- //prepare allocation of CSR sparse block
- int[] rptr = new int[rlen+1];
- int[] indexes = new int[rlen];
- double[] values = new double[rlen];
-
- //sparse-unsafe ctable execution
- //(because input values of 0 are invalid and have to result in
errors)
- //resultBlock guaranteed to be allocated for ctableexpand
- //each row in resultBlock will be allocated and will contain
exactly one value
- int maxCol = 0;
- for( int i=0; i<rlen; i++ ) {
- double v2 = that.get(i, 0);
- maxCol = ctable.execute(i+1, v2, w, maxCol, indexes,
values);
- rptr[i] = i;
- }
- rptr[rlen] = rlen;
-
- //construct sparse CSR block from filled arrays
- ret.sparseBlock = new SparseBlockCSR(rptr, indexes, values,
rlen);
- ((SparseBlockCSR)ret.sparseBlock).compact();
- ret.setNonZeros(ret.sparseBlock.size());
-
- //update meta data (initially unknown number of columns)
- //note: nnz maintained in ctable (via quickset)
- if(updateClen) {
- ret.clen = maxCol;
- }
-
- return ret;
+ return LibMatrixReorg.fusedSeqRexpand(this.getNumRows(), that,
thatScalar, ret, updateClen, 1);
}
/**
diff --git
a/src/test/java/org/apache/sysds/test/component/compress/lib/CLALibReshapeTests.java
b/src/test/java/org/apache/sysds/test/component/compress/lib/CLALibReshapeTests.java
new file mode 100644
index 0000000000..b886c480d8
--- /dev/null
+++
b/src/test/java/org/apache/sysds/test/component/compress/lib/CLALibReshapeTests.java
@@ -0,0 +1,113 @@
+/*
+ * 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.test.component.compress.lib;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlockFactory;
+import org.apache.sysds.runtime.compress.lib.CLALibReshape;
+import org.apache.sysds.runtime.matrix.data.LibMatrixReorg;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.test.TestUtils;
+import org.junit.Test;
+
+public class CLALibReshapeTests {
+ protected static final Log LOG =
LogFactory.getLog(CLALibReshapeTests.class.getName());
+
+ static{
+ Thread.currentThread().setName("test_reshape");
+ }
+
+ @Test
+ public void reshapeSimple() {
+ MatrixBlock mb = TestUtils.generateTestMatrixBlock(1000, 5, 1,
1, 0.5, 235);
+ MatrixBlock m2 =
CompressedMatrixBlockFactory.compress(mb).getLeft();
+
+ MatrixBlock m3 = CLALibReshape.reshape((CompressedMatrixBlock)
m2, 500, 10, false);
+ MatrixBlock ref = LibMatrixReorg.reshape(mb, 500, 10, false);
+
+ TestUtils.compareMatrices(ref, m3, 0);
+ }
+
+ @Test
+ public void reshapeSimple2Rowwise() {
+ MatrixBlock mb = TestUtils.generateTestMatrixBlock(3000, 1, 1,
1, 0.5, 235);
+ MatrixBlock m2 =
CompressedMatrixBlockFactory.compress(mb).getLeft();
+
+ MatrixBlock m3 = CLALibReshape.reshape((CompressedMatrixBlock)
m2, 1500, 2, true);
+ MatrixBlock ref = LibMatrixReorg.reshape(mb, 1500, 2, true);
+
+ TestUtils.compareMatrices(ref, m3, 0);
+ }
+
+ @Test
+ public void reshapeMulti2Rowwise() {
+ MatrixBlock mb = TestUtils.generateTestMatrixBlock(3000, 4, 1,
1, 0.5, 235);
+ MatrixBlock m2 =
CompressedMatrixBlockFactory.compress(mb).getLeft();
+
+ MatrixBlock m3 = CLALibReshape.reshape((CompressedMatrixBlock)
m2, 1500, 8, true);
+ MatrixBlock ref = LibMatrixReorg.reshape(mb, 1500, 8, true);
+
+ TestUtils.compareMatrices(ref, m3, 0);
+ }
+
+
+ @Test
+ public void reshapeMulti2RowwiseSingleThread() {
+ MatrixBlock mb = TestUtils.generateTestMatrixBlock(3000, 4, 1,
1, 0.5, 235);
+ MatrixBlock m2 =
CompressedMatrixBlockFactory.compress(mb).getLeft();
+
+ MatrixBlock m3 = CLALibReshape.reshape((CompressedMatrixBlock)
m2, 1500, 8, true, 1);
+ MatrixBlock ref = LibMatrixReorg.reshape(mb, 1500, 8, true);
+
+ TestUtils.compareMatrices(ref, m3, 0);
+ }
+
+ @Test
+ public void reshapeSimple2RowwiseNotMultiply() {
+ MatrixBlock mb = TestUtils.generateTestMatrixBlock(3000, 2, 1,
1, 0.5, 235);
+ MatrixBlock m2 =
CompressedMatrixBlockFactory.compress(mb).getLeft();
+
+ MatrixBlock m3 = CLALibReshape.reshape((CompressedMatrixBlock)
m2, 2000, 3, true);
+ MatrixBlock ref = LibMatrixReorg.reshape(mb, 2000, 3, true);
+
+ TestUtils.compareMatrices(ref, m3, 0);
+ }
+
+ @Test
+ public void reshapeSimple2ColWise() {
+ MatrixBlock mb = TestUtils.generateTestMatrixBlock(3000, 1, 1,
1, 0.5, 235);
+ MatrixBlock m2 =
CompressedMatrixBlockFactory.compress(mb).getLeft();
+
+ MatrixBlock m3 = CLALibReshape.reshape((CompressedMatrixBlock)
m2, 1500, 2, false);
+ MatrixBlock ref = LibMatrixReorg.reshape(mb, 1500, 2, false);
+
+ TestUtils.compareMatrices(ref, m3, 0);
+ }
+
+ @Test(expected = Exception.class)
+ public void reshapeInvalid() {
+ MatrixBlock mb = TestUtils.generateTestMatrixBlock(1000, 5, 1,
1, 0.5, 235);
+ MatrixBlock m2 =
CompressedMatrixBlockFactory.compress(mb).getLeft();
+
+ CLALibReshape.reshape((CompressedMatrixBlock) m2, 501, 10,
false);
+ }
+}
diff --git
a/src/test/java/org/apache/sysds/test/component/compress/lib/SeqTableTest.java
b/src/test/java/org/apache/sysds/test/component/compress/lib/SeqTableTest.java
new file mode 100644
index 0000000000..8f666aee48
--- /dev/null
+++
b/src/test/java/org/apache/sysds/test/component/compress/lib/SeqTableTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.test.component.compress.lib;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.runtime.compress.lib.CLALibRexpand;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.test.TestUtils;
+import org.junit.Test;
+
+public class SeqTableTest {
+
+ protected static final Log LOG =
LogFactory.getLog(SeqTableTest.class.getName());
+
+ static{
+ CLALibRexpand.ALLOW_COMPRESSED_TABLE_SEQ = true; // allow the
compressed tables.
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void test_notSameDim() throws Exception {
+ MatrixBlock c = new MatrixBlock(20, 1, 0.0);
+ CLALibRexpand.rexpand(10, c);
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void test_toLow() throws Exception {
+ MatrixBlock c = new MatrixBlock(10, 1, -1.0);
+ CLALibRexpand.rexpand(10, c);
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void test_toManyColumn() throws Exception {
+ MatrixBlock c = new MatrixBlock(10, 2, -1.0);
+ CLALibRexpand.rexpand(10, c);
+ }
+
+ @Test
+ public void test_All_NaN() throws Exception {
+ try{
+
+ MatrixBlock c = new MatrixBlock(10, 1, Double.NaN);
+ MatrixBlock ret = CLALibRexpand.rexpand(10, c);
+ assertEquals(0, ret.getNumColumns());
+ }
+ catch(Exception e){
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ @Test
+ public void test_One_NaN() throws Exception {
+ MatrixBlock c = new MatrixBlock(10, 1, 1.0);
+ c.set(3, 1, Double.NaN);
+ MatrixBlock ret = CLALibRexpand.rexpand(10, c);
+ assertEquals(1, ret.getNumColumns());
+ MatrixBlock expected = new MatrixBlock(10, 1, 1.0);
+ expected.set(3, 1, 0.0);
+ TestUtils.compareMatrices(expected, ret, 0.0);
+ }
+
+ @Test
+ public void test_all_one() throws Exception {
+ MatrixBlock c = new MatrixBlock(10, 1, 1.0);
+ MatrixBlock ret = CLALibRexpand.rexpand(10, c, 1);
+ assertEquals(1, ret.getNumColumns());
+ TestUtils.compareMatrices(c, ret, 0);
+ }
+
+}
diff --git
a/src/test/java/org/apache/sysds/test/component/matrix/SeqTableTest.java
b/src/test/java/org/apache/sysds/test/component/matrix/SeqTableTest.java
new file mode 100644
index 0000000000..049b602dd2
--- /dev/null
+++ b/src/test/java/org/apache/sysds/test/component/matrix/SeqTableTest.java
@@ -0,0 +1,106 @@
+/*
+ * 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.test.component.matrix;
+
+import static org.junit.Assert.assertEquals;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.runtime.compress.lib.CLALibRexpand;
+import org.apache.sysds.runtime.matrix.data.LibMatrixReorg;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.test.TestUtils;
+import org.junit.Test;
+
+public class SeqTableTest {
+
+ protected static final Log LOG =
LogFactory.getLog(SeqTableTest.class.getName());
+
+ static{
+ CLALibRexpand.ALLOW_COMPRESSED_TABLE_SEQ = false; // allow the
compressed tables.
+ }
+
+ @Test(expected = Exception.class)
+ public void test_notSameDim() {
+ MatrixBlock c = new MatrixBlock(20, 1, 0.0);
+ LibMatrixReorg.fusedSeqRexpand(10, c, 1);
+ }
+
+ @Test(expected = Exception.class)
+ public void test_toLow() {
+ MatrixBlock c = new MatrixBlock(10, 1, -1.0);
+ LibMatrixReorg.fusedSeqRexpand(10, c, 1);
+ }
+
+ @Test(expected = Exception.class)
+ public void test_toManyColumn() {
+ MatrixBlock c = new MatrixBlock(10, 2, -1.0);
+ LibMatrixReorg.fusedSeqRexpand(10, c, 1);
+ }
+
+ @Test
+ public void test_All_NaN() {
+ MatrixBlock c = new MatrixBlock(10, 1, Double.NaN);
+ MatrixBlock ret = LibMatrixReorg.fusedSeqRexpand(10, c, 1);
+
+ assertEquals(0, ret.getNumColumns());
+ }
+
+ @Test
+ public void test_w_NaN() {
+ MatrixBlock c = new MatrixBlock(10, 1, 1.0);
+ MatrixBlock ret = LibMatrixReorg.fusedSeqRexpand(10, c,
Double.NaN);
+ assertEquals(0, ret.getNumColumns());
+ }
+
+ @Test
+ public void test_all_one() {
+ MatrixBlock c = new MatrixBlock(10, 1, 1.0);
+ MatrixBlock ret = LibMatrixReorg.fusedSeqRexpand(10, c, 1);
+ assertEquals(1, ret.getNumColumns());
+ TestUtils.compareMatrices(c, ret, 0);
+ }
+
+ @Test
+ public void test_all_one_givenMatrixBlock() {
+ MatrixBlock c = new MatrixBlock(10, 1, 1.0);
+ MatrixBlock ret = LibMatrixReorg.fusedSeqRexpand(10, c, 1, new
MatrixBlock(), true);
+ assertEquals(1, ret.getNumColumns());
+ TestUtils.compareMatrices(c, ret, 0);
+ }
+
+ @Test
+ public void test_all_one_givenMatrixBlockWithSize() {
+ MatrixBlock c = new MatrixBlock(10, 1, 1.0);
+ MatrixBlock ret = LibMatrixReorg.fusedSeqRexpand(10, c, 1, new
MatrixBlock(1,2, 0.0), false);
+ assertEquals(2, ret.getNumColumns());
+ MatrixBlock expected = c.append(new MatrixBlock(10, 1, 0.0));
+ TestUtils.compareMatrices(expected, ret, 0);
+ }
+
+ @Test
+ public void test_all_one_givenMatrixBlockWithSize_NaNWeight() {
+ MatrixBlock c = new MatrixBlock(10, 1, 1.0);
+ MatrixBlock ret = LibMatrixReorg.fusedSeqRexpand(10, c,
Double.NaN, new MatrixBlock(1,2, 0.0), false);
+ assertEquals(2, ret.getNumColumns());
+ MatrixBlock expected = new MatrixBlock(10, 2, 0.0);
+ TestUtils.compareMatrices(expected, ret, 0);
+ }
+}
diff --git
a/src/test/java/org/apache/sysds/test/component/resource/RecompilationTest.java
b/src/test/java/org/apache/sysds/test/component/resource/RecompilationTest.java
index ef3af0eb40..e30c373be7 100644
---
a/src/test/java/org/apache/sysds/test/component/resource/RecompilationTest.java
+++
b/src/test/java/org/apache/sysds/test/component/resource/RecompilationTest.java
@@ -235,7 +235,7 @@ public class RecompilationTest extends AutomatedTestBase {
// original compilation used for comparison
Program expectedProgram =
ResourceCompiler.compile(HOME+"mm_test.dml", nvargs);
- Program recompiledProgram = runTest(precompiledProgram,
expectedProgram, driverMemory, numberExecutors, executorMemory);
+ Program recompiledProgram = runTest(precompiledProgram,
expectedProgram, driverMemory, numberExecutors, executorMemory, new
StringBuilder());
Optional<Instruction> mmInstruction = ((BasicProgramBlock)
recompiledProgram.getProgramBlocks().get(0)).getInstructions().stream()
.filter(inst ->
(Objects.equals(expectedSparkExecType, inst instanceof SPInstruction) &&
Objects.equals(inst.getOpcode(), expectedOpcode)))
@@ -257,7 +257,7 @@ public class RecompilationTest extends AutomatedTestBase {
}
// original compilation used for comparison
Program expectedProgram =
ResourceCompiler.compile(HOME+"mm_transpose_test.dml", nvargs);
- Program recompiledProgram = runTest(precompiledProgram,
expectedProgram, driverMemory, numberExecutors, executorMemory);
+ Program recompiledProgram = runTest(precompiledProgram,
expectedProgram, driverMemory, numberExecutors, executorMemory, new
StringBuilder());
Optional<Instruction> mmInstruction = ((BasicProgramBlock)
recompiledProgram.getProgramBlocks().get(0)).getInstructions().stream()
.filter(inst ->
(Objects.equals(expectedSparkExecType, inst instanceof SPInstruction) &&
Objects.equals(inst.getOpcode(), expectedOpcode)))
.findFirst();
@@ -273,8 +273,9 @@ public class RecompilationTest extends AutomatedTestBase {
Map<String,
String> nvargs) throws IOException {
// pre-compiled program using default values to be used as
source for the recompilation
Program precompiledProgram =
generateInitialProgram(HOME+dmlScript, nvargs);
- System.out.println("precompiled");
- System.out.println(Explain.explain(precompiledProgram));
+ StringBuilder sb = new StringBuilder();
+ sb.append("\n\nprecompiled\n");
+ sb.append(Explain.explain(precompiledProgram));
if (numberExecutors > 0) {
ResourceCompiler.setSparkClusterResourceConfigs(driverMemory, driverThreads,
numberExecutors, executorMemory, executorThreads);
} else {
@@ -282,13 +283,13 @@ public class RecompilationTest extends AutomatedTestBase {
}
// original compilation used for comparison
Program expectedProgram =
ResourceCompiler.compile(HOME+dmlScript, nvargs);
- System.out.println("expected");
- System.out.println(Explain.explain(expectedProgram));
- runTest(precompiledProgram, expectedProgram, driverMemory,
numberExecutors, executorMemory);
+ sb.append("\n\nexpected\n");
+ sb.append(Explain.explain(expectedProgram));
+ runTest(precompiledProgram, expectedProgram, driverMemory,
numberExecutors, executorMemory, sb);
}
- private Program runTest(Program precompiledProgram, Program
expectedProgram, long driverMemory, int numberExecutors, long executorMemory) {
- if (DEBUG_MODE)
System.out.println(Explain.explain(expectedProgram));
+ private Program runTest(Program precompiledProgram, Program
expectedProgram, long driverMemory, int numberExecutors, long executorMemory,
StringBuilder sb) {
+ if (DEBUG_MODE) sb.append(Explain.explain(expectedProgram));
Program recompiledProgram;
if (numberExecutors == 0) {
ResourceCompiler.setSingleNodeResourceConfigs(driverMemory, driverThreads);
@@ -303,19 +304,19 @@ public class RecompilationTest extends AutomatedTestBase {
);
recompiledProgram =
ResourceCompiler.doFullRecompilation(precompiledProgram);
}
- System.out.println("recompiled");
- System.out.println(Explain.explain(recompiledProgram));
+ sb.append("\n\nrecompiled\n");
+ sb.append(Explain.explain(recompiledProgram));
- if (DEBUG_MODE)
System.out.println(Explain.explain(recompiledProgram));
- assertEqualPrograms(expectedProgram, recompiledProgram);
+ if (DEBUG_MODE) sb.append(Explain.explain(recompiledProgram));
+ assertEqualPrograms(expectedProgram, recompiledProgram, sb);
return recompiledProgram;
}
- private void assertEqualPrograms(Program expected, Program actual) {
+ private void assertEqualPrograms(Program expected, Program actual,
StringBuilder sb) {
// strip empty blocks basic program blocks
String expectedProgramExplained =
stripGeneralAndReplaceRandoms(Explain.explain(expected));
String actualProgramExplained =
stripGeneralAndReplaceRandoms(Explain.explain(actual));
- Assert.assertEquals(expectedProgramExplained,
actualProgramExplained);
+ Assert.assertEquals(sb.toString(), expectedProgramExplained,
actualProgramExplained);
}
private String stripGeneralAndReplaceRandoms(String explainedProgram) {
diff --git
a/src/test/java/org/apache/sysds/test/functions/compress/wordembedding/WordEmbeddingUseCase.java
b/src/test/java/org/apache/sysds/test/functions/compress/wordembedding/WordEmbeddingUseCase.java
new file mode 100644
index 0000000000..0e66cbeeaf
--- /dev/null
+++
b/src/test/java/org/apache/sysds/test/functions/compress/wordembedding/WordEmbeddingUseCase.java
@@ -0,0 +1,179 @@
+/*
+ * 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.test.functions.compress.wordembedding;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.sysds.common.Types;
+import org.apache.sysds.common.Types.ExecType;
+import org.apache.sysds.hops.OptimizerUtils;
+import org.apache.sysds.runtime.compress.CompressedMatrixBlock;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.util.UtilFunctions;
+import org.apache.sysds.test.AutomatedTestBase;
+import org.apache.sysds.test.TestConfiguration;
+import org.apache.sysds.test.TestUtils;
+import org.junit.Test;
+
+public class WordEmbeddingUseCase extends AutomatedTestBase {
+
+ protected static final Log LOG =
LogFactory.getLog(WordEmbeddingUseCase.class.getName());
+
+ private final static String TEST_DIR =
"functions/compress/wordembedding/";
+
+ protected String getTestClassDir() {
+ return getTestDir();
+ }
+
+ protected String getTestName() {
+ return "wordembedding";
+ }
+
+ protected String getTestDir() {
+ return TEST_DIR;
+ }
+
+ @Test
+ public void testWordEmb() {
+ wordEmb(10, 2, 2, 2, ExecType.CP, "01");
+ }
+
+ @Test
+ public void testWordEmb_medium() {
+ wordEmb(100, 30, 4, 3, ExecType.CP, "01");
+ }
+
+ @Test
+ public void testWordEmb_bigWords() {
+ wordEmb(10, 2, 2, 10, ExecType.CP, "01");
+ }
+
+ @Test
+ public void testWordEmb_longSentences() {
+ wordEmb(100, 30, 5, 2, ExecType.CP, "01");
+ }
+
+ @Test
+ public void testWordEmb_moreUniqueWordsThanSentences() {
+ wordEmb(100, 200, 5, 2, ExecType.CP, "01");
+ }
+
+ @Test
+ public void testWordEmbSP() {
+ wordEmb(10, 2, 2, 2, ExecType.SPARK, "01");
+ }
+
+ @Test
+ public void testWordEmb_mediumSP() {
+ wordEmb(100, 30, 4, 3, ExecType.SPARK, "01");
+ }
+
+ @Test
+ public void testWordEmb_bigWordsSP() {
+ wordEmb(10, 2, 2, 10, ExecType.SPARK, "01");
+ }
+
+ @Test
+ public void testWordEmb_longSentencesSP() {
+ wordEmb(100, 30, 5, 2, ExecType.SPARK, "01");
+ }
+
+ @Test
+ public void testWordEmb_moreUniqueWordsThanSentencesSP() {
+ wordEmb(100, 200, 5, 2, ExecType.SPARK, "01");
+ }
+
+ public void wordEmb(int rows, int unique, int l, int embeddingSize,
ExecType instType, String name) {
+
+ OptimizerUtils.ALLOW_SCRIPT_LEVEL_COMPRESS_COMMAND = true;
+ Types.ExecMode platformOld = setExecMode(instType);
+
+ CompressedMatrixBlock.debug = true;
+
+ try {
+ super.setOutputBuffering(true);
+
loadTestConfiguration(getTestConfiguration(getTestName()));
+ fullDMLScriptName = SCRIPT_DIR + getTestClassDir() +
name + ".dml";
+
+ programArgs = new String[] {"-stats", "100",
"-explain", "-args", input("X"), input("W"), "" + l, output("R")};
+
+ MatrixBlock X = TestUtils.generateTestMatrixBlock(rows,
1, 1, unique + 1, 1.0, 32);
+ X = TestUtils.floor(X);
+ writeBinaryWithMTD("X", X);
+
+ MatrixBlock W =
TestUtils.generateTestMatrixBlock(unique, embeddingSize, 1.0, -1, 1, 32);
+ writeBinaryWithMTD("W", W);
+
+ String r = runTest(null).toString();
+
+ MatrixBlock R = TestUtils.readBinary(output("R"));
+
+ analyzeResult(X, W, R, l);
+
+ if( instType == ExecType.CP &&
heavyHittersContainsString("seq")){
+ fail("cp should not have seq instruction\n" +
r);
+ }
+
+ }
+ catch(Exception e) {
+ e.printStackTrace();
+ assertTrue("Exception in execution: " + e.getMessage(),
false);
+ }
+ finally {
+ rtplatform = platformOld;
+ }
+ }
+
+ private void analyzeResult(MatrixBlock X, MatrixBlock W, MatrixBlock R,
int l) {
+ assertEquals(R.getNumRows() ,X.getNumRows() / l);
+ // assertEquals(W.getNumColumns() , X.getNumColumns() * l);
+
+ for(int i = 0; i < X.getNumRows(); i++) {
+ // for each row in X, it should embed with a W, in
accordance to what value it used
+
+ // the entry to look into W. // as in row
+ int e = UtilFunctions.toInt(X.get(i, 0)) - 1;
+ int rowR = i / l;
+ int offR = i % l;
+
+ for(int j = 0; j < W.getNumColumns(); j++) {
+ assertEquals("i:"+i+" j:" + j,R.get(rowR, offR
* W.getNumColumns() + j), W.get(e, j), 0.0);
+ }
+ }
+ }
+
+ @Override
+ public void setUp() {
+ TestUtils.clearAssertionInformation();
+ addTestConfiguration(getTestName(), new
TestConfiguration(getTestClassDir(), getTestName()));
+ }
+
+ @Override
+ protected File getConfigTemplateFile() {
+ return new
File("./src/test/scripts/functions/compress/SystemDS-config-compress.xml");
+ }
+
+}
diff --git
a/src/test/java/org/apache/sysds/test/functions/federated/primitives/part5/FederatedCtableTest.java
b/src/test/java/org/apache/sysds/test/functions/federated/primitives/part5/FederatedCtableTest.java
index d96f433c0e..4ba19fb3d0 100644
---
a/src/test/java/org/apache/sysds/test/functions/federated/primitives/part5/FederatedCtableTest.java
+++
b/src/test/java/org/apache/sysds/test/functions/federated/primitives/part5/FederatedCtableTest.java
@@ -238,7 +238,11 @@ public class FederatedCtableTest extends AutomatedTestBase
{
compareResults(TOLERANCE);
// check for federated operations
- Assert.assertTrue(heavyHittersContainsString("fed_ctable") ||
heavyHittersContainsString("fed_ctableexpand"));
+ // TODO: add support for ctableexpand back when rewrite change
first parameter to string seq
+ if(heavyHittersContainsString("ctableexpand"))
+ return;
+
+ Assert.assertTrue(heavyHittersContainsString("fed_ctable") ||
heavyHittersContainsString("ctableexpand"));
if(fedOutput) { // verify output is federated
Assert.assertTrue(heavyHittersContainsString("fed_uak+"));
Assert.assertTrue(heavyHittersContainsString("fed_*"));
diff --git
a/src/test/scripts/functions/compress/table/CompressedTableOverwriteTest/01.dml
b/src/test/scripts/functions/compress/table/CompressedTableOverwriteTest/01.dml
new file mode 100644
index 0000000000..0dc9cca559
--- /dev/null
+++
b/src/test/scripts/functions/compress/table/CompressedTableOverwriteTest/01.dml
@@ -0,0 +1,53 @@
+#-------------------------------------------------------------
+#
+# 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.
+#
+#-------------------------------------------------------------
+
+print("Start Test")
+
+X = rand(rows=$rows,cols=1, min=0, max=$unique, sparsity=$sparsity)
+X = floor(X)
+X = X + 1
+
+for(i in 1:$unique){ # ensure all unique values are used.
+ X[i,1] = i
+}
+
+# transform encode path to table command
+F = as.frame(X)
+spec = "{ids:true, dummycode:[1]}"
+[Xt, M] = transformencode(target=F, spec=spec)
+
+
+Xa = table(seq(1, nrow(X)), X)
+
+X_diff = Xt - Xa
+s = max(X_diff) + min(X_diff)
+print(s)
+if(s != 0){
+ # print(toString(t(Xt),sparse=TRUE))
+ # print(toString(t(Xa), sparse=TRUE))
+ # print(toString(X_diff, sparse=TRUE))
+ print(toString(X_diff))
+ print(toString(Xt))
+ print(toString(Xa))
+ print("Failed, the output did not contain the same values after table")
+}
+else
+ print("Success, the output contained the same values after table")
\ No newline at end of file
diff --git a/src/test/scripts/functions/compress/wordembedding/01.dml
b/src/test/scripts/functions/compress/wordembedding/01.dml
new file mode 100644
index 0000000000..2650ae1636
--- /dev/null
+++ b/src/test/scripts/functions/compress/wordembedding/01.dml
@@ -0,0 +1,36 @@
+#-------------------------------------------------------------
+#
+# 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.
+#
+#-------------------------------------------------------------
+
+
+X = read($1)
+W = read($2)
+l = $3
+R_path = $4
+
+Xa = table(seq(1,nrow(X)), X)
+
+Xe = Xa %*% W
+
+R = matrix(Xe, rows = nrow(X) / l, cols = ncol(W) * l )
+
+write(R, R_path)
+
+print("Done")
