Merge branch 'main' into einsum

Author: HubertKrawczyk

.github/workflows/javaTests.yml

@@ -29,6 +29,7 @@ on:
       - '*.html'
       - 'src/main/python/**'
       - 'dev/**'
+      - '.github/workflows/python.yml'
     branches:
       - main
   pull_request:
@@ -38,6 +39,7 @@ on:
       - '*.html'
       - 'src/main/python/**'
       - 'dev/**'
+      - '.github/workflows/python.yml'
     branches:
       - main
 

.github/workflows/python.yml

@@ -45,15 +45,17 @@ on:
 jobs:
   test:
     runs-on: ${{ matrix.os }}
+    timeout-minutes: 60
     strategy:
       fail-fast: false
       matrix:
         python-version: [3.8]
         os: [ubuntu-24.04]
         java: ['17']
         javadist: ['adopt']
+        test_mode: [env, noenv, federated]
 
-    name: ${{ matrix.os }} Java ${{ matrix.java }} ${{ matrix.javadist }} Python ${{ matrix.python-version }}
+    name: ${{ matrix.os }} Java ${{ matrix.java }} ${{ matrix.javadist }} Python ${{ matrix.python-version }}/ ${{ matrix.test_mode}}
     steps:
     - name: Checkout Repository
       uses: actions/checkout@v4
@@ -115,7 +117,6 @@ jobs:
           librosa \
           h5py \
           gensim \
-          black \
           opt-einsum \
           nltk
 
@@ -124,38 +125,29 @@ jobs:
         cd src/main/python
         python create_python_dist.py
 
-    - name: Run all python tests
+    - name: Run tests with env
+      if: ${{ matrix.test_mode == 'env' }}
       run: |
         export SYSTEMDS_ROOT=$(pwd)
         export PATH=$SYSTEMDS_ROOT/bin:$PATH
         export SYSDS_QUIET=1
         export LOG4JPROP=$SYSTEMDS_ROOT/src/test/resources/log4j.properties
         cd src/main/python
         unittest-parallel -t . -s tests -v
-        # python -m unittest discover -s tests -p 'test_*.py'
-        echo "Exit Status: " $?
 
-    - name: Run all python tests no environment
+    - name: Run tests no env
+      if: ${{ matrix.test_mode == 'noenv' }}
       run: |
         export LOG4JPROP=$(pwd)/src/test/resources/log4j.properties
         cd src/main/python
         unittest-parallel -t . -s tests -v
-        # python -m unittest discover -s tests -p 'test_*.py'
-        echo "Exit Status: " $?
 
     - name: Run Federated Python Tests
+      if: ${{ matrix.test_mode == 'federated' }}
       run: |
         export SYSTEMDS_ROOT=$(pwd)
         export PATH=$SYSTEMDS_ROOT/bin:$PATH
         cd src/main/python
         ./tests/federated/runFedTest.sh
 
-    - name: Check formatting according to Black (src/main/python/systemds)
-      run: |
-        black --check --exclude operator/algorithm src/main/python/systemds
-
-    - name: Check formatting according to Black (src/main/python/tests)
-      run: |
-        black --check src/main/python/tests
-
 

.github/workflows/pythonFormatting.yml

@@ -0,0 +1,56 @@
+#-------------------------------------------------------------
+#
+# 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.
+#
+#-------------------------------------------------------------
+
+name: Python Black Format Check
+
+on:
+  push:
+    paths:
+      - 'src/main/python/**'
+    branches:
+      - main
+  pull_request:
+    paths:
+      - 'src/main/python/**'
+    branches:
+      - main
+
+jobs:
+  black:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout Repository
+        uses: actions/checkout@v4
+
+      - name: Setup Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.x'
+
+      - name: Install Black
+        run: |
+          pip install --upgrade pip
+          pip install black
+
+      - name: Run Black Check
+        run: |
+          black --check --exclude operator/algorithm \
+          src/main/python/systemds src/main/python/tests

