From 09c0b23a23450d0aa3bb2d89938ad0f3f5be253c Mon Sep 17 00:00:00 2001 From: dennis zhuang Date: Mon, 17 Aug 2026 09:53:59 +0000 Subject: [PATCH] feat: manage semantic table options via ALTER TABLE SET/UNSET (#8880) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 * 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 * 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 * 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 * 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 * style(grpc-expr): drop a test comment restating the classifier doc Signed-off-by: Dennis Zhuang * 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 --------- Signed-off-by: Dennis Zhuang --- src/common/grpc-expr/src/alter.rs | 256 +++++++-- src/common/grpc-expr/src/lib.rs | 2 +- src/common/meta/AGENTS.md | 5 + .../meta/src/ddl/alter_logical_tables.rs | 57 +- src/common/meta/src/ddl/alter_table.rs | 42 +- .../meta/src/ddl/alter_table/executor.rs | 4 +- .../src/ddl/tests/alter_logical_tables.rs | 103 +++- src/common/meta/src/ddl/tests/alter_table.rs | 59 +- src/common/meta/src/ddl/tests/event/table.rs | 1 + src/common/meta/src/ddl_manager.rs | 162 +++++- src/operator/src/statement/ddl.rs | 209 +++---- src/table/src/metadata.rs | 510 ++++++++++++++++-- src/table/src/requests.rs | 217 +++++++- src/table/src/requests/semantic.rs | 26 +- .../information_schema/table_semantics.result | 118 ++++ .../information_schema/table_semantics.sql | 62 +++ .../common/system/semantic_graph.result | 55 ++ .../common/system/semantic_graph.sql | 29 + 18 files changed, 1630 insertions(+), 287 deletions(-) diff --git a/src/common/grpc-expr/src/alter.rs b/src/common/grpc-expr/src/alter.rs index 46de85da4c..85de11d5cd 100644 --- a/src/common/grpc-expr/src/alter.rs +++ b/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, +) -> Result> { + 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> { + 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 { 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}" + ); + } } diff --git a/src/common/grpc-expr/src/lib.rs b/src/common/grpc-expr/src/lib.rs index 0941cb650d..9b9994a8f0 100644 --- a/src/common/grpc-expr/src/lib.rs +++ b/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}; diff --git a/src/common/meta/AGENTS.md b/src/common/meta/AGENTS.md index fd5da08a93..3684f8f9db 100644 --- a/src/common/meta/AGENTS.md +++ b/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`. diff --git a/src/common/meta/src/ddl/alter_logical_tables.rs b/src/common/meta/src/ddl/alter_logical_tables.rs index 9207f91057..ccf130cec8 100644 --- a/src/common/meta/src/ddl/alter_logical_tables.rs +++ b/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, physical_table_id: TableId, + logical_table_ids: Vec, 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>, /// 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, physical_table_info: Option>, physical_columns: Vec, table_cache_keys_to_invalidate: Vec, diff --git a/src/common/meta/src/ddl/alter_table.rs b/src/common/meta/src/ddl/alter_table.rs index 63342a7bbb..8e24f3ae2a 100644 --- a/src/common/meta/src/ddl/alter_table.rs +++ b/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 { + // 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 { + 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 { // Safety: Checked in `AlterTableProcedure::new`. AlterTableFlow::from_kind(self.task.alter_table.kind.as_ref().unwrap()) } diff --git a/src/common/meta/src/ddl/alter_table/executor.rs b/src/common/meta/src/ddl/alter_table/executor.rs index fbacdd1b2f..65fa45d0a8 100644 --- a/src/common/meta/src/ddl/alter_table/executor.rs +++ b/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 { .. } diff --git a/src/common/meta/src/ddl/tests/alter_logical_tables.rs b/src/common/meta/src/ddl/tests/alter_logical_tables.rs index 351c0fd96e..619e02bb9b 100644 --- a/src/common/meta/src/ddl/tests/alter_logical_tables.rs +++ b/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::>(); + 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::>(); + 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}" + ); +} diff --git a/src/common/meta/src/ddl/tests/alter_table.rs b/src/common/meta/src/ddl/tests/alter_table.rs index b35666e52e..30910eb0ba 100644 --- a/src/common/meta/src/ddl/tests/alter_table.rs +++ b/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()) + ); +} diff --git a/src/common/meta/src/ddl/tests/event/table.rs b/src/common/meta/src/ddl/tests/event/table.rs index 24182fd19a..92fa10ee9e 100644 --- a/src/common/meta/src/ddl/tests/event/table.rs +++ b/src/common/meta/src/ddl/tests/event/table.rs @@ -391,6 +391,7 @@ fn procedure_cases() -> Vec { ), ], 43, + vec![], test_context(), ); let drop_table = DropTableProcedure::new( diff --git a/src/common/meta/src/ddl_manager.rs b/src/common/meta/src/ddl_manager.rs index 7ba408ca2d..d88f08918e 100644 --- a/src/common/meta/src/ddl_manager.rs +++ b/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)> { 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::>(); + 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}"); + } } diff --git a/src/operator/src/statement/ddl.rs b/src/operator/src/statement/ddl.rs index 3c10186ca4..f89f5de73a 100644 --- a/src/operator/src/statement/ddl.rs +++ b/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, - 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..{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) -> Result> { @@ -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 diff --git a/src/table/src/metadata.rs b/src/table/src/metadata.rs index 7b8958e22c..4f7ed1c057 100644 --- a/src/table/src/metadata.rs +++ b/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 { - 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 { + /// 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 { + 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(); + } } diff --git a/src/table/src/requests.rs b/src/table/src/requests.rs index b4a7d5e052..aea45130ed 100644 --- a/src/table/src/requests.rs +++ b/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 { + 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 { + 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, }, - SetRepartitionColumnHint { - column_name: String, + SetAnnotations { + family: AnnotationFamily, + options: Vec<(String, String)>, + }, + UnsetAnnotations { + family: AnnotationFamily, + keys: Vec, }, - UnsetRepartitionColumnHint, SetIndexes { options: Vec, }, diff --git a/src/table/src/requests/semantic.rs b/src/table/src/requests/semantic.rs index 5c0e4c3cae..2daa440c11 100644 --- a/src/table/src/requests/semantic.rs +++ b/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. diff --git a/tests/cases/standalone/common/information_schema/table_semantics.result b/tests/cases/standalone/common/information_schema/table_semantics.result index 5aa39a91fb..b57087b3f5 100644 --- a/tests/cases/standalone/common/information_schema/table_semantics.result +++ b/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 + diff --git a/tests/cases/standalone/common/information_schema/table_semantics.sql b/tests/cases/standalone/common/information_schema/table_semantics.sql index 8fb7c78aef..4acb15294d 100644 --- a/tests/cases/standalone/common/information_schema/table_semantics.sql +++ b/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; diff --git a/tests/cases/standalone/common/system/semantic_graph.result b/tests/cases/standalone/common/system/semantic_graph.result index ed6a3d3f16..6af28cfa2b 100644 --- a/tests/cases/standalone/common/system/semantic_graph.result +++ b/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 diff --git a/tests/cases/standalone/common/system/semantic_graph.sql b/tests/cases/standalone/common/system/semantic_graph.sql index 181a8da66a..bfc9ea8fd7 100644 --- a/tests/cases/standalone/common/system/semantic_graph.sql +++ b/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