servers/otlp/

metrics.rs

// 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 ahash::HashSet;
use api::greptime_proto::io::prometheus::write::v2::histogram::{
    Count as PromCount, ResetHint, ZeroCount as PromZeroCount,
};
use api::greptime_proto::io::prometheus::write::v2::{BucketSpan, Histogram as PromHistogram};
use api::v1::value::ValueData;
use api::v1::{ColumnSchema, RowInsertRequests, SemanticType, Value};
use common_grpc::precision::Precision;
use common_query::native_histogram::{
    encode_native_histogram, native_histogram_column_schema, native_histogram_value_type,
};
use common_query::prelude::{
    GREPTIME_TEMPORALITY_DELTA, OTLP_AGGREGATION_TEMPORALITY_LABEL, greptime_timestamp,
    greptime_value,
};
use common_query::prometheus::PROMETHEUS_STALE_NAN_BITS;
use lazy_static::lazy_static;
use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, KeyValue, any_value};
use otel_arrow_rust::proto::opentelemetry::metrics::v1::{metric, number_data_point, *};
use session::protocol_ctx::{MetricType, OtlpMetricCtx};
use snafu::OptionExt;
use table::requests::{
    METADATA_QUALITY_DECLARED, METRIC_TEMPORALITY_CUMULATIVE, METRIC_TEMPORALITY_DELTA,
    SEMANTIC_METRIC_METADATA_QUALITY, SEMANTIC_METRIC_ORIGINAL_NAME, SEMANTIC_METRIC_TEMPORALITY,
    SEMANTIC_METRIC_TYPE, SEMANTIC_METRIC_UNIT,
};

use crate::error::{self, Result};
use crate::metrics::OTLP_EXPONENTIAL_HISTOGRAM_REJECTED_DATA_POINTS;
use crate::otlp::trace::{KEY_SERVICE_INSTANCE_ID, KEY_SERVICE_NAME, KEY_SERVICE_NAMESPACE};
use crate::query_handler::MetricsIngestOutcome;
use crate::row_writer::{self, MultiTableData, TableData};
pub use crate::semantic::SemanticIndex;
use crate::semantic::{
    METRIC_TYPE_COUNTER, METRIC_TYPE_GAUGE, METRIC_TYPE_HISTOGRAM, METRIC_TYPE_SUMMARY,
    METRIC_TYPE_UPDOWN_COUNTER,
};

mod exemplars;
mod prepare;
mod resource_info;

pub use prepare::{PreparedMetrics, prepare_metrics};
mod translator;

pub use resource_info::OTEL_RESOURCE_INFO_TABLE_NAME;
use resource_info::ResourceInfoData;
pub(crate) use translator::ucum_to_openmetrics_unit;
use translator::{translate_label_name, translate_metric_name};

/// the default column count for table writer
const APPROXIMATE_COLUMN_COUNT: usize = 8;

const COUNT_TABLE_SUFFIX: &str = "_count";
const SUM_TABLE_SUFFIX: &str = "_sum";
const BUCKET_TABLE_SUFFIX: &str = "_bucket";

const JOB_KEY: &str = "job";
const INSTANCE_KEY: &str = "instance";

// see: https://prometheus.io/docs/guides/opentelemetry/#promoting-resource-attributes
const DEFAULT_PROMOTE_ATTRS: [&str; 19] = [
    "service.instance.id",
    "service.name",
    "service.namespace",
    "service.version",
    "cloud.availability_zone",
    "cloud.region",
    "container.name",
    "deployment.environment",
    "deployment.environment.name",
    "k8s.cluster.name",
    "k8s.container.name",
    "k8s.cronjob.name",
    "k8s.daemonset.name",
    "k8s.deployment.name",
    "k8s.job.name",
    "k8s.namespace.name",
    "k8s.pod.name",
    "k8s.replicaset.name",
    "k8s.statefulset.name",
];

lazy_static! {
    static ref DEFAULT_PROMOTE_ATTRS_SET: HashSet<String> =
        HashSet::from_iter(DEFAULT_PROMOTE_ATTRS.iter().map(|s| s.to_string()));
}

const OTEL_SCOPE_NAME: &str = "name";
const OTEL_SCOPE_VERSION: &str = "version";
const OTEL_SCOPE_SCHEMA_URL: &str = "schema_url";
const MIN_EXPONENTIAL_HISTOGRAM_SCALE: i32 = -4;
const MAX_EXPONENTIAL_HISTOGRAM_SCALE: i32 = 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,
    /// Per-table semantic index for the auto-create path to stamp as table
    /// options; covers the descriptor table too.
    pub semantic_index: SemanticIndex,
    /// The synthesized resource descriptor, written separately from the
    /// metric data. See [`resource_info`].
    pub resource_info: Option<RowInsertRequests>,
    pub outcome: MetricsIngestOutcome,
}

struct MetricMetadata {
    name: String,
    original_name: String,
    native_schema: Option<ColumnSchema>,
    semantics: SemanticIndex,
}

impl MetricMetadata {
    fn new(metric: &Metric, ctx: &OtlpMetricCtx) -> Result<Self> {
        let name = translate_metric_name(metric, &ctx.metric_type, ctx.metric_translation_strategy);
        let emitted = emitted_semantic_tables(&ctx.metric_type, &name);
        let mut semantics = SemanticIndex::default();
        record_metric_semantics(&mut semantics, metric, &name, &emitted);
        let native_schema = matches!(metric.data, Some(metric::Data::ExponentialHistogram(_)))
            .then(native_histogram_column_schema)
            .transpose()
            .map_err(|error| {
                error::InternalSnafu {
                    err_msg: error.to_string(),
                }
                .build()
            })?;
        Ok(Self {
            name,
            original_name: metric.name.clone(),
            native_schema,
            semantics,
        })
    }
}

/// Converts OpenTelemetry metrics without validating destination table schemas.
/// Ingestion callers use [`prepare_metrics`], run destination preflight, then call
/// [`PreparedMetrics::finish`] to apply rejections and merge surviving points.
///
/// See
/// <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,
    metric_ctx: &mut OtlpMetricCtx,
) -> Result<MetricsConversion> {
    let prepared = prepare_metrics(request, metric_ctx)?;
    let rejections = vec![None; prepared.num_points()];
    prepared.finish(rejections, &std::collections::HashMap::new())
}

fn validate_sample_kind(table_name: &str, schema: &[ColumnSchema]) -> Result<()> {
    let field_count = schema
        .iter()
        .filter(|column| column.semantic_type == SemanticType::Field as i32)
        .count();
    let has_native_histogram = schema.iter().any(|column| {
        column.semantic_type == SemanticType::Field as i32
            && api::helper::is_column_type_value_eq(
                column.datatype,
                column.datatype_extension.clone(),
                native_histogram_value_type(),
            )
    });
    if has_native_histogram && field_count != 1 {
        return Err(error::InvalidParameterSnafu {
            reason: format!(
                "OTLP metric `{}` cannot mix native histogram and float sample fields",
                table_name
            ),
        }
        .build());
    }
    Ok(())
}

/// The tables a metric emits and their per-table `metric.type`. Histogram fans
/// out into `_bucket` (the histogram) plus `_sum`/`_count` counters; summary
/// fans out into the quantile table plus `_count`/`_sum` counters.
fn emitted_semantic_tables(metric_type: &MetricType, base: &str) -> Vec<(String, &'static str)> {
    match metric_type {
        MetricType::Gauge => vec![(base.to_string(), METRIC_TYPE_GAUGE)],
        MetricType::MonotonicSum => vec![(base.to_string(), METRIC_TYPE_COUNTER)],
        MetricType::NonMonotonicSum => vec![(base.to_string(), METRIC_TYPE_UPDOWN_COUNTER)],
        MetricType::Histogram => vec![
            (
                format!("{base}{BUCKET_TABLE_SUFFIX}"),
                METRIC_TYPE_HISTOGRAM,
            ),
            (format!("{base}{SUM_TABLE_SUFFIX}"), METRIC_TYPE_COUNTER),
            (format!("{base}{COUNT_TABLE_SUFFIX}"), METRIC_TYPE_COUNTER),
        ],
        MetricType::Summary => vec![
            (base.to_string(), METRIC_TYPE_SUMMARY),
            (format!("{base}{COUNT_TABLE_SUFFIX}"), METRIC_TYPE_COUNTER),
            (format!("{base}{SUM_TABLE_SUFFIX}"), METRIC_TYPE_COUNTER),
        ],
        MetricType::ExponentialHistogram => {
            vec![(base.to_string(), METRIC_TYPE_HISTOGRAM)]
        }
        MetricType::Init => vec![],
    }
}

/// Maps OTLP `aggregation_temporality` to the semantic value, or `None` when the
/// instrument has no temporality (gauge/summary) or it is unspecified.
fn temporality_value(data: &metric::Data) -> Option<&'static str> {
    let raw = match data {
        metric::Data::Sum(sum) => sum.aggregation_temporality,
        metric::Data::Histogram(hist) => hist.aggregation_temporality,
        metric::Data::ExponentialHistogram(hist) => hist.aggregation_temporality,
        _ => return None,
    };
    match AggregationTemporality::try_from(raw) {
        Ok(AggregationTemporality::Delta) => Some(METRIC_TEMPORALITY_DELTA),
        Ok(AggregationTemporality::Cumulative) => Some(METRIC_TEMPORALITY_CUMULATIVE),
        _ => None,
    }
}

/// Records the declared metric-level semantic keys for every table this metric
/// emits.
fn record_metric_semantics(
    index: &mut SemanticIndex,
    metric: &Metric,
    name: &str,
    emitted: &[(String, &str)],
) {
    if emitted.is_empty() {
        return;
    }

    let temporality = metric.data.as_ref().and_then(temporality_value);
    let unit = metric.unit.trim();
    // `original_name` is meaningful only when translation renamed the metric.
    let original_name = (name != metric.name.as_str()).then_some(metric.name.as_str());

    for (table, metric_type) in emitted {
        index.record_scalar(table, SEMANTIC_METRIC_TYPE, metric_type);
        index.record_scalar(
            table,
            SEMANTIC_METRIC_METADATA_QUALITY,
            METADATA_QUALITY_DECLARED,
        );
        if let Some(temporality) = temporality {
            index.record_scalar(table, SEMANTIC_METRIC_TEMPORALITY, temporality);
        }
        if !unit.is_empty() {
            index.record_scalar(table, SEMANTIC_METRIC_UNIT, unit);
        }
        if let Some(original_name) = original_name {
            index.record_scalar(table, SEMANTIC_METRIC_ORIGINAL_NAME, original_name);
        }
    }
}

fn from_metric_type(data: &metric::Data) -> MetricType {
    match data {
        metric::Data::Gauge(_) => MetricType::Gauge,
        metric::Data::Sum(s) => {
            if s.is_monotonic {
                MetricType::MonotonicSum
            } else {
                MetricType::NonMonotonicSum
            }
        }
        metric::Data::Histogram(_) => MetricType::Histogram,
        metric::Data::ExponentialHistogram(_) => MetricType::ExponentialHistogram,
        metric::Data::Summary(_) => MetricType::Summary,
    }
}

/// Non-scalar values (arrays, maps, bytes) are not representable as tags.
fn scalar_value_string(value: Option<&AnyValue>) -> Option<String> {
    match value.and_then(|v| v.value.as_ref())? {
        any_value::Value::StringValue(s) => Some(s.clone()),
        any_value::Value::IntValue(v) => Some(v.to_string()),
        any_value::Value::DoubleValue(v) => Some(v.to_string()),
        any_value::Value::BoolValue(v) => Some(v.to_string()),
        _ => None,
    }
}

/// Prometheus-style `(job, instance)` identity. Per the OTel Prometheus
/// compatibility spec `job` folds in `service.namespace` when present; a
/// resource without `service.name` gets no job rather than a fabricated one.
/// Both fields are optional strings, so a named type keeps a swapped
/// destructuring from type-checking at the call sites.
pub(crate) struct ServiceIdentity {
    pub job: Option<String>,
    pub instance: Option<String>,
}