docs/site/dml-language-reference.md

@@ -702,7 +702,8 @@ quantile () | The p-quantile for a random variable X is the value x such that Pr
 quantile () | Returns a column matrix with list of all quantiles requested in P. | Input: (X &lt;(n x 1) matrix&gt;, [W &lt;(n x 1) matrix&gt;),] P &lt;(q x 1) matrix&gt;) <br/> Output: matrix | quantile(X, P) <br/> quantile(X, W, P)
 median() | Computes the median in a given column matrix of values | Input: (X &lt;(n x 1) matrix&gt;, [W &lt;(n x 1) matrix&gt;),]) <br/> Output: &lt;scalar&gt; | median(X) <br/> median(X,W)
 rowSums() <br/> rowMeans() <br/> rowVars() <br/> rowSds() <br/> rowMaxs() <br/> rowMins() | Row-wise computations -- for each row, compute the sum/mean/variance/stdDev/max/min of cell value | Input: matrix <br/> Output: (n x 1) matrix | rowSums(X) <br/> rowMeans(X) <br/> rowVars(X) <br/> rowSds(X) <br/> rowMaxs(X) <br/> rowMins(X)
-cumsum() | Column prefix-sum (For row-prefix sum, use cumsum(t(X)) | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("1 2 3 4 5 6", rows=3, cols=2) <br/> B = cumsum(A) <br/> The output matrix B = [[1, 2], [4, 6], [9, 12]]
+cumsum() | Column prefix-sum | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("1 2 3 4 5 6", rows=3, cols=2) <br/> B = cumsum(A) <br/> The output matrix B = [[1, 2], [4, 6], [9, 12]]
+rowcumsum() | Row prefix-sum | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("1 2 3 4 5 6", rows=2, cols=3) <br/> B = rowcumsum(A) <br/> The output matrix B = [[1, 3, 6], [4, 9, 15]]
 cumprod() | Column prefix-prod (For row-prefix prod, use cumprod(t(X)) | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("1 2 3 4 5 6", rows=3, cols=2) <br/> B = cumprod(A) <br/> The output matrix B = [[1, 2], [3, 8], [15, 48]]
 cummin() | Column prefix-min (For row-prefix min, use cummin(t(X)) | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("3 4 1 6 5 2", rows=3, cols=2) <br/> B = cummin(A) <br/> The output matrix B = [[3, 4], [1, 4], [1, 2]]
 cummax() | Column prefix-max (For row-prefix min, use cummax(t(X)) | Input: matrix <br/> Output: matrix of the same dimensions | A = matrix("3 4 1 6 5 2", rows=3, cols=2) <br/> B = cummax(A) <br/> The output matrix B = [[3, 4], [3, 6], [5, 6]]

src/main/java/org/apache/sysds/api/PythonDMLScript.java

@@ -25,17 +25,40 @@
 import org.apache.log4j.Logger;
 import org.apache.sysds.api.jmlc.Connection;
 
+import org.apache.sysds.common.Types;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.matrix.data.MatrixBlock;
+import org.apache.sysds.runtime.util.CommonThreadPool;
+import org.apache.sysds.runtime.util.UnixPipeUtils;
 import py4j.DefaultGatewayServerListener;
 import py4j.GatewayServer;
 import py4j.Py4JNetworkException;
 
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
 
 public class PythonDMLScript {
 
 	private static final Log LOG = LogFactory.getLog(PythonDMLScript.class.getName());
 	final private Connection _connection;
 	public static GatewayServer GwS;
 
+	private static String fromPythonBase = "py2java";
+	private static String toPythonBase = "java2py";
+	public HashMap<Integer, BufferedInputStream> fromPython = null;
+	public HashMap<Integer, BufferedOutputStream> toPython = null;
+	public String baseDir;
+	private static int BATCH_SIZE = 32*1024;
+
 	/**
 	 * Entry point for Python API.
 	 * 
@@ -78,6 +101,103 @@ public Connection getConnection() {
 		return _connection;
 	}
 
+
+	public void openPipes(String path, int num) throws IOException {
+		fromPython = new HashMap<>(num * 2);
+		toPython = new HashMap<>(num * 2);
+		baseDir = path;
+		for (int i = 0; i < num; i++) {
+			BufferedInputStream pipe_in = UnixPipeUtils.openInput(path + "/" + fromPythonBase + "-" + i, i);
+			LOG.debug("PY2JAVA pipe "+i+" is ready!");
+			fromPython.put(i, pipe_in);
+
+			BufferedOutputStream pipe_out = UnixPipeUtils.openOutput(path + "/" +  toPythonBase + "-" + i, i);
+			toPython.put(i, pipe_out);
+		}
+	}
+
+	public MatrixBlock startReadingMbFromPipe(int id, int rlen, int clen, Types.ValueType type) throws IOException {
+		long limit = (long) rlen * clen;
+		LOG.debug("trying to read matrix from "+id+" with "+rlen+" rows and "+clen+" columns. Total size: "+limit);
+		if(limit > Integer.MAX_VALUE)
+			throw new DMLRuntimeException("Dense NumPy array of size " + limit +
+					" cannot be converted to MatrixBlock");
+		MatrixBlock mb = new MatrixBlock(rlen, clen, false, -1);
+		if(fromPython != null){
+			BufferedInputStream pipe = fromPython.get(id);
+			double[] denseBlock = new double[(int) limit];
+			UnixPipeUtils.readNumpyArrayInBatches(pipe, id, BATCH_SIZE, (int) limit, type, denseBlock, 0);
+			mb.init(denseBlock, rlen, clen);
+		} else {
+			throw new DMLRuntimeException("FIFO Pipes are not initialized.");
+		}
+		mb.recomputeNonZeros();
+		mb.examSparsity();
+		LOG.debug("Reading from Python finished");
+		return mb;
+	}
+
+	public MatrixBlock startReadingMbFromPipes(int[] blockSizes, int rlen, int clen, Types.ValueType type) throws ExecutionException, InterruptedException {
+		long limit = (long) rlen * clen;
+		if(limit > Integer.MAX_VALUE)
+			throw new DMLRuntimeException("Dense NumPy array of size " + limit +
+					" cannot be converted to MatrixBlock");
+		MatrixBlock mb = new MatrixBlock(rlen, clen, false, -1);
+		if(fromPython != null){
+			ExecutorService pool = CommonThreadPool.get();
+			double[] denseBlock = new double[(int) limit];
+			int offsetOut = 0;
+			List<Future<Void>> futures = new ArrayList<>();
+			for (int i = 0; i < blockSizes.length; i++) {
+				BufferedInputStream pipe = fromPython.get(i);
+				int id = i, blockSize = blockSizes[i], _offsetOut = offsetOut;
+				Callable<Void> task = () -> {
+					UnixPipeUtils.readNumpyArrayInBatches(pipe, id, BATCH_SIZE, blockSize, type, denseBlock, _offsetOut);
+					return null;
+				};
+
+				futures.add(pool.submit(task));
+				offsetOut += blockSize;
+			}
+			// Wait for all tasks and propagate exceptions
+			for (Future<Void> f : futures) {
+				f.get();
+			}
+
+			mb.init(denseBlock, rlen, clen);
+		} else {
+			throw new DMLRuntimeException("FIFO Pipes are not initialized.");
+		}
+		mb.recomputeNonZeros();
+		mb.examSparsity();
+		return mb;
+	}
+
+	public void startWritingMbToPipe(int id, MatrixBlock mb) throws IOException {
+		if (toPython != null) {
+			int rlen = mb.getNumRows();
+			int clen = mb.getNumColumns();
+			int numElem = rlen * clen;
+			LOG.debug("Trying to write matrix ["+baseDir + "-"+ id+"] with "+rlen+" rows and "+clen+" columns. Total size: "+numElem*8);
+
+			BufferedOutputStream out = toPython.get(id);
+			long bytes = UnixPipeUtils.writeNumpyArrayInBatches(out, id, BATCH_SIZE, numElem, Types.ValueType.FP64, mb);
+
+			LOG.debug("Writing of " + bytes +" Bytes to Python ["+baseDir + "-"+ id+"] finished");
+		} else {
+			throw new DMLRuntimeException("FIFO Pipes are not initialized.");
+		}
+	}
+
+	public void closePipes() throws IOException {
+		LOG.debug("Closing all pipes in Java");
+		for (BufferedInputStream pipe : fromPython.values())
+			pipe.close();
+		for (BufferedOutputStream pipe : toPython.values())
+			pipe.close();
+		LOG.debug("Closed all pipes in Java");
+	}
+
 	protected static class DMLGateWayListener extends DefaultGatewayServerListener {
 		private static final Log LOG = LogFactory.getLog(DMLGateWayListener.class.getName());
 

src/main/java/org/apache/sysds/common/Builtins.java

@@ -291,6 +291,7 @@ public enum Builtins {
 	ROLL("roll", false),
 	ROUND("round", false),
 	ROW_COUNT_DISTINCT("rowCountDistinct",false),
+	ROWCUMSUM("rowcumsum", false),
 	ROWINDEXMAX("rowIndexMax", false),
 	ROWINDEXMIN("rowIndexMin", false),
 	ROWMAX("rowMaxs", false),

src/main/java/org/apache/sysds/common/Opcodes.java

@@ -34,6 +34,7 @@ public enum Opcodes {
 	UAKP("uak+", InstructionType.AggregateUnary),
 	UARKP("uark+", InstructionType.AggregateUnary),
 	UACKP("uack+", InstructionType.AggregateUnary),
+	UARCKP("uarck+", InstructionType.AggregateUnary),
 	UASQKP("uasqk+", InstructionType.AggregateUnary),
 	UARSQKP("uarsqk+", InstructionType.AggregateUnary),
 	UACSQKP("uacsqk+", InstructionType.AggregateUnary),
@@ -151,6 +152,7 @@ public enum Opcodes {
 	CEIL("ceil", InstructionType.Unary),
 	FLOOR("floor", InstructionType.Unary),
 	UCUMKP("ucumk+", InstructionType.Unary),
+	UROWCUMKP("urowcumk+", InstructionType.Unary),
 	UCUMM("ucum*", InstructionType.Unary),
 	UCUMKPM("ucumk+*", InstructionType.Unary),
 	UCUMMIN("ucummin", InstructionType.Unary),
@@ -383,6 +385,7 @@ public enum Opcodes {
 	UCUMACMIN("ucumacmin", InstructionType.CumsumAggregate),
 	UCUMACMAX("ucumacmax", InstructionType.CumsumAggregate),
 	BCUMOFFKP("bcumoffk+", InstructionType.CumsumOffset),
+	BROWCUMOFFKP("browcumoffk+", InstructionType.CumsumOffset),
 	BCUMOFFM("bcumoff*", InstructionType.CumsumOffset),
 	BCUMOFFPM("bcumoff+*", InstructionType.CumsumOffset),
 	BCUMOFFMIN("bcumoffmin", InstructionType.CumsumOffset),

src/main/java/org/apache/sysds/common/Types.java

@@ -547,7 +547,7 @@ public enum OpOp1 {
 		CEIL, CHOLESKY, COS, COSH, CUMMAX, CUMMIN, CUMPROD, CUMSUM,
 		CUMSUMPROD, DET, DETECTSCHEMA, COLNAMES, EIGEN, EXISTS, EXP, FLOOR, INVERSE,
 		IQM, ISNA, ISNAN, ISINF, LENGTH, LINEAGE, LOG, NCOL, NOT, NROW,
-		MEDIAN, PREFETCH, PRINT, ROUND, SIN, SINH, SIGN, SOFTMAX, SQRT, STOP, _EVICT,
+		MEDIAN, PREFETCH, PRINT, ROUND, ROWCUMSUM, SIN, SINH, SIGN, SOFTMAX, SQRT, STOP, _EVICT,
 		SVD, TAN, TANH, TYPEOF, TRIGREMOTE, SQRT_MATRIX_JAVA,
 		//fused ML-specific operators for performance 
 		SPROP, //sample proportion: P * (1 - P)
@@ -591,6 +591,7 @@ public String toString() {
 				case MULT2:           return Opcodes.MULT2.toString();
 				case NOT:             return Opcodes.NOT.toString();
 				case POW2:            return Opcodes.POW2.toString();
+				case ROWCUMSUM:		  return Opcodes.UROWCUMKP.toString();
 				case TYPEOF:          return Opcodes.TYPEOF.toString();
 				default:              return name().toLowerCase();
 			}
@@ -610,6 +611,7 @@ public static OpOp1 valueOfByOpcode(String opcode) {
 				case "ucummin": return CUMMIN;
 				case "ucum*":   return CUMPROD;
 				case "ucumk+":  return CUMSUM;
+				case "urowcumk+":  return ROWCUMSUM;
 				case "ucumk+*": return CUMSUMPROD;
 				case "detectSchema":    return DETECTSCHEMA;
 				case "*2":      return MULT2;

src/main/java/org/apache/sysds/hops/rewrite/RewriteAlgebraicSimplificationStatic.java

@@ -205,6 +205,7 @@ private void rule_AlgebraicSimplification(Hop hop, boolean descendFirst)
 			hi = simplifyNotOverComparisons(hop, hi, i);         //e.g., !(A>B) -> (A<=B)
 			hi = simplifyMatrixScalarPMOperation(hop, hi, i);    //e.g., a-A-b -> (a-b)-A; a+A-b -> (a-b)+A
 			//hi = removeUnecessaryPPred(hop, hi, i);            //e.g., ppred(X,X,"==")->matrix(1,rows=nrow(X),cols=ncol(X))
+			hi = simplifyTransposedCumsum(hop, hi, i);					 //e.g., t(cumsum(t(X))) -> rowcumsum(X)
 
 			//process childs recursively after rewrites (to investigate pattern newly created by rewrites)
 			if( !descendFirst )
@@ -214,6 +215,28 @@ private void rule_AlgebraicSimplification(Hop hop, boolean descendFirst)
 		hop.setVisited();
 	}
 
+	private static Hop simplifyTransposedCumsum( Hop parent, Hop hi, int pos )
+	{
+		//e.g., t(cumsum(t(X))) -> rowcumsum(X)
+		if(   HopRewriteUtils.isTransposeOperation(hi)
+				&& hi.getInput(0) instanceof UnaryOp
+				&& ((UnaryOp)hi.getInput(0)).getOp() == OpOp1.CUMSUM
+				&& hi.getInput(0).getParent().size() == 1
+				&& HopRewriteUtils.isTransposeOperation(hi.getInput(0).getInput(0), 1)) //inner transpose with single consumer
+		{
+			UnaryOp cumsum=(UnaryOp)hi.getInput(0);
+			Hop innerMatrix = cumsum.getInput(0).getInput(0);
+
+			UnaryOp rowcumsumOp = HopRewriteUtils.createUnary(innerMatrix, OpOp1.ROWCUMSUM);
+			HopRewriteUtils.replaceChildReference(parent,hi, rowcumsumOp, pos);
+
+			hi = rowcumsumOp;
+			LOG.debug("Applied simplifyTransposedCumsum (line "+hi.getBeginLine()+").");
+		}
+
+		return hi;
+	}
+
 	private Hop simplifyMatrixScalarPMOperation(Hop parent, Hop hi, int pos) {
 		if (!(hi instanceof BinaryOp))
 			return hi;