Merge branch 'main' of https://github.com/datafuselabs/databend into add_runtime_bloom_filter_for_merge_into

Author: JackTan25

Cargo.lock

@@ -120,7 +120,8 @@ opendal = { version = "0.45.0", features = [
 sled = { git = "https://github.com/datafuse-extras/sled", tag = "v0.34.7-datafuse.1", default-features = false }
 
 # openraft for debugging
-openraft = { git = "https://github.com/datafuselabs/openraft", tag = "v0.9.0", features = [
+# openraft = { git = "https://github.com/datafuselabs/openraft", tag = "v0.9.0", features = [
+openraft = { version = "0.9.1", features = [
     "serde",
     "tracing-log",
     "generic-snapshot-data",

Cargo.toml

@@ -34,7 +34,7 @@
 
 ## 🐋 Introduction
 
-**Databend** is an open-source, elastic, and workload-aware cloud data warehouse built in Rust, offering a cost-effective [alternative to Snowflake](https://github.com/datafuselabs/databend/issues/13059). It's designed for complex analysis of the world's largest datasets.
+**Databend**, built in Rust, is an open-source cloud data warehouse that serves as a cost-effective [alternative to Snowflake](https://github.com/datafuselabs/databend/issues/13059). With its focus on fast query execution and data ingestion, it's designed for complex analysis of the world's largest datasets.
 
 ## ⚡ Performance
 
@@ -46,6 +46,14 @@
 
 ![Databend vs. Snowflake](https://github.com/datafuselabs/wizard/assets/172204/d796acf0-0a66-4b1d-8754-cd2cd1de04c7)
 
+<div align="center">
+  
+[Data Ingestion Benchmark: Databend Cloud vs. Snowflake](https://docs.databend.com/guides/benchmark/data-ingest)
+
+</div>
+
+![Databend vs. Snowflake](https://github.com/datafuselabs/databend/assets/172204/c61d7a40-f6fe-4fb9-83e8-06ea9599aeb4)
+
 
 ## 🚀 Why Databend
 
@@ -213,6 +221,15 @@ docker run --net=host  datafuselabs/databend
 
 </details>
 
+<details>
+<summary>Accessing Data Lake</summary>
+
+- [Apache Hive](https://docs.databend.com/guides/access-data-lake/hive)
+- [Apache Iceberg](https://docs.databend.com/guides/access-data-lake/iceberg/iceberg-engine)
+- [Delta Lake](https://docs.databend.com/guides/access-data-lake/delta)
+
+</details>
+
 <details>
 <summary>Security</summary>
 
@@ -227,7 +244,8 @@ docker run --net=host  datafuselabs/databend
 <summary>Performance</summary>
 
 - [Review Clickbench](https://databend.com/blog/clickbench-databend-top)
-- [How to Benchmark Databend using TPC-H](https://databend.com/blog/2022/08/08/benchmark-tpc-h)
+- [TPC-H Benchmark: Databend Cloud vs. Snowflake](https://docs.databend.com/guides/benchmark/tpch)
+- [Databend vs. Snowflake: Data Ingestion Benchmark](https://docs.databend.com/guides/benchmark/data-ingest)
 
 </details>
 

README.md

@@ -189,7 +189,7 @@ async fn benchmark_table(client: &Arc<ClientHandle>, prefix: u64, client_num: u6
 
     let res = client
         .create_database(CreateDatabaseReq {
-            create_option: CreateOption::None,
+            create_option: CreateOption::Create,
             name_ident: DatabaseNameIdent {
                 tenant: tenant(),
                 db_name: db_name(),

src/binaries/metabench/main.rs

@@ -74,6 +74,19 @@ pub fn metrics_inc_aggregate_partial_hashtable_allocated_bytes(c: u64) {
     AGGREGATE_PARTIAL_HASHTABLE_ALLOCATED_BYTES.inc_by(c);
 }
 
+pub fn metrics_inc_group_by_partial_spill_count() {
+    let labels = &vec![("spill", "group_by_partial_spill".to_string())];
+    SPILL_COUNT.get_or_create(labels).inc();
+}
+
+pub fn metrics_inc_group_by_partial_spill_cell_count(c: u64) {
+    AGGREGATE_PARTIAL_SPILL_CELL_COUNT.inc_by(c);
+}
+
+pub fn metrics_inc_group_by_partial_hashtable_allocated_bytes(c: u64) {
+    AGGREGATE_PARTIAL_HASHTABLE_ALLOCATED_BYTES.inc_by(c);
+}
+
 pub fn metrics_inc_group_by_spill_write_count() {
     let labels = &vec![("spill", "group_by_spill".to_string())];
     SPILL_WRITE_COUNT.get_or_create(labels).inc();

src/common/metrics/src/metrics/transform.rs

@@ -22,6 +22,7 @@ databend-common-meta-types = { path = "../types" }
 databend-common-proto-conv = { path = "../proto-conv" }
 
 anyhow = { workspace = true }
+async-backtrace = { workspace = true }
 async-trait = { workspace = true }
 chrono = { workspace = true }
 enumflags2 = { workspace = true }

src/meta/api/Cargo.toml

@@ -0,0 +1,81 @@
+// 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::error::Error;
+
+use databend_common_exception::ErrorCode;
+use databend_common_meta_app::tenant_key::TenantResource;
+use databend_common_meta_app::tenant_key_errors::ExistError;
+use databend_common_meta_app::tenant_key_errors::UnknownError;
+use databend_common_meta_types::MetaError;
+
+/// CRUD Error that can be an API level error or a business error.
+#[derive(Clone, Debug, thiserror::Error)]
+pub enum CrudError<E>
+where E: Error + 'static
+{
+    #[error(transparent)]
+    ApiError(#[from] MetaError),
+
+    #[error(transparent)]
+    Business(E),
+}
+
+impl<E> CrudError<E>
+where E: Error + 'static
+{
+    pub fn is_business_error(&self) -> bool {
+        match self {
+            CrudError::ApiError(_e) => false,
+            CrudError::Business(_e) => true,
+        }
+    }
+
+    /// Convert the error into a layered result.
+    pub fn into_result(self) -> Result<Result<(), E>, MetaError> {
+        match self {
+            CrudError::ApiError(meta_err) => Err(meta_err),
+            CrudError::Business(e) => Ok(Err(e)),
+        }
+    }
+}
+
+impl<R> From<ExistError<R>> for CrudError<ExistError<R>>
+where R: TenantResource + 'static
+{
+    fn from(e: ExistError<R>) -> Self {
+        CrudError::Business(e)
+    }
+}
+
+impl<R> From<UnknownError<R>> for CrudError<UnknownError<R>>
+where R: TenantResource + 'static
+{
+    fn from(e: UnknownError<R>) -> Self {
+        CrudError::Business(e)
+    }
+}
+
+impl<E> From<CrudError<E>> for ErrorCode
+where
+    E: Into<ErrorCode>,
+    E: Error,
+{
+    fn from(value: CrudError<E>) -> Self {
+        match value {
+            CrudError::ApiError(meta_err) => meta_err.into(),
+            CrudError::Business(e) => e.into(),
+        }
+    }
+}

src/meta/api/src/crud/errors.rs

@@ -0,0 +1,179 @@
+// 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.
+
+//! A generic CRUD interface for meta data operations.
+
+mod errors;
+use std::marker::PhantomData;
+use std::sync::Arc;
+
+use databend_common_meta_app::schema::CreateOption;
+use databend_common_meta_app::tenant::Tenant;
+use databend_common_meta_app::tenant_key::TIdent;
+use databend_common_meta_app::tenant_key::TenantResource;
+use databend_common_meta_app::tenant_key_errors::ExistError;
+use databend_common_meta_app::tenant_key_errors::UnknownError;
+use databend_common_meta_kvapi::kvapi;
+use databend_common_meta_kvapi::kvapi::DirName;
+use databend_common_meta_kvapi::kvapi::ValueWithName;
+use databend_common_meta_types::MatchSeq;
+use databend_common_meta_types::MatchSeqExt;
+use databend_common_meta_types::MetaError;
+use databend_common_meta_types::NonEmptyString;
+use databend_common_meta_types::SeqV;
+use databend_common_meta_types::SeqValue;
+use databend_common_meta_types::With;
+use databend_common_proto_conv::FromToProto;
+pub use errors::CrudError;
+use futures::TryStreamExt;
+
+use crate::kv_pb_api::KVPbApi;
+use crate::kv_pb_api::UpsertPB;
+
+/// A generic CRUD interface for meta data operations.
+///
+/// - It provide `add`, `update`, `get`, `remove` and `list` operations.
+/// - The key space it operates on is defined by the type [`TIdent<R>`],
+///   which contains a `Tenant` and a `Name`.
+///
+/// One `CrudMgr` instance can only access keys of exactly one [`Tenant`].
+///
+/// [`TIdent<R>`]: TIdent
+pub struct CrudMgr<R> {
+    kv_api: Arc<dyn kvapi::KVApi<Error = MetaError>>,
+    tenant: Tenant,
+    _p: PhantomData<R>,
+}
+
+impl<R> CrudMgr<R> {
+    /// Create a new `CrudMgr` instance providing CRUD access for a key space defined by `R`: [`TenantResource`].
+    pub fn create(
+        kv_api: Arc<dyn kvapi::KVApi<Error = MetaError>>,
+        tenant: &NonEmptyString,
+    ) -> Self {
+        CrudMgr {
+            kv_api,
+            tenant: Tenant::new_nonempty(tenant.clone()),
+            _p: Default::default(),
+        }
+    }
+
+    /// Create a structured key for the given name.
+    fn ident(&self, name: &str) -> TIdent<R> {
+        TIdent::new(self.tenant.clone(), name)
+    }
+}
+
+/// A shortcut
+type ValueOf<R> = <TIdent<R> as kvapi::Key>::ValueType;
+
+impl<R> CrudMgr<R>
+where
+    R: TenantResource + Send + 'static,
+    // As a kvapi::Key, the corresponding value contains a name.
+    ValueOf<R>: ValueWithName + FromToProto + Clone,
+{
+    #[async_backtrace::framed]
+    #[minitrace::trace]
+    pub async fn add(
+        &self,
+        value: ValueOf<R>,
+        create_option: &CreateOption,
+    ) -> Result<(), CrudError<ExistError<R>>> {
+        let ident = self.ident(value.name());
+
+        let seq = MatchSeq::from(*create_option);
+        let upsert = UpsertPB::insert(ident, value.clone()).with(seq);
+
+        let res = self.kv_api.upsert_pb(&upsert).await?;
+
+        if let CreateOption::Create = create_option {
+            if res.prev.is_some() {
+                return Err(ExistError::new(value.name(), "Exist when add").into());
+            }
+        }
+
+        Ok(())
+    }
+
+    #[async_backtrace::framed]
+    #[minitrace::trace]
+    pub async fn update(
+        &self,
+        value: ValueOf<R>,
+        match_seq: MatchSeq,
+    ) -> Result<u64, CrudError<UnknownError<R>>> {
+        let ident = self.ident(value.name());
+        let upsert = UpsertPB::update(ident, value.clone()).with(match_seq);
+
+        let res = self.kv_api.upsert_pb(&upsert).await?;
+
+        match res.result {
+            Some(SeqV { seq, .. }) => Ok(seq),
+            None => Err(UnknownError::new(value.name(), "NotFound when update").into()),
+        }
+    }
+
+    #[async_backtrace::framed]
+    #[minitrace::trace]
+    pub async fn remove(
+        &self,
+        name: &str,
+        seq: MatchSeq,
+    ) -> Result<(), CrudError<UnknownError<R>>> {
+        let ident = self.ident(name);
+
+        let upsert = UpsertPB::delete(ident).with(seq);
+
+        let res = self.kv_api.upsert_pb(&upsert).await?;
+        res.removed_or_else(|e| {
+            UnknownError::new(
+                name,
+                format_args!("NotFound when remove, seq of existing record: {}", e.seq()),
+            )
+        })?;
+
+        Ok(())
+    }
+
+    #[async_backtrace::framed]
+    #[minitrace::trace]
+    pub async fn get(
+        &self,
+        name: &str,
+        seq: MatchSeq,
+    ) -> Result<SeqV<ValueOf<R>>, CrudError<UnknownError<R>>> {
+        let ident = self.ident(name);
+
+        let res = self.kv_api.get_pb(&ident).await?;
+
+        let seq_value = res.ok_or_else(|| UnknownError::new(name, "NotFound when get"))?;
+
+        match seq.match_seq(&seq_value) {
+            Ok(_) => Ok(seq_value),
+            Err(e) => Err(UnknownError::new(name, format_args!("NotFound when get: {}", e)).into()),
+        }
+    }
+
+    #[async_backtrace::framed]
+    #[minitrace::trace]
+    pub async fn list(&self) -> Result<Vec<ValueOf<R>>, MetaError> {
+        let dir_name = DirName::new(self.ident("dummy"));
+
+        let values = self.kv_api.list_pb_values(&dir_name).await?;
+        let values = values.try_collect().await?;
+
+        Ok(values)
+    }
+}

src/meta/api/src/crud/mod.rs

@@ -79,7 +79,7 @@ impl<KV: kvapi::KVApi<Error = MetaError>> DatamaskApi for KV {
 
             if seq > 0 {
                 match req.create_option {
-                    CreateOption::None => {
+                    CreateOption::Create => {
                         return Err(KVAppError::AppError(AppError::DatamaskAlreadyExists(
                             DatamaskAlreadyExists::new(
                                 &name_key.name,

src/meta/api/src/data_mask_api_impl.rs

@@ -41,6 +41,8 @@ pub(crate) mod testing;
 pub mod txn_backoff;
 pub(crate) mod util;
 
+pub mod crud;
+
 pub use background_api::BackgroundApi;
 pub use background_api_test_suite::BackgroundApiTestSuite;
 pub use data_mask_api::DatamaskApi;