diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs index eb1dda480db..2f21cc88259 100644 --- a/src/frontend/src/instance/otlp.rs +++ b/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 { + 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 { + 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 diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 6d2ec9c4807..0581fc04b40 100644 --- a/tests-integration/tests/http.rs +++ b/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();