feat: add tag DDL support (CREATE/DROP/SHOW only) (#19109)

* feat: add tag DDL support (CREATE/DROP/SHOW only)

* fix conversation

* add tag <-> object ref

* refactor

* rename some struct

* extract TagError

* fix conversation

* add reserved TagMeta::drop_on

* split TagError, TagMetaError; add some comments

* retry set_object_tags and drop concurrent tag error

- set_object_tags now perform an optimistic retry with txn_backoff to keep allowed_values consistent with the latest tag meta.
- remove the TagMetaConcurrentModification errors and corresponding exception code, instead retry logic processing.

* meta: add explicit tag allowed_values enforcement flag

  - Proto3 `repeated` fields can’t distinguish `None` from `Some(Vec::new())`, so
    extending TagMeta with a dedicated `enforce_allowed_values` flag lets us tell
    whether a tag declared `ALLOWED_VALUES` even when the list is empty.
  - Split the semantics in Rust too: keep `allowed_values: Vec<String>` strictly
    for ordered values, and use the boolean to record if the constraint is active
    instead of overloading `Option<Vec<_>>`.
  - Update CREATE TAG interpreter, tag validation, error text, and system.tags
    display to respect the new structure, and refresh the proto conversion tests.

* optimize

- represent tag allowed_values as Option<Vec<String>> without an enforce flag
- update tag.proto + conversions to the new AllowedValues message
- simplify validation/error paths and keep system.tags display consistent

Author: TCeason

Cargo.lock

@@ -373,6 +373,18 @@ build_exceptions! {
     IndexColumnIdNotFound(2726),
 }
 
+// Tag Errors [2750-2753]
+build_exceptions! {
+    /// Tag already exists
+    TagAlreadyExists(2750),
+    /// Unknown tag
+    UnknownTag(2751),
+    /// Invalid value for tag
+    NotAllowedTagValue(2752),
+    /// Tag still has references
+    TagHasReferences(2753),
+}
+
 // Cloud and Integration Errors [1701-1703]
 build_exceptions! {
     /// Cloud control connect error

src/common/exception/src/exception_code.rs

@@ -128,7 +128,9 @@ where
         let name_id_values = self.list_id_value(&dir).await?;
 
         let catalog_infos = name_id_values
-            .map(|(name, id, seq_meta)| Arc::new(CatalogInfo::new(name, id, seq_meta.data)))
+            .map(|(name, id_seqv, seq_meta)| {
+                Arc::new(CatalogInfo::new(name, id_seqv.data, seq_meta.data))
+            })
             .collect::<Vec<_>>();
 
         Ok(catalog_infos)

src/meta/api/src/catalog_api.rs

@@ -217,10 +217,10 @@ where
         let index_metas = name_id_metas
             // table_id is not specified
             // or table_id is specified and equals to the given table_id.
-            .filter(|(_k, _id, seq_meta)| {
+            .filter(|(_k, _seq_id, seq_meta)| {
                 req.table_id.is_none() || req.table_id == Some(seq_meta.table_id)
             })
-            .map(|(k, id, seq_meta)| (k.index_name().to_string(), id, seq_meta.data))
+            .map(|(k, id_seqv, seq_meta)| (k.index_name().to_string(), id_seqv.data, seq_meta.data))
             .collect::<Vec<_>>();
 
         Ok(index_metas)

src/meta/api/src/index_api.rs

@@ -45,6 +45,7 @@ pub mod security_api;
 mod sequence_api;
 pub mod serialization_util;
 pub mod table_api;
+pub mod tag_api;
 pub(crate) mod testing;
 pub mod txn_backoff;
 pub mod txn_condition_util;
@@ -72,6 +73,7 @@ pub use data_mask_api::DatamaskApi;
 pub use data_retention_util::get_retention_boundary;
 pub use data_retention_util::is_drop_time_retainable;
 pub use database_api::DatabaseApi;
+pub use databend_common_meta_app::schema::TagError;
 pub use dictionary_api::DictionaryApi;
 // Re-export from new error_util module for backward compatibility
 pub use error_util::assert_table_exist;
@@ -102,6 +104,7 @@ pub use serialization_util::{
     deserialize_struct, deserialize_u64, serialize_struct, serialize_u64,
 };
 pub use table_api::TableApi;
+pub use tag_api::TagApi;
 // Re-export from new txn_condition_util module for backward compatibility
 pub use txn_condition_util::txn_cond_eq_seq;
 pub use txn_condition_util::txn_cond_seq;

src/meta/api/src/lib.rs

@@ -274,9 +274,9 @@ where
         self.get_id_and_value(key).await
     }
 
-    /// list by name prefix, returns a list of `(name, id, value)`
+    /// list by name prefix, returns a list of `(name, SeqV<id>, SeqV<value>)`
     ///
-    /// Returns an iterator of `(name, id, SeqV<value>)` tuples.
+    /// Returns an iterator of `(name, SeqV<id>, SeqV<value>)` tuples.
     /// This function list all the ids by a name prefix,
     /// then get the `id->value` mapping and finally zip these two together.
     // Using `async fn` does not allow using `impl Iterator` in the return type.
@@ -286,7 +286,7 @@ where
         prefix: &DirName<K>,
     ) -> impl Future<
         Output = Result<
-            impl Iterator<Item = (K, DataId<IdRsc>, SeqV<IdRsc::ValueType>)>,
+            impl Iterator<Item = (K, SeqV<DataId<IdRsc>>, SeqV<IdRsc::ValueType>)>,
             MetaError,
         >,
     > + Send {
@@ -295,10 +295,9 @@ where
             let strm = self.list_pb(prefix).await?;
             let name_ids = strm.try_collect::<Vec<_>>().await?;
 
-            let id_idents = name_ids.iter().map(|itm| {
-                let id = itm.seqv.data;
-                id.into_t_ident(tenant)
-            });
+            let id_idents = name_ids
+                .iter()
+                .map(|itm| itm.seqv.data.into_t_ident(tenant));
 
             let strm = self.get_pb_values(id_idents).await?;
             let seq_metas = strm.try_collect::<Vec<_>>().await?;
@@ -308,7 +307,7 @@ where
                     .into_iter()
                     .zip(seq_metas)
                     .filter_map(|(itm, opt_seq_meta)| {
-                        opt_seq_meta.map(|seq_meta| (itm.key, itm.seqv.data, seq_meta))
+                        opt_seq_meta.map(|seq_meta| (itm.key, itm.seqv, seq_meta))
                     });
 
             Ok(name_id_values)