[systemds] branch main updated: [SYSTEMDS-3420] Fix update-in-place handling in spark remove-empty

[mboehm7]

This is an automated email from the ASF dual-hosted git repository.

mboehm7 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 ba250e456c [SYSTEMDS-3420] Fix update-in-place handling in spark 
remove-empty
ba250e456c is described below

commit ba250e456c15b58f57a08cb1beb7bfc1bb830e13
Author: Matthias Boehm <mboe...@gmail.com>
AuthorDate: Sat Aug 13 20:40:23 2022 +0200

    [SYSTEMDS-3420] Fix update-in-place handling in spark remove-empty
    
    Our distributed remove empty operations, perform aggregation cascades
    until the aggregates fit in memory, perform a local cumsum that exploits
    update in-place on the temporary intermediates, and then distribute the
    compute output offsets. In special cases, no aggregation cascades are
    compiled for the cumulative aggregate, leading to invalid update
    in-place. See the example below:
    
    --CP ucumk+ _mVar3602.MATRIX.FP64 _mVar3603.MATRIX.FP64 32 true
    --CP * _mVar3603.MATRIX.FP64 _mVar3602.MATRIX.FP64 _mVar3604.MATRIX.FP64
    32
---
 src/main/java/org/apache/sysds/hops/UnaryOp.java                     | 3 ++-
 .../runtime/instructions/cp/BinaryMatrixMatrixCPInstruction.java     | 5 +----
 2 files changed, 3 insertions(+), 5 deletions(-)

diff --git a/src/main/java/org/apache/sysds/hops/UnaryOp.java 
b/src/main/java/org/apache/sysds/hops/UnaryOp.java
index 430c181552..38db8a50b6 100644
--- a/src/main/java/org/apache/sysds/hops/UnaryOp.java
+++ b/src/main/java/org/apache/sysds/hops/UnaryOp.java
@@ -295,9 +295,10 @@ public class UnaryOp extends MultiThreadedHop
                }
                
                //in-memory cum sum (of partial aggregates)
+               //marked for update in-place if there was at least one 
aggregation level
                if( TEMP.getOutputParameters().getNumRows()!=1 ){
                        int k = OptimizerUtils.getConstrainedNumThreads( 
_maxNumThreads );
-                       Unary unary1 = new Unary( TEMP, _op, DataType.MATRIX, 
ValueType.FP64, ExecType.CP, k, true);
+                       Unary unary1 = new Unary( TEMP, _op, DataType.MATRIX, 
ValueType.FP64, ExecType.CP, k, TEMP!=X);
                        
unary1.getOutputParameters().setDimensions(TEMP.getOutputParameters().getNumRows(),
 clen, blen, -1);
                        setLineNumbers(unary1);
                        TEMP = unary1;
diff --git 
a/src/main/java/org/apache/sysds/runtime/instructions/cp/BinaryMatrixMatrixCPInstruction.java
 
b/src/main/java/org/apache/sysds/runtime/instructions/cp/BinaryMatrixMatrixCPInstruction.java
index c4e9ecd3aa..565210b585 100644
--- 
a/src/main/java/org/apache/sysds/runtime/instructions/cp/BinaryMatrixMatrixCPInstruction.java
+++ 
b/src/main/java/org/apache/sysds/runtime/instructions/cp/BinaryMatrixMatrixCPInstruction.java
@@ -54,10 +54,7 @@ public class BinaryMatrixMatrixCPInstruction extends 
BinaryCPInstruction {
                else {
                        // Perform computation using input matrices, and 
produce the result matrix
                        BinaryOperator bop = (BinaryOperator) _optr;
-
-                       if(compressedLeft && compressedRight)
-                               retBlock = inBlock1.binaryOperations(bop, 
inBlock2, new MatrixBlock());
-                       else if(compressedRight)
+                       if(!compressedLeft && compressedRight)
                                retBlock = ((CompressedMatrixBlock) 
inBlock2).binaryOperationsLeft(bop, inBlock1, new MatrixBlock());
                        else
                                retBlock = inBlock1.binaryOperations(bop, 
inBlock2, new MatrixBlock());