diff --git a/src/common/meta/src/datanode.rs b/src/common/meta/src/datanode.rs index 8af8646212..d50c37c381 100644 --- a/src/common/meta/src/datanode.rs +++ b/src/common/meta/src/datanode.rs @@ -31,6 +31,8 @@ use crate::heartbeat::utils::get_datanode_workloads; const DATANODE_STAT_PREFIX: &str = "__meta_datanode_stat"; pub const REGION_STATISTIC_KEY: &str = "__region_statistic"; +/// The private table name for persisted region statistics. +pub const REGION_STATS_HISTORY_TABLE_NAME: &str = "region_statistics_history"; lazy_static! { pub(crate) static ref DATANODE_LEASE_KEY_PATTERN: Regex = diff --git a/src/meta-srv/src/handler/persist_stats_handler.rs b/src/meta-srv/src/handler/persist_stats_handler.rs index b1378c39e6..fcdb9fa04b 100644 --- a/src/meta-srv/src/handler/persist_stats_handler.rs +++ b/src/meta-srv/src/handler/persist_stats_handler.rs @@ -22,7 +22,7 @@ use client::inserter::{Context as InserterContext, Inserter}; use common_catalog::consts::DEFAULT_PRIVATE_SCHEMA_NAME; use common_macro::{Schema, ToRow}; use common_meta::DatanodeId; -use common_meta::datanode::RegionStat; +use common_meta::datanode::{REGION_STATS_HISTORY_TABLE_NAME, RegionStat}; use common_telemetry::warn; use dashmap::DashMap; use store_api::region_engine::RegionRole; @@ -40,8 +40,6 @@ pub struct PersistStatsHandler { persist_interval: Duration, } -/// The name of the table to persist region stats. -const META_REGION_STATS_HISTORY_TABLE_NAME: &str = "region_statistics_history"; /// The default context to persist region stats. const DEFAULT_CONTEXT: InserterContext = InserterContext { catalog: DEFAULT_CATALOG_NAME, @@ -207,7 +205,7 @@ impl PersistStatsHandler { &DEFAULT_CONTEXT, RowInsertRequests { inserts: vec![RowInsertRequest { - table_name: META_REGION_STATS_HISTORY_TABLE_NAME.to_string(), + table_name: REGION_STATS_HISTORY_TABLE_NAME.to_string(), rows: Some(Rows { schema: PersistRegionStat::schema(), rows, @@ -537,7 +535,7 @@ mod tests { }; assert_eq!( request.table_name, - META_REGION_STATS_HISTORY_TABLE_NAME.to_string() + REGION_STATS_HISTORY_TABLE_NAME.to_string() ); assert_eq!(request.rows.unwrap().rows, vec![expected_row]); diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index ead517ab80..52a73892dc 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -29,12 +29,15 @@ use api::v1::{ use catalog::CatalogManagerRef; use client::{OutputData, OutputMeta}; use common_catalog::consts::{ - PARENT_SPAN_ID_COLUMN, SERVICE_NAME_COLUMN, TRACE_ID_COLUMN, TRACE_TABLE_NAME, - TRACE_TABLE_NAME_SESSION_KEY, default_engine, is_ddl_reserved_table, + DEFAULT_PRIVATE_SCHEMA_NAME, PARENT_SPAN_ID_COLUMN, SERVICE_NAME_COLUMN, TRACE_ID_COLUMN, + TRACE_TABLE_NAME, TRACE_TABLE_NAME_SESSION_KEY, default_engine, is_ddl_reserved_table, trace_operations_table_name, trace_services_table_name, }; +use common_event_recorder::DEFAULT_EVENTS_TABLE_NAME; +use common_frontend::slow_query_event::SLOW_QUERY_TABLE_NAME; use common_grpc_expr::util::ColumnExpr; use common_meta::cache::TableFlownodeSetCacheRef; +use common_meta::datanode::REGION_STATS_HISTORY_TABLE_NAME; use common_meta::node_manager::{AffectedRows, NodeManagerRef}; use common_meta::peer::Peer; use common_meta::rpc::ddl::TriggerReason; @@ -490,9 +493,9 @@ impl Inserter { Ok(inserts) } - /// Returns `None` if auto table creation is allowed, or `Some(reason)` if - /// disabled by either the global config or the request hint. The reason tells - /// which one, for a clearer error. + /// Returns `Some(reason)` if the config or request hint disables automatic + /// table creation. Exempt private system tables are handled by + /// [`Self::is_auto_create_exempt_private_table`]. fn auto_create_disabled_reason(&self, ctx: &QueryContextRef) -> Result> { let auto_create_table_hint = ctx .extension(AUTO_CREATE_TABLE_KEY) @@ -514,6 +517,16 @@ impl Inserter { }) } + /// Returns whether a private system table may infer and reconcile its schema + /// even when automatic table creation is disabled. + fn is_auto_create_exempt_private_table(schema: &str, table: &str) -> bool { + schema == DEFAULT_PRIVATE_SCHEMA_NAME + && matches!( + table, + DEFAULT_EVENTS_TABLE_NAME | SLOW_QUERY_TABLE_NAME | REGION_STATS_HISTORY_TABLE_NAME + ) + } + /// Ensures a trace table has the request-global schema without requiring a /// padded data row to drive on-demand creation or alteration. When /// `alter_existing` is false, a table created after planning is left for the @@ -571,12 +584,21 @@ impl Inserter { let _timer = crate::metrics::CREATE_ALTER_ON_DEMAND .with_label_values(&[auto_create_table_type.as_str()]) .start_timer(); - let catalog = ctx.current_catalog(); let schema = ctx.current_schema(); + let auto_create_disabled_reason = self.auto_create_disabled_reason(ctx)?; + // Enabled batches permit every table, so only disabled batches need a whitelist scan. + let has_auto_create_exempt_table = auto_create_disabled_reason.is_some() + && requests + .inserts + .iter() + .any(|req| Self::is_auto_create_exempt_private_table(&schema, &req.table_name)); let mut table_infos = HashMap::new(); - if let Some(disabled_reason) = self.auto_create_disabled_reason(ctx)? { + // Without exempt tables, verify existing tables and reject missing ones without inferring schemas. + if let Some(disabled_reason) = auto_create_disabled_reason + && !has_auto_create_exempt_table + { let mut instant_table_ids = HashSet::new(); for req in &requests.inserts { let table = match self.get_table(catalog, &schema, &req.table_name).await? { @@ -618,20 +640,25 @@ impl Inserter { let mut per_table_semantics: Option> = None; for req in &mut requests.inserts { + // Mixed batches need a per-table decision so an exempt table cannot authorize others. + let auto_create_allowed = auto_create_disabled_reason.is_none() + || Self::is_auto_create_exempt_private_table(&schema, &req.table_name); match self.get_table(catalog, &schema, &req.table_name).await? { Some(table) => { let table_info = table.table_info(); if table_info.is_ttl_instant_table() { instant_table_ids.insert(table_info.table_id()); } - if let Some(alter_expr) = self.get_alter_table_expr_on_demand( - req, - &table, - ctx, - accommodate_existing_schema, - is_single_value, - auto_create_table_type.alter_existing(), - )? { + if auto_create_allowed + && let Some(alter_expr) = self.get_alter_table_expr_on_demand( + req, + &table, + ctx, + accommodate_existing_schema, + is_single_value, + auto_create_table_type.alter_existing(), + )? + { alter_tables.push(alter_expr); need_refresh_table_infos.insert(( catalog.to_string(), @@ -655,6 +682,17 @@ impl Inserter { } table_infos.insert(table_info.table_id(), table_info); } + None if !auto_create_allowed + && let Some(disabled_reason) = auto_create_disabled_reason => + { + return InvalidInsertRequestSnafu { + reason: format!( + "Table `{}` does not exist, and {}", + req.table_name, disabled_reason, + ), + } + .fail(); + } None => { let semantic_index = per_table_semantics .get_or_insert_with(|| parse_per_table_semantic_index(ctx)) diff --git a/tests-integration/src/cluster.rs b/tests-integration/src/cluster.rs index d0cd2e584c..e40c707fe9 100644 --- a/tests-integration/src/cluster.rs +++ b/tests-integration/src/cluster.rs @@ -169,6 +169,7 @@ pub struct GreptimeDbClusterBuilder { metasrv_wal_config: MetasrvWalConfig, datanode_gc_config: GcConfig, metasrv_gc_config: GcSchedulerOptions, + frontend_auto_create_table: bool, shared_home_dir: Option>, meta_selector: Option, } @@ -202,6 +203,7 @@ impl GreptimeDbClusterBuilder { metasrv_wal_config: MetasrvWalConfig::default(), datanode_gc_config: GcConfig::default(), metasrv_gc_config: GcSchedulerOptions::default(), + frontend_auto_create_table: true, shared_home_dir: None, meta_selector: None, } @@ -248,6 +250,13 @@ impl GreptimeDbClusterBuilder { self } + /// Sets whether the test frontend automatically creates tables on write. + #[must_use] + pub fn with_frontend_auto_create_table(mut self, auto_create_table: bool) -> Self { + self.frontend_auto_create_table = auto_create_table; + self + } + #[must_use] pub fn with_shared_home_dir(mut self, shared_home_dir: Arc) -> Self { self.shared_home_dir = Some(shared_home_dir); @@ -529,7 +538,10 @@ impl GreptimeDbClusterBuilder { } fn build_frontend_options(&self) -> FrontendOptions { - let mut fe_opts = FrontendOptions::default(); + let mut fe_opts = FrontendOptions { + auto_create_table: self.frontend_auto_create_table, + ..Default::default() + }; // Choose a random unused port between [14000, 24000] for local test to avoid conflicts. let port_range = 14000..=24000; diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 16fc2e6563..0704b184e8 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -27,6 +27,7 @@ use common_base::Plugins; use common_catalog::consts::{MIN_USER_FLOW_ID, MIN_USER_TABLE_ID}; use common_config::KvBackendConfig; use common_datasource::object_store::LocalFileAccess; +use common_event_recorder::EventRecorderOptions; use common_meta::cache::LayeredCacheRegistryBuilder; use common_meta::ddl::flow_meta::FlowMetadataAllocator; use common_meta::ddl::table_meta::TableMetadataAllocator; @@ -84,6 +85,7 @@ pub struct GreptimeDbStandaloneBuilder { default_store: Option, plugin: Option, slow_query_options: SlowQueryOptions, + event_recorder_options: EventRecorderOptions, auto_create_table: bool, } @@ -102,6 +104,7 @@ impl GreptimeDbStandaloneBuilder { threshold: Duration::from_secs(1), ..Default::default() }, + event_recorder_options: EventRecorderOptions::default(), auto_create_table: true, } } @@ -112,6 +115,16 @@ impl GreptimeDbStandaloneBuilder { self } + /// Sets the event recorder options for the standalone instance. + #[must_use] + pub fn with_event_recorder_options( + mut self, + event_recorder_options: EventRecorderOptions, + ) -> Self { + self.event_recorder_options = event_recorder_options; + self + } + #[must_use] pub fn with_default_store_type(self, store_type: StorageType) -> Self { Self { @@ -368,6 +381,7 @@ impl GreptimeDbStandaloneBuilder { wal: self.metasrv_wal_config.clone().into(), grpc: GrpcOptions::default().with_server_addr("127.0.0.1:4001"), slow_query: self.slow_query_options.clone(), + event_recorder: self.event_recorder_options.clone(), auto_create_table: self.auto_create_table, // Tests cover the descriptor, so they run with it enabled. otlp: frontend::service_config::OtlpOptions { diff --git a/tests-integration/tests/database_ddl_event.rs b/tests-integration/tests/database_ddl_event.rs index 4dfda4f32b..c08b81211d 100644 --- a/tests-integration/tests/database_ddl_event.rs +++ b/tests-integration/tests/database_ddl_event.rs @@ -15,20 +15,163 @@ use std::sync::Arc; use client::{Database, OutputData}; +use common_event_recorder::{EventRecorderOptions, EventTypeFilter}; use common_test_util::temp_dir::create_temp_dir; use tests_integration::cluster::GreptimeDbClusterBuilder; use tests_integration::standalone::GreptimeDbStandaloneBuilder; use tests_integration::test_util::{ - StorageType, get_test_store_config, setup_authenticated_grpc_database, + StorageType, execute_sql, get_test_store_config, setup_authenticated_grpc_database, }; use crate::event_recorder_test_util::{ - assert_procedure_actor, assert_single_event, find_eventually_string, + assert_eventually_eq, assert_procedure_actor, assert_single_event, find_eventually_string, }; const DATABASE_NAME: &str = "database_ddl_events"; const PROCEDURE_ACTOR: &str = "procedure_actor"; const PROCEDURE_ACTOR_PASSWORD: &str = "procedure_actor_pwd"; +const CREATION_DATABASE_NAME: &str = "event_schema_creation"; +const RECONCILIATION_DATABASE_NAME: &str = "event_schema_reconciliation"; + +#[tokio::test(flavor = "multi_thread")] +async fn test_event_table_auto_creation_with_auto_create_disabled() { + common_telemetry::init_default_ut_logging(); + let standalone = GreptimeDbStandaloneBuilder::new("test_event_table_auto_creation") + .with_auto_create_table(false) + .with_event_recorder_options(EventRecorderOptions { + event_types: Arc::new(EventTypeFilter::Only( + [String::from("create_database")].into_iter().collect(), + )), + ..Default::default() + }) + .build() + .await; + let instance = standalone.fe_instance(); + + let (database, _grpc_server) = setup_authenticated_grpc_database( + instance.clone(), + PROCEDURE_ACTOR, + PROCEDURE_ACTOR_PASSWORD, + ) + .await; + database + .sql(format!("CREATE DATABASE {CREATION_DATABASE_NAME}")) + .await + .unwrap(); + + assert_eventually_eq( + instance, + "SELECT count(*) AS events_tables \ + FROM information_schema.tables \ + WHERE table_catalog = 'greptime' \ + AND table_schema = 'greptime_private' \ + AND table_name = 'events'", + "+---------------+\n| events_tables |\n+---------------+\n| 1 |\n+---------------+", + ) + .await; + + let procedure_id = find_eventually_string( + instance, + &format!( + "SELECT procedure_id FROM greptime_private.events \ + WHERE type = 'create_database' \ + AND schema_name = '{CREATION_DATABASE_NAME}' \ + AND json_path_match(procedure_trigger, '$.type == \"Submitted\"') \ + ORDER BY timestamp DESC LIMIT 1" + ), + "procedure_id", + ) + .await; + assert_procedure_actor(instance, &procedure_id, Some(PROCEDURE_ACTOR)).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_event_table_schema_reconciliation_with_auto_create_disabled() { + common_telemetry::init_default_ut_logging(); + let standalone = GreptimeDbStandaloneBuilder::new("test_event_table_schema_reconciliation") + .with_auto_create_table(false) + .with_event_recorder_options(EventRecorderOptions { + event_types: Arc::new(EventTypeFilter::Only( + [String::from("create_database")].into_iter().collect(), + )), + ..Default::default() + }) + .build() + .await; + let instance = standalone.fe_instance(); + + // Matches the pre-actor procedure-event schema. The recorder filter keeps + // this setup DDL from creating an event before the regression is exercised. + execute_sql( + instance, + r#" + CREATE TABLE greptime_private.events ( + "type" STRING, + payload JSON, + "timestamp" TIMESTAMP(9) NOT NULL, + procedure_id STRING, + procedure_state STRING, + procedure_error STRING, + procedure_trigger JSON, + catalog_name STRING, + schema_name STRING, + event_context JSON, + TIME INDEX ("timestamp"), + PRIMARY KEY ("type") + ) WITH (append_mode = 'true') + "#, + ) + .await; + + assert_eventually_eq( + instance, + "SELECT count(*) AS actor_columns \ + FROM information_schema.columns \ + WHERE table_catalog = 'greptime' \ + AND table_schema = 'greptime_private' \ + AND table_name = 'events' \ + AND column_name = 'actor'", + "+---------------+\n| actor_columns |\n+---------------+\n| 0 |\n+---------------+", + ) + .await; + + let (database, _grpc_server) = setup_authenticated_grpc_database( + instance.clone(), + PROCEDURE_ACTOR, + PROCEDURE_ACTOR_PASSWORD, + ) + .await; + database + .sql(format!("CREATE DATABASE {RECONCILIATION_DATABASE_NAME}")) + .await + .unwrap(); + + assert_eventually_eq( + instance, + "SELECT count(*) AS actor_columns \ + FROM information_schema.columns \ + WHERE table_catalog = 'greptime' \ + AND table_schema = 'greptime_private' \ + AND table_name = 'events' \ + AND column_name = 'actor'", + "+---------------+\n| actor_columns |\n+---------------+\n| 1 |\n+---------------+", + ) + .await; + + let procedure_id = find_eventually_string( + instance, + &format!( + "SELECT procedure_id FROM greptime_private.events \ + WHERE type = 'create_database' \ + AND schema_name = '{RECONCILIATION_DATABASE_NAME}' \ + AND json_path_match(procedure_trigger, '$.type == \"Submitted\"') \ + ORDER BY timestamp DESC LIMIT 1" + ), + "procedure_id", + ) + .await; + assert_procedure_actor(instance, &procedure_id, Some(PROCEDURE_ACTOR)).await; +} #[tokio::test(flavor = "multi_thread")] async fn test_database_ddl_events() { @@ -43,6 +186,7 @@ async fn test_database_ddl_events() { let cluster = GreptimeDbClusterBuilder::new("test_database_ddl_events") .await .with_datanodes(1) + .with_frontend_auto_create_table(false) .with_store_config(store_config) .with_shared_home_dir(Arc::new(home_dir)) .build(true) diff --git a/tests-integration/tests/grpc.rs b/tests-integration/tests/grpc.rs index 737b4fb6c1..f06ffa6522 100644 --- a/tests-integration/tests/grpc.rs +++ b/tests-integration/tests/grpc.rs @@ -26,9 +26,12 @@ use api::v1::{ use auth::user_provider_from_option; use base64::prelude::{BASE64_STANDARD, Engine as _}; use client::{Client, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, Database, OutputData}; -use common_catalog::consts::MITO_ENGINE; +use common_catalog::consts::{DEFAULT_PRIVATE_SCHEMA_NAME, MITO_ENGINE}; +use common_event_recorder::DEFAULT_EVENTS_TABLE_NAME; +use common_frontend::slow_query_event::SLOW_QUERY_TABLE_NAME; use common_grpc::channel_manager::ClientTlsOption; use common_memory_manager::OnExhaustedPolicy; +use common_meta::datanode::REGION_STATS_HISTORY_TABLE_NAME; use common_query::Output; use common_recordbatch::RecordBatches; use common_runtime::Runtime; @@ -98,6 +101,8 @@ macro_rules! grpc_tests { test_auto_create_table, test_auto_create_table_with_hints, test_auto_create_table_disabled_by_config, + test_private_system_tables_auto_create_table_with_global_disabled, + test_private_system_tables_bypass_auto_create_hint, test_otel_arrow_auth, test_otel_arrow_exponential_histogram, test_insert_and_select, @@ -679,6 +684,154 @@ pub async fn test_auto_create_table_disabled_by_config(store_type: StorageType) let _ = fe_grpc_server.shutdown().await; } +pub async fn test_private_system_tables_auto_create_table_with_global_disabled( + store_type: StorageType, +) { + let (_db, fe_grpc_server) = setup_grpc_server_with_auto_create_table_disabled( + store_type, + "test_private_system_tables_auto_create_table_with_global_disabled", + ) + .await; + let addr = fe_grpc_server.bind_addr().unwrap().to_string(); + + let grpc_client = Client::with_urls(vec![addr]); + let db = Database::new( + DEFAULT_CATALOG_NAME, + DEFAULT_PRIVATE_SCHEMA_NAME, + grpc_client, + ); + let (host, cpu, mem, ts) = expect_data(); + + for table_name in [ + DEFAULT_EVENTS_TABLE_NAME, + SLOW_QUERY_TABLE_NAME, + REGION_STATS_HISTORY_TABLE_NAME, + ] { + let result = db + .insert(InsertRequests { + inserts: vec![InsertRequest { + table_name: table_name.to_string(), + columns: vec![host.clone(), cpu.clone(), mem.clone(), ts.clone()], + row_count: 4, + }], + }) + .await; + assert_eq!(result.unwrap(), 4); + } + + let load = Column { + column_name: "load".to_string(), + values: Some(column::Values { + f64_values: vec![0.4, 0.5, 0.6, 0.7], + ..Default::default() + }), + semantic_type: SemanticType::Field as i32, + datatype: ColumnDataType::Float64 as i32, + ..Default::default() + }; + let result = db + .insert(InsertRequests { + inserts: vec![InsertRequest { + table_name: DEFAULT_EVENTS_TABLE_NAME.to_string(), + columns: vec![host, cpu, mem, ts, load], + row_count: 4, + }], + }) + .await; + assert_eq!(result.unwrap(), 4); + + let output = db + .sql(format!("SHOW CREATE TABLE {DEFAULT_EVENTS_TABLE_NAME}")) + .await + .unwrap(); + let record_batches = match output.data { + OutputData::RecordBatches(record_batches) => record_batches, + OutputData::Stream(stream) => RecordBatches::try_collect(stream).await.unwrap(), + OutputData::AffectedRows(_) => unreachable!(), + }; + assert!(record_batches.pretty_print().unwrap().contains("\"load\"")); + + let _ = fe_grpc_server.shutdown().await; +} + +pub async fn test_private_system_tables_bypass_auto_create_hint(store_type: StorageType) { + let (_db, fe_grpc_server) = setup_grpc_server( + store_type, + "test_private_system_tables_bypass_auto_create_hint", + ) + .await; + let addr = fe_grpc_server.bind_addr().unwrap().to_string(); + + let grpc_client = Client::with_urls(vec![addr]); + let db = Database::new( + DEFAULT_CATALOG_NAME, + DEFAULT_PRIVATE_SCHEMA_NAME, + grpc_client, + ); + let (host, cpu, mem, ts) = expect_data(); + + for table_name in [ + DEFAULT_EVENTS_TABLE_NAME, + SLOW_QUERY_TABLE_NAME, + REGION_STATS_HISTORY_TABLE_NAME, + ] { + let result = db + .insert_with_hints( + InsertRequests { + inserts: vec![InsertRequest { + table_name: table_name.to_string(), + columns: vec![host.clone(), cpu.clone(), mem.clone(), ts.clone()], + row_count: 4, + }], + }, + &[("auto_create_table", "false")], + ) + .await; + assert_eq!(result.unwrap(), 4); + } + + let ordinary_table = "ordinary_private_table"; + let result = db + .insert_with_hints( + InsertRequests { + inserts: vec![ + InsertRequest { + table_name: DEFAULT_EVENTS_TABLE_NAME.to_string(), + columns: vec![host.clone(), cpu.clone(), mem.clone(), ts.clone()], + row_count: 4, + }, + InsertRequest { + table_name: ordinary_table.to_string(), + columns: vec![host, cpu, mem, ts], + row_count: 4, + }, + ], + }, + &[("auto_create_table", "false")], + ) + .await; + let err = result.unwrap_err().to_string(); + assert!( + err.contains(ordinary_table) && err.contains("auto_create_table"), + "unexpected error: {err}" + ); + + let output = db.sql("SHOW TABLES").await.unwrap(); + let record_batches = match output.data { + OutputData::RecordBatches(record_batches) => record_batches, + OutputData::Stream(stream) => RecordBatches::try_collect(stream).await.unwrap(), + OutputData::AffectedRows(_) => unreachable!(), + }; + assert!( + !record_batches + .pretty_print() + .unwrap() + .contains(ordinary_table) + ); + + let _ = fe_grpc_server.shutdown().await; +} + fn expect_data() -> (Column, Column, Column, Column) { // testing data: let expected_host_col = Column {