From b07a473bac7c3aefc94b2eb777f46b2eb289f38d Mon Sep 17 00:00:00 2001 From: Bingran Hu Date: Thu, 23 Jul 2026 14:26:44 -0400 Subject: [PATCH 01/10] impl --- .../initialize-orchestration-db.py | 30 ++- .../src/clp_config/package/config.rs | 45 +++- components/clp-rust-utils/src/dataset.rs | 11 + components/compression-coordinator/Cargo.toml | 4 +- .../compression-coordinator/src/error.rs | 31 +++ .../compression-coordinator/src/job_handle.rs | 252 ++++++++++++++++-- .../compression-coordinator/src/partition.rs | 10 + 7 files changed, 350 insertions(+), 33 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py b/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py index 85dac38325..6d91ea1133 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py +++ b/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}` ( diff --git a/components/clp-rust-utils/src/clp_config/package/config.rs b/components/clp-rust-utils/src/clp_config/package/config.rs index 29863686b0..45439942d4 100644 --- a/components/clp-rust-utils/src/clp_config/package/config.rs +++ b/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 `_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 `_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 + /// + /// `_`, 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(), } } } diff --git a/components/clp-rust-utils/src/dataset.rs b/components/clp-rust-utils/src/dataset.rs index e069dc5dfc..c76b50d357 100644 --- a/components/clp-rust-utils/src/dataset.rs +++ b/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 = 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) +} diff --git a/components/compression-coordinator/Cargo.toml b/components/compression-coordinator/Cargo.toml index 134d11a53c..8712993d06 100644 --- a/components/compression-coordinator/Cargo.toml +++ b/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" diff --git a/components/compression-coordinator/src/error.rs b/components/compression-coordinator/src/error.rs index 129503bbf4..d9811d1726 100644 --- a/components/compression-coordinator/src/error.rs +++ b/components/compression-coordinator/src/error.rs @@ -3,9 +3,37 @@ /// Errors returned by the compression coordinator. #[derive(Debug, thiserror::Error)] pub enum Error { + #[error("S3 object metadata {metadata_id} has an empty {field}")] + EmptyS3ObjectMetadataField { + metadata_id: u64, + 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 for ingestion job {ingestion_job_id}; missing IDs: \ + {metadata_ids:?}" + )] + MissingS3ObjectMetadata { + ingestion_job_id: u64, + metadata_ids: Vec, + }, + + #[error("no S3 object metadata was requested for ingestion job {0}")] + NoS3ObjectMetadata(u64), + + #[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 +54,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, } diff --git a/components/compression-coordinator/src/job_handle.rs b/components/compression-coordinator/src/job_handle.rs index 8d47da8a41..1ec3925ae4 100644 --- a/components/compression-coordinator/src/job_handle.rs +++ b/components/compression-coordinator/src/job_handle.rs @@ -1,20 +1,29 @@ //! Handle for driving a single S3 compression job to completion. -use std::{sync::Arc, time::Duration}; +use std::{collections::HashSet, sync::Arc, time::Duration}; use clp_rust_utils::{ clp_config::package::config::Database, dataset::VALID_DATASET_NAME_REGEX, job_config::{ClpIoConfig, CompressionJobId, CompressionJobStatus, InputConfig}, + s3::{ObjectMetadata, S3ObjectMetadataId}, task_io::compression::{ClpSCompressionOption, S3InputSource}, }; +use non_empty_string::NonEmptyString; use spider_core::{ task::ExecutionPolicy, types::id::{JobId as SpiderJobId, ResourceGroupId}, }; use sqlx::MySqlPool; -use crate::{Error, compression_job_submitter::S3CompressionJobSubmitter}; +use crate::{ + Error, + compression_job_submitter::{CompressionJobOutcome, S3CompressionJobSubmitter}, + partition::CompressionInputBuilder, +}; + +const COMPRESSION_JOB_TABLE_NAME: &str = "compression_jobs"; +const INGESTED_S3_OBJECT_METADATA_TABLE_NAME: &str = "ingested_s3_object_metadata"; /// Options for a compression job running in Spider. pub struct SpiderOption { @@ -30,16 +39,16 @@ pub struct SpiderOption { /// /// * `SubmitterType` - The type of the job submitter for Spider job submission. pub struct S3CompressionJobHandle { - _db_pool: MySqlPool, - _db_config: Database, + db_pool: MySqlPool, + db_config: Database, compression_job_id: CompressionJobId, job_submitter: SubmitterType, resource_group_id: ResourceGroupId, - _input_config: InputConfig, + input_config: InputConfig, clp_s_compression_option: ClpSCompressionOption, dataset: Option, - _target_archive_size: u64, + target_archive_size: u64, spider_option: Arc, } @@ -95,15 +104,15 @@ impl S3CompressionJobHandle S3CompressionJobHandle Result<(), Error> { let input_sources = self.prepare_task_inputs().await?; + let num_tasks = input_sources.len(); self.upsert_metadata_tables().await?; @@ -195,7 +205,8 @@ impl S3CompressionJobHandle S3CompressionJobHandle S3CompressionJobHandle Result, Error> { - todo!("implement me!") + let InputConfig::S3ObjectMetadataInputConfig { config } = &self.input_config else { + return Err(Error::UnsupportedInputConfig); + }; + let metadata_ids = &config.s3_object_metadata_ids; + if metadata_ids.is_empty() { + return Err(Error::NoS3ObjectMetadata(config.ingestion_job_id)); + } + + let mut query_builder = sqlx::QueryBuilder::::new(format!( + "SELECT `id`, `bucket`, `key`, `size` FROM \ + `{INGESTED_S3_OBJECT_METADATA_TABLE_NAME}` WHERE `id` IN (" + )); + let mut separated_ids = query_builder.separated(", "); + for metadata_id in metadata_ids { + separated_ids.push_bind(metadata_id); + } + query_builder + .push(") AND `ingestion_job_id` = ") + .push_bind(config.ingestion_job_id); + + let metadata_rows = query_builder + .build_query_as::<(S3ObjectMetadataId, String, String, u64)>() + .fetch_all(&self.db_pool) + .await?; + + let returned_ids: HashSet = metadata_rows + .iter() + .map(|(metadata_id, ..)| *metadata_id) + .collect(); + let requested_ids: HashSet = + metadata_ids.iter().copied().collect(); + let mut missing_ids: Vec = + requested_ids.difference(&returned_ids).copied().collect(); + missing_ids.sort_unstable(); + if !missing_ids.is_empty() { + return Err(Error::MissingS3ObjectMetadata { + ingestion_job_id: config.ingestion_job_id, + metadata_ids: missing_ids, + }); + } + + let mut input_builder = CompressionInputBuilder::from_s3_config( + config.s3_config.clone(), + self.target_archive_size, + ); + for (metadata_id, bucket, key, size) in metadata_rows { + let bucket = NonEmptyString::new(bucket).map_err(|_| { + Error::EmptyS3ObjectMetadataField { + metadata_id, + field: "bucket", + } + })?; + let key = + NonEmptyString::new(key).map_err(|_| Error::EmptyS3ObjectMetadataField { + metadata_id, + field: "key", + })?; + input_builder.add(ObjectMetadata { bucket, key, size })?; + } + + let input_sources = input_builder.into_task_input_sources(); + if input_sources.is_empty() { + return Err(Error::NoTaskInputs); + } + + Ok(input_sources + .into_iter() + .map(|input_source| { + let execution_policy = ExecutionPolicy { + max_num_retry: self.spider_option.compression_task_max_retry, + ..ExecutionPolicy::default() + }; + (input_source, execution_policy) + }) + .collect()) } /// Ensures that the required metadata tables exist for the configured dataset. @@ -265,9 +355,50 @@ impl S3CompressionJobHandle Result<(), Error> { - todo!("implement me!") + let archives_table = self.db_config.archives_table_name(self.dataset.as_deref()); + let column_metadata_table = self + .db_config + .column_metadata_table_name(self.dataset.as_deref()); + + sqlx::query(&format!( + "CREATE TABLE IF NOT EXISTS `{archives_table}` ( + `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`) + )" + )) + .execute(&self.db_pool) + .await + .map_err(|source| Error::MetadataTableCreation { + table: archives_table, + source, + })?; + + sqlx::query(&format!( + "CREATE TABLE IF NOT EXISTS `{column_metadata_table}` ( + `name` VARCHAR(512) NOT NULL, + `type` TINYINT NOT NULL, + PRIMARY KEY (`name`, `type`) + )" + )) + .execute(&self.db_pool) + .await + .map_err(|source| Error::MetadataTableCreation { + table: column_metadata_table, + source, + })?; + + Ok(()) } /// Persists the Spider job ID and marks the compression job as running. @@ -279,9 +410,28 @@ impl S3CompressionJobHandle Result<(), Error> { - todo!("implement me!") + /// * [`Error::TooManyCompressionTasks`] if `num_tasks` exceeds the database column's range. + /// * [`Error::Sqlx`] if the update fails. + async fn persist_spider_job_id( + &self, + spider_job_id: SpiderJobId, + num_tasks: usize, + ) -> Result<(), Error> { + let num_tasks = + i32::try_from(num_tasks).map_err(|_| Error::TooManyCompressionTasks(num_tasks))?; + sqlx::query(&format!( + "UPDATE `{COMPRESSION_JOB_TABLE_NAME}` SET `spider_id` = ?, `status` = ?, \ + `num_tasks` = ?, `start_time` = CURRENT_TIMESTAMP(3), \ + `update_time` = CURRENT_TIMESTAMP() WHERE `id` = ?" + )) + .bind(spider_job_id.get()) + .bind(i32::from(CompressionJobStatus::Running)) + .bind(num_tasks) + .bind(self.compression_job_id) + .execute(&self.db_pool) + .await?; + + Ok(()) } /// Waits for the associated Spider job to complete and finalizes the compression job. @@ -293,9 +443,46 @@ impl S3CompressionJobHandle Result<(), Error> { - todo!("implement me!") + /// * Forwards [`S3CompressionJobSubmitter::run_s3_compression_job_to_completion`]'s return + /// values on failure. + /// * Forwards [`Self::update_job_status`]'s return values on failure. + async fn to_completion(&self, spider_job_id: SpiderJobId) -> Result<(), Error> { + let outcome = self + .job_submitter + .run_s3_compression_job_to_completion( + spider_job_id, + self.spider_option.initial_poll_backoff, + self.spider_option.max_poll_backoff, + ) + .await?; + tracing::info!( + compression_job_id = % self.compression_job_id, + spider_job_id = % spider_job_id, + outcome = ?outcome, + "Compression job reached a terminal state.", + ); + + match outcome { + // The commit task records the successful CLP job status and publishes the archives in + // the same transaction. + CompressionJobOutcome::Succeeded => Ok(()), + CompressionJobOutcome::Failed { error_message } => { + self.update_job_status( + CompressionJobStatus::Failed, + Some(format!( + "The Spider compression job failed: {error_message}" + )), + ) + .await + } + CompressionJobOutcome::Cancelled => { + self.update_job_status( + CompressionJobStatus::Killed, + Some("The Spider compression job was cancelled.".to_owned()), + ) + .await + } + } } /// Updates the compression job status in the CLP database. @@ -304,12 +491,23 @@ impl S3CompressionJobHandle, + job_status: CompressionJobStatus, + status_message: Option, ) -> Result<(), Error> { - todo!("implement me!") + let status_message = status_message.as_ref().map_or("", String::as_str); + sqlx::query(&format!( + "UPDATE `{COMPRESSION_JOB_TABLE_NAME}` SET `status` = ?, `status_msg` = ?, \ + `update_time` = CURRENT_TIMESTAMP() WHERE `id` = ?" + )) + .bind(i32::from(job_status)) + .bind(status_message) + .bind(self.compression_job_id) + .execute(&self.db_pool) + .await?; + + Ok(()) } } diff --git a/components/compression-coordinator/src/partition.rs b/components/compression-coordinator/src/partition.rs index 75f85707f1..2ed31936d5 100644 --- a/components/compression-coordinator/src/partition.rs +++ b/components/compression-coordinator/src/partition.rs @@ -38,6 +38,16 @@ impl CompressionInputBuilder { InputConfig::S3ObjectMetadataInputConfig { config } => config.s3_config.clone(), }; + Self::from_s3_config(s3_config, target_archive_size) + } + + /// Creates an empty builder from the S3 input settings and target archive size. + /// + /// # Returns + /// + /// A newly created [`CompressionInputBuilder`] with an empty buffer. + #[must_use] + pub(crate) fn from_s3_config(s3_config: S3Config, target_archive_size: u64) -> Self { Self { buffer: Vec::new(), partitioned_task_inputs: Vec::new(), From 0ef11ac575f749cdf236483609f5f29a6f675c6c Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Sun, 26 Jul 2026 22:14:42 -0400 Subject: [PATCH 02/10] Include dispatch_time column. --- .../clp-py-utils/clp_py_utils/initialize-orchestration-db.py | 1 + 1 file changed, 1 insertion(+) diff --git a/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py b/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py index 6d91ea1133..839c97ebf6 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py @@ -78,6 +78,7 @@ def main(argv): `clp_binary_version` INT NULL DEFAULT NULL, `clp_config` MEDIUMBLOB NOT NULL, `spider_id` BIGINT UNSIGNED DEFAULT NULL, + `dispatch_time` DATETIME NULL DEFAULT NULL, PRIMARY KEY (`id`) USING BTREE, INDEX `JOB_STATUS` (`status`) USING BTREE, INDEX `JOB_UPDATE_TIME` (`update_time`) USING BTREE, From 8a6be0ae5a95c3386405435768682a28a8688c0c Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Sun, 26 Jul 2026 22:16:50 -0400 Subject: [PATCH 03/10] Revert "Include dispatch_time column." This reverts commit 0ef11ac575f749cdf236483609f5f29a6f675c6c. --- .../clp-py-utils/clp_py_utils/initialize-orchestration-db.py | 1 - 1 file changed, 1 deletion(-) diff --git a/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py b/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py index 839c97ebf6..6d91ea1133 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py @@ -78,7 +78,6 @@ def main(argv): `clp_binary_version` INT NULL DEFAULT NULL, `clp_config` MEDIUMBLOB NOT NULL, `spider_id` BIGINT UNSIGNED DEFAULT NULL, - `dispatch_time` DATETIME NULL DEFAULT NULL, PRIMARY KEY (`id`) USING BTREE, INDEX `JOB_STATUS` (`status`) USING BTREE, INDEX `JOB_UPDATE_TIME` (`update_time`) USING BTREE, From 1a2a2ec4eb429d1ecbcee1b33aaadeab68ef33ed Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Sun, 26 Jul 2026 23:48:22 -0400 Subject: [PATCH 04/10] Review and polish . --- .../compression-coordinator/src/error.rs | 25 +-- .../compression-coordinator/src/job_handle.rs | 152 +++++++++++------- .../compression-coordinator/src/partition.rs | 2 +- 3 files changed, 114 insertions(+), 65 deletions(-) diff --git a/components/compression-coordinator/src/error.rs b/components/compression-coordinator/src/error.rs index d9811d1726..f8814abcbb 100644 --- a/components/compression-coordinator/src/error.rs +++ b/components/compression-coordinator/src/error.rs @@ -1,11 +1,21 @@ //! 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("S3 object metadata {metadata_id} has an empty {field}")] + #[error( + "duplicate S3 object metadata IDs {ids:?} requested for ingestion job {ingestion_job_id}" + )] + DuplicateS3ObjectMetadata { + ingestion_job_id: IngestionJobId, + ids: Vec, + }, + + #[error("S3 object metadata {id} has an empty `{field}`")] EmptyS3ObjectMetadataField { - metadata_id: u64, + id: S3ObjectMetadataId, field: &'static str, }, @@ -19,17 +29,14 @@ pub enum Error { source: sqlx::Error, }, - #[error( - "missing S3 object metadata for ingestion job {ingestion_job_id}; missing IDs: \ - {metadata_ids:?}" - )] + #[error("missing S3 object metadata {id} for ingestion job {ingestion_job_id}")] MissingS3ObjectMetadata { - ingestion_job_id: u64, - metadata_ids: Vec, + ingestion_job_id: IngestionJobId, + id: S3ObjectMetadataId, }, #[error("no S3 object metadata was requested for ingestion job {0}")] - NoS3ObjectMetadata(u64), + NoS3ObjectMetadata(IngestionJobId), #[error("no S3 objects were partitioned into compression task inputs")] NoTaskInputs, diff --git a/components/compression-coordinator/src/job_handle.rs b/components/compression-coordinator/src/job_handle.rs index 1ec3925ae4..9e154d0708 100644 --- a/components/compression-coordinator/src/job_handle.rs +++ b/components/compression-coordinator/src/job_handle.rs @@ -1,6 +1,6 @@ //! Handle for driving a single S3 compression job to completion. -use std::{collections::HashSet, sync::Arc, time::Duration}; +use std::{sync::Arc, time::Duration}; use clp_rust_utils::{ clp_config::package::config::Database, @@ -11,7 +11,7 @@ use clp_rust_utils::{ }; use non_empty_string::NonEmptyString; use spider_core::{ - task::ExecutionPolicy, + task::{ExecutionPolicy, TimeoutPolicy}, types::id::{JobId as SpiderJobId, ResourceGroupId}, }; use sqlx::MySqlPool; @@ -142,7 +142,6 @@ impl S3CompressionJobHandle S3CompressionJobHandle S3CompressionJobHandle Result, Error> { + // NOTE: We batch the metadata fetch to avoid hitting the maximum placeholder limit of + // MySQL. + const FETCH_CHUNK_SIZE: usize = 1000; + let InputConfig::S3ObjectMetadataInputConfig { config } = &self.input_config else { - return Err(Error::UnsupportedInputConfig); + unreachable!("the input config is validated in the factory") }; - let metadata_ids = &config.s3_object_metadata_ids; - if metadata_ids.is_empty() { + if config.s3_object_metadata_ids.is_empty() { return Err(Error::NoS3ObjectMetadata(config.ingestion_job_id)); } - let mut query_builder = sqlx::QueryBuilder::::new(format!( - "SELECT `id`, `bucket`, `key`, `size` FROM \ - `{INGESTED_S3_OBJECT_METADATA_TABLE_NAME}` WHERE `id` IN (" - )); - let mut separated_ids = query_builder.separated(", "); - for metadata_id in metadata_ids { - separated_ids.push_bind(metadata_id); - } - query_builder - .push(") AND `ingestion_job_id` = ") - .push_bind(config.ingestion_job_id); - - let metadata_rows = query_builder - .build_query_as::<(S3ObjectMetadataId, String, String, u64)>() - .fetch_all(&self.db_pool) - .await?; - - let returned_ids: HashSet = metadata_rows - .iter() - .map(|(metadata_id, ..)| *metadata_id) - .collect(); - let requested_ids: HashSet = - metadata_ids.iter().copied().collect(); - let mut missing_ids: Vec = - requested_ids.difference(&returned_ids).copied().collect(); - missing_ids.sort_unstable(); - if !missing_ids.is_empty() { - return Err(Error::MissingS3ObjectMetadata { + let mut sorted_metadata_ids = config.s3_object_metadata_ids.clone(); + sorted_metadata_ids.sort_unstable(); + let mut duplicate_ids: Vec = Vec::new(); + sorted_metadata_ids.dedup_by(|a, b| { + if *a != *b { + return false; + } + if duplicate_ids.last().copied() != Some(*b) { + duplicate_ids.push(*b); + } + true + }); + if !duplicate_ids.is_empty() { + return Err(Error::DuplicateS3ObjectMetadata { ingestion_job_id: config.ingestion_job_id, - metadata_ids: missing_ids, + ids: duplicate_ids, }); } @@ -312,19 +300,54 @@ impl S3CompressionJobHandle::new(format!( + "SELECT `id`, `bucket`, `key`, `size` FROM \ + `{INGESTED_S3_OBJECT_METADATA_TABLE_NAME}` WHERE `id` IN (" + )); + let mut separated_ids = query_builder.separated(", "); + for metadata_id in chunk { + separated_ids.push_bind(metadata_id); + } + query_builder + .push(") AND `ingestion_job_id` = ") + .push_bind(config.ingestion_job_id); + query_builder.push(" ORDER BY `id` ASC"); + + let metadata_rows = query_builder + .build_query_as::() + .fetch_all(&self.db_pool) + .await?; + + let mut row_iter = metadata_rows.into_iter().peekable(); + for &expected_id in chunk { + if row_iter.peek().is_none_or(|row| row.id != expected_id) { + return Err(Error::MissingS3ObjectMetadata { + ingestion_job_id: config.ingestion_job_id, + id: expected_id, + }); } - })?; - let key = - NonEmptyString::new(key).map_err(|_| Error::EmptyS3ObjectMetadataField { - metadata_id, - field: "key", + let row = row_iter + .next() + .expect("row should have been validated by the previous peek"); + let bucket = NonEmptyString::new(row.bucket).map_err(|_| { + Error::EmptyS3ObjectMetadataField { + id: row.id, + field: "bucket", + } + })?; + let key = NonEmptyString::new(row.key).map_err(|_| { + Error::EmptyS3ObjectMetadataField { + id: row.id, + field: "key", + } + })?; + input_builder.add(ObjectMetadata { + bucket, + key, + size: row.size, })?; - input_builder.add(ObjectMetadata { bucket, key, size })?; + } } let input_sources = input_builder.into_task_input_sources(); @@ -335,8 +358,18 @@ impl S3CompressionJobHandle S3CompressionJobHandle S3CompressionJobHandle Self { + pub(crate) const fn from_s3_config(s3_config: S3Config, target_archive_size: u64) -> Self { Self { buffer: Vec::new(), partitioned_task_inputs: Vec::new(), From 107d3cb10825af6bebf924fbabc8855229630989 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Mon, 27 Jul 2026 15:40:08 -0400 Subject: [PATCH 05/10] Review the rest. --- Cargo.lock | 1 + .../src/job_config/compression.rs | 1 + components/compression-coordinator/Cargo.toml | 1 + .../compression-coordinator/src/job_handle.rs | 57 +++++++++++++++---- 4 files changed, 49 insertions(+), 11 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 640fcda85e..717772942f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1009,6 +1009,7 @@ dependencies = [ "async-trait", "clap", "clp-rust-utils", + "const_format", "non-empty-string", "rmp-serde", "serde", diff --git a/components/clp-rust-utils/src/job_config/compression.rs b/components/clp-rust-utils/src/job_config/compression.rs index ddebf2f880..99b73a12d6 100644 --- a/components/clp-rust-utils/src/job_config/compression.rs +++ b/components/clp-rust-utils/src/job_config/compression.rs @@ -18,6 +18,7 @@ pub type CompressionJobId = i32; Serialize, ToSchema, TryFromPrimitive, + sqlx::Type, )] #[repr(i32)] #[strum(ascii_case_insensitive)] diff --git a/components/compression-coordinator/Cargo.toml b/components/compression-coordinator/Cargo.toml index 8712993d06..be6952f983 100644 --- a/components/compression-coordinator/Cargo.toml +++ b/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" } +const_format = "0.2.35" non-empty-string = "0.2.6" rmp-serde = "1.3.1" serde = { version = "1.0.228", features = ["derive"] } diff --git a/components/compression-coordinator/src/job_handle.rs b/components/compression-coordinator/src/job_handle.rs index 9e154d0708..676c8853ca 100644 --- a/components/compression-coordinator/src/job_handle.rs +++ b/components/compression-coordinator/src/job_handle.rs @@ -9,6 +9,7 @@ use clp_rust_utils::{ s3::{ObjectMetadata, S3ObjectMetadataId}, task_io::compression::{ClpSCompressionOption, S3InputSource}, }; +use const_format::formatcp; use non_empty_string::NonEmptyString; use spider_core::{ task::{ExecutionPolicy, TimeoutPolicy}, @@ -443,8 +444,8 @@ impl S3CompressionJobHandle S3CompressionJobHandle Result<(), Error> { let num_tasks = i32::try_from(num_tasks).map_err(|_| Error::TooManyCompressionTasks(num_tasks))?; - sqlx::query(&format!( + sqlx::query(formatcp!( "UPDATE `{COMPRESSION_JOB_TABLE_NAME}` SET `spider_id` = ?, `status` = ?, `num_tasks` \ - = ?, `start_time` = CURRENT_TIMESTAMP(3), `update_time` = CURRENT_TIMESTAMP() WHERE \ - `id` = ?" + = ?, `start_time` = CURRENT_TIMESTAMP(3) WHERE `id` = ?" )) .bind(spider_job_id.get()) - .bind(i32::from(CompressionJobStatus::Running)) + .bind(CompressionJobStatus::Running) .bind(num_tasks) .bind(self.compression_job_id) .execute(&self.db_pool) @@ -479,6 +479,7 @@ impl S3CompressionJobHandle Result<(), Error> { let outcome = self .job_submitter @@ -500,6 +501,12 @@ impl S3CompressionJobHandle Ok(()), CompressionJobOutcome::Failed { error_message } => { + if CompressionJobStatus::Succeeded == self.get_job_status().await? { + // NOTE: The commit task may successfully proceed but fail to report to Spider's + // control unit. In that case, the compression outcome has already committed to + // CLP DB, and thus the job should be considered `Succeeded`. + return Ok(()); + } self.update_job_status( CompressionJobStatus::Failed, Some(format!( @@ -509,6 +516,12 @@ impl S3CompressionJobHandle { + if CompressionJobStatus::Succeeded == self.get_job_status().await? { + // NOTE: The commit task may successfully proceed but fail to report to Spider's + // control unit. In that case, the compression outcome has already committed to + // CLP DB, and thus the job should be considered `Succeeded`. + return Ok(()); + } self.update_job_status( CompressionJobStatus::Killed, Some("The Spider compression job was cancelled.".to_owned()), @@ -518,24 +531,46 @@ impl S3CompressionJobHandle Result { + let job_status: CompressionJobStatus = sqlx::query_scalar(formatcp!( + "SELECT `status` FROM `{COMPRESSION_JOB_TABLE_NAME}` WHERE `id` = ?" + )) + .bind(self.compression_job_id) + .fetch_one(&self.db_pool) + .await?; + + Ok(job_status) + } + /// Updates the compression job status in the CLP database. /// /// # Errors /// /// Returns an error if: /// - /// * [`Error::Sqlx`] if the update fails. + /// * Forwards [`sqlx::query::Query::execute`]'s return values on failure. async fn update_job_status( &self, job_status: CompressionJobStatus, status_message: Option, ) -> Result<(), Error> { let status_message = status_message.as_ref().map_or("", String::as_str); - sqlx::query(&format!( - "UPDATE `{COMPRESSION_JOB_TABLE_NAME}` SET `status` = ?, `status_msg` = ?, \ - `update_time` = CURRENT_TIMESTAMP() WHERE `id` = ?" + sqlx::query(formatcp!( + "UPDATE `{COMPRESSION_JOB_TABLE_NAME}` SET `status` = ?, `status_msg` = ? WHERE `id` \ + = ?" )) - .bind(i32::from(job_status)) + .bind(job_status) .bind(status_message) .bind(self.compression_job_id) .execute(&self.db_pool) From 75f236c8f6bf61fbe667d466fd1cde832a873dbf Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Mon, 27 Jul 2026 15:45:39 -0400 Subject: [PATCH 06/10] Symbol reordering. --- components/compression-coordinator/src/job_handle.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/components/compression-coordinator/src/job_handle.rs b/components/compression-coordinator/src/job_handle.rs index 676c8853ca..373f6b7930 100644 --- a/components/compression-coordinator/src/job_handle.rs +++ b/components/compression-coordinator/src/job_handle.rs @@ -23,9 +23,6 @@ use crate::{ partition::CompressionInputBuilder, }; -const COMPRESSION_JOB_TABLE_NAME: &str = "compression_jobs"; -const INGESTED_S3_OBJECT_METADATA_TABLE_NAME: &str = "ingested_s3_object_metadata"; - /// Options for a compression job running in Spider. pub struct SpiderOption { pub compression_task_max_retry: u32, @@ -580,6 +577,9 @@ impl S3CompressionJobHandle Date: Tue, 28 Jul 2026 18:54:24 -0400 Subject: [PATCH 07/10] Fix logging. --- components/compression-coordinator/src/job_handle.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/compression-coordinator/src/job_handle.rs b/components/compression-coordinator/src/job_handle.rs index 373f6b7930..e3651eb44e 100644 --- a/components/compression-coordinator/src/job_handle.rs +++ b/components/compression-coordinator/src/job_handle.rs @@ -489,7 +489,7 @@ impl S3CompressionJobHandle Date: Tue, 28 Jul 2026 18:59:29 -0400 Subject: [PATCH 08/10] Remove redundant compression_jobs schema setup from initialize-orchestration-db.py; handled by #2421. --- .../initialize-orchestration-db.py | 27 +------------------ 1 file changed, 1 insertion(+), 26 deletions(-) diff --git a/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py b/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py index 1d768e0e5a..5af48908a6 100644 --- a/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py +++ b/components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py @@ -11,7 +11,7 @@ QueryJobStatus, QueryTaskStatus, ) -from mysql.connector.errorcode import ER_DUP_FIELDNAME, ER_DUP_KEYNAME +from mysql.connector.errorcode import ER_DUP_KEYNAME from pydantic import ValidationError from clp_py_utils.clp_config import ( @@ -80,7 +80,6 @@ def main(argv): `spider_id` BIGINT UNSIGNED NULL DEFAULT NULL, `dispatch_time` DATETIME 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, @@ -105,30 +104,6 @@ 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}` ( From b2cee3a45a9dfea250e8178857f20cbd55ee7992 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Tue, 28 Jul 2026 19:07:23 -0400 Subject: [PATCH 09/10] Revert database config. --- .../src/clp_config/package/config.rs | 51 ++++++++----------- 1 file changed, 22 insertions(+), 29 deletions(-) diff --git a/components/clp-rust-utils/src/clp_config/package/config.rs b/components/clp-rust-utils/src/clp_config/package/config.rs index 0891783eae..ac956bd3dd 100644 --- a/components/clp-rust-utils/src/clp_config/package/config.rs +++ b/components/clp-rust-utils/src/clp_config/package/config.rs @@ -143,33 +143,40 @@ pub struct Database { pub table_prefix: String, } +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(), + } + } +} + impl Database { /// # Returns /// - /// The archives table name `_archives`, where a `None` dataset resolves to - /// `default`. + /// The archives table name (`_archives`). #[must_use] pub fn archives_table_name(&self, dataset: Option<&str>) -> String { - self.archive_metadata_table_name("archives", dataset) + format!( + "{}{}_archives", + self.table_prefix, + resolve_dataset_name(dataset) + ) } /// # Returns /// - /// The column-metadata table name `_column_metadata`, where a `None` dataset - /// resolves to `default`. + /// The column-metadata table name (`_column_metadata`). #[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 - /// - /// `_`, where `dataset` defaults to the `CLP_S` default. - fn archive_metadata_table_name(&self, suffix: &str, dataset: Option<&str>) -> String { format!( - "{}{}_{suffix}", + "{}{}_column_metadata", self.table_prefix, resolve_dataset_name(dataset) ) @@ -184,20 +191,6 @@ impl Database { } } -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(), - } - } -} - #[derive(Clone, Debug, Deserialize, Eq, PartialEq)] #[serde(default)] pub struct ApiServer { From 8ca1e9ad69a2ebc366c06cfe944a9f6ecb234cc2 Mon Sep 17 00:00:00 2001 From: LinZhihao-723 Date: Tue, 28 Jul 2026 19:08:05 -0400 Subject: [PATCH 10/10] Remove empty line. --- components/clp-rust-utils/src/clp_config/package/config.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/components/clp-rust-utils/src/clp_config/package/config.rs b/components/clp-rust-utils/src/clp_config/package/config.rs index ac956bd3dd..9f2a481f56 100644 --- a/components/clp-rust-utils/src/clp_config/package/config.rs +++ b/components/clp-rust-utils/src/clp_config/package/config.rs @@ -147,7 +147,6 @@ 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,