fix: enforce auto-create policy in metric batching

Reuse the Inserter auto-create policy before pending-row batch DDL. Keep existing-schema writes working while rejecting disabled table creation and schema evolution.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

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<AlterTableExpr> = tables
             .iter()

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 {

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;

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();