#42 2 integrate extract validator into language container deployer (#46)

ckunki · tkilias · web-flow · commit c6c27ea014d8 · 2024-08-14T13:06:02.000+02:00
* Integrated ExtractValidator into LanguageContainerDeployer * Added file language_container_validator.py to provide backwards compatibility * Fix until manifest is deployed with standard SLCs see #48 * Fixed SQL syntax error * Fixed integration test * Fixed indention for UDF python script Co-authored-by: Torsten Kilias <tkilias@users.noreply.github.com>
diff --git a/exasol/python_extension_common/connections/pyexasol_connection.py b/exasol/python_extension_common/connections/pyexasol_connection.py
@@ -53,7 +53,7 @@ def open_pyexasol_connection(
                                                            pat=saas_token)
     else:
         raise ValueError('Incomplete parameter list. '
-                         'Please either provide the parameters [dns, db_user, db_pass] '
+                         'Please either provide the parameters [dsn, db_user, db_pass] '
                          'for an On-Prem database or [saas_url, saas_account_id, '
                          'saas_database_id or saas_database_name, saas_token] '
                          'for a SaaS database.')
diff --git a/exasol/python_extension_common/deployment/extract_validator.py b/exasol/python_extension_common/deployment/extract_validator.py
@@ -3,15 +3,12 @@
 import pyexasol     # type: ignore
 
 from datetime import datetime, timedelta
+from textwrap import dedent
 from typing import Callable, List
 from tenacity import Retrying
 from tenacity.wait import wait_fixed
 from tenacity.stop import stop_after_delay
 
-from exasol.python_extension_common.deployment.language_container_validator import (
-    temp_schema
-)
-
 MANIFEST_FILE = "exasol-manifest.json"
 
 
