refactor: simplify least-visible-time schema API, add LeastVisibleTim…

…eIdent (#16418)
datafuselabs · Sep 9, 2024 · a0940db · a0940db
1 parent 3bc7a91
commit a0940db
Show file tree

Hide file tree

Showing 8 changed files with 205 additions and 166 deletions.
diff --git a/src/meta/api/src/name_value_api.rs b/src/meta/api/src/name_value_api.rs
@@ -21,6 +21,7 @@ use databend_common_meta_app::tenant_key::ident::TIdent;
 use databend_common_meta_app::tenant_key::resource::TenantResource;
 use databend_common_meta_kvapi::kvapi::KVApi;
 use databend_common_meta_kvapi::kvapi::KeyCodec;
+use databend_common_meta_types::Change;
 use databend_common_meta_types::MatchSeq;
 use databend_common_meta_types::MetaError;
 use databend_common_meta_types::SeqValue;
@@ -46,12 +47,12 @@ use crate::util::txn_op_put_pb;
 pub trait NameValueApi<R, N>: KVApi<Error = MetaError>
 where
     R: TenantResource + Send + Sync + 'static,
-    R::ValueType: FromToProto + Send + Sync + 'static,
+    R::ValueType: FromToProto + Clone + Send + Sync + 'static,
     N: KeyCodec,
     N: fmt::Debug + Clone + Send + Sync + 'static,
 {
     /// Create a `name -> value` mapping.
-    async fn create_name_value(
+    async fn insert_name_value(
         &self,
         name_ident: TIdent<R, N>,
         value: R::ValueType,
@@ -75,7 +76,7 @@ where
     }
 
     /// Create a `name -> value` mapping, with `CreateOption` support
-    async fn create_name_value_with_create_option(
+    async fn insert_name_value_with_create_option(
         &self,
         name_ident: TIdent<R, N>,
         value: R::ValueType,
@@ -97,6 +98,41 @@ where
         Ok(Ok(()))
     }
 
+    /// Update or insert a `name -> value` mapping.
+    ///
+    /// The `update` function is called with the previous value and should output the updated to write back.
+    /// If it outputs `None`, nothing is written back.
+    async fn upsert_name_value_with(
+        &self,
+        name_ident: &TIdent<R, N>,
+        update: impl Fn(Option<R::ValueType>) -> Option<R::ValueType> + Send,
+    ) -> Result<Change<R::ValueType>, MetaTxnError> {
+        debug!(name_ident :? =name_ident; "NameValueApi: {}", func_name!());
+
+        let mut trials = txn_backoff(None, func_name!());
+        loop {
+            trials.next().unwrap()?.await;
+
+            let seq_meta = self.get_pb(name_ident).await?;
+            let seq = seq_meta.seq();
+
+            let updated = match update(seq_meta.clone().into_value()) {
+                Some(x) => x,
+                None => return Ok(Change::new(seq_meta.clone(), seq_meta)),
+            };
+
+            let transition = self
+                .upsert_pb(
+                    &UpsertPB::insert(name_ident.clone(), updated).with(MatchSeq::Exact(seq)),
+                )
+                .await?;
+
+            if transition.is_changed() {
+                return Ok(transition);
+            }
+        }
+    }
+
     /// Update an existent `name -> value` mapping.
     ///
     /// The `update` function is called with the previous value
@@ -174,7 +210,7 @@ impl<R, N, T> NameValueApi<R, N> for T
 where
     T: KVApi<Error = MetaError> + ?Sized,
     R: TenantResource + Send + Sync + 'static,
-    R::ValueType: FromToProto + Send + Sync + 'static,
+    R::ValueType: FromToProto + Clone + Send + Sync + 'static,
     N: KeyCodec,
     N: fmt::Debug + Clone + Send + Sync + 'static,
 {

diff --git a/src/meta/api/src/schema_api.rs b/src/meta/api/src/schema_api.rs
@@ -19,6 +19,7 @@ use databend_common_meta_app::schema::dictionary_id_ident::DictionaryId;
 use databend_common_meta_app::schema::dictionary_name_ident::DictionaryNameIdent;
 use databend_common_meta_app::schema::index_id_ident::IndexId;
 use databend_common_meta_app::schema::index_id_ident::IndexIdIdent;
+use databend_common_meta_app::schema::least_visible_time_ident::LeastVisibleTimeIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CatalogMeta;
 use databend_common_meta_app::schema::CatalogNameIdent;
@@ -49,13 +50,12 @@ use databend_common_meta_app::schema::ExtendLockRevReq;
 use databend_common_meta_app::schema::GcDroppedTableReq;
 use databend_common_meta_app::schema::GetDatabaseReq;
 use databend_common_meta_app::schema::GetIndexReply;
-use databend_common_meta_app::schema::GetLVTReply;
-use databend_common_meta_app::schema::GetLVTReq;
 use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::GetTableReq;
 use databend_common_meta_app::schema::IndexMeta;
 use databend_common_meta_app::schema::IndexNameIdent;
+use databend_common_meta_app::schema::LeastVisibleTime;
 use databend_common_meta_app::schema::ListCatalogReq;
 use databend_common_meta_app::schema::ListDatabaseReq;
 use databend_common_meta_app::schema::ListDictionaryReq;
@@ -72,8 +72,6 @@ use databend_common_meta_app::schema::RenameDatabaseReply;
 use databend_common_meta_app::schema::RenameDatabaseReq;
 use databend_common_meta_app::schema::RenameTableReply;
 use databend_common_meta_app::schema::RenameTableReq;
-use databend_common_meta_app::schema::SetLVTReply;
-use databend_common_meta_app::schema::SetLVTReq;
 use databend_common_meta_app::schema::SetTableColumnMaskPolicyReply;
 use databend_common_meta_app::schema::SetTableColumnMaskPolicyReq;
 use databend_common_meta_app::schema::TableId;
@@ -95,10 +93,12 @@ use databend_common_meta_app::schema::UpdateVirtualColumnReq;
 use databend_common_meta_app::schema::UpsertTableOptionReply;
 use databend_common_meta_app::schema::UpsertTableOptionReq;
 use databend_common_meta_app::schema::VirtualColumnMeta;
+use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_types::seq_value::SeqV;
 use databend_common_meta_types::Change;
 use databend_common_meta_types::MetaError;
 use databend_common_meta_types::MetaId;
+use databend_common_proto_conv::FromToProto;
 
 use crate::kv_app_error::KVAppError;
 use crate::meta_txn_error::MetaTxnError;
@@ -307,8 +307,23 @@ pub trait SchemaApi: Send + Sync {
     -> Result<Vec<Arc<CatalogInfo>>, KVAppError>;
 
     // least visible time
-    async fn set_table_lvt(&self, req: SetLVTReq) -> Result<SetLVTReply, KVAppError>;
-    async fn get_table_lvt(&self, req: GetLVTReq) -> Result<GetLVTReply, KVAppError>;
+
+    /// Updates the table's least visible time (LVT) only if the new value is greater than the existing one.
+    ///
+    /// This function returns the updated LVT if changed, or the existing LVT if no update was necessary.
+    async fn set_table_lvt(
+        &self,
+        name_ident: &LeastVisibleTimeIdent,
+        value: &LeastVisibleTime,
+    ) -> Result<LeastVisibleTime, KVAppError>;
+
+    #[deprecated(note = "use get::<K>() instead")]
+    async fn get_table_lvt(
+        &self,
+        name_ident: &LeastVisibleTimeIdent,
+    ) -> Result<Option<LeastVisibleTime>, KVAppError> {
+        Ok(self.get(name_ident).await?)
+    }
 
     fn name(&self) -> String;
 
@@ -337,4 +352,12 @@ pub trait SchemaApi: Send + Sync {
         &self,
         req: ListDictionaryReq,
     ) -> Result<Vec<(String, DictionaryMeta)>, KVAppError>;
+
+    /// Generic get() implementation for any kvapi::Key.
+    ///
+    /// This method just return an `Option` of the value without seq number.
+    async fn get<K>(&self, name_ident: &K) -> Result<Option<K::ValueType>, MetaError>
+    where
+        K: kvapi::Key + Sync + 'static,
+        K::ValueType: FromToProto + 'static;
 }
diff --git a/src/meta/api/src/schema_api_impl.rs b/src/meta/api/src/schema_api_impl.rs
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::any::type_name;
 use std::cmp::min;
 use std::collections::BTreeMap;
 use std::collections::BTreeSet;
@@ -63,6 +64,7 @@ use databend_common_meta_app::schema::index_id_ident::IndexId;
 use databend_common_meta_app::schema::index_id_ident::IndexIdIdent;
 use databend_common_meta_app::schema::index_id_to_name_ident::IndexIdToNameIdent;
 use databend_common_meta_app::schema::index_name_ident::IndexName;
+use databend_common_meta_app::schema::least_visible_time_ident::LeastVisibleTimeIdent;
 use databend_common_meta_app::schema::CatalogIdToNameIdent;
 use databend_common_meta_app::schema::CatalogInfo;
 use databend_common_meta_app::schema::CatalogMeta;
@@ -105,16 +107,13 @@ use databend_common_meta_app::schema::ExtendLockRevReq;
 use databend_common_meta_app::schema::GcDroppedTableReq;
 use databend_common_meta_app::schema::GetDatabaseReq;
 use databend_common_meta_app::schema::GetIndexReply;
-use databend_common_meta_app::schema::GetLVTReply;
-use databend_common_meta_app::schema::GetLVTReq;
 use databend_common_meta_app::schema::GetTableCopiedFileReply;
 use databend_common_meta_app::schema::GetTableCopiedFileReq;
 use databend_common_meta_app::schema::GetTableReq;
 use databend_common_meta_app::schema::IndexMeta;
 use databend_common_meta_app::schema::IndexNameIdent;
 use databend_common_meta_app::schema::IndexNameIdentRaw;
 use databend_common_meta_app::schema::LeastVisibleTime;
-use databend_common_meta_app::schema::LeastVisibleTimeKey;
 use databend_common_meta_app::schema::ListCatalogReq;
 use databend_common_meta_app::schema::ListDatabaseReq;
 use databend_common_meta_app::schema::ListDictionaryReq;
@@ -131,8 +130,6 @@ use databend_common_meta_app::schema::RenameDatabaseReply;
 use databend_common_meta_app::schema::RenameDatabaseReq;
 use databend_common_meta_app::schema::RenameTableReply;
 use databend_common_meta_app::schema::RenameTableReq;
-use databend_common_meta_app::schema::SetLVTReply;
-use databend_common_meta_app::schema::SetLVTReq;
 use databend_common_meta_app::schema::SetTableColumnMaskPolicyAction;
 use databend_common_meta_app::schema::SetTableColumnMaskPolicyReply;
 use databend_common_meta_app::schema::SetTableColumnMaskPolicyReq;
@@ -188,6 +185,7 @@ use databend_common_meta_types::TxnGetRequest;
 use databend_common_meta_types::TxnGetResponse;
 use databend_common_meta_types::TxnOp;
 use databend_common_meta_types::TxnRequest;
+use databend_common_proto_conv::FromToProto;
 use fastrace::func_name;
 use futures::TryStreamExt;
 use log::debug;
@@ -955,7 +953,7 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
             updated_on: None,
         };
 
-        self.create_name_value_with_create_option(
+        self.insert_name_value_with_create_option(
             req.name_ident.clone(),
             virtual_column_meta,
             req.create_option,
@@ -3029,7 +3027,7 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
         // Revision is unique. if it presents, consider it as success.
         // Thus, we could just ignore create result
         let _create_res = self
-            .create_name_value(key, lock_meta, Some(req.ttl))
+            .insert_name_value(key, lock_meta, Some(req.ttl))
             .await?;
 
         Ok(CreateLockRevReply { revision })
@@ -3186,66 +3184,38 @@ impl<KV: kvapi::KVApi<Error = MetaError> + ?Sized> SchemaApi for KV {
 
     #[logcall::logcall]
     #[fastrace::trace]
-    async fn set_table_lvt(&self, req: SetLVTReq) -> Result<SetLVTReply, KVAppError> {
-        debug!(req :? =(&req); "SchemaApi: {}", func_name!());
-
-        let table_id = req.table_id;
-
-        let mut trials = txn_backoff(None, func_name!());
-        loop {
-            trials.next().unwrap()?.await;
-
-            let lvt_key = LeastVisibleTimeKey { table_id };
-            let (lvt_seq, lvt_opt): (_, Option<LeastVisibleTime>) =
-                get_pb_value(self, &lvt_key).await?;
-            let new_time = match lvt_opt {
-                Some(lvt) => {
-                    if lvt.time >= req.time {
-                        return Ok(SetLVTReply { time: lvt.time });
-                    } else {
-                        req.time
-                    }
+    async fn set_table_lvt(
+        &self,
+        name_ident: &LeastVisibleTimeIdent,
+        value: &LeastVisibleTime,
+    ) -> Result<LeastVisibleTime, KVAppError> {
+        debug!(req :? =(&name_ident, &value); "SchemaApi: {}", func_name!());
+
+        let transition = self
+            .upsert_name_value_with(name_ident, |t: Option<LeastVisibleTime>| {
+                let curr = t.unwrap_or_default();
+                if curr.time >= value.time {
+                    None
+                } else {
+                    Some(value.clone())
                 }
-                None => req.time,
-            };
-
-            let new_lvt = LeastVisibleTime { time: new_time };
-
-            let txn_req = TxnRequest {
-                condition: vec![txn_cond_seq(&lvt_key, Eq, lvt_seq)],
-                if_then: vec![txn_op_put(&lvt_key, serialize_struct(&new_lvt)?)],
-                else_then: vec![],
-            };
-
-            let (succ, _responses) = send_txn(self, txn_req).await?;
-
-            debug!(
-                name :? =(req.table_id),
-                succ = succ;
-                "set_table_lvt"
-            );
+            })
+            .await?;
 
-            if succ {
-                return Ok(SetLVTReply { time: new_time });
-            }
-        }
+        return Ok(transition.result.into_value().unwrap_or_default());
     }
 
     #[logcall::logcall]
     #[fastrace::trace]
-    async fn get_table_lvt(&self, req: GetLVTReq) -> Result<GetLVTReply, KVAppError> {
-        debug!(req :? =(&req); "SchemaApi: {}", func_name!());
-
-        let table_id = req.table_id;
-
-        let lvt_key = LeastVisibleTimeKey { table_id };
-
-        let seq_lvt = self.get_pb(&lvt_key).await?;
-        let lvt_opt = seq_lvt.into_value();
+    async fn get<K>(&self, name_ident: &K) -> Result<Option<K::ValueType>, MetaError>
+    where
+        K: kvapi::Key + Sync + 'static,
+        K::ValueType: FromToProto + 'static,
+    {
+        debug!(req :? =(&name_ident); "SchemaApi::get::<{}>()", typ::<K>());
 
-        Ok(GetLVTReply {
-            time: lvt_opt.map(|time| time.time),
-        })
+        let seq_lvt = self.get_pb(name_ident).await?;
+        Ok(seq_lvt.into_value())
     }
 
     fn name(&self) -> String {
@@ -4553,3 +4523,10 @@ async fn append_update_stream_meta_requests(
     }
     Ok(())
 }
+
+fn typ<K>() -> &'static str {
+    type_name::<K>()
+        .rsplit("::")
+        .next()
+        .unwrap_or("UnknownType")
+}