[SYSTEMDS-3223] Python functions with list arguments

This commit fixes a bug where the sourced functions would not
correctly build the scripts in cases where a list is an input to the
function defined in a sourced script.

Author: Baunsgaard

src/main/python/systemds/context/systemds_context.py

@@ -64,8 +64,7 @@ def __init__(self, port: int = -1):
         if process.poll() is None:
             self.__start_gateway(actual_port)
         else:
-            self.exception_and_close(
-                "Java process stopped before gateway could connect")
+            self.exception_and_close("Java process stopped before gateway could connect")
 
     def get_stdout(self, lines: int = -1):
         """Getter for the stdout of the java subprocess
@@ -89,7 +88,7 @@ def get_stderr(self, lines: int = -1):
         else:
             return [self.__stderr.get() for x in range(lines)]
 
-    def exception_and_close(self, exception_str: str, trace_back_limit: int = None):
+    def exception_and_close(self, exception, trace_back_limit: int = None):
         """
         Method for printing exception, printing stdout and error, while also closing the context correctly.
 
@@ -104,7 +103,7 @@ def exception_and_close(self, exception_str: str, trace_back_limit: int = None):
         if stdErr:
             message += "standard error  :\n" + "\n".join(stdErr)
         message += "\n\n"
-        message += exception_str
+        message += str(exception)
         sys.tracebacklimit = trace_back_limit
         self.close()
         raise RuntimeError(message)

src/main/python/systemds/operator/nodes/list.py

