Merge branch 'main' into subexpression

Author: SkyFan2002

Cargo.lock

@@ -22,7 +22,7 @@
 
 <br>
 
-<img src="https://github.com/databendlabs/databend/assets/172204/9997d8bc-6462-4dbd-90e3-527cf50a709c" alt="databend" />
+<img src="https://github.com/user-attachments/assets/4c288d5c-9365-44f7-8cde-b2c7ebe15622" alt="databend" />
 
 ## Why Databend?
 

README.md

@@ -305,12 +305,7 @@ pub type AsyncDmaFile = DmaFile<AsyncFile>;
 
 impl AsyncDmaFile {
     async fn open_fd(path: impl AsRef<Path>, dio: bool) -> io::Result<AsyncFile> {
-        let flags = if cfg!(target_os = "linux") && dio {
-            OFlags::DIRECT.bits() as i32
-        } else {
-            0
-        };
-
+        let flags = flags_direct_or_empty(dio).bits() as i32;
         AsyncFile::options()
             .read(true)
             .custom_flags(flags)
@@ -319,12 +314,7 @@ impl AsyncDmaFile {
     }
 
     async fn create_fd(path: impl AsRef<Path>, dio: bool) -> io::Result<AsyncFile> {
-        let flags = if cfg!(target_os = "linux") && dio {
-            OFlags::EXCL | OFlags::DIRECT
-        } else {
-            OFlags::EXCL
-        };
-
+        let flags = flags_direct_or_empty(dio) | OFlags::EXCL;
         AsyncFile::options()
             .write(true)
             .create(true)
@@ -421,25 +411,34 @@ impl AsyncDmaFile {
     }
 }
 
+#[cfg(target_os = "linux")]
+fn flags_direct_or_empty(dio: bool) -> OFlags {
+    if dio {
+        OFlags::DIRECT
+    } else {
+        OFlags::empty()
+    }
+}
+
+#[cfg(not(target_os = "linux"))]
+fn flags_direct_or_empty(_dio: bool) -> OFlags {
+    OFlags::empty()
+}
+
 pub type SyncDmaFile = DmaFile<OwnedFd>;
 
 impl SyncDmaFile {
     fn open_fd(path: impl rustix::path::Arg, dio: bool) -> io::Result<OwnedFd> {
-        let flags = if cfg!(target_os = "linux") && dio {
-            OFlags::RDONLY | OFlags::DIRECT
-        } else {
-            OFlags::RDONLY
-        };
+        let flags = OFlags::RDONLY | flags_direct_or_empty(dio);
         rustix::fs::open(path, flags, rustix::fs::Mode::empty()).map_err(io::Error::from)
     }
 
     fn create_fd(path: impl rustix::path::Arg, dio: bool) -> io::Result<OwnedFd> {
-        let flags = if cfg!(target_os = "linux") && dio {
-            OFlags::EXCL | OFlags::CREATE | OFlags::TRUNC | OFlags::WRONLY | OFlags::DIRECT
-        } else {
-            OFlags::EXCL | OFlags::CREATE | OFlags::TRUNC | OFlags::WRONLY
-        };
-
+        let flags = OFlags::EXCL
+            | OFlags::CREATE
+            | OFlags::TRUNC
+            | OFlags::WRONLY
+            | flags_direct_or_empty(dio);
         rustix::fs::open(path, flags, rustix::fs::Mode::from_raw_mode(0o666))
             .map_err(io::Error::from)
     }

src/common/base/src/base/dma.rs

@@ -289,6 +289,8 @@ build_exceptions! {
     AlterTableError(1132),
     /// Constraint error
     ConstraintError(1133),
+    /// Unknown row policy
+    UnknownMaskPolicy(1134),
 }
 
 // Sequence Errors [1124-1126, 3101-3102]

src/common/exception/src/exception_code.rs

@@ -19,7 +19,7 @@ use databend_common_meta_app::tenant::ToTenant;
 use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_kvapi::kvapi::Key;
 use databend_common_meta_types::anyerror::func_name;
-use databend_common_meta_types::protobuf::FetchAddU64Response;
+use databend_common_meta_types::protobuf::FetchIncreaseU64Response;
 use databend_common_meta_types::MetaError;
 use databend_common_meta_types::TxnOp;
 use databend_common_meta_types::TxnRequest;
@@ -46,7 +46,7 @@ where KV: kvapi::KVApi<Error = MetaError> + ?Sized
         self,
         tenant: impl ToTenant,
         count: u64,
-    ) -> Result<Result<FetchAddU64Response, AutoIncrementError>, MetaTxnError> {
+    ) -> Result<Result<FetchIncreaseU64Response, AutoIncrementError>, MetaTxnError> {
         debug!("{}", func_name!());
 
         // Key for the sequence number value.
@@ -69,7 +69,7 @@ where KV: kvapi::KVApi<Error = MetaError> + ?Sized
         );
         debug_assert!(succ);
 
-        let resp = responses[0].try_as_fetch_add_u64().unwrap();
+        let resp = responses[0].try_as_fetch_increase_u64().unwrap();
         let got_delta = resp.delta();
 
         if got_delta < delta {

src/meta/api/src/auto_increment_nextval_impl.rs

@@ -17,6 +17,7 @@ use databend_common_meta_app::data_mask::CreateDatamaskReq;
 use databend_common_meta_app::data_mask::DataMaskId;
 use databend_common_meta_app::data_mask::DataMaskNameIdent;
 use databend_common_meta_app::data_mask::DatamaskMeta;
+use databend_common_meta_app::tenant::Tenant;
 use databend_common_meta_types::MetaError;
 use databend_common_meta_types::SeqV;
 
@@ -40,4 +41,10 @@ pub trait DatamaskApi: Send + Sync {
         &self,
         name_ident: &DataMaskNameIdent,
     ) -> Result<Option<SeqV<DatamaskMeta>>, MetaError>;
+
+    async fn get_data_mask_by_id(
+        &self,
+        tenant: &Tenant,
+        policy_id: u64,
+    ) -> Result<Option<SeqV<DatamaskMeta>>, MetaError>;
 }

src/meta/api/src/data_mask_api.rs

@@ -23,7 +23,7 @@ use databend_common_meta_app::data_mask::MaskPolicyTableIdListIdent;
 use databend_common_meta_app::data_mask::MaskpolicyTableIdList;
 use databend_common_meta_app::id_generator::IdGenerator;
 use databend_common_meta_app::schema::CreateOption;
-use databend_common_meta_app::schema::TableId;
+use databend_common_meta_app::tenant::Tenant;
 use databend_common_meta_app::KeyWithTenant;
 use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_types::MetaError;
@@ -40,7 +40,6 @@ use crate::txn_backoff::txn_backoff;
 use crate::txn_condition_util::txn_cond_eq_seq;
 use crate::txn_core_util::send_txn;
 use crate::txn_core_util::txn_delete_exact;
-use crate::txn_core_util::txn_replace_exact;
 use crate::txn_op_builder_util::txn_op_put_pb;
 
 /// DatamaskApi is implemented upon kvapi::KVApi.
@@ -82,8 +81,6 @@ impl<KV: kvapi::KVApi<Error = MetaError>> DatamaskApi for KV {
 
                         txn_delete_exact(&mut txn, &id_ident, seq_meta.seq);
 
-                        clear_table_column_mask_policy(self, name_ident, &mut txn).await?;
-
                         curr_seq = seq_id.seq;
                     }
                 };
@@ -111,8 +108,9 @@ impl<KV: kvapi::KVApi<Error = MetaError>> DatamaskApi for KV {
                 let id_list = MaskpolicyTableIdList::default();
                 txn.condition.push(txn_cond_eq_seq(name_ident, curr_seq));
                 txn.if_then.extend(vec![
-                    txn_op_put_pb(name_ident, &id, None)?,        // name -> db_id
-                    txn_op_put_pb(&id_ident, &meta, None)?,       // id -> meta
+                    txn_op_put_pb(name_ident, &id, None)?,  // name -> db_id
+                    txn_op_put_pb(&id_ident, &meta, None)?, // id -> meta
+                    // TODO: Tentative retention for compatibility MaskPolicyTableIdListIdent related logic. It can be directly deleted later
                     txn_op_put_pb(&id_list_key, &id_list, None)?, // data mask name -> id_list
                 ]);
 
@@ -143,7 +141,6 @@ impl<KV: kvapi::KVApi<Error = MetaError>> DatamaskApi for KV {
         let mut trials = txn_backoff(None, func_name!());
         loop {
             trials.next().unwrap()?.await;
-
             let mut txn = TxnRequest::default();
 
             let res = self.get_id_and_value(name_ident).await?;
@@ -157,9 +154,8 @@ impl<KV: kvapi::KVApi<Error = MetaError>> DatamaskApi for KV {
 
             txn_delete_exact(&mut txn, name_ident, seq_id.seq);
             txn_delete_exact(&mut txn, &id_ident, seq_meta.seq);
-
+            // TODO: Tentative retention for compatibility MaskPolicyTableIdListIdent related logic. It can be directly deleted later
             clear_table_column_mask_policy(self, name_ident, &mut txn).await?;
-
             let (succ, _responses) = send_txn(self, txn).await?;
             debug!(succ = succ;"{}", func_name!());
 
@@ -179,6 +175,20 @@ impl<KV: kvapi::KVApi<Error = MetaError>> DatamaskApi for KV {
 
         Ok(res.map(|(_, seq_meta)| seq_meta))
     }
+
+    async fn get_data_mask_by_id(
+        &self,
+        tenant: &Tenant,
+        policy_id: u64,
+    ) -> Result<Option<SeqV<DatamaskMeta>>, MetaError> {
+        debug!(req :? =(policy_id); "DatamaskApi: {}", func_name!());
+
+        let id = DataMaskId::new(policy_id);
+        let id_ident = DataMaskIdIdent::new_generic(tenant, id);
+
+        let res = self.get_pb(&id_ident).await?;
+        Ok(res)
+    }
 }
 
 async fn clear_table_column_mask_policy(
@@ -195,30 +205,5 @@ async fn clear_table_column_mask_policy(
     };
 
     txn_delete_exact(txn, &id_list_key, seq_id_list.seq);
-
-    // remove mask policy from table meta
-    for table_id in seq_id_list.data.id_list.into_iter() {
-        let tbid = TableId { table_id };
-
-        let seq_meta = kv_api.get_pb(&tbid).await?;
-
-        let Some(seq_meta) = seq_meta else {
-            continue;
-        };
-
-        let (seq, mut meta) = (seq_meta.seq, seq_meta.data);
-
-        if let Some(column_mask_policy) = meta.column_mask_policy {
-            let new_column_mask_policy = column_mask_policy
-                .into_iter()
-                .filter(|(_, name)| name != name_ident.name())
-                .collect();
-
-            meta.column_mask_policy = Some(new_column_mask_policy);
-
-            txn_replace_exact(txn, &tbid, seq, &meta)?;
-        }
-    }
-
     Ok(())
 }