From 69307e680ee958ae0a6e6e8a4b66b228eb7cac6b Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 17 Apr 2025 00:24:25 -0400 Subject: [PATCH 01/36] Add dataset specification to compress script --- .../clp-package-utils/clp_package_utils/scripts/compress.py | 6 ++++++ .../clp_package_utils/scripts/native/compress.py | 5 +++++ .../job_orchestration/scheduler/job_config.py | 3 +++ 3 files changed, 14 insertions(+) diff --git a/components/clp-package-utils/clp_package_utils/scripts/compress.py b/components/clp-package-utils/clp_package_utils/scripts/compress.py index f957cbef16..3ce977584c 100755 --- a/components/clp-package-utils/clp_package_utils/scripts/compress.py +++ b/components/clp-package-utils/clp_package_utils/scripts/compress.py @@ -82,6 +82,9 @@ def _generate_compress_cmd( compress_cmd.append(parsed_args.tags) if parsed_args.no_progress_reporting is True: compress_cmd.append("--no-progress-reporting") + if parsed_args.dataset is not None: + compress_cmd.append("--dataset") + compress_cmd.append(parsed_args.dataset) compress_cmd.append("--logs-list") compress_cmd.append(str(logs_list_path)) @@ -145,6 +148,9 @@ def main(argv): args_parser.add_argument( "-f", "--path-list", dest="path_list", help="A file listing all paths to compress." ) + args_parser.add_argument( + "--dataset", default="default" help="The name of the log category to compress into." + ) parsed_args = args_parser.parse_args(argv[1:]) diff --git a/components/clp-package-utils/clp_package_utils/scripts/native/compress.py b/components/clp-package-utils/clp_package_utils/scripts/native/compress.py index b37a87317c..ccf2f44b66 100755 --- a/components/clp-package-utils/clp_package_utils/scripts/native/compress.py +++ b/components/clp-package-utils/clp_package_utils/scripts/native/compress.py @@ -140,6 +140,7 @@ def _generate_clp_io_config( paths_to_compress=logs_to_compress, timestamp_key=parsed_args.timestamp_key, path_prefix_to_remove=str(CONTAINER_INPUT_LOGS_ROOT_DIR), + dataset=parsed_args.dataset, ) elif InputType.S3 == input_type: if len(logs_to_compress) != 1: @@ -152,6 +153,7 @@ def _generate_clp_io_config( key_prefix=s3_config.key_prefix + logs_to_compress[0], credentials=s3_config.credentials, timestamp_key=parsed_args.timestamp_key, + dataset=parsed_args.dataset, ) else: raise ValueError(f"Unsupported input type: {input_type}") @@ -203,6 +205,9 @@ def main(argv): args_parser.add_argument( "-t", "--tags", help="A comma-separated list of tags to apply to the compressed archives." ) + args_parser.add_argument( + "--dataset", default="default" help="The name of the log category to compress into." + ) parsed_args = args_parser.parse_args(argv[1:]) # Validate and load config file diff --git a/components/job-orchestration/job_orchestration/scheduler/job_config.py b/components/job-orchestration/job_orchestration/scheduler/job_config.py index 6f4a97e9f6..75c3675e02 100644 --- a/components/job-orchestration/job_orchestration/scheduler/job_config.py +++ b/components/job-orchestration/job_orchestration/scheduler/job_config.py @@ -25,11 +25,13 @@ class FsInputConfig(BaseModel): paths_to_compress: typing.List[str] path_prefix_to_remove: str = None timestamp_key: typing.Optional[str] = None + dataset: str = "default" class S3InputConfig(BaseModel): type: typing.Literal[InputType.S3.value] = InputType.S3.value timestamp_key: typing.Optional[str] = None + dataset: str = "default" region_code: str bucket: str @@ -78,6 +80,7 @@ class ExtractJsonJobConfig(QueryJobConfig): class SearchJobConfig(QueryJobConfig): query_string: str max_num_results: int + dataset: str = "default" tags: typing.Optional[typing.List[str]] = None begin_timestamp: typing.Optional[int] = None end_timestamp: typing.Optional[int] = None From bd230c0e747319d591a983e46cb9354c075f5fab Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 17 Apr 2025 02:42:11 -0400 Subject: [PATCH 02/36] Add missing commas --- .../clp-package-utils/clp_package_utils/scripts/compress.py | 2 +- .../clp_package_utils/scripts/native/compress.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/components/clp-package-utils/clp_package_utils/scripts/compress.py b/components/clp-package-utils/clp_package_utils/scripts/compress.py index 3ce977584c..7de866dd78 100755 --- a/components/clp-package-utils/clp_package_utils/scripts/compress.py +++ b/components/clp-package-utils/clp_package_utils/scripts/compress.py @@ -149,7 +149,7 @@ def main(argv): "-f", "--path-list", dest="path_list", help="A file listing all paths to compress." ) args_parser.add_argument( - "--dataset", default="default" help="The name of the log category to compress into." + "--dataset", default="default", help="The name of the log category to compress into." ) parsed_args = args_parser.parse_args(argv[1:]) diff --git a/components/clp-package-utils/clp_package_utils/scripts/native/compress.py b/components/clp-package-utils/clp_package_utils/scripts/native/compress.py index ccf2f44b66..011865e550 100755 --- a/components/clp-package-utils/clp_package_utils/scripts/native/compress.py +++ b/components/clp-package-utils/clp_package_utils/scripts/native/compress.py @@ -206,7 +206,7 @@ def main(argv): "-t", "--tags", help="A comma-separated list of tags to apply to the compressed archives." ) args_parser.add_argument( - "--dataset", default="default" help="The name of the log category to compress into." + "--dataset", default="default", help="The name of the log category to compress into." ) parsed_args = args_parser.parse_args(argv[1:]) From c501c9e36352e6753779389f1f761f338aeb7629 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 17 Apr 2025 03:41:19 -0400 Subject: [PATCH 03/36] Create table entries upon getting new dataset names --- .../clp-py-utils/clp_py_utils/clp_config.py | 2 + .../initialize-clp-metadata-db.py | 20 +++++++-- .../compress/compression_scheduler.py | 44 ++++++++++++++++++- 3 files changed, 62 insertions(+), 4 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index ffd8fd1fdb..456eed43c0 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -37,6 +37,8 @@ COMPRESSION_JOBS_TABLE_NAME = "compression_jobs" COMPRESSION_TASKS_TABLE_NAME = "compression_tasks" ARCHIVES_TABLE_SUFFIX = "archives" +COLUMN_METADATA_TABLE_SUFFIX = "column_metadata" +DATASETS_TABLE_SUFFIX = "datasets" OS_RELEASE_FILE_PATH = pathlib.Path("etc") / "os-release" diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 83eef63fc4..dcb744e57d 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -6,7 +6,12 @@ from sql_adapter import SQL_Adapter -from clp_py_utils.clp_config import Database +from clp_py_utils.clp_config import ( + ARCHIVES_TABLE_SUFFIX, + COLUMN_METADATA_TABLE_SUFFIX, + Database, + DATASETS_TABLE_SUFFIX, +) from clp_py_utils.core import read_yaml_config_file # Setup logging @@ -37,7 +42,7 @@ def main(argv): ) as metadata_db_cursor: metadata_db_cursor.execute( f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}archives` ( + CREATE TABLE IF NOT EXISTS `{table_prefix}{ARCHIVES_TABLE_SUFFIX}` ( `pagination_id` BIGINT unsigned NOT NULL AUTO_INCREMENT, `id` VARCHAR(64) NOT NULL, `begin_timestamp` BIGINT NOT NULL, @@ -97,7 +102,16 @@ def main(argv): metadata_db_cursor.execute( f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}column_metadata_default` ( + CREATE TABLE IF NOT EXISTS `{table_prefix}{DATASETS_TABLE_SUFFIX}` ( + `name` VARCHAR(512) NOT NULL, + PRIMARY KEY (`name`) + ) + """ + ) + + metadata_db_cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{table_prefix}default_{COLUMN_METADATA_TABLE_SUFFIX}` ( `name` VARCHAR(512) NOT NULL, `type` TINYINT NOT NULL, PRIMARY KEY (`name`, `type`) diff --git a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py index 0a02aaaff7..8a637f1695 100644 --- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py @@ -12,10 +12,14 @@ import msgpack from clp_package_utils.general import CONTAINER_INPUT_LOGS_ROOT_DIR from clp_py_utils.clp_config import ( + ARCHIVES_TABLE_SUFFIX, CLP_METADATA_TABLE_PREFIX, CLPConfig, + COLUMN_METADATA_TABLE_SUFFIX, COMPRESSION_JOBS_TABLE_NAME, COMPRESSION_TASKS_TABLE_NAME, + DATASETS_TABLE_SUFFIX, + StorageEngine, ) 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 @@ -146,7 +150,9 @@ def _process_s3_input( paths_to_compress_buffer.add_file(object_metadata) -def search_and_schedule_new_tasks(db_conn, db_cursor, clp_metadata_db_connection_config): +def search_and_schedule_new_tasks( + db_conn, db_cursor, clp_metadata_db_connection_config, clp_storage_engine: StorageEngine +): """ For all jobs with PENDING status, split the job into tasks and schedule them. """ @@ -154,6 +160,8 @@ def search_and_schedule_new_tasks(db_conn, db_cursor, clp_metadata_db_connection logger.debug("Search and schedule new tasks") + table_prefix = clp_metadata_db_connection_config["table_prefix"] + # Poll for new compression jobs jobs = fetch_new_jobs(db_cursor) db_conn.commit() @@ -203,6 +211,38 @@ def search_and_schedule_new_tasks(db_conn, db_cursor, clp_metadata_db_connection db_conn.commit() continue + if StorageEngine.CLP_S == clp_storage_engine: + dataset = input_config.dataset + query = f"INSERT INTO {table_prefix}{DATASETS_TABLE_SUFFIX} (name) VALUES ({dataset})" + db_cursor.execute(query) + db_cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{table_prefix}{dataset}_{ARCHIVES_TABLE_SUFFIX}` ( + `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}{dataset}_{COLUMN_METADATA_TABLE_SUFFIX}` ( + `name` VARCHAR(512) NOT NULL, + `type` TINYINT NOT NULL, + PRIMARY KEY (`name`, `type`) + ) + """ + ) + db_conn.commit() + paths_to_compress_buffer.flush() tasks = paths_to_compress_buffer.get_tasks() partition_info = paths_to_compress_buffer.get_partition_info() @@ -385,12 +425,14 @@ def main(argv): db_conn.cursor(dictionary=True) ) as db_cursor: # Start Job Processing Loop + clp_storage_engine = clp_config.package.storage_engine while True: try: search_and_schedule_new_tasks( db_conn, db_cursor, sql_adapter.database_config.get_clp_connection_params_and_type(True), + clp_storage_engine, ) poll_running_jobs(db_conn, db_cursor) time.sleep(clp_config.compression_scheduler.jobs_poll_delay) From 6b909d7b308d1600c13972902ba8530fae0dbaef Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 17 Apr 2025 04:00:04 -0400 Subject: [PATCH 04/36] Add local cache --- .../scheduler/compress/compression_scheduler.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py index 8a637f1695..0a0d8c055b 100644 --- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py @@ -46,6 +46,8 @@ scheduled_jobs = {} +seen_datasets = set() + def fetch_new_jobs(db_cursor): db_cursor.execute( @@ -157,6 +159,7 @@ def search_and_schedule_new_tasks( For all jobs with PENDING status, split the job into tasks and schedule them. """ global scheduled_jobs + global seen_datasets logger.debug("Search and schedule new tasks") @@ -211,10 +214,13 @@ def search_and_schedule_new_tasks( db_conn.commit() continue - if StorageEngine.CLP_S == clp_storage_engine: - dataset = input_config.dataset + dataset = input_config.dataset + if StorageEngine.CLP_S == clp_storage_engine and dataset not in seen_datasets: + seen_datasets.add(dataset) query = f"INSERT INTO {table_prefix}{DATASETS_TABLE_SUFFIX} (name) VALUES ({dataset})" db_cursor.execute(query) + db_conn.commit() + db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_prefix}{dataset}_{ARCHIVES_TABLE_SUFFIX}` ( @@ -232,7 +238,9 @@ def search_and_schedule_new_tasks( ) """ ) - metadata_db_cursor.execute( + db_conn.commit() + + db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_prefix}{dataset}_{COLUMN_METADATA_TABLE_SUFFIX}` ( `name` VARCHAR(512) NOT NULL, From 4d14449cd1373c05a55eb84dbcb8c55fbec471d7 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 17 Apr 2025 04:23:12 -0400 Subject: [PATCH 05/36] Move table creation to helpers --- .../clp_py_utils/create-db-tables.py | 2 +- ...ta-db.py => initialize_clp_metadata_db.py} | 70 ++++++++++--------- .../compress/compression_scheduler.py | 40 +++-------- 3 files changed, 47 insertions(+), 65 deletions(-) rename components/clp-py-utils/clp_py_utils/{initialize-clp-metadata-db.py => initialize_clp_metadata_db.py} (69%) diff --git a/components/clp-py-utils/clp_py_utils/create-db-tables.py b/components/clp-py-utils/clp_py_utils/create-db-tables.py index b376fdf640..a97b03b45f 100644 --- a/components/clp-py-utils/clp_py_utils/create-db-tables.py +++ b/components/clp-py-utils/clp_py_utils/create-db-tables.py @@ -26,7 +26,7 @@ def main(argv): # fmt: off cmd = [ - "python3", str(script_dir / "initialize-clp-metadata-db.py"), + "python3", str(script_dir / "initialize_clp_metadata_db.py"), "--config", str(config_file_path), ] # fmt: on diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize_clp_metadata_db.py similarity index 69% rename from components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py rename to components/clp-py-utils/clp_py_utils/initialize_clp_metadata_db.py index dcb744e57d..b16d6910b7 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize_clp_metadata_db.py @@ -12,17 +12,43 @@ Database, DATASETS_TABLE_SUFFIX, ) +from clp_py_utils.clp_logging import get_logger from clp_py_utils.core import read_yaml_config_file # Setup logging -# Create logger -logger = logging.getLogger(__file__) -logger.setLevel(logging.INFO) -# Setup console logging -logging_console_handler = logging.StreamHandler() -logging_formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s") -logging_console_handler.setFormatter(logging_formatter) -logger.addHandler(logging_console_handler) +logger = get_logger("initialize_clp_metadata_db") + + +def create_archives_table(db_cursor, table_name: str) -> None: + db_cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{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_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 main(argv): @@ -40,23 +66,7 @@ def main(argv): 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_TABLE_SUFFIX}` ( - `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`) - ) - """ - ) + create_archives_table(metadata_db_cursor, f"{table_prefix}{ARCHIVES_TABLE_SUFFIX}") metadata_db_cursor.execute( f""" @@ -109,14 +119,8 @@ def main(argv): """ ) - metadata_db_cursor.execute( - f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}default_{COLUMN_METADATA_TABLE_SUFFIX}` ( - `name` VARCHAR(512) NOT NULL, - `type` TINYINT NOT NULL, - PRIMARY KEY (`name`, `type`) - ) - """ + create_column_metadata_table( + db_cursor, f"{table_prefix}default_{COLUMN_METADATA_TABLE_SUFFIX}" ) metadata_db.commit() diff --git a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py index 0a0d8c055b..92250563a7 100644 --- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py @@ -24,6 +24,10 @@ 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 from clp_py_utils.core import read_yaml_config_file +from clp_py_utils.initialize_clp_metadata_db import ( + create_archives_table, + create_column_metadata_table, +) from clp_py_utils.s3_utils import s3_get_object_metadata from clp_py_utils.sql_adapter import SQL_Adapter from job_orchestration.executor.compress.compression_task import compress @@ -217,37 +221,11 @@ def search_and_schedule_new_tasks( dataset = input_config.dataset if StorageEngine.CLP_S == clp_storage_engine and dataset not in seen_datasets: seen_datasets.add(dataset) - query = f"INSERT INTO {table_prefix}{DATASETS_TABLE_SUFFIX} (name) VALUES ({dataset})" - db_cursor.execute(query) - db_conn.commit() - - db_cursor.execute( - f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}{dataset}_{ARCHIVES_TABLE_SUFFIX}` ( - `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`) - ) - """ - ) - db_conn.commit() - - db_cursor.execute( - f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}{dataset}_{COLUMN_METADATA_TABLE_SUFFIX}` ( - `name` VARCHAR(512) NOT NULL, - `type` TINYINT NOT NULL, - PRIMARY KEY (`name`, `type`) - ) - """ + query = f"INSERT INTO {table_prefix}{DATASETS_TABLE_SUFFIX} (name) VALUES (%s)" + db_cursor.execute(query, (dataset,)) + create_archives_table(db_cursor, f"{table_prefix}{dataset}_{ARCHIVES_TABLE_SUFFIX}") + create_column_metadata_table( + db_cursor, f"{table_prefix}{dataset}_{COLUMN_METADATA_TABLE_SUFFIX}" ) db_conn.commit() From e008fb8fa27d75afcb952a031318c708a64fb918 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 17 Apr 2025 04:28:17 -0400 Subject: [PATCH 06/36] Typo fix --- .../clp-py-utils/clp_py_utils/initialize_clp_metadata_db.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/initialize_clp_metadata_db.py b/components/clp-py-utils/clp_py_utils/initialize_clp_metadata_db.py index b16d6910b7..9b664cf51e 100644 --- a/components/clp-py-utils/clp_py_utils/initialize_clp_metadata_db.py +++ b/components/clp-py-utils/clp_py_utils/initialize_clp_metadata_db.py @@ -120,7 +120,7 @@ def main(argv): ) create_column_metadata_table( - db_cursor, f"{table_prefix}default_{COLUMN_METADATA_TABLE_SUFFIX}" + metadata_db_cursor, f"{table_prefix}default_{COLUMN_METADATA_TABLE_SUFFIX}" ) metadata_db.commit() From f600d259199e7ccab4bc3efa2c5ede888e28087c Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 17 Apr 2025 05:14:28 -0400 Subject: [PATCH 07/36] Revert "Move table creation to helpers" This reverts commit 4d14449cd1373c05a55eb84dbcb8c55fbec471d7. --- .../clp_py_utils/create-db-tables.py | 2 +- ...ta_db.py => initialize-clp-metadata-db.py} | 70 +++++++++---------- .../compress/compression_scheduler.py | 40 ++++++++--- 3 files changed, 65 insertions(+), 47 deletions(-) rename components/clp-py-utils/clp_py_utils/{initialize_clp_metadata_db.py => initialize-clp-metadata-db.py} (69%) diff --git a/components/clp-py-utils/clp_py_utils/create-db-tables.py b/components/clp-py-utils/clp_py_utils/create-db-tables.py index a97b03b45f..b376fdf640 100644 --- a/components/clp-py-utils/clp_py_utils/create-db-tables.py +++ b/components/clp-py-utils/clp_py_utils/create-db-tables.py @@ -26,7 +26,7 @@ def main(argv): # fmt: off cmd = [ - "python3", str(script_dir / "initialize_clp_metadata_db.py"), + "python3", str(script_dir / "initialize-clp-metadata-db.py"), "--config", str(config_file_path), ] # fmt: on diff --git a/components/clp-py-utils/clp_py_utils/initialize_clp_metadata_db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py similarity index 69% rename from components/clp-py-utils/clp_py_utils/initialize_clp_metadata_db.py rename to components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 9b664cf51e..dcb744e57d 100644 --- a/components/clp-py-utils/clp_py_utils/initialize_clp_metadata_db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -12,43 +12,17 @@ Database, DATASETS_TABLE_SUFFIX, ) -from clp_py_utils.clp_logging import get_logger from clp_py_utils.core import read_yaml_config_file # Setup logging -logger = get_logger("initialize_clp_metadata_db") - - -def create_archives_table(db_cursor, table_name: str) -> None: - db_cursor.execute( - f""" - CREATE TABLE IF NOT EXISTS `{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_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`) - ) - """ - ) +# Create logger +logger = logging.getLogger(__file__) +logger.setLevel(logging.INFO) +# Setup console logging +logging_console_handler = logging.StreamHandler() +logging_formatter = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s") +logging_console_handler.setFormatter(logging_formatter) +logger.addHandler(logging_console_handler) def main(argv): @@ -66,7 +40,23 @@ def main(argv): with closing(sql_adapter.create_connection(True)) as metadata_db, closing( metadata_db.cursor(dictionary=True) ) as metadata_db_cursor: - create_archives_table(metadata_db_cursor, f"{table_prefix}{ARCHIVES_TABLE_SUFFIX}") + metadata_db_cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{table_prefix}{ARCHIVES_TABLE_SUFFIX}` ( + `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""" @@ -119,8 +109,14 @@ def main(argv): """ ) - create_column_metadata_table( - metadata_db_cursor, f"{table_prefix}default_{COLUMN_METADATA_TABLE_SUFFIX}" + metadata_db_cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{table_prefix}default_{COLUMN_METADATA_TABLE_SUFFIX}` ( + `name` VARCHAR(512) NOT NULL, + `type` TINYINT NOT NULL, + PRIMARY KEY (`name`, `type`) + ) + """ ) metadata_db.commit() diff --git a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py index 92250563a7..0a0d8c055b 100644 --- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py @@ -24,10 +24,6 @@ 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 from clp_py_utils.core import read_yaml_config_file -from clp_py_utils.initialize_clp_metadata_db import ( - create_archives_table, - create_column_metadata_table, -) from clp_py_utils.s3_utils import s3_get_object_metadata from clp_py_utils.sql_adapter import SQL_Adapter from job_orchestration.executor.compress.compression_task import compress @@ -221,11 +217,37 @@ def search_and_schedule_new_tasks( dataset = input_config.dataset if StorageEngine.CLP_S == clp_storage_engine and dataset not in seen_datasets: seen_datasets.add(dataset) - query = f"INSERT INTO {table_prefix}{DATASETS_TABLE_SUFFIX} (name) VALUES (%s)" - db_cursor.execute(query, (dataset,)) - create_archives_table(db_cursor, f"{table_prefix}{dataset}_{ARCHIVES_TABLE_SUFFIX}") - create_column_metadata_table( - db_cursor, f"{table_prefix}{dataset}_{COLUMN_METADATA_TABLE_SUFFIX}" + query = f"INSERT INTO {table_prefix}{DATASETS_TABLE_SUFFIX} (name) VALUES ({dataset})" + db_cursor.execute(query) + db_conn.commit() + + db_cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{table_prefix}{dataset}_{ARCHIVES_TABLE_SUFFIX}` ( + `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`) + ) + """ + ) + db_conn.commit() + + db_cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{table_prefix}{dataset}_{COLUMN_METADATA_TABLE_SUFFIX}` ( + `name` VARCHAR(512) NOT NULL, + `type` TINYINT NOT NULL, + PRIMARY KEY (`name`, `type`) + ) + """ ) db_conn.commit() From a46bb7db88f7065fb0717b9931a26714df7e29a8 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 17 Apr 2025 05:19:34 -0400 Subject: [PATCH 08/36] Create utils file for creating sql tables --- .../initialize-clp-metadata-db.py | 29 +++--------------- .../clp_py_utils/sql_table_schema_utils.py | 30 +++++++++++++++++++ 2 files changed, 34 insertions(+), 25 deletions(-) create mode 100644 components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index dcb744e57d..a3a0585295 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -13,6 +13,7 @@ DATASETS_TABLE_SUFFIX, ) from clp_py_utils.core import read_yaml_config_file +from clp_py_utils.sql_table_schema_utils import create_archives_table, create_column_metadata_table # Setup logging # Create logger @@ -40,23 +41,7 @@ def main(argv): 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_TABLE_SUFFIX}` ( - `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`) - ) - """ - ) + create_archives_table(metadata_db_cursor, f"{table_prefix}{ARCHIVES_TABLE_SUFFIX}") metadata_db_cursor.execute( f""" @@ -109,14 +94,8 @@ def main(argv): """ ) - metadata_db_cursor.execute( - f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}default_{COLUMN_METADATA_TABLE_SUFFIX}` ( - `name` VARCHAR(512) NOT NULL, - `type` TINYINT NOT NULL, - PRIMARY KEY (`name`, `type`) - ) - """ + create_column_metadata_table( + metadata_db_cursor, f"{table_prefix}default_{COLUMN_METADATA_TABLE_SUFFIX}" ) metadata_db.commit() diff --git a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py new file mode 100644 index 0000000000..9d95c4f372 --- /dev/null +++ b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py @@ -0,0 +1,30 @@ +def create_archives_table(db_cursor, table_name: str) -> None: + db_cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{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_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`) + ) + """ + ) From 5f43ab767a512eeed24f0fe6ef956620fe09cb30 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 17 Apr 2025 05:28:40 -0400 Subject: [PATCH 09/36] Apply to dataset table creation --- .../compress/compression_scheduler.py | 35 +++---------------- 1 file changed, 5 insertions(+), 30 deletions(-) diff --git a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py index 0a0d8c055b..a1aa4ed007 100644 --- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py @@ -26,6 +26,7 @@ from clp_py_utils.core import read_yaml_config_file from clp_py_utils.s3_utils import s3_get_object_metadata from clp_py_utils.sql_adapter import SQL_Adapter +from clp_py_utils.sql_table_schema_utils import create_archives_table, create_column_metadata_table from job_orchestration.executor.compress.compression_task import compress from job_orchestration.scheduler.compress.partition import PathsToCompressBuffer from job_orchestration.scheduler.constants import CompressionJobStatus, CompressionTaskStatus @@ -217,37 +218,11 @@ def search_and_schedule_new_tasks( dataset = input_config.dataset if StorageEngine.CLP_S == clp_storage_engine and dataset not in seen_datasets: seen_datasets.add(dataset) - query = f"INSERT INTO {table_prefix}{DATASETS_TABLE_SUFFIX} (name) VALUES ({dataset})" + query = f'INSERT INTO {table_prefix}{DATASETS_TABLE_SUFFIX} (name) VALUES ("{dataset}")' db_cursor.execute(query) - db_conn.commit() - - db_cursor.execute( - f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}{dataset}_{ARCHIVES_TABLE_SUFFIX}` ( - `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`) - ) - """ - ) - db_conn.commit() - - db_cursor.execute( - f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}{dataset}_{COLUMN_METADATA_TABLE_SUFFIX}` ( - `name` VARCHAR(512) NOT NULL, - `type` TINYINT NOT NULL, - PRIMARY KEY (`name`, `type`) - ) - """ + create_archives_table(db_cursor, f"{table_prefix}{dataset}_{ARCHIVES_TABLE_SUFFIX}") + create_column_metadata_table( + db_cursor, f"{table_prefix}{dataset}_{COLUMN_METADATA_TABLE_SUFFIX}" ) db_conn.commit() From 3f19234bd3ed9b72df80da9a41c27acd4aebee9a Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 17 Apr 2025 05:42:01 -0400 Subject: [PATCH 10/36] Remove unrelated change --- .../job-orchestration/job_orchestration/scheduler/job_config.py | 1 - 1 file changed, 1 deletion(-) diff --git a/components/job-orchestration/job_orchestration/scheduler/job_config.py b/components/job-orchestration/job_orchestration/scheduler/job_config.py index 75c3675e02..7744f4e5e1 100644 --- a/components/job-orchestration/job_orchestration/scheduler/job_config.py +++ b/components/job-orchestration/job_orchestration/scheduler/job_config.py @@ -80,7 +80,6 @@ class ExtractJsonJobConfig(QueryJobConfig): class SearchJobConfig(QueryJobConfig): query_string: str max_num_results: int - dataset: str = "default" tags: typing.Optional[typing.List[str]] = None begin_timestamp: typing.Optional[int] = None end_timestamp: typing.Optional[int] = None From 5ffcf92eb8eca44b7e1d3290a7d646b0b9cc8ac4 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 17 Apr 2025 12:53:51 -0400 Subject: [PATCH 11/36] Move datasets table creation into compression runtime. Optimize the local cache generation. --- .../initialize-clp-metadata-db.py | 15 +------ .../clp_py_utils/sql_table_schema_utils.py | 11 +++++ .../compress/compression_scheduler.py | 45 +++++++++++++++---- 3 files changed, 49 insertions(+), 22 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index a3a0585295..2b07e6c9bc 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -13,7 +13,7 @@ DATASETS_TABLE_SUFFIX, ) from clp_py_utils.core import read_yaml_config_file -from clp_py_utils.sql_table_schema_utils import create_archives_table, create_column_metadata_table +from clp_py_utils.sql_table_schema_utils import create_archives_table # Setup logging # Create logger @@ -85,19 +85,6 @@ def main(argv): """ ) - metadata_db_cursor.execute( - f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}{DATASETS_TABLE_SUFFIX}` ( - `name` VARCHAR(512) NOT NULL, - PRIMARY KEY (`name`) - ) - """ - ) - - create_column_metadata_table( - metadata_db_cursor, f"{table_prefix}default_{COLUMN_METADATA_TABLE_SUFFIX}" - ) - metadata_db.commit() except: logger.exception("Failed to create clp metadata tables.") diff --git a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py index 9d95c4f372..ec25cd9db5 100644 --- a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py +++ b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py @@ -28,3 +28,14 @@ def create_column_metadata_table(db_cursor, table_name: str) -> None: ) """ ) + + +def create_datasets_table(db_cursor, table_name: str) -> None: + db_cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{table_name}` ( + `name` VARCHAR(512) NOT NULL, + PRIMARY KEY (`name`) + ) + """ + ) diff --git a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py index a1aa4ed007..581b4b2137 100644 --- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py @@ -6,6 +6,7 @@ import time from contextlib import closing from pathlib import Path +from typing import Set import brotli import celery @@ -26,7 +27,11 @@ from clp_py_utils.core import read_yaml_config_file from clp_py_utils.s3_utils import s3_get_object_metadata from clp_py_utils.sql_adapter import SQL_Adapter -from clp_py_utils.sql_table_schema_utils import create_archives_table, create_column_metadata_table +from clp_py_utils.sql_table_schema_utils import ( + create_archives_table, + create_column_metadata_table, + create_datasets_table, +) from job_orchestration.executor.compress.compression_task import compress from job_orchestration.scheduler.compress.partition import PathsToCompressBuffer from job_orchestration.scheduler.constants import CompressionJobStatus, CompressionTaskStatus @@ -47,8 +52,6 @@ scheduled_jobs = {} -seen_datasets = set() - def fetch_new_jobs(db_cursor): db_cursor.execute( @@ -153,14 +156,30 @@ def _process_s3_input( paths_to_compress_buffer.add_file(object_metadata) +def init_datasets_table(db_conn, db_cursor, clp_metadata_db_connection_config) -> Set[str]: + table_prefix = clp_metadata_db_connection_config["table_prefix"] + table_name = f"{table_prefix}{DATASETS_TABLE_SUFFIX}" + + create_datasets_table(db_cursor, table_name) + db_conn.commit() + + db_cursor.execute(f"SELECT name FROM {table_name}") + rows = cursor.fetchall() + db_conn.commit() + return {str(row[0]) for row in rows} + + def search_and_schedule_new_tasks( - db_conn, db_cursor, clp_metadata_db_connection_config, clp_storage_engine: StorageEngine -): + db_conn, + db_cursor, + clp_metadata_db_connection_config, + clp_storage_engine: StorageEngine, + seen_datasets: Set[str], +) -> None: """ For all jobs with PENDING status, split the job into tasks and schedule them. """ global scheduled_jobs - global seen_datasets logger.debug("Search and schedule new tasks") @@ -218,6 +237,7 @@ def search_and_schedule_new_tasks( dataset = input_config.dataset if StorageEngine.CLP_S == clp_storage_engine and dataset not in seen_datasets: seen_datasets.add(dataset) + # TODO: ensure that the dataset string doesn't violate SQL table naming rules query = f'INSERT INTO {table_prefix}{DATASETS_TABLE_SUFFIX} (name) VALUES ("{dataset}")' db_cursor.execute(query) create_archives_table(db_cursor, f"{table_prefix}{dataset}_{ARCHIVES_TABLE_SUFFIX}") @@ -407,15 +427,24 @@ def main(argv): with closing(sql_adapter.create_connection(True)) as db_conn, closing( db_conn.cursor(dictionary=True) ) as db_cursor: - # Start Job Processing Loop + clp_metadata_db_connection_config = ( + sql_adapter.database_config.get_clp_connection_params_and_type(True) + ) clp_storage_engine = clp_config.package.storage_engine + if StorageEngine.CLP_S == clp_storage_engine: + seen_datasets = init_datasets_table( + db_conn, db_cursor, clp_metadata_db_connection_config + ) + + # Start Job Processing Loop while True: try: search_and_schedule_new_tasks( db_conn, db_cursor, - sql_adapter.database_config.get_clp_connection_params_and_type(True), + clp_metadata_db_connection_config, clp_storage_engine, + seen_datasets, ) poll_running_jobs(db_conn, db_cursor) time.sleep(clp_config.compression_scheduler.jobs_poll_delay) From 279c3397bc9ecc5871af7dcaeb90f7a93f27703f Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Fri, 18 Apr 2025 17:46:57 -0400 Subject: [PATCH 12/36] Revert --dataset interface changes --- .../clp-package-utils/clp_package_utils/scripts/compress.py | 6 ------ .../clp_package_utils/scripts/native/compress.py | 5 ----- 2 files changed, 11 deletions(-) diff --git a/components/clp-package-utils/clp_package_utils/scripts/compress.py b/components/clp-package-utils/clp_package_utils/scripts/compress.py index 7de866dd78..f957cbef16 100755 --- a/components/clp-package-utils/clp_package_utils/scripts/compress.py +++ b/components/clp-package-utils/clp_package_utils/scripts/compress.py @@ -82,9 +82,6 @@ def _generate_compress_cmd( compress_cmd.append(parsed_args.tags) if parsed_args.no_progress_reporting is True: compress_cmd.append("--no-progress-reporting") - if parsed_args.dataset is not None: - compress_cmd.append("--dataset") - compress_cmd.append(parsed_args.dataset) compress_cmd.append("--logs-list") compress_cmd.append(str(logs_list_path)) @@ -148,9 +145,6 @@ def main(argv): args_parser.add_argument( "-f", "--path-list", dest="path_list", help="A file listing all paths to compress." ) - args_parser.add_argument( - "--dataset", default="default", help="The name of the log category to compress into." - ) parsed_args = args_parser.parse_args(argv[1:]) diff --git a/components/clp-package-utils/clp_package_utils/scripts/native/compress.py b/components/clp-package-utils/clp_package_utils/scripts/native/compress.py index 011865e550..b37a87317c 100755 --- a/components/clp-package-utils/clp_package_utils/scripts/native/compress.py +++ b/components/clp-package-utils/clp_package_utils/scripts/native/compress.py @@ -140,7 +140,6 @@ def _generate_clp_io_config( paths_to_compress=logs_to_compress, timestamp_key=parsed_args.timestamp_key, path_prefix_to_remove=str(CONTAINER_INPUT_LOGS_ROOT_DIR), - dataset=parsed_args.dataset, ) elif InputType.S3 == input_type: if len(logs_to_compress) != 1: @@ -153,7 +152,6 @@ def _generate_clp_io_config( key_prefix=s3_config.key_prefix + logs_to_compress[0], credentials=s3_config.credentials, timestamp_key=parsed_args.timestamp_key, - dataset=parsed_args.dataset, ) else: raise ValueError(f"Unsupported input type: {input_type}") @@ -205,9 +203,6 @@ def main(argv): args_parser.add_argument( "-t", "--tags", help="A comma-separated list of tags to apply to the compressed archives." ) - args_parser.add_argument( - "--dataset", default="default", help="The name of the log category to compress into." - ) parsed_args = args_parser.parse_args(argv[1:]) # Validate and load config file From 2ba9525b65f5c89989c4e07a5d39af390a8e3b42 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 09:30:15 -0400 Subject: [PATCH 13/36] Remove unrelated changes --- .../compress/compression_scheduler.py | 58 +------------------ .../job_orchestration/scheduler/job_config.py | 2 - 2 files changed, 2 insertions(+), 58 deletions(-) diff --git a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py index 581b4b2137..0a02aaaff7 100644 --- a/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py +++ b/components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py @@ -6,32 +6,22 @@ import time from contextlib import closing from pathlib import Path -from typing import Set import brotli import celery import msgpack from clp_package_utils.general import CONTAINER_INPUT_LOGS_ROOT_DIR from clp_py_utils.clp_config import ( - ARCHIVES_TABLE_SUFFIX, CLP_METADATA_TABLE_PREFIX, CLPConfig, - COLUMN_METADATA_TABLE_SUFFIX, COMPRESSION_JOBS_TABLE_NAME, COMPRESSION_TASKS_TABLE_NAME, - DATASETS_TABLE_SUFFIX, - StorageEngine, ) 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 from clp_py_utils.core import read_yaml_config_file from clp_py_utils.s3_utils import s3_get_object_metadata from clp_py_utils.sql_adapter import SQL_Adapter -from clp_py_utils.sql_table_schema_utils import ( - create_archives_table, - create_column_metadata_table, - create_datasets_table, -) from job_orchestration.executor.compress.compression_task import compress from job_orchestration.scheduler.compress.partition import PathsToCompressBuffer from job_orchestration.scheduler.constants import CompressionJobStatus, CompressionTaskStatus @@ -156,26 +146,7 @@ def _process_s3_input( paths_to_compress_buffer.add_file(object_metadata) -def init_datasets_table(db_conn, db_cursor, clp_metadata_db_connection_config) -> Set[str]: - table_prefix = clp_metadata_db_connection_config["table_prefix"] - table_name = f"{table_prefix}{DATASETS_TABLE_SUFFIX}" - - create_datasets_table(db_cursor, table_name) - db_conn.commit() - - db_cursor.execute(f"SELECT name FROM {table_name}") - rows = cursor.fetchall() - db_conn.commit() - return {str(row[0]) for row in rows} - - -def search_and_schedule_new_tasks( - db_conn, - db_cursor, - clp_metadata_db_connection_config, - clp_storage_engine: StorageEngine, - seen_datasets: Set[str], -) -> None: +def search_and_schedule_new_tasks(db_conn, db_cursor, clp_metadata_db_connection_config): """ For all jobs with PENDING status, split the job into tasks and schedule them. """ @@ -183,8 +154,6 @@ def search_and_schedule_new_tasks( logger.debug("Search and schedule new tasks") - table_prefix = clp_metadata_db_connection_config["table_prefix"] - # Poll for new compression jobs jobs = fetch_new_jobs(db_cursor) db_conn.commit() @@ -234,18 +203,6 @@ def search_and_schedule_new_tasks( db_conn.commit() continue - dataset = input_config.dataset - if StorageEngine.CLP_S == clp_storage_engine and dataset not in seen_datasets: - seen_datasets.add(dataset) - # TODO: ensure that the dataset string doesn't violate SQL table naming rules - query = f'INSERT INTO {table_prefix}{DATASETS_TABLE_SUFFIX} (name) VALUES ("{dataset}")' - db_cursor.execute(query) - create_archives_table(db_cursor, f"{table_prefix}{dataset}_{ARCHIVES_TABLE_SUFFIX}") - create_column_metadata_table( - db_cursor, f"{table_prefix}{dataset}_{COLUMN_METADATA_TABLE_SUFFIX}" - ) - db_conn.commit() - paths_to_compress_buffer.flush() tasks = paths_to_compress_buffer.get_tasks() partition_info = paths_to_compress_buffer.get_partition_info() @@ -427,24 +384,13 @@ def main(argv): with closing(sql_adapter.create_connection(True)) as db_conn, closing( db_conn.cursor(dictionary=True) ) as db_cursor: - clp_metadata_db_connection_config = ( - sql_adapter.database_config.get_clp_connection_params_and_type(True) - ) - clp_storage_engine = clp_config.package.storage_engine - if StorageEngine.CLP_S == clp_storage_engine: - seen_datasets = init_datasets_table( - db_conn, db_cursor, clp_metadata_db_connection_config - ) - # Start Job Processing Loop while True: try: search_and_schedule_new_tasks( db_conn, db_cursor, - clp_metadata_db_connection_config, - clp_storage_engine, - seen_datasets, + sql_adapter.database_config.get_clp_connection_params_and_type(True), ) poll_running_jobs(db_conn, db_cursor) time.sleep(clp_config.compression_scheduler.jobs_poll_delay) diff --git a/components/job-orchestration/job_orchestration/scheduler/job_config.py b/components/job-orchestration/job_orchestration/scheduler/job_config.py index 7744f4e5e1..6f4a97e9f6 100644 --- a/components/job-orchestration/job_orchestration/scheduler/job_config.py +++ b/components/job-orchestration/job_orchestration/scheduler/job_config.py @@ -25,13 +25,11 @@ class FsInputConfig(BaseModel): paths_to_compress: typing.List[str] path_prefix_to_remove: str = None timestamp_key: typing.Optional[str] = None - dataset: str = "default" class S3InputConfig(BaseModel): type: typing.Literal[InputType.S3.value] = InputType.S3.value timestamp_key: typing.Optional[str] = None - dataset: str = "default" region_code: str bucket: str From 9d9d0d6df779eb304bdc8403e0c53a48e1dca867 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 09:42:30 -0400 Subject: [PATCH 14/36] Move all table creations into utility file --- .../clp-py-utils/clp_py_utils/clp_config.py | 4 ++ .../initialize-clp-metadata-db.py | 50 +++-------------- .../clp_py_utils/sql_table_schema_utils.py | 56 +++++++++++++++++-- 3 files changed, 63 insertions(+), 47 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_config.py b/components/clp-py-utils/clp_py_utils/clp_config.py index 456eed43c0..f5e49c8a0b 100644 --- a/components/clp-py-utils/clp_py_utils/clp_config.py +++ b/components/clp-py-utils/clp_py_utils/clp_config.py @@ -36,9 +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" diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 2b07e6c9bc..314d8cdaa4 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -7,10 +7,12 @@ from sql_adapter import SQL_Adapter from clp_py_utils.clp_config import ( + ARCHIVE_TAGS_TABLE_SUFFIX, ARCHIVES_TABLE_SUFFIX, - COLUMN_METADATA_TABLE_SUFFIX, Database, DATASETS_TABLE_SUFFIX, + FILES_TABLE_SUFFIX, + TAGS_TABLE_SUFFIX, ) from clp_py_utils.core import read_yaml_config_file from clp_py_utils.sql_table_schema_utils import create_archives_table @@ -42,49 +44,11 @@ def main(argv): metadata_db.cursor(dictionary=True) ) as metadata_db_cursor: create_archives_table(metadata_db_cursor, f"{table_prefix}{ARCHIVES_TABLE_SUFFIX}") - - 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`) - ) - """ + create_tags_table(metadata_db_cursor, f"{table_prefix}{TAGS_TABLE_SUFFIX}") + create_archive_tags_table( + metadata_db_cursor, f"{table_prefix}{ARCHIVE_TAGS_TABLE_SUFFIX}" ) - - 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 - """ - ) - + create_files_table(metadata_db_cursor, f"{table_prefix}{FILES_TABLE_SUFFIX}") metadata_db.commit() except: logger.exception("Failed to create clp metadata tables.") diff --git a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py index ec25cd9db5..13e1affb18 100644 --- a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py +++ b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py @@ -18,18 +18,54 @@ def create_archives_table(db_cursor, table_name: str) -> None: ) -def create_column_metadata_table(db_cursor, table_name: str) -> None: +def create_tags_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`) + `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, table_name: str) -> None: + metadata_db_cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{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 `{table_prefix}archives` (`id`), + FOREIGN KEY (`tag_id`) REFERENCES `{table_prefix}tags` (`tag_id`) ) """ ) +def create_files_table(db_cursor, table_name: str) -> None: + metadata_db_cursor.execute( + f""" + CREATE TABLE IF NOT EXISTS `{table_name}` ( + `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_datasets_table(db_cursor, table_name: str) -> None: db_cursor.execute( f""" @@ -39,3 +75,15 @@ def create_datasets_table(db_cursor, table_name: str) -> None: ) """ ) + + +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`) + ) + """ + ) From bd33ebc0aa81a8472afa5b2e8c341c427725198f Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 10:56:52 -0400 Subject: [PATCH 15/36] Fix bugs --- .../clp_py_utils/initialize-clp-metadata-db.py | 7 ++++++- .../clp-py-utils/clp_py_utils/sql_table_schema_utils.py | 2 +- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 314d8cdaa4..5bf166ead5 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -15,7 +15,12 @@ TAGS_TABLE_SUFFIX, ) from clp_py_utils.core import read_yaml_config_file -from clp_py_utils.sql_table_schema_utils import create_archives_table +from clp_py_utils.sql_table_schema_utils import ( + create_archive_tags_table, + create_archives_table, + create_files_table, + create_tags_table, +) # Setup logging # Create logger diff --git a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py index 13e1affb18..387ec0627d 100644 --- a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py +++ b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py @@ -32,7 +32,7 @@ def create_tags_table(db_cursor, table_name: str) -> None: def create_archive_tags_table(db_cursor, table_name: str) -> None: - metadata_db_cursor.execute( + db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_name}` ( `archive_id` VARCHAR(64) NOT NULL, From 01618f807702d269830a3a1b2be966b86eaee12b Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 10:58:01 -0400 Subject: [PATCH 16/36] Fix datasets table schema --- components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py index 387ec0627d..292f913dfb 100644 --- a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py +++ b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py @@ -70,7 +70,8 @@ def create_datasets_table(db_cursor, table_name: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_name}` ( - `name` VARCHAR(512) NOT NULL, + `name` VARCHAR(255) NOT NULL, + `archive_storage_directory` VARCHAR(4096) NOT NULL, PRIMARY KEY (`name`) ) """ From 45258ce5d66a87304a82379d993a7024293e8e91 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 11:14:56 -0400 Subject: [PATCH 17/36] Group metadata db tables creation into a single function --- .../initialize-clp-metadata-db.py | 24 ++---------- .../clp_py_utils/sql_table_schema_utils.py | 38 +++++++++++++++++-- 2 files changed, 39 insertions(+), 23 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 5bf166ead5..ab70fb546c 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -6,21 +6,9 @@ from sql_adapter import SQL_Adapter -from clp_py_utils.clp_config import ( - ARCHIVE_TAGS_TABLE_SUFFIX, - ARCHIVES_TABLE_SUFFIX, - Database, - DATASETS_TABLE_SUFFIX, - FILES_TABLE_SUFFIX, - TAGS_TABLE_SUFFIX, -) +from clp_py_utils.clp_config import Database, DATASETS_TABLE_SUFFIX from clp_py_utils.core import read_yaml_config_file -from clp_py_utils.sql_table_schema_utils import ( - create_archive_tags_table, - create_archives_table, - create_files_table, - create_tags_table, -) +from clp_py_utils.sql_table_schema_utils import create_metadata_db_tables # Setup logging # Create logger @@ -42,18 +30,14 @@ def main(argv): database_config = Database.parse_obj(read_yaml_config_file(parsed_args.config)) if database_config is None: raise ValueError(f"Database configuration file '{parsed_args.config}' is empty.") + logger.info(database_config) 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: - create_archives_table(metadata_db_cursor, f"{table_prefix}{ARCHIVES_TABLE_SUFFIX}") - create_tags_table(metadata_db_cursor, f"{table_prefix}{TAGS_TABLE_SUFFIX}") - create_archive_tags_table( - metadata_db_cursor, f"{table_prefix}{ARCHIVE_TAGS_TABLE_SUFFIX}" - ) - create_files_table(metadata_db_cursor, f"{table_prefix}{FILES_TABLE_SUFFIX}") + create_metadata_db_tables(metadata_db_cursor, table_prefix) metadata_db.commit() except: logger.exception("Failed to create clp metadata tables.") diff --git a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py index 292f913dfb..f75bb43e11 100644 --- a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py +++ b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py @@ -1,3 +1,13 @@ +import typing + +from clp_py_utils.clp_config import ( + ARCHIVE_TAGS_TABLE_SUFFIX, + ARCHIVES_TABLE_SUFFIX, + FILES_TABLE_SUFFIX, + TAGS_TABLE_SUFFIX, +) + + def create_archives_table(db_cursor, table_name: str) -> None: db_cursor.execute( f""" @@ -31,15 +41,17 @@ def create_tags_table(db_cursor, table_name: str) -> None: ) -def create_archive_tags_table(db_cursor, table_name: str) -> None: +def create_archive_tags_table( + db_cursor, table_name: str, archives_table_name: str, tags_table_name: str +) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{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 `{table_prefix}archives` (`id`), - FOREIGN KEY (`tag_id`) REFERENCES `{table_prefix}tags` (`tag_id`) + FOREIGN KEY (`archive_id`) REFERENCES `{archives_table_name}` (`id`), + FOREIGN KEY (`tag_id`) REFERENCES `{tags_table_name}` (`tag_id`) ) """ ) @@ -88,3 +100,23 @@ def create_column_metadata_table(db_cursor, table_name: str) -> None: ) """ ) + + +def create_metadata_db_tables( + db_cursor, table_prefix: str, dataset: typing.Optional[str] = None +) -> None: + if dataset is not None: + table_prefix = f"{table_prefix}{dataset}_" + create_column_metadata_table(db_cursor, f"{table_prefix}{COLUMN_METADATA_TABLE_SUFFIX}") + + 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}" + files_table_name = f"{table_prefix}{FILES_TABLE_SUFFIX}" + + 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, files_table_name) From c8bf33d3253622cb08b2d9146d1bb6f10d5fca44 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 11:20:20 -0400 Subject: [PATCH 18/36] Fix typo --- components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py index f75bb43e11..2c8e875319 100644 --- a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py +++ b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py @@ -58,7 +58,7 @@ def create_archive_tags_table( def create_files_table(db_cursor, table_name: str) -> None: - metadata_db_cursor.execute( + db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_name}` ( `id` VARCHAR(64) NOT NULL, From f18089b447f6192a422a7533407fd25ed55e1e36 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 11:26:39 -0400 Subject: [PATCH 19/36] Remove logging statements --- .../clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index ab70fb546c..47e13fbd10 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -6,7 +6,7 @@ from sql_adapter import SQL_Adapter -from clp_py_utils.clp_config import Database, DATASETS_TABLE_SUFFIX +from clp_py_utils.clp_config import Database from clp_py_utils.core import read_yaml_config_file from clp_py_utils.sql_table_schema_utils import create_metadata_db_tables @@ -30,7 +30,6 @@ def main(argv): database_config = Database.parse_obj(read_yaml_config_file(parsed_args.config)) if database_config is None: raise ValueError(f"Database configuration file '{parsed_args.config}' is empty.") - logger.info(database_config) 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"] From 943020faae06d6aa26f7884a746d07e938db80b0 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 11:44:15 -0400 Subject: [PATCH 20/36] Add missing import --- components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py | 1 + 1 file changed, 1 insertion(+) diff --git a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py index 2c8e875319..87000f5a45 100644 --- a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py +++ b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py @@ -3,6 +3,7 @@ from clp_py_utils.clp_config import ( ARCHIVE_TAGS_TABLE_SUFFIX, ARCHIVES_TABLE_SUFFIX, + COLUMN_METADATA_TABLE_SUFFIX, FILES_TABLE_SUFFIX, TAGS_TABLE_SUFFIX, ) From ff83cc7f672d4661b82406d91921901d9747e23e Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 18:28:48 -0400 Subject: [PATCH 21/36] Add type for db_cursor --- .../clp_py_utils/sql_table_schema_utils.py | 20 ++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py index 87000f5a45..1a5bf82510 100644 --- a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py +++ b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py @@ -1,5 +1,8 @@ import typing +from mariadb import Cursor as MariaDBCursor +from mysql.connector.cursor import MySQLCursor + from clp_py_utils.clp_config import ( ARCHIVE_TAGS_TABLE_SUFFIX, ARCHIVES_TABLE_SUFFIX, @@ -9,7 +12,7 @@ ) -def create_archives_table(db_cursor, table_name: str) -> None: +def create_archives_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_name}` ( @@ -29,7 +32,7 @@ def create_archives_table(db_cursor, table_name: str) -> None: ) -def create_tags_table(db_cursor, table_name: str) -> None: +def create_tags_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_name}` ( @@ -43,7 +46,10 @@ def create_tags_table(db_cursor, table_name: str) -> None: def create_archive_tags_table( - db_cursor, table_name: str, archives_table_name: str, tags_table_name: str + db_cursor: MariaDbCursor | MySQLCursor, + table_name: str, + archives_table_name: str, + tags_table_name: str, ) -> None: db_cursor.execute( f""" @@ -58,7 +64,7 @@ def create_archive_tags_table( ) -def create_files_table(db_cursor, table_name: str) -> None: +def create_files_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_name}` ( @@ -79,7 +85,7 @@ def create_files_table(db_cursor, table_name: str) -> None: ) -def create_datasets_table(db_cursor, table_name: str) -> None: +def create_datasets_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_name}` ( @@ -91,7 +97,7 @@ def create_datasets_table(db_cursor, table_name: str) -> None: ) -def create_column_metadata_table(db_cursor, table_name: str) -> None: +def create_column_metadata_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_name}` ( @@ -104,7 +110,7 @@ def create_column_metadata_table(db_cursor, table_name: str) -> None: def create_metadata_db_tables( - db_cursor, table_prefix: str, dataset: typing.Optional[str] = None + db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str, dataset: typing.Optional[str] = None ) -> None: if dataset is not None: table_prefix = f"{table_prefix}{dataset}_" From c0cacfdde08eef82fd466da7d5e4b61e9e8bd370 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 18:36:58 -0400 Subject: [PATCH 22/36] Change table creation functions from public to private and change to use table prefix --- .../clp_py_utils/sql_table_schema_utils.py | 65 ++++++++----------- 1 file changed, 28 insertions(+), 37 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py index 1a5bf82510..d79ffb07e8 100644 --- a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py +++ b/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py @@ -7,15 +7,16 @@ ARCHIVE_TAGS_TABLE_SUFFIX, ARCHIVES_TABLE_SUFFIX, COLUMN_METADATA_TABLE_SUFFIX, + DATASETS_TABLE_SUFFIX, FILES_TABLE_SUFFIX, TAGS_TABLE_SUFFIX, ) -def create_archives_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) -> None: +def _create_archives_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str) -> None: db_cursor.execute( f""" - CREATE TABLE IF NOT EXISTS `{table_name}` ( + CREATE TABLE IF NOT EXISTS `{table_prefix}{ARCHIVES_TABLE_SUFFIX}` ( `pagination_id` BIGINT unsigned NOT NULL AUTO_INCREMENT, `id` VARCHAR(64) NOT NULL, `begin_timestamp` BIGINT NOT NULL, @@ -32,10 +33,10 @@ def create_archives_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: st ) -def create_tags_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) -> None: +def _create_tags_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str) -> None: db_cursor.execute( f""" - CREATE TABLE IF NOT EXISTS `{table_name}` ( + CREATE TABLE IF NOT EXISTS `{table_prefix}{TAGS_TABLE_SUFFIX}` ( `tag_id` INT unsigned NOT NULL AUTO_INCREMENT, `tag_name` VARCHAR(255) NOT NULL, UNIQUE KEY (`tag_name`) USING BTREE, @@ -45,29 +46,24 @@ def create_tags_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) - ) -def create_archive_tags_table( - db_cursor: MariaDbCursor | MySQLCursor, - table_name: str, - archives_table_name: str, - tags_table_name: str, -) -> None: +def _create_archive_tags_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str) -> None: db_cursor.execute( f""" - CREATE TABLE IF NOT EXISTS `{table_name}` ( + CREATE TABLE IF NOT EXISTS `{table_prefix}{ARCHIVE_TAGS_TABLE_SUFFIX}` ( `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`) + FOREIGN KEY (`archive_id`) REFERENCES `{table_prefix}{ARCHIVES_TABLE_SUFFIX}` (`id`), + FOREIGN KEY (`tag_id`) REFERENCES `{table_prefix}{TAGS_TABLE_SUFFIX}` (`tag_id`) ) """ ) -def create_files_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) -> None: +def _create_files_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str) -> None: db_cursor.execute( f""" - CREATE TABLE IF NOT EXISTS `{table_name}` ( + 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, @@ -85,25 +81,27 @@ def create_files_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) ) -def create_datasets_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) -> None: +def _create_column_metadata_table( + db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str +) -> None: db_cursor.execute( f""" - CREATE TABLE IF NOT EXISTS `{table_name}` ( - `name` VARCHAR(255) NOT NULL, - `archive_storage_directory` VARCHAR(4096) NOT NULL, - PRIMARY KEY (`name`) + CREATE TABLE IF NOT EXISTS `{table_prefix}{COLUMN_METADATA_TABLE_SUFFIX}` ( + `name` VARCHAR(512) NOT NULL, + `type` TINYINT NOT NULL, + PRIMARY KEY (`name`, `type`) ) """ ) -def create_column_metadata_table(db_cursor: MariaDbCursor | MySQLCursor, table_name: str) -> None: +def create_datasets_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: 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`) + 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`) ) """ ) @@ -114,16 +112,9 @@ def create_metadata_db_tables( ) -> None: if dataset is not None: table_prefix = f"{table_prefix}{dataset}_" - create_column_metadata_table(db_cursor, f"{table_prefix}{COLUMN_METADATA_TABLE_SUFFIX}") + _create_column_metadata_table(db_cursor, table_prefix) - 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}" - files_table_name = f"{table_prefix}{FILES_TABLE_SUFFIX}" - - 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, files_table_name) + _create_archives_table(db_cursor, table_prefix) + _create_tags_table(db_cursor, table_prefix) + _create_archive_tags_table(db_cursor, table_prefix) + _create_files_table(db_cursor, table_prefix) From 9ca99d9d0915b70bc454f4e6f4d3c56729b16a61 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 18:40:11 -0400 Subject: [PATCH 23/36] rename utils file --- .../{sql_table_schema_utils.py => clp_metadata_db_utils.py} | 0 .../clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py | 2 +- 2 files changed, 1 insertion(+), 1 deletion(-) rename components/clp-py-utils/clp_py_utils/{sql_table_schema_utils.py => clp_metadata_db_utils.py} (100%) diff --git a/components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py similarity index 100% rename from components/clp-py-utils/clp_py_utils/sql_table_schema_utils.py rename to components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 47e13fbd10..5a77f0cd5d 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -8,7 +8,7 @@ from clp_py_utils.clp_config import Database from clp_py_utils.core import read_yaml_config_file -from clp_py_utils.sql_table_schema_utils import create_metadata_db_tables +from clp_py_utils.clp_metadata_db_utils.py import create_metadata_db_tables # Setup logging # Create logger From d07e3f40cd6be61b4900ed2a7fa30c4addcd7b9f Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 18:45:35 -0400 Subject: [PATCH 24/36] Fix typo --- components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py index d79ffb07e8..f1ba6878ff 100644 --- a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py +++ b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py @@ -1,6 +1,6 @@ import typing -from mariadb import Cursor as MariaDBCursor +from mariadb import Cursor as MariaDbCursor from mysql.connector.cursor import MySQLCursor from clp_py_utils.clp_config import ( From 6d1b356ae94e792987dba38d73f5dba99c635d78 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Mon, 21 Apr 2025 18:47:03 -0400 Subject: [PATCH 25/36] Use pipe syntax --- components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py index f1ba6878ff..fae06060d3 100644 --- a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py +++ b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py @@ -108,7 +108,7 @@ def create_datasets_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: def create_metadata_db_tables( - db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str, dataset: typing.Optional[str] = None + db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str, dataset: str | None = None ) -> None: if dataset is not None: table_prefix = f"{table_prefix}{dataset}_" From ad67fd9e0113e7740ce006a4f4e088d88c865feb Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Tue, 22 Apr 2025 11:47:29 -0400 Subject: [PATCH 26/36] Propagate storage engine config to metadata db table creation code --- .../clp_package_utils/scripts/start_clp.py | 1 + .../clp_py_utils/clp_metadata_db_utils.py | 21 ++++++----------- .../clp_py_utils/create-db-tables.py | 11 +++++++++ .../initialize-clp-metadata-db.py | 23 +++++++++++++++---- 4 files changed, 37 insertions(+), 19 deletions(-) diff --git a/components/clp-package-utils/clp_package_utils/scripts/start_clp.py b/components/clp-package-utils/clp_package_utils/scripts/start_clp.py index 525ddd54d8..660458744b 100755 --- a/components/clp-package-utils/clp_package_utils/scripts/start_clp.py +++ b/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 diff --git a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py index fae06060d3..0517256721 100644 --- a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py +++ b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py @@ -1,8 +1,5 @@ import typing -from mariadb import Cursor as MariaDbCursor -from mysql.connector.cursor import MySQLCursor - from clp_py_utils.clp_config import ( ARCHIVE_TAGS_TABLE_SUFFIX, ARCHIVES_TABLE_SUFFIX, @@ -13,7 +10,7 @@ ) -def _create_archives_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str) -> None: +def _create_archives_table(db_cursor, table_prefix: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_prefix}{ARCHIVES_TABLE_SUFFIX}` ( @@ -33,7 +30,7 @@ def _create_archives_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: ) -def _create_tags_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str) -> None: +def _create_tags_table(db_cursor, table_prefix: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_prefix}{TAGS_TABLE_SUFFIX}` ( @@ -46,7 +43,7 @@ def _create_tags_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str ) -def _create_archive_tags_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str) -> None: +def _create_archive_tags_table(db_cursor, table_prefix: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_prefix}{ARCHIVE_TAGS_TABLE_SUFFIX}` ( @@ -60,7 +57,7 @@ def _create_archive_tags_table(db_cursor: MariaDbCursor | MySQLCursor, table_pre ) -def _create_files_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str) -> None: +def _create_files_table(db_cursor, table_prefix: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_prefix}{FILES_TABLE_SUFFIX}` ( @@ -81,9 +78,7 @@ def _create_files_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: st ) -def _create_column_metadata_table( - db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str -) -> None: +def _create_column_metadata_table(db_cursor, table_prefix: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_prefix}{COLUMN_METADATA_TABLE_SUFFIX}` ( @@ -95,7 +90,7 @@ def _create_column_metadata_table( ) -def create_datasets_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str) -> None: +def create_datasets_table(db_cursor, table_prefix: str) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_prefix}{DATASETS_TABLE_SUFFIX}` ( @@ -107,9 +102,7 @@ def create_datasets_table(db_cursor: MariaDbCursor | MySQLCursor, table_prefix: ) -def create_metadata_db_tables( - db_cursor: MariaDbCursor | MySQLCursor, table_prefix: str, dataset: str | None = None -) -> None: +def create_metadata_db_tables(db_cursor, table_prefix: str, dataset: str | None = None) -> None: if dataset is not None: table_prefix = f"{table_prefix}{dataset}_" _create_column_metadata_table(db_cursor, table_prefix) diff --git a/components/clp-py-utils/clp_py_utils/create-db-tables.py b/components/clp-py-utils/clp_py_utils/create-db-tables.py index b376fdf640..1fb3f07c6f 100644 --- a/components/clp-py-utils/clp_py_utils/create-db-tables.py +++ b/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,9 +20,17 @@ 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() @@ -28,6 +38,7 @@ def main(argv): 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) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 5a77f0cd5d..20e345a2d3 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -6,9 +6,9 @@ from sql_adapter import SQL_Adapter -from clp_py_utils.clp_config import Database -from clp_py_utils.core import read_yaml_config_file +from clp_py_utils.clp_config import Database, StorageEngine from clp_py_utils.clp_metadata_db_utils.py import create_metadata_db_tables +from clp_py_utils.core import read_yaml_config_file # Setup logging # Create logger @@ -24,19 +24,32 @@ 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="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) + 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: - create_metadata_db_tables(metadata_db_cursor, table_prefix) + if StorageEngine.CLP_S == storage_engine: + create_metadata_db_tables(metadata_db_cursor, table_prefix, dataset="default") + else: + create_metadata_db_tables(metadata_db_cursor, table_prefix) metadata_db.commit() except: logger.exception("Failed to create clp metadata tables.") From 5f2cde5e4b2544fd666dea56ef26d3faf22c9846 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Tue, 22 Apr 2025 12:02:07 -0400 Subject: [PATCH 27/36] Syntax fix --- .../clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 20e345a2d3..cd1d30a090 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -1,13 +1,14 @@ #!/usr/bin/env python3 import argparse import logging +import pathlib import sys from contextlib import closing from sql_adapter import SQL_Adapter from clp_py_utils.clp_config import Database, StorageEngine -from clp_py_utils.clp_metadata_db_utils.py import create_metadata_db_tables +from clp_py_utils.clp_metadata_db_utils import create_metadata_db_tables from clp_py_utils.core import read_yaml_config_file # Setup logging From b62e26a0f61e4af676633f0a8e7c0cb73e7f7c6b Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Tue, 22 Apr 2025 12:06:32 -0400 Subject: [PATCH 28/36] Add dataset TODO --- .../clp_py_utils/initialize-clp-metadata-db.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index cd1d30a090..cc1b778086 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -7,7 +7,11 @@ from sql_adapter import SQL_Adapter -from clp_py_utils.clp_config import Database, StorageEngine +from clp_py_utils.clp_config import ( + CLP_DEFAULT_DATASET_NAME, + Database, + StorageEngine, +) from clp_py_utils.clp_metadata_db_utils import create_metadata_db_tables from clp_py_utils.core import read_yaml_config_file @@ -48,9 +52,11 @@ def main(argv): metadata_db.cursor(dictionary=True) ) as metadata_db_cursor: if StorageEngine.CLP_S == storage_engine: - create_metadata_db_tables(metadata_db_cursor, table_prefix, dataset="default") - else: - create_metadata_db_tables(metadata_db_cursor, table_prefix) + create_metadata_db_tables( + metadata_db_cursor, table_prefix, dataset=CLP_DEFAULT_DATASET_NAME + ) + # TODO: Remove the default db tables for CLP_S after the dataset feature is implemented. + create_metadata_db_tables(metadata_db_cursor, table_prefix) metadata_db.commit() except: logger.exception("Failed to create clp metadata tables.") From 58d46da431f73ef25b146cabbc35b2c4dcb39276 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Tue, 22 Apr 2025 12:34:05 -0400 Subject: [PATCH 29/36] Add dataset table creation --- .../clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index cc1b778086..ed773a2c3b 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -12,7 +12,10 @@ Database, StorageEngine, ) -from clp_py_utils.clp_metadata_db_utils import create_metadata_db_tables +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 @@ -55,6 +58,7 @@ def main(argv): create_metadata_db_tables( metadata_db_cursor, table_prefix, dataset=CLP_DEFAULT_DATASET_NAME ) + create_datasets_table(metadata_db_cursor, table_prefix) # TODO: Remove the default db tables for CLP_S after the dataset feature is implemented. create_metadata_db_tables(metadata_db_cursor, table_prefix) metadata_db.commit() From 32c369dcb5b22cf2d0a4464240b8cd45be18418a Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Tue, 22 Apr 2025 12:45:22 -0400 Subject: [PATCH 30/36] Add todo for dataset table default entries --- .../clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py | 1 + 1 file changed, 1 insertion(+) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index ed773a2c3b..505811a4b0 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -58,6 +58,7 @@ def main(argv): create_metadata_db_tables( metadata_db_cursor, table_prefix, dataset=CLP_DEFAULT_DATASET_NAME ) + # TODO: populate the dataset table with proper default name and default path. create_datasets_table(metadata_db_cursor, table_prefix) # TODO: Remove the default db tables for CLP_S after the dataset feature is implemented. create_metadata_db_tables(metadata_db_cursor, table_prefix) From 84ee7375437414b86da6503c9ec1642574d39228 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Tue, 22 Apr 2025 13:15:23 -0400 Subject: [PATCH 31/36] Add docstring --- .../clp_py_utils/clp_metadata_db_utils.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py index 0517256721..96ad27e762 100644 --- a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py +++ b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py @@ -91,6 +91,11 @@ def _create_column_metadata_table(db_cursor, table_prefix: str) -> None: def create_datasets_table(db_cursor, table_prefix: str) -> None: + """ + Creates the table to store the datasets information (for clp-s only). + :param db_cursor: The database cursor to execute the table creation. + :param table_prefix: A string prefix applied to the dataset table name. + """ db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_prefix}{DATASETS_TABLE_SUFFIX}` ( @@ -103,6 +108,13 @@ def create_datasets_table(db_cursor, table_prefix: str) -> None: def create_metadata_db_tables(db_cursor, table_prefix: str, dataset: str | None = None) -> None: + """ + Creates the standard set of SQL tables in the metadata database for recording compression tasks' + results. + :param db_cursor: The database cursor to execute the table creations. + :param table_prefix: A string prefix applied to all table names. + :param dataset: If provided, dataset-specific set of tables are created (for clp-s only). + """ if dataset is not None: table_prefix = f"{table_prefix}{dataset}_" _create_column_metadata_table(db_cursor, table_prefix) From 6fd25369de2fe792f8d71da9c84de1bb964b62a2 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Wed, 23 Apr 2025 09:09:51 -0400 Subject: [PATCH 32/36] Apply suggestions from code review Co-authored-by: kirkrodrigues <2454684+kirkrodrigues@users.noreply.github.com> --- .../clp_py_utils/clp_metadata_db_utils.py | 15 ++++++++------- .../clp_py_utils/initialize-clp-metadata-db.py | 2 +- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py index 96ad27e762..06c580124d 100644 --- a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py +++ b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py @@ -1,4 +1,3 @@ -import typing from clp_py_utils.clp_config import ( ARCHIVE_TAGS_TABLE_SUFFIX, @@ -92,9 +91,10 @@ def _create_column_metadata_table(db_cursor, table_prefix: str) -> None: def create_datasets_table(db_cursor, table_prefix: str) -> None: """ - Creates the table to store the datasets information (for clp-s only). + Creates the dataset information table. + :param db_cursor: The database cursor to execute the table creation. - :param table_prefix: A string prefix applied to the dataset table name. + :param table_prefix: A string to prepend to the table name. """ db_cursor.execute( f""" @@ -109,11 +109,12 @@ def create_datasets_table(db_cursor, table_prefix: str) -> None: def create_metadata_db_tables(db_cursor, table_prefix: str, dataset: str | None = None) -> None: """ - Creates the standard set of SQL tables in the metadata database for recording compression tasks' - results. + 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 prefix applied to all table names. - :param dataset: If provided, dataset-specific set of tables are created (for clp-s only). + :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 and the + column-metadata table will be created. """ if dataset is not None: table_prefix = f"{table_prefix}{dataset}_" diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 505811a4b0..121af8bd3d 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -37,7 +37,7 @@ def main(argv): type=str, choices=[engine.value for engine in StorageEngine], required=True, - help="Compression storage engine to use.", + help="Storage engine to create tables for.", ) parsed_args = args_parser.parse_args(argv[1:]) From ab2654acf7ef220ff5892ed36430714db869cca7 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Wed, 23 Apr 2025 09:15:43 -0400 Subject: [PATCH 33/36] Address review concerns --- .../clp_py_utils/clp_metadata_db_utils.py | 16 ++++++++++++---- .../clp_py_utils/initialize-clp-metadata-db.py | 4 ++-- 2 files changed, 14 insertions(+), 6 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py index 06c580124d..cf476b06af 100644 --- a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py +++ b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py @@ -1,3 +1,4 @@ +from __future__ import annotations from clp_py_utils.clp_config import ( ARCHIVE_TAGS_TABLE_SUFFIX, @@ -42,15 +43,17 @@ def _create_tags_table(db_cursor, table_prefix: str) -> None: ) -def _create_archive_tags_table(db_cursor, table_prefix: str) -> None: +def _create_archive_tags_table( + db_cursor, table_prefix: str, archives_table_name: str, tags_table_name: str +) -> None: db_cursor.execute( f""" CREATE TABLE IF NOT EXISTS `{table_prefix}{ARCHIVE_TAGS_TABLE_SUFFIX}` ( `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_TABLE_SUFFIX}` (`id`), - FOREIGN KEY (`tag_id`) REFERENCES `{table_prefix}{TAGS_TABLE_SUFFIX}` (`tag_id`) + FOREIGN KEY (`archive_id`) REFERENCES `{archives_table_name}` (`id`), + FOREIGN KEY (`tag_id`) REFERENCES `{tags_table_name}` (`tag_id`) ) """ ) @@ -122,5 +125,10 @@ def create_metadata_db_tables(db_cursor, table_prefix: str, dataset: str | None _create_archives_table(db_cursor, table_prefix) _create_tags_table(db_cursor, table_prefix) - _create_archive_tags_table(db_cursor, table_prefix) + _create_archive_tags_table( + db_cursor, + table_prefix, + f"{table_prefix}{ARCHIVES_TABLE_SUFFIX}", + f"{table_prefix}{TAGS_TABLE_SUFFIX}", + ) _create_files_table(db_cursor, table_prefix) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 121af8bd3d..4e37a509c2 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -1,9 +1,9 @@ #!/usr/bin/env python3 import argparse import logging -import pathlib import sys from contextlib import closing +from pathlib import Path from sql_adapter import SQL_Adapter @@ -41,7 +41,7 @@ def main(argv): ) parsed_args = args_parser.parse_args(argv[1:]) - config_file_path = pathlib.Path(parsed_args.config) + config_file_path = Path(parsed_args.config) storage_engine = StorageEngine(parsed_args.storage_engine) try: From eaa05506a4fdf9685e7ea0f8d3a860d352a1ec46 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Wed, 23 Apr 2025 10:30:51 -0400 Subject: [PATCH 34/36] Stil create the column metadata for the indexer --- .../clp_py_utils/clp_metadata_db_utils.py | 35 +++++++++++-------- .../initialize-clp-metadata-db.py | 11 +++--- 2 files changed, 25 insertions(+), 21 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py index cf476b06af..9e9ea4003f 100644 --- a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py +++ b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py @@ -3,6 +3,7 @@ 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, @@ -10,10 +11,10 @@ ) -def _create_archives_table(db_cursor, table_prefix: str) -> None: +def _create_archives_table(db_cursor, archives_table_name: str) -> None: db_cursor.execute( f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}{ARCHIVES_TABLE_SUFFIX}` ( + 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, @@ -30,10 +31,10 @@ def _create_archives_table(db_cursor, table_prefix: str) -> None: ) -def _create_tags_table(db_cursor, table_prefix: str) -> None: +def _create_tags_table(db_cursor, tags_table_name: str) -> None: db_cursor.execute( f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}{TAGS_TABLE_SUFFIX}` ( + 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, @@ -44,11 +45,11 @@ def _create_tags_table(db_cursor, table_prefix: str) -> None: def _create_archive_tags_table( - db_cursor, table_prefix: str, archives_table_name: str, tags_table_name: str + 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 `{table_prefix}{ARCHIVE_TAGS_TABLE_SUFFIX}` ( + 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`), @@ -80,10 +81,10 @@ def _create_files_table(db_cursor, table_prefix: str) -> None: ) -def _create_column_metadata_table(db_cursor, table_prefix: str) -> None: +def _create_column_metadata_table(db_cursor, table_name: str) -> None: db_cursor.execute( f""" - CREATE TABLE IF NOT EXISTS `{table_prefix}{COLUMN_METADATA_TABLE_SUFFIX}` ( + CREATE TABLE IF NOT EXISTS `{table_name}` ( `name` VARCHAR(512) NOT NULL, `type` TINYINT NOT NULL, PRIMARY KEY (`name`, `type`) @@ -121,14 +122,18 @@ def create_metadata_db_tables(db_cursor, table_prefix: str, dataset: str | None """ if dataset is not None: table_prefix = f"{table_prefix}{dataset}_" - _create_column_metadata_table(db_cursor, table_prefix) - _create_archives_table(db_cursor, table_prefix) - _create_tags_table(db_cursor, table_prefix) + 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}" + 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, - table_prefix, - f"{table_prefix}{ARCHIVES_TABLE_SUFFIX}", - f"{table_prefix}{TAGS_TABLE_SUFFIX}", + 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) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 4e37a509c2..5929e8971e 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -8,7 +8,6 @@ from sql_adapter import SQL_Adapter from clp_py_utils.clp_config import ( - CLP_DEFAULT_DATASET_NAME, Database, StorageEngine, ) @@ -54,13 +53,13 @@ def main(argv): with closing(sql_adapter.create_connection(True)) as metadata_db, closing( metadata_db.cursor(dictionary=True) ) as metadata_db_cursor: + # TODO: After the dataset feature is implemented: + # 1. Populate the `clp-datasets` table with the name and path for the `default` + # dataset. + # 2. Create a set of metadata tables for the `default` dataset. + # 3. Remove the default db tables for CLP_S. if StorageEngine.CLP_S == storage_engine: - create_metadata_db_tables( - metadata_db_cursor, table_prefix, dataset=CLP_DEFAULT_DATASET_NAME - ) - # TODO: populate the dataset table with proper default name and default path. create_datasets_table(metadata_db_cursor, table_prefix) - # TODO: Remove the default db tables for CLP_S after the dataset feature is implemented. create_metadata_db_tables(metadata_db_cursor, table_prefix) metadata_db.commit() except: From bc5ee6e84454be9526d77f78aad23670605b2ab2 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Wed, 23 Apr 2025 10:33:17 -0400 Subject: [PATCH 35/36] add comment --- components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py | 1 + 1 file changed, 1 insertion(+) diff --git a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py index 9e9ea4003f..726a29f11f 100644 --- a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py +++ b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py @@ -126,6 +126,7 @@ def create_metadata_db_tables(db_cursor, table_prefix: str, dataset: str | None 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 the default column metadata table name after updating the indexer. column_metadata_table_name = ( f"{table_prefix}{COLUMN_METADATA_TABLE_SUFFIX}_{CLP_DEFAULT_DATASET_NAME}" ) From 9a82570750cd95920b8ab7342baec4e7670136f9 Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Wed, 23 Apr 2025 14:10:51 -0400 Subject: [PATCH 36/36] Apply suggestions from code review Co-authored-by: kirkrodrigues <2454684+kirkrodrigues@users.noreply.github.com> --- .../clp-py-utils/clp_py_utils/clp_metadata_db_utils.py | 8 +++++--- .../clp_py_utils/initialize-clp-metadata-db.py | 8 +++----- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py index 726a29f11f..f5420aaf16 100644 --- a/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py +++ b/components/clp-py-utils/clp_py_utils/clp_metadata_db_utils.py @@ -117,8 +117,7 @@ def create_metadata_db_tables(db_cursor, table_prefix: str, dataset: str | None :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 and the - column-metadata table will be created. + :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}_" @@ -126,7 +125,10 @@ def create_metadata_db_tables(db_cursor, table_prefix: str, dataset: str | None 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 the default column metadata table name after updating the indexer. + + # 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}" ) diff --git a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py index 5929e8971e..0dc547964e 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-clp-metadata-db.py @@ -53,11 +53,9 @@ def main(argv): with closing(sql_adapter.create_connection(True)) as metadata_db, closing( metadata_db.cursor(dictionary=True) ) as metadata_db_cursor: - # TODO: After the dataset feature is implemented: - # 1. Populate the `clp-datasets` table with the name and path for the `default` - # dataset. - # 2. Create a set of metadata tables for the `default` dataset. - # 3. Remove the default db tables for CLP_S. + # 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)