refactor(clp-py-utils): Move CLP metadata DB table-creation logic into utility module; Conditionally create the datasets table when using the CLP_S storage engine.

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

@@ -231,6 +231,7 @@ def create_db_tables(
         "python3",
         str(clp_py_utils_dir / "create-db-tables.py"),
         "--config", str(container_clp_config.logs_directory / db_config_filename),
+        "--storage-engine", str(container_clp_config.package.storage_engine),
     ]
     # fmt: on
 

components/clp-py-utils/clp_py_utils/clp_config.py

@@ -36,7 +36,13 @@
 QUERY_TASKS_TABLE_NAME = "query_tasks"
 COMPRESSION_JOBS_TABLE_NAME = "compression_jobs"
 COMPRESSION_TASKS_TABLE_NAME = "compression_tasks"
+
+ARCHIVE_TAGS_TABLE_SUFFIX = "archive_tags"
 ARCHIVES_TABLE_SUFFIX = "archives"
+COLUMN_METADATA_TABLE_SUFFIX = "column_metadata"
+DATASETS_TABLE_SUFFIX = "datasets"
+FILES_TABLE_SUFFIX = "files"
+TAGS_TABLE_SUFFIX = "tags"
 
 OS_RELEASE_FILE_PATH = pathlib.Path("etc") / "os-release"
 

components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py