pub(crate) fn service_identity(attrs: &[KeyValue]) -> ServiceIdentity {
    let mut name = None;
    let mut namespace = None;
    let mut instance = None;
    for kv in attrs {
        match kv.key.as_str() {
            KEY_SERVICE_NAME => name = scalar_value_string(kv.value.as_ref()),
            KEY_SERVICE_NAMESPACE => namespace = scalar_value_string(kv.value.as_ref()),
            KEY_SERVICE_INSTANCE_ID => instance = scalar_value_string(kv.value.as_ref()),
            _ => {}
        }
    }
    let job = name.map(|name| match namespace {
        Some(ns) if !ns.is_empty() => format!("{ns}/{name}"),
        _ => name,
    });
    ServiceIdentity { job, instance }
}

fn string_key_value(key: &str, value: String) -> KeyValue {
    KeyValue {
        key: key.to_string(),
        value: Some(AnyValue {
            value: Some(any_value::Value::StringValue(value)),
        }),
    }
}

fn process_resource_attrs(attrs: &mut Vec<KeyValue>, metric_ctx: &OtlpMetricCtx) {
    // remap the service identity attributes to job and instance
    let ServiceIdentity { job, instance } = service_identity(attrs);

    // if promote all, then exclude the list, else, include the list
    if metric_ctx.promote_all_resource_attrs {
        attrs.retain(|kv| !metric_ctx.resource_attrs.contains(&kv.key));
    } else {
        attrs.retain(|kv| {
            metric_ctx.resource_attrs.contains(&kv.key)
                || DEFAULT_PROMOTE_ATTRS_SET.contains(&kv.key)
        });
    }

    if let Some(job) = job {
        attrs.push(string_key_value(JOB_KEY, job));
    }
    if let Some(instance) = instance {
        attrs.push(string_key_value(INSTANCE_KEY, instance));
    }
}

fn process_scope_attrs(scope: &ScopeMetrics, metric_ctx: &OtlpMetricCtx) -> Option<Vec<KeyValue>> {
    if !metric_ctx.promote_scope_attrs {
        return None;
    }

    // persist scope attrs with name, version and schema_url
    scope.scope.as_ref().map(|s| {
        let mut attrs = s.attributes.clone();
        attrs.push(KeyValue {
            key: OTEL_SCOPE_NAME.to_string(),
            value: Some(AnyValue {
                value: Some(any_value::Value::StringValue(s.name.clone())),
            }),
        });
        attrs.push(KeyValue {
            key: OTEL_SCOPE_VERSION.to_string(),
            value: Some(AnyValue {
                value: Some(any_value::Value::StringValue(s.version.clone())),
            }),
        });
        attrs.push(KeyValue {
            key: OTEL_SCOPE_SCHEMA_URL.to_string(),
            value: Some(AnyValue {
                value: Some(any_value::Value::StringValue(scope.schema_url.clone())),
            }),
        });
        attrs
    })
}

#[allow(clippy::too_many_arguments)]
fn encode_metrics(
    table_writer: &mut MultiTableData,
    metric: &Metric,
    metadata: &MetricMetadata,
    resource_attrs: Option<&Vec<KeyValue>>,
    scope_attrs: Option<&Vec<KeyValue>>,
    metric_ctx: &OtlpMetricCtx,
    outcome: &mut MetricsIngestOutcome,
    exemplars: &mut Vec<exemplars::RowExemplars>,
) -> Result<()> {
    let name = &metadata.name;

    if let Some(data) = &metric.data {
        match data {
            metric::Data::Gauge(gauge) => {
                encode_gauge(
                    table_writer,
                    name,
                    gauge,
                    resource_attrs,
                    scope_attrs,
                    metric_ctx,
                    exemplars,
                )?;
                add_accepted_data_points(outcome, gauge.data_points.len())?;
            }
            metric::Data::Sum(sum) => {
                encode_sum(
                    table_writer,
                    name,
                    sum,
                    resource_attrs,
                    scope_attrs,
                    metric_ctx,
                    exemplars,
                )?;
                add_accepted_data_points(outcome, sum.data_points.len())?;
            }
            metric::Data::Summary(summary) => {
                encode_summary(
                    table_writer,
                    name,
                    summary,
                    resource_attrs,
                    scope_attrs,
                    metric_ctx,
                )?;
                add_accepted_data_points(outcome, summary.data_points.len())?;
            }
            metric::Data::Histogram(hist) => encode_histogram(
                table_writer,
                name,
                hist,
                resource_attrs,
                scope_attrs,
                metric_ctx,
                outcome,
                exemplars,
            )?,
            metric::Data::ExponentialHistogram(hist) => encode_exponential_histogram(
                table_writer,
                metadata,
                hist,
                resource_attrs,
                scope_attrs,
                metric_ctx,
                outcome,
                exemplars,
            )?,
        }
    }

    Ok(())
}

fn add_accepted_data_points(outcome: &mut MetricsIngestOutcome, count: usize) -> Result<()> {
    let count = i64::try_from(count).map_err(|_| {
        error::InvalidParameterSnafu {
            reason: "OTLP metrics data-point count exceeds i64",
        }
        .build()
    })?;
    outcome.accepted_data_points =
        outcome
            .accepted_data_points
            .checked_add(count)
            .ok_or_else(|| {
                error::InvalidParameterSnafu {
                    reason: "OTLP accepted data-point count overflows i64",
                }
                .build()
            })?;
    Ok(())
}

fn reject_data_points(
    outcome: &mut MetricsIngestOutcome,
    count: usize,
    reason: impl FnOnce() -> String,
) -> Result<()> {
    if count == 0 {
        return Ok(());
    }
    let count = i64::try_from(count).map_err(|_| {
        error::InvalidParameterSnafu {
            reason: "OTLP rejected data-point count exceeds i64",
        }
        .build()
    })?;
    outcome.rejected_data_points =
        outcome
            .rejected_data_points
            .checked_add(count)
            .ok_or_else(|| {
                error::InvalidParameterSnafu {
                    reason: "OTLP rejected data-point count overflows i64",
                }
                .build()
            })?;
    append_rejection_message(&mut outcome.error_message, reason);
    Ok(())
}

fn append_rejection_message(message: &mut Option<String>, reason: impl FnOnce() -> String) {
    let message = message.get_or_insert_with(String::new);
    let separator = if message.is_empty() { "" } else { "; " };
    let Some(available) = MAX_REJECTION_MESSAGE_BYTES.checked_sub(message.len()) else {
        return;
    };
    if available <= separator.len() {
        return;
    }
    let reason = reason();
    message.push_str(separator);

    let available = MAX_REJECTION_MESSAGE_BYTES - message.len();
    if reason.len() <= available {
        message.push_str(&reason);
        return;
    }

    const ELLIPSIS: &str = "...";
    let mut end = available.saturating_sub(ELLIPSIS.len());
    while !reason.is_char_boundary(end) {
        end -= 1;
    }
    message.push_str(&reason[..end]);
    if available >= ELLIPSIS.len() {
        message.push_str(ELLIPSIS);
    }
}

#[allow(clippy::too_many_arguments)]
fn encode_exponential_histogram(
    table_writer: &mut MultiTableData,
    metadata: &MetricMetadata,
    histogram: &ExponentialHistogram,
    resource_attrs: Option<&Vec<KeyValue>>,
    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) {
        Ok(temporality) => temporality,
        Err(rejection) => {
            reject_data_points(outcome, histogram.data_points.len(), || {
                rejection.message(name)
            })?;
            OTLP_EXPONENTIAL_HISTOGRAM_REJECTED_DATA_POINTS
                .with_label_values(&[rejection.reason_label()])
                .inc_by(histogram.data_points.len() as u64);
            return Ok(());
        }
    };

    let column_schema = metadata
        .native_schema
        .as_ref()
        .context(error::InternalSnafu {
            err_msg: "missing native histogram schema",
        })?;
    for (index, data_point) in histogram.data_points.iter().enumerate() {
        let (value, timestamp_nanos) = match exponential_histogram_value(data_point, temporality) {
            Ok(value) => value,
            Err(reason) => {
                reject_data_points(outcome, 1, || {
                    format!("metric `{name}` data point {index}: {reason}")
                })?;
                OTLP_EXPONENTIAL_HISTOGRAM_REJECTED_DATA_POINTS
                    .with_label_values(&["invalid_data_point"])
                    .inc();
                continue;
            }
        };

        let table = table_writer.get_or_default_table_data(
            name,
            APPROXIMATE_COLUMN_COUNT,
            histogram.data_points.len(),
        );
        let mut row = table.alloc_one_row();
        write_tags_and_timestamp(
            table,
            &mut row,
            resource_attrs,
            scope_attrs,
            Some(data_point.attributes.as_ref()),
            timestamp_nanos,
            metric_ctx,
        )?;
        write_temporality_tag(
            table,
            &mut row,
            temporality == AggregationTemporality::Delta,
        )?;
        row_writer::write_by_schema(
            table,
            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)?;
    }

    Ok(())
}

#[derive(Debug)]
pub(crate) enum ExponentialHistogramRejection {
    UnspecifiedTemporality,
}

impl ExponentialHistogramRejection {
    fn reason_label(&self) -> &'static str {
        match self {
            Self::UnspecifiedTemporality => "unspecified_temporality",
        }
    }

    fn message(&self, name: &str) -> String {
        match self {
            Self::UnspecifiedTemporality => format!(
                "metric `{name}` has unspecified OTLP exponential histogram temporality; delta or cumulative temporality is required"
            ),
        }
    }
}

/// Validate temporality before processing points; delta values remain raw intervals.
pub(crate) fn exponential_histogram_gate(
    histogram: &ExponentialHistogram,
) -> std::result::Result<AggregationTemporality, ExponentialHistogramRejection> {
    match AggregationTemporality::try_from(histogram.aggregation_temporality) {
        Ok(temporality @ (AggregationTemporality::Cumulative | AggregationTemporality::Delta)) => {
            Ok(temporality)
        }
        _ => Err(ExponentialHistogramRejection::UnspecifiedTemporality),
    }
}

pub(crate) fn exponential_histogram_value(
    data_point: &ExponentialHistogramDataPoint,
    temporality: AggregationTemporality,
) -> std::result::Result<(ValueData, i64), String> {
    if data_point.start_time_unix_nano > data_point.time_unix_nano {
        return Err(format!(
            "start_time_unix_nano {} exceeds time_unix_nano {}",
            data_point.start_time_unix_nano, data_point.time_unix_nano
        ));
    }

    let timestamp_nanos = i64::try_from(data_point.time_unix_nano)
        .map_err(|_| format!("time_unix_nano {} overflows i64", data_point.time_unix_nano))?;
    let timestamp = timestamp_millis(data_point.time_unix_nano, "time_unix_nano")?;
    let start_timestamp =
        timestamp_millis(data_point.start_time_unix_nano, "start_time_unix_nano")?;

    let reset_hint = if temporality == AggregationTemporality::Delta {
        ResetHint::Gauge
    } else {
        ResetHint::Unspecified
    } as i32;
    let no_recorded_value = data_point.flags & DataPointFlags::NoRecordedValueMask as u32 != 0;
    let histogram = if no_recorded_value {
        PromHistogram {
            sum: f64::from_bits(PROMETHEUS_STALE_NAN_BITS),
            schema: 0,
            zero_threshold: 0.0,
            reset_hint,
            timestamp,
            start_timestamp,
            count: Some(PromCount::CountInt(0)),
            zero_count: Some(PromZeroCount::ZeroCountInt(0)),
            ..Default::default()
        }
    } else {
        if data_point.scale < MIN_EXPONENTIAL_HISTOGRAM_SCALE {
            return Err(format!(
                "scale {} is unsupported; minimum supported scale is {}",
                data_point.scale, MIN_EXPONENTIAL_HISTOGRAM_SCALE
            ));
        }
        if !data_point.zero_threshold.is_finite() || data_point.zero_threshold < 0.0 {
            return Err(format!(
                "zero_threshold {} must be finite and non-negative",
                data_point.zero_threshold
            ));
        }

        let downscale_shift = if data_point.scale > MAX_EXPONENTIAL_HISTOGRAM_SCALE {
            let shift = data_point
                .scale
                .checked_sub(MAX_EXPONENTIAL_HISTOGRAM_SCALE)
                .ok_or_else(|| "downscale shift overflows i32".to_string())?;
            u32::try_from(shift).map_err(|_| "downscale shift exceeds u32".to_string())?
        } else {
            0
        };
        let (positive_spans, positive_deltas, positive_count) =
            convert_bucket_range("positive", data_point.positive.as_ref(), downscale_shift)?;
        let (negative_spans, negative_deltas, negative_count) =
            convert_bucket_range("negative", data_point.negative.as_ref(), downscale_shift)?;
        let bucket_count = data_point
            .zero_count
            .checked_add(positive_count)
            .and_then(|count| count.checked_add(negative_count))
            .ok_or_else(|| "bucket observation total overflows u64".to_string())?;
        if bucket_count != data_point.count {
            return Err(format!(
                "buckets contain {bucket_count} observations, declared count is {}",
                data_point.count
            ));
        }

        i64::try_from(data_point.count)
            .map_err(|_| format!("count {} overflows i64", data_point.count))?;
        i64::try_from(data_point.zero_count)
            .map_err(|_| format!("zero_count {} overflows i64", data_point.zero_count))?;

        let sum = match data_point.sum {
            Some(sum) if sum.is_nan() => f64::NAN,
            Some(sum) => sum,
            None => f64::NAN,
        };
        PromHistogram {
            sum,
            schema: data_point.scale.min(MAX_EXPONENTIAL_HISTOGRAM_SCALE),
            zero_threshold: data_point.zero_threshold,
            negative_spans,
            negative_deltas,
            positive_spans,
            positive_deltas,
            reset_hint,
            timestamp,
            start_timestamp,
            count: Some(PromCount::CountInt(data_point.count)),
            zero_count: Some(PromZeroCount::ZeroCountInt(data_point.zero_count)),
            ..Default::default()
        }
    };

    encode_native_histogram(&histogram)
        .map(|value| (value, timestamp_nanos))
        .map_err(|error| format!("OTLP exponential histogram cannot be encoded: {error}"))
}

