feat: add vacuum impl

Cargo.toml

@@ -70,6 +70,7 @@ members = [
     "src/query/storages/parquet",
     "src/query/storages/result_cache",
     "src/query/users",
+    "src/query/ee-features/vacuum-handler",
     # databend-query
     "src/query/service",
     # enterprise

src/binaries/Cargo.toml

@@ -39,6 +39,7 @@ common-config = { path = "../query/config" }
 common-exception = { path = "../common/exception" }
 common-expression = { path = "../query/expression" }
 common-grpc = { path = "../common/grpc" }
+common-license = { path = "../common/license" }
 common-meta-api = { path = "../meta/api" }
 common-meta-app = { path = "../meta/app" }
 common-meta-client = { path = "../meta/client" }

src/binaries/query/oss_main.rs

@@ -22,6 +22,8 @@ use common_base::mem_allocator::GlobalAllocator;
 use common_base::runtime::Runtime;
 use common_config::InnerConfig;
 use common_exception::Result;
+use common_license::license_manager::LicenseManager;
+use common_license::license_manager::OssLicenseManager;
 
 use crate::entry::init_services;
 use crate::entry::start_services;
@@ -47,5 +49,7 @@ fn main() {
 async fn main_entrypoint() -> Result<()> {
     let conf: InnerConfig = InnerConfig::load()?;
     init_services(&conf).await?;
+    // init oss license manager
+    OssLicenseManager::init()?;
     start_services(&conf).await
 }

src/common/license/src/license_manager.rs

@@ -15,6 +15,7 @@
 use std::sync::Arc;
 
 use common_base::base::GlobalInstance;
+use common_exception::ErrorCode;
 use common_exception::Result;
 use jwt_simple::claims::JWTClaims;
 
@@ -61,6 +62,33 @@ pub struct LicenseManagerWrapper {
 unsafe impl Send for LicenseManagerWrapper {}
 unsafe impl Sync for LicenseManagerWrapper {}
 
+pub struct OssLicenseManager {}
+
+impl LicenseManager for OssLicenseManager {
+    fn init() -> Result<()> {
+        let rm = OssLicenseManager {};
+        let wrapper = LicenseManagerWrapper {
+            manager: Box::new(rm),
+        };
+        GlobalInstance::set(Arc::new(wrapper));
+        Ok(())
+    }
+
+    fn instance() -> Arc<Box<dyn LicenseManager>> {
+        GlobalInstance::get()
+    }
+
+    fn is_active(&self) -> bool {
+        false
+    }
+
+    fn make_license(_raw: &str) -> Result<JWTClaims<LicenseInfo>> {
+        Err(ErrorCode::LicenceDenied(
+            "Need Commercial License".to_string(),
+        ))
+    }
+}
+
 pub fn get_license_manager() -> Arc<LicenseManagerWrapper> {
     GlobalInstance::get()
 }

src/query/README.md

@@ -21,3 +21,4 @@ Databend Query is a Distributed Query Engine at scale.
 - [`streams`](./streams/) contains data sources and streams.
 - [`users`](./users/), role-based access and control.
 - [`ee`](ee/) contains enterprise functionalities.
+- [`ee-features`](ee-features/) contains enterprise features.

src/query/ee-features/README.md

@@ -0,0 +1 @@
+`ee-features` contains all the enterprise version feature traits.

src/query/ee-features/vacuum-handler/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "vacuum-handler"
+description = "vacuum handler"
+version = { workspace = true }
+authors = { workspace = true }
+license = { workspace = true }
+publish = { workspace = true }
+edition = { workspace = true }
+
+[lib]
+doctest = false
+test = false
+
+[dependencies]
+# Workspace dependencies
+common-base = { path = "../../../common/base" }
+common-catalog = { path = "../../catalog" }
+common-exception = { path = "../../../common/exception" }
+common-storages-fuse = { path = "../../storages/fuse" }
+
+async-backtrace = { workspace = true }
+async-trait = "0.1.57"
+chrono = { workspace = true }
+
+[build-dependencies]
+
+[package.metadata.cargo-machete]
+ignored = ["match-template"]

src/query/ee-features/vacuum-handler/src/lib.rs

@@ -0,0 +1,19 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod vacuum_handler;
+
+pub use vacuum_handler::get_vacuum_handler;
+pub use vacuum_handler::VacuumHandler;
+pub use vacuum_handler::VacuumHandlerWrapper;

src/query/ee-features/vacuum-handler/src/vacuum_handler.rs

@@ -0,0 +1,58 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use chrono::DateTime;
+use chrono::Utc;
+use common_base::base::GlobalInstance;
+use common_catalog::table_context::TableContext;
+use common_exception::Result;
+use common_storages_fuse::FuseTable;
+
+#[async_trait::async_trait]
+pub trait VacuumHandler: Sync + Send {
+    async fn do_vacuum(
+        &self,
+        fuse_table: &FuseTable,
+        ctx: Arc<dyn TableContext>,
+        retention_time: DateTime<Utc>,
+    ) -> Result<()>;
+}
+
+pub struct VacuumHandlerWrapper {
+    handler: Box<dyn VacuumHandler>,
+}
+
+impl VacuumHandlerWrapper {
+    pub fn new(handler: Box<dyn VacuumHandler>) -> Self {
+        Self { handler }
+    }
+
+    #[async_backtrace::framed]
+    pub async fn do_vacuum(
+        &self,
+        fuse_table: &FuseTable,
+        ctx: Arc<dyn TableContext>,
+        retention_time: DateTime<Utc>,
+    ) -> Result<()> {
+        self.handler
+            .do_vacuum(fuse_table, ctx, retention_time)
+            .await
+    }
+}
+
+pub fn get_vacuum_handler() -> Arc<VacuumHandlerWrapper> {
+    GlobalInstance::get()
+}

src/query/ee/Cargo.toml

@@ -15,10 +15,20 @@ test = false
 # Workspace dependencies
 async-backtrace = { workspace = true }
 common-base = { path = "../../common/base" }
+common-catalog = { path = "../catalog" }
 common-config = { path = "../config" }
 common-exception = { path = "../../common/exception" }
 common-license = { path = "../../common/license" }
+common-storages-fuse = { path = "../storages/fuse" }
+databend-query = { path = "../service" }
+storages-common-cache = { path = "../storages/common/cache" }
+storages-common-table-meta = { path = "../storages/common/table-meta" }
+vacuum-handler = { path = "../ee-features/vacuum-handler" }
+
+async-trait = "0.1.57"
+chrono = { workspace = true }
 jwt-simple = "0.11.0"
+tracing = "0.1.36"
 
 [build-dependencies]
 common-building = { path = "../../common/building" }

src/query/ee/src/enterprise_services.rs

@@ -17,12 +17,14 @@ use common_exception::Result;
 use common_license::license_manager::LicenseManager;
 
 use crate::license::license_mgr::RealLicenseManager;
+use crate::storages::fuse::operations::RealVacuumHandler;
 
 pub struct EnterpriseServices;
 impl EnterpriseServices {
     #[async_backtrace::framed]
     pub async fn init(_config: InnerConfig) -> Result<()> {
         RealLicenseManager::init()?;
+        RealVacuumHandler::init()?;
         Ok(())
     }
 }

src/query/ee/src/lib.rs

@@ -14,3 +14,4 @@
 
 pub mod enterprise_services;
 pub mod license;
+pub mod storages;

src/query/ee/src/license/mod.rs

@@ -13,3 +13,4 @@
 // limitations under the License.
 
 pub mod license_mgr;
+pub use license_mgr::RealLicenseManager;

src/query/ee/src/storages/fuse/io/mod.rs

@@ -0,0 +1,15 @@
+// Copyright 2023 Databend Cloud
+//
+// Licensed under the Elastic License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.elastic.co/licensing/elastic-license
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod snapshots;