[DO NOT MERGE] JAVA 17 BWARE COMMIT V3

bin/systemds

@@ -408,6 +408,8 @@ if [ $SYSDS_QUIET == 0 ]; then
   print_out "#"
 fi
 
+java --version
+
 # Build the command to run
 if [ $WORKER == 1 ]; then
   print_out "#  starting Federated worker on port $PORT"
@@ -422,6 +424,7 @@ elif [ "$FEDMONITORING" == 1 ]; then
   print_out "#  starting Federated backend monitoring on port $PORT"
   CMD=" \
   java $SYSTEMDS_STANDALONE_OPTS \
+  --add-modules=jdk.incubator.vector \
   $LOG4JPROPFULL \
   -jar $SYSTEMDS_JAR_FILE \
   -fedMonitoring $PORT \
@@ -433,6 +436,7 @@ elif [ $SYSDS_DISTRIBUTED == 0 ]; then
   CMD=" \
   java $SYSTEMDS_STANDALONE_OPTS \
   $LOG4JPROPFULL \
+  --add-modules=jdk.incubator.vector \
   -jar $SYSTEMDS_JAR_FILE \
   -f $SCRIPT_FILE \
   -exec $SYSDS_EXEC_MODE \
@@ -442,6 +446,7 @@ else
   print_out "#  Running script $SCRIPT_FILE distributed with opts: $*"
   CMD=" \
   spark-submit $SYSTEMDS_DISTRIBUTED_OPTS \
+  --add-modules=jdk.incubator.vector \
   $SYSTEMDS_JAR_FILE \
   -f $SCRIPT_FILE \
   -exec $SYSDS_EXEC_MODE \

pom.xml

@@ -65,7 +65,7 @@
 		<maven-compiler-plugin.version>3.11.0</maven-compiler-plugin.version>
 		<maven-antrun-plugin.version>3.1.0</maven-antrun-plugin.version>
 		<!-- Set java compile level via argument, ex: 1.8 1.9 10 11-->
-		<java.level>11</java.level>
+		<java.level>17</java.level>
 		<java.version>{java.level}</java.version>
 		<!-->Testing settings<!-->
 		<maven.test.skip>false</maven.test.skip>
@@ -343,6 +343,9 @@
 					<source>${java.level}</source>
 					<target>${java.level}</target>
 					<release>${java.level}</release>
+					<compilerArgs>
+						<arg>--add-modules=jdk.incubator.vector</arg>
+					</compilerArgs>
 				</configuration>
 			</plugin>
 
@@ -365,6 +368,7 @@
 					<systemPropertyVariables>
 						<log4j.configurationFile>file:src/test/resources/log4j.properties</log4j.configurationFile>
 					</systemPropertyVariables>
+					<argLine>--add-modules=jdk.incubator.vector</argLine>
 				</configuration>
 			</plugin>
 

scripts/builtin/l2svm.dml

