#30: Correction for variadic inputs (#32)

* Corrected variadic input access

* Removed DeprecationWarning

* Added validations

* Called getattr directly

* Updated  naming

Author: umitbuyuksahin

exasol_udf_mock_python/mock_context_run_wrapper.py

@@ -9,11 +9,12 @@ def _disallowed_function(*args, **kw):
 
 class MockContextRunWrapper:
 
-    def __init__(
-            self, mock_context: MockContext, input_type: str, output_type: str):
+    def __init__(self, mock_context: MockContext, input_type: str,
+                 output_type: str, is_variadic_input: bool):
         self._output_type = output_type
         self._input_type = input_type
         self._mock_context = mock_context
+        self._is_variadic_input = is_variadic_input
         if self._output_type == "RETURNS":
             self.emit = _disallowed_function
         else:
@@ -28,7 +29,30 @@ def __init__(
             self.size = self._mock_context.size
 
     def __getattr__(self, name):
+        """
+        Variadic UDFs' columns are only integer values. Since integers are not
+        valid identifier in python, this method cannot be used by variadic UDFs.
+        """
+        if self._is_variadic_input:
+            raise RuntimeError(f"E-UDF-CL-SL-PYTHON-1085: Iterator has no "
+                               f"object with name '{name}'")
         return self._mock_context.__getattr__(name)
 
     def __getitem__(self, item):
-        return self._mock_context._data[item]
+        """
+        Variadic UDFs can retrieve items by index. The index value can be given
+        as an integer (e.g. ctx[1]) or as a string integer (e.g. ctx["1"]).
+
+        Non-variadic UDFs can retrieve items by index, that can be either an
+        integer (e.g. ctx[1]) or a column name (e.g. ctx["col_name"]). They do
+        not accept string integers as index.
+        """
+        item = int(item) if self._is_variadic_input else item
+        if isinstance(item, int):
+            return self._mock_context._data[item]
+        else:
+            try:
+                return self._mock_context.__getattr__(item)
+            except KeyError:
+                raise RuntimeError(f"E-UDF-CL-SL-PYTHON-1082: Column with name "
+                                   f"'{item}' does not exist")

exasol_udf_mock_python/mock_meta_data.py

@@ -1,9 +1,7 @@
 import re
+import dill
 import textwrap
 from typing import List
-
-import dill
-
 from exasol_udf_mock_python.column import Column
 
 
@@ -15,6 +13,7 @@ def __init__(
             input_type: str,
             output_columns: List[Column],
             output_type: str,
+            is_variadic_input: bool = False,
             script_name: str="TEST_UDF",
             script_schema: str="TEST_SCHEMA",
             current_user: str="sys",
@@ -30,6 +29,13 @@ def __init__(
             statement_id: int="123456789",
             memory_limit: int=4*1073741824,
             ):
+
+        assert input_type.upper() in ["SET", "SCALAR"]
+        assert output_type.upper() in ["EMITS", "RETURNS"]
+        if is_variadic_input:
+            for i in range(len(input_columns)):
+                assert str(i+1) == str(input_columns[i].name)
+
         self._script_language = "PYTHON3"
         self._script_name = script_name
         self._script_schema = script_schema
@@ -54,6 +60,7 @@ def __init__(
         self._validate_column_defintions(output_columns)
         self._output_column_count = len(output_columns)
         self._output_columns = output_columns
+        self._is_variadic_input = is_variadic_input
 
     def _extract_script_code(self, script_code_wrapper_function):
         function_code = textwrap.dedent(dill.source.getsource(script_code_wrapper_function))
@@ -173,5 +180,9 @@ def output_column_count(self):
     def output_columns(self):
         return self._output_columns
 
+    @property
+    def is_variadic_input(self):
+        return self._is_variadic_input
+
     def __repr__(self):
-        return str(self.__class__) + ": " + str(self.__dict__)
+        return str(self.__class__) + ": " + str(self.__dict__)

exasol_udf_mock_python/udf_mock_executor.py

@@ -1,17 +1,19 @@
 from multiprocessing import Lock
 from typing import Dict, Any, Iterator, List, Union, Callable, Tuple
-
 from exasol_udf_mock_python.group import Group
 from exasol_udf_mock_python.mock_context import MockContext
 from exasol_udf_mock_python.mock_context_run_wrapper import MockContextRunWrapper
 from exasol_udf_mock_python.mock_exa_environment import MockExaEnvironment
 
+
 def _loop_groups(ctx:MockContext, exa:MockExaEnvironment, runfunc:Callable):
     while ctx._next_group():
         _wrapped_run(ctx, exa, runfunc)
 
-def _wrapped_run(ctx:MockContext, exa:MockExaEnvironment, runfunc:Callable):
-    wrapped_ctx = MockContextRunWrapper(ctx, exa.meta.input_type, exa.meta.output_type)
+
+def _wrapped_run(ctx:MockContext, exa: MockExaEnvironment, runfunc: Callable):
+    wrapped_ctx = MockContextRunWrapper(
+        ctx, exa.meta.input_type, exa.meta.output_type, exa.meta.is_variadic_input)
     if exa.meta.input_type == "SET":
         if exa.meta.output_type == "RETURNS":
             run_with_returns(ctx, runfunc, wrapped_ctx)

tests/test_executor_context_set_emits.py

@@ -1,5 +1,5 @@
 import pytest
-
+import unittest
 from exasol_udf_mock_python.column import Column
 from exasol_udf_mock_python.group import Group
 from exasol_udf_mock_python.mock_exa_environment import MockExaEnvironment
@@ -366,26 +366,59 @@ def run(ctx):
         result = executor.run([Group([(1,), (2,), (3,), (4,), (5,), (6,)])], exa)
 
 
-def test_context_parameters():
+def test_access_variadic_inputs():
     def udf_wrapper():
         def run(ctx):
-            ctx.emit(ctx[0], ctx.t1)
-            ctx.emit(ctx[1], ctx.t2)
-            ctx.emit(ctx[2], ctx.t3)
+            ctx.emit(ctx[0])
+            ctx.emit(ctx[1])
+            ctx.emit(ctx["2"])
+
+    input_columns = [Column("1", int, "INTEGER"),
+                     Column("2", int, "INTEGER"),
+                     Column("3", int, "INTEGER")]
+    output_columns = [Column("o1", int, "INTEGER")]
+    meta = MockMetaData(
+        script_code_wrapper_function=udf_wrapper,
+        input_type="SET",
+        input_columns=input_columns,
+        output_type="EMITS",
+        output_columns=output_columns,
+        is_variadic_input=True)
+
+    input_data = [(1, 2, 3), (4, 5, 6)]
+    exa = MockExaEnvironment(meta)
+    executor = UDFMockExecutor()
+    result = executor.run([Group(input_data)], exa)
+    for i, group in enumerate(result):
+        result_row = group.rows
+        assert len(result_row) == len(input_columns)
+        for j in range(len(result_row)):
+            assert len(result_row[j]) == len(output_columns)
+            assert input_data[i][j] == result_row[j][0]
+
+
+def test_access_non_variadic_inputs():
+    def udf_wrapper():
+        def run(ctx):
+            ctx.emit(ctx[0])
+            ctx.emit(ctx[1])
+            ctx.emit(ctx["t3"])
+            ctx.emit(ctx["4"])
 
     input_columns = [Column("t1", int, "INTEGER"),
                      Column("t2", int, "INTEGER"),
-                     Column("t3", int, "INTEGER")]
-    output_columns = [Column("o1", int, "INTEGER"),
-                      Column("o2", int, "INTEGER")]
+                     Column("t3", int, "INTEGER"),
+                     Column("4", int, "INTEGER")]
+    output_columns = [Column("o1", int, "INTEGER")]
     meta = MockMetaData(
         script_code_wrapper_function=udf_wrapper,
         input_type="SET",
         input_columns=input_columns,
         output_type="EMITS",
-        output_columns=output_columns
-    )
-    input_data = [(1, 2, 3), (4, 5, 6)]
+        output_columns=output_columns,
+        is_variadic_input=False)
+
+    input_data = [(1, 2, 3, 4), (5, 6, 7, 8)]
     exa = MockExaEnvironment(meta)
     executor = UDFMockExecutor()
     result = executor.run([Group(input_data)], exa)
@@ -394,4 +427,112 @@ def run(ctx):
         assert len(result_row) == len(input_columns)
         for j in range(len(result_row)):
             assert len(result_row[j]) == len(output_columns)
-            assert input_data[i][j] == result_row[j][0] == result_row[j][1]
+            assert input_data[i][j] == result_row[j][0]
+
+
+class InvalidTestsForVariadicInputs(unittest.TestCase):
+    def test_access_variadic_inputs_by_name(self):
+        def udf_wrapper():
+            def run(ctx):
+                ctx.emit(ctx.t1)
+
+        input_columns = [Column("1", int, "INTEGER")]
+        output_columns = [Column("o1", int, "INTEGER")]
+        meta = MockMetaData(
+            script_code_wrapper_function=udf_wrapper,
+            input_type="SET",
+            input_columns=input_columns,
+            output_type="EMITS",
+            output_columns=output_columns,
+            is_variadic_input=True)
+
+        input_data = [(1,)]
+        exa = MockExaEnvironment(meta)
+        executor = UDFMockExecutor()
+        with self.assertRaises(RuntimeError):
+            result = executor.run([Group(input_data)], exa)
+
+    def test_invalid_variadic_input_columns_name(self):
+        def udf_wrapper():
+            def run(ctx):
+                ctx.emit(ctx[0])
+
+        invalid_input_columns_list = [
+            [Column("t1", int, "INTEGER")],
+            [Column("1", int, "INTEGER"), Column("3", int, "INTEGER")]
+        ]
+        for input_columns in invalid_input_columns_list:
+            output_columns = [Column("o1", int, "INTEGER")]
+            with self.assertRaises(AssertionError):
+                meta = MockMetaData(
+                    script_code_wrapper_function=udf_wrapper,
+                    input_type="SET",
+                    input_columns=input_columns,
+                    output_type="EMITS",
+                    output_columns=output_columns,
+                    is_variadic_input=True)
+
+                input_data = [(1,)]
+                exa = MockExaEnvironment(meta)
+                executor = UDFMockExecutor()
+                result = executor.run([Group(input_data)], exa)
+
+
+class InvalidTestsForNonVariadicInputs(unittest.TestCase):
+    def test_invalid_access_to_inputs(self):
+        def udf_wrapper():
+            def run(ctx):
+                ctx.emit(ctx["1"])
+
+        input_columns = [Column("t1", int, "INTEGER")]
+        output_columns = [Column("o1", int, "INTEGER")]
+
+        with self.assertRaises(RuntimeError):
+            meta = MockMetaData(
+                script_code_wrapper_function=udf_wrapper,
+                input_type="SET",
+                input_columns=input_columns,
+                output_type="EMITS",
+                output_columns=output_columns,
+                is_variadic_input=False)
+
+            input_data = [(1,)]
+            exa = MockExaEnvironment(meta)
+            executor = UDFMockExecutor()
+            result = executor.run([Group(input_data)], exa)
+
+
+class InvalidTestsForUDFTypes(unittest.TestCase):
+    def test_invalid_input_types(self):
+        def udf_wrapper():
+            def run(ctx):
+                ctx.emit(ctx.t1)
+
+        input_columns = [Column("1", int, "INTEGER")]
+        output_columns = [Column("o1", int, "INTEGER")]
+        for input_type in ["SETS", "SCALARS", "INVALID"]:
+            with self.assertRaises(AssertionError):
+                MockMetaData(
+                    script_code_wrapper_function=udf_wrapper,
+                    input_type=input_type,
+                    input_columns=input_columns,
+                    output_type="EMITS",
+                    output_columns=output_columns,
+                    is_variadic_input=True)
+
+    def test_invalid_output_types(self):
+        def udf_wrapper():
+            def run(ctx):
+                ctx.emit(ctx.t1)
+
+        input_columns = [Column("1", int, "INTEGER")]
+        output_columns = [Column("o1", int, "INTEGER")]
+        for output_types in ["EMIT", "RETURN", "INVALID"]:
+            with self.assertRaises(AssertionError):
+                MockMetaData(
+                    script_code_wrapper_function=udf_wrapper,
+                    input_type="SET",
+                    input_columns=input_columns,
+                    output_type=output_types,
+                    output_columns=output_columns,
+                    is_variadic_input=True)

tests/test_group.py

@@ -1,4 +1,4 @@
-from collections import Iterable
+from collections.abc import Iterable
 
 from exasol_udf_mock_python.group import Group, IterableWithSize