[WIP] JAVA 17 BWARE COMMIT

src/main/java/org/apache/sysds/hops/AggBinaryOp.java

@@ -439,8 +439,7 @@ private boolean isApplicableForTransitiveSparkExecType(boolean left)
 				|| (left && !isLeftTransposeRewriteApplicable(true)))
 			&& getInput(index).getParent().size()==1 //bagg is only parent
 			&& !getInput(index).areDimsBelowThreshold() 
-			&& (getInput(index).optFindExecType() == ExecType.SPARK
-				|| (getInput(index) instanceof DataOp && ((DataOp)getInput(index)).hasOnlyRDD()))
+			&& getInput(index).hasSparkOutput()
 			&& getInput(index).getOutputMemEstimate()>getOutputMemEstimate();
 	}
 

src/main/java/org/apache/sysds/hops/BinaryOp.java

@@ -747,8 +747,8 @@ protected ExecType optFindExecType(boolean transitive) {
 
 		checkAndSetForcedPlatform();
 
-		DataType dt1 = getInput().get(0).getDataType();
-		DataType dt2 = getInput().get(1).getDataType();
+		final DataType dt1 = getInput(0).getDataType();
+		final DataType dt2 = getInput(1).getDataType();
 
 		if( _etypeForced != null ) {
 			setExecType(_etypeForced);
@@ -796,18 +796,28 @@ else if ( dt1 == DataType.SCALAR && dt2 == DataType.MATRIX ) {
 			checkAndSetInvalidCPDimsAndSize();
 		}
 
-		//spark-specific decision refinement (execute unary scalar w/ spark input and
+		// spark-specific decision refinement (execute unary scalar w/ spark input and
 		// single parent also in spark because it's likely cheap and reduces intermediates)
-		if(transitive && _etype == ExecType.CP && _etypeForced != ExecType.CP && _etypeForced != ExecType.FED &&
-			getDataType().isMatrix() // output should be a matrix
-			&& (dt1.isScalar() || dt2.isScalar()) // one side should be scalar
-			&& supportsMatrixScalarOperations() // scalar operations
-			&& !(getInput().get(dt1.isScalar() ? 1 : 0) instanceof DataOp) // input is not checkpoint
-			&& getInput().get(dt1.isScalar() ? 1 : 0).getParent().size() == 1 // unary scalar is only parent
-			&& !HopRewriteUtils.isSingleBlock(getInput().get(dt1.isScalar() ? 1 : 0)) // single block triggered exec
-			&& getInput().get(dt1.isScalar() ? 1 : 0).optFindExecType() == ExecType.SPARK) {
-			// pull unary scalar operation into spark
-			_etype = ExecType.SPARK;
+		if(transitive // we allow transitive Spark operations. continue sequences of spark operations
+			&& _etype == ExecType.CP // The instruction is currently in CP
+			&& _etypeForced != ExecType.CP // not forced CP
+			&& _etypeForced != ExecType.FED // not federated
+			&& (getDataType().isMatrix() || getDataType().isFrame()) // output should be a matrix or frame
+		) {
+			final boolean v1 = getInput(0).isScalarOrVectorBellowBlockSize();
+			final boolean v2 = getInput(1).isScalarOrVectorBellowBlockSize();
+			final boolean left = v1 == true; // left side is the vector or scalar
+			final Hop sparkIn = getInput(left ? 1 : 0);
+			if((v1 ^ v2) // XOR only one side is allowed to be a vector or a scalar.
+				&& (supportsMatrixScalarOperations() || op == OpOp2.APPLY_SCHEMA)  // supported operation
+				&& sparkIn.getParent().size() == 1 // only one parent
+				&& !HopRewriteUtils.isSingleBlock(sparkIn) // single block triggered exec
+				&& sparkIn.optFindExecType() == ExecType.SPARK // input was spark op.
+				&& !(sparkIn instanceof DataOp) // input is not checkpoint
+			) {
+				// pull operation into spark
+				_etype = ExecType.SPARK;
+			}
 		}
 
 		if( OptimizerUtils.ALLOW_BINARY_UPDATE_IN_PLACE &&
@@ -837,7 +847,7 @@ else if( (op == OpOp2.CBIND && getDataType().isList())
 				|| (op == OpOp2.RBIND && getDataType().isList())) {
 			_etype = ExecType.CP;
 		}
-		
+
 		//mark for recompile (forever)
 		setRequiresRecompileIfNecessary();
 

src/main/java/org/apache/sysds/hops/Hop.java

@@ -1040,6 +1040,12 @@ public final String toString() {
 	// ========================================================================================
 
 
+	protected boolean isScalarOrVectorBellowBlockSize(){
+		return getDataType().isScalar() || (dimsKnown() && 
+			(( _dc.getRows() == 1 && _dc.getCols() < ConfigurationManager.getBlocksize())
+			|| _dc.getCols() == 1 && _dc.getRows() < ConfigurationManager.getBlocksize()));
+	}
+
 	protected boolean isVector() {
 		return (dimsKnown() && (_dc.getRows() == 1 || _dc.getCols() == 1) );
 	}
@@ -1624,6 +1630,11 @@ protected void setMemoryAndComputeEstimates(Lop lop) {
 		lop.setComputeEstimate(ComputeCost.getHOPComputeCost(this));
 	}
 
+	protected boolean hasSparkOutput(){
+		return (this.optFindExecType() == ExecType.SPARK
+			|| (this instanceof DataOp && ((DataOp)this).hasOnlyRDD()));
+	}
+
 	/**
 	 * Set parse information.
 	 *

src/main/java/org/apache/sysds/hops/UnaryOp.java

@@ -366,7 +366,11 @@ protected double computeOutputMemEstimate( long dim1, long dim2, long nnz )
 		} else {
 			sparsity = OptimizerUtils.getSparsity(dim1, dim2, nnz);
 		}
-		return OptimizerUtils.estimateSizeExactSparsity(dim1, dim2, sparsity);
+
+		if(getDataType() == DataType.FRAME)
+			return OptimizerUtils.estimateSizeExactFrame(dim1, dim2);
+		else
+			return OptimizerUtils.estimateSizeExactSparsity(dim1, dim2, sparsity);
 	}
 
 	@Override
@@ -463,6 +467,13 @@ public boolean isMetadataOperation() {
 			|| _op == OpOp1.CAST_AS_LIST;
 	}
 
+	private boolean isDisallowedSparkOps(){
+		return isCumulativeUnaryOperation() 
+			|| isCastUnaryOperation()
+			|| _op==OpOp1.MEDIAN
+			|| _op==OpOp1.IQM;
+	}
+
 	@Override
 	protected ExecType optFindExecType(boolean transitive) 
 	{
@@ -493,19 +504,22 @@ else if ( getInput().get(0).areDimsBelowThreshold() || getInput().get(0).isVecto
 			checkAndSetInvalidCPDimsAndSize();
 		}
 
+
 		//spark-specific decision refinement (execute unary w/ spark input and 
 		//single parent also in spark because it's likely cheap and reduces intermediates)
-		if( _etype == ExecType.CP && _etypeForced != ExecType.CP
-			&& getInput().get(0).optFindExecType() == ExecType.SPARK 
-			&& getDataType().isMatrix() 
-			&& !isCumulativeUnaryOperation() && !isCastUnaryOperation()
-			&& _op!=OpOp1.MEDIAN && _op!=OpOp1.IQM
-			&& !(getInput().get(0) instanceof DataOp)    //input is not checkpoint
-			&& getInput().get(0).getParent().size()==1 ) //unary is only parent
-		{
+		if(_etype == ExecType.CP // currently CP instruction
+			&& _etype != ExecType.SPARK /// currently not SP.
+			&& _etypeForced != ExecType.CP // not forced as CP instruction
+			&& getInput(0).hasSparkOutput() // input is a spark instruction
+			&& (getDataType().isMatrix() || getDataType().isFrame()) // output is a matrix or frame
+			&& !isDisallowedSparkOps() // is invalid spark instruction
+			// && !(getInput().get(0) instanceof DataOp) // input is not checkpoint
+			// && getInput(0).getParent().size() <= 1// unary is only parent
+		) {
 			//pull unary operation into spark 
 			_etype = ExecType.SPARK;
 		}
+
 
 		//mark for recompile (forever)
 		setRequiresRecompileIfNecessary();
@@ -520,7 +534,7 @@ && getInput().get(0).getParent().size()==1 ) //unary is only parent
 		} else {
 			setRequiresRecompileIfNecessary();
 		}
-		
+
 		return _etype;
 	}
 

src/main/java/org/apache/sysds/runtime/compress/CompressedMatrixBlock.java

@@ -58,6 +58,7 @@
 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.CLALibReorg;
 import org.apache.sysds.runtime.compress.lib.CLALibReplace;
 import org.apache.sysds.runtime.compress.lib.CLALibReshape;
 import org.apache.sysds.runtime.compress.lib.CLALibRexpand;
@@ -72,7 +73,6 @@
 import org.apache.sysds.runtime.data.DenseBlock;
 import org.apache.sysds.runtime.data.SparseBlock;
 import org.apache.sysds.runtime.data.SparseRow;
-import org.apache.sysds.runtime.functionobjects.SwapIndex;
 import org.apache.sysds.runtime.instructions.InstructionUtils;
 import org.apache.sysds.runtime.instructions.cp.CM_COV_Object;
 import org.apache.sysds.runtime.instructions.cp.ScalarObject;
@@ -633,21 +633,7 @@ public MatrixBlock replaceOperations(MatrixValue result, double pattern, double
 
 	@Override
 	public MatrixBlock reorgOperations(ReorgOperator op, MatrixValue ret, int startRow, int startColumn, int length) {
-		if(op.fn instanceof SwapIndex && this.getNumColumns() == 1) {
-			MatrixBlock tmp = decompress(op.getNumThreads());
-			long nz = tmp.setNonZeros(tmp.getNonZeros());
-			tmp = new MatrixBlock(tmp.getNumColumns(), tmp.getNumRows(), tmp.getDenseBlockValues());
-			tmp.setNonZeros(nz);
-			return tmp;
-		}
-		else {
-			// Allow transpose to be compressed output. In general we need to have a transposed flag on
-			// the compressed matrix. https://issues.apache.org/jira/browse/SYSTEMDS-3025
-			String message = op.getClass().getSimpleName() + " -- " + op.fn.getClass().getSimpleName();
-			MatrixBlock tmp = getUncompressed(message, op.getNumThreads());
-			return tmp.reorgOperations(op, ret, startRow, startColumn, length);
-		}
-
+		return CLALibReorg.reorg(this, op, (MatrixBlock) ret, startRow, startColumn, length);
 	}
 
 	public boolean isOverlapping() {
@@ -1215,8 +1201,8 @@ public void examSparsity(boolean allowCSR, int k) {
 	}
 
 	@Override
-	public void sparseToDense(int k) {
-		// do nothing
+	public MatrixBlock sparseToDense(int k) {
+		return this; // do nothing
 	}
 
 	@Override

src/main/java/org/apache/sysds/runtime/compress/lib/CLALibReorg.java

@@ -0,0 +1,158 @@
+/*
+ * 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.compress.CompressedMatrixBlock;
+import org.apache.sysds.runtime.compress.DMLCompressionException;
+import org.apache.sysds.runtime.compress.colgroup.AColGroup;
+import org.apache.sysds.runtime.data.DenseBlock;
+import org.apache.sysds.runtime.data.SparseBlock;
+import org.apache.sysds.runtime.data.SparseBlockMCSR;
+import org.apache.sysds.runtime.functionobjects.SwapIndex;
+import org.apache.sysds.runtime.matrix.data.LibMatrixReorg;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.matrix.operators.ReorgOperator;
+import org.apache.sysds.runtime.util.CommonThreadPool;
+
+public class CLALibReorg {
+
+	protected static final Log LOG = LogFactory.getLog(CLALibReorg.class.getName());
+
+	public static boolean warned = false;
+
+	public static MatrixBlock reorg(CompressedMatrixBlock cmb, ReorgOperator op, MatrixBlock ret, int startRow,
+		int startColumn, int length) {
+		// SwapIndex is transpose
+		if(op.fn instanceof SwapIndex && cmb.getNumColumns() == 1) {
+			MatrixBlock tmp = cmb.decompress(op.getNumThreads());
+			long nz = tmp.setNonZeros(tmp.getNonZeros());
+			if(tmp.isInSparseFormat())
+				return LibMatrixReorg.transpose(tmp); // edge case...
+			else
+				tmp = new MatrixBlock(tmp.getNumColumns(), tmp.getNumRows(), tmp.getDenseBlockValues());
+			tmp.setNonZeros(nz);
+			return tmp;
+		}
+		else if(op.fn instanceof SwapIndex) {
+			if(cmb.getCachedDecompressed() != null)
+				return cmb.getCachedDecompressed().reorgOperations(op, ret, startRow, startColumn, length);
+
+			return transpose(cmb, ret, op.getNumThreads());
+		}
+		else {
+			// Allow transpose to be compressed output. In general we need to have a transposed flag on
+			// the compressed matrix. https://issues.apache.org/jira/browse/SYSTEMDS-3025
+			String message = !warned ? op.getClass().getSimpleName() + " -- " + op.fn.getClass().getSimpleName() : null;
+			MatrixBlock tmp = cmb.getUncompressed(message, op.getNumThreads());
+			warned = true;
+			return tmp.reorgOperations(op, ret, startRow, startColumn, length);
+		}
+	}
+
+	private static MatrixBlock transpose(CompressedMatrixBlock cmb, MatrixBlock ret, int k) {
+
+		final long nnz = cmb.getNonZeros();
+		final int nRow = cmb.getNumRows();
+		final int nCol = cmb.getNumColumns();
+		final boolean sparseOut = MatrixBlock.evalSparseFormatInMemory(nRow, nCol, nnz);
+		if(sparseOut)
+			return transposeSparse(cmb, ret, k, nRow, nCol, nnz);
+		else
+			return transposeDense(cmb, ret, k, nRow, nCol, nnz);
+	}
+
+	private static MatrixBlock transposeSparse(CompressedMatrixBlock cmb, MatrixBlock ret, int k, int nRow, int nCol,
+		long nnz) {
+		if(ret == null)
+			ret = new MatrixBlock(nCol, nRow, true, nnz);
+		else
+			ret.reset(nCol, nRow, true, nnz);
+
+		ret.allocateAndResetSparseBlock(true, SparseBlock.Type.MCSR);
+
+		final int nColOut = ret.getNumColumns();
+
+		if(k > 1)
+			decompressToTransposedSparseParallel((SparseBlockMCSR) ret.getSparseBlock(), cmb.getColGroups(), nColOut, k);
+		else
+			decompressToTransposedSparseSingleThread((SparseBlockMCSR) ret.getSparseBlock(), cmb.getColGroups(), nColOut);
+
+		return ret;
+	}
+
+	private static MatrixBlock transposeDense(CompressedMatrixBlock cmb, MatrixBlock ret, int k, int nRow, int nCol,
+		long nnz) {
+		if(ret == null)
+			ret = new MatrixBlock(nCol, nRow, false, nnz);
+		else
+			ret.reset(nCol, nRow, false, nnz);
+
+		// TODO: parallelize
+		ret.allocateDenseBlock();
+
+		decompressToTransposedDense(ret.getDenseBlock(), cmb.getColGroups(), nRow, 0, nRow);
+		return ret;
+	}
+
+	private static void decompressToTransposedDense(DenseBlock ret, List<AColGroup> groups, int rlen, int rl, int ru) {
+		for(int i = 0; i < groups.size(); i++) {
+			AColGroup g = groups.get(i);
+			g.decompressToDenseBlockTransposed(ret, rl, ru);
+		}
+	}
+
+	private static void decompressToTransposedSparseSingleThread(SparseBlockMCSR ret, List<AColGroup> groups,
+		int nColOut) {
+		for(int i = 0; i < groups.size(); i++) {
+			AColGroup g = groups.get(i);
+			g.decompressToSparseBlockTransposed(ret, nColOut);
+		}
+	}
+
+	private static void decompressToTransposedSparseParallel(SparseBlockMCSR ret, List<AColGroup> groups, int nColOut,
+		int k) {
+		final ExecutorService pool = CommonThreadPool.get(k);
+		try {
+			final List<Future<?>> tasks = new ArrayList<>(groups.size());
+
+			for(int i = 0; i < groups.size(); i++) {
+				final AColGroup g = groups.get(i);
+				tasks.add(pool.submit(() -> g.decompressToSparseBlockTransposed(ret, nColOut)));
+			}
+
+			for(Future<?> f : tasks)
+				f.get();
+
+		}
+		catch(Exception e) {
+			throw new DMLCompressionException("Failed to parallel decompress transpose sparse", e);
+		}
+		finally {
+			pool.shutdown();
+		}
+	}
+}

src/main/java/org/apache/sysds/runtime/compress/lib/CLALibReshape.java

@@ -161,7 +161,7 @@ private void checkValidity() {
 
 	private boolean shouldItBeCompressedOutputs() {
 		// The number of rows in the reshaped allocations is fairly large.
-		return rlen > COMPRESSED_RESHAPE_THRESHOLD && rowwise &&
+		return rlen > COMPRESSED_RESHAPE_THRESHOLD &&
 			// 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;
 	}

src/main/java/org/apache/sysds/runtime/instructions/SPInstructionParser.java

@@ -39,6 +39,7 @@
 import org.apache.sysds.lops.WeightedUnaryMM;
 import org.apache.sysds.lops.WeightedUnaryMMR;
 import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.instructions.cp.CPInstruction.CPType;
 import org.apache.sysds.runtime.instructions.cp.CPOperand;
 import org.apache.sysds.runtime.instructions.spark.AggregateTernarySPInstruction;
 import org.apache.sysds.runtime.instructions.spark.AggregateUnarySPInstruction;
@@ -195,6 +196,7 @@ public class SPInstructionParser extends InstructionParser
 		String2SPInstructionType.put( "freplicate", SPType.Binary);
 		String2SPInstructionType.put( "mapdropInvalidLength", SPType.Binary);
 		String2SPInstructionType.put( "valueSwap", SPType.Binary);
+		String2SPInstructionType.put( "applySchema"  , SPType.Binary);
 		String2SPInstructionType.put( "_map", SPType.Ternary); // _map refers to the operation map
 		// Relational Instruction Opcodes
 		String2SPInstructionType.put( "=="   , SPType.Binary);