perf: optimize OTLP conversion and sparse key encoding

Reduce label normalization and scalar attribute allocations, add exact key capacity planning, and preserve decode experiments for comparison. Files: `src/servers/src/otlp/metrics.rs`, `src/servers/src/otlp/metrics/translator.rs`, `src/mito-codec/src/row_converter/sparse.rs`, `src/servers/Cargo.toml`, `src/servers/benches/bench_otlp_decode.rs`, `src/servers/benches/otlp_decode/capacity_hint.rs`, `src/servers/benches/otlp_decode/README.md`. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
2026-10-12 23:12:28 +00:00 · 2026-09-14 10:48:03 +08:00
parent 3ee1114bb4
commit 94c80cf982
7 changed files with 679 additions and 75 deletions
@@ -301,13 +301,33 @@ impl SparsePrimaryKeyCodec {
        Ok(())
    }

+    /// Exact length of a sparse metric key, including table id and TSID.
+    pub fn encoded_raw_key_len<'a>(
+        tag_values: impl IntoIterator<Item = &'a [u8]>,
+    ) -> Result<usize> {
+        tag_values.into_iter().try_fold(22usize, |length, value| {
+            value
+                .len()
+                .div_ceil(8)
+                .checked_mul(9)
+                .and_then(|encoded| encoded.checked_add(6))
+                .and_then(|encoded| length.checked_add(encoded))
+                .ok_or_else(|| {
+                    InvalidSparsePrimaryKeySnafu {
+                        reason: "encoded primary key length overflows usize",
+                    }
+                    .build()
+                })
+        })
+    }
+
    pub fn encode_raw_tag_value<'a, I>(&self, row: I, buffer: &mut Vec<u8>) -> Result<()>
    where
        I: Iterator<Item = (ColumnId, &'a [u8])>,
    {
        for (tag_column_id, tag_value) in row {
            let value_len = tag_value.len();
-            buffer.reserve(6 + value_len / 8 * 9);
+            buffer.reserve(6 + value_len.div_ceil(8) * 9);
            buffer.put_u32(tag_column_id);
            buffer.put_u8(1);
            buffer.put_u8(!tag_value.is_empty() as u8);
@@ -630,6 +650,38 @@ mod tests {

    use super::*;

+    #[test]
+    fn raw_key_length_matches_encoding_at_chunk_boundaries() {
+        let codec = SparsePrimaryKeyCodec::schemaless();
+        for lengths in [
+            vec![],
+            vec![0],
+            vec![1, 7, 8, 9],
+            vec![15, 16, 17],
+            vec![0, 128, 257],
+        ] {
+            let values: Vec<_> = lengths
+                .into_iter()
+                .map(|length| vec![b'x'; length])
+                .collect();
+            let length =
+                SparsePrimaryKeyCodec::encoded_raw_key_len(values.iter().map(Vec::as_slice))
+                    .unwrap();
+            let mut encoded = Vec::with_capacity(length);
+            codec.encode_internal(42, 99, &mut encoded).unwrap();
+            codec
+                .encode_raw_tag_value(
+                    values
+                        .iter()
+                        .enumerate()
+                        .map(|(index, value)| (index as u32 + 10, value.as_slice())),
+                    &mut encoded,
+                )
+                .unwrap();
+            assert_eq!(length, encoded.len());
+        }
+    }
+
    fn test_region_metadata() -> RegionMetadataRef {
        let mut builder = RegionMetadataBuilder::new(RegionId::new(1, 1));
        builder
@@ -181,6 +181,10 @@ common-version.workspace = true
 name = "bench_prom"
 harness = false

+[[bench]]
+name = "bench_otlp_decode"
+harness = false
+
 [[bench]]
 name = "prom_decode"
 harness = false
@@ -0,0 +1,153 @@
+// 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::hint::black_box;
+use std::path::PathBuf;
+use std::time::Duration;
+
+use bytes::Bytes;
+use criterion::{Criterion, Throughput, criterion_group, criterion_main};
+use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
+use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
+    Gauge, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics, metric,
+};
+use prost::Message;
+
+#[path = "otlp_decode/capacity_hint.rs"]
+mod decode;
+
+fn decode_metrics(c: &mut Criterion) {
+    let directory = std::env::var_os("OTLP_DECODE_CORPUS")
+        .expect("set OTLP_DECODE_CORPUS to uncompressed sample_loader OTLP .pb requests");
+    let mut paths: Vec<PathBuf> = std::fs::read_dir(directory)
+        .unwrap()
+        .map(|entry| entry.unwrap().path())
+        .filter(|path| path.extension().is_some_and(|extension| extension == "pb"))
+        .collect();
+    paths.sort();
+    let corpus: Vec<Bytes> = paths
+        .into_iter()
+        .map(|path| std::fs::read(path).unwrap().into())
+        .collect();
+    assert!(!corpus.is_empty());
+    for payload in &corpus {
+        assert_eq!(
+            ExportMetricsServiceRequest::decode(payload.clone()).unwrap(),
+            ExportMetricsServiceRequest::decode(payload.as_ref()).unwrap()
+        );
+        assert_eq!(
+            ExportMetricsServiceRequest::decode(payload.clone()).unwrap(),
+            decode::decode_metrics(payload.clone()).unwrap()
+        );
+    }
+    if let Some(report) = std::env::var_os("OTLP_DECODE_CAPACITY_REPORT") {
+        let standard: usize = corpus
+            .iter()
+            .map(|payload| {
+                attribute_capacity(&ExportMetricsServiceRequest::decode(payload.clone()).unwrap())
+            })
+            .sum();
+        let hinted: usize = corpus
+            .iter()
+            .map(|payload| attribute_capacity(&decode::decode_metrics(payload.clone()).unwrap()))
+            .sum();
+        std::fs::write(report, format!("standard_attribute_capacity_bytes={standard}\nhinted_attribute_capacity_bytes={hinted}\n")).unwrap();
+    }
+    let mut group = c.benchmark_group("otlp_metrics_decode");
+    group.sample_size(50);
+    group.warm_up_time(Duration::from_secs(2));
+    group.measurement_time(Duration::from_secs(5));
+    group.throughput(Throughput::Bytes(
+        corpus.iter().map(|payload| payload.len() as u64).sum(),
+    ));
+    group.bench_function("bytes", |b| {
+        b.iter(|| {
+            for payload in &corpus {
+                black_box(ExportMetricsServiceRequest::decode(black_box(payload.clone())).unwrap());
+            }
+        })
+    });
+    group.bench_function("slice", |b| {
+        b.iter(|| {
+            for payload in &corpus {
+                black_box(
+                    ExportMetricsServiceRequest::decode(black_box(payload.as_ref())).unwrap(),
+                );
+            }
+        })
+    });
+    group.bench_function("capacity_hint", |b| {
+        b.iter(|| {
+            for payload in &corpus {
+                black_box(decode::decode_metrics(black_box(payload.clone())).unwrap());
+            }
+        })
+    });
+    group.finish();
+
+    let small = Bytes::from(
+        ExportMetricsServiceRequest {
+            resource_metrics: vec![ResourceMetrics {
+                scope_metrics: vec![ScopeMetrics {
+                    metrics: vec![Metric {
+                        data: Some(metric::Data::Gauge(Gauge {
+                            data_points: vec![NumberDataPoint::default()],
+                        })),
+                        ..Default::default()
+                    }],
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+        }
+        .encode_to_vec(),
+    );
+    let mut group = c.benchmark_group("otlp_small_decode");
+    group.measurement_time(Duration::from_secs(3));
+    group.bench_function("bytes", |b| {
+        b.iter(|| black_box(ExportMetricsServiceRequest::decode(black_box(small.clone())).unwrap()))
+    });
+    group.bench_function("capacity_hint", |b| {
+        b.iter(|| black_box(decode::decode_metrics(black_box(small.clone())).unwrap()))
+    });
+    group.finish();
+}
+
+fn attribute_capacity(request: &ExportMetricsServiceRequest) -> usize {
+    request
+        .resource_metrics
+        .iter()
+        .flat_map(|resource| &resource.scope_metrics)
+        .flat_map(|scope| &scope.metrics)
+        .map(|metric| {
+            let points = match &metric.data {
+                Some(metric::Data::Gauge(gauge)) => &gauge.data_points,
+                Some(metric::Data::Sum(sum)) => &sum.data_points,
+                _ => return 0,
+            };
+            points
+                .iter()
+                .map(|point| {
+                    point.attributes.capacity()
+                        * std::mem::size_of::<
+                            otel_arrow_rust::proto::opentelemetry::common::v1::KeyValue,
+                        >()
+                })
+                .sum::<usize>()
+        })
+        .sum()
+}
+
+criterion_group!(benches, decode_metrics);
+criterion_main!(benches);
@@ -0,0 +1,55 @@
+# OTLP metrics decode experiment
+
+This experiment uses uncompressed protobuf requests generated by
+`metrics-bench-suite/bin/sample_loader` with `configs/samples_1750`,
+`--protocol otlp`, `--replica 0`, and `--max-samples 1000`.
+Save HTTP request bodies as `.pb` files, then run:
+
+```sh
+OTLP_DECODE_CORPUS=/path/to/corpus \
+OTLP_DECODE_CAPACITY_REPORT=/path/to/capacity.txt \
+cargo bench -p servers --bench bench_otlp_decode
+```
+
+The harness checks decoded-message equality before timing. Each iteration
+decodes and drops the entire corpus; it does not append successive requests
+into an uncleared message. Throughput is input protobuf bytes, not database
+ingestion throughput. The capacity report measures attribute-vector backing
+storage, excluding strings and the rest of the message; it is not process RSS.
+
+## Candidates
+
+- **bytes:** current prost decoder over `Bytes`.
+- **slice:** prost decoder over `&[u8]`.
+- **capacity_hint:** bounded preallocation of scalar data-point attributes.
+  This follows the enterprise Remote Write decoder's focus on reducing
+  repeated-container allocations. It still delegates wire validation,
+  recursion handling and unknown fields to prost. It is benchmark-only.
+
+## Local results (2026-09-10)
+
+Release-mode isolated harness, the actual pinned `otel-arrow-rust` message
+types, prost/prost-build 0.14.1, bytes 1.11.1; 12 suite requests, each containing
+1,000 input samples. The isolated harness builds only benchmark dependencies,
+not a release database binary. Criterion: 50 samples, 2 seconds warm-up,
+5 seconds measurement for the real corpus.
+
+| Decoder | Corpus time, final run | Throughput |
+| --- | ---: | ---: |
+| bytes | 17.568 ms | 334.79 MiB/s |
+| slice | 18.596 ms | 316.30 MiB/s |
+| capacity hint (32) | 17.437 ms | 337.32 MiB/s |
+
+An earlier repeat measured 17.637 ms vs 17.327 ms for bytes vs capacity hint.
+The observed improvement therefore ranged from approximately 0.7% to 1.8%.
+A capacity hint of 16 measured 17.775 ms vs a 17.635 ms baseline.
+
+The 32-entry hint increased aggregate attribute-vector capacity from
+15,980,160 to 21,414,848 bytes (+34%). On a single empty scalar data point,
+baseline decoding measured 87.130 ns vs 89.584 ns with the hint (+2.8%).
+
+**Conclusion:** retain the current production decoder. The small, variable
+improvement does not justify additional decoding wrappers and memory overhead.
+Keep the candidate here so future alternatives can be compared with the same
+corpus. These results do not establish an end-to-end ingestion speedup: the
+standalone smoke test's default 3-second batching interval dominates latency.
@@ -0,0 +1,166 @@
+// 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 bytes::{Buf, BufMut, Bytes};
+use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
+use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
+    Gauge, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics, Sum, metric,
+};
+use prost::encoding::{DecodeContext, WireType, decode_varint, message};
+use prost::{DecodeError, Message};
+
+/// Decode scalar metrics with bounded attribute capacity hints. Prost still
+/// handles field validation, unknown fields, oneofs and recursive messages.
+pub fn decode_metrics(bytes: Bytes) -> Result<ExportMetricsServiceRequest, DecodeError> {
+    RequestDecoder::decode(bytes).map(|request| request.0)
+}
+
+macro_rules! decoder {
+    ($name:ident, $inner:ty, |$this:ident, $tag:ident, $wire:ident, $buf:ident, $ctx:ident| $merge:block) => {
+        #[derive(Debug, Default)]
+        struct $name($inner);
+
+        impl Message for $name {
+            fn encode_raw(&self, buf: &mut impl BufMut) {
+                self.0.encode_raw(buf);
+            }
+
+            fn merge_field(
+                &mut self,
+                $tag: u32,
+                $wire: WireType,
+                $buf: &mut impl Buf,
+                $ctx: DecodeContext,
+            ) -> Result<(), DecodeError> {
+                let $this = self;
+                $merge
+            }
+
+            fn encoded_len(&self) -> usize {
+                self.0.encoded_len()
+            }
+
+            fn clear(&mut self) {
+                self.0.clear();
+            }
+        }
+    };
+}
+
+decoder!(
+    RequestDecoder,
+    ExportMetricsServiceRequest,
+    |this, tag, wire, buf, ctx| {
+        if tag == 1 {
+            let mut resource = ResourceDecoder::default();
+            message::merge(wire, &mut resource, buf, ctx)?;
+            this.0.resource_metrics.push(resource.0);
+            Ok(())
+        } else {
+            this.0.merge_field(tag, wire, buf, ctx)
+        }
+    }
+);
+
+decoder!(
+    ResourceDecoder,
+    ResourceMetrics,
+    |this, tag, wire, buf, ctx| {
+        if tag == 2 {
+            let mut scope = ScopeDecoder::default();
+            message::merge(wire, &mut scope, buf, ctx)?;
+            this.0.scope_metrics.push(scope.0);
+            Ok(())
+        } else {
+            this.0.merge_field(tag, wire, buf, ctx)
+        }
+    }
+);
+
+decoder!(ScopeDecoder, ScopeMetrics, |this, tag, wire, buf, ctx| {
+    if tag == 2 {
+        let mut metric = MetricDecoder::default();
+        message::merge(wire, &mut metric, buf, ctx)?;
+        this.0.metrics.push(metric.0);
+        Ok(())
+    } else {
+        this.0.merge_field(tag, wire, buf, ctx)
+    }
+});
+
+decoder!(MetricDecoder, Metric, |this, tag, wire, buf, ctx| {
+    match tag {
+        5 => {
+            let mut gauge = GaugeDecoder(match this.0.data.take() {
+                Some(metric::Data::Gauge(gauge)) => gauge,
+                _ => Gauge::default(),
+            });
+            message::merge(wire, &mut gauge, buf, ctx)?;
+            this.0.data = Some(metric::Data::Gauge(gauge.0));
+            Ok(())
+        }
+        7 => {
+            let mut sum = SumDecoder(match this.0.data.take() {
+                Some(metric::Data::Sum(sum)) => sum,
+                _ => Sum::default(),
+            });
+            message::merge(wire, &mut sum, buf, ctx)?;
+            this.0.data = Some(metric::Data::Sum(sum.0));
+            Ok(())
+        }
+        _ => this.0.merge_field(tag, wire, buf, ctx),
+    }
+});
+
+decoder!(GaugeDecoder, Gauge, |this, tag, wire, buf, ctx| {
+    if tag == 1 {
+        this.0
+            .data_points
+            .push(decode_number_point(wire, buf, ctx)?);
+        Ok(())
+    } else {
+        this.0.merge_field(tag, wire, buf, ctx)
+    }
+});
+
+decoder!(SumDecoder, Sum, |this, tag, wire, buf, ctx| {
+    if tag == 1 {
+        this.0
+            .data_points
+            .push(decode_number_point(wire, buf, ctx)?);
+        Ok(())
+    } else {
+        this.0.merge_field(tag, wire, buf, ctx)
+    }
+});
+
+fn decode_number_point(
+    wire: WireType,
+    buf: &mut impl Buf,
+    ctx: DecodeContext,
+) -> Result<NumberDataPoint, DecodeError> {
+    // Large, label-rich samples otherwise repeatedly grow a Vec while decoding.
+    // The hint is capped independently of the untrusted length; small points
+    // retain prost's normal allocation behavior.
+    let capacity = match decode_varint(&mut buf.chunk()) {
+        Ok(len) if len >= 256 => 32,
+        _ => 0,
+    };
+    let mut point = NumberDataPoint {
+        attributes: Vec::with_capacity(capacity),
+        ..Default::default()
+    };
+    message::merge(wire, &mut point, buf, ctx)?;
+    Ok(point)
+}
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::borrow::Cow;
+
 use ahash::HashSet;
 use api::greptime_proto::io::prometheus::write::v2::histogram::{
    Count as PromCount, ResetHint, ZeroCount as PromZeroCount,
@@ -125,7 +127,7 @@ pub struct MetricsConversion {
 /// <https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/metrics/v1/metrics.proto>
 /// for data structure of OTLP metrics.
 pub fn to_grpc_insert_requests(
-    request: ExportMetricsServiceRequest,
+    mut request: ExportMetricsServiceRequest,
    metric_ctx: &mut OtlpMetricCtx,
 ) -> Result<MetricsConversion> {
    let mut table_writer = MultiTableData::default();
@@ -133,7 +135,7 @@ pub fn to_grpc_insert_requests(
    let mut outcome = MetricsIngestOutcome::default();
    let mut resource_info = ResourceInfoData::default();

-    for resource in &request.resource_metrics {
+    for resource in &mut request.resource_metrics {
        if metric_ctx.resource_info
            && !metric_ctx.is_legacy
            && let Some(r) = resource.resource.as_ref()
@@ -147,10 +149,10 @@ pub fn to_grpc_insert_requests(
            attrs
        });

-        for scope in &resource.scope_metrics {
+        for scope in &mut resource.scope_metrics {
            let scope_attrs = process_scope_attrs(scope, metric_ctx);

-            for metric in &scope.metrics {
+            for metric in &mut scope.metrics {
                if metric.data.is_none() {
                    continue;
                }
@@ -458,7 +460,7 @@ fn process_scope_attrs(scope: &ScopeMetrics, metric_ctx: &OtlpMetricCtx) -> Opti

 fn encode_metrics(
    table_writer: &mut MultiTableData,
-    metric: &Metric,
+    metric: &mut Metric,
    resource_attrs: Option<&Vec<KeyValue>>,
    scope_attrs: Option<&Vec<KeyValue>>,
    metric_ctx: &OtlpMetricCtx,
@@ -475,7 +477,7 @@ fn encode_metrics(
        )
    };

-    let emitted = if let Some(data) = &metric.data {
+    let emitted = if let Some(data) = &mut metric.data {
        match data {
            metric::Data::Gauge(gauge) => {
                encode_gauge(
@@ -666,7 +668,7 @@ fn encode_exponential_histogram(
            &mut row,
            resource_attrs,
            scope_attrs,
-            Some(data_point.attributes.as_ref()),
+            DataPointAttributes::Borrowed(&data_point.attributes),
            timestamp_nanos,
            metric_ctx,
        )?;
@@ -922,6 +924,30 @@ enum AttributeType {
    Legacy,
 }

+fn attribute_key<'a>(
+    name: &'a str,
+    attribute_type: AttributeType,
+    metric_ctx: &OtlpMetricCtx,
+) -> Result<Cow<'a, str>> {
+    let key = match attribute_type {
+        AttributeType::Resource | AttributeType::DataPoint => {
+            translate_label_name(name, metric_ctx.metric_translation_strategy)
+        }
+        AttributeType::Scope => Cow::Owned(format!(
+            "otel_scope_{}",
+            translate_label_name(name, metric_ctx.metric_translation_strategy)
+        )),
+        AttributeType::Legacy => Cow::Owned(legacy_normalize_otlp_name(name)),
+    };
+    if key == OTLP_AGGREGATION_TEMPORALITY_LABEL {
+        return Err(error::InvalidOtlpMetricInputSnafu {
+            reason: format!("OTLP attribute `{name}` resolves to reserved label `{key}`"),
+        }
+        .build());
+    }
+    Ok(key)
+}
+
 fn write_attributes(
    writer: &mut TableData,
    row: &mut Vec<Value>,
@@ -939,27 +965,7 @@ fn write_attributes(
        let Some(value) = scalar_value_string(attr.value.as_ref()) else {
            continue;
        };
-        let key = match attribute_type {
-            AttributeType::Resource | AttributeType::DataPoint => {
-                translate_label_name(&attr.key, metric_ctx.metric_translation_strategy)
-            }
-            AttributeType::Scope => {
-                format!(
-                    "otel_scope_{}",
-                    translate_label_name(&attr.key, metric_ctx.metric_translation_strategy)
-                )
-            }
-            AttributeType::Legacy => legacy_normalize_otlp_name(&attr.key),
-        };
-        if key == OTLP_AGGREGATION_TEMPORALITY_LABEL {
-            return Err(error::InvalidOtlpMetricInputSnafu {
-                reason: format!(
-                    "OTLP attribute `{}` resolves to reserved label `{}`",
-                    attr.key, OTLP_AGGREGATION_TEMPORALITY_LABEL
-                ),
-            }
-            .build());
-        }
+        let key = attribute_key(&attr.key, attribute_type, metric_ctx)?;
        tags.push((key, value));
    }
    row_writer::write_tags(writer, tags.into_iter(), row)?;
@@ -967,6 +973,37 @@ fn write_attributes(
    Ok(())
 }

+fn write_owned_attributes(
+    writer: &mut TableData,
+    row: &mut Vec<Value>,
+    attrs: Vec<KeyValue>,
+    attribute_type: AttributeType,
+    metric_ctx: &OtlpMetricCtx,
+) -> Result<()> {
+    let mut tags = Vec::with_capacity(attrs.len());
+    for attr in attrs {
+        let value = match attr.value.and_then(|value| value.value) {
+            Some(any_value::Value::StringValue(value)) => value,
+            Some(any_value::Value::IntValue(value)) => value.to_string(),
+            Some(any_value::Value::DoubleValue(value)) => value.to_string(),
+            _ => continue,
+        };
+        let key = match attribute_key(&attr.key, attribute_type, metric_ctx)? {
+            Cow::Borrowed(_) => attr.key,
+            Cow::Owned(key) => key,
+        };
+        tags.push((key, value));
+    }
+    row_writer::write_tags(writer, tags.into_iter(), row)
+}
+
+enum DataPointAttributes<'a> {
+    // Histograms and summaries fan out the same attributes into multiple rows.
+    Borrowed(&'a Vec<KeyValue>),
+    // Gauge/sum points emit one row, so their strings can move into that row.
+    Owned(Vec<KeyValue>),
+}
+
 fn write_timestamp(
    table: &mut TableData,
    row: &mut Vec<Value>,
@@ -1036,7 +1073,7 @@ fn write_tags_and_timestamp(
    row: &mut Vec<Value>,
    resource_attrs: Option<&Vec<KeyValue>>,
    scope_attrs: Option<&Vec<KeyValue>>,
-    data_point_attrs: Option<&Vec<KeyValue>>,
+    data_point_attrs: DataPointAttributes<'_>,
    timestamp_nanos: i64,
    metric_ctx: &OtlpMetricCtx,
 ) -> Result<()> {
@@ -1049,13 +1086,6 @@ fn write_tags_and_timestamp(
            metric_ctx,
        )?;
        write_attributes(table, row, scope_attrs, AttributeType::Legacy, metric_ctx)?;
-        write_attributes(
-            table,
-            row,
-            data_point_attrs,
-            AttributeType::Legacy,
-            metric_ctx,
-        )?;
    } else {
        // TODO(shuiyisong): check `__type__` and `__unit__` tags in prometheus
        write_attributes(
@@ -1066,13 +1096,20 @@ fn write_tags_and_timestamp(
            metric_ctx,
        )?;
        write_attributes(table, row, scope_attrs, AttributeType::Scope, metric_ctx)?;
-        write_attributes(
-            table,
-            row,
-            data_point_attrs,
-            AttributeType::DataPoint,
-            metric_ctx,
-        )?;
+    }
+
+    let attribute_type = if metric_ctx.is_legacy {
+        AttributeType::Legacy
+    } else {
+        AttributeType::DataPoint
+    };
+    match data_point_attrs {
+        DataPointAttributes::Borrowed(attrs) => {
+            write_attributes(table, row, Some(attrs), attribute_type, metric_ctx)?
+        }
+        DataPointAttributes::Owned(attrs) => {
+            write_owned_attributes(table, row, attrs, attribute_type, metric_ctx)?
+        }
    }

    write_timestamp(table, row, timestamp_nanos, metric_ctx.is_legacy)?;
@@ -1087,7 +1124,7 @@ fn write_tags_and_timestamp(
 fn encode_gauge(
    table_writer: &mut MultiTableData,
    name: &str,
-    gauge: &Gauge,
+    gauge: &mut Gauge,
    resource_attrs: Option<&Vec<KeyValue>>,
    scope_attrs: Option<&Vec<KeyValue>>,
    metric_ctx: &OtlpMetricCtx,
@@ -1098,14 +1135,14 @@ fn encode_gauge(
        gauge.data_points.len(),
    );

-    for data_point in &gauge.data_points {
+    for data_point in &mut gauge.data_points {
        let mut row = table.alloc_one_row();
        write_tags_and_timestamp(
            table,
            &mut row,
            resource_attrs,
            scope_attrs,
-            Some(data_point.attributes.as_ref()),
+            DataPointAttributes::Owned(std::mem::take(&mut data_point.attributes)),
            data_point.time_unix_nano as i64,
            metric_ctx,
        )?;
@@ -1122,7 +1159,7 @@ fn encode_gauge(
 fn encode_sum(
    table_writer: &mut MultiTableData,
    name: &str,
-    sum: &Sum,
+    sum: &mut Sum,
    resource_attrs: Option<&Vec<KeyValue>>,
    scope_attrs: Option<&Vec<KeyValue>>,
    metric_ctx: &OtlpMetricCtx,
@@ -1137,14 +1174,14 @@ fn encode_sum(
        sum.data_points.len(),
    );

-    for data_point in &sum.data_points {
+    for data_point in &mut sum.data_points {
        let mut row = table.alloc_one_row();
        write_tags_and_timestamp(
            table,
            &mut row,
            resource_attrs,
            scope_attrs,
-            Some(data_point.attributes.as_ref()),
+            DataPointAttributes::Owned(std::mem::take(&mut data_point.attributes)),
            data_point.time_unix_nano as i64,
            metric_ctx,
        )?;
@@ -1256,7 +1293,7 @@ fn encode_histogram(
                &mut bucket_row,
                resource_attrs,
                scope_attrs,
-                Some(data_point.attributes.as_ref()),
+                DataPointAttributes::Borrowed(&data_point.attributes),
                data_point.time_unix_nano as i64,
                metric_ctx,
            )?;
@@ -1279,7 +1316,7 @@ fn encode_histogram(
                &mut sum_row,
                resource_attrs,
                scope_attrs,
-                Some(data_point.attributes.as_ref()),
+                DataPointAttributes::Borrowed(&data_point.attributes),
                data_point.time_unix_nano as i64,
                metric_ctx,
            )?;
@@ -1304,7 +1341,7 @@ fn encode_histogram(
            &mut count_row,
            resource_attrs,
            scope_attrs,
-            Some(data_point.attributes.as_ref()),
+            DataPointAttributes::Borrowed(&data_point.attributes),
            data_point.time_unix_nano as i64,
            metric_ctx,
        )?;
@@ -1401,7 +1438,7 @@ fn encode_summary(
                &mut row,
                resource_attrs,
                scope_attrs,
-                Some(data_point.attributes.as_ref()),
+                DataPointAttributes::Borrowed(&data_point.attributes),
                data_point.time_unix_nano as i64,
                metric_ctx,
            )?;
@@ -1442,7 +1479,7 @@ fn encode_summary(
                        &mut row,
                        resource_attrs,
                        scope_attrs,
-                        Some(data_point.attributes.as_ref()),
+                        DataPointAttributes::Borrowed(&data_point.attributes),
                        data_point.time_unix_nano as i64,
                        metric_ctx,
                    )?;
@@ -1468,7 +1505,7 @@ fn encode_summary(
                    &mut row,
                    resource_attrs,
                    scope_attrs,
-                    Some(data_point.attributes.as_ref()),
+                    DataPointAttributes::Borrowed(&data_point.attributes),
                    data_point.time_unix_nano as i64,
                    metric_ctx,
                )?;
@@ -1495,7 +1532,7 @@ fn encode_summary(
                    &mut row,
                    resource_attrs,
                    scope_attrs,
-                    Some(data_point.attributes.as_ref()),
+                    DataPointAttributes::Borrowed(&data_point.attributes),
                    data_point.time_unix_nano as i64,
                    metric_ctx,
                )?;
@@ -1535,6 +1572,101 @@ mod tests {
        }
    }

+    #[test]
+    fn owned_scalar_attributes_preserve_borrowed_row_semantics() {
+        use session::protocol_ctx::OtlpMetricTranslationStrategy::*;
+
+        let resource = vec![keyvalue("shared.key", "resource")];
+        let scope = vec![keyvalue("shared.key", "scope")];
+        let attributes = vec![
+            keyvalue("shared.key", "first"),
+            keyvalue("shared_key", "normalized collision"),
+            keyvalue("shared.key", "last"),
+            keyvalue("é汉", "unicode"),
+            KeyValue {
+                key: "int".into(),
+                value: Some(AnyValue {
+                    value: Some(Val::IntValue(-123)),
+                }),
+            },
+            KeyValue {
+                key: "double".into(),
+                value: Some(AnyValue {
+                    value: Some(Val::DoubleValue(2.5)),
+                }),
+            },
+            KeyValue {
+                key: "ignored_bool".into(),
+                value: Some(AnyValue {
+                    value: Some(Val::BoolValue(true)),
+                }),
+            },
+            KeyValue {
+                key: "ignored_missing".into(),
+                value: None,
+            },
+        ];
+        for is_legacy in [false, true] {
+            for strategy in [
+                UnderscoreEscapingWithSuffixes,
+                UnderscoreEscapingWithoutSuffixes,
+                NoUtf8EscapingWithSuffixes,
+                NoTranslation,
+            ] {
+                let ctx = OtlpMetricCtx {
+                    is_legacy,
+                    metric_translation_strategy: strategy,
+                    ..Default::default()
+                };
+                let mut borrowed = TableData::new(0, 1);
+                let mut borrowed_row = borrowed.alloc_one_row();
+                write_tags_and_timestamp(
+                    &mut borrowed,
+                    &mut borrowed_row,
+                    Some(&resource),
+                    Some(&scope),
+                    DataPointAttributes::Borrowed(&attributes),
+                    123_000_000,
+                    &ctx,
+                )
+                .unwrap();
+                borrowed.add_row(borrowed_row);
+                let mut owned = TableData::new(0, 1);
+                let mut owned_row = owned.alloc_one_row();
+                write_tags_and_timestamp(
+                    &mut owned,
+                    &mut owned_row,
+                    Some(&resource),
+                    Some(&scope),
+                    DataPointAttributes::Owned(attributes.clone()),
+                    123_000_000,
+                    &ctx,
+                )
+                .unwrap();
+                owned.add_row(owned_row);
+                assert_eq!(
+                    borrowed.into_schema_and_rows(),
+                    owned.into_schema_and_rows()
+                );
+            }
+        }
+    }
+
+    #[test]
+    fn owned_scalar_attributes_reject_reserved_label() {
+        let mut table = TableData::new(0, 1);
+        let mut row = table.alloc_one_row();
+        let error = write_owned_attributes(
+            &mut table,
+            &mut row,
+            vec![keyvalue(OTLP_AGGREGATION_TEMPORALITY_LABEL, "reserved")],
+            AttributeType::DataPoint,
+            &OtlpMetricCtx::default(),
+        )
+        .unwrap_err();
+        assert!(matches!(error, error::Error::InvalidOtlpMetricInput { .. }));
+    }
+
    fn descriptor_ctx() -> OtlpMetricCtx {
        OtlpMetricCtx {
            resource_info: true,
@@ -1716,11 +1848,11 @@ mod tests {
                ..Default::default()
            },
        ];
-        let gauge = Gauge { data_points };
+        let mut gauge = Gauge { data_points };
        encode_gauge(
            &mut tables,
            "datamon",
-            &gauge,
+            &mut gauge,
            Some(&vec![]),
            Some(&vec![keyvalue("scope", "otel")]),
            &OtlpMetricCtx::default(),
@@ -1763,14 +1895,14 @@ mod tests {
                ..Default::default()
            },
        ];
-        let sum = Sum {
+        let mut sum = Sum {
            data_points,
            ..Default::default()
        };
        encode_sum(
            &mut tables,
            "datamon",
-            &sum,
+            &mut sum,
            Some(&vec![]),
            Some(&vec![keyvalue("scope", "otel")]),
            &OtlpMetricCtx::default(),
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::borrow::Cow;
+
 use ahash::HashMap;
 use lazy_static::lazy_static;
 use otel_arrow_rust::proto::opentelemetry::metrics::v1::Metric;
@@ -88,11 +90,11 @@ pub fn translate_metric_name(
    }
 }

-pub fn translate_label_name(name: &str, strategy: OtlpMetricTranslationStrategy) -> String {
+pub fn translate_label_name(name: &str, strategy: OtlpMetricTranslationStrategy) -> Cow<'_, str> {
    if strategy.should_escape() {
        normalize_label_name(name)
    } else {
-        name.to_string()
+        Cow::Borrowed(name)
    }
 }

@@ -269,21 +271,34 @@ pub(crate) fn clean_unit_name(name: &str) -> String {
 }

 // See https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/145942706622aba5c276ca47f48df438228bfea4/pkg/translator/prometheus/normalize_label.go#L27
-pub fn normalize_label_name(name: &str) -> String {
-    if name.is_empty() {
-        return name.to_string();
-    }
-
-    let n = NON_ALPHA_NUM_CHAR.replace_all(name, UNDERSCORE);
-    if let Some((_, first)) = n.char_indices().next()
-        && first.is_ascii_digit()
+pub fn normalize_label_name(name: &str) -> Cow<'_, str> {
+    // Most attribute names already use the Prometheus ASCII alphabet. Borrow
+    // them so an existing column does not allocate a fresh name for every row.
+    let normalized = if name
+        .bytes()
+        .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
    {
-        return format!("key_{}", n);
+        Cow::Borrowed(name)
+    } else {
+        // Match the old Unicode regex: each invalid code point, not each UTF-8
+        // byte, becomes one underscore. Consecutive replacements stay distinct.
+        Cow::Owned(
+            name.chars()
+                .map(|ch| if ch.is_ascii_alphanumeric() { ch } else { '_' })
+                .collect::<String>(),
+        )
+    };
+    if normalized
+        .as_bytes()
+        .first()
+        .is_some_and(u8::is_ascii_digit)
+    {
+        return Cow::Owned(format!("key_{normalized}"));
    }
-    if n.starts_with(UNDERSCORE) && !n.starts_with(DOUBLE_UNDERSCORE) {
-        return format!("key{}", n);
+    if normalized.starts_with(UNDERSCORE) && !normalized.starts_with(DOUBLE_UNDERSCORE) {
+        return Cow::Owned(format!("key{normalized}"));
    }
-    n.to_string()
+    normalized
 }

 /// Normalize otlp instrumentation, metric and attribute names
@@ -306,6 +321,33 @@ mod tests {

    use super::*;

+    #[test]
+    fn label_scan_matches_regex_normalization() {
+        fn reference(name: &str) -> String {
+            let normalized = NON_ALPHA_NUM_CHAR.replace_all(name, UNDERSCORE);
+            if normalized.starts_with(|ch: char| ch.is_ascii_digit()) {
+                format!("key_{normalized}")
+            } else if normalized.starts_with('_') && !normalized.starts_with("__") {
+                format!("key{normalized}")
+            } else {
+                normalized.into_owned()
+            }
+        }
+        // Covers prefix rules and one replacement per Unicode scalar, including
+        // adjacent invalid characters and already-normalized underscores.
+        let alphabet = [
+            "", "a", "Z", "0", "_", ".", "-", ":", " ", "\n", "\0", "é", "汉", "💡", "\u{0301}",
+        ];
+        for first in alphabet {
+            for second in alphabet {
+                for third in alphabet {
+                    let name = format!("{first}{second}{third}");
+                    assert_eq!(normalize_label_name(&name), reference(&name), "{name:?}");
+                }
+            }
+        }
+    }
+
    #[test]
    fn test_legacy_normalize_otlp_name() {
        assert_eq!(