feat(otlp): store validated metric exemplars (#9923)

* feat(otlp): persist metric exemplars

OTLP metric conversion now emits exemplars through the shared storage contract. It validates timestamps, IDs, values and label budgets, and associates histogram exemplars only with emitted buckets that contain their values. Invalid observations are dropped and counted while valid points and siblings remain ingestible.

Frontend ingestion authorizes the emitted destinations and retains confirmed completions across metric and exemplar writes, including logical-table batching. Includes conversion and schema-failure isolation tests; tests that also query the new Prometheus endpoint land in the final layer.

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* feat(otlp): bind exemplars to source metric identities

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix(otlp): bound exemplar memory and preserve series

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix(otlp): skip exemplars without metric engine

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix(otlp): skip exemplars for ordinary tables

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* refactor(exemplar)!: prefix the shared table name

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix(otlp): avoid waiting with body permits held

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

---------

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

src/common/catalog/src/consts.rs

@@ -224,6 +224,8 @@ pub const SPAN_KIND_SERVER: &str = "SPAN_KIND_SERVER";
 pub const SPAN_STATUS_ERROR: &str = "STATUS_CODE_ERROR";
 pub const TRACE_TABLE_NAME: &str = "opentelemetry_traces";
 pub const TRACE_TABLE_NAME_SESSION_KEY: &str = "trace_table_name";
+/// Shared exemplar destination within each source metric's schema.
+pub const METRICS_EXEMPLARS_TABLE: &str = concatcp!(DEFAULT_CATALOG_NAME, "_metrics_exemplars");
 // ---- End of special table and fields ----
 
 /// Generate the trace services table name from the trace table name by adding `_services` suffix.

src/common/query/src/prelude.rs

@@ -91,8 +91,6 @@ pub fn greptime_native_histogram() -> &'static str {
 const GREPTIME_TIMESTAMP: &str = "greptime_timestamp";
 /// Default value column name constant for backward compatibility.
 const GREPTIME_VALUE: &str = "greptime_value";
-/// Shared exemplar destination within each source metric's schema.
-pub const METRICS_EXEMPLARS_TABLE: &str = "metrics_exemplars";
 /// Query-visible source metric name stored in the exemplar table.
 pub const EXEMPLAR_METRIC_COLUMN: &str = "greptime_metric";
 /// JSON2 object containing exemplar-owned labels.

src/frontend/AGENTS.md

@@ -74,8 +74,14 @@ including caches owned by pipeline and protocol handlers.
   These tables are ingestion-managed; manual mutation requires restarting
   serving frontends.
 
-- **OTLP metrics** (`instance/otlp.rs`): authorize all emitted destinations before
-  schema preflight can reject points or expose destination schemas.
+- **OTLP metrics** (`instance/otlp.rs`): authorize source tables before schema
+  preflight can reject points or expose their schemas. Skip and count exemplars
+  when metric-engine ingestion is disabled or the source is an ordinary table,
+  retaining metric rows. Authorize retained exemplar writes before accessing
+  the exemplar destination or reserving expansion memory.
+  Reserve exemplar expansion with the transport's memory limiter before building
+  rows; retain the permit through `WriteProgress` until writes complete.
+  Expansion must not wait while the request holds transport memory permits.
 
 - **Logical-table batching** (`instance/logical_batcher.rs`): `Services` initializes
   one shared batcher for opted-in HTTP Prom and OTLP metric-engine

src/frontend/src/instance/otlp.rs

@@ -20,16 +20,18 @@ use std::collections::HashSet;
 use std::sync::Arc;
 
 use api::v1::RowInsertRequests;
+use api::v1::value::ValueData;
 use async_trait::async_trait;
 use auth::{
     OTLP_WRITE, PermissionChecker, PermissionCheckerRef, PermissionReq, PermissionTableTarget,
     PermissionTableTargets,
 };
 use client::Output;
-use common_catalog::consts::{trace_operations_table_name, trace_services_table_name};
+use common_catalog::consts::{
+    METRICS_EXEMPLARS_TABLE, trace_operations_table_name, trace_services_table_name,
+};
 use common_error::ext::BoxedError;
-use common_query::prelude::GREPTIME_PHYSICAL_TABLE;
-use common_query::{OutputData, OutputMeta};
+use common_query::prelude::{EXEMPLAR_METRIC_COLUMN, GREPTIME_PHYSICAL_TABLE};
 use common_telemetry::{tracing, warn};
 use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
 use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
@@ -44,9 +46,11 @@ use servers::otlp::trace::span::TraceSpanGroup;
 use servers::query_handler::{
     MetricsIngestOutcome, OpenTelemetryProtocolHandler, PipelineHandlerRef, TraceIngestOutcome,
 };
+use servers::request_memory_limiter::ServerMemoryLimiter;
 use session::context::QueryContextRef;
 use session::protocol_ctx::OtlpMetricCtx;
-use snafu::ResultExt;
+use session::write_progress::WriteProgress;
+use snafu::{IntoError, ResultExt};
 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,
@@ -96,6 +100,7 @@ impl OpenTelemetryProtocolHandler for Instance {
         &self,
         request: ExportMetricsServiceRequest,
         ctx: QueryContextRef,
+        memory_limiter: Option<(ServerMemoryLimiter, u64)>,
     ) -> ServerResult<MetricsIngestOutcome> {
         self.plugins
             .get::<PermissionCheckerRef>()
@@ -117,7 +122,7 @@ impl OpenTelemetryProtocolHandler for Instance {
         metric_ctx.resource_info = self.otlp_resource_info;
 
         let mut prepared = otlp::metrics::prepare_metrics(request, &mut metric_ctx)?;
-        // Check all destinations before preflight can expose schemas or discard points.
+        // Authorize source tables before preflight can expose schemas or discard points.
         let targets = {
             let catalog = ctx.current_catalog();
             let schema = ctx.current_schema();
@@ -127,7 +132,7 @@ impl OpenTelemetryProtocolHandler for Instance {
                 .iter()
                 .filter(|request| table_names.insert(request.table_name.as_str()))
                 .map(|request| PermissionTableTarget::new(catalog, &schema, &request.table_name))
-                .collect()
+                .collect::<Vec<_>>()
         };
         self.check_table_permission(
             &ctx,
@@ -152,6 +157,7 @@ impl OpenTelemetryProtocolHandler for Instance {
         let otlp::metrics::MetricsConversion {
             requests,
             rows,
+            mut exemplars,
             semantic_index,
             resource_info,
             mut outcome,
@@ -167,7 +173,73 @@ impl OpenTelemetryProtocolHandler for Instance {
             return Ok(outcome);
         }
 
-        let ctx = admit_write(rows as u64, &ctx)
+        // The engine option only controls new tables; existing ordinary sources cannot bind exemplars.
+        let before = exemplars.len();
+        exemplars.retain(|exemplar| {
+            preflight
+                .physical_tables
+                .contains_key(exemplar.metric.as_ref())
+        });
+        servers::metrics::EXEMPLARS_DROPPED_TOTAL
+            .with_label_values(&["otlp"])
+            .inc_by((before - exemplars.len()) as u64);
+        if !exemplars.is_empty() {
+            self.check_table_permission(
+                &ctx,
+                PermissionReq::Action(OTLP_WRITE),
+                PermissionTableTargets::resolved(vec![PermissionTableTarget::new(
+                    ctx.current_catalog(),
+                    ctx.current_schema(),
+                    METRICS_EXEMPLARS_TABLE,
+                )]),
+            )
+            .context(AuthSnafu)?;
+        }
+
+        let ctx = if !exemplars.is_empty()
+            && let Some((limiter, held_bytes)) = memory_limiter
+        {
+            let bytes =
+                servers::exemplar::estimated_row_bytes(std::iter::once(exemplars.as_slice()))
+                    .and_then(|bytes| u64::try_from(bytes).ok())
+                    .ok_or_else(|| {
+                        error::MemoryLimitExceededSnafu.into_error(
+                            common_memory_manager::Error::MemoryLimitExceeded {
+                                requested_bytes: u64::MAX,
+                                limit_bytes: limiter.limit_bytes(),
+                            },
+                        )
+                    })?;
+            limiter.check_request_fits(bytes, held_bytes)?;
+            // Holding body permits while waiting for expansion can stall concurrent requests.
+            let reservation = Arc::new(if held_bytes > 0 {
+                limiter.try_acquire(bytes).ok_or_else(|| {
+                    error::MemoryLimitExceededSnafu.into_error(
+                        common_memory_manager::Error::MemoryLimitExceeded {
+                            requested_bytes: bytes,
+                            limit_bytes: limiter.limit_bytes(),
+                        },
+                    )
+                })?
+            } else {
+                limiter.acquire(bytes).await?
+            });
+            let mut ctx = ctx.fork();
+            // Queued and detached writes retain the reservation after caller cancellation.
+            ctx.set_write_progress(Arc::new(WriteProgress::with_reservation(reservation)));
+            Arc::new(ctx)
+        } else {
+            ctx
+        };
+        let exemplar_rows = exemplars.len();
+        let (exemplars, _) = servers::exemplar::into_rows(exemplars)?;
+        let total_rows = rows.checked_add(exemplar_rows).ok_or_else(|| {
+            error::InvalidOtlpMetricInputSnafu {
+                reason: "combined metric and exemplar row count overflow",
+            }
+            .build()
+        })?;
+        let ctx = admit_write(total_rows as u64, &ctx)
             .await
             .map_err(BoxedError::new)
             .context(error::ExecuteGrpcQuerySnafu)?;
@@ -187,10 +259,107 @@ impl OpenTelemetryProtocolHandler for Instance {
             Arc::new(c)
         };
 
-        let output = self
-            .handle_otlp_metric_row_inserts(requests, ctx.clone(), &metric_ctx)
-            .await?;
-        outcome.write_cost = output.meta.cost;
+        if exemplar_rows > 0 {
+            let confirmed = Arc::new(ctx.with_confirmed_writes());
+            let source_names = exemplars
+                .inserts
+                .iter()
+                .filter_map(|request| request.rows.as_ref())
+                .flat_map(|rows| {
+                    let index = rows
+                        .schema
+                        .iter()
+                        .position(|c| c.column_name == EXEMPLAR_METRIC_COLUMN);
+                    rows.rows.iter().filter_map(move |row| {
+                        match &row.values.get(index?)?.value_data {
+                            Some(ValueData::StringValue(name)) => Some(name.as_str()),
+                            _ => None,
+                        }
+                    })
+                })
+                .collect::<HashSet<_>>();
+            let (source_requests, other): (Vec<_>, Vec<_>) = requests
+                .inserts
+                .into_iter()
+                .partition(|request| source_names.contains(request.table_name.as_str()));
+            let prepared = self
+                .inserter
+                .prepare_metric_row_inserts(
+                    RowInsertRequests {
+                        inserts: source_requests,
+                    },
+                    confirmed.clone(),
+                    &self.statement_executor,
+                    ctx.extension(PHYSICAL_TABLE_PARAM)
+                        .unwrap_or(GREPTIME_PHYSICAL_TABLE)
+                        .to_string(),
+                )
+                .await
+                .map_err(BoxedError::new)
+                .context(error::ExecuteGrpcQuerySnafu);
+            let source_ready = prepared.is_ok();
+            let sources = prepared
+                .as_ref()
+                .map(|request| request.source_tables().cloned().collect::<Vec<_>>())
+                .unwrap_or_default();
+            let (metric_output, other_output, exemplar_output) = tokio::join!(
+                async {
+                    self.inserter
+                        .handle_prepared_metric_inserts(prepared?, confirmed.clone())
+                        .await
+                        .map_err(BoxedError::new)
+                        .context(error::ExecuteGrpcQuerySnafu)
+                },
+                async {
+                    if other.is_empty() {
+                        Ok(Output::new_with_affected_rows(0))
+                    } else {
+                        self.handle_otlp_metric_row_inserts(
+                            RowInsertRequests { inserts: other },
+                            confirmed.clone(),
+                            &metric_ctx,
+                        )
+                        .await
+                    }
+                },
+                async {
+                    // Failed source preparation must not bind exemplars to a later replacement table.
+                    if !source_ready {
+                        return Ok(Output::new_with_affected_rows(0));
+                    }
+                    self.inserter
+                        .handle_exemplar_inserts(
+                            exemplars,
+                            confirmed.clone(),
+                            &self.statement_executor,
+                            &sources,
+                        )
+                        .await
+                        .map_err(BoxedError::new)
+                        .context(error::ExecuteGrpcQuerySnafu)
+                },
+            );
+            let mut progress = common_query::WriteProgress::default();
+            for output in [metric_output, other_output, exemplar_output] {
+                match output {
+                    Ok(output) => {
+                        outcome.write_cost += output.meta.cost;
+                        progress.merge(output.write_progress().await);
+                    }
+                    Err(error) => {
+                        progress.merge(common_query::WriteProgress::failed(BoxedError::new(error)))
+                    }
+                }
+            }
+            progress
+                .into_result()
+                .context(error::ExecuteGrpcQuerySnafu)?;
+        } else {
+            let output = self
+                .handle_otlp_metric_row_inserts(requests, ctx.clone(), &metric_ctx)
+                .await?;
+            outcome.write_cost = output.meta.cost;
+        }
 
         // Derived enrichment follows the accepted metric submission, which may
         // still be queued in asynchronous mode. Failures remain warning-only
@@ -369,8 +538,8 @@ impl Instance {
 
         // OTLP tables have one sample field in both the ordinary and physical paths.
         if let Some(batcher) = batcher {
-            let (rows, cost) = batcher
-                .submit_with(requests, ctx.clone(), |mut requests| {
+            let (mut output, cost) = batcher
+                .submit_output_with(requests, ctx.clone(), |mut requests| {
                     let ctx = ctx.clone();
                     async move {
                         Inserter::meter_row_inserts(&mut requests, &ctx)
@@ -380,10 +549,8 @@ impl Instance {
                     }
                 })
                 .await?;
-            Ok(Output::new(
-                OutputData::AffectedRows(rows as usize),
-                OutputMeta::new_with_cost(cost as _),
-            ))
+            output.meta.cost = cost as _;
+            Ok(output)
         } else if !metric_ctx.with_metric_engine {
             self.handle_row_inserts(requests, ctx.clone(), false, true)
                 .await

src/frontend/src/server.rs

@@ -432,6 +432,7 @@ where
                 )
                 .with_influxdb_handler(instance.clone())
                 .with_otlp_handler(instance.clone(), with_metric_engine)
+                .with_memory_limiter(self.server_memory_limiter.clone())
                 .with_batching_enabled(postgres_batching),
             ) as Box<dyn Server>;
 

src/operator/src/insert.rs

@@ -39,9 +39,10 @@ use catalog::CatalogManagerRef;
 use client::{OutputData, OutputMeta};
 use common_base::readable_size::ReadableSize;
 use common_catalog::consts::{
-    DEFAULT_PRIVATE_SCHEMA_NAME, PARENT_SPAN_ID_COLUMN, SERVICE_NAME_COLUMN, SPAN_ID_COLUMN,
-    TIMESTAMP_END_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,
+    DEFAULT_PRIVATE_SCHEMA_NAME, METRICS_EXEMPLARS_TABLE, PARENT_SPAN_ID_COLUMN,
+    SERVICE_NAME_COLUMN, SPAN_ID_COLUMN, TIMESTAMP_END_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;
@@ -54,8 +55,7 @@ use common_meta::rpc::ddl::TriggerReason;
 use common_query::Output;
 use common_query::native_histogram::{is_native_histogram_value_type, native_histogram_value_type};
 use common_query::prelude::{
-    EXEMPLAR_LABELS_COLUMN, EXEMPLAR_METRIC_COLUMN, METRICS_EXEMPLARS_TABLE, greptime_timestamp,
-    greptime_value,
+    EXEMPLAR_LABELS_COLUMN, EXEMPLAR_METRIC_COLUMN, greptime_timestamp, greptime_value,
 };
 use common_telemetry::tracing_context::TracingContext;
 use common_telemetry::{debug, error, warn};
@@ -2208,7 +2208,9 @@ pub fn validate_exemplar_table(table: &TableInfo) -> Result<()> {
                 .get(store_api::metric_engine_consts::PRIMARY_KEY_ENCODING)
                 .is_some_and(|v| v.eq_ignore_ascii_case("sparse")),
         InvalidInsertRequestSnafu {
-            reason: "metrics_exemplars requires Mito, append_mode=true and primary_key_encoding=sparse"
+            reason: format!(
+                "{METRICS_EXEMPLARS_TABLE} requires Mito, append_mode=true and primary_key_encoding=sparse"
+            )
         }
     );
     let columns = meta.schema.column_schemas();

src/operator/src/insert/otlp.rs

@@ -30,7 +30,9 @@ use common_time::timestamp::TimeUnit;
 use datatypes::schema::ColumnSchema as DataColumn;
 use session::context::QueryContextRef;
 use snafu::prelude::*;
-use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, PHYSICAL_TABLE_METADATA_KEY};
+use store_api::metric_engine_consts::{
+    LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY,
+};
 use table::TableRef;
 
 use crate::error::{self, InvalidInsertRequestSnafu, Result};
@@ -43,6 +45,7 @@ use crate::req_convert::insert::{ImpureDefaultFiller, value_matches_type};
 pub struct MetricInsertPreflight {
     /// Rejections in batch order, then row order.
     pub rejections: Vec<Vec<Option<String>>>,
+    /// Backing tables for existing or newly created logical metric-engine destinations.
     pub physical_tables: std::collections::HashMap<String, String>,
 }
 
@@ -91,7 +94,7 @@ impl Inserter {
                 tables.insert(request.table_name.clone(), table);
             }
             let backing = match tables[&request.table_name].as_ref() {
-                Some(table) => table
+                Some(table) if table.table_info().meta.engine == METRIC_ENGINE_NAME => table
                     .table_info()
                     .meta
                     .options
@@ -99,7 +102,7 @@ impl Inserter {
                     .get(LOGICAL_TABLE_METADATA_KEY)
                     .cloned(),
                 None if metric_engine => Some(physical_table.to_string()),
-                None => None,
+                _ => None,
             };
             if let Some(backing) = backing {
                 physical_tables.insert(request.table_name.clone(), backing);

src/servers/src/exemplar.rs

@@ -19,10 +19,10 @@ use std::sync::Arc;
 
 use api::v1::value::ValueData;
 use api::v1::{ColumnDataType, Row, RowInsertRequests, Value};
+use common_catalog::consts::METRICS_EXEMPLARS_TABLE;
 use common_grpc::precision::Precision;
 use common_query::prelude::{
-    EXEMPLAR_LABELS_COLUMN, EXEMPLAR_METRIC_COLUMN, METRICS_EXEMPLARS_TABLE, greptime_timestamp,
-    greptime_value,
+    EXEMPLAR_LABELS_COLUMN, EXEMPLAR_METRIC_COLUMN, greptime_timestamp, greptime_value,
 };
 use datatypes::json;
 use metric_engine::row_modifier::TsidGenerator;
@@ -50,8 +50,10 @@ pub struct Exemplar {
 pub type ExemplarRequests = BTreeMap<Option<String>, Vec<Exemplar>>;
 
 /// Bounds row capacity after each database's metrics share one exemplar schema.
-pub(crate) fn estimated_row_bytes(requests: &ExemplarRequests) -> Option<usize> {
-    requests.values().try_fold(0usize, |total, exemplars| {
+pub fn estimated_row_bytes<'a>(
+    requests: impl IntoIterator<Item = &'a [Exemplar]>,
+) -> Option<usize> {
+    requests.into_iter().try_fold(0usize, |total, exemplars| {
         let mut tags = BTreeSet::new();
         let mut payload = 0usize;
         // Shared series data is copied into each output row.

src/servers/src/grpc/memory_limit.rs

@@ -63,6 +63,12 @@ pub(crate) struct PreDecodeMemoryReservation {
     _guard: Arc<MemoryGuard<RequestMemoryMetrics>>,
 }
 
+impl PreDecodeMemoryReservation {
+    pub(crate) fn reserved_bytes(&self) -> u64 {
+        self._guard.granted_bytes()
+    }
+}
+
 #[derive(Clone)]
 pub struct MemoryLimiterExtensionLayer {
     limiter: ServerMemoryLimiter,

src/servers/src/http/memory_limit.rs

@@ -155,7 +155,12 @@ pub async fn decoded_body_accounting_middleware(
         .and_then(|v| v.parse::<u64>().ok())
         .unwrap_or(0);
 
-    let accounting = BodyMemoryAccounting::default();
+    // Share the wire-byte ledger so handlers see all permits held by this request.
+    let accounting = req
+        .extensions()
+        .get::<BodyMemoryAccounting>()
+        .cloned()
+        .unwrap_or_default();
     // Retain the permits for the whole request, like the global middleware.
     let retained_accounting = accounting.clone();
     let (mut parts, body) = req.into_parts();

src/servers/src/http/otlp.rs

@@ -45,10 +45,12 @@ use crate::http::extractor::{
     LogTableName, OtlpMetricOptions, PipelineInfo, SelectInfoWrapper, TraceTableName,
 };
 use crate::http::header::{CONTENT_TYPE_PROTOBUF, write_cost_header_map};
+use crate::http::memory_limit::BodyMemoryAccounting;
 use crate::metrics::METRIC_HTTP_OPENTELEMETRY_LOGS_ELAPSED;
 use crate::query_handler::{
     MetricsIngestOutcome, OpenTelemetryProtocolHandlerRef, PipelineHandler, TraceIngestOutcome,
 };
+use crate::request_memory_limiter::ServerMemoryLimiter;
 
 #[derive(Clone, prost::Message)]
 pub struct GoogleRpcStatus {
@@ -87,6 +89,7 @@ pub async fn metrics(
     Extension(mut query_ctx): Extension<QueryContext>,
     http_opts: OtlpMetricOptions,
     content_type: Option<TypedHeader<ContentType>>,
+    extensions: http::Extensions,
     bytes: Bytes,
 ) -> Result<OtlpMetricsResponse> {
     if is_json_content_type(content_type.as_ref().map(|h| &h.0)) {
@@ -122,7 +125,16 @@ pub async fn metrics(
     }));
     let query_ctx = Arc::new(query_ctx);
 
-    match handler.metrics(request, query_ctx).await {
+    let memory_limiter = extensions
+        .get::<ServerMemoryLimiter>()
+        .cloned()
+        .map(|limiter| {
+            let held_bytes = extensions
+                .get::<BodyMemoryAccounting>()
+                .map_or(0, BodyMemoryAccounting::reserved_bytes);
+            (limiter, held_bytes)
+        });
+    match handler.metrics(request, query_ctx, memory_limiter).await {
         Ok(outcome) => {
             if outcome.accepted_data_points == 0 && outcome.rejected_data_points > 0 {
                 Ok(OtlpMetricsResponse::Failure(outcome))

src/servers/src/http/prom_store.rs

@@ -235,7 +235,7 @@ async fn remote_write_v1(
         query_ctx
     } else {
         // Reserve for row expansion before copying shared metric names and series labels.
-        let bytes = crate::exemplar::estimated_row_bytes(&exemplars)
+        let bytes = crate::exemplar::estimated_row_bytes(exemplars.values().map(Vec::as_slice))
             .and_then(|bytes| u64::try_from(bytes).ok())
             .ok_or_else(|| {
                 error::MemoryLimitExceededSnafu.into_error(

src/servers/src/metrics.rs

@@ -130,7 +130,7 @@ lazy_static! {
     ).unwrap();
     pub static ref EXEMPLARS_DROPPED_TOTAL: IntCounterVec = register_int_counter_vec!(
         "greptime_servers_exemplars_dropped_total",
-        "Exemplars dropped during input validation",
+        "Exemplars dropped during ingestion",
         &["protocol"]
     ).unwrap();
     /// The native histograms count of Prometheus remote write per protocol version.

src/servers/src/otel_arrow.rs

@@ -33,7 +33,9 @@ use tonic::{Request, Response, Status, Streaming};
 
 use crate::error;
 use crate::grpc::context_auth;
+use crate::grpc::memory_limit::PreDecodeMemoryReservation;
 use crate::query_handler::{MetricsIngestOutcome, OpenTelemetryProtocolHandlerRef};
+use crate::request_memory_limiter::ServerMemoryLimiter;
 
 const EXPONENTIAL_HISTOGRAM_UNSUPPORTED: &str = "OTel Arrow exponential histograms are unsupported because the Arrow wire format omits zero_threshold";
 
@@ -120,6 +122,8 @@ impl ArrowMetricsService for OtelArrowServiceHandler<OpenTelemetryProtocolHandle
         };
 
         let handler = self.handler.clone();
+        let memory_limiter = extensions.get::<ServerMemoryLimiter>().cloned();
+        let decode_reservation = extensions.get::<PreDecodeMemoryReservation>().cloned();
 
         // handles incoming requests
         common_runtime::spawn_global(async move {
@@ -159,7 +163,16 @@ impl ArrowMetricsService for OtelArrowServiceHandler<OpenTelemetryProtocolHandle
                 };
                 let has_exponential_histogram_data_points =
                     remove_exponential_histograms(&mut request);
-                let outcome = match handler.metrics(request, query_ctx.clone()).await {
+                let memory_limiter = memory_limiter.clone().map(|limiter| {
+                    let held_bytes = decode_reservation
+                        .as_ref()
+                        .map_or(0, PreDecodeMemoryReservation::reserved_bytes);
+                    (limiter, held_bytes)
+                });
+                let outcome = match handler
+                    .metrics(request, query_ctx.clone(), memory_limiter)
+                    .await
+                {
                     Ok(outcome) => outcome,
                     Err(error::Error::InvalidOtlpMetricInput { reason }) => {
                         let _ = sender

src/servers/src/otlp/metrics.rs

@@ -51,6 +51,7 @@ use crate::semantic::{
     METRIC_TYPE_UPDOWN_COUNTER,
 };
 
+mod exemplars;
 mod prepare;
 mod resource_info;
 
@@ -110,6 +111,8 @@ const MAX_REJECTION_MESSAGE_BYTES: usize = 512;
 /// Result of converting one OTLP metrics request.
 #[derive(Debug)]
 pub struct MetricsConversion {
+    /// Kept normalized until the frontend reserves memory for row expansion.
+    pub exemplars: Vec<crate::exemplar::Exemplar>,
     pub requests: RowInsertRequests,
     /// Row count of `requests`; the resource descriptor is not counted.
     pub rows: usize,
@@ -391,6 +394,7 @@ fn process_scope_attrs(scope: &ScopeMetrics, metric_ctx: &OtlpMetricCtx) -> Opti
     })
 }
 
+#[allow(clippy::too_many_arguments)]
 fn encode_metrics(
     table_writer: &mut MultiTableData,
     metric: &Metric,
@@ -399,6 +403,7 @@ fn encode_metrics(
     scope_attrs: Option<&Vec<KeyValue>>,
     metric_ctx: &OtlpMetricCtx,
     outcome: &mut MetricsIngestOutcome,
+    exemplars: &mut Vec<exemplars::RowExemplars>,
 ) -> Result<()> {
     let name = &metadata.name;
 
@@ -412,6 +417,7 @@ fn encode_metrics(
                     resource_attrs,
                     scope_attrs,
                     metric_ctx,
+                    exemplars,
                 )?;
                 add_accepted_data_points(outcome, gauge.data_points.len())?;
             }
@@ -423,6 +429,7 @@ fn encode_metrics(
                     resource_attrs,
                     scope_attrs,
                     metric_ctx,
+                    exemplars,
                 )?;
                 add_accepted_data_points(outcome, sum.data_points.len())?;
             }
@@ -445,6 +452,7 @@ fn encode_metrics(
                 scope_attrs,
                 metric_ctx,
                 outcome,
+                exemplars,
             )?,
             metric::Data::ExponentialHistogram(hist) => encode_exponential_histogram(
                 table_writer,
@@ -454,6 +462,7 @@ fn encode_metrics(
                 scope_attrs,
                 metric_ctx,
                 outcome,
+                exemplars,
             )?,
         }
     }
@@ -538,6 +547,7 @@ fn append_rejection_message(message: &mut Option<String>, reason: impl FnOnce()
     }
 }
 
+#[allow(clippy::too_many_arguments)]
 fn encode_exponential_histogram(
     table_writer: &mut MultiTableData,
     metadata: &MetricMetadata,
@@ -546,6 +556,7 @@ fn encode_exponential_histogram(
     scope_attrs: Option<&Vec<KeyValue>>,
     metric_ctx: &OtlpMetricCtx,
     outcome: &mut MetricsIngestOutcome,
+    exemplars: &mut Vec<exemplars::RowExemplars>,
 ) -> Result<()> {
     let name = &metadata.name;
     let temporality = match exponential_histogram_gate(histogram) {
@@ -606,6 +617,13 @@ fn encode_exponential_histogram(
             std::iter::once((column_schema.clone(), Some(value))),
             &mut row,
         )?;
+        exemplars::append(
+            exemplars,
+            &data_point.exemplars,
+            name,
+            table.num_rows(),
+            metric_ctx,
+        );
         table.add_row(row);
         add_accepted_data_points(outcome, 1)?;
     }
@@ -898,11 +916,15 @@ fn write_attributes(
                 )
             }
         };
-        if key == OTLP_AGGREGATION_TEMPORALITY_LABEL {
+        // OTLP attribute keys may include __name__, but Prometheus metric identity comes from
+        // the OTLP metric name. Reject collisions after translation and scope prefixing.
+        // https://opentelemetry.io/docs/specs/otel/compatibility/prometheus_and_openmetrics/#metric-metadata-1
+        if key == OTLP_AGGREGATION_TEMPORALITY_LABEL || key == crate::prom_store::METRIC_NAME_LABEL
+        {
             return Err(error::InvalidOtlpMetricInputSnafu {
                 reason: format!(
                     "OTLP attribute `{}` resolves to reserved label `{}`",
-                    attr.key, OTLP_AGGREGATION_TEMPORALITY_LABEL
+                    attr.key, key
                 ),
             }
             .build());
@@ -1026,6 +1048,7 @@ fn encode_gauge(
     resource_attrs: Option<&Vec<KeyValue>>,
     scope_attrs: Option<&Vec<KeyValue>>,
     metric_ctx: &OtlpMetricCtx,
+    exemplars: &mut Vec<exemplars::RowExemplars>,
 ) -> Result<()> {
     let table = table_writer.get_or_default_table_data(
         name,
@@ -1046,6 +1069,13 @@ fn encode_gauge(
         )?;
 
         write_data_point_value(table, &mut row, greptime_value(), &data_point.value)?;
+        exemplars::append(
+            exemplars,
+            &data_point.exemplars,
+            name,
+            table.num_rows(),
+            metric_ctx,
+        );
         table.add_row(row);
     }
 
@@ -1061,6 +1091,7 @@ fn encode_sum(
     resource_attrs: Option<&Vec<KeyValue>>,
     scope_attrs: Option<&Vec<KeyValue>>,
     metric_ctx: &OtlpMetricCtx,
+    exemplars: &mut Vec<exemplars::RowExemplars>,
 ) -> Result<()> {
     let is_delta = matches!(
         AggregationTemporality::try_from(sum.aggregation_temporality),
@@ -1094,6 +1125,13 @@ fn encode_sum(
         } else {
             write_data_point_value(table, &mut row, greptime_value(), &data_point.value)?;
         }
+        exemplars::append(
+            exemplars,
+            &data_point.exemplars,
+            name,
+            table.num_rows(),
+            metric_ctx,
+        );
         table.add_row(row);
     }
 
@@ -1115,6 +1153,7 @@ const HISTOGRAM_LE_COLUMN: &str = "le";
 /// quantile functions on this table.
 /// Delta points retain their temporality tag and interval values. Bucket counts
 /// are prefix-summed within each point, never accumulated across timestamps.
+#[allow(clippy::too_many_arguments)]
 fn encode_histogram(
     table_writer: &mut MultiTableData,
     name: &str,
@@ -1123,6 +1162,7 @@ fn encode_histogram(
     scope_attrs: Option<&Vec<KeyValue>>,
     metric_ctx: &OtlpMetricCtx,
     outcome: &mut MetricsIngestOutcome,
+    exemplars: &mut Vec<exemplars::RowExemplars>,
 ) -> Result<()> {
     let normalized_name = name;
 
@@ -1185,7 +1225,28 @@ fn encode_histogram(
             }
             values
         };
-        for (bound, value) in bucket_values {
+        let mut by_bucket = vec![Vec::new(); bucket_values.len()];
+        for exemplar in &data_point.exemplars {
+            let value = match exemplars::value(exemplar) {
+                Ok(value) => value,
+                Err(_) => {
+                    crate::metrics::EXEMPLARS_DROPPED_TOTAL
+                        .with_label_values(&["otlp"])
+                        .inc();
+                    continue;
+                }
+            };
+            let Some(bucket) = bucket_values.iter().position(|(bound, _)| {
+                value <= *bound || (value.is_nan() && *bound == f64::INFINITY)
+            }) else {
+                crate::metrics::EXEMPLARS_DROPPED_TOTAL
+                    .with_label_values(&["otlp"])
+                    .inc();
+                continue;
+            };
+            by_bucket[bucket].push(exemplar);
+        }
+        for ((bound, value), bucket_exemplars) in bucket_values.into_iter().zip(by_bucket) {
             let mut bucket_row = bucket_table.alloc_one_row();
             write_tags_and_timestamp(
                 bucket_table,
@@ -1197,9 +1258,17 @@ fn encode_histogram(
                 metric_ctx,
             )?;
             write_temporality_tag(bucket_table, &mut bucket_row, is_delta)?;
+            // Preserve existing OTLP series identities, including the "inf" bucket.
             row_writer::write_tag(bucket_table, HISTOGRAM_LE_COLUMN, bound, &mut bucket_row)?;
             row_writer::write_f64(bucket_table, greptime_value(), value, &mut bucket_row)?;
 
+            exemplars::append(
+                exemplars,
+                bucket_exemplars,
+                &bucket_table_name,
+                bucket_table.num_rows(),
+                metric_ctx,
+            );
             bucket_table.add_row(bucket_row);
         }
 
@@ -1411,14 +1480,15 @@ fn encode_summary(
 mod tests {
     use api::v1::ColumnDataType;
     use common_query::prelude::set_default_prefix;
-    use otel_arrow_rust::proto::opentelemetry::common::v1::AnyValue;
     use otel_arrow_rust::proto::opentelemetry::common::v1::any_value::Value as Val;
+    use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, InstrumentationScope};
     use otel_arrow_rust::proto::opentelemetry::metrics::v1::number_data_point::Value;
     use otel_arrow_rust::proto::opentelemetry::metrics::v1::summary_data_point::ValueAtQuantile;
     use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
         AggregationTemporality, HistogramDataPoint, NumberDataPoint, SummaryDataPoint,
     };
     use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource;
+    use session::protocol_ctx::OtlpMetricTranslationStrategy;
 
     use super::*;
 
@@ -1536,8 +1606,19 @@ mod tests {
     }
 
     #[test]
-    fn test_descriptors_exclude_points_rejected_by_destination_preflight() {
+    fn test_descriptors_and_exemplars_exclude_points_rejected_by_destination_preflight() {
         let mut request = gauge_request(vec![keyvalue("service.name", "api")], "accepted");
+        let Some(metric::Data::Gauge(gauge)) = request.resource_metrics[0].scope_metrics[0].metrics
+            [0]
+        .data
+        .as_mut() else {
+            unreachable!()
+        };
+        gauge.data_points[0].exemplars.push(Exemplar {
+            time_unix_nano: 1,
+            value: Some(exemplar::Value::AsDouble(0.5)),
+            ..Default::default()
+        });
         let mut rejected = request.resource_metrics[0].clone();
         let metric = &mut rejected.scope_metrics[0].metrics[0];
         metric.name = "rejected".into();
@@ -1547,7 +1628,28 @@ mod tests {
         gauge.data_points[0].time_unix_nano =
             2 * common_catalog::consts::SEMANTIC_GRAPH_WINDOW_NANOS as u64;
         request.resource_metrics.push(rejected);
-        let prepared = prepare_metrics(request, &mut descriptor_ctx()).unwrap();
+        let mut prepared = prepare_metrics(
+            request,
+            &mut OtlpMetricCtx {
+                with_metric_engine: true,
+                ..descriptor_ctx()
+            },
+        )
+        .unwrap();
+        for request in &mut prepared.requests {
+            let rows = request.rows.as_mut().unwrap();
+            rows.schema.push(ColumnSchema {
+                column_name: "defaulted_host".into(),
+                datatype: api::v1::ColumnDataType::String as i32,
+                semantic_type: SemanticType::Tag as i32,
+                ..Default::default()
+            });
+            for row in &mut rows.rows {
+                row.values.push(api::v1::Value {
+                    value_data: Some(ValueData::StringValue("unknown".into())),
+                });
+            }
+        }
         let conversion = prepared
             .finish(
                 vec![None, Some("incompatible schema".into())],
@@ -1561,6 +1663,22 @@ mod tests {
             ),
             (1, 1)
         );
+        assert_eq!(conversion.exemplars.len(), 1);
+        assert_eq!(
+            conversion.exemplars[0].series_labels["defaulted_host"],
+            "unknown"
+        );
+        let (exemplars, _) = crate::exemplar::into_rows(conversion.exemplars).unwrap();
+        let exemplar_rows = exemplars.inserts[0].rows.as_ref().unwrap();
+        let metric = exemplar_rows
+            .schema
+            .iter()
+            .position(|c| c.column_name == common_query::prelude::EXEMPLAR_METRIC_COLUMN)
+            .unwrap();
+        assert_eq!(
+            exemplar_rows.rows[0].values[metric].value_data,
+            Some(ValueData::StringValue("accepted".into()))
+        );
         let descriptors = conversion.resource_info.unwrap();
         let rows = descriptors.inserts[0].rows.as_ref().unwrap();
         assert_eq!(rows.rows.len(), 1);
@@ -1750,6 +1868,85 @@ mod tests {
         assert_eq!(attr_value(&attrs, "instance").as_deref(), Some("inst-1"));
     }
 
+    #[test]
+    fn test_reject_metric_name_attribute() {
+        set_default_prefix(None).unwrap();
+        for (strategy, key) in [
+            (
+                OtlpMetricTranslationStrategy::UnderscoreEscapingWithSuffixes,
+                "__name__",
+            ),
+            (
+                OtlpMetricTranslationStrategy::UnderscoreEscapingWithoutSuffixes,
+                "__name__",
+            ),
+            (
+                OtlpMetricTranslationStrategy::NoUtf8EscapingWithSuffixes,
+                "__name__",
+            ),
+            (OtlpMetricTranslationStrategy::NoTranslation, "__name__"),
+            (
+                OtlpMetricTranslationStrategy::UnderscoreEscapingWithSuffixes,
+                "..name..",
+            ),
+        ] {
+            for source in [
+                AttributeType::Resource,
+                AttributeType::Scope,
+                AttributeType::DataPoint,
+            ] {
+                let mut request = gauge_request(vec![], "source_metric");
+                let resource = &mut request.resource_metrics[0];
+                let scope = &mut resource.scope_metrics[0];
+                let Some(metric::Data::Gauge(gauge)) = scope.metrics[0].data.as_mut() else {
+                    unreachable!();
+                };
+                gauge.data_points[0].exemplars.push(Exemplar {
+                    value: Some(exemplar::Value::AsDouble(0.25)),
+                    time_unix_nano: 1_000_000,
+                    ..Default::default()
+                });
+                let attributes = vec![keyvalue(key, "wrong_metric")];
+                match source {
+                    AttributeType::Resource => {
+                        resource.resource.as_mut().unwrap().attributes = attributes;
+                    }
+                    AttributeType::Scope => {
+                        scope.scope = Some(InstrumentationScope {
+                            attributes,
+                            ..Default::default()
+                        });
+                    }
+                    AttributeType::DataPoint => gauge.data_points[0].attributes = attributes,
+                }
+                let mut ctx = OtlpMetricCtx {
+                    with_metric_engine: true,
+                    metric_translation_strategy: strategy,
+                    promote_all_resource_attrs: true,
+                    promote_scope_attrs: true,
+                    ..Default::default()
+                };
+                let result = to_grpc_insert_requests(request, &mut ctx);
+                if source == AttributeType::Scope {
+                    let conversion = result.unwrap();
+                    assert_eq!(1, conversion.exemplars.len());
+                    assert!(
+                        column_names(&conversion.requests, "source_metric")
+                            .contains(&"otel_scope___name__".to_string())
+                    );
+                } else {
+                    let conversion = result.unwrap();
+                    assert_eq!(conversion.outcome.accepted_data_points, 0);
+                    assert_eq!(conversion.outcome.rejected_data_points, 1);
+                    assert_eq!(conversion.exemplars.len(), 0);
+                    assert!(conversion.outcome.error_message.unwrap().contains(&format!(
+                        "OTLP attribute `{key}` resolves to reserved label `__name__`"
+                    )));
+                }
+            }
+        }
+    }
+
     #[test]
     fn test_boolean_attributes_keep_series_distinct() {
         set_default_prefix(None).unwrap();
@@ -1834,6 +2031,7 @@ mod tests {
             Some(&vec![]),
             Some(&vec![keyvalue("scope", "otel")]),
             &OtlpMetricCtx::default(),
+            &mut Vec::new(),
         )
         .unwrap();
 
@@ -1884,6 +2082,7 @@ mod tests {
             Some(&vec![]),
             Some(&vec![keyvalue("scope", "otel")]),
             &OtlpMetricCtx::default(),
+            &mut Vec::new(),
         )
         .unwrap();
 
@@ -2020,6 +2219,7 @@ mod tests {
             Some(&vec![keyvalue("scope", "otel")]),
             &OtlpMetricCtx::default(),
             &mut outcome,
+            &mut Vec::new(),
         )
         .unwrap();
 
@@ -2643,6 +2843,247 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_invalid_exemplars_preserve_metric_payload() {
+        use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar};
+
+        let valid = Exemplar {
+            time_unix_nano: 1,
+            value: Some(exemplar::Value::AsDouble(0.5)),
+            ..Default::default()
+        };
+        for invalid in [
+            Exemplar::default(),
+            Exemplar {
+                trace_id: vec![1],
+                ..valid.clone()
+            },
+            Exemplar {
+                span_id: vec![1],
+                ..valid.clone()
+            },
+            Exemplar {
+                time_unix_nano: u64::MAX,
+                ..valid.clone()
+            },
+            Exemplar {
+                filtered_attributes: vec![keyvalue("http.url", &"x".repeat(128))],
+                ..valid.clone()
+            },
+            Exemplar {
+                filtered_attributes: vec![keyvalue("a", "1"), keyvalue("a", "2")],
+                ..valid.clone()
+            },
+        ] {
+            let exemplars = vec![invalid.clone(), valid.clone(), invalid, valid.clone()];
+            let point = NumberDataPoint {
+                time_unix_nano: 1,
+                value: Some(Value::AsDouble(1.0)),
+                exemplars: exemplars.clone(),
+                ..Default::default()
+            };
+            let metrics = vec![
+                Metric {
+                    name: "gauge".into(),
+                    data: Some(metric::Data::Gauge(Gauge {
+                        data_points: vec![point.clone()],
+                    })),
+                    ..Default::default()
+                },
+                Metric {
+                    name: "sum".into(),
+                    data: Some(metric::Data::Sum(Sum {
+                        data_points: vec![point],
+                        aggregation_temporality: AggregationTemporality::Cumulative as i32,
+                        ..Default::default()
+                    })),
+                    ..Default::default()
+                },
+                exponential_metric(
+                    "histogram",
+                    vec![ExponentialHistogramDataPoint {
+                        exemplars,
+                        ..exponential_point()
+                    }],
+                    AggregationTemporality::Cumulative,
+                ),
+            ];
+            let conversion = to_grpc_insert_requests(
+                metrics_request(metrics),
+                &mut OtlpMetricCtx {
+                    with_metric_engine: true,
+                    ..Default::default()
+                },
+            )
+            .unwrap();
+            assert_eq!(conversion.rows, 3);
+            assert_eq!(conversion.outcome.accepted_data_points, 3);
+            assert_eq!(conversion.outcome.rejected_data_points, 0);
+            assert_eq!(conversion.exemplars.len(), 6);
+        }
+    }
+
+    #[test]
+    fn test_histogram_exemplars_follow_emitted_buckets() {
+        use common_query::prelude::EXEMPLAR_METRIC_COLUMN;
+        use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar};
+        use prost::Message;
+
+        for mode in ["delta", "empty", "stale", "missing_infinity"] {
+            let mut point = HistogramDataPoint {
+                time_unix_nano: 100,
+                count: 4,
+                explicit_bounds: vec![1.0, 2.0],
+                bucket_counts: vec![1, 1, 2],
+                exemplars: [5.0, 1.0, 0.5, f64::NEG_INFINITY, f64::INFINITY, f64::NAN]
+                    .into_iter()
+                    .enumerate()
+                    .map(|(i, value)| Exemplar {
+                        time_unix_nano: i as u64 + 1,
+                        value: Some(exemplar::Value::AsDouble(value)),
+                        ..Default::default()
+                    })
+                    .collect(),
+                ..Default::default()
+            };
+            point.exemplars.extend([
+                Exemplar::default(),
+                Exemplar {
+                    trace_id: vec![1],
+                    value: Some(exemplar::Value::AsDouble(1.0)),
+                    ..Default::default()
+                },
+            ]);
+            match mode {
+                "empty" => {
+                    point.explicit_bounds.clear();
+                    point.bucket_counts.clear();
+                }
+                "stale" => point.flags = DataPointFlags::NoRecordedValueMask as u32,
+                "missing_infinity" => {
+                    // Cumulative conversion tolerates a missing infinity bucket.
+                    point.explicit_bounds = vec![1.0];
+                    point.bucket_counts = vec![1];
+                    point.count = 1;
+                }
+                _ => {}
+            }
+            let mut histogram = Histogram {
+                data_points: vec![point],
+                aggregation_temporality: if mode == "missing_infinity" {
+                    AggregationTemporality::Cumulative
+                } else {
+                    AggregationTemporality::Delta
+                } as i32,
+            };
+            if mode != "missing_infinity" {
+                histogram.data_points.push(HistogramDataPoint {
+                    explicit_bounds: vec![1.0],
+                    bucket_counts: vec![4],
+                    count: 4,
+                    exemplars: vec![Exemplar::default()],
+                    ..Default::default()
+                });
+            }
+            let mut request = gauge_request(vec![], "gauge");
+            request.resource_metrics[0].scope_metrics[0]
+                .metrics
+                .push(Metric {
+                    name: "histogram".into(),
+                    data: Some(metric::Data::Histogram(histogram.clone())),
+                    ..Default::default()
+                });
+            let ctx = OtlpMetricCtx {
+                with_metric_engine: true,
+                ..Default::default()
+            };
+            let mut conversion =
+                to_grpc_insert_requests(request.clone(), &mut ctx.clone()).unwrap();
+            assert_eq!(conversion.outcome.accepted_data_points, 2, "{mode}");
+            assert_eq!(
+                conversion.outcome.rejected_data_points,
+                i64::from(mode != "missing_infinity"),
+                "{mode}"
+            );
+
+            // Exemplars must not change either the histogram or the independent gauge payload.
+            for point in &mut histogram.data_points {
+                point.exemplars.clear();
+            }
+            request.resource_metrics[0].scope_metrics[0].metrics[1].data =
+                Some(metric::Data::Histogram(histogram));
+            let mut baseline = to_grpc_insert_requests(request, &mut ctx.clone()).unwrap();
+            conversion
+                .requests
+                .inserts
+                .sort_by(|a, b| a.table_name.cmp(&b.table_name));
+            baseline
+                .requests
+                .inserts
+                .sort_by(|a, b| a.table_name.cmp(&b.table_name));
+            // Compare wire bytes so stale NaN values compare by their bits.
+            assert_eq!(
+                conversion.requests.encode_to_vec(),
+                baseline.requests.encode_to_vec(),
+                "{mode}"
+            );
+
+            let expected = match mode {
+                "empty" => vec![
+                    (1, "inf"),
+                    (2, "inf"),
+                    (3, "inf"),
+                    (4, "inf"),
+                    (5, "inf"),
+                    (6, "inf"),
+                ],
+                "missing_infinity" => vec![(2, "1"), (3, "1"), (4, "1")],
+                _ => vec![
+                    (1, "inf"),
+                    (2, "1"),
+                    (3, "1"),
+                    (4, "1"),
+                    (5, "inf"),
+                    (6, "inf"),
+                ],
+            };
+            assert_eq!(conversion.exemplars.len(), expected.len(), "{mode}");
+            let (exemplars, _) = crate::exemplar::into_rows(conversion.exemplars).unwrap();
+            let rows = exemplars.inserts[0].rows.as_ref().unwrap();
+            let index = |name| {
+                rows.schema
+                    .iter()
+                    .position(|c| c.column_name == name)
+                    .unwrap()
+            };
+            let mut actual = Vec::new();
+            for row in &rows.rows {
+                assert_eq!(
+                    row.values[index(EXEMPLAR_METRIC_COLUMN)].value_data,
+                    Some(ValueData::StringValue("histogram_bucket".into()))
+                );
+                let Some(ValueData::TimestampNanosecondValue(timestamp)) =
+                    row.values[index(greptime_timestamp())].value_data
+                else {
+                    panic!("missing exemplar timestamp")
+                };
+                let Some(ValueData::StringValue(ref bound)) = row.values[index("le")].value_data
+                else {
+                    panic!("missing bucket label")
+                };
+                actual.push((timestamp, bound.as_str()));
+                if timestamp == 3 {
+                    assert_eq!(
+                        row.values[index(greptime_value())].value_data,
+                        Some(ValueData::F64Value(0.5))
+                    );
+                }
+            }
+            actual.sort_unstable();
+            assert_eq!(actual, expected, "{mode}");
+        }
+    }
+
     fn exponential_metric(
         name: impl Into<String>,
         data_points: Vec<ExponentialHistogramDataPoint>,

src/servers/src/otlp/metrics/exemplars.rs

@@ -0,0 +1,311 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::{BTreeMap, HashSet};
+use std::sync::Arc;
+
+use api::v1::{ColumnSchema, SemanticType, Value, value::ValueData};
+use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Exemplar, exemplar};
+use session::protocol_ctx::OtlpMetricCtx;
+use snafu::{OptionExt, ensure};
+
+use crate::error::{InvalidOtlpMetricInputSnafu, Result};
+use crate::exemplar::{self as normalized, Exemplar as Observation};
+use crate::otlp::metrics::{scalar_value_string, translate_label_name};
+
+/// Observations whose inherited labels are bound after source-row normalization.
+pub(crate) struct RowExemplars {
+    pub(crate) metric: Arc<str>,
+    /// Starts as an emitted table's row offset; preparation maps it into `row_locations`.
+    pub(crate) row: usize,
+    observations: Vec<Observation>,
+}
+
+impl RowExemplars {
+    /// Drops observations whose inherited tags cannot be represented by the exemplar schema.
+    pub(crate) fn finish(mut self, schema: &[ColumnSchema], row: &[Value]) -> Vec<Observation> {
+        let labels = normalized::labels(
+            schema.iter().zip(row).filter_map(|(column, value)| {
+                if column.semantic_type != SemanticType::Tag as i32 {
+                    return None;
+                }
+                match &value.value_data {
+                    Some(ValueData::StringValue(value)) => {
+                        Some((column.column_name.clone(), value.clone()))
+                    }
+                    _ => None,
+                }
+            }),
+            false,
+            |name| !name.is_empty(),
+        );
+        let Ok(labels) = labels else {
+            crate::metrics::EXEMPLARS_DROPPED_TOTAL
+                .with_label_values(&["otlp"])
+                .inc_by(self.observations.len() as u64);
+            return Vec::new();
+        };
+        let labels = Arc::new(labels);
+        for observation in &mut self.observations {
+            observation.series_labels = labels.clone();
+        }
+        self.observations
+    }
+}
+
+pub(super) fn value(exemplar: &Exemplar) -> Result<f64> {
+    match exemplar
+        .value
+        .as_ref()
+        .context(InvalidOtlpMetricInputSnafu {
+            reason: "exemplar is missing its value",
+        })? {
+        exemplar::Value::AsDouble(value) => Ok(*value),
+        exemplar::Value::AsInt(value) => Ok(*value as f64),
+    }
+}
+
+/// Collects exemplars for metric-engine ingestion, retaining each emitted row's identity
+/// until destination defaults have been filled.
+pub(crate) fn append<'a>(
+    output: &mut Vec<RowExemplars>,
+    exemplars: impl IntoIterator<Item = &'a Exemplar>,
+    metric: &str,
+    row: usize,
+    ctx: &OtlpMetricCtx,
+) {
+    let mut exemplars = exemplars.into_iter().peekable();
+    if exemplars.peek().is_none() {
+        return;
+    }
+    if !ctx.with_metric_engine {
+        crate::metrics::EXEMPLARS_DROPPED_TOTAL
+            .with_label_values(&["otlp"])
+            .inc_by(exemplars.count() as u64);
+        return;
+    }
+    let validate_name = |name: &[u8]| !name.is_empty();
+    let series_labels = Arc::new(BTreeMap::new());
+    let metric: Arc<str> = metric.into();
+    let mut pending = RowExemplars {
+        metric: metric.clone(),
+        row,
+        observations: Vec::new(),
+    };
+    for exemplar in exemplars {
+        let observation: Result<_> = (|| {
+            let mut names = HashSet::new();
+            let mut attributes = Vec::new();
+            for attribute in &exemplar.filtered_attributes {
+                ensure!(
+                    names.insert(&attribute.key),
+                    InvalidOtlpMetricInputSnafu {
+                        reason: format!("duplicate filtered exemplar attribute: {}", attribute.key)
+                    }
+                );
+                if let Some(value) = scalar_value_string(attribute.value.as_ref()) {
+                    let name =
+                        translate_label_name(&attribute.key, ctx.metric_translation_strategy);
+                    attributes.push((name, value));
+                }
+            }
+            let mut labels = normalized::labels(attributes, true, validate_name)?;
+            for (name, bytes, length) in [
+                ("trace_id", &exemplar.trace_id, 16),
+                ("span_id", &exemplar.span_id, 8),
+            ] {
+                if !bytes.is_empty() {
+                    ensure!(
+                        bytes.len() == length,
+                        InvalidOtlpMetricInputSnafu {
+                            reason: format!("exemplar {name} must contain {length} bytes")
+                        }
+                    );
+                    labels.insert(name.to_string(), hex::encode(bytes));
+                }
+            }
+            let labels = normalized::labels(labels, true, validate_name)?;
+            let timestamp = i64::try_from(exemplar.time_unix_nano).map_err(|_| {
+                InvalidOtlpMetricInputSnafu {
+                    reason: "exemplar timestamp exceeds signed nanosecond range",
+                }
+                .build()
+            })?;
+            Ok(Observation {
+                metric: metric.clone(),
+                series_labels: series_labels.clone(),
+                timestamp,
+                value: value(exemplar)?,
+                labels,
+            })
+        })();
+        match observation {
+            Ok(observation) => pending.observations.push(observation),
+            Err(_) => crate::metrics::EXEMPLARS_DROPPED_TOTAL
+                .with_label_values(&["otlp"])
+                .inc(),
+        }
+    }
+    if !pending.observations.is_empty() {
+        output.push(pending);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::otlp::metrics::string_key_value;
+    use crate::row_writer;
+    use crate::row_writer::TableData;
+
+    #[test]
+    fn test_exemplars_require_metric_engine() {
+        let exemplar = Exemplar {
+            time_unix_nano: 1,
+            value: Some(exemplar::Value::AsDouble(0.5)),
+            ..Default::default()
+        };
+        let dropped = crate::metrics::EXEMPLARS_DROPPED_TOTAL.with_label_values(&["otlp"]);
+        for enabled in [false, true] {
+            let before = dropped.get();
+            let mut output = Vec::new();
+            append(
+                &mut output,
+                [&exemplar],
+                "latency",
+                0,
+                &OtlpMetricCtx {
+                    with_metric_engine: enabled,
+                    ..Default::default()
+                },
+            );
+            assert_eq!(output.len(), usize::from(enabled));
+            assert_eq!(dropped.get() - before, u64::from(!enabled));
+        }
+    }
+
+    #[test]
+    fn test_exemplar_labels_follow_otlp_translation_strategy() {
+        use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
+            Gauge, NumberDataPoint, number_data_point,
+        };
+        use session::protocol_ctx::OtlpMetricTranslationStrategy;
+
+        use crate::otlp::metrics::encode_gauge;
+        use crate::row_writer::MultiTableData;
+
+        for strategy in [
+            OtlpMetricTranslationStrategy::NoTranslation,
+            OtlpMetricTranslationStrategy::NoUtf8EscapingWithSuffixes,
+            OtlpMetricTranslationStrategy::UnderscoreEscapingWithSuffixes,
+            OtlpMetricTranslationStrategy::UnderscoreEscapingWithoutSuffixes,
+        ] {
+            for name in ["request.method", "请求"] {
+                let gauge = Gauge {
+                    data_points: vec![NumberDataPoint {
+                        attributes: vec![string_key_value(name, "GET".into())],
+                        time_unix_nano: 1,
+                        value: Some(number_data_point::Value::AsDouble(1.0)),
+                        exemplars: vec![Exemplar {
+                            filtered_attributes: vec![string_key_value(name, "POST".into())],
+                            time_unix_nano: 1,
+                            value: Some(exemplar::Value::AsDouble(0.5)),
+                            ..Default::default()
+                        }],
+                        ..Default::default()
+                    }],
+                };
+                let mut tables = MultiTableData::default();
+                let mut output = Vec::new();
+                encode_gauge(
+                    &mut tables,
+                    "requests",
+                    &gauge,
+                    None,
+                    None,
+                    &OtlpMetricCtx {
+                        with_metric_engine: true,
+                        metric_translation_strategy: strategy,
+                        ..Default::default()
+                    },
+                    &mut output,
+                )
+                .unwrap();
+                let (requests, rows) = tables.into_row_insert_requests();
+                assert_eq!(rows, 1);
+                let rows = requests.inserts[0].rows.as_ref().unwrap();
+                let output = output
+                    .into_iter()
+                    .flat_map(|pending| {
+                        let row = pending.row;
+                        pending.finish(&rows.schema, &rows.rows[row].values)
+                    })
+                    .collect::<Vec<_>>();
+                let name = translate_label_name(name, strategy);
+                assert_eq!(output.len(), 1);
+                assert_eq!(output[0].series_labels[&name], "GET");
+                assert_eq!(output[0].labels[&name], "POST");
+                assert!(
+                    requests.inserts[0]
+                        .rows
+                        .as_ref()
+                        .unwrap()
+                        .schema
+                        .iter()
+                        .any(|column| column.column_name == name)
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn own_time_labels_and_dedicated_id_precedence() {
+        let mut table = TableData::new(2, 1);
+        let mut row = table.alloc_one_row();
+        row_writer::write_tag(&mut table, "job", "api", &mut row).unwrap();
+        let exemplar = Exemplar {
+            filtered_attributes: vec![
+                string_key_value("trace_id", "fallback".into()),
+                string_key_value("span_id", "fallback_span".into()),
+                string_key_value("request.id", "r1".into()),
+            ],
+            time_unix_nano: 1720000000123456789,
+            trace_id: vec![0xab; 16],
+            span_id: vec![0xcd; 8],
+            value: Some(exemplar::Value::AsDouble(0.32)),
+        };
+        let ctx = OtlpMetricCtx {
+            with_metric_engine: true,
+            ..Default::default()
+        };
+        let mut output = Vec::new();
+        append(&mut output, [&exemplar], "latency", table.num_rows(), &ctx);
+        let first = output.pop().unwrap().finish(table.columns(), &row);
+        assert_eq!(first[0].timestamp, 1720000000123456789);
+        assert_eq!(first[0].series_labels["job"], "api");
+        assert_eq!(first[0].labels["trace_id"], "ab".repeat(16));
+        assert_eq!(first[0].labels["span_id"], "cd".repeat(8));
+        let mut without_id = exemplar.clone();
+        without_id.trace_id.clear();
+        append(
+            &mut output,
+            [&without_id],
+            "latency",
+            table.num_rows(),
+            &ctx,
+        );
+        let second = output.pop().unwrap().finish(table.columns(), &row);
+        assert_eq!(second[0].labels["trace_id"], "fallback");
+    }
+}

src/servers/src/otlp/metrics/prepare.rs

@@ -31,6 +31,7 @@ use table::requests::{
 
 use crate::error::{self, Result};
 use crate::metrics::OTLP_EXPONENTIAL_HISTOGRAM_REJECTED_DATA_POINTS;
+use crate::otlp::metrics::exemplars::RowExemplars;
 use crate::otlp::metrics::{
     MetricMetadata, MetricsConversion, OTEL_RESOURCE_INFO_TABLE_NAME, ResourceInfoData,
     add_accepted_data_points, encode_metrics, from_metric_type, has_no_recorded_value,
@@ -56,6 +57,7 @@ struct PreparedPoint {
     resource: usize,
     timestamp: i64,
     rows: Range<usize>,
+    exemplars: Vec<RowExemplars>,
 }
 
 #[derive(Default)]
@@ -213,6 +215,7 @@ impl PreparedMetrics {
         let invalid_interval =
             !matches!(metric.data, Some(metric::Data::Gauge(_))) && start > timestamp;
         let mut outcome = MetricsIngestOutcome::default();
+        let mut exemplars = Vec::new();
         let encoded = if !exponential && (timestamp > i64::MAX as u64 || invalid_interval) {
             Err(error::InvalidOtlpMetricInputSnafu {
                 reason: "invalid OTLP data-point timestamp or interval",
@@ -227,6 +230,7 @@ impl PreparedMetrics {
                 scope_attrs,
                 ctx,
                 &mut outcome,
+                &mut exemplars,
             )
         };
         if let Err(err) = encoded {
@@ -259,6 +263,16 @@ impl PreparedMetrics {
         }
         let first_row = self.row_locations.len();
         for (name, schema, rows) in writer.tables.drain_rows() {
+            for pending in exemplars
+                .iter_mut()
+                .filter(|pending| pending.metric.as_ref() == name)
+            {
+                pending.row = self.row_locations.len().checked_add(pending.row).context(
+                    error::InternalSnafu {
+                        err_msg: "exemplar source row offset overflow",
+                    },
+                )?;
+            }
             let batch = match writer.batches.get(name) {
                 Some(&(columns, batch)) if columns == schema.len() => batch,
                 _ => {
@@ -313,6 +327,7 @@ impl PreparedMetrics {
             resource,
             timestamp: timestamp as i64,
             rows: first_row..self.row_locations.len(),
+            exemplars,
         });
         Ok(true)
     }
@@ -412,6 +427,28 @@ impl PreparedMetrics {
             rejections[index] =
                 Some("incompatible metric sample kinds or column definitions".to_string());
         }
+        // Bind before survivor merging moves batch schemas and row values.
+        let mut exemplars = Vec::new();
+        for (point, rejection) in self.points.iter_mut().zip(&rejections) {
+            if rejection.is_some() {
+                continue;
+            }
+            for pending in std::mem::take(&mut point.exemplars) {
+                let &(batch, row) =
+                    self.row_locations
+                        .get(pending.row)
+                        .context(error::InternalSnafu {
+                            err_msg: "missing exemplar source row",
+                        })?;
+                let rows = self.requests[batch]
+                    .rows
+                    .as_ref()
+                    .context(error::InternalSnafu {
+                        err_msg: "missing prepared batch",
+                    })?;
+                exemplars.extend(pending.finish(&rows.schema, &rows.rows[row].values));
+            }
+        }
         let reuse_batches = rejections.iter().all(Option::is_none) && {
             let mut tables = HashSet::new();
             self.requests.len() <= 1 || self.requests.iter().all(|r| tables.insert(&r.table_name))
@@ -539,6 +576,7 @@ impl PreparedMetrics {
             );
         }
         Ok(MetricsConversion {
+            exemplars,
             requests,
             rows,
             semantic_index,

src/servers/src/postgres.rs

@@ -47,6 +47,7 @@ use crate::query_handler::sql::ServerSqlQueryHandlerRef;
 use crate::query_handler::{
     CopyInHandlerRef, InfluxdbLineProtocolHandlerRef, OpenTelemetryProtocolHandlerRef,
 };
+use crate::request_memory_limiter::ServerMemoryLimiter;
 
 pub(crate) struct GreptimeDBStartupParameters {
     version: String,
@@ -94,6 +95,7 @@ pub struct PostgresServerHandlerInner {
     /// HTTP wiring passes from its options.
     otlp_handler: Option<OpenTelemetryProtocolHandlerRef>,
     otlp_with_metric_engine: bool,
+    memory_limiter: ServerMemoryLimiter,
     force_tls: bool,
     auth_driver: AuthDriver<Arc<GreptimeDBStartupParameters>, UserInfoRef>,
     /// SCRAM selected; failures get the uniform metric label.
@@ -118,6 +120,8 @@ pub(crate) struct MakePostgresServerHandler {
     otlp_handler: Option<OpenTelemetryProtocolHandlerRef>,
     #[builder(default)]
     otlp_with_metric_engine: bool,
+    #[builder(default)]
+    memory_limiter: ServerMemoryLimiter,
     user_provider: Option<UserProviderRef>,
     #[builder(default = "Arc::new(GreptimeDBStartupParameters::new())")]
     param_provider: Arc<GreptimeDBStartupParameters>,
@@ -171,6 +175,7 @@ impl MakePostgresServerHandler {
             influxdb_handler: self.influxdb_handler.clone(),
             otlp_handler: self.otlp_handler.clone(),
             otlp_with_metric_engine: self.otlp_with_metric_engine,
+            memory_limiter: self.memory_limiter.clone(),
             force_tls: self.force_tls,
             auth_driver: AuthDriver::new(self.param_provider.clone())
                 .with_pid_secret_key_generator(Arc::new(ZeroPidSecretKeyGenerator)),

src/servers/src/postgres/copy_in/mod.rs

@@ -721,7 +721,11 @@ impl PostgresServerHandlerInner {
                 let request = ExportMetricsServiceRequest::decode(bytes.as_slice())
                     .map_err(|e| bad_copy_data(format!("invalid OTLP payload: {e}")))?;
                 let outcome = otlp_handler
-                    .metrics(request, Arc::new(ctx))
+                    .metrics(
+                        request,
+                        Arc::new(ctx),
+                        Some((self.memory_limiter.clone(), 0)),
+                    )
                     .await
                     .map_err(convert_err)?;
                 // Same failure semantics as the HTTP route: a payload

src/servers/src/postgres/server.rs

@@ -33,6 +33,7 @@ use crate::query_handler::CopyInHandlerRef;
 use crate::query_handler::InfluxdbLineProtocolHandlerRef;
 use crate::query_handler::OpenTelemetryProtocolHandlerRef;
 use crate::query_handler::sql::ServerSqlQueryHandlerRef;
+use crate::request_memory_limiter::ServerMemoryLimiter;
 use crate::server::{AbortableStream, BaseTcpServer, Server};
 use crate::tls::ReloadableTlsServerConfig;
 
@@ -45,6 +46,7 @@ pub struct PostgresServer {
     influxdb_handler: Option<InfluxdbLineProtocolHandlerRef>,
     otlp_handler: Option<OpenTelemetryProtocolHandlerRef>,
     otlp_with_metric_engine: bool,
+    memory_limiter: ServerMemoryLimiter,
     bind_addr: Option<SocketAddr>,
     process_manager: Option<ProcessManagerRef>,
 }
@@ -85,6 +87,7 @@ impl PostgresServer {
             influxdb_handler: None,
             otlp_handler: None,
             otlp_with_metric_engine: false,
+            memory_limiter: ServerMemoryLimiter::default(),
             bind_addr: None,
             process_manager,
         }
@@ -96,6 +99,12 @@ impl PostgresServer {
         self
     }
 
+    /// Sets the shared memory limiter used for OTLP exemplar expansion.
+    pub fn with_memory_limiter(mut self, limiter: ServerMemoryLimiter) -> Self {
+        self.memory_limiter = limiter;
+        self
+    }
+
     /// Wires the influxdb line-protocol handler used by table-less
     /// `COPY FROM STDIN WITH (FORMAT influxdb)` flushes — the same
     /// handler the HTTP influxdb write endpoints use, so a flushed batch
@@ -137,6 +146,7 @@ impl PostgresServer {
             influxdb_handler: self.influxdb_handler.clone(),
             otlp_handler: self.otlp_handler.clone(),
             otlp_with_metric_engine: self.otlp_with_metric_engine,
+            memory_limiter: self.memory_limiter.clone(),
             user_provider: make_handler.user_provider.clone(),
             param_provider: make_handler.param_provider.clone(),
             force_tls: make_handler.force_tls,

src/servers/src/prom_remote_write/decode.rs

@@ -596,7 +596,7 @@ mod tests {
             );
         }
         assert!(
-            crate::exemplar::estimated_row_bytes(&requests).unwrap()
+            crate::exemplar::estimated_row_bytes(requests.values().map(Vec::as_slice)).unwrap()
                 >= count * (metric.len() + label.len())
         );
     }

src/servers/src/prom_remote_write/v2.rs

@@ -561,7 +561,9 @@ fn convert_remote_write_v2(
             checked_total(histogram_count_total, counts.histograms, "histogram")?;
     }
 
-    budget.reserve(crate::exemplar::estimated_row_bytes(&exemplars))?;
+    budget.reserve(crate::exemplar::estimated_row_bytes(
+        exemplars.values().map(Vec::as_slice),
+    ))?;
 
     Ok(RemoteWriteV2WriteRequests {
         samples: into_context_req(sample_tables),

src/servers/src/query_handler.rs

@@ -56,6 +56,7 @@ use crate::error::Result;
 use crate::http::jaeger::QueryTraceParams;
 use crate::influxdb::InfluxdbRequest;
 use crate::opentsdb::codec::DataPoint;
+use crate::request_memory_limiter::ServerMemoryLimiter;
 pub type OpentsdbProtocolHandlerRef = Arc<dyn OpentsdbProtocolHandler + Send + Sync>;
 pub type InfluxdbLineProtocolHandlerRef = Arc<dyn InfluxdbLineProtocolHandler + Send + Sync>;
 pub type PromStoreProtocolHandlerRef = Arc<dyn PromStoreProtocolHandler + Send + Sync>;
@@ -200,11 +201,13 @@ pub trait PromStoreProtocolHandler {
 
 #[async_trait]
 pub trait OpenTelemetryProtocolHandler: PipelineHandler {
-    /// Handling opentelemetry metrics request
+    /// Handles metrics, reserving row expansion against the transport's limiter.
+    /// The accompanying byte count includes permits already held by this request.
     async fn metrics(
         &self,
         request: ExportMetricsServiceRequest,
         ctx: QueryContextRef,
+        memory_limiter: Option<(ServerMemoryLimiter, u64)>,
     ) -> Result<MetricsIngestOutcome>;
 
     /// Handling opentelemetry traces request

src/servers/src/request_memory_limiter.rs

@@ -70,7 +70,7 @@ impl ServerMemoryLimiter {
     }
 
     /// Rejects requests whose additional bytes cannot fit alongside their own held permits.
-    pub(crate) fn check_request_fits(&self, bytes: u64, held_bytes: u64) -> Result<()> {
+    pub fn check_request_fits(&self, bytes: u64, held_bytes: u64) -> Result<()> {
         let requested_bytes = held_bytes.checked_add(bytes);
         let limit_bytes = self.limit_bytes();
         match requested_bytes {

tests-integration/src/instance.rs

@@ -778,7 +778,7 @@ mod tests {
 
         let errors = frontend::OTLP_RESOURCE_INFO_WRITE_ERRORS.get();
         let outcome =
-            OpenTelemetryProtocolHandler::metrics(&*instance, request, QueryContext::arc())
+            OpenTelemetryProtocolHandler::metrics(&*instance, request, QueryContext::arc(), None)
                 .await
                 .unwrap();
         assert_eq!(

tests-integration/src/tests/instance_test.rs

@@ -16,7 +16,7 @@ use std::env;
 use std::sync::Arc;
 
 use client::{DEFAULT_SCHEMA_NAME, OutputData};
-use common_catalog::consts::DEFAULT_CATALOG_NAME;
+use common_catalog::consts::{DEFAULT_CATALOG_NAME, METRICS_EXEMPLARS_TABLE};
 use common_error::ext::{ErrorExt, RetryHint};
 use common_error::status_code::StatusCode;
 use common_query::Output;
@@ -4560,7 +4560,7 @@ async fn test_exemplar_preserves_physical_table_routing(#[case] batching: bool)
             let table = frontend
                 .catalog_manager()
                 .table(
-                    "greptime",
+                    DEFAULT_CATALOG_NAME,
                     "public",
                     &format!("routed_{version}_{suffix}"),
                     None,
@@ -4574,7 +4574,7 @@ async fn test_exemplar_preserves_physical_table_routing(#[case] batching: bool)
             );
         }
         let output = execute_sql(frontend, &format!(
-            "SELECT count(*) FROM metrics_exemplars e JOIN physical_a m ON e.__table_id = m.__table_id AND e.__tsid = m.__tsid WHERE e.greptime_metric = 'routed_{version}_a'"
+            "SELECT count(*) FROM {METRICS_EXEMPLARS_TABLE} e JOIN physical_a m ON e.__table_id = m.__table_id AND e.__tsid = m.__tsid WHERE e.greptime_metric = 'routed_{version}_a'"
         )).await;
         check_output_stream(
             output.data,
@@ -4583,7 +4583,12 @@ async fn test_exemplar_preserves_physical_table_routing(#[case] batching: bool)
         .await;
         let table = frontend
             .catalog_manager()
-            .table("greptime", "public", "metrics_exemplars", None)
+            .table(
+                DEFAULT_CATALOG_NAME,
+                "public",
+                METRICS_EXEMPLARS_TABLE,
+                None,
+            )
             .await
             .unwrap()
             .unwrap();
@@ -4776,7 +4781,7 @@ async fn test_exemplar_precreated_table_options(
     execute_sql(frontend, "CREATE TABLE source_physical (ts TIMESTAMP TIME INDEX, val DOUBLE) ENGINE=metric WITH ('physical_metric_table'='')").await;
     execute_sql(frontend, "CREATE TABLE metric_source (job STRING PRIMARY KEY, ts TIMESTAMP TIME INDEX, val DOUBLE) ENGINE=metric WITH ('on_physical_table'='source_physical')").await;
     execute_sql(frontend, &format!(
-        "CREATE TABLE metrics_exemplars (__table_id INT UNSIGNED NOT NULL, __tsid BIGINT UNSIGNED NOT NULL, greptime_metric STRING, greptime_timestamp TIMESTAMP({precision}) TIME INDEX, greptime_value DOUBLE, exemplar_labels JSON2, PRIMARY KEY (__table_id, __tsid, greptime_metric)) ENGINE=mito WITH ('append_mode'='{append_mode}', 'primary_key_encoding'='{encoding}', 'compaction.type'='twcs')"
+        "CREATE TABLE {METRICS_EXEMPLARS_TABLE} (__table_id INT UNSIGNED NOT NULL, __tsid BIGINT UNSIGNED NOT NULL, greptime_metric STRING, greptime_timestamp TIMESTAMP({precision}) TIME INDEX, greptime_value DOUBLE, exemplar_labels JSON2, PRIMARY KEY (__table_id, __tsid, greptime_metric)) ENGINE=mito WITH ('append_mode'='{append_mode}', 'primary_key_encoding'='{encoding}', 'compaction.type'='twcs')"
     )).await;
     let (request, _) = into_rows(vec![Exemplar {
         metric: "metric_source".into(),
@@ -4795,15 +4800,20 @@ async fn test_exemplar_precreated_table_options(
         let error = result.unwrap_err();
         assert_eq!(error.status_code(), StatusCode::InvalidArguments);
         assert!(
-            error.output_msg().contains(
-                "metrics_exemplars requires Mito, append_mode=true and primary_key_encoding=sparse"
-            ),
+            error.output_msg().contains(&format!(
+                "{METRICS_EXEMPLARS_TABLE} requires Mito, append_mode=true and primary_key_encoding=sparse"
+            )),
             "{error:?}"
         );
     }
     let table = frontend
         .catalog_manager()
-        .table("greptime", "public", "metrics_exemplars", None)
+        .table(
+            DEFAULT_CATALOG_NAME,
+            "public",
+            METRICS_EXEMPLARS_TABLE,
+            None,
+        )
         .await
         .unwrap()
         .unwrap();
@@ -4815,10 +4825,10 @@ async fn test_exemplar_precreated_table_options(
     let sql = if accepted {
         let timestamp = 1_234_567_890 / 10_i64.pow(9 - precision);
         format!(
-            "SELECT count(*) FROM metrics_exemplars WHERE job = 'api' AND greptime_value = 2.0 AND CAST(greptime_timestamp AS BIGINT) = {timestamp}"
+            "SELECT count(*) FROM {METRICS_EXEMPLARS_TABLE} WHERE job = 'api' AND greptime_value = 2.0 AND CAST(greptime_timestamp AS BIGINT) = {timestamp}"
         )
     } else {
-        "SELECT count(*) FROM metrics_exemplars".into()
+        format!("SELECT count(*) FROM {METRICS_EXEMPLARS_TABLE}")
     };
     let output = execute_sql(frontend, &sql).await;
     check_output_stream(
@@ -4875,7 +4885,12 @@ async fn test_exemplar_source_identity_policy() {
     assert!(
         frontend
             .catalog_manager()
-            .table("greptime", "public", "metrics_exemplars", None)
+            .table(
+                DEFAULT_CATALOG_NAME,
+                "public",
+                METRICS_EXEMPLARS_TABLE,
+                None
+            )
             .await
             .unwrap()
             .is_none()
@@ -4885,7 +4900,7 @@ async fn test_exemplar_source_identity_policy() {
     execute_sql(frontend, create).await;
     let source = frontend
         .catalog_manager()
-        .table("greptime", "public", "metric_source", None)
+        .table(DEFAULT_CATALOG_NAME, "public", "metric_source", None)
         .await
         .unwrap()
         .unwrap()
@@ -4895,7 +4910,7 @@ async fn test_exemplar_source_identity_policy() {
     execute_sql(frontend, create).await;
     let replacement = frontend
         .catalog_manager()
-        .table("greptime", "public", "metric_source", None)
+        .table(DEFAULT_CATALOG_NAME, "public", "metric_source", None)
         .await
         .unwrap()
         .unwrap()
@@ -4915,7 +4930,12 @@ async fn test_exemplar_source_identity_policy() {
         .unwrap();
     let table = frontend
         .catalog_manager()
-        .table("greptime", "public", "metrics_exemplars", None)
+        .table(
+            DEFAULT_CATALOG_NAME,
+            "public",
+            METRICS_EXEMPLARS_TABLE,
+            None,
+        )
         .await
         .unwrap()
         .unwrap();
@@ -4940,7 +4960,7 @@ async fn test_exemplar_source_identity_policy() {
     assert_eq!(info.meta.primary_key_indices, [0, 1, 2, 3]);
     let output = execute_sql(
         frontend,
-        "SELECT __table_id FROM metrics_exemplars ORDER BY __table_id",
+        &format!("SELECT __table_id FROM {METRICS_EXEMPLARS_TABLE} ORDER BY __table_id"),
     )
     .await;
     let OutputData::Stream(stream) = output.data else {

tests-integration/tests/grpc.rs

@@ -35,7 +35,7 @@ use auth::user_provider_from_option;
 use base64::prelude::{BASE64_STANDARD, Engine as _};
 use catalog::process_manager::{ProcessManager, ProcessManagerRef};
 use client::{Client, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, Database, OutputData};
-use common_catalog::consts::{DEFAULT_PRIVATE_SCHEMA_NAME, MITO_ENGINE};
+use common_catalog::consts::{DEFAULT_PRIVATE_SCHEMA_NAME, METRICS_EXEMPLARS_TABLE, 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;
@@ -133,6 +133,7 @@ macro_rules! grpc_tests {
                 test_otel_arrow_auth,
                 test_otel_arrow_delta_histogram,
                 test_otel_arrow_exponential_histogram,
+                test_otel_arrow_exemplars_without_metric_engine,
                 test_insert_and_select,
                 test_dbname,
                 test_grpc_message_size_ok,
@@ -1241,6 +1242,103 @@ pub async fn test_otel_arrow_delta_histogram(store_type: StorageType) {
     let _ = server.shutdown().await;
 }
 
+pub async fn test_otel_arrow_exemplars_without_metric_engine(store_type: StorageType) {
+    use datatypes::arrow::array::BinaryArray;
+    use otel_arrow_rust::Consumer;
+    use otel_arrow_rust::proto::opentelemetry::metrics::v1::metric;
+
+    let (_instance, server) = setup_grpc_server(
+        store_type,
+        "test_otel_arrow_exemplars_without_metric_engine",
+    )
+    .await;
+    let addr = server.bind_addr().unwrap().to_string();
+    let db = Database::new_with_dbname(
+        format!("{DEFAULT_CATALOG_NAME}-{DEFAULT_SCHEMA_NAME}"),
+        Client::with_urls(vec![addr.clone()]),
+    );
+    let mut client = ArrowMetricsServiceClient::connect(format!("http://{addr}"))
+        .await
+        .unwrap();
+    let mut batch = gauge_arrow_batch(20, false);
+    let exemplars = ArrowRecordBatch::try_from_iter(vec![
+        (
+            arrow_consts::PARENT_ID,
+            Arc::new(UInt32Array::from(vec![0_u32])) as ArrayRef,
+        ),
+        (
+            arrow_consts::TIME_UNIX_NANO,
+            Arc::new(TimestampNanosecondArray::from(vec![2_000_000_000])) as ArrayRef,
+        ),
+        (
+            arrow_consts::DOUBLE_VALUE,
+            Arc::new(Float64Array::from(vec![0.5])) as ArrayRef,
+        ),
+        (
+            arrow_consts::SPAN_ID,
+            Arc::new(BinaryArray::from_vec(vec![&[1_u8; 8]])) as ArrayRef,
+        ),
+        (
+            arrow_consts::TRACE_ID,
+            Arc::new(BinaryArray::from_vec(vec![&[2_u8; 16]])) as ArrayRef,
+        ),
+    ])
+    .unwrap();
+    batch.arrow_payloads.push(ArrowPayload {
+        schema_id: "exemplars-20".into(),
+        r#type: ArrowPayloadType::NumberDpExemplars as i32,
+        record: serialize_arrow_record_batch(&exemplars),
+    });
+    let decoded = Consumer::default()
+        .consume_metrics_batches(&mut batch.clone())
+        .unwrap();
+    let Some(metric::Data::Gauge(gauge)) =
+        &decoded.resource_metrics[0].scope_metrics[0].metrics[0].data
+    else {
+        panic!("expected a gauge")
+    };
+    assert_eq!(gauge.data_points[0].exemplars.len(), 1);
+
+    let request = Request::new(futures::stream::iter([
+        batch,
+        delta_histogram_arrow_batch(21, &[(3, &[1, 2], &[1.0])]),
+    ]));
+    let mut response = client.arrow_metrics(request).await.unwrap().into_inner();
+    for batch_id in [20, 21] {
+        let status = response.message().await.unwrap().unwrap();
+        assert_eq!(status.batch_id, batch_id);
+        assert_eq!(status.status_code, ArrowStatusCode::Ok as i32);
+        assert!(status.status_message.is_empty());
+    }
+    assert!(response.message().await.unwrap().is_none());
+    for (sql, count) in [
+        ("SELECT COUNT(*) AS c FROM otel_arrow_gauge", 1),
+        (
+            "SELECT COUNT(*) AS c FROM otel_arrow_delta_histogram_bucket",
+            2,
+        ),
+        (
+            &format!(
+                "SELECT COUNT(*) AS c FROM information_schema.tables WHERE table_name='{METRICS_EXEMPLARS_TABLE}'"
+            ),
+            0,
+        ),
+    ] {
+        let OutputData::Stream(stream) = db.sql(sql).await.unwrap().data else {
+            panic!("expected a stream")
+        };
+        assert_eq!(
+            RecordBatches::try_collect(stream)
+                .await
+                .unwrap()
+                .pretty_print()
+                .unwrap(),
+            format!("+---+\n| c |\n+---+\n| {count} |\n+---+"),
+        );
+    }
+    server.shutdown().await.unwrap();
+}
+
 pub async fn test_otel_arrow_exponential_histogram(store_type: StorageType) {
     let (_instance, server) =
         setup_grpc_server(store_type, "test_otel_arrow_exponential_histogram").await;