fn timestamp_millis(timestamp_nanos: u64, name: &str) -> std::result::Result<i64, String> {
    i64::try_from(timestamp_nanos / 1_000_000)
        .map_err(|_| format!("{name} {timestamp_nanos} milliseconds overflow i64"))
}

fn convert_bucket_range(
    name: &str,
    buckets: Option<&exponential_histogram_data_point::Buckets>,
    downscale_shift: u32,
) -> std::result::Result<(Vec<BucketSpan>, Vec<i64>, u64), String> {
    let Some(buckets) = buckets else {
        return Ok((Vec::new(), Vec::new(), 0));
    };
    if buckets.bucket_counts.is_empty() {
        return Ok((Vec::new(), Vec::new(), 0));
    }

    let mut merged = Vec::<(i32, u64)>::with_capacity(buckets.bucket_counts.len());
    let mut total = 0u64;
    for (position, count) in buckets.bucket_counts.iter().copied().enumerate() {
        let position =
            i32::try_from(position).map_err(|_| format!("{name} bucket length exceeds i32"))?;
        let source_index = buckets.offset.checked_add(position).ok_or_else(|| {
            format!(
                "{name} bucket index overflows i32 at offset {} and position {position}",
                buckets.offset
            )
        })?;
        let target_index = downscale_bucket_index(source_index, downscale_shift)?
            .checked_add(1)
            .ok_or_else(|| format!("{name} shifted bucket index overflows i32"))?;
        if let Some((last_index, last_count)) = merged.last_mut() {
            if *last_index == target_index {
                *last_count = last_count
                    .checked_add(count)
                    .ok_or_else(|| format!("{name} merged bucket count overflows u64"))?;
                total = total
                    .checked_add(count)
                    .ok_or_else(|| format!("{name} bucket count total overflows u64"))?;
                continue;
            }
            let next_index = last_index
                .checked_add(1)
                .ok_or_else(|| format!("{name} target bucket index overflows i32"))?;
            if target_index != next_index {
                return Err(format!(
                    "{name} bucket indexes are not contiguous after downscaling"
                ));
            }
        }
        total = total
            .checked_add(count)
            .ok_or_else(|| format!("{name} bucket count total overflows u64"))?;
        merged.push((target_index, count));
    }

    let mut spans = Vec::<BucketSpan>::new();
    let mut deltas = Vec::with_capacity(merged.len());
    let mut previous_index = None::<i32>;
    let mut previous = 0i64;
    for (index, count) in merged {
        if count == 0 {
            continue;
        }
        match (spans.last_mut(), previous_index) {
            (Some(span), Some(prev_index)) if prev_index.checked_add(1) == Some(index) => {
                span.length = span
                    .length
                    .checked_add(1)
                    .ok_or_else(|| format!("{name} bucket span length exceeds u32"))?;
            }
            (_, prev_index) => {
                let offset = match prev_index {
                    Some(prev_index) => index
                        .checked_sub(prev_index)
                        .and_then(|gap| gap.checked_sub(1))
                        .ok_or_else(|| format!("{name} bucket span offset overflows i32"))?,
                    None => index,
                };
                spans.push(BucketSpan { offset, length: 1 });
            }
        }
        let count = i64::try_from(count)
            .map_err(|_| format!("{name} bucket count {count} overflows i64"))?;
        let delta = count
            .checked_sub(previous)
            .ok_or_else(|| format!("{name} bucket delta overflows i64"))?;
        deltas.push(delta);
        previous_index = Some(index);
        // Bucket deltas continue across span gaps; omitted zero buckets do not reset them.
        previous = count;
    }

    Ok((spans, deltas, total))
}

fn downscale_bucket_index(index: i32, downscale_shift: u32) -> std::result::Result<i32, String> {
    if downscale_shift == 0 {
        return Ok(index);
    }
    if downscale_shift >= i32::BITS {
        return Ok(if index < 0 { -1 } else { 0 });
    }

    let divisor = 1i64
        .checked_shl(downscale_shift)
        .ok_or_else(|| format!("downscale shift {downscale_shift} is invalid"))?;
    i32::try_from(i64::from(index).div_euclid(divisor))
        .map_err(|_| format!("downscaled bucket index {index} overflows i32"))
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum AttributeType {
    Resource,
    Scope,
    DataPoint,
}

fn write_attributes(
    writer: &mut TableData,
    row: &mut Vec<Value>,
    attrs: Option<&Vec<KeyValue>>,
    attribute_type: AttributeType,
    metric_ctx: &OtlpMetricCtx,
) -> Result<()> {
    let Some(attrs) = attrs else {
        return Ok(());
    };

    let mut tags = Vec::with_capacity(attrs.len());
    for attr in attrs {
        // TODO(sunng87): allow different type of values
        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)
                )
            }
        };
        // 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, key
                ),
            }
            .build());
        }
        tags.push((key, value));
    }
    row_writer::write_tags(writer, tags.into_iter(), row)?;

    Ok(())
}

fn write_timestamp(
    table: &mut TableData,
    row: &mut Vec<Value>,
    time_nano: i64,
    metric_ctx: &OtlpMetricCtx,
) -> Result<()> {
    // Keep the full nanosecond precision whenever the request is headed for
    // the metric engine: `Inserter::handle_metric_row_inserts` converts the
    // timestamps to the physical table's time index unit (which may be
    // micro/nanosecond). Only the non-metric prometheus-compatible path is
    // fixed to milliseconds, to keep auto-created mito tables on the
    // millisecond time index.
    if metric_ctx.with_metric_engine {
        row_writer::write_ts_to_nanos(
            table,
            greptime_timestamp(),
            Some(time_nano),
            Precision::Nanosecond,
            row,
        )
    } else {
        row_writer::write_ts_to_millis(
            table,
            greptime_timestamp(),
            Some(time_nano / 1000000),
            Precision::Millisecond,
            row,
        )
    }
}

fn write_data_point_value(
    table: &mut TableData,
    row: &mut Vec<Value>,
    field: &str,
    value: &Option<number_data_point::Value>,
) -> Result<()> {
    match value {
        Some(number_data_point::Value::AsInt(val)) => {
            // we coerce all values to f64
            row_writer::write_f64(table, field, *val as f64, row)?;
        }
        Some(number_data_point::Value::AsDouble(val)) => {
            row_writer::write_f64(table, field, *val, row)?;
        }
        _ => {}
    }
    Ok(())
}

fn write_temporality_tag(
    table: &mut TableData,
    row: &mut Vec<Value>,
    is_delta: bool,
) -> Result<()> {
    if is_delta {
        row_writer::write_tag(
            table,
            OTLP_AGGREGATION_TEMPORALITY_LABEL,
            GREPTIME_TEMPORALITY_DELTA,
            row,
        )?;
    }
    Ok(())
}

fn has_no_recorded_value(flags: u32) -> bool {
    flags & DataPointFlags::NoRecordedValueMask as u32 != 0
}

fn write_tags_and_timestamp(
    table: &mut TableData,
    row: &mut Vec<Value>,
    resource_attrs: Option<&Vec<KeyValue>>,
    scope_attrs: Option<&Vec<KeyValue>>,
    data_point_attrs: Option<&Vec<KeyValue>>,
    timestamp_nanos: i64,
    metric_ctx: &OtlpMetricCtx,
) -> Result<()> {
    // TODO(shuiyisong): check `__type__` and `__unit__` tags in prometheus
    write_attributes(
        table,
        row,
        resource_attrs,
        AttributeType::Resource,
        metric_ctx,
    )?;
    write_attributes(table, row, scope_attrs, AttributeType::Scope, metric_ctx)?;
    write_attributes(
        table,
        row,
        data_point_attrs,
        AttributeType::DataPoint,
        metric_ctx,
    )?;

    write_timestamp(table, row, timestamp_nanos, metric_ctx)?;

    Ok(())
}

/// encode this gauge metric
///
/// note that there can be multiple data points in the request, it's going to be
/// stored as multiple rows
fn encode_gauge(
    table_writer: &mut MultiTableData,
    name: &str,
    gauge: &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,
        APPROXIMATE_COLUMN_COUNT,
        gauge.data_points.len(),
    );

    for data_point in &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()),
            data_point.time_unix_nano as i64,
            metric_ctx,
        )?;

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

    Ok(())
}

/// Encodes sums, preserving delta points as interval values with a temporality tag.
/// Ingestion is stateless: values are never accumulated across timestamps.
fn encode_sum(
    table_writer: &mut MultiTableData,
    name: &str,
    sum: &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),
        Ok(AggregationTemporality::Delta)
    );
    let table = table_writer.get_or_default_table_data(
        name,
        APPROXIMATE_COLUMN_COUNT,
        sum.data_points.len(),
    );

    for data_point in &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()),
            data_point.time_unix_nano as i64,
            metric_ctx,
        )?;
        write_temporality_tag(table, &mut row, is_delta)?;
        if has_no_recorded_value(data_point.flags) {
            row_writer::write_f64(
                table,
                greptime_value(),
                f64::from_bits(PROMETHEUS_STALE_NAN_BITS),
                &mut row,
            )?;
        } 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);
    }

    Ok(())
}

const HISTOGRAM_LE_COLUMN: &str = "le";

