Merge branch 'main' into fix/cluster_hang_3

Author: zhang2014

.github/actions/setup_bendsql/action.yml

@@ -4,7 +4,7 @@ inputs:
   version:
     description: "bendsql version"
     required: false
-    default: v0.3.8
+    default: v0.3.9
 runs:
   using: "composite"
   steps:

Cargo.lock

@@ -4,11 +4,10 @@ title: 'Conditional Functions'
 
 | Function                                                   | Description                                                                                                                                              | Example                                                                          | Result   |
 |------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------------|----------|
-| **IF(expr1, expr2, expr3)**                                | Return expr2 if expr1 is TRUE. Otherwise return expr3. expr2 and expr3 must have the lowest common type.                                                 | **IF(1 > 2, 3, 4)**                                                              | 4        |
+| **IF(cond1, expr1, [cond2, expr2, ...], expr_else)**       | If cond1 is TRUE, it returns expr1. Otherwise if cond2 is TRUE, it returns expr2, and so on.                                                             | **IF(1 > 2, 3, 4 < 5, 6, 7)**                                                    | 6        |
 | **IFNULL(expr1, expr2)**                                   | Return expr1 if it is not NULL. Otherwise return expr2. They must have the same data type.                                                               | **IFNULL(0, NULL)**                                                              | 0        |
 | **value [ NOT ] IN (value1, value2, ...)**                 | Check whether value is (or is not) one of the members of an explicit list.                                                                               | **1 not in (2, 3)**                                                              | 1(TRUE)  |
 | **expr1 IS [ NOT ] DISTINCT FROM expr2**                   | Compares whether two expressions are equal (or not equal) with awareness of nullability, meaning it treats NULLs as known values for comparing equality. | **NULL is distinct from NULL**                                                   | 0(FALSE) |
 | **IS_NOT_NULL(expr)**                                      | Check whether the value is not NULL.                                                                                                                     | **IS_NOT_NULL(1)**                                                               | 1(TRUE)  |
 | **IS_NULL(expr)**                                          | Check whether the value is NULL.                                                                                                                         | **IS_NULL(1)**                                                                   | 0(FALSE) |
-| **MULTI_IF(cond1, expr1, [cond2, expr2, ...], expr_else)** | If cond1 is TRUE, it returns expr1. Otherwise if cond2 is TRUE, it returns expr2, and so on.                                                             | **MULTI_IF(1 > 2, 3, 4 < 5, 6, 7)**                                              | 6        |
 | **NULLIF(expr1, expr2)**                                   | Return NULL if two expressions are equal. Otherwise return expr1. They must have the same data type.                                                     | **NULLIF(0, NULL)**                                                              | 0        |

benchmark/clickbench/tpch/queries.sql

@@ -7,7 +7,7 @@ select version();
 SQL
 
 for t in customer lineitem nation orders partsupp part region supplier; do
-  echo "DROP TABLE IF EXISTS $t;" | bendsql query
+	echo "DROP TABLE IF EXISTS $t;" | bendsql query
 done
 
 cat <<SQL | bendsql query
@@ -113,8 +113,8 @@ cat <<SQL | bendsql query
 SQL
 
 for t in customer lineitem nation orders partsupp part region supplier; do
-  echo "loading into $t ..."
-  cat <<SQL | bendsql query
+	echo "loading into $t ..."
+	cat <<SQL | bendsql query
 COPY INTO $t FROM 's3://repo.databend.rs/datasets/tpch10/${t}/'
 credentials=(aws_key_id='$REPO_ACCESS_KEY_ID' aws_secret_key='$REPO_SECRET_ACCESS_KEY') pattern ='${t}.*'
 file_format=(type='CSV' field_delimiter='|' record_delimiter='\\n' skip_header=0);

docs/doc/15-sql-functions/11-conditional-functions/index.md

@@ -21,6 +21,7 @@ anyhow = { workspace = true }
 async-trait = "0.1"
 bytes = "1"
 chrono = { workspace = true }
+flagset = "0.4"
 futures = "0.3"
 opendal = { workspace = true }
 regex = "1.6.0"

scripts/benchmark/query/load/tpch10.sh

@@ -55,7 +55,6 @@ pub use parquet::read_parquet_schema_async;
 
 mod stage;
 pub use stage::init_stage_operator;
-pub use stage::FileWithMeta;
 pub use stage::StageFileInfo;
 pub use stage::StageFileStatus;
 pub use stage::StageFilesInfo;

src/common/storage/Cargo.toml

@@ -33,20 +33,6 @@ use regex::Regex;
 use crate::init_operator;
 use crate::DataOperator;
 
