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

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

@@ -169,6 +169,18 @@ impl Database {
         )
     }
 
+    /// # Returns
+    ///
+    /// The column-metadata table name (`<prefix><dataset>_column_metadata`).
+    #[must_use]
+    pub fn column_metadata_table_name(&self, dataset: Option<&str>) -> String {
+        format!(
+            "{}{}_column_metadata",
+            self.table_prefix,
+            resolve_dataset_name(dataset)
+        )
+    }
+
     /// # Returns
     ///
     /// The datasets table name `<prefix>datasets`.

components/compression-coordinator/Cargo.toml

@@ -16,6 +16,8 @@ anyhow = "1.0.100"
 async-trait = "0.1.89"
 clap = { version = "4.6.4", features = ["derive"] }
 clp-rust-utils = { path = "../clp-rust-utils" }
+const_format = "0.2.35"
+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 +27,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,
 }