refactor(clp-package): Replace hard-coded CLP metadata table names with table_prefix from connection params and table suffix constants.

components/clp-package-utils/clp_package_utils/scripts/native/archive_manager.py

@@ -7,7 +7,12 @@
 from contextlib import closing
 from pathlib import Path
 
-from clp_py_utils.clp_config import Database
+from clp_py_utils.clp_config import (
+    ARCHIVE_TAGS_TABLE_SUFFIX,
+    ARCHIVES_TABLE_SUFFIX,
+    Database,
+    FILES_TABLE_SUFFIX,
+)
 from clp_py_utils.sql_adapter import SQL_Adapter
 
 from clp_package_utils.general import (
@@ -245,7 +250,12 @@ def _find_archives(
             db_conn.cursor(dictionary=True)
         ) as db_cursor:
             query_params: typing.List[int] = [begin_ts]
-            query: str = f"SELECT id FROM `{table_prefix}archives` WHERE begin_timestamp >= %s"
+            query: str = (
+                f"""
+                SELECT id FROM `{table_prefix}{ARCHIVES_TABLE_SUFFIX}`
+                WHERE begin_timestamp >= %s
+                """
+            )
             if end_ts is not None:
                 query += " AND end_timestamp <= %s"
                 query_params.append(end_ts)
@@ -308,7 +318,7 @@ def _delete_archives(
 
             db_cursor.execute(
                 f"""
-                DELETE FROM `{table_prefix}archives`
+                DELETE FROM `{table_prefix}{ARCHIVES_TABLE_SUFFIX}`
                 WHERE {query_criteria}
                 RETURNING id
                 """,
@@ -327,14 +337,14 @@ def _delete_archives(
 
             db_cursor.execute(
                 f"""
-                DELETE FROM `{table_prefix}files`
+                DELETE FROM `{table_prefix}{FILES_TABLE_SUFFIX}`
                 WHERE archive_id in ({ids_list_string})
                 """
             )
 
             db_cursor.execute(
                 f"""
-                DELETE FROM `{table_prefix}archive_tags`
+                DELETE FROM `{table_prefix}{ARCHIVE_TAGS_TABLE_SUFFIX}`
                 WHERE archive_id in ({ids_list_string})
                 """
             )

components/clp-package-utils/clp_package_utils/scripts/native/decompress.py

@@ -9,7 +9,11 @@
 from typing import Optional
 
 import yaml
-from clp_py_utils.clp_config import CLP_METADATA_TABLE_PREFIX, CLPConfig, Database
+from clp_py_utils.clp_config import (
+    CLPConfig,
+    Database,
+    FILES_TABLE_SUFFIX,
+)
 from clp_py_utils.sql_adapter import SQL_Adapter
 from job_orchestration.scheduler.constants import QueryJobStatus, QueryJobType
 from job_orchestration.scheduler.job_config import (
@@ -44,11 +48,13 @@ def get_orig_file_id(db_config: Database, path: str) -> Optional[str]:
     only one of them.
     """
     sql_adapter = SQL_Adapter(db_config)
+    clp_db_connection_params = db_config.get_clp_connection_params_and_type(True)
+    table_prefix = clp_db_connection_params["table_prefix"]
     with closing(sql_adapter.create_connection(True)) as db_conn, closing(
         db_conn.cursor(dictionary=True)
     ) as db_cursor:
         db_cursor.execute(
-            f"SELECT orig_file_id FROM `{CLP_METADATA_TABLE_PREFIX}files` WHERE path = (%s)",
+            f"SELECT orig_file_id FROM `{table_prefix}{FILES_TABLE_SUFFIX}` WHERE path = (%s)",
             (path,),
         )
         results = db_cursor.fetchall()

components/clp-package-utils/clp_package_utils/scripts/start_clp.py

@@ -14,14 +14,15 @@
 import yaml
 from clp_py_utils.clp_config import (
     ALL_TARGET_NAME,
+    ARCHIVES_TABLE_SUFFIX,
     AwsAuthType,
-    CLP_METADATA_TABLE_PREFIX,
     CLPConfig,
     COMPRESSION_JOBS_TABLE_NAME,
     COMPRESSION_SCHEDULER_COMPONENT_NAME,
     COMPRESSION_WORKER_COMPONENT_NAME,
     CONTROLLER_TARGET_NAME,
     DB_COMPONENT_NAME,
+    FILES_TABLE_SUFFIX,
     LOG_VIEWER_WEBUI_COMPONENT_NAME,
     QUERY_JOBS_TABLE_NAME,
     QUERY_SCHEDULER_COMPONENT_NAME,
@@ -861,13 +862,15 @@ def start_webui(instance_id: str, clp_config: CLPConfig, mounts: CLPDockerMounts
     container_webui_logs_dir = pathlib.Path("/") / "var" / "log" / component_name
 
     # Read and update settings.json
+    clp_db_connection_params = clp_config.database.get_clp_connection_params_and_type(True)
+    table_prefix = clp_db_connection_params["table_prefix"]
     meteor_settings_updates = {
         "private": {
             "SqlDbHost": clp_config.database.host,
             "SqlDbPort": clp_config.database.port,
             "SqlDbName": clp_config.database.name,
-            "SqlDbClpArchivesTableName": f"{CLP_METADATA_TABLE_PREFIX}archives",
-            "SqlDbClpFilesTableName": f"{CLP_METADATA_TABLE_PREFIX}files",
+            "SqlDbClpArchivesTableName": f"{table_prefix}{ARCHIVES_TABLE_SUFFIX}",
+            "SqlDbClpFilesTableName": f"{table_prefix}{FILES_TABLE_SUFFIX}",
             "SqlDbCompressionJobsTableName": COMPRESSION_JOBS_TABLE_NAME,
             "SqlDbQueryJobsTableName": QUERY_JOBS_TABLE_NAME,
         },

components/job-orchestration/job_orchestration/executor/compress/compression_task.py

@@ -10,6 +10,7 @@
 from celery.app.task import Task
 from celery.utils.log import get_task_logger
 from clp_py_utils.clp_config import (
+    ARCHIVE_TAGS_TABLE_SUFFIX,
     ARCHIVES_TABLE_SUFFIX,
     CLP_DEFAULT_DATASET_NAME,
     COMPRESSION_JOBS_TABLE_NAME,
@@ -70,7 +71,10 @@ def increment_compression_job_metadata(db_cursor, job_id, kv):
 
 def update_tags(db_cursor, table_prefix, archive_id, tag_ids):
     db_cursor.executemany(
-        f"INSERT INTO {table_prefix}archive_tags (archive_id, tag_id) VALUES (%s, %s)",
+        f"""
+        INSERT INTO {table_prefix}{ARCHIVE_TAGS_TABLE_SUFFIX} (archive_id, tag_id)
+        VALUES (%s, %s)
+        """,
         [(archive_id, tag_id) for tag_id in tag_ids],
     )
 

components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py

@@ -12,10 +12,10 @@
 import msgpack
 from clp_package_utils.general import CONTAINER_INPUT_LOGS_ROOT_DIR
 from clp_py_utils.clp_config import (
-    CLP_METADATA_TABLE_PREFIX,
     CLPConfig,
     COMPRESSION_JOBS_TABLE_NAME,
     COMPRESSION_TASKS_TABLE_NAME,
+    TAGS_TABLE_SUFFIX,
 )
 from clp_py_utils.clp_logging import get_logger, get_logging_formatter, set_logging_level
 from clp_py_utils.compression import validate_path_and_get_info
@@ -235,13 +235,14 @@ def search_and_schedule_new_tasks(db_conn, db_cursor, clp_metadata_db_connection
 
         tag_ids = None
         if clp_io_config.output.tags:
+            table_prefix = clp_metadata_db_connection_config["table_prefix"]
             db_cursor.executemany(
-                f"INSERT IGNORE INTO {CLP_METADATA_TABLE_PREFIX}tags (tag_name) VALUES (%s)",
+                f"INSERT IGNORE INTO {table_prefix}{TAGS_TABLE_SUFFIX} (tag_name) VALUES (%s)",
                 [(tag,) for tag in clp_io_config.output.tags],
             )
             db_conn.commit()
             db_cursor.execute(
-                f"SELECT tag_id FROM {CLP_METADATA_TABLE_PREFIX}tags WHERE tag_name IN (%s)"
+                f"SELECT tag_id FROM {table_prefix}{TAGS_TABLE_SUFFIX} WHERE tag_name IN (%s)"
                 % ", ".join(["%s"] * len(clp_io_config.output.tags)),
                 clp_io_config.output.tags,
             )

components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py

@@ -31,10 +31,13 @@
 import msgpack
 import pymongo
 from clp_py_utils.clp_config import (
-    CLP_METADATA_TABLE_PREFIX,
+    ARCHIVE_TAGS_TABLE_SUFFIX,
+    ARCHIVES_TABLE_SUFFIX,
     CLPConfig,
+    FILES_TABLE_SUFFIX,
     QUERY_JOBS_TABLE_NAME,
     QUERY_TASKS_TABLE_NAME,
+    TAGS_TABLE_SUFFIX,
 )
 from clp_py_utils.clp_logging import get_logger, get_logging_formatter, set_logging_level
 from clp_py_utils.core import read_yaml_config_file
@@ -104,11 +107,17 @@ def create_stream_extraction_job(self) -> QueryJob: ...
 
 
 class IrExtractionHandle(StreamExtractionHandle):
-    def __init__(self, job_id: str, job_config: Dict[str, Any], db_conn):
+    def __init__(
+        self,
+        job_id: str,
+        job_config: Dict[str, Any],
+        db_conn,
+        table_prefix: str,
+    ):
         super().__init__(job_id)
         self.__job_config = ExtractIrJobConfig.parse_obj(job_config)
         self._archive_id, self.__file_split_id = get_archive_and_file_split_ids_for_ir_extraction(
-            db_conn, self.__job_config
+            db_conn, table_prefix, self.__job_config
         )
         if self._archive_id is None:
             raise ValueError("Job parameters don't resolve to an existing archive")
@@ -145,11 +154,17 @@ def create_stream_extraction_job(self) -> QueryJob:
 
 
 class JsonExtractionHandle(StreamExtractionHandle):
-    def __init__(self, job_id: str, job_config: Dict[str, Any], db_conn):
+    def __init__(
+        self,
+        job_id: str,
+        job_config: Dict[str, Any],
+        db_conn,
+        table_prefix: str,
+    ):
         super().__init__(job_id)
         self.__job_config = ExtractJsonJobConfig.parse_obj(job_config)
         self._archive_id = self.__job_config.archive_id
-        if not archive_exists(db_conn, self._archive_id):
+        if not archive_exists(db_conn, table_prefix, self._archive_id):
             raise ValueError(f"Archive {self._archive_id} doesn't exist")
 
     def get_stream_id(self) -> str:
@@ -368,10 +383,11 @@ def insert_query_tasks_into_db(db_conn, job_id, archive_ids: List[str]) -> List[
 @exception_default_value(default=[])
 def get_archives_for_search(
     db_conn,
+    table_prefix: str,
     search_config: SearchJobConfig,
 ):
-    query = f"""SELECT id as archive_id, end_timestamp 
-            FROM {CLP_METADATA_TABLE_PREFIX}archives
+    query = f"""SELECT id as archive_id, end_timestamp
+            FROM {table_prefix}{ARCHIVES_TABLE_SUFFIX}
             """
     filter_clauses = []
     if search_config.end_timestamp is not None:
@@ -380,8 +396,8 @@ def get_archives_for_search(
         filter_clauses.append(f"end_timestamp >= {search_config.begin_timestamp}")
     if search_config.tags is not None:
         filter_clauses.append(
-            f"id IN (SELECT archive_id FROM {CLP_METADATA_TABLE_PREFIX}archive_tags WHERE "
-            f"tag_id IN (SELECT tag_id FROM {CLP_METADATA_TABLE_PREFIX}tags WHERE tag_name IN "
+            f"id IN (SELECT archive_id FROM {table_prefix}{ARCHIVE_TAGS_TABLE_SUFFIX} WHERE "
+            f"tag_id IN (SELECT tag_id FROM {table_prefix}{TAGS_TABLE_SUFFIX} WHERE tag_name IN "
             f"(%s)))" % ", ".join(["%s" for _ in search_config.tags])
         )
     if len(filter_clauses) > 0:
@@ -399,12 +415,13 @@ def get_archives_for_search(
 
 def get_archive_and_file_split_ids_for_ir_extraction(
     db_conn,
+    table_prefix: str,
     extract_ir_config: ExtractIrJobConfig,
 ) -> Tuple[Optional[str], Optional[str]]:
     orig_file_id = extract_ir_config.orig_file_id
     msg_ix = extract_ir_config.msg_ix
 
-    results = get_archive_and_file_split_ids(db_conn, orig_file_id, msg_ix)
+    results = get_archive_and_file_split_ids(db_conn, table_prefix, orig_file_id, msg_ix)
     if len(results) == 0:
         logger.error(f"No matching file splits for orig_file_id={orig_file_id}, msg_ix={msg_ix}")
         return None, None
@@ -420,6 +437,7 @@ def get_archive_and_file_split_ids_for_ir_extraction(
 @exception_default_value(default=[])
 def get_archive_and_file_split_ids(
     db_conn,
+    table_prefix: str,
     orig_file_id: str,
     msg_ix: int,
 ):
@@ -429,16 +447,17 @@ def get_archive_and_file_split_ids(
     1. The file split's original file id = `orig_file_id`
     2. The file split includes the message with index = `msg_ix`
     :param db_conn:
+    :param table_prefix:
     :param orig_file_id: Original file id of the split
     :param msg_ix: Index of the message that the file split must include
     :return: A list of (archive id, file split id) on success. An empty list if
     an exception occurs while interacting with the database.
     """
 
-    query = f"""SELECT archive_id, id as file_split_id 
-            FROM {CLP_METADATA_TABLE_PREFIX}files WHERE
-            orig_file_id = '{orig_file_id}' AND 
-            begin_message_ix <= {msg_ix} AND 
+    query = f"""SELECT archive_id, id as file_split_id
+            FROM {table_prefix}{FILES_TABLE_SUFFIX} WHERE
+            orig_file_id = '{orig_file_id}' AND
+            begin_message_ix <= {msg_ix} AND
             (begin_message_ix + num_messages) > {msg_ix}
             """
 
@@ -451,12 +470,10 @@ def get_archive_and_file_split_ids(
 @exception_default_value(default=False)
 def archive_exists(
     db_conn,
+    table_prefix: str,
     archive_id: str,
 ) -> bool:
-    query = f"""SELECT 1 
-                FROM {CLP_METADATA_TABLE_PREFIX}archives WHERE
-                id = %s
-                """
+    query = f"SELECT 1 FROM {table_prefix}{ARCHIVES_TABLE_SUFFIX} WHERE id = %s"
     with contextlib.closing(db_conn.cursor(dictionary=True)) as cursor:
         cursor.execute(query, (archive_id,))
         if cursor.fetchone():
@@ -611,6 +628,8 @@ def handle_pending_query_jobs(
         and job.get_type() == QueryJobType.SEARCH_OR_AGGREGATION
     ]
 
+    table_prefix = clp_metadata_db_conn_params["table_prefix"]
+
     with contextlib.closing(db_conn_pool.connect()) as db_conn:
         for job in fetch_new_query_jobs(db_conn):
             job_id = str(job["job_id"])
@@ -623,7 +642,7 @@ def handle_pending_query_jobs(
                     continue
 
                 search_config = SearchJobConfig.parse_obj(job_config)
-                archives_for_search = get_archives_for_search(db_conn, search_config)
+                archives_for_search = get_archives_for_search(db_conn, table_prefix, search_config)
                 if len(archives_for_search) == 0:
                     if set_job_or_task_status(
                         db_conn,
@@ -662,9 +681,9 @@ def handle_pending_query_jobs(
                 job_handle: StreamExtractionHandle
                 try:
                     if QueryJobType.EXTRACT_IR == job_type:
-                        job_handle = IrExtractionHandle(job_id, job_config, db_conn)
+                        job_handle = IrExtractionHandle(job_id, job_config, db_conn, table_prefix)
                     else:
-                        job_handle = JsonExtractionHandle(job_id, job_config, db_conn)
+                        job_handle = JsonExtractionHandle(job_id, job_config, db_conn, table_prefix)
                 except ValueError:
                     logger.exception("Failed to initialize extraction job handle")
                     if not set_job_or_task_status(