@@ -0,0 +1,142 @@
+from __future__ import annotations
+
+from clp_py_utils.clp_config import (
+    ARCHIVE_TAGS_TABLE_SUFFIX,
+    ARCHIVES_TABLE_SUFFIX,
+    CLP_DEFAULT_DATASET_NAME,
+    COLUMN_METADATA_TABLE_SUFFIX,
+    DATASETS_TABLE_SUFFIX,
+    FILES_TABLE_SUFFIX,
+    TAGS_TABLE_SUFFIX,
+)
+
+
+def _create_archives_table(db_cursor, archives_table_name: str) -> None:
+    db_cursor.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS `{archives_table_name}` (
+            `pagination_id` BIGINT unsigned NOT NULL AUTO_INCREMENT,
+            `id` VARCHAR(64) NOT NULL,
+            `begin_timestamp` BIGINT NOT NULL,
+            `end_timestamp` BIGINT NOT NULL,
+            `uncompressed_size` BIGINT NOT NULL,
+            `size` BIGINT NOT NULL,
+            `creator_id` VARCHAR(64) NOT NULL,
+            `creation_ix` INT NOT NULL,
+            KEY `archives_creation_order` (`creator_id`,`creation_ix`) USING BTREE,
+            UNIQUE KEY `archive_id` (`id`) USING BTREE,
+            PRIMARY KEY (`pagination_id`)
+        )
+        """
+    )
+
+
+def _create_tags_table(db_cursor, tags_table_name: str) -> None:
+    db_cursor.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS `{tags_table_name}` (
+            `tag_id` INT unsigned NOT NULL AUTO_INCREMENT,
+            `tag_name` VARCHAR(255) NOT NULL,
+            UNIQUE KEY (`tag_name`) USING BTREE,
+            PRIMARY KEY (`tag_id`)
+        )
+        """
+    )
+
+
+def _create_archive_tags_table(
+    db_cursor, archive_tags_table_name: str, archives_table_name: str, tags_table_name: str
+) -> None:
+    db_cursor.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS `{archive_tags_table_name}` (
+            `archive_id` VARCHAR(64) NOT NULL,
+            `tag_id` INT unsigned NOT NULL,
+            PRIMARY KEY (`archive_id`,`tag_id`),
+            FOREIGN KEY (`archive_id`) REFERENCES `{archives_table_name}` (`id`),
+            FOREIGN KEY (`tag_id`) REFERENCES `{tags_table_name}` (`tag_id`)
+        )
+        """
+    )
+
+
+def _create_files_table(db_cursor, table_prefix: str) -> None:
+    db_cursor.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS `{table_prefix}{FILES_TABLE_SUFFIX}` (
+            `id` VARCHAR(64) NOT NULL,
+            `orig_file_id` VARCHAR(64) NOT NULL,
+            `path` VARCHAR(12288) NOT NULL,
+            `begin_timestamp` BIGINT NOT NULL,
+            `end_timestamp` BIGINT NOT NULL,
+            `num_uncompressed_bytes` BIGINT NOT NULL,
+            `begin_message_ix` BIGINT NOT NULL,
+            `num_messages` BIGINT NOT NULL,
+            `archive_id` VARCHAR(64) NOT NULL,
+            KEY `files_path` (path(768)) USING BTREE,
+            KEY `files_archive_id` (`archive_id`) USING BTREE,
+            PRIMARY KEY (`id`)
+        ) ROW_FORMAT=DYNAMIC
+        """
+    )
+
+
+def _create_column_metadata_table(db_cursor, table_name: str) -> None:
+    db_cursor.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS `{table_name}` (
+            `name` VARCHAR(512) NOT NULL,
+            `type` TINYINT NOT NULL,
+            PRIMARY KEY (`name`, `type`)
+        )
+        """
+    )
+
+
+def create_datasets_table(db_cursor, table_prefix: str) -> None:
+    """
+    Creates the dataset information table.
+
+    :param db_cursor: The database cursor to execute the table creation.
+    :param table_prefix: A string to prepend to the table name.
+    """
+    db_cursor.execute(
+        f"""
+        CREATE TABLE IF NOT EXISTS `{table_prefix}{DATASETS_TABLE_SUFFIX}` (
+            `name` VARCHAR(255) NOT NULL,
+            `archive_storage_directory` VARCHAR(4096) NOT NULL,
+            PRIMARY KEY (`name`)
+        )
+        """
+    )
+
+
+def create_metadata_db_tables(db_cursor, table_prefix: str, dataset: str | None = None) -> None:
+    """
+    Creates the standard set of tables for CLP's metadata.
+
+    :param db_cursor: The database cursor to execute the table creations.
+    :param table_prefix: A string to prepend to all table names.
+    :param dataset: If set, all tables will be named in a dataset-specific manner.
+    """
+    if dataset is not None:
+        table_prefix = f"{table_prefix}{dataset}_"
+
+    archives_table_name = f"{table_prefix}{ARCHIVES_TABLE_SUFFIX}"
+    tags_table_name = f"{table_prefix}{TAGS_TABLE_SUFFIX}"
+    archive_tags_table_name = f"{table_prefix}{ARCHIVE_TAGS_TABLE_SUFFIX}"
+
+    # TODO: Update this to
+    # {table_prefix}{CLP_DEFAULT_DATASET_NAME}_{COLUMN_METADATA_TABLE_SUFFIX} when we can also
+    # change the indexer to match.
+    column_metadata_table_name = (
+        f"{table_prefix}{COLUMN_METADATA_TABLE_SUFFIX}_{CLP_DEFAULT_DATASET_NAME}"
+    )
+
+    _create_archives_table(db_cursor, archives_table_name)
+    _create_tags_table(db_cursor, tags_table_name)
+    _create_archive_tags_table(
+        db_cursor, archive_tags_table_name, archives_table_name, tags_table_name
+    )
+    _create_files_table(db_cursor, table_prefix)
+    _create_column_metadata_table(db_cursor, column_metadata_table_name)

components/clp-py-utils/clp_py_utils/create-db-tables.py

@@ -4,6 +4,8 @@
 import subprocess
 import sys
 
+from clp_py_utils.clp_config import StorageEngine
+
 # Setup logging
 # Create logger
 logger = logging.getLogger(__file__)
@@ -18,16 +20,25 @@
 def main(argv):
     args_parser = argparse.ArgumentParser(description="Creates database tables for CLP.")
     args_parser.add_argument("--config", required=True, help="Database config file.")
+    args_parser.add_argument(
+        "--storage-engine",
+        type=str,
+        choices=[engine.value for engine in StorageEngine],
+        required=True,
+        help="Compression storage engine to use.",
+    )
     parsed_args = args_parser.parse_args(argv[1:])
 
     config_file_path = pathlib.Path(parsed_args.config)
+    storage_engine = StorageEngine(parsed_args.storage_engine)
 
     script_dir = pathlib.Path(__file__).parent.resolve()
 
     # fmt: off
     cmd = [
         "python3", str(script_dir / "initialize-clp-metadata-db.py"),
         "--config", str(config_file_path),
+        "--storage-engine", str(storage_engine),
     ]
     # fmt: on
     subprocess.run(cmd, check=True)

components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py

@@ -3,10 +3,18 @@
 import logging
 import sys
 from contextlib import closing
+from pathlib import Path
 
 from sql_adapter import SQL_Adapter
 
-from clp_py_utils.clp_config import Database
+from clp_py_utils.clp_config import (
+    Database,
+    StorageEngine,
+)
+from clp_py_utils.clp_metadata_db_utils import (
+    create_datasets_table,
+    create_metadata_db_tables,
+)
 from clp_py_utils.core import read_yaml_config_file
 
 # Setup logging
@@ -23,88 +31,34 @@
 def main(argv):
     args_parser = argparse.ArgumentParser(description="Sets up CLP's metadata tables.")
     args_parser.add_argument("--config", required=True, help="Database config file.")
+    args_parser.add_argument(
+        "--storage-engine",
+        type=str,
+        choices=[engine.value for engine in StorageEngine],
+        required=True,
+        help="Storage engine to create tables for.",
+    )
     parsed_args = args_parser.parse_args(argv[1:])
 
+    config_file_path = Path(parsed_args.config)
+    storage_engine = StorageEngine(parsed_args.storage_engine)
+
     try:
-        database_config = Database.parse_obj(read_yaml_config_file(parsed_args.config))
+        database_config = Database.parse_obj(read_yaml_config_file(config_file_path))
         if database_config is None:
-            raise ValueError(f"Database configuration file '{parsed_args.config}' is empty.")
+            raise ValueError(f"Database configuration file '{config_file_path}' is empty.")
         sql_adapter = SQL_Adapter(database_config)
         clp_db_connection_params = database_config.get_clp_connection_params_and_type(True)
         table_prefix = clp_db_connection_params["table_prefix"]
         with closing(sql_adapter.create_connection(True)) as metadata_db, closing(
             metadata_db.cursor(dictionary=True)
         ) as metadata_db_cursor:
-            metadata_db_cursor.execute(
-                f"""
-                CREATE TABLE IF NOT EXISTS `{table_prefix}archives` (
-                    `pagination_id` BIGINT unsigned NOT NULL AUTO_INCREMENT,
-                    `id` VARCHAR(64) NOT NULL,
-                    `begin_timestamp` BIGINT NOT NULL,
-                    `end_timestamp` BIGINT NOT NULL,
-                    `uncompressed_size` BIGINT NOT NULL,
-                    `size` BIGINT NOT NULL,
-                    `creator_id` VARCHAR(64) NOT NULL,
-                    `creation_ix` INT NOT NULL,
-                    KEY `archives_creation_order` (`creator_id`,`creation_ix`) USING BTREE,
-                    UNIQUE KEY `archive_id` (`id`) USING BTREE,
-                    PRIMARY KEY (`pagination_id`)
-                )
-                """
-            )
-
-            metadata_db_cursor.execute(
-                f"""
-                CREATE TABLE IF NOT EXISTS `{table_prefix}tags` (
-                    `tag_id` INT unsigned NOT NULL AUTO_INCREMENT,
-                    `tag_name` VARCHAR(255) NOT NULL,
-                    UNIQUE KEY (`tag_name`) USING BTREE,
-                    PRIMARY KEY (`tag_id`)
-                )
-                """
-            )
-
-            metadata_db_cursor.execute(
-                f"""
-                CREATE TABLE IF NOT EXISTS `{table_prefix}archive_tags` (
-                    `archive_id` VARCHAR(64) NOT NULL,
-                    `tag_id` INT unsigned NOT NULL,
-                    PRIMARY KEY (`archive_id`,`tag_id`),
-                    FOREIGN KEY (`archive_id`) REFERENCES `{table_prefix}archives` (`id`),
-                    FOREIGN KEY (`tag_id`) REFERENCES `{table_prefix}tags` (`tag_id`)
-                )
-                """
-            )
-
-            metadata_db_cursor.execute(
-                f"""
-                CREATE TABLE IF NOT EXISTS `{table_prefix}files` (
-                    `id` VARCHAR(64) NOT NULL,
-                    `orig_file_id` VARCHAR(64) NOT NULL,
-                    `path` VARCHAR(12288) NOT NULL,
-                    `begin_timestamp` BIGINT NOT NULL,
-                    `end_timestamp` BIGINT NOT NULL,
-                    `num_uncompressed_bytes` BIGINT NOT NULL,
-                    `begin_message_ix` BIGINT NOT NULL,
-                    `num_messages` BIGINT NOT NULL,
-                    `archive_id` VARCHAR(64) NOT NULL,
-                    KEY `files_path` (path(768)) USING BTREE,
-                    KEY `files_archive_id` (`archive_id`) USING BTREE,
-                    PRIMARY KEY (`id`)
-                ) ROW_FORMAT=DYNAMIC
-                """
-            )
-
-            metadata_db_cursor.execute(
-                f"""
-                CREATE TABLE IF NOT EXISTS `{table_prefix}column_metadata_default` (
-                    `name` VARCHAR(512) NOT NULL,
-                    `type` TINYINT NOT NULL,
-                    PRIMARY KEY (`name`, `type`)
-                )
-                """
-            )
-
+            # TODO: After the dataset feature is fully implemented, for clp-json:
+            # 1. Populate the datasets table with the name and path for the "default" dataset.
+            # 2. Change the metadata tables to be specific to the "default" dataset.
+            if StorageEngine.CLP_S == storage_engine:
+                create_datasets_table(metadata_db_cursor, table_prefix)
+            create_metadata_db_tables(metadata_db_cursor, table_prefix)
             metadata_db.commit()
     except:
         logger.exception("Failed to create clp metadata tables.")