@@ -74,8 +71,7 @@ def _create_manifest_udf(self, language_alias: str, udf_name: str):
         Much more a later statement "CREATE SCRIPT" will fail with an error
         message. Hence we need to use a retry here, as well.
         """
-        self._pyexasol_conn.execute(
-            f"""
+        self._pyexasol_conn.execute(dedent(f"""
             CREATE OR REPLACE {language_alias} SET SCRIPT
                 {udf_name}(my_path VARCHAR(256))
                 EMITS (node INTEGER, manifest BOOL) AS
@@ -84,7 +80,7 @@ def run(ctx):
                 ctx.emit(exa.meta.node_id, os.path.isfile(ctx.my_path))
             /
             """
-        )
+        ))
 
     def _check_all_nodes_with_retry(self, udf_name: str, nproc: int, manifest: str, timeout: timedelta):
         for attempt in Retrying(
@@ -97,7 +93,7 @@ def _check_all_nodes_with_retry(self, udf_name: str, nproc: int, manifest: str,
     def _check_all_nodes(self, udf_name: str, nproc: int, manifest: str):
         result = self._pyexasol_conn.execute(
             f"""
-            SELECT {udf_name}({manifest})
+            SELECT {udf_name}('{manifest}')
             FROM VALUES BETWEEN 1 AND {nproc} t(i) GROUP BY i
             """
         ).fetchall()
@@ -117,8 +113,9 @@ def verify_all_nodes(self, schema: str, language_alias: str, bfs_archive_path: b
         still nodes pending, for which the extraction could not be verified,
         yet.
         """
-        manifest = f"{bfs_archive_path.as_udf_path()}/{MANIFEST_FILE}"
-        nproc = self._pyexasol_conn.execute("SELECT nproc()").fetchone()
+        # manifest = f"{bfs_archive_path.as_udf_path()}/{MANIFEST_FILE}"
+        manifest = "/exaudf/exaudfclient_py3"
+        nproc = self._pyexasol_conn.execute("SELECT nproc()").fetchone()[0]
         udf_name = _udf_name(schema, language_alias)
         start = datetime.now()
         try:
diff --git a/exasol/python_extension_common/deployment/language_container_deployer.py b/exasol/python_extension_common/deployment/language_container_deployer.py
@@ -2,6 +2,7 @@
 from textwrap import dedent
 from typing import List, Optional, Dict
 from pathlib import Path, PurePosixPath
+from datetime import timedelta
 import logging
 import tempfile
 import ssl
@@ -10,9 +11,8 @@
 import exasol.bucketfs as bfs   # type: ignore
 from exasol.saas.client.api_access import (get_connection_params, get_database_id)      # type: ignore
 
-from exasol.python_extension_common.deployment.language_container_validator import (
-    wait_language_container, temp_schema
-)
+from exasol.python_extension_common.deployment.temp_schema import temp_schema
+from exasol.python_extension_common.deployment.extract_validator import ExtractValidator
 
 
 logger = logging.getLogger(__name__)
@@ -92,11 +92,21 @@ class LanguageContainerDeployer:
     def __init__(self,
                  pyexasol_connection: pyexasol.ExaConnection,
                  language_alias: str,
-                 bucketfs_path: bfs.path.PathLike) -> None:
+                 bucketfs_path: bfs.path.PathLike,
+                 extract_validator: ExtractValidator|None = None,
+                 ) -> None:
 
         self._bucketfs_path = bucketfs_path
         self._language_alias = language_alias
         self._pyexasol_conn = pyexasol_connection
+        if extract_validator:
+            self._extract_validator = extract_validator
+        else:
+            self._extract_validator = ExtractValidator(
+                pyexasol_connection,
+                timeout=timedelta(minutes=5),
+                interval=timedelta(seconds=10),
+            )
         logger.debug("Init %s", LanguageContainerDeployer.__name__)
 
     def download_and_run(self, url: str,
@@ -124,6 +134,9 @@ def download_and_run(self, url: str,
             self.run(Path(tmp_file.name), bucket_file_path, alter_system, allow_override,
                      wait_for_completion)
 
+    def _upload_path(self, bucket_file_path: str|None) -> bfs.path.PathLike:
+        return self._bucketfs_path / bucket_file_path
+
     def run(self, container_file: Optional[Path] = None,
             bucket_file_path: Optional[str] = None,
             alter_system: bool = True,
@@ -162,10 +175,12 @@ def run(self, container_file: Optional[Path] = None,
         self.activate_container(bucket_file_path, LanguageActivationLevel.Session,
                                 allow_override)
 
-        # Maybe wait until the container becomes operational.
+        # Optionally wait until the container is extracted on all nodes of the
+        # database cluster.
         if container_file and wait_for_completion:
             with temp_schema(self._pyexasol_conn) as schema:
-                wait_language_container(self._pyexasol_conn, self._language_alias, schema)
+                self._extract_validator.verify_all_nodes(
+                    schema, self._language_alias, self._upload_path(bucket_file_path))
 
         if not alter_system:
             message = dedent(f"""
@@ -180,8 +195,7 @@ def run(self, container_file: Optional[Path] = None,
             """)
             print(message)
 
-    def upload_container(self, container_file: Path,
-                         bucket_file_path: Optional[str] = None) -> None:
+    def upload_container(self, container_file: Path, bucket_file_path: Optional[str] = None) -> None:
         """
         Upload the language container to the BucketFS.
 
@@ -192,8 +206,7 @@ def upload_container(self, container_file: Path,
             raise RuntimeError(f"Container file {container_file} "
                                f"is not a file.")
         with open(container_file, "br") as f:
-            file_path = self._bucketfs_path / bucket_file_path
-            file_path.write(f)
+            self._upload_path(bucket_file_path).write(f)
         logging.debug("Container is uploaded to bucketfs")
 
     def activate_container(self, bucket_file_path: str,
@@ -333,7 +346,7 @@ def create(cls,
                                                 path=path_in_bucket)
         else:
             raise ValueError('Incomplete parameter list. '
-                             'Please either provide the parameters [dns, db_user, '
+                             'Please either provide the parameters [dsn, db_user, '
                              'db_password, bucketfs_host, bucketfs_port, bucketfs_name, '
                              'bucket, bucketfs_user, bucketfs_password] for an On-Prem '
                              'database or [saas_url, saas_account_id, saas_database_id, '
diff --git a/exasol/python_extension_common/deployment/language_container_validator.py b/exasol/python_extension_common/deployment/language_container_validator.py
@@ -1,149 +1,4 @@
-from __future__ import annotations
-from typing import Generator
-from datetime import timedelta
-import random
-import string
-from contextlib import contextmanager
-from textwrap import dedent
+# This file is only for providing backwards compatibility.
+# As function temp_schema() has been moved to file temp_schema.py.
 
-from tenacity import retry
-from tenacity.wait import wait_fixed
-from tenacity.stop import stop_after_delay, stop_after_attempt
-
-import pyexasol     # type: ignore
-
-_DUMMY_UDF_NAME = 'DUMMY_UDF'
-
-
-def _get_test_udf_name(schema: str | None) -> str:
-
-    if schema:
-        return f'"{schema}"."{_DUMMY_UDF_NAME}"'
-    return f'"{_DUMMY_UDF_NAME}"'
-
-
-def _create_dummy_udf(conn: pyexasol.ExaConnection, language_alias: str,
-                      schema: str | None) -> None:
-
-    # The dummy UDF returns the ID of the node it is running at.
-    udf_name = _get_test_udf_name(schema)
-    sql = dedent(f"""
-    CREATE OR REPLACE {language_alias} SET SCRIPT {udf_name}(i DECIMAL(10, 0))
-    RETURNS DECIMAL(10, 0) AS
-
-    def run(ctx):
-        return exa.meta.node_id
-    /
-    """)
-    conn.execute(sql)
-
-
-def _call_dummy_udf(conn: pyexasol.ExaConnection, schema: str | None) -> None:
-
-    # First, need to find out the number of nodes.
-    sql = "SELECT NPROC();"
-    nproc = conn.execute(sql).fetchval()
-
-    # This query should run the dummy udf on all nodes.
-    udf_name = _get_test_udf_name(schema)
-    sql = dedent(f"""
-    SELECT {udf_name}(i) FROM VALUES BETWEEN 1 AND {nproc} t(i)
-    GROUP BY i;
-    """)
-    # The expected result is a collection of distinct node IDs from 0 to nproc - 1.
-    result = conn.execute(sql).fetchall()
-    set_result = {row[0] for row in result}
-
-    assert set_result == set(range(nproc))
-
-
-def _delete_dummy_udf(conn: pyexasol.ExaConnection, schema: str | None) -> None:
-
-    udf_name = _get_test_udf_name(schema)
-    sql = dedent(f"""
-    DROP SCRIPT IF EXISTS {udf_name};
-    """)
-    conn.execute(sql)
-
-
-@retry(reraise=True, stop=stop_after_attempt(3))
-def _create_random_schema(conn: pyexasol.ExaConnection, schema_name_length: int) -> str:
-
-    schema = ''.join(random.choice(string.ascii_letters)
-                     for _ in range(schema_name_length))
-    sql = f'CREATE SCHEMA "{schema}";'
-    conn.execute(query=sql)
-    return schema
-
-
-def _delete_schema(conn: pyexasol.ExaConnection, schema: str) -> None:
-
-    sql = f'DROP SCHEMA IF EXISTS "{schema}" CASCADE;'
-    conn.execute(query=sql)
-
-
-def validate_language_container(conn: pyexasol.ExaConnection,
-                                language_alias: str,
-                                schema: str | None = None
-                                ) -> None:
-    """
-    Runs a test to check if a language container has been installed and is now
-    operational. Will raise an exception if this is not the case.
-
-    conn            - pyexasol connection. The language container must be activated either
-                    at the SYSTEM level or at the SESSION associated with this connection.
-    language_alias  - Language alias of the language container.
-    schema          - The schema to run the tests in. If not specified the current schema
-                    is assumed.
-    """
-    try:
-        _create_dummy_udf(conn, language_alias, schema)
-        _call_dummy_udf(conn, schema)
-    finally:
-        _delete_dummy_udf(conn, schema)
-
-
-def wait_language_container(conn: pyexasol.ExaConnection,
-                            language_alias: str,
-                            schema: str | None = None,
-                            timeout: timedelta = timedelta(minutes=5),
-                            interval: timedelta = timedelta(seconds=5),
-                            ) -> None:
-    """
-    Keeps calling validate_language_container until it succeeds or the timeout expires.
-
-    conn            - pyexasol connection. The language container must be activated either
-                    at the SYSTEM level or at the SESSION associated with this connection.
-    language_alias  - Language alias of the language container.
-    schema          - The schema to run the tests in. If not specified the current schema
-                    is assumed.
-    timeout         - Will give up after this timeout expires. The last exception thrown
-                    by the validate_language_container will be re-raised.
-    interval        - The calls to validate_language_container are spaced by this time
-                    interval.
-    """
-    @retry(reraise=True, wait=wait_fixed(interval), stop=stop_after_delay(timeout))
-    def repeat_validate_language_container():
-        validate_language_container(conn, language_alias, schema)
-
-    repeat_validate_language_container()
-
-
-@contextmanager
-def temp_schema(conn: pyexasol.ExaConnection,
-                schema_name_length: int = 20
-                ) -> Generator[str, None, None]:
-    """
-    A context manager for running an operation in a newly created temporary schema.
-    The schema will be deleted after the operation is competed. Note, that all objects
-    created in this schema will be deleted with it. Returns the name of the created schema.
-
-    conn                - pyexasol connection.
-    schema_name_length  - Number of characters in the temporary schema name.
-    """
-    schema = ''
-    try:
-        schema = _create_random_schema(conn, schema_name_length)
-        yield schema
-    finally:
-        _delete_schema(conn, schema)
+from exasol.python_extension_common.deployment.temp_schema import temp_schema
diff --git a/exasol/python_extension_common/deployment/temp_schema.py b/exasol/python_extension_common/deployment/temp_schema.py
@@ -0,0 +1,41 @@
+import pyexasol     # type: ignore
+import random
+import string
+from contextlib import contextmanager
+
+from typing import Generator
+from tenacity import retry
+from tenacity.stop import stop_after_attempt
+
+@retry(reraise=True, stop=stop_after_attempt(3))
+def _create_random_schema(conn: pyexasol.ExaConnection, schema_name_length: int) -> str:
+
+    schema = ''.join(random.choice(string.ascii_letters)
+                     for _ in range(schema_name_length))
+    sql = f'CREATE SCHEMA "{schema}";'
+    conn.execute(query=sql)
+    return schema
+
+
+def _delete_schema(conn: pyexasol.ExaConnection, schema: str) -> None:
+    sql = f'DROP SCHEMA IF EXISTS "{schema}" CASCADE;'
+
+
+@contextmanager
+def temp_schema(conn: pyexasol.ExaConnection,
+                schema_name_length: int = 20
+                ) -> Generator[str, None, None]:
+    """
+    A context manager for running an operation in a newly created temporary schema.
+    The schema will be deleted after the operation is competed. Note, that all objects
+    created in this schema will be deleted with it. Returns the name of the created schema.
+
+    conn                - pyexasol connection.
+    schema_name_length  - Number of characters in the temporary schema name.
+    """
+    schema = ''
+    try:
+        schema = _create_random_schema(conn, schema_name_length)
+        yield schema
+    finally:
+        _delete_schema(conn, schema)
diff --git a/test/unit/deployment/test_extract_validator.py b/test/unit/deployment/test_extract_validator.py
@@ -48,7 +48,7 @@ def execute(self, *args, **kwargs):
         return self
 
     def fetchone(self):
-        return next(self.values)
+        return [ next(self.values) ]
 
     def fetchall(self):
         return [ v for v in self.values ]
diff --git a/test/unit/deployment/test_language_container_deployer.py b/test/unit/deployment/test_language_container_deployer.py
diff --git a/test/unit/deployment/test_language_container_validator.py b/test/unit/deployment/test_language_container_validator.py
