fix: preserve OTLP metric table placement

Fall back to the Inserter when an existing OTLP metric table belongs to a different physical table. Cover the custom physical-table placement path with an HTTP integration regression test.

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

src/frontend/src/instance/otlp.rs

@@ -46,6 +46,7 @@ use servers::query_handler::{
 };
 use session::context::QueryContextRef;
 use snafu::ResultExt;
+use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME};
 use table::requests::{
     OTLP_METRIC_COMPAT_KEY, OTLP_METRIC_COMPAT_PROM, SEMANTIC_PER_TABLE_INDEX_KEY,
     SEMANTIC_SIGNAL_TYPE, SEMANTIC_SOURCE, SIGNAL_TYPE_LOG, SIGNAL_TYPE_METRIC,
@@ -56,6 +57,64 @@ use self::trace_ingest::trace_conventions;
 use crate::instance::Instance;
 use crate::metrics::{OTLP_LOGS_ROWS, OTLP_METRICS_ROWS, OTLP_RESOURCE_INFO_WRITE_ERRORS};
 
+impl Instance {
+    async fn metric_table_matches_physical_table(
+        &self,
+        table_name: &str,
+        physical_table: &str,
+        ctx: &QueryContextRef,
+    ) -> ServerResult<bool> {
+        let table = self
+            .catalog_manager()
+            .table(
+                ctx.current_catalog(),
+                &ctx.current_schema(),
+                table_name,
+                Some(ctx.as_ref()),
+            )
+            .await
+            .map_err(BoxedError::new)
+            .context(error::ExecuteGrpcQuerySnafu)?;
+        Ok(table.is_none_or(|table| {
+            metric_table_info_matches_physical_table(&table.table_info(), physical_table)
+        }))
+    }
+
+    async fn existing_metric_tables_match_physical_table(
+        &self,
+        requests: &api::v1::RowInsertRequests,
+        ctx: &QueryContextRef,
+    ) -> ServerResult<bool> {
+        let physical_table = ctx
+            .extension(PHYSICAL_TABLE_PARAM)
+            .unwrap_or(GREPTIME_PHYSICAL_TABLE);
+
+        for request in &requests.inserts {
+            if !self
+                .metric_table_matches_physical_table(&request.table_name, physical_table, ctx)
+                .await?
+            {
+                return Ok(false);
+            }
+        }
+
+        Ok(true)
+    }
+}
+
+fn metric_table_info_matches_physical_table(
+    table_info: &table::metadata::TableInfo,
+    physical_table: &str,
+) -> bool {
+    table_info.meta.engine == METRIC_ENGINE_NAME
+        && table_info
+            .meta
+            .options
+            .extra_options
+            .get(LOGICAL_TABLE_METADATA_KEY)
+            .is_some_and(|table| table == physical_table)
+}
+
 fn trace_permission_targets(
     table_name: &str,
     groups: &[TraceSpanGroup],
@@ -167,7 +226,11 @@ impl OpenTelemetryProtocolHandler for Instance {
 
         let batching_candidate =
             !metric_ctx.is_legacy && metric_ctx.with_metric_engine && metric_row_batcher.is_some();
-        let batchable = batching_candidate && is_scalar_metric_batchable(&requests)?;
+        let batchable = batching_candidate
+            && is_scalar_metric_batchable(&requests)?
+            && self
+                .existing_metric_tables_match_physical_table(&requests, &ctx)
+                .await?;
         let write_cost =
             if let Some(metric_row_batcher) = metric_row_batcher.as_ref().filter(|_| batchable) {
                 metric_row_batcher

tests-integration/tests/http.rs

@@ -7042,6 +7042,61 @@ async fn test_otlp_metrics_batching_flushes_multiple_small_requests() {
     guard.remove_all().await;
 }
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_otlp_metrics_batching_respects_existing_logical_table_placement() {
+    common_telemetry::init_default_ut_logging();
+    let (app, mut guard) = setup_test_otlp_metrics_app_with_frontend_batched(
+        StorageType::File,
+        "test_otlp_metrics_batching_respects_existing_logical_table_placement",
+    )
+    .await;
+    let client = TestClient::new(app).await;
+
+    let response = send_otlp_metrics_json(
+        &client,
+        r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"default_seed","gauge":{"dataPoints":[{"timeUnixNano":"1000000","asDouble":1.0}]}}]}]}]}"#,
+    )
+    .await;
+    assert_eq!(StatusCode::OK, response.status());
+
+    for sql in [
+        r#"CREATE TABLE otlp_custom_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("physical_metric_table" = "")"#,
+        r#"CREATE TABLE placement_metric (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ("on_physical_table" = "otlp_custom_physical", "otlp_metric_compat" = "prom")"#,
+    ] {
+        let response = execute_sql(&client, sql).await;
+        assert_eq!(
+            StatusCode::OK,
+            response.status(),
+            "{}",
+            response.text().await
+        );
+    }
+
+    let response = send_otlp_metrics_json(
+        &client,
+        r#"{"resourceMetrics":[{"scopeMetrics":[{"metrics":[{"name":"placement_metric","gauge":{"dataPoints":[{"timeUnixNano":"2000000","asDouble":2.0}]}}]}]}]}"#,
+    )
+    .await;
+    assert_eq!(StatusCode::OK, response.status());
+
+    validate_data(
+        "otlp_batched_custom_physical_table",
+        &client,
+        "SELECT greptime_value FROM placement_metric",
+        "[[2.0]]",
+    )
+    .await;
+    validate_data(
+        "otlp_batched_default_physical_table_baseline",
+        &client,
+        "SELECT greptime_value FROM default_seed",
+        "[[1.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();