fix(compression_scheduler): Only fetch datasets when using the clp-s storage engine (fixes #1214). (#1215)

Co-authored-by: Bingran Hu <[email protected]>

Author: kirkrodrigues

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

@@ -154,23 +154,25 @@ def _process_s3_input(
 
 
 def search_and_schedule_new_tasks(
+    clp_config: CLPConfig,
     db_conn,
     db_cursor,
     clp_metadata_db_connection_config: Dict[str, Any],
-    clp_archive_output: ArchiveOutput,
 ):
     """
     For all jobs with PENDING status, splits the job into tasks and schedules them.
+    :param clp_config:
     :param db_conn:
     :param db_cursor:
     :param clp_metadata_db_connection_config:
-    :param clp_archive_output:
     """
     global scheduled_jobs
 
-    existing_datasets = fetch_existing_datasets(
-        db_cursor, clp_metadata_db_connection_config["table_prefix"]
-    )
+    existing_datasets: Set[str] = set()
+    if StorageEngine.CLP_S == clp_config.package.storage_engine:
+        existing_datasets = fetch_existing_datasets(
+            db_cursor, clp_metadata_db_connection_config["table_prefix"]
+        )
 
     logger.debug("Search and schedule new tasks")
 
@@ -193,7 +195,7 @@ def search_and_schedule_new_tasks(
                 db_cursor,
                 table_prefix,
                 dataset,
-                clp_archive_output,
+                clp_config.archive_output,
             )
 
             # NOTE: This assumes we never delete a dataset when compression jobs are being scheduled
@@ -428,10 +430,10 @@ def main(argv):
         while True:
             try:
                 search_and_schedule_new_tasks(
+                    clp_config,
                     db_conn,
                     db_cursor,
                     clp_metadata_db_connection_config,
-                    clp_config.archive_output,
                 )
                 poll_running_jobs(db_conn, db_cursor)
                 time.sleep(clp_config.compression_scheduler.jobs_poll_delay)