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

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

@@ -10,6 +10,7 @@ use serde::Deserialize;
 use crate::clp_config::AwsAuthentication;
 use crate::clp_config::S3Config;
 use crate::dataset::resolve_dataset_name;
+use crate::types::non_empty_string::ExpectedNonEmpty;
 
 /// Mirror of `clp_py_utils.clp_config.ClpConfig`.
 ///
@@ -381,6 +382,29 @@ 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`.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the derived object key is empty. This should never happen since the key always
+    /// contains at least the `/` separator.
+    #[must_use]
+    pub fn dataset_archive_object_key(
+        &self,
+        dataset: Option<&str>,
+        archive_id: &str,
+    ) -> NonEmptyString {
+        NonEmptyString::from_string(format!(
+            "{}/{archive_id}",
+            self.dataset_archive_storage_directory(dataset)
+        ))
+    }
 }
 
 impl Default for ArchiveOutput {
@@ -699,6 +723,42 @@ 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")
+                .as_str(),
+            "LIB1/default/abc"
+        );
+        assert_eq!(
+            archive_output
+                .dataset_archive_object_key(Some("mydataset"), "abc")
+                .as_str(),
+            "LIB1/mydataset/abc"
+        );
+    }
+
     #[test]
     fn deserialize_database_ignores_provided_table_prefix() {
         let database_json = serde_json::json!({

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

@@ -8,6 +8,7 @@ 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,
@@ -28,4 +29,100 @@ pub struct ClpSQueryOption {
 
 /// The output handler that `clp-s` writes a query task's results to.
 #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
-pub enum OutputHandle {}
+#[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/task/compression/compress.rs

@@ -12,7 +12,6 @@ use std::process::Stdio;
 use anyhow::Context;
 use clp_rust_utils::aws::AWS_DEFAULT_REGION;
 use clp_rust_utils::clp_config::S3Config;
-use clp_rust_utils::clp_config::package::config::ArchiveOutput;
 use clp_rust_utils::clp_config::package::config::ArchiveOutputStorage;
 use clp_rust_utils::clp_config::package::config::Database;
 use clp_rust_utils::clp_config::package::config::SpiderTaskExecutorConfig;
@@ -129,7 +128,10 @@ pub(super) fn compress(
             ArchiveFinisher {
                 client: client.clone(),
                 bucket: bucket.clone(),
-                key: create_archive_s3_key(&config.archive_output, dataset.as_deref(), &archive.id),
+                key: config
+                    .archive_output
+                    .dataset_archive_object_key(dataset.as_deref(), &archive.id)
+                    .into_inner(),
                 indexer_bin: indexer_bin.clone(),
                 database: config.database.clone(),
                 dataset: dataset.clone(),
@@ -536,23 +538,6 @@ fn extract_s3_output_config(config: &SpiderTaskExecutorConfig) -> anyhow::Result
     }
 }
 
-/// Builds the S3 object key for an archive by appending `archive_id` to
-/// [`ArchiveOutput::dataset_archive_storage_directory`].
-///
-/// # Returns
-///
-/// The archive's S3 object key.
-fn create_archive_s3_key(
-    archive_output: &ArchiveOutput,
-    dataset: Option<&str>,
-    archive_id: &str,
-) -> String {
-    format!(
-        "{}/{archive_id}",
-        archive_output.dataset_archive_storage_directory(dataset)
-    )
-}
-
 /// Uploads a local file to S3 through `PutObject`.
 ///
 /// # Errors
@@ -827,9 +812,6 @@ mod tests {
     use std::path::PathBuf;
 
     use clp_rust_utils::clp_config::AwsAuthentication;
-    use clp_rust_utils::clp_config::S3Config;
-    use clp_rust_utils::clp_config::package::config::ArchiveOutput;
-    use clp_rust_utils::clp_config::package::config::ArchiveOutputStorage;
     use clp_rust_utils::clp_config::package::config::ClpDbNames;
     use clp_rust_utils::clp_config::package::config::Database;
     use clp_rust_utils::task_io::compression::ArchiveMetadata;
@@ -842,7 +824,6 @@ mod tests {
     use super::build_indexer_args;
     use super::build_log_converter_args;
     use super::build_s3_logs_list;
-    use super::create_archive_s3_key;
     use super::parse_archive_stats;
 
     #[test]
@@ -1003,30 +984,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn archive_s3_key_joins_prefix_dataset_and_id() {
-        let archive_output = ArchiveOutput {
-            storage: ArchiveOutputStorage::S3 {
-                staging_directory: "var/data/staged-archives".to_owned(),
-                s3_config: S3Config {
-                    bucket: NonEmptyString::try_from("bucket".to_string())
-                        .expect("bucket is non-empty"),
-                    region_code: None,
-                    key_prefix: NonEmptyString::try_from("LIB1/".to_string())
-                        .expect("key prefix is non-empty"),
-                    endpoint_url: None,
-                    aws_authentication: AwsAuthentication::Default,
-                },
-            },
-            ..ArchiveOutput::default()
-        };
-
-        assert_eq!(
-            create_archive_s3_key(&archive_output, None, "abc"),
-            "LIB1/default/abc"
-        );
-    }
-
     #[test]
     fn build_indexer_args_uses_mysql_and_expected_order() {
         let database = Database {

components/clp-tdl-package/src/task/query/mod.rs

@@ -1,21 +1,33 @@
-//! The query-task signatures registered with Spider.
+//! The query tasks: the `#[task]` wrappers Spider invokes and their implementations.
 
 use clp_rust_utils::job_config::ArchiveId;
 use clp_rust_utils::job_config::QueryJobId;
 use clp_rust_utils::task_io::query::ClpSQueryOption;
 use clp_rust_utils::task_io::query::OutputHandle;
 use non_empty_string::NonEmptyString;
 use spider_tdl::TaskContext;
+use spider_tdl::TdlError;
 use spider_tdl::task;
 
+mod search;
+
 #[task(name = "query::clp_s_search")]
 pub(crate) fn clp_s_search_task(
-    _ctx: TaskContext,
-    _query_job_id: QueryJobId,
-    _clp_s_query_option: ClpSQueryOption,
-    _dataset: Option<NonEmptyString>,
-    _archive_id: ArchiveId,
-    _output_handle: OutputHandle,
-) -> Result<(), spider_tdl::TdlError> {
-    todo!("clp-s search task is not implemented")
+    ctx: TaskContext,
+    query_job_id: QueryJobId,
+    clp_s_query_option: ClpSQueryOption,
+    dataset: Option<NonEmptyString>,
+    archive_id: ArchiveId,
+    output_handle: OutputHandle,
+) -> Result<(), TdlError> {
+    search::search(
+        &ctx,
+        crate::common::spider_task_executor_config(),
+        query_job_id,
+        &clp_s_query_option,
+        archive_id.into_inner(),
+        dataset.as_ref().map(NonEmptyString::as_str),
+        &output_handle,
+    )
+    .map_err(|e| TdlError::ExecutionError(format!("{e:#}")))
 }