Compare temp branch with current (Do not merge)

exasol_advanced_analytics_framework/query_handler/context/connection_name.py

@@ -0,0 +1,22 @@
+from exasol_data_science_utils_python.schema.dbobject_name import DBObjectName
+from exasol_data_science_utils_python.schema.dbobject_name_impl import DBObjectNameImpl
+from typeguard import typechecked
+
+
+class ConnectionName(DBObjectName):
+    """A DBObjectName class which represents the name of a connection object"""
+
+    @typechecked
+    def __init__(self, connection_name: str):
+        super().__init__(connection_name.upper())
+
+
+class ConnectionNameImpl(DBObjectNameImpl, ConnectionName):
+
+    @property
+    def fully_qualified(self) -> str:
+        return self.quoted_name
+
+    @typechecked
+    def __init__(self, connection_name: str):
+        super().__init__(connection_name)

exasol_advanced_analytics_framework/query_handler/context/connection_name_proxy.py

@@ -0,0 +1,17 @@
+from exasol_advanced_analytics_framework.query_handler.context.connection_name import ConnectionName, ConnectionNameImpl
+from exasol_advanced_analytics_framework.query_handler.context.proxy.db_object_name_proxy import DBObjectNameProxy
+from exasol_advanced_analytics_framework.query_handler.query.drop_connection_query import DropConnectionQuery
+from exasol_advanced_analytics_framework.query_handler.query.query import Query
+
+
+class ConnectionNameProxy(DBObjectNameProxy[ConnectionName], ConnectionName):
+
+    @property
+    def fully_qualified(self) -> str:
+        return self.quoted_name
+
+    def get_cleanup_query(self) -> Query:
+        return DropConnectionQuery(self._db_object_name)
+
+    def __init__(self, connection_name: ConnectionName, global_counter_value: int):
+        super().__init__(connection_name, global_counter_value)

exasol_advanced_analytics_framework/query_handler/context/proxy/drop_udf_query.py

@@ -0,0 +1,17 @@
+from exasol_data_science_utils_python.schema.udf_name import UDFName
+
+from exasol_advanced_analytics_framework.query_handler.query.drop_query import DropQuery
+
+
+class DropUDFQuery(DropQuery):
+
+    def __init__(self, udf_name: UDFName):
+        self._udf_name = udf_name
+
+    @property
+    def query_string(self) -> str:
+        return f"DROP SCRIPT IF EXISTS {self._udf_name.fully_qualified};"
+
+    @property
+    def table_name(self) -> UDFName:
+        return self._udf_name

exasol_advanced_analytics_framework/query_handler/context/proxy/udf_name_proxy.py

@@ -0,0 +1,15 @@
+from exasol_data_science_utils_python.schema.udf_name import UDFName
+
+from exasol_advanced_analytics_framework.query_handler.context.proxy.db_object_name_with_schema_proxy import \
+    DBObjectNameWithSchemaProxy
+from exasol_advanced_analytics_framework.query_handler.context.proxy.drop_udf_query import DropUDFQuery
+from exasol_advanced_analytics_framework.query_handler.query.query import Query
+
+
+class UDFNameProxy(DBObjectNameWithSchemaProxy[UDFName], UDFName):
+
+    def get_cleanup_query(self) -> Query:
+        return DropUDFQuery(self._db_object_name)
+
+    def __init__(self, script_name: UDFName, global_counter_value: int):
+        super().__init__(script_name, global_counter_value)

exasol_advanced_analytics_framework/query_handler/context/query_handler_context.py

@@ -2,13 +2,23 @@
 from abc import ABC
 
 from exasol_data_science_utils_python.schema.table_name import TableName
+from exasol_data_science_utils_python.schema.udf_name import UDFName
 from exasol_data_science_utils_python.schema.view_name import ViewName
 
+from exasol_advanced_analytics_framework.query_handler.context.connection_name import ConnectionName
 from exasol_advanced_analytics_framework.query_handler.context.proxy.bucketfs_location_proxy import \
     BucketFSLocationProxy
 
 
 class QueryHandlerContext(ABC):
