feat(clp-tdl-package): Implement the query::clp_s_search task.

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

@@ -381,6 +381,20 @@ impl ArchiveOutput {
             .to_string_lossy()
             .into_owned()
     }
+
+    /// Derives the S3 object key of an archive in a dataset.
+    ///
+    /// # Returns
+    ///
+    /// The dataset's archive storage directory joined with `archive_id`, where a `None` dataset
+    /// resolves to `default`.
+    #[must_use]
+    pub fn dataset_archive_object_key(&self, dataset: Option<&str>, archive_id: &str) -> String {
+        format!(
+            "{}/{archive_id}",
+            self.dataset_archive_storage_directory(dataset)
+        )
+    }
 }
 
 impl Default for ArchiveOutput {
@@ -699,6 +713,38 @@ mod tests {
         );
     }
 
+    #[test]
+    fn dataset_archive_object_key_joins_prefix_dataset_and_id() {
+        use non_empty_string::NonEmptyString;
+
+        use crate::clp_config::AwsAuthentication;
+        use crate::clp_config::S3Config;
+        use crate::types::non_empty_string::ExpectedNonEmpty;
+
+        let archive_output = ArchiveOutput {
+            storage: ArchiveOutputStorage::S3 {
+                staging_directory: "var/data/staged-archives".to_owned(),
+                s3_config: S3Config {
+                    bucket: NonEmptyString::from_static_str("bucket"),
+                    region_code: None,
+                    key_prefix: NonEmptyString::from_static_str("LIB1/"),
+                    endpoint_url: None,
+                    aws_authentication: AwsAuthentication::Default,
+                },
+            },
+            ..ArchiveOutput::default()
+        };
+
+        assert_eq!(
+            archive_output.dataset_archive_object_key(None, "abc"),
+            "LIB1/default/abc"
+        );
+        assert_eq!(
+            archive_output.dataset_archive_object_key(Some("mydataset"), "abc"),
+            "LIB1/mydataset/abc"
+        );
+    }
+
     #[test]
     fn deserialize_database_ignores_provided_table_prefix() {
         let database_json = serde_json::json!({

components/clp-rust-utils/src/job_config/search.rs

@@ -5,6 +5,8 @@ use serde::Serialize;
 
 pub const QUERY_JOBS_TABLE_NAME: &str = "query_jobs";
 
+pub type QueryJobId = i32;
+
 /// Mirror of `job_orchestration.scheduler.job_config.AggregationConfig`. Must be kept in sync.
 #[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
 #[serde(default)]

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

@@ -1 +1,2 @@
 pub mod compression;
+pub mod query;

components/clp-rust-utils/src/task_io/query.rs

@@ -0,0 +1,128 @@
+//! Protocol types exchanged with the Spider (Huntsman) tasks that run CLP query jobs.
+
+use std::num::NonZeroU32;
+
+use non_empty_string::NonEmptyString;
+use serde::Deserialize;
+use serde::Serialize;
+
+/// `clp-s` options for a query job.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct ClpSQueryOption {
+    /// The query string passed positionally to `clp-s`.
+    pub query_string: NonEmptyString,
+
+    /// The per-archive result limit. When absent, the task omits `--max-num-results` and uses the
+    /// `clp-s` default.
+    pub max_num_results: Option<NonZeroU32>,
+
+    /// Inclusive `--tge` bound in Unix epoch milliseconds.
+    pub begin_timestamp_millisecs: Option<i64>,
+
+    /// Inclusive `--tle` bound in Unix epoch milliseconds.
+    pub end_timestamp_millisecs: Option<i64>,
+
+    /// Whether `clp-s` performs a case-insensitive search.
+    pub ignore_case: bool,
+}
+
+/// The output handler that `clp-s` writes a query task's results to.
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields, tag = "type")]
+pub enum OutputHandle {
+    /// The results cache, addressed by a MongoDB URI whose path names the database. The collection
+    /// is the query job's ID.
+    #[serde(rename = "results_cache")]
+    ResultsCache { uri: NonEmptyString },
+
+    /// A file per archive. Not yet supported by the Spider query flow.
+    #[serde(rename = "file")]
+    File,
+}
+
+#[cfg(test)]
+mod tests {
+    use std::num::NonZeroU32;
+
+    use non_empty_string::NonEmptyString;
+
+    use super::ClpSQueryOption;
+    use super::OutputHandle;
+    use crate::types::non_empty_string::ExpectedNonEmpty;
+
+    #[test]
+    fn clp_s_query_option_with_timestamp_bounds_round_trips_through_msgpack() {
+        let expected = ClpSQueryOption {
+            query_string: NonEmptyString::from_static_str("level:error"),
+            max_num_results: Some(NonZeroU32::new(1_000).expect("1,000 is nonzero")),
+            begin_timestamp_millisecs: Some(1_700_000_000_001),
+            end_timestamp_millisecs: Some(1_700_000_000_999),
+            ignore_case: true,
+        };
+
+        let serialized = rmp_serde::to_vec(&expected).expect("query options should serialize");
+        let actual: ClpSQueryOption =
+            rmp_serde::from_slice(&serialized).expect("query options should deserialize");
+
+        assert_eq!(expected, actual);
+    }
+
+    #[test]
+    fn clp_s_query_option_without_timestamp_bounds_round_trips_through_msgpack() {
+        let expected = ClpSQueryOption {
+            query_string: NonEmptyString::from_static_str("*"),
+            max_num_results: Some(NonZeroU32::new(1).expect("1 is nonzero")),
+            begin_timestamp_millisecs: None,
+            end_timestamp_millisecs: None,
+            ignore_case: false,
+        };
+
+        let serialized = rmp_serde::to_vec(&expected).expect("query options should serialize");
+        let actual: ClpSQueryOption =
+            rmp_serde::from_slice(&serialized).expect("query options should deserialize");
+
+        assert_eq!(expected, actual);
+    }
+
+    #[test]
+    fn clp_s_query_option_without_max_num_results_round_trips_through_msgpack() {
+        let expected = ClpSQueryOption {
+            query_string: NonEmptyString::from_static_str("*"),
+            max_num_results: None,
+            begin_timestamp_millisecs: None,
+            end_timestamp_millisecs: None,
+            ignore_case: false,
+        };
+
+        let serialized = rmp_serde::to_vec(&expected).expect("query options should serialize");
+        let actual: ClpSQueryOption =
+            rmp_serde::from_slice(&serialized).expect("query options should deserialize");
+
+        assert_eq!(expected, actual);
+    }
+
+    #[test]
+    fn output_handle_results_cache_round_trips_through_msgpack() {
+        let expected = OutputHandle::ResultsCache {
+            uri: NonEmptyString::from_static_str("mongodb://results-cache:27017/clp-query-results"),
+        };
+
+        let serialized = rmp_serde::to_vec(&expected).expect("output handle should serialize");
+        let actual: OutputHandle =
+            rmp_serde::from_slice(&serialized).expect("output handle should deserialize");
+
+        assert_eq!(expected, actual);
+    }
+
+    #[test]
+    fn output_handle_file_round_trips_through_msgpack() {
+        let expected = OutputHandle::File;
+
+        let serialized = rmp_serde::to_vec(&expected).expect("output handle should serialize");
+        let actual: OutputHandle =
+            rmp_serde::from_slice(&serialized).expect("output handle should deserialize");
+
+        assert_eq!(expected, actual);
+    }
+}

components/clp-tdl-package/README.md

@@ -12,3 +12,7 @@ documented below.
 
 * `compression::clp_s_s3_compress`: Compress inputs from S3 using `clp-s`.
 * `compression::commit`: Commit compression task outcomes to the CLP metadata database.
+
+### Query
+
+* `query::clp_s_search`: Search a single archive using the `clp-s` engine.

components/clp-tdl-package/src/lib.rs

@@ -1,4 +1,4 @@
-//! Spider TDL task package `clp`: the CLP compression tasks the Spider task executor loads.
+//! Spider TDL package `clp`, providing CLP compression and query tasks for Spider task executors.
 
 pub mod common;
 mod task;
@@ -28,5 +28,9 @@ fn package_init() -> Result<(), TdlError> {
 spider_tdl::register_tdl_package! {
     package_name: "clp",
     init: package_init,
-    tasks: [task::compression::s3_compress_task, task::compression::commit_task],
+    tasks: [
+        task::compression::s3_compress_task,
+        task::compression::commit_task,
+        task::query::clp_s_search_task,
+    ],
 }