feat(compression-coordinator): Complete S3 compression job handle implementation.

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

@@ -11,7 +11,7 @@
     QueryJobStatus,
     QueryTaskStatus,
 )
-from mysql.connector.errorcode import ER_DUP_KEYNAME
+from mysql.connector.errorcode import ER_DUP_FIELDNAME, ER_DUP_KEYNAME
 from pydantic import ValidationError
 
 from clp_py_utils.clp_config import (
@@ -77,10 +77,12 @@ def main(argv):
                     `num_tasks_completed` INT NOT NULL DEFAULT '0',
                     `clp_binary_version` INT NULL DEFAULT NULL,
                     `clp_config` MEDIUMBLOB NOT NULL,
+                    `spider_id` BIGINT UNSIGNED DEFAULT 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
                 """
             )
@@ -100,6 +102,30 @@ def main(argv):
                 if not (hasattr(err, "errno") and err.errno == ER_DUP_KEYNAME):
                     raise
 
+            # Add Spider fields to existing tables created before compression jobs were submitted
+            # through Spider.
+            try:
+                scheduling_db_cursor.execute(
+                    f"""
+                    ALTER TABLE `{COMPRESSION_JOBS_TABLE_NAME}`
+                    ADD COLUMN `spider_id` BIGINT UNSIGNED DEFAULT NULL
+                    """
+                )
+            except Exception as err:
+                if not (hasattr(err, "errno") and err.errno == ER_DUP_FIELDNAME):
+                    raise
+
+            try:
+                scheduling_db_cursor.execute(
+                    f"""
+                    ALTER TABLE `{COMPRESSION_JOBS_TABLE_NAME}`
+                    ADD INDEX `JOB_SPIDER_ID` (`spider_id`) USING BTREE
+                    """
+                )
+            except Exception as err:
+                if not (hasattr(err, "errno") and err.errno == ER_DUP_KEYNAME):
+                    raise
+
             scheduling_db_cursor.execute(
                 f"""
                 CREATE TABLE IF NOT EXISTS `{COMPRESSION_TASKS_TABLE_NAME}` (

components/clp-rust-utils/src/clp_config/package/config.rs

@@ -1,6 +1,9 @@
 use serde::Deserialize;
 
-use crate::clp_config::{AwsAuthentication, S3Config};
+use crate::{
+    clp_config::{AwsAuthentication, S3Config},
+    dataset::resolve_dataset_name,
+};
 
 /// Mirror of `clp_py_utils.clp_config.ClpConfig`.
 ///
@@ -89,14 +92,54 @@ pub struct Database {
     pub host: String,
     pub port: u16,
     pub names: ClpDbNames,
+
+    #[serde(skip)]
+    pub table_prefix: String,
+}
+
+impl Database {
+    /// # Returns
+    ///
+    /// The archives table name `<prefix><dataset>_archives`, where a `None` dataset resolves to
+    /// `default`.
+    #[must_use]
+    pub fn archives_table_name(&self, dataset: Option<&str>) -> String {
+        self.archive_metadata_table_name("archives", dataset)
+    }
+
+    /// # Returns
+    ///
+    /// The column-metadata table name `<prefix><dataset>_column_metadata`, where a `None` dataset
+    /// resolves to `default`.
+    #[must_use]
+    pub fn column_metadata_table_name(&self, dataset: Option<&str>) -> String {
+        self.archive_metadata_table_name("column_metadata", dataset)
+    }
+
+    /// Builds a per-dataset archive-metadata table name.
+    ///
+    /// # Returns
+    ///
+    /// `<prefix><dataset>_<suffix>`, where `dataset` defaults to the `CLP_S` default.
+    fn archive_metadata_table_name(&self, suffix: &str, dataset: Option<&str>) -> String {
+        format!(
+            "{}{}_{suffix}",
+            self.table_prefix,
+            resolve_dataset_name(dataset)
+        )
+    }
 }
 
 impl Default for Database {
     fn default() -> Self {
+        /// Mirror of `clp_py_utils.clp_config.CLP_METADATA_TABLE_PREFIX`.
+        const CLP_METADATA_TABLE_PREFIX: &str = "clp_";
+
         Self {
             host: "localhost".to_owned(),
             port: 3306,
             names: ClpDbNames::default(),
+            table_prefix: CLP_METADATA_TABLE_PREFIX.to_owned(),
         }
     }
 }

components/clp-rust-utils/src/dataset.rs

@@ -2,5 +2,16 @@ use std::sync::LazyLock;
 
 use regex::Regex;
 
+/// The default dataset name (mirror of `clp_py_utils.clp_config.CLP_DEFAULT_DATASET_NAME`).
+pub const CLP_DEFAULT_DATASET_NAME: &str = "default";
+
 pub static VALID_DATASET_NAME_REGEX: LazyLock<Regex> =
     LazyLock::new(|| Regex::new(r"^[a-zA-Z0-9_]+$").unwrap());
+
+/// # Returns
+///
+/// `dataset` when set, otherwise the `CLP_S` default dataset name [`CLP_DEFAULT_DATASET_NAME`].
+#[must_use]
+pub fn resolve_dataset_name(dataset: Option<&str>) -> &str {
+    dataset.unwrap_or(CLP_DEFAULT_DATASET_NAME)
+}

components/compression-coordinator/Cargo.toml

@@ -16,6 +16,7 @@ anyhow = "1.0.100"
 async-trait = "0.1.89"
 clap = { version = "4.6.4", features = ["derive"] }
 clp-rust-utils = { path = "../clp-rust-utils" }
+non-empty-string = "0.2.6"
 rmp-serde = "1.3.1"
 serde = { version = "1.0.228", features = ["derive"] }
 spider-client = { git = "https://github.com/y-scope/spider.git", branch = "main" }
@@ -25,6 +26,3 @@ strsim = "0.11.1"
 thiserror = "2.0.18"
 tokio = { version = "1.52.3", features = ["time", "rt-multi-thread"] }
 tracing = "0.1.44"
-
-[dev-dependencies]
-non-empty-string = "0.2.6"

components/compression-coordinator/src/error.rs

@@ -1,11 +1,46 @@
 //! The crate-level error type for the compression coordinator.
 
+use clp_rust_utils::{job_config::ingestion::JobId as IngestionJobId, s3::S3ObjectMetadataId};
+
 /// Errors returned by the compression coordinator.
 #[derive(Debug, thiserror::Error)]
 pub enum Error {
+    #[error(
+        "duplicate S3 object metadata IDs {ids:?} requested for ingestion job {ingestion_job_id}"
+    )]
+    DuplicateS3ObjectMetadata {
+        ingestion_job_id: IngestionJobId,
+        ids: Vec<S3ObjectMetadataId>,
+    },
+
+    #[error("S3 object metadata {id} has an empty `{field}`")]
+    EmptyS3ObjectMetadataField {
+        id: S3ObjectMetadataId,
+        field: &'static str,
+    },
+
     #[error("invalid dataset: {0}")]
     InvalidDataset(String),
 
+    #[error("failed to create metadata table `{table}`: {source}")]
+    MetadataTableCreation {
+        table: String,
+        #[source]
+        source: sqlx::Error,
+    },
+
+    #[error("missing S3 object metadata {id} for ingestion job {ingestion_job_id}")]
+    MissingS3ObjectMetadata {
+        ingestion_job_id: IngestionJobId,
+        id: S3ObjectMetadataId,
+    },
+
+    #[error("no S3 object metadata was requested for ingestion job {0}")]
+    NoS3ObjectMetadata(IngestionJobId),
+
+    #[error("no S3 objects were partitioned into compression task inputs")]
+    NoTaskInputs,
+
     #[error("S3 bucket mismatch: expected `{0}`, but got `{1}`")]
     S3BucketMismatch(String, String),
 
@@ -26,6 +61,9 @@ pub enum Error {
     #[error("failed to serialize a task input: {0}")]
     TaskInputSerialization(#[from] rmp_serde::encode::Error),
 
+    #[error("number of compression tasks {0} exceeds `i32::MAX`")]
+    TooManyCompressionTasks(usize),
+
     #[error("unsupported input config")]
     UnsupportedInputConfig,
 }