+
+    @abc.abstractmethod
+    def get_temporary_name(self) -> str:
+        """
+        Returns a temporary name
+        """
+        pass
+
     @abc.abstractmethod
     def get_temporary_table_name(self) -> TableName:
         """
@@ -26,11 +36,28 @@ def get_temporary_view_name(self) -> ViewName:
 
         pass
 
+    @abc.abstractmethod
+    def get_temporary_udf_name(self) -> UDFName:
+        """
+        This function registers a new temporary script without creating it.
+        After the release of this context the framework will issue a cleanup query.
+        """
+
+        pass
+
+    @abc.abstractmethod
+    def get_temporary_connection_name(self) -> ConnectionName:
+        """
+        This function registers a new temporary connection without creating it.
+        After the release of this context the framework will issue a cleanup query.
+        """
+
+        pass
+
     @abc.abstractmethod
     def get_temporary_bucketfs_location(self) -> BucketFSLocationProxy:
         """
         This function registers a new temporary bucketfs file without creating it.
         After the release of this context the framework will remove it.
         """
         pass
-

exasol_advanced_analytics_framework/query_handler/context/top_level_query_handler_context.py

@@ -7,14 +7,19 @@
 from exasol_data_science_utils_python.schema.schema_name import SchemaName
 from exasol_data_science_utils_python.schema.table_name import TableName
 from exasol_data_science_utils_python.schema.table_name_builder import TableNameBuilder
+from exasol_data_science_utils_python.schema.udf_name import UDFName
+from exasol_data_science_utils_python.schema.udf_name_builder import UDFNameBuilder
 from exasol_data_science_utils_python.schema.view_name import ViewName
 from exasol_data_science_utils_python.schema.view_name_builder import ViewNameBuilder
 
+from exasol_advanced_analytics_framework.query_handler.context.connection_name_proxy import ConnectionNameProxy
+from exasol_advanced_analytics_framework.query_handler.context.connection_name import ConnectionName, ConnectionNameImpl
 from exasol_advanced_analytics_framework.query_handler.context.proxy.bucketfs_location_proxy import \
     BucketFSLocationProxy
 from exasol_advanced_analytics_framework.query_handler.context.proxy.db_object_name_proxy import DBObjectNameProxy
 from exasol_advanced_analytics_framework.query_handler.context.proxy.object_proxy import ObjectProxy
 from exasol_advanced_analytics_framework.query_handler.context.proxy.table_name_proxy import TableNameProxy
+from exasol_advanced_analytics_framework.query_handler.context.proxy.udf_name_proxy import UDFNameProxy
 from exasol_advanced_analytics_framework.query_handler.context.proxy.view_name_proxy import ViewNameProxy
 from exasol_advanced_analytics_framework.query_handler.context.scope_query_handler_context import \
     ScopeQueryHandlerContext, Connection
@@ -110,6 +115,20 @@ def _get_temporary_view_name(self) -> ViewName:
             schema=SchemaName(schema_name=self._temporary_schema_name))
         return temporary_view_name
 
+    def _get_temporary_udf_name(self) -> UDFName:
+        self._check_if_released()
+        temporary_name = self._get_temporary_db_object_name()
+        temporary_script_name = UDFNameBuilder.create(
+            name=temporary_name,
+            schema=SchemaName(schema_name=self._temporary_schema_name))
+        return temporary_script_name
+
+    def _get_temporary_connection_name(self) -> ConnectionName:
+        self._check_if_released()
+        temporary_name = self._get_temporary_db_object_name()
+        temporary_connection_name = ConnectionNameImpl(connection_name=temporary_name)
+        return temporary_connection_name
+
     def _get_temporary_db_object_name(self) -> str:
         temporary_name = f"{self._temporary_db_object_name_prefix}_{self._get_counter_value()}"
         return temporary_name
@@ -118,6 +137,11 @@ def _own_object(self, object_proxy: ObjectProxy):
         self._register_object(object_proxy)
         self._owned_object_proxies.add(object_proxy)
 
+    def get_temporary_name(self) -> str:
+        self._check_if_released()
+        temporary_name = self._get_temporary_db_object_name()
+        return temporary_name
+
     def get_temporary_table_name(self) -> TableName:
         self._check_if_released()
         temporary_table_name = self._get_temporary_table_name()
@@ -134,6 +158,22 @@ def get_temporary_view_name(self) -> ViewName:
         self._own_object(object_proxy)
         return object_proxy
 
+    def get_temporary_udf_name(self) -> UDFName:
+        self._check_if_released()
+        temporary_script_name = self._get_temporary_udf_name()
+        object_proxy = UDFNameProxy(temporary_script_name,
+                                    self._global_temporary_object_counter.get_current_value())
+        self._own_object(object_proxy)
+        return object_proxy
+
+    def get_temporary_connection_name(self) -> ConnectionName:
+        self._check_if_released()
+        temporary_connection_name = self._get_temporary_connection_name()
+        object_proxy = ConnectionNameProxy(connection_name=temporary_connection_name,
+                                           global_counter_value=self._global_temporary_object_counter.get_current_value())
+        self._own_object(object_proxy)
+        return object_proxy
+
     def get_temporary_bucketfs_location(self) -> BucketFSLocationProxy:
         self._check_if_released()
         temporary_path = self._get_temporary_path()

exasol_advanced_analytics_framework/query_handler/query/drop_connection_query.py

@@ -0,0 +1,16 @@
+from exasol_advanced_analytics_framework.query_handler.context.connection_name import ConnectionName
+from exasol_advanced_analytics_framework.query_handler.query.drop_query import DropQuery
+
+
+class DropConnectionQuery(DropQuery):
+
+    def __init__(self, connection_name: ConnectionName):
+        self._connection_name = connection_name
+
+    @property
+    def query_string(self) -> str:
+        return f"DROP CONNECTION IF EXISTS {self._connection_name.fully_qualified};"
+
+    @property
+    def connection_name(self) -> ConnectionName:
+        return self._connection_name

exasol_advanced_analytics_framework/query_handler/query/drop_query.py

@@ -3,3 +3,7 @@
 
 class DropyQuery(Query):
     pass
+
+
+class DropQuery(Query):
+    pass

exasol_advanced_analytics_framework/query_handler/query/drop_table_query.py

@@ -1,8 +1,6 @@
 from exasol_data_science_utils_python.schema.table_name import TableName
-from exasol_advanced_analytics_framework.query_handler.query.query import Query
 
-class DropQuery(Query):
-    pass
+from exasol_advanced_analytics_framework.query_handler.query.drop_query import DropQuery
 
 
 class DropTableQuery(DropQuery):
@@ -15,5 +13,5 @@ def query_string(self) -> str:
         return f"DROP TABLE IF EXISTS {self._table_name.fully_qualified};"
 
     @property
-    def table_name(self)->TableName:
-        return self._table_name
+    def table_name(self) -> TableName:
+        return self._table_name

exasol_advanced_analytics_framework/query_handler/query/drop_view_query.py

@@ -1,7 +1,7 @@
 from exasol_data_science_utils_python.schema.table_name import TableName
 from exasol_data_science_utils_python.schema.view_name import ViewName
 
-from exasol_advanced_analytics_framework.query_handler.query.drop_table_query import DropQuery
+from exasol_advanced_analytics_framework.query_handler.query.drop_query import DropQuery
 
 
 class DropViewQuery(DropQuery):

exasol_advanced_analytics_framework/testing/mock_query_handler_runner.py

@@ -1,4 +1,6 @@
 import logging
+import textwrap
+from inspect import cleandoc
 from typing import Callable, TypeVar, Generic, Tuple, Union, List
 
 from exasol_data_science_utils_python.udf_utils.sql_executor import SQLExecutor
@@ -97,12 +99,18 @@ def release_and_create_query_handler_context_of_input_query(self):
 
     def _wrap_return_query(self, input_query: SelectQueryWithColumnDefinition) -> Tuple[str, str]:
         temporary_view_name = self._state.input_query_query_handler_context.get_temporary_view_name()
-        input_query_create_view_string = \
-            f"CREATE VIEW {temporary_view_name.fully_qualified} AS {input_query.query_string};"
+        input_query_create_view_string = cleandoc(
+            f"""
+CREATE OR REPLACE VIEW {temporary_view_name.fully_qualified} AS 
+{input_query.query_string};
+""")
         full_qualified_columns = [col.name.fully_qualified
                                   for col in input_query.output_columns]
-        columns_str = ",".join(full_qualified_columns)
-        input_query_string = \
-            f"SELECT {columns_str} " \
-            f"FROM {temporary_view_name.fully_qualified};"
+        columns_str = ",\n".join(full_qualified_columns)
+        input_query_string = cleandoc(
+            f"""
+SELECT 
+{textwrap.indent(columns_str, " " * 4)}
+FROM {temporary_view_name.fully_qualified};
+""")
         return input_query_create_view_string, input_query_string

exasol_advanced_analytics_framework/udf_communication/all_gather_operation.py

@@ -0,0 +1,27 @@
+from typing import List, Optional
+
+from pydantic import BaseModel
+
+from exasol_advanced_analytics_framework.udf_communication.communicator_protocol import CommunicatorProtocol
+from exasol_advanced_analytics_framework.udf_communication.serialization import serialize_message, deserialize_message
+
+
+class AllGatherResult(BaseModel):
+    gather_result: List[bytes]
+
+
+class AllGatherOperation:
+
+    def __init__(self, communicator: CommunicatorProtocol, value: bytes):
+        self._value = value
+        self._communicator = communicator
+
+    def __call__(self) -> List[bytes]:
+        gather_result = self._communicator.gather(self._value)
+        broadcast_value: Optional[bytes] = None
+        if self._communicator.is_multi_node_leader():
+            all_gather_result = AllGatherResult(gather_result=gather_result)
+            broadcast_value = serialize_message(all_gather_result)
+        broadcast_result = self._communicator.broadcast(broadcast_value)
+        all_gather_result = deserialize_message(broadcast_result, AllGatherResult)
+        return all_gather_result.gather_result