/// Encode histogram data. This function returns 3 insert requests for 3 tables.
///
/// The implementation has been following Prometheus histogram table format:
///
/// - A `%metric%_bucket` table including `greptime_le` tag that stores bucket upper
///   limit, and `greptime_value` for bucket count
/// - A `%metric%_sum` table storing sum of samples
/// -  A `%metric%_count` table storing count of samples.
///
/// By its Prometheus compatibility, we hope to be able to use prometheus
/// 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,
    hist: &Histogram,
    resource_attrs: Option<&Vec<KeyValue>>,
    scope_attrs: Option<&Vec<KeyValue>>,
    metric_ctx: &OtlpMetricCtx,
    outcome: &mut MetricsIngestOutcome,
    exemplars: &mut Vec<exemplars::RowExemplars>,
) -> Result<()> {
    let normalized_name = name;

    let bucket_table_name = format!("{}{}", normalized_name, BUCKET_TABLE_SUFFIX);
    let sum_table_name = format!("{}{}", normalized_name, SUM_TABLE_SUFFIX);
    let count_table_name = format!("{}{}", normalized_name, COUNT_TABLE_SUFFIX);

    let is_delta = matches!(
        AggregationTemporality::try_from(hist.aggregation_temporality),
        Ok(AggregationTemporality::Delta)
    );
    let stale_value = f64::from_bits(PROMETHEUS_STALE_NAN_BITS);
    for (index, data_point) in hist.data_points.iter().enumerate() {
        if let Some(reason) = histogram_data_point_rejection(data_point, is_delta) {
            reject_data_points(outcome, 1, || {
                format!("metric `{name}` data point {index}: {reason}")
            })?;
            continue;
        }

        let bucket_table =
            table_writer.get_or_default_table_data(&bucket_table_name, APPROXIMATE_COLUMN_COUNT, 0);
        let no_recorded_value = has_no_recorded_value(data_point.flags);
        if no_recorded_value {
            bucket_table.reserve_rows(data_point.explicit_bounds.len());
            bucket_table.reserve_rows(1);
        } else {
            bucket_table.reserve_rows(data_point.bucket_counts.len().max(1));
        }
        let bucket_values = if no_recorded_value {
            data_point
                .explicit_bounds
                .iter()
                .copied()
                .chain(std::iter::once(f64::INFINITY))
                .map(|bound| (bound, stale_value))
                .collect::<Vec<_>>()
        } else if data_point.bucket_counts.is_empty() && data_point.explicit_bounds.is_empty() {
            // OTLP count-only histograms map to one implicit Prometheus infinity bucket.
            vec![(f64::INFINITY, data_point.count as f64)]
        } else {
            let mut accumulated_count = 0u64;
            let mut values = Vec::with_capacity(data_point.bucket_counts.len());
            for (idx, count) in data_point.bucket_counts.iter().enumerate() {
                accumulated_count = accumulated_count.checked_add(*count).ok_or_else(|| {
                    error::InvalidParameterSnafu {
                        reason: format!(
                            "metric `{name}` data point {index}: bucket prefix overflows u64"
                        ),
                    }
                    .build()
                })?;
                let bound =
                    data_point.explicit_bounds.get(idx).copied().or_else(|| {
                        (idx == data_point.explicit_bounds.len()).then_some(f64::INFINITY)
                    });
                if let Some(bound) = bound {
                    values.push((bound, accumulated_count as f64));
                }
            }
            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,
                &mut bucket_row,
                resource_attrs,
                scope_attrs,
                Some(data_point.attributes.as_ref()),
                data_point.time_unix_nano as i64,
                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);
        }

        if let Some(sum) = data_point.sum {
            let sum_table = table_writer.get_or_default_table_data(
                &sum_table_name,
                APPROXIMATE_COLUMN_COUNT,
                hist.data_points.len(),
            );
            let mut sum_row = sum_table.alloc_one_row();
            write_tags_and_timestamp(
                sum_table,
                &mut sum_row,
                resource_attrs,
                scope_attrs,
                Some(data_point.attributes.as_ref()),
                data_point.time_unix_nano as i64,
                metric_ctx,
            )?;
            write_temporality_tag(sum_table, &mut sum_row, is_delta)?;
            row_writer::write_f64(
                sum_table,
                greptime_value(),
                if no_recorded_value { stale_value } else { sum },
                &mut sum_row,
            )?;
            sum_table.add_row(sum_row);
        }

        let count_table = table_writer.get_or_default_table_data(
            &count_table_name,
            APPROXIMATE_COLUMN_COUNT,
            hist.data_points.len(),
        );
        let mut count_row = count_table.alloc_one_row();
        write_tags_and_timestamp(
            count_table,
            &mut count_row,
            resource_attrs,
            scope_attrs,
            Some(data_point.attributes.as_ref()),
            data_point.time_unix_nano as i64,
            metric_ctx,
        )?;
        write_temporality_tag(count_table, &mut count_row, is_delta)?;
        row_writer::write_f64(
            count_table,
            greptime_value(),
            if no_recorded_value {
                stale_value
            } else {
                data_point.count as f64
            },
            &mut count_row,
        )?;
        count_table.add_row(count_row);
        add_accepted_data_points(outcome, 1)?;
    }

    Ok(())
}

pub(crate) fn histogram_data_point_rejection(
    data_point: &HistogramDataPoint,
    is_delta: bool,
) -> Option<String> {
    if has_no_recorded_value(data_point.flags) {
        return None;
    }

    if is_delta {
        let valid_empty_layout =
            data_point.bucket_counts.is_empty() && data_point.explicit_bounds.is_empty();
        let expected_buckets = data_point.explicit_bounds.len().checked_add(1);
        if !valid_empty_layout && expected_buckets != Some(data_point.bucket_counts.len()) {
            return Some(format!(
                "bucket_counts length {} must equal explicit_bounds length {} plus one",
                data_point.bucket_counts.len(),
                data_point.explicit_bounds.len()
            ));
        }
        if data_point
            .explicit_bounds
            .iter()
            .any(|bound| !bound.is_finite())
            || data_point
                .explicit_bounds
                .windows(2)
                .any(|bounds| bounds[0] >= bounds[1])
        {
            return Some("explicit_bounds must be finite and strictly increasing".to_string());
        }
        if data_point.count == 0 && data_point.sum.is_some_and(|sum| sum != 0.0) {
            return Some("sum must be absent or zero when count is zero".to_string());
        }
    }

    let bucket_total = data_point
        .bucket_counts
        .iter()
        .try_fold(0u64, |total, count| total.checked_add(*count));
    let Some(bucket_total) = bucket_total else {
        return Some("bucket prefix overflows u64".to_string());
    };
    if is_delta && !data_point.bucket_counts.is_empty() && bucket_total != data_point.count {
        return Some(format!(
            "buckets contain {bucket_total} observations, declared count is {}",
            data_point.count
        ));
    }

    None
}

