Err(existent) => match req.create_option {                CreateOption::Create => Err(TagError::already_exists(                    req.name_ident.tag_name().to_string(),                )),                CreateOption::CreateIfNotExists => Ok(CreateTagReply {                    tag_id: *existent.data,                }),                CreateOption::CreateOrReplace => unreachable!("CreateOrReplace handled above"),            },

    async fn drop_tag(&self, req: DropTagReq) -> Result<Result<(), TagError>, MetaTxnError> {        debug!(req :? =(&req); "SchemaApi: {}", func_name!());        let references = collect_tag_references(self,&ListTagReferencesReq {            tenant: req.name_ident.tenant().clone(),            tag_name: Some(req.name_ident.tag_name().to_string()),            object_type: None,        })        .await        .map_err(MetaTxnError::from)?;        if !references.is_empty() {            return Ok(Err(TagError::has_references(                req.name_ident.tag_name().to_string(),                references.len(),            )));        }        // NOTE: The reference check above and the delete operation below are not transactional.        // A concurrent SET/UNSET TAGS call may introduce new references in between, which is an        // acceptable trade-off for the initial version of tag DDL support.        let removed = self.remove_id_value(&req.name_ident, |_| vec![]).await?;        if removed.is_none()&& !req.if_exists {            return Ok(Err(TagError::not_found(                req.name_ident.tag_name().to_string(),                "drop_tag",            )));        }        Ok(Ok(()))    }

messageTagMeta {uint64ver=100;uint64min_reader_ver=101;repeatedstringallowed_values=1;stringcomment=2;stringcreated_on=3;optionalstringupdated_on=4;}messageTagRefValue {uint64ver=100;uint64min_reader_ver=101;uint64tag_id=1;stringvalue=2;stringcreated_on=3;}

                if req.if_exists {                    return Ok(Ok(()));                }                return Ok(Err(TagError::not_found(                    req.name_ident.tag_name().to_string(),                    "drop_tag",                )));

#[logcall::logcall]#[fastrace::trace]asyncfnlist_tags(&self,tenant:&Tenant) ->Result<Vec<TagInfo>,MetaError>{debug!(tenant:? =(tenant);"SchemaApi: {}", func_name!());let tag_name_key =TagNameIdent::new(tenant,"dummy");let tag_dir =DirName::new(tag_name_key);let tags =self.list_id_value(&tag_dir).await?.map(|(name_ident, tag_id, meta_seqv)|TagInfo{name: name_ident.tag_name().to_string(),tag_id:*tag_id,meta: meta_seqv,}).collect();Ok(tags)}

/// list by name prefix, returns a list of `(name, id, value)`////// Returns an iterator of `(name, 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.// Thus we use `impl Future` instead.fn list_id_value(&self,prefix:&DirName<K>,) ->implFuture<Output =Result<implIterator<Item =(K,DataId<IdRsc>,SeqV<IdRsc::ValueType>)>,MetaError,>,> +Send{

Thus, maps are encoded almost exactly like a repeated message field: as a sequence of LEN-typed records, with two fields each. The exception is that order is not guaranteed to be preserved with maps during serialization.

        let (succ, _) = send_txn(self, TxnRequest::new(txn_conditions, txn_ops)).await?;        if succ {            Ok(Ok(()))        } else {            // Tag metadata was modified concurrently (e.g., allowed_values changed)            Ok(Err(TagError::concurrent_modification()))        }

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]pub struct TagMeta {

#[logcall::logcall]#[fastrace::trace]asyncfn set_object_tags(&self,req:SetObjectTagsReq,) ->Result<Result<(),TagError>,MetaError>{     ...for(((tag_id, tag_value), meta_opt), tag_meta_key) in         req.tags.iter().zip(tag_metas).zip(tag_meta_keys){// Verify tag existsletSome(meta_seqv) = meta_optelse{returnOk(Err(TagError::not_found(*tag_id)));};// Check allowed_values: None means any value is allowedifletSome(allowed) =&meta_seqv.data.allowed_values{if !allowed.contains(tag_value){returnOk(Err(TagError::not_allowed_value(*tag_id,                     tag_value.clone(),Some(allowed.clone()),)));}}         ...}

- Example:    CREATE TAG env_tag      ALLOWED_VALUES ('blue','red','green')      PROPAGATE= ON_DEPENDENCY      ON_CONFLICT= ALLOWED_VALUES_SEQUENCE;--    Suppose a view inherits env_tag from two base tables, one tagged env_tag = 'red' and the other env_tag = 'blue'. Because 'blue' appears before 'red' in the allowed list, the view’s final env_tag value becomes 'blue'. If the conflict involved 'green' and 'red', 'red' would win, and so on.

                return Ok(Ok(Some((id_seqv, meta_seqv))));            } else {                let reference_count = self.list_pb_vec(&refs_dir).await?.len();                if reference_count > 0 {                    return Ok(Err(TagError::tag_has_references(                        name_ident.tag_name().to_string(),                        reference_count,                    )));                }            }

#[derive(Clone,Debug, thiserror::Error,PartialEq,Eq)]pubenumTagError{#[error(transparent)]UnknownTagId(#[from]UnknownError<TagIdResource,TagId>),

pub enum TagError {    #[error(transparent)]    UnknownTagId(#[from] UnknownError<TagIdResource, TagId>),

/// Attach multiple tags to an object in a single transaction./// Validates tag existence and allowed_values constraints.////// The txn uses optimistic locking:/// - We must re-fetch `TagMeta` and verify values on every retry so each attempt uses the///   latest `allowed_values`./// - The generated conditions include `tag_meta.seq`, ensuring no concurrent update sneaks in///   between validation and commit. If any `TagMeta` changes, the txn fails and the loop///   re-executes the entire validation/write process.

    pub fn not_allowed_value(        tag_id: u64,        tag_value: impl Into<String>,        allowed_values: Option<Vec<String>>,    ) -> Self {        let allowed_values_display = match allowed_values {            Some(values) if !values.is_empty() => format!(                ". Allowed values: [{}]",                values                    .into_iter()                    .map(|v| format!("'{v}'"))                    .collect::<Vec<_>>()                    .join(", ")            ),            _ => String::new(),

createtablet(idint);create tag tag1 allowed_values'x','y';ALTER TAG tag1  DROP ALLOWED_VALUES'x';ALTER TAG tag1  DROP ALLOWED_VALUES'y';altertable t1set tag tag1='y';-- expect err.

message TagMeta {  uint64 ver = 100;  uint64 min_reader_ver = 101;  // Ordered list of string values that may be set on this tag.  // Matches CREATE TAG ... ALLOWED_VALUES semantics: must be declared before  // other parameters, accepts up to 5,000 entries, and when omitted any string  // (including empty) is accepted. Propagating tags rely on this order to  // resolve conflicts when ON_CONFLICT = ALLOWED_VALUES_SEQUENCE (e.g. `CREATE  // TAG my_tag ALLOWED_VALUES 'blue', 'red' PROPAGATE = ON_DEPENDENCY  // ON_CONFLICT = ALLOWED_VALUES_SEQUENCE` always picks `'blue'`). A conflict  // arises when the same downstream object inherits different values:  //  //   Tag definition: my_tag ALLOWED_VALUES 'blue', 'red'  //   Object A --(my_tag='blue')--> \  //                                   -> Object C (final = 'blue')  //   Object B --(my_tag='red')-->  /  //  // We keep a repeated field instead of a map to avoid undefined iteration  // order or manually maintained indices. With maps, gaps like `{blue:1, red:3}`  // (after deletes/edits) would desync the conflict priority from the declared  // order, so we store the sequence directly.  repeated string allowed_values = 1;  string comment = 2;  string created_on = 3;  optional string updated_on = 4;  optional string drop_on = 5;  // Distinguishes between `CREATE TAG ... ALLOWED_VALUES` (even if the list is  // currently empty) and tags that never configured allowed values. When false,  // the tag accepts any string. When true, only entries from `allowed_values`  // are valid; the list may be empty when all values have been dropped but the  // constraint still exists until an UNSET command runs.  bool enforce_allowed_values = 6;}

  repeated string allowed_values = 1;  string comment = 2;  string created_on = 3;  optional string updated_on = 4;  optional string drop_on = 5;// Distinguishes between `CREATE TAG ... ALLOWED_VALUES` (even if the list is// currently empty) and tags that never configured allowed values. When false,// the tag accepts any string. When true, only entries from `allowed_values`// are valid; the list may be empty when all values have been dropped but the// constraint still exists until an UNSET command runs.  bool enforce_allowed_values = 6;

message TagMeta {  uint64 ver = 100;  uint64 min_reader_ver = 101;  message AllowedValues {    // Ordered list of string values that may be set on this tag.    // Matches CREATE TAG ... ALLOWED_VALUES semantics: must be declared before    // other parameters, accepts up to 5,000 entries, and when omitted any string    // (including empty) is accepted. Propagating tags rely on this order to    // resolve conflicts when ON_CONFLICT = ALLOWED_VALUES_SEQUENCE (e.g. `CREATE    // TAG my_tag ALLOWED_VALUES 'blue', 'red' PROPAGATE = ON_DEPENDENCY    // ON_CONFLICT = ALLOWED_VALUES_SEQUENCE` always picks `'blue'`). A conflict    // arises when the same downstream object inherits different values:    //    //   Tag definition: my_tag ALLOWED_VALUES 'blue', 'red'    //   Object A --(my_tag='blue')--> \    //                                   -> Object C (final = 'blue')    //   Object B --(my_tag='red')-->  /    //    // We keep a repeated field instead of a map to avoid undefined iteration    // order or manually maintained indices. With maps, gaps like `{blue:1, red:3}`    // (after deletes/edits) would desync the conflict priority from the declared    // order, so we store the sequence directly.    repeated string values = 1;  }  optional AllowedValues allowed_values = 1;  string comment = 2;  string created_on = 3;  optional string updated_on = 4;  optional string drop_on = 5;}