feat(query-coordinator): Fail aggregation and write-to-file query jobs as unsupported.

components/query-coordinator/src/coordination.rs

@@ -167,58 +167,6 @@ impl Coordinator {
         Ok((coordinator, cancellation_token))
     }
 
-    /// Spawns a detached handle to drive each query job that a previous coordinator instance had
-    /// already submitted to Spider.
-    ///
-    /// A job whose handle cannot be constructed is logged and skipped.
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if:
-    ///
-    /// * Forwards [`Self::fetch_submitted_running_jobs`]'s return values on failure.
-    async fn recover_submitted_jobs(&self) -> Result<(), Error> {
-        // NOTE: The current implementation does not enforce concurrency limits for recovered jobs
-        // since they were already submitted to Spider. See #2472.
-        for SubmittedJob {
-            id: job_id,
-            spider_job_id,
-            search_job_config,
-            job_creation_timestamp_millisecs,
-        } in self.fetch_submitted_running_jobs().await?
-        {
-            tracing::info!(
-                job_id = % job_id,
-                spider_job_id = % spider_job_id,
-                "Recovering a previously submitted job."
-            );
-            let Ok(job_handle) = QueryJobHandle::new(
-                self.job_handle_context.clone(),
-                job_id,
-                self.spider_client.clone(),
-                self.resource_group_id,
-                search_job_config,
-                self.output_handle.clone(),
-                job_creation_timestamp_millisecs,
-            )
-            .inspect_err(|e| {
-                tracing::error!(
-                    error = % e,
-                    job_id = % job_id,
-                    "Failed to create the query job handle for recovery. Skipping."
-                );
-            }) else {
-                continue;
-            };
-            tokio::spawn(async move {
-                // `QueryJobHandle::recover` already logs and persists its own failures.
-                let _ = job_handle.recover(spider_job_id).await;
-            });
-        }
-
-        Ok(())
-    }
-
     /// Runs the coordinator's poll loop until cancelled.
     ///
     /// On each iteration, this method fetches the pending query jobs, spawns a detached handle to
@@ -267,6 +215,64 @@ impl Coordinator {
         Ok(())
     }
 
+    /// Spawns a detached handle to drive each query job that a previous coordinator instance had
+    /// already submitted to Spider.
+    ///
+    /// A job whose handle cannot be constructed is marked [`QueryJobStatus::Failed`] and skipped.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if:
+    ///
+    /// * Forwards [`Self::fetch_submitted_running_jobs`]'s return values on failure.
+    async fn recover_submitted_jobs(&self) -> Result<(), Error> {
+        // NOTE: The current implementation does not enforce concurrency limits for recovered jobs
+        // since they were already submitted to Spider. See #2472.
+        for SubmittedJob {
+            id: job_id,
+            spider_job_id,
+            search_job_config,
+            job_creation_timestamp_millisecs,
+        } in self.fetch_submitted_running_jobs().await?
+        {
+            tracing::info!(
+                job_id = % job_id,
+                spider_job_id = % spider_job_id,
+                "Recovering a previously submitted job."
+            );
+            let job_handle = match QueryJobHandle::new(
+                self.job_handle_context.clone(),
+                job_id,
+                self.spider_client.clone(),
+                self.resource_group_id,
+                search_job_config,
+                self.output_handle.clone(),
+                job_creation_timestamp_millisecs,
+            ) {
+                Ok(job_handle) => job_handle,
+                Err(e) => {
+                    tracing::error!(
+                        error = % e,
+                        job_id = % job_id,
+                        "Failed to create the query job handle for recovery. Skipping."
+                    );
+                    self.mark_job_failed(
+                        job_id,
+                        &format!("Failed to create the query job handle for recovery: {e}"),
+                    )
+                    .await;
+                    continue;
+                }
+            };
+            tokio::spawn(async move {
+                // `QueryJobHandle::recover` already logs and persists its own failures.
+                let _ = job_handle.recover(spider_job_id).await;
+            });
+        }
+
+        Ok(())
+    }
+
     /// Marks the query job identified by `job_id` as [`QueryJobStatus::Failed`].
     ///
     /// This is a best-effort update; if it fails, the error is logged and otherwise ignored.
@@ -295,8 +301,7 @@ impl Coordinator {
     /// the job-handler semaphore.
     ///
     /// A job whose config cannot be deserialized is marked [`QueryJobStatus::Failed`] and skipped;
-    /// a job whose handle cannot be constructed is marked failed and skipped as well. Aggregation
-    /// jobs are left undispatched for another scheduler.
+    /// a job whose handle cannot be constructed is marked failed and skipped as well.
     ///
     /// # Returns
     ///
@@ -338,14 +343,6 @@ impl Coordinator {
                         continue;
                     }
                 };
-            if search_job_config.aggregation_config.is_some() {
-                tracing::info!(
-                    job_id = % job_id,
-                    "Aggregation jobs are currently not supported by `query-coordinator`. \
-                     They will be handled by `search-scheduler` instead. Skipping."
-                );
-                continue;
-            }
             tracing::info!(job_id = % job_id, "Scheduling new job.");
             let job_handle = match QueryJobHandle::new(
                 self.job_handle_context.clone(),
@@ -531,14 +528,6 @@ impl Coordinator {
                     continue;
                 }
             };
-            if search_job_config.aggregation_config.is_some() {
-                tracing::error!(
-                    job_id = % row.id,
-                    "Aggregation jobs are not supposed to be handled by `query-coordinator` yet. \
-                     It should not have a Spider job; The database might be corrupted. Skipping."
-                );
-                continue;
-            }
             recovery_context.push(SubmittedJob {
                 id: row.id,
                 spider_job_id: row.spider_job_id,

components/query-coordinator/src/error.rs

@@ -48,4 +48,7 @@ pub enum Error {
 
     #[error("number of query tasks {0} exceeds `i32::MAX`")]
     TooManyQueryTasks(usize),
+
+    #[error("unsupported query job configuration: {0}")]
+    UnsupportedJobConfig(String),
 }

components/query-coordinator/src/job_handle.rs

@@ -80,6 +80,9 @@ impl<SubmitterType: QueryJobSubmitter> QueryJobHandle<SubmitterType> {
     ///
     /// Returns an error if:
     ///
+    /// * [`Error::UnsupportedJobConfig`] if:
+    ///   * The job carries an aggregation config.
+    ///   * The job writes its results to files.
     /// * [`Error::InvalidQueryJobConfig`] if:
     ///   * The query string is empty.
     ///   * The begin timestamp exceeds the end timestamp.
@@ -92,6 +95,17 @@ impl<SubmitterType: QueryJobSubmitter> QueryJobHandle<SubmitterType> {
         output_handle: OutputHandle,
         job_creation_timestamp_millisecs: i64,
     ) -> Result<Self, Error> {
+        if search_job_config.aggregation_config.is_some() {
+            return Err(Error::UnsupportedJobConfig(
+                "aggregation jobs are not supported".to_owned(),
+            ));
+        }
+        if search_job_config.write_to_file {
+            return Err(Error::UnsupportedJobConfig(
+                "writing query results to files is not supported".to_owned(),
+            ));
+        }
+
         let query_string = NonEmptyString::try_from(search_job_config.query_string.clone())
             .map_err(|_| {
                 Error::InvalidQueryJobConfig("query string must not be empty".to_owned())