@@ -88,7 +88,8 @@ m_l2svm = function(Matrix[Double] X, Matrix[Double] Y, Boolean intercept = FALSE
   w = matrix(0, rows=ncol(X), cols=1)
   Xw = matrix(0, rows=nrow(X), cols=1)
 
-  g_old = t(X) %*% Y
+  ## rewrite t(X) %*% Y
+  g_old = t(t(Y) %*% X)
   s = g_old
 
   iter = 0
@@ -121,7 +122,9 @@ m_l2svm = function(Matrix[Double] X, Matrix[Double] Y, Boolean intercept = FALSE
     sv = (out > 0)
     out = sv * out
     obj = 0.5 * sum(out * out) + reg/2 * sum(w * w)
-    g_new = t(X) %*% (out * Y) - reg * w
+    ## rewrite t(X) %*% (out * Y) - reg * w
+    g_new = t(t(out*Y) %*% X) - reg * w
+    # g_new = t(X) %*% (out * Y) - reg * w
 
     if(verbose) {
       colstr = ifelse(columnId!=-1, "-- MSVM class="+columnId+": ", "")

scripts/builtin/lmCG.dml

@@ -95,15 +95,17 @@ m_lmCG = function(Matrix[Double] X, Matrix[Double] y, Integer icpt = 0,
   lambda = scale_lambda * regularization
   beta_unscaled = matrix(0, rows = m_ext, cols = 1)
 
-  if(max_iteration == 0){
+  if((max_iteration == 0) | (max_iteration > m_ext)){
     max_iteration = m_ext
   }
   i = 0
 
   # BEGIN THE CONJUGATE GRADIENT ALGORITHM
   if(verbose) print("Running the CG algorithm...")
 
-  r = - t(X) %*% y
+  # Equivalent to - t(X) %*% y, we have a rewrite to detect it
+  # But CLA does not do it correctly... TODO to fix that rewrite.
+  r = - t(t(y) %*% X)
 
   if(intercept_status == 2){
     r = scale_X * r + shift_X %*% r [m_ext, ]
@@ -116,8 +118,10 @@ m_lmCG = function(Matrix[Double] X, Matrix[Double] y, Integer icpt = 0,
   if(verbose){
     print("||r|| initial value = " + sqrt(norm_r2_initial) +
       ", target value = " + sqrt(norm_r2_target))
+    print("Max Iteraton: " + max_iteration)
   }
 
+
   while(i < max_iteration & norm_r2 > norm_r2_target){
     if(intercept_status == 2){
       ssX_p = scale_X * p
@@ -128,10 +132,15 @@ m_lmCG = function(Matrix[Double] X, Matrix[Double] y, Integer icpt = 0,
 
     q = t(X) %*% (X %*% ssX_p)
 
+
+
+
     if(intercept_status == 2) {
       q = scale_X * q + shift_X %*% q [m_ext, ]
     }
 
+    qi = q;
+
     q += lambda * p
     a = norm_r2 / sum(p * q)
     beta_unscaled += a * p
@@ -142,7 +151,8 @@ m_lmCG = function(Matrix[Double] X, Matrix[Double] y, Integer icpt = 0,
     i = i + 1
     if(verbose){
       print("Iteration " + i + ":  ||r|| / ||r init|| = "
-        + sqrt(norm_r2 / norm_r2_initial))
+        + sqrt(norm_r2 / norm_r2_initial)
+        + "   :  " + sum(qi))
     }
   }
 

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

@@ -446,8 +446,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

@@ -752,8 +752,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 ) {
 			_etype = _etypeForced;
@@ -802,18 +802,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 &&
@@ -843,7 +853,7 @@ else if( (op == OpOp2.CBIND && getDataType().isList())
 				|| (op == OpOp2.RBIND && getDataType().isList())) {
 			_etype = ExecType.CP;
 		}
-		
+
 		//mark for recompile (forever)
 		setRequiresRecompileIfNecessary();
 
@@ -1160,17 +1170,35 @@ && getInput().get(0) == that2.getInput().get(0)
 	}
 
 	public boolean supportsMatrixScalarOperations() {
-		return ( op==OpOp2.PLUS ||op==OpOp2.MINUS
-				||op==OpOp2.MULT ||op==OpOp2.DIV
-				||op==OpOp2.MODULUS ||op==OpOp2.INTDIV
-				||op==OpOp2.LESS ||op==OpOp2.LESSEQUAL
-				||op==OpOp2.GREATER ||op==OpOp2.GREATEREQUAL
-				||op==OpOp2.EQUAL ||op==OpOp2.NOTEQUAL
-				||op==OpOp2.MIN ||op==OpOp2.MAX
-				||op==OpOp2.LOG ||op==OpOp2.POW
-				||op==OpOp2.AND ||op==OpOp2.OR ||op==OpOp2.XOR
-				||op==OpOp2.BITWAND ||op==OpOp2.BITWOR ||op==OpOp2.BITWXOR
-				||op==OpOp2.BITWSHIFTL ||op==OpOp2.BITWSHIFTR);
+		switch(op) {
+			case PLUS:
+			case MINUS:
+			case MULT:
+			case DIV:
+			case MODULUS:
+			case INTDIV:
+			case LESS:
+			case LESSEQUAL:
+			case GREATER:
+			case GREATEREQUAL:
+			case EQUAL:
+			case NOTEQUAL:
+			case MIN:
+			case MAX:
+			case LOG:
+			case POW:
+			case AND:
+			case OR:
+			case XOR:
+			case BITWAND:
+			case BITWOR:
+			case BITWXOR:
+			case BITWSHIFTL:
+			case BITWSHIFTR:
+				return true;
+			default:
+				return false;
+		}
 	}
 
 	public boolean isPPredOperation() {

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

@@ -386,8 +386,8 @@ public boolean allowsAllExecTypes()
 	protected double computeOutputMemEstimate( long dim1, long dim2, long nnz )
 	{		
 		double ret = 0;
-
-		if ( getDataType() == DataType.SCALAR ) 
+		final DataType dt = getDataType();
+		if ( dt == DataType.SCALAR ) 
 		{
 			switch( getValueType() ) 
 			{
@@ -406,6 +406,11 @@ protected double computeOutputMemEstimate( long dim1, long dim2, long nnz )
 					ret = 0;
 			}
 		}
+		else if(dt == DataType.FRAME) {
+			if(_op == OpOpData.PERSISTENTREAD || _op == OpOpData.TRANSIENTREAD) {
+				ret = OptimizerUtils.estimateSizeExactFrame(dim1, dim2);
+			}
+		}
 		else //MATRIX / FRAME
 		{
 			if(   _op == OpOpData.PERSISTENTREAD 

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

@@ -1064,6 +1064,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) );
 	}
@@ -1648,6 +1654,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/OptimizerUtils.java

@@ -64,6 +64,7 @@
 import org.apache.sysds.runtime.util.IndexRange;
 import org.apache.sysds.runtime.util.UtilFunctions;
 import org.apache.sysds.utils.stats.InfrastructureAnalyzer;
+import org.apache.sysds.utils.MemoryEstimates;
 
 public class OptimizerUtils 
 {
@@ -787,6 +788,15 @@ public static long estimateSizeExactSparsity(long nrows, long ncols, long nnz)
 		double sp = getSparsity(nrows, ncols, nnz);
 		return estimateSizeExactSparsity(nrows, ncols, sp);
 	}
+
+
+	public static long estimateSizeExactFrame(long nRows, long nCols){
+		if(nRows > Integer.MAX_VALUE)
+			return Long.MAX_VALUE;
+
+		// assuming String arrays and on average 8 characters per value.
+		return (long)MemoryEstimates.stringArrayCost((int)nRows, 8) * nCols;
+	}
 
 	/**
 	 * Estimates the footprint (in bytes) for an in-memory representation of a

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

@@ -21,6 +21,7 @@
 
 import org.apache.sysds.api.DMLScript;
 import org.apache.sysds.common.Types.DataType;
+import org.apache.sysds.common.Types.ExecType;
 import org.apache.sysds.common.Types.OpOp2;
 import org.apache.sysds.common.Types.OpOp3;
 import org.apache.sysds.common.Types.OpOpDG;
@@ -33,8 +34,8 @@
 import org.apache.sysds.lops.CentralMoment;
 import org.apache.sysds.lops.CoVariance;
 import org.apache.sysds.lops.Ctable;
+import org.apache.sysds.lops.Data;
 import org.apache.sysds.lops.Lop;
-import org.apache.sysds.common.Types.ExecType;
 import org.apache.sysds.lops.LopsException;
 import org.apache.sysds.lops.PickByCount;
 import org.apache.sysds.lops.SortKeys;
@@ -284,14 +285,19 @@ private void constructLopsCtable() {
 		// F=ctable(A,B,W)
 
 		DataType dt1 = getInput().get(0).getDataType(); 
+
+
 		DataType dt2 = getInput().get(1).getDataType(); 
 		DataType dt3 = getInput().get(2).getDataType(); 
 		Ctable.OperationTypes ternaryOpOrig = Ctable.findCtableOperationByInputDataTypes(dt1, dt2, dt3);
 
 		// Compute lops for all inputs
 		Lop[] inputLops = new Lop[getInput().size()];
 		for(int i=0; i < getInput().size(); i++) {
-			inputLops[i] = getInput().get(i).constructLops();
+			if(i == 0 && HopRewriteUtils.isSequenceSizeOfA(getInput(0), getInput(1)))
+				inputLops[i] = Data.createLiteralLop(ValueType.INT64, "" +getInput(1).getDim(0));
+			else
+				inputLops[i] = getInput().get(i).constructLops();
 		}
 
 		ExecType et = optFindExecType();