@@ -76,11 +76,10 @@ def pass_python_data_to_prepared_script(self, sds, var_name: str, prepared_scrip
 
     def code_line(self, var_name: str, unnamed_input_vars: Sequence[str],
                   named_input_vars: Dict[str, str]) -> str:
-        inputs_comma_sep = create_params_string(
-            unnamed_input_vars, named_input_vars)
-        return f'{var_name}={self.operation}({inputs_comma_sep});'
+        code_line = super().code_line(var_name, unnamed_input_vars, named_input_vars)
+        return code_line
 
-    def compute(self, verbose: bool = False, lineage: bool = False) -> Union[np.array]:
+    def compute(self, verbose: bool = False, lineage: bool = False) -> np.array:
         return super().compute(verbose, lineage)
 
     def __str__(self):

src/main/python/systemds/operator/nodes/scalar.py

@@ -57,7 +57,7 @@ def code_line(self, var_name: str, unnamed_input_vars: Sequence[str],
         else:
             return super().code_line(var_name, unnamed_input_vars, named_input_vars)
 
-    def compute(self, verbose: bool = False, lineage: bool = False) -> Union[np.array]:
+    def compute(self, verbose: bool = False, lineage: bool = False):
         return super().compute(verbose, lineage)
 
     def _parse_output_result_variables(self, result_variables):

src/main/python/systemds/script_building/script.py

@@ -19,11 +19,12 @@
 #
 # -------------------------------------------------------------
 
-from typing import Any, Collection, KeysView, Tuple, Union, Optional, Dict, TYPE_CHECKING, List
+from typing import (TYPE_CHECKING, Any, Collection, Dict, KeysView, List,
+                    Optional, Tuple, Union)
 
+from py4j.protocol import Py4JNetworkError
 from py4j.java_collections import JavaArray
-from py4j.java_gateway import JavaObject, JavaGateway
-
+from py4j.java_gateway import JavaGateway, JavaObject
 from systemds.script_building.dag import DAGNode, OutputType
 from systemds.utils.consts import VALID_INPUT_TYPES
 
@@ -79,9 +80,14 @@ def execute(self) -> JavaObject:
             self.__prepare_script()
             ret = self.prepared_script.executeScript()
             return ret
+        except Py4JNetworkError:
+            exception_str = "Py4JNetworkError: no connection to JVM, most likely due to previous crash or closed JVM from calls to close()"
+            trace_back_limit = 0
         except Exception as e:
-            self.sds_context.exception_and_close(e)
-            return None
+            exception_str = str(e)
+            trace_back_limit = None
+        self.sds_context.exception_and_close(exception_str, trace_back_limit)
+        
 
     def execute_with_lineage(self) -> Tuple[JavaObject, str]:
         """If not already created, create a preparedScript from our DMLCode, pass python local data to our prepared
@@ -104,9 +110,13 @@ def execute_with_lineage(self) -> Tuple[JavaObject, str]:
                     traces.append(self.prepared_script.getLineageTrace(output))
                 return ret, traces
 
+        except Py4JNetworkError:
+            exception_str = "Py4JNetworkError: no connection to JVM, most likely due to previous crash or closed JVM from calls to close()"
+            trace_back_limit = 0
         except Exception as e:
-            self.sds_context.exception_and_close(e)
-            return None, None
+            exception_str = str(e)
+            trace_back_limit = None
+        self.sds_context.exception_and_close(exception_str, trace_back_limit)
 
     def __prepare_script(self):
         gateway = self.sds_context.java_gateway
@@ -190,15 +200,13 @@ def _dfs_dag_nodes(self, dag_node: VALID_INPUT_TYPES) -> str:
         # for each node do the dfs operation and save the variable names in `input_var_names`
         # get variable names of unnamed parameters
 
-        unnamed_input_vars = [self._dfs_dag_nodes(
-            input_node) for input_node in dag_node.unnamed_input_nodes]
+        unnamed_input_vars = []
+        for un_node in dag_node.unnamed_input_nodes:
+            unnamed_input_vars.append(self._dfs_dag_nodes(un_node))
 
         named_input_vars = {}
         for name, input_node in dag_node.named_input_nodes.items():
             named_input_vars[name] = self._dfs_dag_nodes(input_node)
-            if isinstance(input_node, DAGNode) and input_node._output_type == OutputType.LIST:
-                dag_node.dml_name = named_input_vars[name] + name
-                return dag_node.dml_name
 
         # check if the node gets a name after multireturns
         # If it has, great, return that name
@@ -212,6 +220,7 @@ def _dfs_dag_nodes(self, dag_node: VALID_INPUT_TYPES) -> str:
 
         code_line = dag_node.code_line(
             dag_node.dml_name, unnamed_input_vars, named_input_vars)
+
         self.add_code(code_line)
         return dag_node.dml_name
 

src/main/python/tests/matrix/test_print.py

@@ -22,6 +22,7 @@
 import unittest
 
 import numpy as np
+from time import sleep
 from systemds.context import SystemDSContext
 
 
@@ -32,6 +33,9 @@ class TestPrint(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.sds = SystemDSContext()
+        sleep(1.0)
+        cls.sds.get_stdout()
+        cls.sds.get_stdout()
 
     @classmethod
     def tearDownClass(cls):

src/main/python/tests/script/test_dml_script.py

@@ -20,7 +20,7 @@
 # -------------------------------------------------------------
 
 import unittest
-import time
+from time import sleep
 
 from systemds.context import SystemDSContext
 from systemds.script_building import DMLScript
@@ -35,6 +35,9 @@ class Test_DMLScript(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         cls.sds = SystemDSContext()
+        sleep(1.0)
+        cls.sds.get_stdout()
+        cls.sds.get_stdout()
 
     @classmethod
     def tearDownClass(cls):
@@ -44,7 +47,7 @@ def test_simple_print_1(self):
         script = DMLScript(self.sds)
         script.add_code('print("Hello")')
         script.execute()
-        time.sleep(0.5)
+        sleep(0.5)
         stdout = self.sds.get_stdout(100)
         self.assertListEqual(["Hello"], stdout)
 
@@ -54,7 +57,7 @@ def test_simple_print_2(self):
         script.add_code('print("World")')
         script.add_code('print("!")')
         script.execute()
-        time.sleep(0.5)
+        sleep(0.5)
         stdout = self.sds.get_stdout(100)
         self.assertListEqual(['Hello', 'World', '!'], stdout)
 
@@ -65,7 +68,7 @@ def test_multiple_executions_1(self):
         scr_a.add_code('y = x + 1')
         scr_a.add_code('print(y)')
         scr_a.execute()
-        time.sleep(0.5)
+        sleep(0.5)
         stdout = self.sds.get_stdout(100)
         self.assertEqual("4", stdout[0])
         self.assertEqual("5", stdout[1])

src/main/python/tests/source/source_with_list_input.dml

@@ -0,0 +1,29 @@
+#-------------------------------------------------------------
+#
+# 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.
+#
+#-------------------------------------------------------------
+
+func = function(list[unknown] a) return (matrix[double] b){
+    b = as.matrix(a[1])
+}
+
+func2 = function(list[unknown] a) return (matrix[double] b, matrix[double] c){
+    b = as.matrix(a[1])
+    c = as.matrix(a[2])
+}

src/main/python/tests/source/test_source_list.py

@@ -0,0 +1,63 @@
+# -------------------------------------------------------------
+#
+# 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.
+#
+# -------------------------------------------------------------
+
+import unittest
+
+import numpy as np
+from systemds.context import SystemDSContext
+from systemds.operator.algorithm.builtin.scale import scale
+
+
+class TestSource_01(unittest.TestCase):
+
+    sds: SystemDSContext = None
+    source_path: str = "./tests/source/source_with_list_input.dml"
+
+    @classmethod
+    def setUpClass(cls):
+        cls.sds = SystemDSContext()
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.sds.close()
+
+    def test_single_return(self):
+        arr = self.sds.array(self.sds.full((10, 10), 4))
+        c = self.sds.source(self.source_path, "test").func(arr)
+        res = c.sum().compute()
+        self.assertTrue(res == 10*10*4)
+
+    def test_input_multireturn(self):
+        m = self.sds.full((10, 10), 2)
+        [a, b, c] = scale(m, True, True)
+        arr = self.sds.array(a, b, c)
+        c = self.sds.source(self.source_path, "test").func(arr)
+        res = c.sum().compute()
+        self.assertTrue(res == 0)
+
+    # [SYSTEMDS-3224] https://issues.apache.org/jira/browse/SYSTEMDS-3224
+    # def test_multi_return(self):
+    #     arr = self.sds.array(
+    #         self.sds.full((10, 10), 4),
+    #         self.sds.full((3, 3), 5))
+    #     [b, c] = self.sds.source(self.source_path, "test", True).func2(arr)
+    #     res = c.sum().compute()
+    #     self.assertTrue(res == 10*10*4)