feat: add compaction server supporting remote compaction service

Cargo.toml

@@ -24,7 +24,7 @@ license = "Apache-2.0"
 [workspace]
 resolver = "2"
 # In alphabetical order
-members = [
+members = [ 
     "horaectl",
     "integration_tests",
     "integration_tests/sdk/rust",
@@ -33,7 +33,8 @@ members = [
     "src/catalog",
     "src/catalog_impls",
     "src/cluster",
-    "src/common_types",
+    "src/common_types", 
+    "src/compaction_client",
     "src/components/alloc_tracker",
     "src/components/arena",
     "src/components/arrow_ext",
@@ -101,7 +102,8 @@ thiserror = "1"
 bytes_ext = { path = "src/components/bytes_ext" }
 catalog = { path = "src/catalog" }
 catalog_impls = { path = "src/catalog_impls" }
-horaedbproto = { git = "https://github.com/apache/incubator-horaedb-proto.git", rev = "19ece8f771fc0b3e8e734072cc3d8040de6c74cb" }
+# TODO(leslie): modify it when the related pr in incubator-horaedb-proto is merged.
+horaedbproto = { git = "https://github.com/LeslieKid/incubator-horaedb-proto.git", rev = "c8a073af23a7e5d29ec88011dd3c5abeffd37d23" }
 codec = { path = "src/components/codec" }
 chrono = "0.4"
 clap = { version = "4.5.1", features = ["derive"] }
@@ -110,6 +112,7 @@ cluster = { path = "src/cluster" }
 criterion = "0.5"
 horaedb-client = "1.0.2"
 common_types = { path = "src/common_types" }
+compaction_client = { path = "src/compaction_client" }
 datafusion = { git = "https://github.com/CeresDB/arrow-datafusion.git", rev = "e21b03154" }
 datafusion-proto = { git = "https://github.com/CeresDB/arrow-datafusion.git", rev = "e21b03154" }
 derive_builder = "0.12"

src/analytic_engine/src/compaction/mod.rs

@@ -20,15 +20,17 @@
 use std::{collections::HashMap, fmt, str::FromStr, sync::Arc};
 
 use common_types::COMPACTION_STRATEGY;
+use generic_error::{BoxError, GenericError};
+use macros::define_result;
 use serde::{Deserialize, Serialize};
 use size_ext::ReadableSize;
-use snafu::{ensure, Backtrace, GenerateBacktrace, ResultExt, Snafu};
+use snafu::{ensure, Backtrace, GenerateBacktrace, OptionExt, ResultExt, Snafu};
 use time_ext::TimeUnit;
 use tokio::sync::oneshot;
 
 use crate::{
     compaction::picker::{CommonCompactionPicker, CompactionPickerRef},
-    sst::file::{FileHandle, Level},
+    sst::file::{FileHandle, FileMeta, FilePurgeQueue, Level},
     table::data::TableDataRef,
 };
 
@@ -72,8 +74,22 @@ pub enum Error {
     },
     #[snafu(display("Invalid compaction option value, err: {}", error))]
     InvalidOption { error: String, backtrace: Backtrace },
+
+    #[snafu(display("Empty file meta.\nBacktrace:\n{}", backtrace))]
+    EmptyFileMeta { backtrace: Backtrace },
+
+    #[snafu(display("Failed to convert file meta, err:{}", source))]
+    ConvertFileMeta { source: GenericError },
+
+    #[snafu(display("Empty purge queue.\nBacktrace:\n{}", backtrace))]
+    EmptyPurgeQueue { backtrace: Backtrace },
+
+    #[snafu(display("Failed to convert level, err:{}", source))]
+    ConvertLevel { source: GenericError },
 }
 
+define_result!(Error);
+
 #[derive(Debug, Clone, Copy, Deserialize, Default, PartialEq, Serialize)]
 pub enum CompactionStrategy {
     #[default]
@@ -145,7 +161,7 @@ impl CompactionStrategy {
     pub(crate) fn parse_from(
         value: &str,
         options: &HashMap<String, String>,
-    ) -> Result<CompactionStrategy, Error> {
+    ) -> Result<CompactionStrategy> {
         match value.trim().to_lowercase().as_str() {
             DEFAULT_STRATEGY => Ok(CompactionStrategy::Default),
             STC_STRATEGY => Ok(CompactionStrategy::SizeTiered(
@@ -182,7 +198,7 @@ impl CompactionStrategy {
 }
 
 impl SizeTieredCompactionOptions {
-    pub(crate) fn validate(&self) -> Result<(), Error> {
+    pub(crate) fn validate(&self) -> Result<()> {
         ensure!(
             self.bucket_high > self.bucket_low,
             InvalidOption {
@@ -215,7 +231,7 @@ impl SizeTieredCompactionOptions {
 
     pub(crate) fn parse_from(
         options: &HashMap<String, String>,
-    ) -> Result<SizeTieredCompactionOptions, Error> {
+    ) -> Result<SizeTieredCompactionOptions> {
         let mut opts = SizeTieredCompactionOptions::default();
         if let Some(v) = options.get(BUCKET_LOW_KEY) {
             opts.bucket_low = v.parse().context(ParseFloat {
@@ -278,7 +294,7 @@ impl TimeWindowCompactionOptions {
         );
     }
 
-    pub(crate) fn validate(&self) -> Result<(), Error> {
+    pub(crate) fn validate(&self) -> Result<()> {
         if !Self::valid_timestamp_unit(self.timestamp_resolution) {
             return InvalidOption {
                 error: format!(
@@ -294,7 +310,7 @@ impl TimeWindowCompactionOptions {
 
     pub(crate) fn parse_from(
         options: &HashMap<String, String>,
-    ) -> Result<TimeWindowCompactionOptions, Error> {
+    ) -> Result<TimeWindowCompactionOptions> {
         let mut opts = TimeWindowCompactionOptions {
             size_tiered: SizeTieredCompactionOptions::parse_from(options)?,
             ..Default::default()
@@ -326,6 +342,67 @@ pub struct CompactionInputFiles {
     pub output_level: Level,
 }
 
+impl TryFrom<horaedbproto::compaction_service::CompactionInputFiles> for CompactionInputFiles {
+    type Error = Error;
+
+    fn try_from(value: horaedbproto::compaction_service::CompactionInputFiles) -> Result<Self> {
+        let level: Level = value.level.try_into().box_err().context(ConvertLevel)?;
+        let output_level: Level = value
+            .output_level
+            .try_into()
+            .box_err()
+            .context(ConvertLevel)?;
+
+        let mut files: Vec<FileHandle> = Vec::with_capacity(value.files.len());
+        for file in value.files {
+            let meta: FileMeta = file
+                .meta
+                .context(EmptyFileMeta)?
+                .try_into()
+                .box_err()
+                .context(ConvertFileMeta)?;
+
+            let purge_queue: FilePurgeQueue = file.purge_queue.context(EmptyPurgeQueue)?.into();
+
+            files.push({
+                let handle = FileHandle::new(meta, purge_queue);
+                handle.set_being_compacted(file.being_compacted);
+                handle
+            });
+        }
+
+        Ok(CompactionInputFiles {
+            level,
+            files,
+            output_level,
+        })
+    }
+}
+
+impl From<CompactionInputFiles> for horaedbproto::compaction_service::CompactionInputFiles {
+    fn from(value: CompactionInputFiles) -> Self {
+        let mut files = Vec::with_capacity(value.files.len());
+        for file in value.files {
+            let handle = horaedbproto::compaction_service::FileHandle {
+                meta: Some(file.meta().into()),
+                purge_queue: Some(horaedbproto::compaction_service::FilePurgeQueue {
+                    space_id: file.space_id(),
+                    table_id: file.table_id().into(),
+                }),
+                being_compacted: file.being_compacted(),
+                metrics: Some(horaedbproto::compaction_service::SstMetrics {}),
+            };
+            files.push(handle);
+        }
+
+        Self {
+            level: value.level.as_u32(),
+            files,
+            output_level: value.output_level.as_u32(),
+        }
+    }
+}
+
 #[derive(Debug, Default, Clone)]
 pub struct ExpiredFiles {
     /// Level of the expired files.

src/analytic_engine/src/compaction/runner/local_runner.rs

@@ -45,6 +45,7 @@ use crate::{
 const MAX_RECORD_BATCHES_IN_FLIGHT_WHEN_COMPACTION_READ: usize = 64;
 
 /// Executor carrying for actual compaction work
+#[derive(Clone)]
 pub struct LocalCompactionRunner {
     runtime: Arc<Runtime>,
     scan_options: ScanOptions,