refactor: use uuid::Uuid as JobId

src/mito2/src/compaction.rs

@@ -296,7 +296,7 @@ impl CompactionScheduler {
  let result = remote_job_scheduler
  .schedule(
  RemoteJob::CompactionJob(remote_compaction_job),
- Arc::new(DefaultNotifier {
+ Box::new(DefaultNotifier {
  request_sender: request_sender.clone(),
  waiters: std::mem::take(&mut waiters),
  }),
@@ -305,7 +305,7 @@ impl CompactionScheduler {
 
  if let Ok(job_id) = result {
  info!(
- "Scheduled remote compaction job {} for region {}",
+ "Scheduled remote compaction job {:?} for region {}",
  job_id, region_id
  );
 

src/mito2/src/error.rs

@@ -761,6 +761,14 @@ pub enum Error {
  reason: String,
  },
 
+ #[snafu(display("Failed to parse job id"))]
+ ParseJobId {
+ #[snafu(implicit)]
+ location: Location,
+ #[snafu(source)]
+ error: uuid::Error,
+ },
+
  #[snafu(display("Operation is not supported: {}", err_msg))]
  UnsupportedOperation {
  err_msg: String,
@@ -819,7 +827,8 @@ impl ErrorExt for Error {
  | InvalidMetadata { .. }
  | InvalidRegionOptions { .. }
  | InvalidWalReadRequest { .. }
- | PartitionOutOfRange { .. } => StatusCode::InvalidArguments,
+ | PartitionOutOfRange { .. }
+ | ParseJobId { .. } => StatusCode::InvalidArguments,
 
  InvalidRegionRequestSchemaVersion { .. } => StatusCode::RequestOutdated,
 

src/mito2/src/schedule/remote_job_scheduler.rs

@@ -16,12 +16,15 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use common_telemetry::error;
+use serde::{Deserialize, Serialize};
+use snafu::ResultExt;
 use store_api::storage::RegionId;
 use tokio::sync::mpsc::Sender;
+use uuid::Uuid;
 
 use crate::compaction::compactor::CompactionRegion;
 use crate::compaction::picker::PickerOutput;
-use crate::error::Result;
+use crate::error::{ParseJobIdSnafu, Result};
 use crate::manifest::action::RegionEdit;
 use crate::request::{
  BackgroundNotify, CompactionFailed, CompactionFinished, OutputTx, WorkerRequest,
@@ -50,7 +53,7 @@ pub type RemoteJobSchedulerRef = Arc<dyn RemoteJobScheduler>;
 #[async_trait::async_trait]
 pub trait RemoteJobScheduler: Send + Sync + 'static {
  /// Sends a job to the scheduler and returns a UUID for the job.
- async fn schedule(&self, job: RemoteJob, notifier: Arc<dyn Notifier>) -> Result<String>;
+ async fn schedule(&self, job: RemoteJob, notifier: Box<dyn Notifier>) -> Result<JobId>;
 }
 
 /// Notifier is used to notify the mito engine when a remote job is completed.
@@ -60,6 +63,22 @@ pub trait Notifier: Send + Sync + 'static {
  async fn notify(&mut self, result: RemoteJobResult);
 }
 
+/// Unique id for a remote job.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub struct JobId(Uuid);
+
+impl JobId {
+ /// Parses job id from string.
+ pub fn parse_str(input: &str) -> Result<JobId> {
+ Uuid::parse_str(input).map(JobId).context(ParseJobIdSnafu)
+ }
+
+ /// Covert job id to string.
+ pub fn to_string(&self) -> String {
+ self.0.to_string()
+ }
+}
+
 /// RemoteJob is a job that can be executed remotely. For example, a remote compaction job.
 #[derive(Clone)]
 #[allow(dead_code)]
@@ -85,7 +104,7 @@ pub enum RemoteJobResult {
 /// CompactionJobResult is the result of a compaction job.
 #[allow(dead_code)]
 pub struct CompactionJobResult {
- pub job_id: String,
+ pub job_id: JobId,
  pub region_id: RegionId,
  pub start_time: Instant,
  pub region_edit: Result<RegionEdit>,