feat: manage semantic table options via ALTER TABLE SET/UNSET (#8880)

* fix(meta): actually acquire logical table locks in alter-logical-tables procedure

The procedure listed its logical table locks from table_info_values,
which is only filled during Prepare, while procedure lock keys are
fixed at submission — so the logical locks were never acquired. Today
every writer of a logical table's info is serialized by the physical
table lock, which hides the problem; a metadata-only alter procedure
targeting a single logical table would race it.

Resolve the logical table ids at submission, persist them in the
procedure state (serde(default): state dumped by older versions keeps
the previous behavior), lock physical + logical tables, and re-check
the resolved ids against the locked set at Prepare so a table dropped
and recreated after submission cannot be mutated without a lock.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* feat: manage semantic table options via ALTER TABLE SET/UNSET

CREATE TABLE accepts greptime.semantic.* options, but ALTER TABLE SET
routed every option through SetRegionOption, whose closed match
rejects them — tables auto-created by ingestion could never receive
semantic declarations after the fact.

Semantic options are pure metadata markers no region consumes, so
they now take a metadata-only alter, following the repartition-hint
precedent:

- New AlterKind::SetAnnotations/UnsetAnnotations carrying an
  AnnotationFamily (currently only Semantic), so future marker-style
  option families reuse the same machinery. The converter classifies
  a SET/UNSET batch by key prefix and rejects batches that mix
  annotation keys with regular options.
- The procedure reuses the MetadataOnly flow: no region dispatch,
  table-info update plus cache invalidation only.
- Validation lives in the table-meta mutation layer, so it runs at
  frontend verification and again in the procedure's prepare step
  under the table lock: SET is strict (known key, value domain,
  entity columns exist and render as strings); UNSET is lenient
  inside the namespace so stale keys can be cleaned up.
  ModifyColumnTypes re-checks columns referenced by entity
  declarations at the same layer, closing a verify-then-execute race.
- Logical metric tables are supported: an annotation alter submits a
  regular alter-table task locking only the logical table, and the
  DDL manager's physical-route guard admits it.
- create_table_info re-checks semantic value domains for gRPC-built
  expressions that bypass the SQL parser.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor(table): centralize annotation option classification and validation

Address review feedback on the AnnotationFamily abstraction: with only
one variant that every consumer immediately destructured, the
generality was fake. Make it real and exhaustive instead:

- AnnotationFamily gains RepartitionHint: repartition.column.hint is
  the same kind of marker option (pure metadata, no region consumes
  it) and previously had a hand-rolled special case in the converter,
  the metadata-only classifier, and a dedicated AlterKind pair — all
  deleted, one classification API remains. Per-family logical-table
  eligibility (allows_logical_tables) replaces the hard-coded
  Semantic check in the DDL manager guard.
- One validation core in the table crate (check_annotation) serves
  both DDL entry points. CREATE and ALTER previously duplicated the
  rules; each keeps its existing error variants, status codes and
  messages via thin adapters over a typed error (ALTER missing column
  stays 4002 TableColumnNotFound, CREATE stays InvalidArguments).
- The batch classifier returns Result instead of swallowing the
  mixed-batch error: a mixed SET on a logical table now reports the
  actual problem instead of UnexpectedLogicalRouteTable, and the flow
  classifiers propagate instead of guessing. The converter also moves
  its owned payloads instead of cloning them.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* test(meta): cover logical-table annotation alter routing

The route-guard branch admitting metadata-only annotation alters on
logical tables was only exercised end to end by sqlness. Pin it at the
DDL manager level: a semantic SET on a logical table succeeds, updates
only the logical table's metadata and dispatches nothing to datanodes;
a mixed batch reports its own error instead of the route guard's; the
repartition hint stays rejected on logical routes.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* fix(table): keep entity guard on ADD COLUMN and report missing columns first

Review follow-ups: the old verify_alter loop scanned the post-alter
schema, so it also caught DROP COLUMN followed by re-adding the
declared column with a non-string type — the mutation-layer move only
kept the MODIFY path. Guard add_columns the same way (this also covers
ingestion auto-alter). And run the MODIFY drift check after the
existence lookup, so altering a dropped-but-still-declared column
reports ColumnNotExists (4002) like every other MODIFY on a missing
column.

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* style(grpc-expr): drop a test comment restating the classifier doc

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

* refactor(table): rename annotation validation helpers per review

check_annotation* validated and normalized; align the names with the
validate_and_normalize_* convention nearby, and spell out
AnnotationContext (Cx is not used in this repo).

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

---------

Signed-off-by: Dennis Zhuang <killme2008@gmail.com>

src/common/grpc-expr/src/alter.rs

@@ -29,8 +29,8 @@ use snafu::{OptionExt, ResultExt, ensure};
 use store_api::region_request::{SetRegionOption, UnsetRegionOption};
 use table::metadata::{TableId, TableMeta};
 use table::requests::{
-    AddColumnRequest, AlterKind, AlterTableRequest, ModifyColumnTypeRequest,
-    REPARTITION_COLUMN_HINT_KEY, SetDefaultRequest, SetIndexOption, UnsetIndexOption,
+    AddColumnRequest, AlterKind, AlterTableRequest, AnnotationFamily, ModifyColumnTypeRequest,
+    SetDefaultRequest, SetIndexOption, UnsetIndexOption,
 };
 
 use crate::error::{
@@ -44,6 +44,57 @@ use crate::error::{
 const LOCATION_TYPE_FIRST: i32 = LocationType::First as i32;
 const LOCATION_TYPE_AFTER: i32 = LocationType::After as i32;
 
+/// Classifies a SET/UNSET key batch: `Ok(Some(family))` when every key belongs
+/// to the same annotation family, `Ok(None)` when none does, and an error on a
+/// mixed batch — annotation alters skip region dispatch, so they cannot share
+/// a statement with options that regions must see.
+fn annotation_family_of_keys<'a>(
+    mut keys: impl Iterator<Item = &'a str>,
+) -> Result<Option<AnnotationFamily>> {
+    let Some(first) = keys.next() else {
+        return Ok(None);
+    };
+    let family = AnnotationFamily::of_key(first);
+    let mut count = 1usize;
+    for key in keys {
+        count += 1;
+        let this = AnnotationFamily::of_key(key);
+        if this != family {
+            return error::InvalidTableOptionRequestSnafu {
+                err_msg: family.or(this).unwrap().mixed_batch_error(),
+            }
+            .fail();
+        }
+    }
+    if let Some(family) = family
+        && family.requires_single_key()
+        && count > 1
+    {
+        return error::InvalidTableOptionRequestSnafu {
+            err_msg: family.mixed_batch_error(),
+        }
+        .fail();
+    }
+    Ok(family)
+}
+
+/// Returns the annotation family when `kind` is a SET/UNSET whose keys all
+/// belong to one family — the alters that only rewrite table metadata and skip
+/// region dispatch. A mixed batch is an error; never interpret it as "not an
+/// annotation alter", or the batch falls through to a path that reports a
+/// misleading error (or dispatches to regions).
+pub fn annotation_alter_family(kind: &Kind) -> Result<Option<AnnotationFamily>> {
+    match kind {
+        Kind::SetTableOptions(api::v1::SetTableOptions { table_options }) => {
+            annotation_family_of_keys(table_options.iter().map(|option| option.key.as_str()))
+        }
+        Kind::UnsetTableOptions(api::v1::UnsetTableOptions { keys }) => {
+            annotation_family_of_keys(keys.iter().map(|key| key.as_str()))
+        }
+        _ => Ok(None),
+    }
+}
+
 fn set_index_option_from_proto(set_index: api::v1::SetIndex) -> Result<SetIndexOption> {
     let options = set_index.options.context(MissingAlterIndexOptionSnafu)?;
     Ok(match options {
@@ -164,20 +215,15 @@ pub fn alter_expr_to_request(
             AlterKind::RenameTable { new_table_name }
         }
         Kind::SetTableOptions(api::v1::SetTableOptions { table_options }) => {
-            let repartition_column_hint = table_options
-                .iter()
-                .find(|option| option.key == REPARTITION_COLUMN_HINT_KEY);
-            if let Some(option) = repartition_column_hint {
-                ensure!(
-                    table_options.len() == 1,
-                    error::InvalidTableOptionRequestSnafu {
-                        err_msg: format!(
-                            "{REPARTITION_COLUMN_HINT_KEY} must be altered separately"
-                        ),
-                    }
-                );
-                AlterKind::SetRepartitionColumnHint {
-                    column_name: option.value.clone(),
+            if let Some(family) = annotation_family_of_keys(
+                table_options.iter().map(|option| option.key.as_str()),
+            )? {
+                AlterKind::SetAnnotations {
+                    family,
+                    options: table_options
+                        .into_iter()
+                        .map(|option| (option.key, option.value))
+                        .collect(),
                 }
             } else {
                 AlterKind::SetTableOptions {
@@ -190,19 +236,8 @@ pub fn alter_expr_to_request(
             }
         }
         Kind::UnsetTableOptions(api::v1::UnsetTableOptions { keys }) => {
-            let unset_repartition_column_hint = keys
-                .iter()
-                .any(|key| key.as_str() == REPARTITION_COLUMN_HINT_KEY);
-            if unset_repartition_column_hint {
-                ensure!(
-                    keys.len() == 1,
-                    error::InvalidTableOptionRequestSnafu {
-                        err_msg: format!(
-                            "{REPARTITION_COLUMN_HINT_KEY} must be altered separately"
-                        ),
-                    }
-                );
-                AlterKind::UnsetRepartitionColumnHint
+            if let Some(family) = annotation_family_of_keys(keys.iter().map(|key| key.as_str()))? {
+                AlterKind::UnsetAnnotations { family, keys }
             } else {
                 AlterKind::UnsetTableOptions {
                     keys: keys
@@ -359,6 +394,7 @@ mod tests {
         Option as PbOption, SemanticType, SetTableOptions, UnsetTableOptions,
     };
     use datatypes::prelude::ConcreteDataType;
+    use table::requests::REPARTITION_COLUMN_HINT_KEY;
 
     use super::*;
 
@@ -563,8 +599,12 @@ mod tests {
 
         let alter_request = alter_expr_to_request(1, expr, None).unwrap();
         match alter_request.alter_kind {
-            AlterKind::SetRepartitionColumnHint { column_name } => {
-                assert_eq!("host", column_name);
+            AlterKind::SetAnnotations { family, options } => {
+                assert_eq!(AnnotationFamily::RepartitionHint, family);
+                assert_eq!(
+                    vec![(REPARTITION_COLUMN_HINT_KEY.to_string(), "host".to_string())],
+                    options
+                );
             }
             _ => unreachable!(),
         }
@@ -595,6 +635,30 @@ mod tests {
             err.to_string()
                 .contains("repartition.column.hint must be altered separately")
         );
+
+        // Duplicate hint entries are not a meaningful batch either.
+        let dup = AlterTableExpr {
+            catalog_name: "test_catalog".to_string(),
+            schema_name: "test_schema".to_string(),
+            table_name: "monitor".to_string(),
+            kind: Some(Kind::SetTableOptions(SetTableOptions {
+                table_options: vec![
+                    PbOption {
+                        key: REPARTITION_COLUMN_HINT_KEY.to_string(),
+                        value: "host".to_string(),
+                    },
+                    PbOption {
+                        key: REPARTITION_COLUMN_HINT_KEY.to_string(),
+                        value: "region".to_string(),
+                    },
+                ],
+            })),
+        };
+        let err = alter_expr_to_request(1, dup, None).unwrap_err();
+        assert!(
+            err.to_string()
+                .contains("repartition.column.hint must be altered separately")
+        );
     }
 
     #[test]
@@ -608,10 +672,138 @@ mod tests {
             })),
         };
 
+        let alter_request = alter_expr_to_request(1, expr, None).unwrap();
+        match alter_request.alter_kind {
+            AlterKind::UnsetAnnotations { family, keys } => {
+                assert_eq!(AnnotationFamily::RepartitionHint, family);
+                assert_eq!(vec![REPARTITION_COLUMN_HINT_KEY.to_string()], keys);
+            }
+            _ => unreachable!(),
+        }
+    }
+
+    #[test]
+    fn test_semantic_options_classified_as_annotations() {
+        let expr = AlterTableExpr {
+            catalog_name: "test_catalog".to_string(),
+            schema_name: "test_schema".to_string(),
+            table_name: "monitor".to_string(),
+            kind: Some(Kind::SetTableOptions(SetTableOptions {
+                table_options: vec![
+                    PbOption {
+                        key: "greptime.semantic.signal_type".to_string(),
+                        value: "trace".to_string(),
+                    },
+                    PbOption {
+                        key: "greptime.semantic.entity.host.id".to_string(),
+                        value: "host".to_string(),
+                    },
+                ],
+            })),
+        };
+
+        let alter_request = alter_expr_to_request(1, expr, None).unwrap();
+        let AlterKind::SetAnnotations { family, options } = alter_request.alter_kind else {
+            panic!(
+                "expected SetAnnotations, got {:?}",
+                alter_request.alter_kind
+            );
+        };
+        assert_eq!(family, AnnotationFamily::Semantic);
+        assert_eq!(options.len(), 2);
+
+        let expr = AlterTableExpr {
+            catalog_name: "test_catalog".to_string(),
+            schema_name: "test_schema".to_string(),
+            table_name: "monitor".to_string(),
+            kind: Some(Kind::UnsetTableOptions(UnsetTableOptions {
+                keys: vec!["greptime.semantic.signal_type".to_string()],
+            })),
+        };
         let alter_request = alter_expr_to_request(1, expr, None).unwrap();
         assert!(matches!(
             alter_request.alter_kind,
-            AlterKind::UnsetRepartitionColumnHint
+            AlterKind::UnsetAnnotations {
+                family: AnnotationFamily::Semantic,
+                ..
+            }
         ));
     }
+
+    #[test]
+    fn test_semantic_options_reject_mixed_batch() {
+        let mixed_set = AlterTableExpr {
+            catalog_name: "test_catalog".to_string(),
+            schema_name: "test_schema".to_string(),
+            table_name: "monitor".to_string(),
+            kind: Some(Kind::SetTableOptions(SetTableOptions {
+                table_options: vec![
+                    PbOption {
+                        key: "greptime.semantic.signal_type".to_string(),
+                        value: "trace".to_string(),
+                    },
+                    PbOption {
+                        key: table::requests::TTL_KEY.to_string(),
+                        value: "7d".to_string(),
+                    },
+                ],
+            })),
+        };
+        let err = alter_expr_to_request(1, mixed_set, None).unwrap_err();
+        assert!(err.to_string().contains("altered separately"), "{err}");
+
+        let mixed_unset = AlterTableExpr {
+            catalog_name: "test_catalog".to_string(),
+            schema_name: "test_schema".to_string(),
+            table_name: "monitor".to_string(),
+            kind: Some(Kind::UnsetTableOptions(UnsetTableOptions {
+                keys: vec![
+                    "ttl".to_string(),
+                    "greptime.semantic.signal_type".to_string(),
+                ],
+            })),
+        };
+        let err = alter_expr_to_request(1, mixed_unset, None).unwrap_err();
+        assert!(err.to_string().contains("altered separately"), "{err}");
+    }
+
+    #[test]
+    fn test_annotation_alter_family() {
+        let mixed = Kind::SetTableOptions(SetTableOptions {
+            table_options: vec![
+                PbOption {
+                    key: "greptime.semantic.signal_type".to_string(),
+                    value: "trace".to_string(),
+                },
+                PbOption {
+                    key: "ttl".to_string(),
+                    value: "7d".to_string(),
+                },
+            ],
+        });
+        let err = annotation_alter_family(&mixed).unwrap_err();
+        assert!(
+            err.to_string().contains("must be altered separately"),
+            "{err}"
+        );
+
+        // Two annotation families cannot share a batch either.
+        let cross = Kind::SetTableOptions(SetTableOptions {
+            table_options: vec![
+                PbOption {
+                    key: "greptime.semantic.signal_type".to_string(),
+                    value: "trace".to_string(),
+                },
+                PbOption {
+                    key: REPARTITION_COLUMN_HINT_KEY.to_string(),
+                    value: "host".to_string(),
+                },
+            ],
+        });
+        let err = annotation_alter_family(&cross).unwrap_err();
+        assert!(
+            err.to_string().contains("must be altered separately"),
+            "{err}"
+        );
+    }
 }

src/common/grpc-expr/src/lib.rs

@@ -16,4 +16,4 @@ mod alter;
 pub mod error;
 pub mod util;
 
-pub use alter::{alter_expr_to_request, create_table_schema};
+pub use alter::{alter_expr_to_request, annotation_alter_family, create_table_schema};

src/common/meta/AGENTS.md

@@ -26,6 +26,11 @@ and durable DDL procedures. Metasrv process and service wiring live in
   invalidation.
 - Procedure changes must preserve persisted state and `TYPE_NAME`; new
   procedures need loader registration in `src/common/meta/src/ddl_manager.rs`.
+- Marker-style table options that no region consumes (`greptime.semantic.*`,
+  `repartition.column.hint`) go through `AnnotationFamily` in
+  `src/table/src/requests.rs` — classification, validation, logical-table
+  eligibility, and the metadata-only alter flow come with it. Do not add
+  per-option special cases to the alter conversion or DDL routing.
 - `KvBackend` behavior changes should extend the shared tests in
   `src/common/meta/src/kv_backend/test.rs`.
 

src/common/meta/src/ddl/alter_logical_tables.rs

@@ -24,7 +24,7 @@ use common_procedure::{Context, EventContext, EventTrigger, LockKey, Procedure,
 use common_telemetry::{debug, error, info, warn};
 pub use executor::make_alter_region_request;
 use serde::{Deserialize, Serialize};
-use snafu::ResultExt;
+use snafu::{ResultExt, ensure};
 use store_api::metadata::ColumnMetadata;
 use store_api::metric_engine_consts::ALTER_PHYSICAL_EXTENSION_KEY;
 use strum::AsRefStr;
@@ -40,7 +40,7 @@ use crate::ddl::event::table::{
     TableDdlEvent, TableDdlEventType, TableDdlLocator, alter_table_kind_name,
 };
 use crate::ddl::utils::{extract_column_metadatas, map_to_procedure_error, sync_follower_regions};
-use crate::error::Result;
+use crate::error::{self, Result};
 use crate::instruction::CacheIdent;
 use crate::key::DeserializedValueWithBytes;
 use crate::key::table_info::TableInfoValue;
@@ -88,6 +88,7 @@ impl AlterLogicalTablesProcedure {
     pub fn new(
         tasks: Vec<AlterTableTask>,
         physical_table_id: TableId,
+        logical_table_ids: Vec<TableId>,
         context: DdlContext,
     ) -> Self {
         Self {
@@ -97,6 +98,7 @@ impl AlterLogicalTablesProcedure {
                 tasks,
                 table_info_values: vec![],
                 physical_table_id,
+                logical_table_ids,
                 physical_table_info: None,
                 physical_columns: vec![],
                 table_cache_keys_to_invalidate: vec![],
@@ -147,6 +149,26 @@ impl AlterLogicalTablesProcedure {
             );
         }
 
+        // Locks are fixed at submission. A logical table that resolves to an
+        // id outside the locked set here was dropped and recreated in between,
+        // so this procedure holds no lock for it. Procedures restored from
+        // pre-upgrade state have no recorded ids and keep the old behavior.
+        if !self.data.logical_table_ids.is_empty() {
+            for value in &table_info_values {
+                let table_id = value.get_inner_ref().table_info.ident.table_id;
+                ensure!(
+                    self.data.logical_table_ids.contains(&table_id),
+                    error::UnexpectedSnafu {
+                        err_msg: format!(
+                            "logical table {} (id {table_id}) is not covered by the \
+                             procedure locks; retry the statement",
+                            value.get_inner_ref().table_info.name
+                        ),
+                    }
+                );
+            }
+        }
+
         // Updates the procedure state.
         retain_unskipped(&mut self.data.tasks, &skip_alter);
         self.data.physical_table_info = Some(physical_table_info);
@@ -305,17 +327,29 @@ impl Procedure for AlterLogicalTablesProcedure {
         // CatalogLock, SchemaLock,
         // TableLock
         // TableNameLock(s)
-        let mut lock_key = Vec::with_capacity(2 + 1 + self.data.tasks.len());
+        let mut lock_key = Vec::with_capacity(2 + 1 + self.data.logical_table_ids.len());
         let table_ref = self.data.tasks[0].table_ref();
         lock_key.push(CatalogLock::Read(table_ref.catalog).into());
         lock_key.push(SchemaLock::read(table_ref.catalog, table_ref.schema).into());
         lock_key.push(TableLock::Write(self.data.physical_table_id).into());
-        lock_key.extend(
-            self.data
-                .table_info_values
-                .iter()
-                .map(|table| TableLock::Write(table.table_info.ident.table_id).into()),
-        );
+        if self.data.logical_table_ids.is_empty() {
+            // Pre-upgrade procedure state has no `logical_table_ids`, but a
+            // dump taken after `Prepare` still carries the resolved table
+            // snapshots — recover the logical locks from them.
+            lock_key.extend(
+                self.data
+                    .table_info_values
+                    .iter()
+                    .map(|table| TableLock::Write(table.table_info.ident.table_id).into()),
+            );
+        } else {
+            lock_key.extend(
+                self.data
+                    .logical_table_ids
+                    .iter()
+                    .map(|table_id| TableLock::Write(*table_id).into()),
+            );
+        }
 
         LockKey::new(lock_key)
     }
@@ -366,6 +400,11 @@ pub struct AlterTablesData {
     table_info_values: Vec<DeserializedValueWithBytes<TableInfoValue>>,
     /// Physical table info
     physical_table_id: TableId,
+    /// Logical table ids resolved at submission time, so `lock_key` can name
+    /// them before `Prepare` runs (procedure locks are fixed at submission).
+    /// Empty when restored from pre-upgrade procedure state.
+    #[serde(default)]
+    logical_table_ids: Vec<TableId>,
     physical_table_info: Option<DeserializedValueWithBytes<TableInfoValue>>,
     physical_columns: Vec<ColumnMetadata>,
     table_cache_keys_to_invalidate: Vec<CacheIdent>,

src/common/meta/src/ddl/alter_table.rs

@@ -21,7 +21,7 @@ use std::vec;
 
 use api::region::RegionResponse;
 use api::v1::alter_table_expr::Kind;
-use api::v1::{RenameTable, SetTableOptions, UnsetTableOptions};
+use api::v1::{RenameTable, SetTableOptions};
 use async_trait::async_trait;
 use common_catalog::consts::{METRIC_ENGINE, MITO_ENGINE};
 use common_error::ext::BoxedError;
@@ -38,7 +38,7 @@ use store_api::metric_engine_consts::TABLE_COLUMN_METADATA_EXTENSION_KEY;
 use store_api::storage::RegionId;
 use strum::AsRefStr;
 use table::metadata::{TableId, TableInfo};
-use table::requests::{REPARTITION_COLUMN_HINT_KEY, SKIP_WAL_KEY};
+use table::requests::SKIP_WAL_KEY;
 use table::table_reference::TableReference;
 
 use crate::ddl::DdlContext;
@@ -51,7 +51,8 @@ use crate::ddl::utils::{
     sync_follower_regions,
 };
 use crate::error::{
-    AbortProcedureSnafu, NoLeaderSnafu, PutPoisonSnafu, Result, RetryLaterSnafu, UnsupportedSnafu,
+    AbortProcedureSnafu, ConvertAlterTableRequestSnafu, NoLeaderSnafu, PutPoisonSnafu, Result,
+    RetryLaterSnafu, UnsupportedSnafu,
 };
 use crate::key::table_info::TableInfoValue;
 use crate::key::{DeserializedValueWithBytes, RegionDistribution};
@@ -201,7 +202,7 @@ impl AlterTableProcedure {
             self.data.region_distribution =
                 Some(region_distribution(&physical_table_route.region_routes));
         }
-        self.data.state = self.data.flow().after_prepare();
+        self.data.state = self.data.flow()?.after_prepare();
         Ok(Status::executing(true))
     }
 
@@ -249,7 +250,7 @@ impl AlterTableProcedure {
         ensure!(!leaders.is_empty(), NoLeaderSnafu { table_id });
         // Puts the poison before submitting alter region requests to datanodes.
         self.put_poison(ctx_provider, procedure_id).await?;
-        let flow = self.data.flow();
+        let flow = self.data.flow()?;
         if flow == AlterTableFlow::MetadataFirst {
             let results = self
                 .executor
@@ -340,7 +341,7 @@ impl AlterTableProcedure {
                 "altering table result doesn't contains extension key `{TABLE_COLUMN_METADATA_EXTENSION_KEY}`,leaving the table's column metadata unchanged"
             );
         }
-        self.data.state = self.data.flow().after_regions();
+        self.data.state = self.data.flow()?.after_regions();
         Ok(())
     }
 
@@ -372,7 +373,7 @@ impl AlterTableProcedure {
         let table_info_value = self.data.table_info_value.as_ref().unwrap();
         // Safety: Checked in `AlterTableProcedure::new`.
         let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
-        let flow = self.data.flow();
+        let flow = self.data.flow()?;
         let metadata_only_alter = flow == AlterTableFlow::MetadataOnly;
 
         // Gets the table info from the cache or builds it.
@@ -469,17 +470,12 @@ pub(crate) fn only_enables_skip_wal(alter_kind: &Kind) -> bool {
         && table_options[0].value == "true"
 }
 
-fn is_metadata_only_alter(alter_kind: &Kind) -> bool {
-    match alter_kind {
-        Kind::RenameTable { .. } => true,
-        Kind::SetTableOptions(SetTableOptions { table_options }) => {
-            table_options.len() == 1 && table_options[0].key.as_str() == REPARTITION_COLUMN_HINT_KEY
-        }
-        Kind::UnsetTableOptions(UnsetTableOptions { keys }) => {
-            keys.len() == 1 && keys[0].as_str() == REPARTITION_COLUMN_HINT_KEY
-        }
-        _ => false,
-    }
+fn is_metadata_only_alter(alter_kind: &Kind) -> Result<bool> {
+    // A mixed annotation batch is an error, never "not metadata-only": falling
+    // through to the region-first flow would dispatch it to regions.
+    let family = common_grpc_expr::annotation_alter_family(alter_kind)
+        .context(ConvertAlterTableRequestSnafu)?;
+    Ok(family.is_some() || matches!(alter_kind, Kind::RenameTable { .. }))
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -490,14 +486,14 @@ enum AlterTableFlow {
 }
 
 impl AlterTableFlow {
-    fn from_kind(kind: &Kind) -> Self {
-        if only_enables_skip_wal(kind) {
+    fn from_kind(kind: &Kind) -> Result<Self> {
+        Ok(if only_enables_skip_wal(kind) {
             Self::MetadataFirst
-        } else if is_metadata_only_alter(kind) {
+        } else if is_metadata_only_alter(kind)? {
             Self::MetadataOnly
         } else {
             Self::RegionFirst
-        }
+        })
     }
 
     fn after_prepare(self) -> AlterTableState {
@@ -647,7 +643,7 @@ impl AlterTableData {
             .map(|value| &value.table_info)
     }
 
-    fn flow(&self) -> AlterTableFlow {
+    fn flow(&self) -> Result<AlterTableFlow> {
         // Safety: Checked in `AlterTableProcedure::new`.
         AlterTableFlow::from_kind(self.task.alter_table.kind.as_ref().unwrap())
     }

src/common/meta/src/ddl/alter_table/executor.rs

@@ -335,8 +335,8 @@ fn build_new_table_info(
         | AlterKind::ModifyColumnTypes { .. }
         | AlterKind::SetTableOptions { .. }
         | AlterKind::UnsetTableOptions { .. }
-        | AlterKind::SetRepartitionColumnHint { .. }
-        | AlterKind::UnsetRepartitionColumnHint
+        | AlterKind::SetAnnotations { .. }
+        | AlterKind::UnsetAnnotations { .. }
         | AlterKind::SetIndexes { .. }
         | AlterKind::UnsetIndexes { .. }
         | AlterKind::DropDefaults { .. }

src/common/meta/src/ddl/tests/alter_logical_tables.rs

@@ -43,6 +43,7 @@ use crate::error::Error::{AlterLogicalTablesInvalidArguments, TableNotFound};
 use crate::error::Result;
 use crate::key::table_name::TableNameKey;
 use crate::key::table_route::{PhysicalTableRouteValue, TableRouteValue};
+use crate::lock_key::TableLock;
 use crate::rpc::ddl::AlterTableTask;
 use crate::rpc::router::{Region, RegionRoute};
 use crate::test_util::{MockDatanodeManager, new_ddl_context};
@@ -162,7 +163,8 @@ async fn test_on_prepare_check_schema() {
         ),
     ];
     let physical_table_id = 1024u32;
-    let mut procedure = AlterLogicalTablesProcedure::new(tasks, physical_table_id, ddl_context);
+    let mut procedure =
+        AlterLogicalTablesProcedure::new(tasks, physical_table_id, vec![], ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, AlterLogicalTablesInvalidArguments { .. });
 }
@@ -177,7 +179,8 @@ async fn test_on_prepare_check_alter_kind() {
         "new_table1",
     )];
     let physical_table_id = 1024u32;
-    let mut procedure = AlterLogicalTablesProcedure::new(tasks, physical_table_id, ddl_context);
+    let mut procedure =
+        AlterLogicalTablesProcedure::new(tasks, physical_table_id, vec![], ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, AlterLogicalTablesInvalidArguments { .. });
 }
@@ -197,7 +200,7 @@ async fn test_on_prepare_different_physical_table() {
         make_alter_logical_table_add_column_task(None, "table2", vec!["column2".to_string()]),
     ];
 
-    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy1_id, ddl_context);
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy1_id, vec![], ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, AlterLogicalTablesInvalidArguments { .. });
 }
@@ -218,7 +221,7 @@ async fn test_on_prepare_logical_table_not_exists() {
         make_alter_logical_table_add_column_task(None, "table2", vec!["column2".to_string()]),
     ];
 
-    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context);
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, vec![], ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, TableNotFound { .. });
 }
@@ -241,7 +244,7 @@ async fn test_on_prepare() {
         make_alter_logical_table_add_column_task(None, "table3", vec!["column3".to_string()]),
     ];
 
-    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context);
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, vec![], ddl_context);
     let result = procedure.on_prepare().await;
     assert_matches!(
         result,
@@ -277,7 +280,8 @@ async fn test_on_update_metadata() {
         make_alter_logical_table_add_column_task(None, "table3", vec!["new_col".to_string()]),
     ];
 
-    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context.clone());
+    let mut procedure =
+        AlterLogicalTablesProcedure::new(tasks, phy_id, vec![], ddl_context.clone());
     let mut status = procedure.on_prepare().await.unwrap();
     assert_matches!(
         status,
@@ -362,7 +366,8 @@ async fn test_on_part_duplicate_alter_request() {
         make_alter_logical_table_add_column_task(None, "table2", vec!["col_0".to_string()]),
     ];
 
-    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context.clone());
+    let mut procedure =
+        AlterLogicalTablesProcedure::new(tasks, phy_id, vec![], ddl_context.clone());
     let mut status = procedure.on_prepare().await.unwrap();
     assert_matches!(
         status,
@@ -448,7 +453,8 @@ async fn test_on_part_duplicate_alter_request() {
         ),
     ];
 
-    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context.clone());
+    let mut procedure =
+        AlterLogicalTablesProcedure::new(tasks, phy_id, vec![], ddl_context.clone());
     let mut status = procedure.on_prepare().await.unwrap();
     assert_matches!(
         status,
@@ -620,7 +626,7 @@ async fn test_on_submit_alter_region_request() {
         make_alter_logical_table_add_column_task(None, "table2", vec!["mew_col".to_string()]),
     ];
 
-    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context);
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, vec![], ddl_context);
     procedure.on_prepare().await.unwrap();
     procedure.on_submit_alter_region_requests().await.unwrap();
     let mut results = Vec::new();
@@ -646,3 +652,82 @@ async fn test_on_submit_alter_region_request() {
         })
     );
 }
+
+#[test]
+fn test_lock_key_covers_submission_resolved_logical_tables() {
+    let node_manager = Arc::new(MockDatanodeManager::new(()));
+    let ddl_context = new_ddl_context(node_manager);
+    let tasks = vec![make_alter_logical_table_add_column_task(
+        None,
+        "table1",
+        vec!["column1".to_string()],
+    )];
+    let procedure = AlterLogicalTablesProcedure::new(tasks, 1024, vec![1025, 1026], ddl_context);
+    let keys = procedure.lock_key();
+    let keys = keys.keys_to_lock().collect::<Vec<_>>();
+    for table_id in [1024u32, 1025, 1026] {
+        let expected: common_procedure::StringKey = TableLock::Write(table_id).into();
+        assert!(keys.contains(&&expected), "missing lock for {table_id}");
+    }
+}
+
+#[tokio::test]
+async fn test_from_json_without_logical_table_ids() {
+    let node_manager = Arc::new(MockDatanodeManager::new(()));
+    let ddl_context = new_ddl_context(node_manager);
+    let phy_id = create_physical_table(&ddl_context, "phy").await;
+    let logical_id = create_logical_table(ddl_context.clone(), phy_id, "table1").await;
+    let tasks = vec![make_alter_logical_table_add_column_task(
+        None,
+        "table1",
+        vec!["column1".to_string()],
+    )];
+    let mut procedure =
+        AlterLogicalTablesProcedure::new(tasks, phy_id, vec![logical_id], ddl_context.clone());
+    procedure.on_prepare().await.unwrap();
+    let json = procedure.dump().unwrap();
+
+    // Pre-upgrade procedure state carries no `logical_table_ids`, but a
+    // post-Prepare dump still holds the resolved table snapshots.
+    let mut value: serde_json::Value = serde_json::from_str(&json).unwrap();
+    value
+        .as_object_mut()
+        .unwrap()
+        .remove("logical_table_ids")
+        .unwrap();
+    let restored = AlterLogicalTablesProcedure::from_json(&value.to_string(), ddl_context).unwrap();
+
+    let keys = restored.lock_key();
+    let keys = keys.keys_to_lock().collect::<Vec<_>>();
+    for table_id in [phy_id, logical_id] {
+        let expected: common_procedure::StringKey = TableLock::Write(table_id).into();
+        assert!(keys.contains(&&expected), "missing lock for {table_id}");
+    }
+}
+
+#[tokio::test]
+async fn test_on_prepare_rejects_table_outside_locked_set() {
+    let node_manager = Arc::new(MockDatanodeManager::new(()));
+    let ddl_context = new_ddl_context(node_manager);
+
+    let phy_id = create_physical_table(&ddl_context, "phy").await;
+    create_logical_table(ddl_context.clone(), phy_id, "table1").await;
+
+    let tasks = vec![make_alter_logical_table_add_column_task(
+        None,
+        "table1",
+        vec!["column1".to_string()],
+    )];
+
+    // The table resolved at prepare time is not in the locked set, as if it
+    // had been dropped and recreated after submission.
+    let stale_locked_id = phy_id + 1000;
+    let mut procedure =
+        AlterLogicalTablesProcedure::new(tasks, phy_id, vec![stale_locked_id], ddl_context);
+    let err = procedure.on_prepare().await.unwrap_err();
+    assert!(
+        err.to_string()
+            .contains("not covered by the procedure locks"),
+        "{err}"
+    );
+}

src/common/meta/src/ddl/tests/alter_table.rs

@@ -29,7 +29,7 @@ use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
 use common_procedure::store::poison_store::PoisonStore;
 use common_procedure::{Procedure, ProcedureId, Status};
-use common_procedure_test::MockContextProvider;
+use common_procedure_test::{MockContextProvider, execute_procedure_until_done};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::ColumnSchema;
 use store_api::metadata::ColumnMetadata;
@@ -1121,3 +1121,60 @@ async fn test_on_submit_alter_request_with_exist_poison() {
         .unwrap_err();
     assert_matches!(err, Error::PutPoison { .. });
 }
+
+#[tokio::test]
+async fn test_semantic_annotation_alter_is_metadata_only() {
+    let (tx, mut rx) = mpsc::channel(8);
+    let node_manager = Arc::new(MockDatanodeManager::new(DatanodeWatcher::new(tx)));
+    let ddl_context = new_ddl_context(node_manager);
+    let table_id = 1024;
+    let table_name = "foo";
+    let task = test_create_table_task(table_name, table_id);
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(
+            task.table_info.clone(),
+            prepare_table_route(table_id),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    let alter_table_task = AlterTableTask {
+        alter_table: AlterTableExpr {
+            catalog_name: DEFAULT_CATALOG_NAME.to_string(),
+            schema_name: DEFAULT_SCHEMA_NAME.to_string(),
+            table_name: table_name.to_string(),
+            kind: Some(Kind::SetTableOptions(SetTableOptions {
+                table_options: vec![api::v1::Option {
+                    key: "greptime.semantic.signal_type".to_string(),
+                    value: "metric".to_string(),
+                }],
+            })),
+        },
+    };
+    let mut procedure =
+        AlterTableProcedure::new(table_id, alter_table_task, ddl_context.clone()).unwrap();
+    execute_procedure_until_done(&mut procedure).await;
+
+    // Metadata-only: no region request reaches any datanode.
+    rx.try_recv().unwrap_err();
+
+    let table_info = ddl_context
+        .table_metadata_manager
+        .table_info_manager()
+        .get(table_id)
+        .await
+        .unwrap()
+        .unwrap()
+        .into_inner()
+        .table_info;
+    assert_eq!(
+        table_info
+            .meta
+            .options
+            .extra_options
+            .get("greptime.semantic.signal_type"),
+        Some(&"metric".to_string())
+    );
+}

src/common/meta/src/ddl/tests/event/table.rs

@@ -391,6 +391,7 @@ fn procedure_cases() -> Vec<ProcedureCase> {
             ),
         ],
         43,
+        vec![],
         test_context(),
     );
     let drop_table = DropTableProcedure::new(

src/common/meta/src/ddl_manager.rs

@@ -53,7 +53,7 @@ use crate::ddl::truncate_table::TruncateTableProcedure;
 use crate::ddl::undrop_table::UndropTableProcedure;
 use crate::ddl::{DdlContext, utils};
 use crate::error::{
-    self, CreateRepartitionProcedureSnafu, EmptyDdlTasksSnafu,
+    self, ConvertAlterTableRequestSnafu, CreateRepartitionProcedureSnafu, EmptyDdlTasksSnafu,
     PersistRepartitionGcRequirementSnafu, ProcedureOutputSnafu, RegisterProcedureLoaderSnafu,
     RegisterRepartitionProcedureLoaderSnafu, Result, SubmitProcedureSnafu, TableInfoNotFoundSnafu,
     TableNotFoundSnafu, TableRouteNotFoundSnafu, UnexpectedLogicalRouteTableSnafu,
@@ -554,8 +554,26 @@ impl DdlManager {
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
 
-        let procedure =
-            AlterLogicalTablesProcedure::new(alter_table_tasks, physical_table_id, context);
+        // Resolve the logical table ids up front: procedure locks are fixed
+        // at submission, so `lock_key` cannot derive them during `Prepare`.
+        let logical_table_ids = {
+            let table_refs = alter_table_tasks
+                .iter()
+                .map(|task| task.table_ref())
+                .collect::<Vec<_>>();
+            utils::table_id::get_all_table_ids_by_names(
+                self.table_metadata_manager().table_name_manager(),
+                &table_refs,
+            )
+            .await?
+        };
+
+        let procedure = AlterLogicalTablesProcedure::new(
+            alter_table_tasks,
+            physical_table_id,
+            logical_table_ids,
+            context,
+        );
 
         let procedure_with_id =
             ProcedureWithId::with_random_id(Box::new(procedure)).with_context(procedure_context);
@@ -1053,8 +1071,17 @@ async fn handle_alter_table_task(
         .get(table_id)
         .await?
         .context(TableRouteNotFoundSnafu { table_id })?;
+    // Classify before the route guard: a mixed annotation batch must surface
+    // its own error here, not a misleading "non-physical route" one. Families
+    // that only rewrite the logical table's metadata may target logical tables.
+    let annotation_family = match alter_table_task.alter_table.kind.as_ref() {
+        Some(kind) => common_grpc_expr::annotation_alter_family(kind)
+            .context(ConvertAlterTableRequestSnafu)?,
+        None => None,
+    };
     ensure!(
-        table_route_value.is_physical(),
+        table_route_value.is_physical()
+            || annotation_family.is_some_and(|family| family.allows_logical_tables()),
         UnexpectedLogicalRouteTableSnafu {
             err_msg: format!("{:?} is a non-physical TableRouteValue.", table_ref),
         }
@@ -1499,7 +1526,9 @@ mod tests {
     use common_event_recorder::{PersistentEventContext, ProcedureEventInput, TriggerReason};
     use common_procedure::local::LocalManager;
     use common_procedure::test_util::InMemoryPoisonStore;
-    use common_procedure::{BoxedProcedure, ProcedureContext, ProcedureManagerRef};
+    use common_procedure::{
+        BoxedProcedure, ProcedureContext, ProcedureManager, ProcedureManagerRef,
+    };
     use store_api::storage::TableId;
     use table::table_name::TableName;
 
@@ -1900,4 +1929,127 @@ mod tests {
             assert!(matches!(err, crate::error::Error::Unsupported { .. }));
         }
     }
+
+    async fn ddl_manager_with_context(ddl_context: DdlContext) -> DdlManager {
+        let kv_backend = Arc::new(MemoryKvBackend::new());
+        let state_store = Arc::new(KvStateStore::new(kv_backend.clone()));
+        let poison_manager = Arc::new(InMemoryPoisonStore::default());
+        let procedure_manager = Arc::new(LocalManager::new(
+            Default::default(),
+            state_store,
+            poison_manager,
+            None,
+            None,
+        ));
+        procedure_manager.start().await.unwrap();
+        let ddl_manager = DdlManager::new(
+            ddl_context,
+            procedure_manager,
+            Arc::new(DummyRepartitionProcedureFactory),
+        );
+        ddl_manager.register_loaders().unwrap();
+        ddl_manager
+    }
+
+    fn set_options_expr(table_name: &str, options: &[(&str, &str)]) -> api::v1::AlterTableExpr {
+        api::v1::AlterTableExpr {
+            catalog_name: common_catalog::consts::DEFAULT_CATALOG_NAME.to_string(),
+            schema_name: common_catalog::consts::DEFAULT_SCHEMA_NAME.to_string(),
+            table_name: table_name.to_string(),
+            kind: Some(api::v1::alter_table_expr::Kind::SetTableOptions(
+                api::v1::SetTableOptions {
+                    table_options: options
+                        .iter()
+                        .map(|(key, value)| api::v1::Option {
+                            key: key.to_string(),
+                            value: value.to_string(),
+                        })
+                        .collect(),
+                },
+            )),
+        }
+    }
+
+    #[tokio::test]
+    async fn test_logical_table_annotation_alter_routing() {
+        let (tx, mut rx) = tokio::sync::mpsc::channel(8);
+        let node_manager = Arc::new(crate::test_util::MockDatanodeManager::new(
+            crate::ddl::test_util::datanode_handler::DatanodeWatcher::new(tx),
+        ));
+        let ddl_context = crate::test_util::new_ddl_context(node_manager);
+        let phy_id = crate::ddl::test_util::create_physical_table(&ddl_context, "phy").await;
+        let logical_id =
+            crate::ddl::test_util::create_logical_table(ddl_context.clone(), phy_id, "logical")
+                .await;
+        let ddl_manager = ddl_manager_with_context(ddl_context.clone()).await;
+
+        // A semantic alter on a logical table passes the route guard, updates
+        // only the logical table's metadata, and dispatches nothing.
+        ddl_manager
+            .submit_ddl_task(
+                ExecutorContext {
+                    query_context: Some(QueryContext::default()),
+                    ..Default::default()
+                },
+                SubmitDdlTaskRequest::new(DdlTask::new_alter_table(set_options_expr(
+                    "logical",
+                    &[("greptime.semantic.signal_type", "metric")],
+                ))),
+            )
+            .await
+            .unwrap();
+        rx.try_recv().unwrap_err();
+        let table_info = ddl_manager
+            .table_metadata_manager()
+            .table_info_manager()
+            .get(logical_id)
+            .await
+            .unwrap()
+            .unwrap()
+            .into_inner()
+            .table_info;
+        assert_eq!(
+            table_info
+                .meta
+                .options
+                .extra_options
+                .get("greptime.semantic.signal_type"),
+            Some(&"metric".to_string())
+        );
+
+        // A mixed batch fails with its own error, not the route guard's.
+        let err = ddl_manager
+            .submit_ddl_task(
+                ExecutorContext {
+                    query_context: Some(QueryContext::default()),
+                    ..Default::default()
+                },
+                SubmitDdlTaskRequest::new(DdlTask::new_alter_table(set_options_expr(
+                    "logical",
+                    &[("greptime.semantic.source", "prometheus"), ("ttl", "7d")],
+                ))),
+            )
+            .await
+            .unwrap_err();
+        let msg = common_error::ext::ErrorExt::output_msg(&err);
+        assert!(msg.contains("must be altered separately"), "{msg}");
+
+        // The repartition hint drives physical repartitioning; on a logical
+        // route it stays rejected by the guard.
+        let err = ddl_manager
+            .submit_ddl_task(
+                ExecutorContext {
+                    query_context: Some(QueryContext::default()),
+                    ..Default::default()
+                },
+                SubmitDdlTaskRequest::new(DdlTask::new_alter_table(set_options_expr(
+                    "logical",
+                    &[("repartition.column.hint", "host")],
+                ))),
+            )
+            .await
+            .unwrap_err();
+        let msg = common_error::ext::ErrorExt::output_msg(&err);
+        assert!(msg.contains("non-physical TableRouteValue"), "{msg}");
+    }
 }

src/operator/src/statement/ddl.rs

@@ -98,8 +98,8 @@ use table::TableRef;
 use table::dist_table::DistTable;
 use table::metadata::{self, TableId, TableInfo, TableMeta, TableType};
 use table::requests::{
-    AlterKind, AlterTableRequest, COMMENT_KEY, DDL_TIMEOUT, DDL_WAIT, REPARTITION_COLUMN_HINT_KEY,
-    TableOptions, parse_entity_columns, parse_entity_option_key,
+    AlterKind, AlterTableRequest, AnnotationContext, COMMENT_KEY, DDL_TIMEOUT, DDL_WAIT,
+    TableOptions, validate_and_normalize_annotation_options,
 };
 use table::table_name::TableName;
 use table::table_reference::TableReference;
@@ -1881,8 +1881,20 @@ impl StatementExecutor {
 
             (req, invalidate_keys)
         } else {
-            // This is logical table
-            let req = SubmitDdlTaskRequest::new(DdlTask::new_alter_logical_tables(vec![expr]));
+            // This is logical table. Annotation alters only rewrite its own
+            // metadata; `AlterLogicalTablesProcedure` only handles column adds.
+            let annotation_alter = match expr.kind.as_ref() {
+                Some(kind) => common_grpc_expr::annotation_alter_family(kind)
+                    .context(AlterExprToRequestSnafu)?
+                    .is_some_and(|family| family.allows_logical_tables()),
+                None => false,
+            };
+            let task = if annotation_alter {
+                DdlTask::new_alter_table(expr)
+            } else {
+                DdlTask::new_alter_logical_tables(vec![expr])
+            };
+            let req = SubmitDdlTaskRequest::new(task);
 
             let mut invalidate_keys = vec![
                 CacheIdent::TableId(physical_table_id),
@@ -2338,19 +2350,6 @@ pub fn verify_alter(
         .context(error::BuildTableMetaSnafu { table_name })?;
 
     validate_json2_columns_append_mode(&new_meta.schema, &new_meta.options)?;
-    // A column referenced by an entity declaration may be dropped (the
-    // read-time derivation skips the stale declaration), but must not change
-    // to a type without a stable string form.
-    for (key, value) in &new_meta.options.extra_options {
-        if parse_entity_option_key(key).is_none() {
-            continue;
-        }
-        for col in &parse_entity_columns(value) {
-            if let Some(column) = new_meta.schema.column_schema_by_name(col) {
-                ensure_entity_column_renders_as_string(key, col, &column.data_type)?;
-            }
-        }
-    }
 
     Ok(true)
 }
@@ -2402,14 +2401,7 @@ pub fn create_table_info(
 
     validate_json2_columns_append_mode(&schema, &table_options)?;
 
-    validate_repartition_column_hint(
-        &mut table_options,
-        &column_name_to_index_map,
-        &partition_key_indices,
-        &create_table.time_index,
-    )?;
-
-    validate_entity_semantic_options(&table_options, &schema)?;
+    validate_and_normalize_annotations(&mut table_options, &schema, &partition_key_indices)?;
 
     let meta = TableMeta {
         schema,
@@ -2469,61 +2461,6 @@ fn validate_json2_columns_append_mode(schema: &Schema, table_options: &TableOpti
     Ok(())
 }
 
-fn validate_repartition_column_hint(
-    table_options: &mut TableOptions,
-    column_name_to_index_map: &HashMap<String, usize>,
-    partition_key_indices: &[usize],
-    time_index: &str,
-) -> Result<()> {
-    let Some(column_name) = table_options
-        .extra_options
-        .get(REPARTITION_COLUMN_HINT_KEY)
-        .map(|value| value.trim().to_string())
-    else {
-        return Ok(());
-    };
-
-    ensure!(
-        !column_name.is_empty(),
-        InvalidPartitionRuleSnafu {
-            reason: format!("{REPARTITION_COLUMN_HINT_KEY} expects exactly one column name"),
-        }
-    );
-
-    ensure!(
-        !column_name.contains(','),
-        InvalidPartitionRuleSnafu {
-            reason: format!("{REPARTITION_COLUMN_HINT_KEY} expects exactly one column name"),
-        }
-    );
-
-    ensure!(
-        partition_key_indices.is_empty(),
-        InvalidPartitionRuleSnafu {
-            reason: format!(
-                "cannot set {REPARTITION_COLUMN_HINT_KEY} on a table with partition metadata"
-            ),
-        }
-    );
-
-    column_name_to_index_map
-        .get(&column_name)
-        .context(ColumnNotFoundSnafu { msg: &column_name })?;
-
-    ensure!(
-        column_name != time_index,
-        InvalidPartitionRuleSnafu {
-            reason: format!("cannot set {REPARTITION_COLUMN_HINT_KEY} to the time index column"),
-        }
-    );
-
-    table_options
-        .extra_options
-        .insert(REPARTITION_COLUMN_HINT_KEY.to_string(), column_name);
-
-    Ok(())
-}
-
 /// Rejects DDL against read-only schemas and computed entity-graph tables.
 fn ensure_table_writable(schema: &str, table: &str) -> Result<()> {
     ensure!(
@@ -2557,57 +2494,34 @@ fn ensure_table_definition_writable(schema: &str, table: &str) -> Result<()> {
     Ok(())
 }
 
-/// Whether `CAST(column AS Utf8)` yields a stable entity-id string — the
-/// binary-backed and nested types do not, and without the DDL check the
-/// failure would surface only when the graph is scanned.
-fn has_stable_string_form(data_type: &datatypes::prelude::ConcreteDataType) -> bool {
-    use datatypes::prelude::ConcreteDataType;
-    !matches!(
-        data_type,
-        ConcreteDataType::Binary(_)
-            | ConcreteDataType::Json(_)
-            | ConcreteDataType::Vector(_)
-            | ConcreteDataType::List(_)
-            | ConcreteDataType::Struct(_)
-            | ConcreteDataType::Dictionary(_)
-            | ConcreteDataType::Null(_)
-    )
-}
-
-fn ensure_entity_column_renders_as_string(
-    key: &str,
-    col: &str,
-    data_type: &datatypes::prelude::ConcreteDataType,
+/// CREATE-side annotation validation: one rule source in the table crate,
+/// mapped onto this crate's existing error variants so client-visible codes
+/// and messages stay put.
+fn validate_and_normalize_annotations(
+    options: &mut TableOptions,
+    schema: &Schema,
+    partition_key_indices: &[usize],
 ) -> Result<()> {
-    ensure!(
-        has_stable_string_form(data_type),
-        InvalidSqlSnafu {
-            err_msg: format!(
-                "entity column `{col}` (option `{key}`) has type `{data_type}`, which cannot \
-                 render as a string",
-            ),
+    use table::requests::AnnotationValidationError as CheckError;
+    let cx = AnnotationContext {
+        schema,
+        partition_key_indices,
+    };
+    validate_and_normalize_annotation_options(options, &cx).map_err(|e| match e {
+        CheckError::ColumnNotFound { column } => ColumnNotFoundSnafu { msg: column }.build(),
+        e @ (CheckError::UnknownKey { .. }
+        | CheckError::InvalidValue { .. }
+        | CheckError::ColumnNotStringForm { .. }) => InvalidSqlSnafu {
+            err_msg: e.to_string(),
         }
-    );
-    Ok(())
-}
-
-/// Validates `greptime.semantic.entity.<type>.{id|descriptive|scope}` options
-/// against the table schema: every named column must exist (tag or field) and
-/// have a stable string form — the derivation renders id, scope and
-/// descriptive values as strings.
-fn validate_entity_semantic_options(table_options: &TableOptions, schema: &Schema) -> Result<()> {
-    for (key, value) in &table_options.extra_options {
-        if parse_entity_option_key(key).is_none() {
-            continue;
-        };
-        for col in &parse_entity_columns(value) {
-            let column = schema
-                .column_schema_by_name(col)
-                .context(ColumnNotFoundSnafu { msg: col })?;
-            ensure_entity_column_renders_as_string(key, col, &column.data_type)?;
+        .build(),
+        e @ (CheckError::NotSingleColumn
+        | CheckError::PartitionMetadataConflict
+        | CheckError::TimeIndexConflict) => InvalidPartitionRuleSnafu {
+            reason: e.to_string(),
         }
-    }
-    Ok(())
+        .build(),
+    })
 }
 
 fn find_partition_columns(partitions: &Option<Partitions>) -> Result<Vec<String>> {
@@ -2904,6 +2818,7 @@ mod test {
     use sql::parser::{ParseOptions, ParserContext};
     use sql::statements::statement::Statement;
     use sqlparser::parser::Parser;
+    use table::requests::REPARTITION_COLUMN_HINT_KEY;
 
     use super::*;
     use crate::expr_helper;
@@ -3133,7 +3048,7 @@ mod test {
     }
 
     #[test]
-    fn test_validate_entity_semantic_options() {
+    fn test_validate_and_normalize_annotations() {
         let schema = Schema::new(vec![
             ColumnSchema::new("service_name", ConcreteDataType::string_datatype(), true),
             ColumnSchema::new("host_id", ConcreteDataType::string_datatype(), true),
@@ -3147,6 +3062,10 @@ mod test {
                 .collect(),
             ..Default::default()
         };
+        let check = |pairs: &[(&str, &str)]| {
+            let mut options = opts(pairs);
+            validate_and_normalize_annotations(&mut options, &schema, &[]).map(|()| options)
+        };
 
         // Any existing column with a string form may be an id, tag or field.
         for pairs in [
@@ -3156,22 +3075,27 @@ mod test {
             )][..],
             &[("greptime.semantic.entity.service.id", "value")][..],
         ] {
-            assert!(validate_entity_semantic_options(&opts(pairs), &schema).is_ok());
+            assert!(check(pairs).is_ok());
         }
 
-        let missing = validate_entity_semantic_options(
-            &opts(&[("greptime.semantic.entity.service.id", "nope")]),
-            &schema,
-        )
-        .unwrap_err();
+        // Missing columns keep this crate's error variant and status code.
+        let missing = check(&[("greptime.semantic.entity.service.id", "nope")]).unwrap_err();
         assert!(matches!(missing, error::Error::ColumnNotFound { .. }));
+        assert_eq!(
+            common_error::status_code::StatusCode::InvalidArguments,
+            common_error::ext::ErrorExt::status_code(&missing)
+        );
 
-        let binary_id = validate_entity_semantic_options(
-            &opts(&[("greptime.semantic.entity.service.id", "payload")]),
-            &schema,
-        )
-        .unwrap_err();
+        let binary_id = check(&[("greptime.semantic.entity.service.id", "payload")]).unwrap_err();
         assert!(matches!(binary_id, error::Error::InvalidSql { .. }));
+
+        // The SQL parser checks keys and value domains, but gRPC expressions
+        // bypass it.
+        let bad_value = check(&[("greptime.semantic.signal_type", "garbage")]).unwrap_err();
+        assert!(matches!(bad_value, error::Error::InvalidSql { .. }));
+
+        let unknown_key = check(&[("greptime.semantic.nonsense", "x")]).unwrap_err();
+        assert!(matches!(unknown_key, error::Error::InvalidSql { .. }));
     }
 
     #[test]
@@ -3559,9 +3483,10 @@ WITH ('repartition.column.hint' = ' host ')",
             }],
         }));
         let err = verify_alter(1, table_info.clone(), modify).unwrap_err();
+        let msg = common_error::ext::ErrorExt::output_msg(&err);
         assert!(
-            err.to_string().contains("cannot render as a string"),
-            "{err}"
+            msg.contains("must keep a type that renders as a string"),
+            "{msg}"
         );
 
         // Dropping a declared column stays allowed: the derivation skips the

src/table/src/metadata.rs

@@ -37,8 +37,10 @@ use store_api::storage::{ColumnDescriptor, ColumnDescriptorBuilder, ColumnId};
 
 use crate::error::{self, Result};
 use crate::requests::{
-    AddColumnRequest, AlterKind, ModifyColumnTypeRequest, REPARTITION_COLUMN_HINT_KEY,
-    SetDefaultRequest, SetIndexOption, TableOptions, UnsetIndexOption,
+    AddColumnRequest, AlterKind, AnnotationContext, AnnotationFamily, AnnotationValidationError,
+    ModifyColumnTypeRequest, REPARTITION_COLUMN_HINT_KEY, SetDefaultRequest, SetIndexOption,
+    TableOptions, UnsetIndexOption, has_stable_string_form, parse_entity_columns,
+    parse_entity_option_key, validate_and_normalize_annotation,
 };
 use crate::table_reference::TableReference;
 
@@ -331,10 +333,12 @@ impl TableMeta {
             AlterKind::RenameTable { .. } => Ok(self.new_meta_builder()),
             AlterKind::SetTableOptions { options } => self.set_table_options(options),
             AlterKind::UnsetTableOptions { keys } => self.unset_table_options(keys),
-            AlterKind::SetRepartitionColumnHint { column_name } => {
-                self.set_repartition_column_hint(table_name, column_name)
+            AlterKind::SetAnnotations { family, options } => {
+                self.set_annotations(table_name, *family, options)
+            }
+            AlterKind::UnsetAnnotations { family, keys } => {
+                self.unset_annotations(table_name, *family, keys)
             }
-            AlterKind::UnsetRepartitionColumnHint => self.unset_repartition_column_hint(),
             AlterKind::SetIndexes { options } => self.set_indexes(table_name, options),
             AlterKind::UnsetIndexes { options } => self.unset_indexes(table_name, options),
             AlterKind::DropDefaults { names } => self.drop_defaults(table_name, names),
@@ -422,67 +426,92 @@ impl TableMeta {
         self.set_table_options(&requests)
     }
 
-    fn set_repartition_column_hint(
+    /// Applies an annotation SET. Validation lives here, on the mutation
+    /// path, so it runs both at frontend verification and again inside the
+    /// alter procedure's prepare step — under the table lock, against fresh
+    /// metadata.
+    fn set_annotations(
         &self,
         table_name: &str,
-        column_name: &str,
+        family: AnnotationFamily,
+        options: &[(String, String)],
     ) -> Result<TableMetaBuilder> {
-        let column_name = column_name.trim();
         ensure!(
-            !column_name.is_empty() && !column_name.contains(','),
+            !family.requires_single_key() || options.len() == 1,
             error::InvalidAlterRequestSnafu {
                 table: table_name,
-                err: format!("{REPARTITION_COLUMN_HINT_KEY} expects exactly one column name"),
+                err: family.mixed_batch_error(),
             }
         );
-
-        ensure!(
-            self.partition_key_indices.is_empty(),
-            error::InvalidAlterRequestSnafu {
-                table: table_name,
-                err: format!(
-                    "cannot set {REPARTITION_COLUMN_HINT_KEY} on a table with partition metadata"
-                ),
-            }
-        );
-
-        let column_index = self
-            .schema
-            .column_index_by_name(column_name)
-            .with_context(|| error::ColumnNotExistsSnafu {
-                column_name,
-                table_name,
-            })?;
-
-        if let Some(time_index) = self.schema.timestamp_index() {
+        let cx = AnnotationContext {
+            schema: &self.schema,
+            partition_key_indices: &self.partition_key_indices,
+        };
+        let mut new_options = self.options.clone();
+        for (key, value) in options {
             ensure!(
-                column_index != time_index,
+                AnnotationFamily::of_key(key) == Some(family),
                 error::InvalidAlterRequestSnafu {
                     table: table_name,
                     err: format!(
-                        "cannot set {REPARTITION_COLUMN_HINT_KEY} to the time index column"
+                        "`{key}` is outside the `{}` annotation namespace",
+                        family.namespace()
                     ),
                 }
             );
+            let checked = validate_and_normalize_annotation(family, &cx, key, value).map_err(
+                |e| match e {
+                    AnnotationValidationError::ColumnNotFound { column } => {
+                        error::ColumnNotExistsSnafu {
+                            column_name: column,
+                            table_name,
+                        }
+                        .build()
+                    }
+                    other => error::InvalidAlterRequestSnafu {
+                        table: table_name,
+                        err: other.to_string(),
+                    }
+                    .build(),
+                },
+            )?;
+            new_options.extra_options.insert(key.clone(), checked);
         }
-
-        let mut new_options = self.options.clone();
-        new_options.extra_options.insert(
-            REPARTITION_COLUMN_HINT_KEY.to_string(),
-            column_name.to_string(),
-        );
-
         let mut builder = self.new_meta_builder();
         builder.options(new_options);
         Ok(builder)
     }
 
-    fn unset_repartition_column_hint(&self) -> Result<TableMetaBuilder> {
+    /// Applies an annotation UNSET. Deliberately lenient inside the family's
+    /// namespace: keys this version does not recognise may still be removed,
+    /// so options left behind by other versions can be cleaned up.
+    fn unset_annotations(
+        &self,
+        table_name: &str,
+        family: AnnotationFamily,
+        keys: &[String],
+    ) -> Result<TableMetaBuilder> {
+        ensure!(
+            !family.requires_single_key() || keys.len() == 1,
+            error::InvalidAlterRequestSnafu {
+                table: table_name,
+                err: family.mixed_batch_error(),
+            }
+        );
         let mut new_options = self.options.clone();
-        new_options
-            .extra_options
-            .remove(REPARTITION_COLUMN_HINT_KEY);
-
+        for key in keys {
+            ensure!(
+                AnnotationFamily::of_key(key) == Some(family),
+                error::InvalidAlterRequestSnafu {
+                    table: table_name,
+                    err: format!(
+                        "`{key}` is outside the `{}` annotation namespace",
+                        family.namespace()
+                    ),
+                }
+            );
+            new_options.extra_options.remove(key);
+        }
         let mut builder = self.new_meta_builder();
         builder.options(new_options);
         Ok(builder)
@@ -773,6 +802,24 @@ impl TableMeta {
                     },
                 );
 
+                // A dropped column may leave a stale entity declaration behind;
+                // re-adding it must not hand the declaration a type without a
+                // stable string form.
+                if !has_stable_string_form(&col_to_add.column_schema.data_type)
+                    && let Some(key) =
+                        entity_option_referencing(&self.options, &col_to_add.column_schema.name)
+                {
+                    return error::InvalidAlterRequestSnafu {
+                        table: table_name,
+                        err: format!(
+                            "column `{}` is referenced by entity option `{key}` and must \
+                             keep a type that renders as a string, got `{}`",
+                            col_to_add.column_schema.name, col_to_add.column_schema.data_type
+                        ),
+                    }
+                    .fail();
+                }
+
                 new_columns.push(col_to_add.clone());
             }
         }
@@ -966,6 +1013,25 @@ impl TableMeta {
                     table_name,
                 })?;
 
+            // A column referenced by an entity declaration may be dropped (the
+            // read-time derivation skips the stale declaration), but must not
+            // change to a type without a stable string form. Checked after the
+            // existence lookup so a missing column keeps reporting
+            // `ColumnNotExists` regardless of stale declarations.
+            if !has_stable_string_form(&col_to_change.target_type)
+                && let Some(key) = entity_option_referencing(&self.options, change_column_name)
+            {
+                return error::InvalidAlterRequestSnafu {
+                    table: table_name,
+                    err: format!(
+                        "column `{change_column_name}` is referenced by entity option \
+                         `{key}` and must keep a type that renders as a string, got `{}`",
+                        col_to_change.target_type
+                    ),
+                }
+                .fail();
+            }
+
             let column = &table_schema.column_schemas()[index];
 
             ensure!(
@@ -1408,6 +1474,14 @@ impl TableInfo {
     }
 }
 
+fn entity_option_referencing<'a>(options: &'a TableOptions, column: &str) -> Option<&'a str> {
+    options.extra_options.iter().find_map(|(key, value)| {
+        (parse_entity_option_key(key).is_some()
+            && parse_entity_columns(value).iter().any(|c| c == column))
+        .then_some(key.as_str())
+    })
+}
+
 /// Set column fulltext options if it passed the validation.
 ///
 /// Options allowed to modify:
@@ -1738,8 +1812,12 @@ mod tests {
             .build()
             .unwrap();
 
-        let alter_kind = AlterKind::SetRepartitionColumnHint {
-            column_name: " col1 ".to_string(),
+        let alter_kind = AlterKind::SetAnnotations {
+            family: AnnotationFamily::RepartitionHint,
+            options: vec![(
+                REPARTITION_COLUMN_HINT_KEY.to_string(),
+                " col1 ".to_string(),
+            )],
         };
         let new_meta = meta
             .builder_with_alter_kind("my_table", &alter_kind)
@@ -1767,8 +1845,9 @@ mod tests {
             .build()
             .unwrap();
 
-        let alter_kind = AlterKind::SetRepartitionColumnHint {
-            column_name: " ".to_string(),
+        let alter_kind = AlterKind::SetAnnotations {
+            family: AnnotationFamily::RepartitionHint,
+            options: vec![(REPARTITION_COLUMN_HINT_KEY.to_string(), " ".to_string())],
         };
         let err = meta
             .builder_with_alter_kind("my_table", &alter_kind)
@@ -1791,8 +1870,12 @@ mod tests {
             .build()
             .unwrap();
 
-        let alter_kind = AlterKind::SetRepartitionColumnHint {
-            column_name: "col1,col2".to_string(),
+        let alter_kind = AlterKind::SetAnnotations {
+            family: AnnotationFamily::RepartitionHint,
+            options: vec![(
+                REPARTITION_COLUMN_HINT_KEY.to_string(),
+                "col1,col2".to_string(),
+            )],
         };
         let err = meta
             .builder_with_alter_kind("my_table", &alter_kind)
@@ -1815,8 +1898,12 @@ mod tests {
             .build()
             .unwrap();
 
-        let alter_kind = AlterKind::SetRepartitionColumnHint {
-            column_name: "missing".to_string(),
+        let alter_kind = AlterKind::SetAnnotations {
+            family: AnnotationFamily::RepartitionHint,
+            options: vec![(
+                REPARTITION_COLUMN_HINT_KEY.to_string(),
+                "missing".to_string(),
+            )],
         };
         let err = meta
             .builder_with_alter_kind("my_table", &alter_kind)
@@ -1836,8 +1923,9 @@ mod tests {
             .build()
             .unwrap();
 
-        let alter_kind = AlterKind::SetRepartitionColumnHint {
-            column_name: "ts".to_string(),
+        let alter_kind = AlterKind::SetAnnotations {
+            family: AnnotationFamily::RepartitionHint,
+            options: vec![(REPARTITION_COLUMN_HINT_KEY.to_string(), "ts".to_string())],
         };
         let err = meta
             .builder_with_alter_kind("my_table", &alter_kind)
@@ -1861,8 +1949,9 @@ mod tests {
             .build()
             .unwrap();
 
-        let alter_kind = AlterKind::SetRepartitionColumnHint {
-            column_name: "col1".to_string(),
+        let alter_kind = AlterKind::SetAnnotations {
+            family: AnnotationFamily::RepartitionHint,
+            options: vec![(REPARTITION_COLUMN_HINT_KEY.to_string(), "col1".to_string())],
         };
         let err = meta
             .builder_with_alter_kind("my_table", &alter_kind)
@@ -1891,7 +1980,13 @@ mod tests {
             .unwrap();
 
         let new_meta = meta
-            .builder_with_alter_kind("my_table", &AlterKind::UnsetRepartitionColumnHint)
+            .builder_with_alter_kind(
+                "my_table",
+                &AlterKind::UnsetAnnotations {
+                    family: AnnotationFamily::RepartitionHint,
+                    keys: vec![REPARTITION_COLUMN_HINT_KEY.to_string()],
+                },
+            )
             .unwrap()
             .build()
             .unwrap();
@@ -2627,4 +2722,305 @@ mod tests {
         };
         assert_eq!(actual, expected);
     }
+
+    use crate::requests::{SEMANTIC_METRIC_UNIT, SEMANTIC_SIGNAL_TYPE};
+
+    /// `host` is a tag, `service` and `note` are string fields, `payload` is a
+    /// binary field.
+    fn semantic_test_meta() -> TableMeta {
+        let column_schemas = vec![
+            ColumnSchema::new("host", ConcreteDataType::string_datatype(), true),
+            ColumnSchema::new(
+                "ts",
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            )
+            .with_time_index(true),
+            ColumnSchema::new("payload", ConcreteDataType::binary_datatype(), true),
+            ColumnSchema::new("service", ConcreteDataType::string_datatype(), true),
+            ColumnSchema::new("note", ConcreteDataType::string_datatype(), true),
+        ];
+        let schema = Arc::new(
+            SchemaBuilder::try_from(column_schemas)
+                .unwrap()
+                .build()
+                .unwrap(),
+        );
+        TableMetaBuilder::empty()
+            .schema(schema)
+            .primary_key_indices(vec![0])
+            .engine("engine")
+            .next_column_id(5)
+            .build()
+            .unwrap()
+    }
+
+    #[test]
+    fn test_set_semantic_annotations() {
+        let meta = semantic_test_meta();
+        // `service` is a plain field: entity columns may be tags or fields.
+        let alter_kind = AlterKind::SetAnnotations {
+            family: AnnotationFamily::Semantic,
+            options: vec![
+                (SEMANTIC_SIGNAL_TYPE.to_string(), "trace".to_string()),
+                (
+                    "greptime.semantic.entity.service.id".to_string(),
+                    "service".to_string(),
+                ),
+            ],
+        };
+        let new_meta = meta
+            .builder_with_alter_kind("my_table", &alter_kind)
+            .unwrap()
+            .build()
+            .unwrap();
+        assert_eq!(
+            new_meta.options.extra_options.get(SEMANTIC_SIGNAL_TYPE),
+            Some(&"trace".to_string())
+        );
+        assert_eq!(
+            new_meta
+                .options
+                .extra_options
+                .get("greptime.semantic.entity.service.id"),
+            Some(&"service".to_string())
+        );
+    }
+
+    #[test]
+    fn test_repartition_hint_batch_must_be_single_key() {
+        let meta = TableMetaBuilder::empty()
+            .schema(Arc::new(new_test_schema()))
+            .primary_key_indices(vec![0])
+            .engine("engine")
+            .next_column_id(3)
+            .build()
+            .unwrap();
+
+        // Direct gRPC can hand the mutation layer a duplicated batch the
+        // converter never saw; last-write-wins must not silently apply.
+        let dup_set = AlterKind::SetAnnotations {
+            family: AnnotationFamily::RepartitionHint,
+            options: vec![
+                (REPARTITION_COLUMN_HINT_KEY.to_string(), "col1".to_string()),
+                (REPARTITION_COLUMN_HINT_KEY.to_string(), "col2".to_string()),
+            ],
+        };
+        let err = meta
+            .builder_with_alter_kind("my_table", &dup_set)
+            .err()
+            .unwrap();
+        assert!(
+            err.to_string().contains("must be altered separately"),
+            "{err}"
+        );
+
+        let dup_unset = AlterKind::UnsetAnnotations {
+            family: AnnotationFamily::RepartitionHint,
+            keys: vec![
+                REPARTITION_COLUMN_HINT_KEY.to_string(),
+                REPARTITION_COLUMN_HINT_KEY.to_string(),
+            ],
+        };
+        let err = meta
+            .builder_with_alter_kind("my_table", &dup_unset)
+            .err()
+            .unwrap();
+        assert!(
+            err.to_string().contains("must be altered separately"),
+            "{err}"
+        );
+    }
+
+    #[test]
+    fn test_set_semantic_annotations_rejects_invalid() {
+        let meta = semantic_test_meta();
+        let cases = [
+            (
+                "greptime.semantic.unknown_key",
+                "x",
+                "unknown semantic option",
+            ),
+            (SEMANTIC_SIGNAL_TYPE, "garbage", "invalid value"),
+            (
+                "greptime.semantic.entity.host.id",
+                "no_such_column",
+                "no_such_column",
+            ),
+            (
+                "greptime.semantic.entity.host.id",
+                "payload",
+                "cannot render as a string",
+            ),
+        ];
+        for (key, value, needle) in cases {
+            let alter_kind = AlterKind::SetAnnotations {
+                family: AnnotationFamily::Semantic,
+                options: vec![(key.to_string(), value.to_string())],
+            };
+            let err = meta
+                .builder_with_alter_kind("my_table", &alter_kind)
+                .err()
+                .unwrap();
+            assert!(
+                err.to_string().contains(needle),
+                "key `{key}`: unexpected error `{err}`"
+            );
+        }
+
+        // ALTER keeps missing columns on the 4002 contract (pinned by sqlness);
+        // the shared validator must not collapse it into InvalidArguments.
+        let missing = meta
+            .builder_with_alter_kind(
+                "my_table",
+                &AlterKind::SetAnnotations {
+                    family: AnnotationFamily::Semantic,
+                    options: vec![(
+                        "greptime.semantic.entity.host.id".to_string(),
+                        "no_such_column".to_string(),
+                    )],
+                },
+            )
+            .err()
+            .unwrap();
+        assert_eq!(
+            common_error::status_code::StatusCode::TableColumnNotFound,
+            common_error::ext::ErrorExt::status_code(&missing)
+        );
+    }
+
+    #[test]
+    fn test_unset_semantic_annotations() {
+        let mut meta = semantic_test_meta();
+        meta.options
+            .extra_options
+            .insert(SEMANTIC_SIGNAL_TYPE.to_string(), "trace".to_string());
+        // A key this version does not recognise, still inside the namespace.
+        meta.options
+            .extra_options
+            .insert("greptime.semantic.future_key".to_string(), "x".to_string());
+
+        let alter_kind = AlterKind::UnsetAnnotations {
+            family: AnnotationFamily::Semantic,
+            keys: vec![
+                SEMANTIC_SIGNAL_TYPE.to_string(),
+                "greptime.semantic.future_key".to_string(),
+                // Absent key: removal is a no-op, not an error.
+                SEMANTIC_METRIC_UNIT.to_string(),
+            ],
+        };
+        let new_meta = meta
+            .builder_with_alter_kind("my_table", &alter_kind)
+            .unwrap()
+            .build()
+            .unwrap();
+        assert!(
+            !new_meta
+                .options
+                .extra_options
+                .contains_key(SEMANTIC_SIGNAL_TYPE)
+        );
+        assert!(
+            !new_meta
+                .options
+                .extra_options
+                .contains_key("greptime.semantic.future_key")
+        );
+
+        let outside = AlterKind::UnsetAnnotations {
+            family: AnnotationFamily::Semantic,
+            keys: vec!["ttl".to_string()],
+        };
+        let err = meta
+            .builder_with_alter_kind("my_table", &outside)
+            .err()
+            .unwrap();
+        assert!(err.to_string().contains("annotation namespace"), "{err}");
+    }
+
+    #[test]
+    fn test_modify_entity_column_type_keeps_string_form() {
+        let mut meta = semantic_test_meta();
+        meta.options.extra_options.insert(
+            "greptime.semantic.entity.service.id".to_string(),
+            "service".to_string(),
+        );
+
+        let alter_kind = AlterKind::ModifyColumnTypes {
+            columns: vec![ModifyColumnTypeRequest {
+                column_name: "service".to_string(),
+                target_type: ConcreteDataType::binary_datatype(),
+            }],
+        };
+        let err = meta
+            .builder_with_alter_kind("my_table", &alter_kind)
+            .err()
+            .unwrap();
+        assert!(
+            err.to_string()
+                .contains("must keep a type that renders as a string"),
+            "{err}"
+        );
+
+        // An unreferenced column may still change to a non-string form.
+        let alter_kind = AlterKind::ModifyColumnTypes {
+            columns: vec![ModifyColumnTypeRequest {
+                column_name: "note".to_string(),
+                target_type: ConcreteDataType::binary_datatype(),
+            }],
+        };
+        meta.builder_with_alter_kind("my_table", &alter_kind)
+            .unwrap();
+    }
+
+    #[test]
+    fn test_stale_entity_declaration_guards_readd_and_reports_missing_column() {
+        // A stale declaration: `gone` was dropped after being declared.
+        let mut meta = semantic_test_meta();
+        meta.options.extra_options.insert(
+            "greptime.semantic.entity.service.id".to_string(),
+            "gone".to_string(),
+        );
+
+        // MODIFY on the missing column keeps the ColumnNotExists contract
+        // (4002), stale declaration or not.
+        let modify = AlterKind::ModifyColumnTypes {
+            columns: vec![ModifyColumnTypeRequest {
+                column_name: "gone".to_string(),
+                target_type: ConcreteDataType::binary_datatype(),
+            }],
+        };
+        let err = meta
+            .builder_with_alter_kind("my_table", &modify)
+            .err()
+            .unwrap();
+        assert_eq!(
+            common_error::status_code::StatusCode::TableColumnNotFound,
+            common_error::ext::ErrorExt::status_code(&err)
+        );
+
+        // Re-adding the declared column must not hand the declaration a
+        // non-string type...
+        let add = |ty: ConcreteDataType| AlterKind::AddColumns {
+            columns: vec![AddColumnRequest {
+                column_schema: ColumnSchema::new("gone", ty, true),
+                is_key: false,
+                location: None,
+                add_if_not_exists: false,
+            }],
+        };
+        let err = meta
+            .builder_with_alter_kind("my_table", &add(ConcreteDataType::binary_datatype()))
+            .err()
+            .unwrap();
+        assert!(
+            err.to_string()
+                .contains("must keep a type that renders as a string"),
+            "{err}"
+        );
+
+        // ...while a string re-add re-satisfies it.
+        meta.builder_with_alter_kind("my_table", &add(ConcreteDataType::string_datatype()))
+            .unwrap();
+    }
 }

src/table/src/requests.rs

@@ -27,7 +27,7 @@ use common_time::range::TimestampRange;
 use datatypes::data_type::ConcreteDataType;
 use datatypes::prelude::VectorRef;
 use datatypes::schema::{
-    ColumnDefaultConstraint, ColumnSchema, FulltextOptions, SkippingIndexOptions,
+    ColumnDefaultConstraint, ColumnSchema, FulltextOptions, Schema, SkippingIndexOptions,
 };
 use greptime_proto::v1::region::compact_request;
 use once_cell::sync::Lazy;
@@ -301,6 +301,211 @@ pub struct ModifyColumnTypeRequest {
     pub target_type: ConcreteDataType,
 }
 
+/// A family of annotation table options: pure metadata markers that no region
+/// consumes. Setting or unsetting them only rewrites the table's
+/// `extra_options`, so the alter skips region dispatch entirely.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub enum AnnotationFamily {
+    /// `greptime.semantic.*` options (see the [`semantic`] module).
+    Semantic,
+    /// `repartition.column.hint`, consumed by the auto-repartition planner.
+    RepartitionHint,
+}
+
+impl AnnotationFamily {
+    /// The key namespace: a prefix for [`Self::Semantic`], the exact key for
+    /// [`Self::RepartitionHint`].
+    pub fn namespace(self) -> &'static str {
+        match self {
+            Self::Semantic => SEMANTIC_PREFIX,
+            Self::RepartitionHint => REPARTITION_COLUMN_HINT_KEY,
+        }
+    }
+
+    pub fn of_key(key: &str) -> Option<Self> {
+        if key.starts_with(SEMANTIC_PREFIX) {
+            Some(Self::Semantic)
+        } else if key == REPARTITION_COLUMN_HINT_KEY {
+            Some(Self::RepartitionHint)
+        } else {
+            None
+        }
+    }
+
+    /// Whether this family may be altered on logical metric tables. Only
+    /// families whose values nothing on the physical side consumes qualify;
+    /// the repartition hint drives physical region repartitioning.
+    pub fn allows_logical_tables(self) -> bool {
+        match self {
+            Self::Semantic => true,
+            Self::RepartitionHint => false,
+        }
+    }
+
+    /// Whether this family's SET/UNSET batch must contain exactly one key.
+    /// The repartition hint is a single marker; a batch with several hint
+    /// entries (duplicates included) has no meaningful order.
+    pub fn requires_single_key(self) -> bool {
+        matches!(self, Self::RepartitionHint)
+    }
+
+    /// The error for a SET/UNSET batch mixing this family with other options.
+    pub fn mixed_batch_error(self) -> String {
+        match self {
+            Self::Semantic => format!(
+                "`{SEMANTIC_PREFIX}*` options must be altered separately from other table options"
+            ),
+            Self::RepartitionHint => {
+                format!("{REPARTITION_COLUMN_HINT_KEY} must be altered separately")
+            }
+        }
+    }
+}
+
+/// Table shape an annotation option is validated against.
+pub struct AnnotationContext<'a> {
+    pub schema: &'a Schema,
+    pub partition_key_indices: &'a [usize],
+}
+
+/// Why an annotation option was rejected. Typed so each DDL entry point maps
+/// rules onto its existing error variants and status codes: ALTER keeps
+/// missing columns as `TableColumnNotFound` (4002), CREATE keeps its
+/// `InvalidArguments` family — the rules converge, the contracts do not.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum AnnotationValidationError {
+    UnknownKey {
+        key: String,
+    },
+    InvalidValue {
+        key: String,
+        value: String,
+    },
+    ColumnNotFound {
+        column: String,
+    },
+    ColumnNotStringForm {
+        key: String,
+        column: String,
+        ty: ConcreteDataType,
+    },
+    NotSingleColumn,
+    PartitionMetadataConflict,
+    TimeIndexConflict,
+}
+
+impl fmt::Display for AnnotationValidationError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::UnknownKey { key } => write!(f, "unknown semantic option `{key}`"),
+            Self::InvalidValue { key, value } => {
+                write!(f, "invalid value `{value}` for semantic option `{key}`")
+            }
+            Self::ColumnNotFound { column } => write!(f, "column `{column}` not found"),
+            Self::ColumnNotStringForm { key, column, ty } => write!(
+                f,
+                "entity column `{column}` (option `{key}`) has type `{ty}`, \
+                 which cannot render as a string"
+            ),
+            Self::NotSingleColumn => write!(
+                f,
+                "{REPARTITION_COLUMN_HINT_KEY} expects exactly one column name"
+            ),
+            Self::PartitionMetadataConflict => write!(
+                f,
+                "cannot set {REPARTITION_COLUMN_HINT_KEY} on a table with partition metadata"
+            ),
+            Self::TimeIndexConflict => write!(
+                f,
+                "cannot set {REPARTITION_COLUMN_HINT_KEY} to the time index column"
+            ),
+        }
+    }
+}
+
+/// Validates one annotation option and returns the value to store — the
+/// repartition hint is trimmed to the bare column name, semantic values pass
+/// through unchanged.
+pub(crate) fn validate_and_normalize_annotation(
+    family: AnnotationFamily,
+    cx: &AnnotationContext<'_>,
+    key: &str,
+    value: &str,
+) -> std::result::Result<String, AnnotationValidationError> {
+    match family {
+        AnnotationFamily::Semantic => {
+            if !is_semantic_option_key(key) {
+                return Err(AnnotationValidationError::UnknownKey {
+                    key: key.to_string(),
+                });
+            }
+            if !validate_semantic_option(key, value) {
+                return Err(AnnotationValidationError::InvalidValue {
+                    key: key.to_string(),
+                    value: value.to_string(),
+                });
+            }
+            if parse_entity_option_key(key).is_some() {
+                for column in parse_entity_columns(value) {
+                    let schema = cx.schema.column_schema_by_name(&column).ok_or_else(|| {
+                        AnnotationValidationError::ColumnNotFound {
+                            column: column.clone(),
+                        }
+                    })?;
+                    if !has_stable_string_form(&schema.data_type) {
+                        return Err(AnnotationValidationError::ColumnNotStringForm {
+                            key: key.to_string(),
+                            column,
+                            ty: schema.data_type.clone(),
+                        });
+                    }
+                }
+            }
+            Ok(value.to_string())
+        }
+        AnnotationFamily::RepartitionHint => {
+            let column_name = value.trim();
+            if column_name.is_empty() || column_name.contains(',') {
+                return Err(AnnotationValidationError::NotSingleColumn);
+            }
+            if !cx.partition_key_indices.is_empty() {
+                return Err(AnnotationValidationError::PartitionMetadataConflict);
+            }
+            let column_index = cx.schema.column_index_by_name(column_name).ok_or_else(|| {
+                AnnotationValidationError::ColumnNotFound {
+                    column: column_name.to_string(),
+                }
+            })?;
+            if cx.schema.timestamp_index() == Some(column_index) {
+                return Err(AnnotationValidationError::TimeIndexConflict);
+            }
+            Ok(column_name.to_string())
+        }
+    }
+}
+
+/// CREATE-side entry: validates every annotation option present in `options`
+/// and writes normalized values back in place.
+pub fn validate_and_normalize_annotation_options(
+    options: &mut TableOptions,
+    cx: &AnnotationContext<'_>,
+) -> std::result::Result<(), AnnotationValidationError> {
+    let mut normalized = Vec::new();
+    for (key, value) in &options.extra_options {
+        let Some(family) = AnnotationFamily::of_key(key) else {
+            continue;
+        };
+        let checked = validate_and_normalize_annotation(family, cx, key, value)?;
+        if checked != *value {
+            normalized.push((key.clone(), checked));
+        }
+    }
+    for (key, value) in normalized {
+        options.extra_options.insert(key, value);
+    }
+    Ok(())
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum AlterKind {
     AddColumns {
@@ -321,10 +526,14 @@ pub enum AlterKind {
     UnsetTableOptions {
         keys: Vec<UnsetRegionOption>,
     },
-    SetRepartitionColumnHint {
-        column_name: String,
+    SetAnnotations {
+        family: AnnotationFamily,
+        options: Vec<(String, String)>,
+    },
+    UnsetAnnotations {
+        family: AnnotationFamily,
+        keys: Vec<String>,
     },
-    UnsetRepartitionColumnHint,
     SetIndexes {
         options: Vec<SetIndexOption>,
     },

src/table/src/requests/semantic.rs

@@ -30,6 +30,8 @@
 //! [`crate::requests::validate_table_option`], so they are accepted both on the
 //! ingestion auto-create path and on explicit `CREATE TABLE ... WITH (...)` DDL.
 
+use datatypes::prelude::ConcreteDataType;
+
 /// Reserved prefix for every public semantic table-option key.
 pub const SEMANTIC_PREFIX: &str = "greptime.semantic.";
 
@@ -90,8 +92,10 @@ pub const SEMANTIC_ENTITY_PREFIX: &str = "greptime.semantic.entity.";
 /// is the `service_name` tag column, declaring the logical `service` entity.
 pub const SEMANTIC_ENTITY_SERVICE_ID: &str = "greptime.semantic.entity.service.id";
 
-/// The role a set of columns plays for an entity: `id` (identifying attributes,
-/// must be tag columns — enforced at DDL time), `descriptive`, or `scope`.
+/// The role a set of columns plays for an entity: `id` (identifying
+/// attributes), `descriptive`, or `scope`. Columns may be tags or fields; DDL
+/// validation only requires that they exist and render as stable strings
+/// ([`has_stable_string_form`]).
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum EntityRole {
     Id,
@@ -182,6 +186,24 @@ pub fn is_entity_option_key(key: &str) -> bool {
     parse_entity_option_key(key).is_some()
 }
 
+/// Returns true if a column of `data_type` renders as a stable string — the
+/// requirement for entity id/descriptive/scope columns. The read-time
+/// derivation casts them to strings, so a type without a stable string form
+/// would fail only when the graph is scanned; DDL validation rejects it up
+/// front instead.
+pub fn has_stable_string_form(data_type: &ConcreteDataType) -> bool {
+    !matches!(
+        data_type,
+        ConcreteDataType::Binary(_)
+            | ConcreteDataType::Json(_)
+            | ConcreteDataType::Vector(_)
+            | ConcreteDataType::List(_)
+            | ConcreteDataType::Struct(_)
+            | ConcreteDataType::Dictionary(_)
+            | ConcreteDataType::Null(_)
+    )
+}
+
 /// Tokenizes an entity option's comma-separated column list (trimmed, empty
 /// tokens dropped). [`validate_semantic_option`] rejects empty tokens at DDL
 /// time, so readers only ever drop what validation already refused.

tests/cases/standalone/common/information_schema/table_semantics.result

@@ -86,3 +86,121 @@ DROP TABLE plain_table;
 
 Affected Rows: 0
 
+-- ALTER TABLE manages semantic declarations on existing tables: SET appears in
+-- the view, UNSET disappears from it.
+CREATE TABLE altered_semantics (
+  ts TIMESTAMP TIME INDEX,
+  svc STRING,
+  payload BINARY,
+  val DOUBLE,
+);
+
+Affected Rows: 0
+
+ALTER TABLE altered_semantics SET 'greptime.semantic.signal_type' = 'metric', 'greptime.semantic.entity.service.id' = 'svc';
+
+Affected Rows: 0
+
+SELECT table_name, signal_type, semantic_options
+FROM information_schema.table_semantics
+WHERE table_name = 'altered_semantics';
+
++-------------------+-------------+-----------------------------+
+| table_name        | signal_type | semantic_options            |
++-------------------+-------------+-----------------------------+
+| altered_semantics | metric      | {"entity.service.id":"svc"} |
++-------------------+-------------+-----------------------------+
+
+-- Semantic options never share a statement with regular table options.
+ALTER TABLE altered_semantics SET 'greptime.semantic.source' = 'prometheus', 'ttl' = '7d';
+
+Error: 1004(InvalidArguments), Invalid table option request: `greptime.semantic.*` options must be altered separately from other table options
+
+-- Unknown semantic keys are rejected on SET.
+ALTER TABLE altered_semantics SET 'greptime.semantic.nonsense' = 'x';
+
+Error: 1004(InvalidArguments), Invalid alter table(altered_semantics) request: unknown semantic option `greptime.semantic.nonsense`
+
+-- Values outside the key's domain are rejected.
+ALTER TABLE altered_semantics SET 'greptime.semantic.signal_type' = 'garbage';
+
+Error: 1004(InvalidArguments), Invalid alter table(altered_semantics) request: invalid value `garbage` for semantic option `greptime.semantic.signal_type`
+
+-- Entity columns must exist.
+ALTER TABLE altered_semantics SET 'greptime.semantic.entity.host.id' = 'no_such_column';
+
+Error: 4002(TableColumnNotFound), Column no_such_column not exists in table altered_semantics
+
+-- Entity columns must render as strings.
+ALTER TABLE altered_semantics SET 'greptime.semantic.entity.host.id' = 'payload';
+
+Error: 1004(InvalidArguments), Invalid alter table(altered_semantics) request: entity column `payload` (option `greptime.semantic.entity.host.id`) has type `Binary`, which cannot render as a string
+
+ALTER TABLE altered_semantics UNSET 'greptime.semantic.entity.service.id';
+
+Affected Rows: 0
+
+SELECT table_name, signal_type, semantic_options
+FROM information_schema.table_semantics
+WHERE table_name = 'altered_semantics';
+
++-------------------+-------------+------------------+
+| table_name        | signal_type | semantic_options |
++-------------------+-------------+------------------+
+| altered_semantics | metric      |                  |
++-------------------+-------------+------------------+
+
+DROP TABLE altered_semantics;
+
+Affected Rows: 0
+
+-- Logical metric tables take the same metadata-only path.
+CREATE TABLE phy_sem (ts TIMESTAMP TIME INDEX, val DOUBLE) engine=metric with ("physical_metric_table" = "");
+
+Affected Rows: 0
+
+CREATE TABLE logical_sem (ts TIMESTAMP TIME INDEX, val DOUBLE, host STRING PRIMARY KEY) engine=metric with ("on_physical_table" = "phy_sem");
+
+Affected Rows: 0
+
+ALTER TABLE logical_sem SET 'greptime.semantic.signal_type' = 'metric', 'greptime.semantic.entity.host.id' = 'host';
+
+Affected Rows: 0
+
+SELECT table_name, signal_type, semantic_options
+FROM information_schema.table_semantics
+WHERE table_name = 'logical_sem';
+
++-------------+-------------+---------------------------+
+| table_name  | signal_type | semantic_options          |
++-------------+-------------+---------------------------+
+| logical_sem | metric      | {"entity.host.id":"host"} |
++-------------+-------------+---------------------------+
+
+-- Regular options still cannot be altered on logical tables.
+ALTER TABLE logical_sem SET 'ttl' = '7d';
+
+Error: 1004(InvalidArguments), Alter logical tables invalid arguments: Only support add columns operation
+
+ALTER TABLE logical_sem UNSET 'greptime.semantic.entity.host.id';
+
+Affected Rows: 0
+
+SELECT table_name, signal_type, semantic_options
+FROM information_schema.table_semantics
+WHERE table_name = 'logical_sem';
+
++-------------+-------------+------------------+
+| table_name  | signal_type | semantic_options |
++-------------+-------------+------------------+
+| logical_sem | metric      |                  |
++-------------+-------------+------------------+
+
+DROP TABLE logical_sem;
+
+Affected Rows: 0
+
+DROP TABLE phy_sem;
+
+Affected Rows: 0
+

tests/cases/standalone/common/information_schema/table_semantics.sql

@@ -45,3 +45,65 @@ DROP TABLE metrics_tagged;
 DROP TABLE traces_tagged;
 
 DROP TABLE plain_table;
+
+-- ALTER TABLE manages semantic declarations on existing tables: SET appears in
+-- the view, UNSET disappears from it.
+CREATE TABLE altered_semantics (
+  ts TIMESTAMP TIME INDEX,
+  svc STRING,
+  payload BINARY,
+  val DOUBLE,
+);
+
+ALTER TABLE altered_semantics SET 'greptime.semantic.signal_type' = 'metric', 'greptime.semantic.entity.service.id' = 'svc';
+
+SELECT table_name, signal_type, semantic_options
+FROM information_schema.table_semantics
+WHERE table_name = 'altered_semantics';
+
+-- Semantic options never share a statement with regular table options.
+ALTER TABLE altered_semantics SET 'greptime.semantic.source' = 'prometheus', 'ttl' = '7d';
+
+-- Unknown semantic keys are rejected on SET.
+ALTER TABLE altered_semantics SET 'greptime.semantic.nonsense' = 'x';
+
+-- Values outside the key's domain are rejected.
+ALTER TABLE altered_semantics SET 'greptime.semantic.signal_type' = 'garbage';
+
+-- Entity columns must exist.
+ALTER TABLE altered_semantics SET 'greptime.semantic.entity.host.id' = 'no_such_column';
+
+-- Entity columns must render as strings.
+ALTER TABLE altered_semantics SET 'greptime.semantic.entity.host.id' = 'payload';
+
+ALTER TABLE altered_semantics UNSET 'greptime.semantic.entity.service.id';
+
+SELECT table_name, signal_type, semantic_options
+FROM information_schema.table_semantics
+WHERE table_name = 'altered_semantics';
+
+DROP TABLE altered_semantics;
+
+-- Logical metric tables take the same metadata-only path.
+CREATE TABLE phy_sem (ts TIMESTAMP TIME INDEX, val DOUBLE) engine=metric with ("physical_metric_table" = "");
+
+CREATE TABLE logical_sem (ts TIMESTAMP TIME INDEX, val DOUBLE, host STRING PRIMARY KEY) engine=metric with ("on_physical_table" = "phy_sem");
+
+ALTER TABLE logical_sem SET 'greptime.semantic.signal_type' = 'metric', 'greptime.semantic.entity.host.id' = 'host';
+
+SELECT table_name, signal_type, semantic_options
+FROM information_schema.table_semantics
+WHERE table_name = 'logical_sem';
+
+-- Regular options still cannot be altered on logical tables.
+ALTER TABLE logical_sem SET 'ttl' = '7d';
+
+ALTER TABLE logical_sem UNSET 'greptime.semantic.entity.host.id';
+
+SELECT table_name, signal_type, semantic_options
+FROM information_schema.table_semantics
+WHERE table_name = 'logical_sem';
+
+DROP TABLE logical_sem;
+
+DROP TABLE phy_sem;

tests/cases/standalone/common/system/semantic_graph.result

@@ -119,6 +119,61 @@ drop table graph_app_metrics;
 
 Affected Rows: 0
 
+-- Declarations can be added after the fact: a table created without semantic
+-- options joins the graph once ALTER TABLE declares its entities, and leaves
+-- it again on UNSET.
+create table graph_late_metrics (
+  ts timestamp time index,
+  svc string,
+  env string,
+  latency double
+);
+
+Affected Rows: 0
+
+insert into graph_late_metrics values (now(), 'checkout', 'eu-1', 1.5);
+
+Affected Rows: 1
+
+select entity_type, entity_id from greptime_private.semantic_entities order by entity_type, entity_id;
+
+++
+++
+
+alter table graph_late_metrics set 'greptime.semantic.entity.service.id' = 'svc', 'greptime.semantic.entity.service.scope' = 'env';
+
+Affected Rows: 0
+
+select entity_type, entity_id, scope from greptime_private.semantic_entities order by entity_type, entity_id;
+
++-------------+-----------+-------+
+| entity_type | entity_id | scope |
++-------------+-----------+-------+
+| service     | checkout  | eu-1  |
++-------------+-----------+-------+
+
+-- A column referenced by a declaration must keep a string-renderable type.
+alter table graph_late_metrics modify column svc binary;
+
+Error: 1004(InvalidArguments), Invalid alter table(graph_late_metrics) request: column `svc` is referenced by entity option `greptime.semantic.entity.service.id` and must keep a type that renders as a string, got `Binary`
+
+alter table graph_late_metrics unset 'greptime.semantic.entity.service.id';
+
+Affected Rows: 0
+
+alter table graph_late_metrics unset 'greptime.semantic.entity.service.scope';
+
+Affected Rows: 0
+
+select entity_type, entity_id from greptime_private.semantic_entities order by entity_type, entity_id;
+
+++
+++
+
+drop table graph_late_metrics;
+
+Affected Rows: 0
+
 -- Calls derivation over trace-v1 tables, all branches in one scan: the client
 -- span and its child server span land in different tables and still pair into
 -- one edge with RED metrics; the epoch-timestamped pair falls outside the

tests/cases/standalone/common/system/semantic_graph.sql

@@ -66,6 +66,35 @@ order by rel_type, src_id;
 
 drop table graph_app_metrics;
 
+-- Declarations can be added after the fact: a table created without semantic
+-- options joins the graph once ALTER TABLE declares its entities, and leaves
+-- it again on UNSET.
+create table graph_late_metrics (
+  ts timestamp time index,
+  svc string,
+  env string,
+  latency double
+);
+
+insert into graph_late_metrics values (now(), 'checkout', 'eu-1', 1.5);
+
+select entity_type, entity_id from greptime_private.semantic_entities order by entity_type, entity_id;
+
+alter table graph_late_metrics set 'greptime.semantic.entity.service.id' = 'svc', 'greptime.semantic.entity.service.scope' = 'env';
+
+select entity_type, entity_id, scope from greptime_private.semantic_entities order by entity_type, entity_id;
+
+-- A column referenced by a declaration must keep a string-renderable type.
+alter table graph_late_metrics modify column svc binary;
+
+alter table graph_late_metrics unset 'greptime.semantic.entity.service.id';
+
+alter table graph_late_metrics unset 'greptime.semantic.entity.service.scope';
+
+select entity_type, entity_id from greptime_private.semantic_entities order by entity_type, entity_id;
+
+drop table graph_late_metrics;
+
 -- Calls derivation over trace-v1 tables, all branches in one scan: the client
 -- span and its child server span land in different tables and still pair into
 -- one edge with RED metrics; the epoch-timestamped pair falls outside the