-pub struct FileWithMeta {
-    pub path: String,
-    pub metadata: Metadata,
-}
-
-impl FileWithMeta {
-    fn new(path: &str, meta: Metadata) -> Self {
-        Self {
-            path: path.to_string(),
-            metadata: meta,
-        }
-    }
-}
-
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
 pub enum StageFileStatus {
     NeedCopy,
@@ -79,11 +65,10 @@ impl StageFileInfo {
             creator: None,
         }
     }
-}
 
-impl From<FileWithMeta> for StageFileInfo {
-    fn from(value: FileWithMeta) -> Self {
-        StageFileInfo::new(value.path, &value.metadata)
+    /// NOTE: update this query when add new meta
+    pub fn meta_query() -> flagset::FlagSet<Metakey> {
+        Metakey::ContentLength | Metakey::ContentMd5 | Metakey::LastModified | Metakey::Etag
     }
 }
 
@@ -122,7 +107,7 @@ impl StageFilesInfo {
         }
     }
 
-    pub async fn list(&self, operator: &Operator, first_only: bool) -> Result<Vec<FileWithMeta>> {
+    pub async fn list(&self, operator: &Operator, first_only: bool) -> Result<Vec<StageFileInfo>> {
         if let Some(files) = &self.files {
             let mut res = Vec::new();
             for file in files {
@@ -132,7 +117,7 @@ impl StageFilesInfo {
                     .to_string();
                 let meta = operator.stat(&full_path).await?;
                 if meta.mode().is_file() {
-                    res.push(FileWithMeta::new(&full_path, meta))
+                    res.push(StageFileInfo::new(full_path, &meta))
                 } else {
                     return Err(ErrorCode::BadArguments(format!(
                         "{full_path} is not a file"
@@ -149,15 +134,15 @@ impl StageFilesInfo {
         }
     }
 
-    pub async fn first_file(&self, operator: &Operator) -> Result<FileWithMeta> {
+    pub async fn first_file(&self, operator: &Operator) -> Result<StageFileInfo> {
         let mut files = self.list(operator, true).await?;
         match files.pop() {
             None => Err(ErrorCode::BadArguments("no file found")),
             Some(f) => Ok(f),
         }
     }
 
-    pub fn blocking_first_file(&self, operator: &Operator) -> Result<FileWithMeta> {
+    pub fn blocking_first_file(&self, operator: &Operator) -> Result<StageFileInfo> {
         let mut files = self.blocking_list(operator, true)?;
         match files.pop() {
             None => Err(ErrorCode::BadArguments("no file found")),
@@ -169,7 +154,7 @@ impl StageFilesInfo {
         &self,
         operator: &Operator,
         first_only: bool,
-    ) -> Result<Vec<FileWithMeta>> {
+    ) -> Result<Vec<StageFileInfo>> {
         if let Some(files) = &self.files {
             let mut res = Vec::new();
             for file in files {
@@ -179,7 +164,7 @@ impl StageFilesInfo {
                     .to_string();
                 let meta = operator.blocking().stat(&full_path)?;
                 if meta.mode().is_file() {
-                    res.push(FileWithMeta::new(&full_path, meta))
+                    res.push(StageFileInfo::new(full_path, &meta))
                 } else {
                     return Err(ErrorCode::BadArguments(format!(
                         "{full_path} is not a file"
@@ -201,11 +186,11 @@ impl StageFilesInfo {
         path: &str,
         pattern: Option<Regex>,
         first_only: bool,
-    ) -> Result<Vec<FileWithMeta>> {
+    ) -> Result<Vec<StageFileInfo>> {
         let root_meta = operator.stat(path).await;
         match root_meta {
             Ok(meta) => match meta.mode() {
-                EntryMode::FILE => return Ok(vec![FileWithMeta::new(path, meta)]),
+                EntryMode::FILE => return Ok(vec![StageFileInfo::new(path.to_string(), &meta)]),
                 EntryMode::DIR => {}
                 EntryMode::Unknown => {
                     return Err(ErrorCode::BadArguments("object mode is unknown"));
@@ -224,10 +209,9 @@ impl StageFilesInfo {
         let mut files = Vec::new();
         let mut list = operator.scan(path).await?;
         while let Some(obj) = list.try_next().await? {
-            // todo(youngsofun): not always need Metakey::Complete
-            let meta = operator.metadata(&obj, Metakey::Complete).await?;
+            let meta = operator.metadata(&obj, StageFileInfo::meta_query()).await?;
             if check_file(obj.path(), meta.mode(), &pattern) {
-                files.push(FileWithMeta::new(obj.path(), meta));
+                files.push(StageFileInfo::new(obj.path().to_string(), &meta));
                 if first_only {
                     return Ok(files);
                 }
@@ -253,13 +237,13 @@ fn blocking_list_files_with_pattern(
     path: &str,
     pattern: Option<Regex>,
     first_only: bool,
-) -> Result<Vec<FileWithMeta>> {
+) -> Result<Vec<StageFileInfo>> {
     let operator = operator.blocking();
 
     let root_meta = operator.stat(path);
     match root_meta {
         Ok(meta) => match meta.mode() {
-            EntryMode::FILE => return Ok(vec![FileWithMeta::new(path, meta)]),
+            EntryMode::FILE => return Ok(vec![StageFileInfo::new(path.to_string(), &meta)]),
             EntryMode::DIR => {}
             EntryMode::Unknown => return Err(ErrorCode::BadArguments("object mode is unknown")),
         },
@@ -277,9 +261,9 @@ fn blocking_list_files_with_pattern(
     let list = operator.list(path)?;
     for obj in list {
         let obj = obj?;
-        let meta = operator.metadata(&obj, Metakey::Complete)?;
+        let meta = operator.metadata(&obj, StageFileInfo::meta_query())?;
         if check_file(obj.path(), meta.mode(), &pattern) {
-            files.push(FileWithMeta::new(obj.path(), meta));
+            files.push(StageFileInfo::new(obj.path().to_string(), &meta));
             if first_only {
                 return Ok(files);
             }

src/common/storage/src/lib.rs

@@ -90,13 +90,21 @@ impl GrantEntry {
         &self.privileges
     }
 
-    pub fn verify_privilege(&self, object: &GrantObject, privilege: UserPrivilegeType) -> bool {
+    pub fn verify_privilege(
+        &self,
+        object: &GrantObject,
+        privileges: Vec<UserPrivilegeType>,
+    ) -> bool {
         // the verified object should be smaller than the object inside my grant entry.
         if !self.object.contains(object) {
             return false;
         }
 
-        self.privileges.contains(privilege)
+        let mut priv_set = UserPrivilegeSet::empty();
+        for privilege in privileges {
+            priv_set.set_privilege(privilege)
+        }
+        self.privileges.contains(BitFlags::from(priv_set))
     }
 
     pub fn matches_entry(&self, object: &GrantObject) -> bool {
@@ -156,10 +164,14 @@ impl UserGrantSet {
         self.roles.remove(role);
     }
 
-    pub fn verify_privilege(&self, object: &GrantObject, privilege: UserPrivilegeType) -> bool {
+    pub fn verify_privilege(
+        &self,
+        object: &GrantObject,
+        privilege: Vec<UserPrivilegeType>,
+    ) -> bool {
         self.entries
             .iter()
-            .any(|e| e.verify_privilege(object, privilege))
+            .any(|e| e.verify_privilege(object, privilege.clone()))
     }
 
     pub fn grant_privileges(&mut self, object: &GrantObject, privileges: UserPrivilegeSet) {

src/common/storage/src/stage.rs

@@ -58,6 +58,10 @@ pub enum UserPrivilegeType {
     Grant = 1 << 12,
     // Privilege to Create Stage.
     CreateStage = 1 << 13,
+    // Privilege to Drop role.
+    DropRole = 1 << 14,
+    // Privilege to Drop user.
+    DropUser = 1 << 15,
     // TODO: remove this later
     Set = 1 << 4,
 }
@@ -73,7 +77,9 @@ const ALL_PRIVILEGES: BitFlags<UserPrivilegeType> = make_bitflags!(
         | Alter
         | Super
         | CreateUser
+        | DropUser
         | CreateRole
+        | DropRole
         | Grant
         | CreateStage
         | Set
@@ -93,7 +99,9 @@ impl std::fmt::Display for UserPrivilegeType {
             UserPrivilegeType::Alter => "ALTER",
             UserPrivilegeType::Super => "SUPER",
             UserPrivilegeType::CreateUser => "CREATE USER",
+            UserPrivilegeType::DropUser => "DROP USER",
             UserPrivilegeType::CreateRole => "CREATE ROLE",
+            UserPrivilegeType::DropRole => "DROP ROLE",
             UserPrivilegeType::CreateStage => "CREATE STAGE",
             UserPrivilegeType::Grant => "GRANT",
             UserPrivilegeType::Set => "SET",
@@ -121,8 +129,7 @@ impl UserPrivilegeSet {
     /// on databases and tables, and has some Global only privileges.
     pub fn available_privileges_on_global() -> Self {
         let database_privs = Self::available_privileges_on_database();
-        let privs =
-            make_bitflags!(UserPrivilegeType::{ Usage | Super | CreateUser | CreateRole | Grant });
+        let privs = make_bitflags!(UserPrivilegeType::{ Usage | Super | CreateUser | DropUser | CreateRole | DropRole | Grant });
         (database_privs.privileges | privs).into()
     }