fn encode_summary(
    table_writer: &mut MultiTableData,
    name: &str,
    summary: &Summary,
    resource_attrs: Option<&Vec<KeyValue>>,
    scope_attrs: Option<&Vec<KeyValue>>,
    metric_ctx: &OtlpMetricCtx,
) -> Result<()> {
    // 1. quantile table
    // 2. count table
    // 3. sum table

    let metric_name = name;
    let count_name = format!("{}{}", metric_name, COUNT_TABLE_SUFFIX);
    let sum_name = format!("{}{}", metric_name, SUM_TABLE_SUFFIX);

    for data_point in &summary.data_points {
        {
            let quantile_table = table_writer.get_or_default_table_data(
                metric_name,
                APPROXIMATE_COLUMN_COUNT,
                summary.data_points.len(),
            );

            for quantile in &data_point.quantile_values {
                let mut row = quantile_table.alloc_one_row();
                write_tags_and_timestamp(
                    quantile_table,
                    &mut row,
                    resource_attrs,
                    scope_attrs,
                    Some(data_point.attributes.as_ref()),
                    data_point.time_unix_nano as i64,
                    metric_ctx,
                )?;
                row_writer::write_tag(quantile_table, "quantile", quantile.quantile, &mut row)?;
                row_writer::write_f64(quantile_table, greptime_value(), quantile.value, &mut row)?;
                quantile_table.add_row(row);
            }
        }
        {
            let count_table = table_writer.get_or_default_table_data(
                &count_name,
                APPROXIMATE_COLUMN_COUNT,
                summary.data_points.len(),
            );
            let mut row = count_table.alloc_one_row();
            write_tags_and_timestamp(
                count_table,
                &mut row,
                resource_attrs,
                scope_attrs,
                Some(data_point.attributes.as_ref()),
                data_point.time_unix_nano as i64,
                metric_ctx,
            )?;

            row_writer::write_f64(
                count_table,
                greptime_value(),
                data_point.count as f64,
                &mut row,
            )?;

            count_table.add_row(row);
        }
        {
            let sum_table = table_writer.get_or_default_table_data(
                &sum_name,
                APPROXIMATE_COLUMN_COUNT,
                summary.data_points.len(),
            );

            let mut row = sum_table.alloc_one_row();
            write_tags_and_timestamp(
                sum_table,
                &mut row,
                resource_attrs,
                scope_attrs,
                Some(data_point.attributes.as_ref()),
                data_point.time_unix_nano as i64,
                metric_ctx,
            )?;

            row_writer::write_f64(sum_table, greptime_value(), data_point.sum, &mut row)?;

            sum_table.add_row(row);
        }
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use api::v1::ColumnDataType;
    use common_query::prelude::set_default_prefix;
    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::*;

    mod delta;

    fn keyvalue(key: &str, value: &str) -> KeyValue {
        KeyValue {
            key: key.into(),
            value: Some(AnyValue {
                value: Some(Val::StringValue(value.into())),
            }),
        }
    }

    fn bool_keyvalue(key: &str, value: bool) -> KeyValue {
        KeyValue {
            key: key.into(),
            value: Some(AnyValue {
                value: Some(Val::BoolValue(value)),
            }),
        }
    }

    fn descriptor_ctx() -> OtlpMetricCtx {
        OtlpMetricCtx {
            resource_info: true,
            ..Default::default()
        }
    }

    fn attr_value(attrs: &[KeyValue], key: &str) -> Option<String> {
        attrs
            .iter()
            .find(|kv| kv.key == key)
            .and_then(|kv| scalar_value_string(kv.value.as_ref()))
    }

    fn gauge_request(
        resource_attrs: Vec<KeyValue>,
        metric_name: &str,
    ) -> ExportMetricsServiceRequest {
        use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource;
        ExportMetricsServiceRequest {
            resource_metrics: vec![ResourceMetrics {
                resource: Some(Resource {
                    attributes: resource_attrs,
                    ..Default::default()
                }),
                scope_metrics: vec![ScopeMetrics {
                    metrics: vec![Metric {
                        name: metric_name.to_string(),
                        data: Some(metric::Data::Gauge(Gauge {
                            data_points: vec![NumberDataPoint {
                                time_unix_nano: 1_000_000,
                                value: Some(Value::AsInt(1)),
                                ..Default::default()
                            }],
                        })),
                        ..Default::default()
                    }],
                    ..Default::default()
                }],
                ..Default::default()
            }],
        }
    }

    fn column_names(request: &RowInsertRequests, table: &str) -> Vec<String> {
        request
            .inserts
            .iter()
            .find(|r| r.table_name == table)
            .unwrap_or_else(|| panic!("missing table {table}"))
            .rows
            .as_ref()
            .unwrap()
            .schema
            .iter()
            .map(|c| c.column_name.clone())
            .collect()
    }

    #[test]
    fn test_conversion_synthesizes_resource_descriptor() {
        set_default_prefix(None).unwrap();
        let request = gauge_request(
            vec![keyvalue("service.name", "api"), keyvalue("host.id", "h-1")],
            "my_gauge",
        );
        let conversion = to_grpc_insert_requests(request, &mut descriptor_ctx()).unwrap();

        // descriptor keeps raw OTel keys while the metric table's labels went
        // through the (default underscore-escaping) translation strategy
        let resource_info = conversion.resource_info.expect("descriptor synthesized");
        let descriptor_cols = column_names(&resource_info, OTEL_RESOURCE_INFO_TABLE_NAME);
        assert!(descriptor_cols.contains(&"host.id".to_string()));
        assert!(descriptor_cols.contains(&"service.name".to_string()));
        assert!(descriptor_cols.contains(&"job".to_string()));
        let metric_cols = column_names(&conversion.requests, "my_gauge");
        assert!(metric_cols.contains(&"service_name".to_string()));
        assert!(!metric_cols.contains(&"service.name".to_string()));
        // host.id is not in the promote list: only the descriptor keeps it
        assert!(!metric_cols.contains(&"host_id".to_string()));

        let decoded = decode(&conversion.semantic_index);
        let t = &decoded[OTEL_RESOURCE_INFO_TABLE_NAME];
        assert_eq!(
            t.get(SEMANTIC_METRIC_TYPE).map(String::as_str),
            Some("info")
        );
        assert_eq!(
            t.get(SEMANTIC_METRIC_METADATA_QUALITY).map(String::as_str),
            Some("declared")
        );
    }

    #[test]
    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();
        let Some(metric::Data::Gauge(gauge)) = metric.data.as_mut() else {
            unreachable!()
        };
        gauge.data_points[0].time_unix_nano =
            2 * common_catalog::consts::SEMANTIC_GRAPH_WINDOW_NANOS as u64;
        request.resource_metrics.push(rejected);
        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())],
                &Default::default(),
            )
            .unwrap();
        assert_eq!(
            (
                conversion.outcome.accepted_data_points,
                conversion.outcome.rejected_data_points
            ),
            (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);
        let timestamp = rows
            .schema
            .iter()
            .position(|c| c.column_name == greptime_timestamp())
            .unwrap();
        assert_eq!(
            rows.rows[0].values[timestamp].value_data,
            Some(ValueData::TimestampMillisecondValue(1))
        );
    }

    #[test]
    fn test_finish_rejects_mismatched_point_counts() {
        for (requests, rejections) in [(0, 1), (2, 1), (1, 0), (1, 2)] {
            let mut prepared = prepare_metrics(
                gauge_request(vec![], "gauge"),
                &mut OtlpMetricCtx::default(),
            )
            .unwrap();
            prepared.requests.resize(requests, Default::default());
            let err = prepared
                .finish(vec![None; rejections], &Default::default())
                .unwrap_err();
            assert!(matches!(err, error::Error::Internal { .. }));
        }
    }

    #[test]
    fn test_conversion_describes_only_accepted_exponential_histogram_windows() {
        set_default_prefix(None).unwrap();
        let window = common_catalog::consts::SEMANTIC_GRAPH_WINDOW_NANOS as u64;
        let mut valid = exponential_point();
        valid.time_unix_nano = window + 1_000_000;

        let mut invalid_count = valid.clone();
        invalid_count.time_unix_nano += 8_000_000;
        invalid_count.count += 1;
        let mut invalid_scale = valid.clone();
        invalid_scale.time_unix_nano += window;
        invalid_scale.scale = -5;
        // OTLP bucket conversion succeeds, but the shared native encoder rejects this index.
        let invalid_native = ExponentialHistogramDataPoint {
            time_unix_nano: 3 * window + 1_000_000,
            count: 1,
            positive: Some(exponential_buckets(2000, vec![1])),
            ..Default::default()
        };
        assert!(
            exponential_histogram_value(&invalid_native, AggregationTemporality::Cumulative)
                .unwrap_err()
                .contains("out of range")
        );
        let stale = ExponentialHistogramDataPoint {
            time_unix_nano: 4 * window + 1_000_000,
            count: u64::MAX,
            scale: i32::MIN,
            flags: DataPointFlags::NoRecordedValueMask as u32,
            ..Default::default()
        };
        let invalid_time = ExponentialHistogramDataPoint {
            time_unix_nano: 5 * window + 1_000_000,
            start_time_unix_nano: 5 * window + 2_000_000,
            ..stale.clone()
        };
        let rejected = vec![invalid_count, invalid_scale, invalid_native, invalid_time];
        let mut mixed = rejected.clone();
        mixed.extend([valid, stale]);
        for (points, expected) in [
            (rejected, vec![]),
            (
                mixed,
                vec![
                    (window / 1_000_000 + 1) as i64,
                    (4 * window / 1_000_000 + 1) as i64,
                ],
            ),
        ] {
            let mut request = metrics_request(vec![exponential_metric(
                "latency",
                points,
                AggregationTemporality::Cumulative,
            )]);
            request.resource_metrics[0].resource = Some(Resource {
                attributes: vec![keyvalue("service.name", "api")],
                ..Default::default()
            });
            let conversion = to_grpc_insert_requests(request, &mut descriptor_ctx()).unwrap();
            assert_eq!(conversion.outcome.rejected_data_points, 4);
            assert_eq!(
                conversion.outcome.accepted_data_points,
                expected.len() as i64
            );
            assert_eq!(conversion.rows, expected.len());
            assert_eq!(conversion.resource_info.is_some(), !expected.is_empty());
            let actual = conversion
                .resource_info
                .map(|descriptor| {
                    let rows = descriptor.inserts[0].rows.as_ref().unwrap();
                    let ts_index = rows
                        .schema
                        .iter()
                        .position(|column| column.column_name == greptime_timestamp())
                        .unwrap();
                    rows.rows
                        .iter()
                        .map(|row| match row.values[ts_index].value_data {
                            Some(ValueData::TimestampMillisecondValue(ts)) => ts,
                            _ => panic!("expected descriptor timestamp"),
                        })
                        .collect::<Vec<_>>()
                })
                .unwrap_or_default();
            assert_eq!(actual, expected);
        }
    }

    #[test]
    fn test_conversion_skips_descriptor_on_metric_name_collision() {
        set_default_prefix(None).unwrap();
        for (enabled, attributes, warns) in [
            (true, vec![keyvalue("service.name", "api")], true),
            (false, vec![keyvalue("service.name", "api")], false),
            (true, vec![keyvalue("unprojected", "value")], false),
        ] {
            let logs = tempfile::NamedTempFile::new().unwrap();
            let writer = logs.reopen().unwrap();
            let subscriber = common_telemetry::tracing_subscriber::fmt()
                .without_time()
                .with_ansi(false)
                .with_writer(move || writer.try_clone().unwrap())
                .finish();
            let conversion = tracing::subscriber::with_default(subscriber, || {
                let mut ctx = descriptor_ctx();
                ctx.resource_info = enabled;
                to_grpc_insert_requests(
                    gauge_request(attributes, OTEL_RESOURCE_INFO_TABLE_NAME),
                    &mut ctx,
                )
                .unwrap()
            });
            let diagnostics = std::fs::read_to_string(logs.path()).unwrap();
            assert_eq!(
                diagnostics.contains("Skipping OTLP resource descriptor"),
                warns
            );
            assert!(conversion.resource_info.is_none());
            assert!(
                conversion
                    .requests
                    .inserts
                    .iter()
                    .any(|r| r.table_name == OTEL_RESOURCE_INFO_TABLE_NAME)
            );
        }
    }

    #[test]
    fn test_job_composition_follows_service_namespace() {
        let mut attrs = vec![
            keyvalue("service.name", "api"),
            keyvalue("service.namespace", "shop"),
        ];
        process_resource_attrs(&mut attrs, &OtlpMetricCtx::default());
        assert_eq!(attr_value(&attrs, "job").as_deref(), Some("shop/api"));

        let mut attrs = vec![keyvalue("service.name", "api")];
        process_resource_attrs(&mut attrs, &OtlpMetricCtx::default());
        assert_eq!(attr_value(&attrs, "job").as_deref(), Some("api"));

        let mut attrs = vec![
            keyvalue("service.name", "api"),
            keyvalue("service.namespace", ""),
        ];
        process_resource_attrs(&mut attrs, &OtlpMetricCtx::default());
        assert_eq!(attr_value(&attrs, "job").as_deref(), Some("api"));

        // no service.name: no job is fabricated
        let mut attrs = vec![
            keyvalue("service.namespace", "shop"),
            keyvalue("service.instance.id", "inst-1"),
        ];
        process_resource_attrs(&mut attrs, &OtlpMetricCtx::default());
        assert_eq!(attr_value(&attrs, "job"), None);
        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();
        let request = metrics_request(vec![Metric {
            name: "bool_repro".to_string(),
            data: Some(metric::Data::Sum(Sum {
                data_points: vec![
                    NumberDataPoint {
                        attributes: vec![bool_keyvalue("failed", false)],
                        start_time_unix_nano: 1_000_000,
                        time_unix_nano: 2_000_000,
                        value: Some(Value::AsDouble(100.0)),
                        ..Default::default()
                    },
                    NumberDataPoint {
                        attributes: vec![bool_keyvalue("failed", true)],
                        start_time_unix_nano: 1_000_000,
                        time_unix_nano: 2_000_000,
                        value: Some(Value::AsDouble(7.0)),
                        ..Default::default()
                    },
                ],
                aggregation_temporality: AggregationTemporality::Cumulative as i32,
                is_monotonic: true,
            })),
            ..Default::default()
        }]);
        let conversion = to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap();

        let rows = conversion
            .requests
            .inserts
            .iter()
            .find(|insert| insert.table_name == "bool_repro_total")
            .expect("missing bool_repro_total table")
            .rows
            .as_ref()
            .unwrap();
        assert_eq!(2, rows.rows.len());
        let failed = rows
            .schema
            .iter()
            .position(|column| column.column_name == "failed")
            .expect("boolean attribute `failed` was dropped from the label set");
        let values = rows
            .rows
            .iter()
            .map(|row| row.values[failed].value_data.clone())
            .collect::<Vec<_>>();
        assert_eq!(
            vec![
                Some(ValueData::StringValue("false".to_string())),
                Some(ValueData::StringValue("true".to_string())),
            ],
            values
        );
    }

    #[test]
    fn test_encode_gauge() {
        let mut tables = MultiTableData::default();

        let data_points = vec![
            NumberDataPoint {
                attributes: vec![keyvalue("host", "testsevrer")],
                time_unix_nano: 100,
                value: Some(Value::AsInt(100)),
                ..Default::default()
            },
            NumberDataPoint {
                attributes: vec![keyvalue("host", "testserver")],
                time_unix_nano: 105,
                value: Some(Value::AsInt(105)),
                ..Default::default()
            },
        ];
        let gauge = Gauge { data_points };
        encode_gauge(
            &mut tables,
            "datamon",
            &gauge,
            Some(&vec![]),
            Some(&vec![keyvalue("scope", "otel")]),
            &OtlpMetricCtx::default(),
            &mut Vec::new(),
        )
        .unwrap();

        let table = tables.get_or_default_table_data("datamon", 0, 0);
        assert_eq!(table.num_rows(), 2);
        assert_eq!(table.num_columns(), 4);
        assert_eq!(
            table
                .columns()
                .iter()
                .map(|c| &c.column_name)
                .collect::<Vec<&String>>(),
            vec![
                "otel_scope_scope",
                "host",
                greptime_timestamp(),
                greptime_value()
            ]
        );
    }

    #[test]
    fn test_encode_sum() {
        let mut tables = MultiTableData::default();

        let data_points = vec![
            NumberDataPoint {
                attributes: vec![keyvalue("host", "testserver")],
                time_unix_nano: 100,
                value: Some(Value::AsInt(100)),
                ..Default::default()
            },
            NumberDataPoint {
                attributes: vec![keyvalue("host", "testserver")],
                time_unix_nano: 105,
                value: Some(Value::AsInt(0)),
                ..Default::default()
            },
        ];
        let sum = Sum {
            data_points,
            ..Default::default()
        };
        encode_sum(
            &mut tables,
            "datamon",
            &sum,
            Some(&vec![]),
            Some(&vec![keyvalue("scope", "otel")]),
            &OtlpMetricCtx::default(),
            &mut Vec::new(),
        )
        .unwrap();

        let table = tables.get_or_default_table_data("datamon", 0, 0);
        assert_eq!(table.num_rows(), 2);
        assert_eq!(table.num_columns(), 4);
        assert_eq!(
            table
                .columns()
                .iter()
                .map(|c| &c.column_name)
                .collect::<Vec<&String>>(),
            vec![
                "otel_scope_scope",
                "host",
                greptime_timestamp(),
                greptime_value()
            ]
        );
    }

    #[test]
    fn test_encode_summary() {
        let mut tables = MultiTableData::default();

        let data_points = vec![SummaryDataPoint {
            attributes: vec![keyvalue("host", "testserver")],
            time_unix_nano: 100,
            count: 25,
            sum: 5400.0,
            quantile_values: vec![
                ValueAtQuantile {
                    quantile: 0.90,
                    value: 1000.0,
                },
                ValueAtQuantile {
                    quantile: 0.95,
                    value: 3030.0,
                },
            ],
            ..Default::default()
        }];
        let summary = Summary { data_points };
        encode_summary(
            &mut tables,
            "datamon",
            &summary,
            Some(&vec![]),
            Some(&vec![keyvalue("scope", "otel")]),
            &OtlpMetricCtx::default(),
        )
        .unwrap();

        let table = tables.get_or_default_table_data("datamon", 0, 0);
        assert_eq!(table.num_rows(), 2);
        assert_eq!(table.num_columns(), 5);
        assert_eq!(
            table
                .columns()
                .iter()
                .map(|c| &c.column_name)
                .collect::<Vec<&String>>(),
            vec![
                "otel_scope_scope",
                "host",
                greptime_timestamp(),
                "quantile",
                greptime_value()
            ]
        );

        let table = tables.get_or_default_table_data("datamon_count", 0, 0);
        assert_eq!(table.num_rows(), 1);
        assert_eq!(table.num_columns(), 4);
        assert_eq!(
            table
                .columns()
                .iter()
                .map(|c| &c.column_name)
                .collect::<Vec<&String>>(),
            vec![
                "otel_scope_scope",
                "host",
                greptime_timestamp(),
                greptime_value()
            ]
        );

        let table = tables.get_or_default_table_data("datamon_sum", 0, 0);
        assert_eq!(table.num_rows(), 1);
        assert_eq!(table.num_columns(), 4);
        assert_eq!(
            table
                .columns()
                .iter()
                .map(|c| &c.column_name)
                .collect::<Vec<&String>>(),
            vec![
                "otel_scope_scope",
                "host",
                greptime_timestamp(),
                greptime_value()
            ]
        );
    }

    #[test]
    fn test_encode_histogram() {
        let mut tables = MultiTableData::default();
        let mut outcome = MetricsIngestOutcome::default();

        let data_points = vec![HistogramDataPoint {
            attributes: vec![keyvalue("host", "testserver")],
            time_unix_nano: 100,
            start_time_unix_nano: 23,
            count: 25,
            sum: Some(100.),
            max: Some(200.),
            min: Some(0.03),
            bucket_counts: vec![2, 4, 6, 9, 4],
            explicit_bounds: vec![0.1, 1., 10., 100.],
            ..Default::default()
        }];

        let histogram = Histogram {
            data_points,
            aggregation_temporality: AggregationTemporality::Delta.into(),
        };
        encode_histogram(
            &mut tables,
            "histo",
            &histogram,
            Some(&vec![]),
            Some(&vec![keyvalue("scope", "otel")]),
            &OtlpMetricCtx::default(),
            &mut outcome,
            &mut Vec::new(),
        )
        .unwrap();

        assert_eq!(3, tables.num_tables());
        assert_eq!(1, outcome.accepted_data_points);

        // bucket table
        let bucket_table = tables.get_or_default_table_data("histo_bucket", 0, 0);
        assert_eq!(bucket_table.num_rows(), 5);
        assert_eq!(bucket_table.num_columns(), 6);
        assert_eq!(
            bucket_table
                .columns()
                .iter()
                .map(|c| &c.column_name)
                .collect::<Vec<&String>>(),
            vec![
                "otel_scope_scope",
                "host",
                greptime_timestamp(),
                OTLP_AGGREGATION_TEMPORALITY_LABEL,
                "le",
                greptime_value(),
            ]
        );

        let sum_table = tables.get_or_default_table_data("histo_sum", 0, 0);
        assert_eq!(sum_table.num_rows(), 1);
        assert_eq!(sum_table.num_columns(), 5);
        assert_eq!(
            sum_table
                .columns()
                .iter()
                .map(|c| &c.column_name)
                .collect::<Vec<&String>>(),
            vec![
                "otel_scope_scope",
                "host",
                greptime_timestamp(),
                OTLP_AGGREGATION_TEMPORALITY_LABEL,
                greptime_value()
            ]
        );

        let count_table = tables.get_or_default_table_data("histo_count", 0, 0);
        assert_eq!(count_table.num_rows(), 1);
        assert_eq!(count_table.num_columns(), 5);
        assert_eq!(
            count_table
                .columns()
                .iter()
                .map(|c| &c.column_name)
                .collect::<Vec<&String>>(),
            vec![
                "otel_scope_scope",
                "host",
                greptime_timestamp(),
                OTLP_AGGREGATION_TEMPORALITY_LABEL,
                greptime_value()
            ]
        );
    }

    use std::collections::BTreeMap;

    use table::requests::validate_semantic_option;

    fn decode(index: &SemanticIndex) -> BTreeMap<String, BTreeMap<String, String>> {
        let nested: BTreeMap<String, BTreeMap<String, BTreeMap<String, String>>> =
            serde_json::from_str(&index.encode("public").expect("non-empty index")).unwrap();
        nested.into_values().next().unwrap()
    }

    fn record(metric: &Metric, metric_type: MetricType, name: &str) -> SemanticIndex {
        let ctx = OtlpMetricCtx {
            metric_type,
            ..Default::default()
        };
        let mut index = SemanticIndex::default();
        record_metric_semantics(
            &mut index,
            metric,
            name,
            &emitted_semantic_tables(&ctx.metric_type, name),
        );
        index
    }

    #[test]
    fn test_metric_type_constants_validate() {
        for value in [
            METRIC_TYPE_COUNTER,
            METRIC_TYPE_UPDOWN_COUNTER,
            METRIC_TYPE_GAUGE,
            METRIC_TYPE_HISTOGRAM,
            METRIC_TYPE_SUMMARY,
        ] {
            assert!(
                validate_semantic_option(SEMANTIC_METRIC_TYPE, value),
                "metric.type value `{value}` must be in the vocabulary domain"
            );
        }
        for value in ["delta", "cumulative"] {
            assert!(validate_semantic_option(SEMANTIC_METRIC_TEMPORALITY, value));
        }
    }

    #[test]
    fn test_record_monotonic_sum() {
        let metric = Metric {
            name: "claude_code.cost.usage".to_string(),
            unit: "USD".to_string(),
            data: Some(metric::Data::Sum(Sum {
                aggregation_temporality: AggregationTemporality::Delta as i32,
                is_monotonic: true,
                ..Default::default()
            })),
            ..Default::default()
        };
        let index = record(
            &metric,
            MetricType::MonotonicSum,
            "claude_code_cost_usage_USD_total",
        );
        let decoded = decode(&index);
        let t = &decoded["claude_code_cost_usage_USD_total"];

        assert_eq!(
            t.get(SEMANTIC_METRIC_TYPE).map(String::as_str),
            Some("counter")
        );
        assert_eq!(
            t.get(SEMANTIC_METRIC_TEMPORALITY).map(String::as_str),
            Some("delta")
        );
        assert_eq!(t.get(SEMANTIC_METRIC_UNIT).map(String::as_str), Some("USD"));
        assert_eq!(
            t.get(SEMANTIC_METRIC_ORIGINAL_NAME).map(String::as_str),
            Some("claude_code.cost.usage")
        );
        assert_eq!(
            t.get(SEMANTIC_METRIC_METADATA_QUALITY).map(String::as_str),
            Some("declared")
        );
    }

    #[test]
    fn test_record_non_monotonic_sum() {
        let metric = Metric {
            name: "queue_size".to_string(),
            data: Some(metric::Data::Sum(Sum {
                aggregation_temporality: AggregationTemporality::Cumulative as i32,
                is_monotonic: false,
                ..Default::default()
            })),
            ..Default::default()
        };
        let index = record(&metric, MetricType::NonMonotonicSum, "queue_size");
        let decoded = decode(&index);
        let t = &decoded["queue_size"];
        assert_eq!(
            t.get(SEMANTIC_METRIC_TYPE).map(String::as_str),
            Some("updown_counter")
        );
        assert_eq!(
            t.get(SEMANTIC_METRIC_TEMPORALITY).map(String::as_str),
            Some("cumulative")
        );
        // Name unchanged by translation -> no original_name.
        assert_eq!(t.get(SEMANTIC_METRIC_ORIGINAL_NAME), None);
    }

    #[test]
    fn test_record_gauge_has_no_temporality() {
        let metric = Metric {
            name: "temperature".to_string(),
            data: Some(metric::Data::Gauge(Gauge::default())),
            ..Default::default()
        };
        let index = record(&metric, MetricType::Gauge, "temperature");
        let decoded = decode(&index);
        let t = &decoded["temperature"];
        assert_eq!(
            t.get(SEMANTIC_METRIC_TYPE).map(String::as_str),
            Some("gauge")
        );
        assert_eq!(t.get(SEMANTIC_METRIC_TEMPORALITY), None);
    }

    #[test]
    fn test_record_histogram_fans_out_with_distinct_types() {
        let metric = Metric {
            name: "request.duration".to_string(),
            unit: "s".to_string(),
            data: Some(metric::Data::Histogram(Histogram {
                aggregation_temporality: AggregationTemporality::Cumulative as i32,
                ..Default::default()
            })),
            ..Default::default()
        };
        let index = record(&metric, MetricType::Histogram, "request_duration");
        let decoded = decode(&index);

        let bucket = &decoded["request_duration_bucket"];
        assert_eq!(
            bucket.get(SEMANTIC_METRIC_TYPE).map(String::as_str),
            Some("histogram")
        );
        assert_eq!(
            bucket.get(SEMANTIC_METRIC_UNIT).map(String::as_str),
            Some("s")
        );

        for companion in ["request_duration_sum", "request_duration_count"] {
            let t = &decoded[companion];
            assert_eq!(
                t.get(SEMANTIC_METRIC_TYPE).map(String::as_str),
                Some("counter")
            );
            assert_eq!(
                t.get(SEMANTIC_METRIC_TEMPORALITY).map(String::as_str),
                Some("cumulative")
            );
        }
    }

    #[test]
    fn test_record_summary_fans_out() {
        let metric = Metric {
            name: "rpc.latency".to_string(),
            data: Some(metric::Data::Summary(Summary::default())),
            ..Default::default()
        };
        let index = record(&metric, MetricType::Summary, "rpc_latency");
        let decoded = decode(&index);

        assert_eq!(
            decoded["rpc_latency"]
                .get(SEMANTIC_METRIC_TYPE)
                .map(String::as_str),
            Some("summary")
        );
        // Summary has no temporality.
        assert_eq!(
            decoded["rpc_latency"].get(SEMANTIC_METRIC_TEMPORALITY),
            None
        );
        for companion in ["rpc_latency_count", "rpc_latency_sum"] {
            assert_eq!(
                decoded[companion]
                    .get(SEMANTIC_METRIC_TYPE)
                    .map(String::as_str),
                Some("counter")
            );
        }
    }

    fn exponential_buckets(
        offset: i32,
        bucket_counts: Vec<u64>,
    ) -> exponential_histogram_data_point::Buckets {
        exponential_histogram_data_point::Buckets {
            offset,
            bucket_counts,
        }
    }

    fn exponential_point() -> ExponentialHistogramDataPoint {
        ExponentialHistogramDataPoint {
            start_time_unix_nano: 1_000_000,
            time_unix_nano: 2_000_000,
            count: 28,
            sum: None,
            scale: 9,
            zero_count: 7,
            positive: Some(exponential_buckets(-3, vec![1, 2, 3, 4, 5, 6])),
            zero_threshold: 0.0,
            ..Default::default()
        }
    }

    fn native_field(value: &ValueData, name: &str) -> Option<ValueData> {
        let ValueData::StructValue(value) = value else {
            panic!("expected native histogram Struct value");
        };
        let index = common_query::native_histogram::NATIVE_HISTOGRAM_FIELD_NAMES
            .iter()
            .position(|field| *field == name)
            .unwrap();
        value.items[index].value_data.clone()
    }

    fn i32_list(value: Option<ValueData>) -> Vec<i32> {
        let Some(ValueData::ListValue(value)) = value else {
            panic!("expected i32 list");
        };
        value
            .items
            .into_iter()
            .map(|item| match item.value_data {
                Some(ValueData::I32Value(value)) => value,
                _ => panic!("expected i32 value"),
            })
            .collect()
    }

    fn i64_list(value: Option<ValueData>) -> Vec<i64> {
        let Some(ValueData::ListValue(value)) = value else {
            panic!("expected i64 list");
        };
        value
            .items
            .into_iter()
            .map(|item| match item.value_data {
                Some(ValueData::I64Value(value)) => value,
                _ => panic!("expected i64 value"),
            })
            .collect()
    }

    #[test]
    fn test_downscale_bucket_index_uses_signed_floor_division() {
        assert_eq!(downscale_bucket_index(-3, 1).unwrap(), -2);
        assert_eq!(downscale_bucket_index(-2, 1).unwrap(), -1);
        assert_eq!(downscale_bucket_index(-1, 1).unwrap(), -1);
        assert_eq!(downscale_bucket_index(0, 1).unwrap(), 0);
        assert_eq!(downscale_bucket_index(1, 1).unwrap(), 0);
        assert_eq!(downscale_bucket_index(2, 1).unwrap(), 1);
        assert_eq!(downscale_bucket_index(i32::MIN, 32).unwrap(), -1);
        assert_eq!(downscale_bucket_index(i32::MAX, 32).unwrap(), 0);
    }

    #[test]
    fn test_convert_bucket_range_downscales_before_prometheus_shift() {
        let buckets = exponential_buckets(-3, vec![1, 2, 3, 4, 5, 6]);
        let (spans, deltas, total) = convert_bucket_range("positive", Some(&buckets), 1).unwrap();

        assert_eq!(
            spans,
            vec![BucketSpan {
                offset: -1,
                length: 4
            }]
        );
        assert_eq!(deltas, vec![1, 4, 4, -3]);
        assert_eq!(total, 21);
    }

    #[test]
    fn test_convert_bucket_range_compacts_zero_runs() {
        let large = (1u64 << 53) + 1;
        for (offset, counts, shift, expected_spans, expected_deltas, expected_total) in [
            (
                -3,
                vec![0, 3, 0, 0, 7, 9, 0],
                0,
                vec![(-1, 1), (2, 2)],
                vec![3, 4, 2],
                19,
            ),
            (
                -4,
                vec![0, 0, 2, 3, 0, 0, 0, 0, 7, 0],
                1,
                vec![(0, 1), (2, 1)],
                vec![5, 2],
                12,
            ),
            (0, vec![0, 0, 0], 0, vec![], vec![], 0),
            (0, vec![], 0, vec![], vec![], 0),
            (i32::MIN, vec![0, 1], 0, vec![(i32::MIN + 2, 1)], vec![1], 1),
            (i32::MAX - 1, vec![1], 0, vec![(i32::MAX, 1)], vec![1], 1),
            (
                0,
                vec![large, 0, large + 2],
                0,
                vec![(1, 1), (1, 1)],
                vec![large as i64, 2],
                large * 2 + 2,
            ),
        ] {
            let buckets = exponential_buckets(offset, counts);
            let (spans, deltas, total) =
                convert_bucket_range("positive", Some(&buckets), shift).unwrap();
            assert_eq!(
                spans
                    .iter()
                    .map(|span| (span.offset, span.length))
                    .collect::<Vec<_>>(),
                expected_spans,
                "{buckets:?}, shift={shift}"
            );
            assert_eq!(deltas, expected_deltas, "{buckets:?}, shift={shift}");
            assert_eq!(total, expected_total);
        }

        // Compaction must not hide invalid source indexes, even for empty buckets.
        let buckets = exponential_buckets(i32::MAX, vec![0]);
        assert!(
            convert_bucket_range("positive", Some(&buckets), 0)
                .unwrap_err()
                .contains("shifted bucket index overflows")
        );
    }

    #[test]
    fn test_exponential_histogram_compaction_preserves_integer_counts() {
        use common_query::native_histogram::{
            COUNT_I64_FIELD, NEGATIVE_BUCKETS_I64_FIELD, NEGATIVE_SPAN_LENGTHS_FIELD,
            NEGATIVE_SPAN_OFFSETS_FIELD, POSITIVE_BUCKETS_I64_FIELD, POSITIVE_SPAN_LENGTHS_FIELD,
            POSITIVE_SPAN_OFFSETS_FIELD,
        };

        let large = (1u64 << 53) + 1;
        let buckets = exponential_buckets(-2, vec![0, large, 0, 3, 0]);
        let point = ExponentialHistogramDataPoint {
            count: 2 * (large + 3),
            positive: Some(buckets.clone()),
            negative: Some(buckets),
            ..Default::default()
        };
        let (value, _) =
            exponential_histogram_value(&point, AggregationTemporality::Cumulative).unwrap();
        assert_eq!(
            native_field(&value, COUNT_I64_FIELD),
            Some(ValueData::I64Value(point.count as i64))
        );
        for (offsets, lengths, counts) in [
            (
                POSITIVE_SPAN_OFFSETS_FIELD,
                POSITIVE_SPAN_LENGTHS_FIELD,
                POSITIVE_BUCKETS_I64_FIELD,
            ),
            (
                NEGATIVE_SPAN_OFFSETS_FIELD,
                NEGATIVE_SPAN_LENGTHS_FIELD,
                NEGATIVE_BUCKETS_I64_FIELD,
            ),
        ] {
            assert_eq!(i32_list(native_field(&value, offsets)), vec![0, 1]);
            assert_eq!(i32_list(native_field(&value, lengths)), vec![1, 1]);
            assert_eq!(
                i64_list(native_field(&value, counts)),
                vec![large as i64, 3]
            );
        }
    }

    #[test]
    fn test_exponential_histogram_value_uses_integer_family() {
        use common_query::native_histogram::{
            COUNT_F64_FIELD, COUNT_I64_FIELD, POSITIVE_BUCKETS_I64_FIELD,
            POSITIVE_SPAN_LENGTHS_FIELD, POSITIVE_SPAN_OFFSETS_FIELD, SCHEMA_FIELD, SUM_FIELD,
            ZERO_COUNT_I64_FIELD,
        };

        let (value, timestamp_nanos) =
            exponential_histogram_value(&exponential_point(), AggregationTemporality::Cumulative)
                .unwrap();

        assert_eq!(timestamp_nanos, 2_000_000);
        assert_eq!(
            native_field(&value, SCHEMA_FIELD),
            Some(ValueData::I32Value(8))
        );
        assert_eq!(
            native_field(&value, COUNT_I64_FIELD),
            Some(ValueData::I64Value(28))
        );
        assert_eq!(
            native_field(&value, ZERO_COUNT_I64_FIELD),
            Some(ValueData::I64Value(7))
        );
        assert_eq!(native_field(&value, COUNT_F64_FIELD), None);
        assert_eq!(
            i32_list(native_field(&value, POSITIVE_SPAN_OFFSETS_FIELD)),
            vec![-1]
        );
        assert_eq!(
            i32_list(native_field(&value, POSITIVE_SPAN_LENGTHS_FIELD)),
            vec![4]
        );
        assert_eq!(
            i64_list(native_field(&value, POSITIVE_BUCKETS_I64_FIELD)),
            vec![1, 5, 9, 6]
        );
        let Some(ValueData::F64Value(sum)) = native_field(&value, SUM_FIELD) else {
            panic!("expected histogram sum");
        };
        assert_eq!(sum.to_bits(), f64::NAN.to_bits());
        assert_ne!(sum.to_bits(), PROMETHEUS_STALE_NAN_BITS);
    }

    #[test]
    fn test_no_recorded_value_ignores_other_value_fields() {
        use common_query::native_histogram::{
            COUNT_I64_FIELD, POSITIVE_BUCKETS_I64_FIELD, SCHEMA_FIELD, SUM_FIELD,
            ZERO_COUNT_I64_FIELD, ZERO_THRESHOLD_FIELD,
        };

        let point = ExponentialHistogramDataPoint {
            start_time_unix_nano: 1_000_000,
            time_unix_nano: 2_000_000,
            count: u64::MAX,
            sum: Some(1.0),
            scale: i32::MIN,
            zero_count: u64::MAX,
            positive: Some(exponential_buckets(i32::MAX, vec![u64::MAX])),
            flags: DataPointFlags::NoRecordedValueMask as u32,
            zero_threshold: f64::NAN,
            ..Default::default()
        };
        let (value, _) =
            exponential_histogram_value(&point, AggregationTemporality::Cumulative).unwrap();

        assert_eq!(
            native_field(&value, SCHEMA_FIELD),
            Some(ValueData::I32Value(0))
        );
        assert_eq!(
            native_field(&value, ZERO_THRESHOLD_FIELD),
            Some(ValueData::F64Value(0.0))
        );
        assert_eq!(
            native_field(&value, COUNT_I64_FIELD),
            Some(ValueData::I64Value(0))
        );
        assert_eq!(
            native_field(&value, ZERO_COUNT_I64_FIELD),
            Some(ValueData::I64Value(0))
        );
        assert!(i64_list(native_field(&value, POSITIVE_BUCKETS_I64_FIELD)).is_empty());
        let Some(ValueData::F64Value(sum)) = native_field(&value, SUM_FIELD) else {
            panic!("expected histogram sum");
        };
        assert_eq!(sum.to_bits(), PROMETHEUS_STALE_NAN_BITS);
    }

    #[test]
    fn test_non_flag_nan_sum_is_normalized() {
        use common_query::native_histogram::SUM_FIELD;

        let point = ExponentialHistogramDataPoint {
            sum: Some(f64::from_bits(PROMETHEUS_STALE_NAN_BITS)),
            ..Default::default()
        };
        let (value, _) =
            exponential_histogram_value(&point, AggregationTemporality::Cumulative).unwrap();
        let Some(ValueData::F64Value(sum)) = native_field(&value, SUM_FIELD) else {
            panic!("expected histogram sum");
        };
        assert_eq!(sum.to_bits(), f64::NAN.to_bits());
        assert_ne!(sum.to_bits(), PROMETHEUS_STALE_NAN_BITS);
    }

    #[test]
    fn test_exponential_histogram_rejects_invalid_values() {
        let mut cases = Vec::new();

        let mut point = exponential_point();
        point.scale = -5;
        cases.push((point, "scale -5 is unsupported"));

        let mut point = exponential_point();
        point.zero_threshold = f64::INFINITY;
        cases.push((point, "must be finite and non-negative"));

        let mut point = exponential_point();
        point.start_time_unix_nano = point.time_unix_nano + 1;
        cases.push((point, "start_time_unix_nano"));

        let mut point = exponential_point();
        point.count = 27;
        cases.push((point, "declared count is 27"));

        let point = ExponentialHistogramDataPoint {
            count: u64::MAX,
            zero_count: u64::MAX,
            ..Default::default()
        };
        cases.push((point, "count 18446744073709551615 overflows i64"));

        let point = ExponentialHistogramDataPoint {
            count: 1,
            scale: 8,
            positive: Some(exponential_buckets(i32::MAX, vec![1])),
            ..Default::default()
        };
        cases.push((point, "shifted bucket index overflows i32"));

        for (point, expected) in cases {
            let error = exponential_histogram_value(&point, AggregationTemporality::Cumulative)
                .unwrap_err();
            assert!(
                error.contains(expected),
                "expected {expected:?}, got {error}"
            );
        }

        let buckets = exponential_buckets(0, vec![u64::MAX, 1]);
        let error = convert_bucket_range("positive", Some(&buckets), 1).unwrap_err();
        assert!(
            error.contains("merged bucket count overflows u64"),
            "{error}"
        );

        let buckets = exponential_buckets(i32::MAX, vec![1, 1]);
        let error = convert_bucket_range("positive", Some(&buckets), 1).unwrap_err();
        assert!(error.contains("bucket index overflows i32"), "{error}");
    }

    fn metrics_request(metrics: Vec<Metric>) -> ExportMetricsServiceRequest {
        ExportMetricsServiceRequest {
            resource_metrics: vec![ResourceMetrics {
                scope_metrics: vec![ScopeMetrics {
                    metrics,
                    ..Default::default()
                }],
                ..Default::default()
            }],
        }
    }

    #[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>,
        temporality: AggregationTemporality,
    ) -> Metric {
        Metric {
            name: name.into(),
            data: Some(metric::Data::ExponentialHistogram(ExponentialHistogram {
                data_points,
                aggregation_temporality: temporality as i32,
            })),
            ..Default::default()
        }
    }

    fn histogram_metric(name: impl Into<String>) -> Metric {
        Metric {
            name: name.into(),
            data: Some(metric::Data::Histogram(Histogram {
                data_points: vec![HistogramDataPoint {
                    start_time_unix_nano: 1_000_000,
                    time_unix_nano: 2_000_000,
                    count: 1,
                    sum: Some(1.0),
                    bucket_counts: vec![1],
                    ..Default::default()
                }],
                aggregation_temporality: AggregationTemporality::Cumulative as i32,
            })),
            ..Default::default()
        }
    }

    #[test]
    fn test_metric_engine_path_keeps_nanosecond_precision() {
        let time_unix_nano = 1_704_067_200_123_456_789u64;
        let request = metrics_request(vec![Metric {
            name: "my_gauge".to_string(),
            data: Some(metric::Data::Gauge(Gauge {
                data_points: vec![NumberDataPoint {
                    time_unix_nano,
                    value: Some(Value::AsDouble(1.0)),
                    ..Default::default()
                }],
            })),
            ..Default::default()
        }]);

        // The metric engine path keeps the full nanosecond precision here;
        // `Inserter::handle_metric_row_inserts` converts the timestamps to
        // the physical table's time index unit afterwards.
        let mut metric_ctx = OtlpMetricCtx {
            with_metric_engine: true,
            ..Default::default()
        };
        let MetricsConversion { requests, .. } =
            to_grpc_insert_requests(request, &mut metric_ctx).unwrap();

        let rows = requests.inserts[0].rows.as_ref().unwrap();
        let ts_index = rows
            .schema
            .iter()
            .position(|column| column.column_name == greptime_timestamp())
            .unwrap();
        assert_eq!(
            rows.schema[ts_index].datatype,
            ColumnDataType::TimestampNanosecond as i32
        );
        assert!(matches!(
            rows.rows[0].values[ts_index].value_data,
            Some(ValueData::TimestampNanosecondValue(
                1_704_067_200_123_456_789
            ))
        ));

        // The non-metric prometheus-compatible path stays millisecond so
        // auto-created mito tables keep the millisecond time index.
        let mut compat_ctx = OtlpMetricCtx::default();
        let request = metrics_request(vec![Metric {
            name: "my_gauge".to_string(),
            data: Some(metric::Data::Gauge(Gauge {
                data_points: vec![NumberDataPoint {
                    time_unix_nano,
                    value: Some(Value::AsDouble(1.0)),
                    ..Default::default()
                }],
            })),
            ..Default::default()
        }]);
        let MetricsConversion { requests, .. } =
            to_grpc_insert_requests(request, &mut compat_ctx).unwrap();

        let rows = requests.inserts[0].rows.as_ref().unwrap();
        let ts_index = rows
            .schema
            .iter()
            .position(|column| column.column_name == greptime_timestamp())
            .unwrap();
        assert_eq!(
            rows.schema[ts_index].datatype,
            ColumnDataType::TimestampMillisecond as i32
        );
        assert!(matches!(
            rows.rows[0].values[ts_index].value_data,
            Some(ValueData::TimestampMillisecondValue(1_704_067_200_123))
        ));
    }

    #[test]
    fn test_exponential_histogram_rejection_metrics() {
        // Other conversion tests update the same process-global counters.
        const ISOLATED_ENV: &str = "GREPTIME_TEST_OTLP_REJECTION_METRICS_ISOLATED";
        if std::env::var_os(ISOLATED_ENV).is_none() {
            let output = std::process::Command::new(std::env::current_exe().unwrap())
                .args([
                    "--exact",
                    "otlp::metrics::tests::test_exponential_histogram_rejection_metrics",
                ])
                .env(ISOLATED_ENV, "1")
                .output()
                .unwrap();
            let stdout = String::from_utf8_lossy(&output.stdout);
            assert!(
                output.status.success() && stdout.contains("1 passed"),
                "isolated rejection metrics test failed\nstdout:\n{stdout}\nstderr:\n{}",
                String::from_utf8_lossy(&output.stderr)
            );
            return;
        }

        let mut invalid = exponential_point();
        invalid.scale = -5;
        for (temporality, points, reason, accepted, rejected) in [
            (
                AggregationTemporality::Delta,
                vec![exponential_point(); 2],
                "invalid_data_point",
                2,
                0,
            ),
            (
                AggregationTemporality::Unspecified,
                vec![exponential_point(); 2],
                "unspecified_temporality",
                0,
                2,
            ),
            (
                AggregationTemporality::Cumulative,
                vec![exponential_point(), invalid],
                "invalid_data_point",
                1,
                1,
            ),
            (
                AggregationTemporality::Cumulative,
                vec![exponential_point()],
                "invalid_data_point",
                1,
                0,
            ),
        ] {
            let counter =
                OTLP_EXPONENTIAL_HISTOGRAM_REJECTED_DATA_POINTS.with_label_values(&[reason]);
            let before = counter.get();
            let mut request =
                metrics_request(vec![exponential_metric("latency", points, temporality)]);
            request.resource_metrics[0].resource = Some(Resource {
                attributes: vec![keyvalue("service.name", "api")],
                ..Default::default()
            });
            let mut ctx = descriptor_ctx();
            let conversion = to_grpc_insert_requests(request, &mut ctx).unwrap();
            assert_eq!(conversion.outcome.accepted_data_points, accepted);
            assert_eq!(conversion.outcome.rejected_data_points, rejected);
            assert_eq!(counter.get() - before, rejected as u64, "{reason}");
            assert_eq!(conversion.resource_info.is_some(), accepted > 0);

            let before = counter.get();
            let empty = metrics_request(vec![exponential_metric("empty", vec![], temporality)]);
            assert_eq!(
                to_grpc_insert_requests(empty, &mut ctx)
                    .unwrap()
                    .outcome
                    .rejected_data_points,
                0
            );
            assert_eq!(counter.get(), before, "empty metric: {reason}");
        }
    }

    #[test]
    fn test_exponential_histogram_default_context_and_partial_outcome() {
        let request = metrics_request(vec![
            Metric {
                name: "temperature".to_string(),
                data: Some(metric::Data::Gauge(Gauge {
                    data_points: vec![NumberDataPoint::default()],
                })),
                ..Default::default()
            },
            exponential_metric(
                "latency",
                vec![exponential_point()],
                AggregationTemporality::Cumulative,
            ),
            exponential_metric(
                "delta_latency",
                vec![exponential_point()],
                AggregationTemporality::Delta,
            ),
        ]);
        let MetricsConversion {
            requests,
            semantic_index,
            outcome,
            ..
        } = to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap();

        assert_eq!(outcome.accepted_data_points, 3);
        assert_eq!(outcome.rejected_data_points, 0);
        assert!(outcome.error_message.is_none());
        assert_eq!(requests.inserts.len(), 3);
        let semantics = decode(&semantic_index);
        assert!(semantics.contains_key("temperature"));
        assert!(semantics.contains_key("latency"));
        assert_eq!(
            semantics["delta_latency"][SEMANTIC_METRIC_TEMPORALITY],
            "delta"
        );

        let empty = metrics_request(vec![exponential_metric(
            "empty",
            vec![],
            AggregationTemporality::Cumulative,
        )]);
        let outcome = to_grpc_insert_requests(empty, &mut OtlpMetricCtx::default())
            .unwrap()
            .outcome;
        assert_eq!(outcome.rejected_data_points, 0);
        assert_eq!(outcome.error_message, None);
    }

    #[test]
    fn test_exponential_histogram_cannot_share_table_with_scalar_metric() {
        let request = metrics_request(vec![
            Metric {
                name: "latency".to_string(),
                data: Some(metric::Data::Gauge(Gauge {
                    data_points: vec![NumberDataPoint {
                        value: Some(number_data_point::Value::AsDouble(1.0)),
                        ..Default::default()
                    }],
                })),
                ..Default::default()
            },
            exponential_metric(
                "latency",
                vec![exponential_point()],
                AggregationTemporality::Cumulative,
            ),
        ]);
        let mut ctx = OtlpMetricCtx::default();

        let conversion = to_grpc_insert_requests(request, &mut ctx).unwrap();
        assert_eq!(conversion.outcome.accepted_data_points, 0);
        assert_eq!(conversion.outcome.rejected_data_points, 2);
        assert!(conversion.requests.inserts.is_empty());
    }

    #[test]
    fn test_histogram_cannot_replace_exponential_histogram_table() {
        let request = metrics_request(vec![
            exponential_metric(
                "latency_bucket",
                vec![exponential_point()],
                AggregationTemporality::Cumulative,
            ),
            histogram_metric("latency"),
        ]);
        let mut ctx = OtlpMetricCtx::default();

        let conversion = to_grpc_insert_requests(request, &mut ctx).unwrap();
        assert_eq!(conversion.outcome.accepted_data_points, 0);
        assert_eq!(conversion.outcome.rejected_data_points, 2);
        assert!(conversion.requests.inserts.is_empty());
    }

    #[test]
    fn test_histograms_with_same_name_across_resources_are_merged() {
        let metric = histogram_metric("latency");
        let request = ExportMetricsServiceRequest {
            resource_metrics: ["service-a", "service-b"]
                .into_iter()
                .map(|service| ResourceMetrics {
                    resource: Some(Resource {
                        attributes: vec![keyvalue("service.name", service)],
                        ..Default::default()
                    }),
                    scope_metrics: vec![ScopeMetrics {
                        metrics: vec![metric.clone()],
                        ..Default::default()
                    }],
                    ..Default::default()
                })
                .collect(),
        };

        let MetricsConversion {
            requests,
            rows,
            outcome,
            ..
        } = to_grpc_insert_requests(request, &mut OtlpMetricCtx::default()).unwrap();

        assert_eq!(outcome.accepted_data_points, 2);
        assert_eq!(rows, 6);
        assert_eq!(requests.inserts.len(), 3);
        for request in requests.inserts {
            assert_eq!(
                request.rows.unwrap().rows.len(),
                2,
                "{}",
                request.table_name
            );
        }
    }

    #[test]
    fn test_exponential_histogram_rejects_temporality_before_stale_point() {
        let stale = ExponentialHistogramDataPoint {
            flags: DataPointFlags::NoRecordedValueMask as u32,
            ..Default::default()
        };
        let request = metrics_request(vec![exponential_metric(
            "latency",
            vec![stale],
            AggregationTemporality::Unspecified,
        )]);
        let mut ctx = OtlpMetricCtx::default();
        let MetricsConversion {
            requests,
            rows,
            semantic_index,
            outcome,
            ..
        } = to_grpc_insert_requests(request, &mut ctx).unwrap();

        assert_eq!(outcome.accepted_data_points, 0);
        assert_eq!(outcome.rejected_data_points, 1);
        assert_eq!(rows, 0);
        assert!(requests.inserts.is_empty());
        assert!(semantic_index.is_empty());
    }

    #[test]
    fn test_exponential_histogram_storage_paths_share_struct() {
        use common_query::prelude::greptime_native_histogram;

        let mut point = exponential_point();
        point.sum = Some(42.0);
        let request = metrics_request(vec![exponential_metric(
            "request.duration",
            vec![point],
            AggregationTemporality::Cumulative,
        )]);
        let mut ordinary_ctx = OtlpMetricCtx::default();
        let ordinary_requests = to_grpc_insert_requests(request.clone(), &mut ordinary_ctx)
            .unwrap()
            .requests;
        let mut metric_ctx = OtlpMetricCtx {
            with_metric_engine: true,
            ..Default::default()
        };
        let metric_requests = to_grpc_insert_requests(request, &mut metric_ctx)
            .unwrap()
            .requests;

        let ordinary_insert = &ordinary_requests.inserts[0];
        let metric_insert = &metric_requests.inserts[0];
        assert_eq!(ordinary_insert.table_name, "request_duration");
        assert_eq!(metric_insert.table_name, "request_duration");
        let ordinary_rows = ordinary_insert.rows.as_ref().unwrap();
        let metric_rows = metric_insert.rows.as_ref().unwrap();
        let field = greptime_native_histogram();
        let ordinary_histogram = ordinary_rows.rows[0].values[ordinary_rows
            .schema
            .iter()
            .position(|column| column.column_name == field)
            .unwrap()]
        .clone();
        let metric_histogram = metric_rows.rows[0].values[metric_rows
            .schema
            .iter()
            .position(|column| column.column_name == field)
            .unwrap()]
        .clone();
        assert_eq!(ordinary_histogram, metric_histogram);
        assert!(matches!(
            ordinary_rows.rows[0].values[ordinary_rows
                .schema
                .iter()
                .position(|column| column.column_name == greptime_timestamp())
                .unwrap()]
            .value_data,
            Some(ValueData::TimestampMillisecondValue(2))
        ));
        assert!(matches!(
            metric_rows.rows[0].values[metric_rows
                .schema
                .iter()
                .position(|column| column.column_name == greptime_timestamp())
                .unwrap()]
            .value_data,
            Some(ValueData::TimestampNanosecondValue(2_000_000))
        ));
    }

    #[test]
    fn test_rejection_message_is_bounded() {
        let request = metrics_request(vec![exponential_metric(
            "x".repeat(1_000),
            vec![ExponentialHistogramDataPoint::default()],
            AggregationTemporality::Unspecified,
        )]);
        let outcome = to_grpc_insert_requests(request, &mut OtlpMetricCtx::default())
            .unwrap()
            .outcome;

        assert_eq!(outcome.rejected_data_points, 1);
        assert!(outcome.error_message.unwrap().len() <= MAX_REJECTION_MESSAGE_BYTES);
    }

    #[test]
    fn test_rejection_message_reason_is_lazy_after_cap() {
        let mut outcome = MetricsIngestOutcome {
            error_message: Some("x".repeat(MAX_REJECTION_MESSAGE_BYTES)),
            ..Default::default()
        };
        let mut reason_built = false;

        reject_data_points(&mut outcome, 1, || {
            reason_built = true;
            "unused".to_string()
        })
        .unwrap();

        assert_eq!(outcome.rejected_data_points, 1);
        assert!(!reason_built);
    }
}