diff --git a/src/frontend/src/instance/prom_store.rs b/src/frontend/src/instance/prom_store.rs index af2a95857c7..ee07d9e8fd5 100644 --- a/src/frontend/src/instance/prom_store.rs +++ b/src/frontend/src/instance/prom_store.rs @@ -293,6 +293,10 @@ impl PendingRowsSchemaAlterer for Instance { if tables.is_empty() { return Ok(()); } + self.inserter + .ensure_auto_create_allowed(&ctx) + .map_err(BoxedError::new) + .context(error::ExecuteGrpcQuerySnafu)?; let create_type = auto_create_table_type_for_prom_remote_write(&ctx, with_metric_engine); if let Some(physical_table) = required_physical_table_for_create_type(&create_type) { @@ -397,6 +401,10 @@ impl PendingRowsSchemaAlterer for Instance { if tables.is_empty() { return Ok(()); } + self.inserter + .ensure_auto_create_allowed(&ctx) + .map_err(BoxedError::new) + .context(error::ExecuteGrpcQuerySnafu)?; let alter_exprs: Vec = tables .iter() diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index 52a73892dc0..fabdf21ef7e 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -517,6 +517,17 @@ impl Inserter { }) } + /// Ensures automatic table creation and schema evolution are allowed for the request. + pub fn ensure_auto_create_allowed(&self, ctx: &QueryContextRef) -> Result<()> { + if let Some(reason) = self.auto_create_disabled_reason(ctx)? { + return InvalidInsertRequestSnafu { + reason: reason.to_string(), + } + .fail(); + } + Ok(()) + } + /// 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 { diff --git a/tests-integration/src/test_util.rs b/tests-integration/src/test_util.rs index 040d014487c..db14d0357ff 100644 --- a/tests-integration/src/test_util.rs +++ b/tests-integration/src/test_util.rs @@ -478,7 +478,20 @@ pub async fn setup_test_otlp_metrics_app_with_frontend_batched( store_type: StorageType, name: &str, ) -> (Router, TestGuard) { - let instance = setup_standalone_instance(name, store_type).await; + setup_test_otlp_metrics_app_with_frontend_batched_and_auto_create_table(store_type, name, true) + .await +} + +pub async fn setup_test_otlp_metrics_app_with_frontend_batched_and_auto_create_table( + store_type: StorageType, + name: &str, + auto_create_table: bool, +) -> (Router, TestGuard) { + let instance = GreptimeDbStandaloneBuilder::new(name) + .with_default_store_type(store_type) + .with_auto_create_table(auto_create_table) + .build() + .await; let frontend = instance.fe_instance().clone(); let mut options = instance.opts.clone(); options.prom_store.with_metric_engine = true; diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 0581fc04b40..0745a3537c3 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -81,8 +81,10 @@ use tests_integration::test_util::{ StorageType, setup_test_http_app, setup_test_http_app_with_frontend, setup_test_http_app_with_frontend_and_slow_query_threshold, setup_test_http_app_with_frontend_and_user_provider, - setup_test_otlp_metrics_app_with_frontend_batched, setup_test_prom_app_with_frontend, - setup_test_prom_app_with_frontend_batched, setup_test_prom_app_with_frontend_native_histogram, + setup_test_otlp_metrics_app_with_frontend_batched, + setup_test_otlp_metrics_app_with_frontend_batched_and_auto_create_table, + setup_test_prom_app_with_frontend, setup_test_prom_app_with_frontend_batched, + setup_test_prom_app_with_frontend_native_histogram, }; use urlencoding::encode; use yaml_rust::YamlLoader; @@ -7097,6 +7099,75 @@ async fn test_otlp_metrics_batching_respects_existing_logical_table_placement() guard.remove_all().await; } +#[tokio::test(flavor = "multi_thread")] +async fn test_otlp_metrics_batching_preserves_disabled_auto_create_policy() { + common_telemetry::init_default_ut_logging(); + let (app, mut guard) = setup_test_otlp_metrics_app_with_frontend_batched_and_auto_create_table( + StorageType::File, + "test_otlp_metrics_batching_preserves_disabled_auto_create_policy", + false, + ) + .await; + let client = TestClient::new(app).await; + + let missing_response = send_otlp_metrics_json( + &client, + r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"missing_metric","gauge":{"dataPoints":[{"timeUnixNano":"1000000","asDouble":1.0}]}}]}]}]}"#, + ) + .await; + assert_eq!(StatusCode::BAD_REQUEST, missing_response.status()); + validate_data( + "otlp_batched_auto_create_disabled_missing_tables", + &client, + "SELECT count(*) FROM information_schema.tables WHERE table_schema = 'public' AND table_name IN ('missing_metric', 'greptime_physical_table')", + "[[0]]", + ) + .await; + + for sql in [ + r#"CREATE TABLE greptime_physical_table (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("physical_metric_table" = "")"#, + r#"CREATE TABLE existing_metric (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("on_physical_table" = "greptime_physical_table", "otlp_metric_compat" = "prom")"#, + ] { + let response = execute_sql(&client, sql).await; + assert_eq!( + StatusCode::OK, + response.status(), + "{}", + response.text().await + ); + } + + let existing_response = send_otlp_metrics_json( + &client, + r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"existing_metric","gauge":{"dataPoints":[{"timeUnixNano":"2000000","asDouble":2.0}]}}]}]}]}"#, + ) + .await; + assert_eq!(StatusCode::OK, existing_response.status()); + + let alter_response = send_otlp_metrics_json( + &client, + r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"existing_metric","gauge":{"dataPoints":[{"attributes":[{"key":"host","value":{"stringValue":"frontend"}}],"timeUnixNano":"3000000","asDouble":3.0}]}}]}]}]}"#, + ) + .await; + assert_eq!(StatusCode::BAD_REQUEST, alter_response.status()); + validate_data( + "otlp_batched_auto_create_disabled_existing_rows", + &client, + "SELECT greptime_value FROM existing_metric ORDER BY greptime_timestamp", + "[[2.0]]", + ) + .await; + validate_data( + "otlp_batched_auto_create_disabled_schema", + &client, + "SELECT count(*) FROM information_schema.columns WHERE table_schema = 'public' AND table_name = 'existing_metric' AND column_name = 'host'", + "[[0]]", + ) + .await; + + guard.remove_all().await; +} + #[tokio::test(flavor = "multi_thread")] async fn test_otlp_metrics_batching_preserves_fanout_semantic_options() { common_telemetry::init_default_ut_logging();