feat(clp-package): Add Python package config, metadata-DB column, and scheduler handoff for the compression coordinator.

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

@@ -778,6 +778,34 @@ class LogIngestor(BaseModel):
     logging_level: LoggingLevelRust = "INFO"
 
 
+class Spider(BaseModel):
+    host: DomainStr = "localhost"
+    port: Port = 6000
+
+
+class SpiderResourceGroup(BaseModel):
+    name: NonEmptyStr
+
+
+class PollingBackoff(BaseModel):
+    init_backoff_millisecs: PositiveInt
+    max_backoff_millisecs: PositiveInt
+
+
+class CompressionCoordinator(BaseModel):
+    resource_group: SpiderResourceGroup = SpiderResourceGroup(name="compression-coordinator")
+    job_polling_interval_millisecs: PositiveInt = 100
+    result_polling: PollingBackoff = PollingBackoff(
+        init_backoff_millisecs=100, max_backoff_millisecs=1000
+    )
+    compression_task_max_retry: NonNegativeInt = 1
+    commit_task_max_retry: NonNegativeInt = 1
+    database_connection_pool_size: PositiveInt = 10
+    termination_timeout_secs: PositiveInt = 30
+    commit_task_soft_timeout_secs: PositiveInt = 45
+    commit_task_hard_timeout_secs: PositiveInt = 60
+
+
 class Presto(BaseModel):
     DEFAULT_PORT: ClassVar[int] = 8080
 
@@ -830,6 +858,8 @@ class ClpConfig(BaseModel):
     garbage_collector: GarbageCollector = GarbageCollector()
     api_server: ApiServer | None = ApiServer()
     log_ingestor: LogIngestor | None = LogIngestor()
+    spider: Spider | None = None
+    compression_coordinator: CompressionCoordinator | None = None
     credentials_file_path: SerializablePath = CLP_DEFAULT_CREDENTIALS_FILE_PATH
 
     mcp_server: McpServer | None = None
@@ -1029,6 +1059,21 @@ def validate_log_ingestor_config(self):
             raise ValueError(msg)
         return self
 
+    @model_validator(mode="after")
+    def validate_compression_coordinator_config(self):
+        if self.compression_coordinator is None:
+            return self
+        if self.package.storage_engine != StorageEngine.CLP_S:
+            msg = (
+                "compression-coordinator is only compatible with storage engine "
+                f"`{StorageEngine.CLP_S}`."
+            )
+            raise ValueError(msg)
+        if self.spider is None:
+            msg = "compression-coordinator requires Spider to be configured."
+            raise ValueError(msg)
+        return self
+
     @model_validator(mode="after")
     def validate_presto_config(self):
         query_engine = self.webui.query_engine

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

@@ -68,19 +68,23 @@ def main(argv):
                     `status_msg` VARCHAR(512) NOT NULL DEFAULT '',
                     `creation_time` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
                     `start_time` DATETIME(3) NULL DEFAULT NULL,
-                    `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP(),
+                    `update_time` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP()
+                        ON UPDATE CURRENT_TIMESTAMP(),
                     `duration` FLOAT NULL DEFAULT NULL,
                     `original_size` BIGINT NOT NULL DEFAULT '0',
                     `uncompressed_size` BIGINT NOT NULL DEFAULT '0',
                     `compressed_size` BIGINT NOT NULL DEFAULT '0',
                     `num_tasks` INT NOT NULL DEFAULT '0',
                     `num_tasks_completed` INT NOT NULL DEFAULT '0',
                     `clp_binary_version` INT NULL DEFAULT NULL,
+                    `spider_id` BIGINT UNSIGNED NULL DEFAULT NULL,
+                    `dispatch_time` DATETIME NULL DEFAULT NULL,
                     `clp_config` MEDIUMBLOB NOT NULL,
                     PRIMARY KEY (`id`) USING BTREE,
                     INDEX `JOB_STATUS` (`status`) USING BTREE,
                     INDEX `JOB_UPDATE_TIME` (`update_time`) USING BTREE,
-                    INDEX `JOB_START_TIME_STATUS` (`start_time`, `status`) USING BTREE
+                    INDEX `JOB_START_TIME_STATUS` (`start_time`, `status`) USING BTREE,
+                    INDEX `JOB_SPIDER_ID` (`spider_id`) USING BTREE
                 ) ROW_FORMAT=DYNAMIC
                 """
             )

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

@@ -467,6 +467,16 @@ def _schedule_job(
                 )
                 return
         elif input_type == InputType.S3_OBJECT_METADATA.value:
+            if clp_config.compression_coordinator is not None:
+                # NOTE: These jobs are left in PENDING and will eventually be picked up and
+                # scheduled by the compression-coordinator.
+                logger.info(
+                    "compression-coordinator is configured to handle compression jobs submitted"
+                    " by log-ingestor. Skipping job %d.",
+                    job_id,
+                )
+                return
+
             try:
                 _process_s3_object_metadata_input(
                     input_config, paths_to_compress_buffer, db_context