servers/otlp/trace/

v1.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 std::collections::{HashMap, HashSet};

use api::v1::column_data_type_extension::TypeExt;
use api::v1::helper::time_index_column_schema;
use api::v1::value::ValueData;
use api::v1::{
    ColumnDataType, ColumnDataTypeExtension, ColumnSchema, JsonTypeExtension, RowInsertRequests,
    SemanticType, Value,
};
use common_catalog::consts::{trace_operations_table_name, trace_services_table_name};
use common_grpc::precision::Precision;
use opentelemetry_proto::tonic::common::v1::KeyValue;
use opentelemetry_proto::tonic::common::v1::any_value::Value as OtlpValue;

use crate::error::Result;
use crate::otlp::trace::attributes::Attributes;
use crate::otlp::trace::span::TraceSpan;
use crate::otlp::trace::{
    DURATION_NANO_COLUMN, KEY_SERVICE_NAME, PARENT_SPAN_ID_COLUMN, SCOPE_NAME_COLUMN,
    SCOPE_VERSION_COLUMN, SERVICE_NAME_COLUMN, SPAN_EVENTS_COLUMN, SPAN_ID_COLUMN,
    SPAN_KIND_COLUMN, SPAN_NAME_COLUMN, SPAN_STATUS_CODE, SPAN_STATUS_MESSAGE_COLUMN,
    TIMESTAMP_COLUMN, TIMESTAMP_END_COLUMN, TRACE_ID_COLUMN, TRACE_STATE_COLUMN, TraceAuxData,
};
use crate::otlp::utils::any_value_to_jsonb;
use crate::row_writer::{self, MultiTableData, TableData};

const APPROXIMATE_COLUMN_COUNT: usize = 30;

// Use a timestamp(2100-01-01 00:00:00) as large as possible.
const MAX_TIMESTAMP: i64 = 4102444800000000000;

struct FixedTraceColumnIndexes {
    timestamp: usize,
    timestamp_end: usize,
    duration_nano: usize,
    parent_span_id: usize,
    trace_id: usize,
    span_id: usize,
    span_kind: usize,
    span_name: usize,
    span_status_code: usize,
    span_status_message: usize,
    trace_state: usize,
    scope_name: usize,
    scope_version: usize,
}

impl FixedTraceColumnIndexes {
    fn resolve(writer: &mut TableData) -> Result<Self> {
        let timestamp = writer.ensure_column(time_index_column_schema(
            TIMESTAMP_COLUMN,
            ColumnDataType::TimestampNanosecond,
        ))?;
        let mut field = |name: &str, datatype: ColumnDataType| {
            writer.ensure_column(ColumnSchema {
                column_name: name.to_string(),
                datatype: datatype as i32,
                semantic_type: SemanticType::Field as i32,
                ..Default::default()
            })
        };

        Ok(Self {
            timestamp,
            timestamp_end: field(TIMESTAMP_END_COLUMN, ColumnDataType::TimestampNanosecond)?,
            duration_nano: field(DURATION_NANO_COLUMN, ColumnDataType::Int64)?,
            parent_span_id: field(PARENT_SPAN_ID_COLUMN, ColumnDataType::String)?,
            trace_id: field(TRACE_ID_COLUMN, ColumnDataType::String)?,
            span_id: field(SPAN_ID_COLUMN, ColumnDataType::String)?,
            span_kind: field(SPAN_KIND_COLUMN, ColumnDataType::String)?,
            span_name: field(SPAN_NAME_COLUMN, ColumnDataType::String)?,
            span_status_code: field(SPAN_STATUS_CODE, ColumnDataType::String)?,
            span_status_message: field(SPAN_STATUS_MESSAGE_COLUMN, ColumnDataType::String)?,
            trace_state: field(TRACE_STATE_COLUMN, ColumnDataType::String)?,
            scope_name: field(SCOPE_NAME_COLUMN, ColumnDataType::String)?,
            scope_version: field(SCOPE_VERSION_COLUMN, ColumnDataType::String)?,
        })
    }
}

/// Distinguishes raw bytes from JSONB values that both use a binary protobuf value.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TraceBinaryType {
    /// Raw OTLP bytes with no datatype extension.
    Binary,
    /// An OTLP array or key-value list encoded with the JSONB extension.
    Json,
}

impl TraceBinaryType {
    /// Applies this logical binary type to a column schema.
    pub fn apply_to_schema(self, schema: &mut ColumnSchema) {
        schema.datatype = ColumnDataType::Binary as i32;
        schema.datatype_extension = match self {
            Self::Binary => None,
            Self::Json => Some(ColumnDataTypeExtension {
                type_ext: Some(TypeExt::JsonType(JsonTypeExtension::JsonBinary.into())),
            }),
        };
    }
}

/// Per-column observations collected while building one trace chunk.
#[derive(Default)]
struct TraceBatchColumnSchema {
    value_types: Vec<ColumnDataType>,
    present_rows: Vec<usize>,
    binary_types: Vec<(usize, TraceBinaryType)>,
}

impl TraceBatchColumnSchema {
    /// Records a row once while preserving encounter order.
    fn observe_row(&mut self, row_index: usize) {
        if self.present_rows.last() != Some(&row_index) {
            self.present_rows.push(row_index);
        }
    }

    /// Returns whether raw binary and JSONB values share this column.
    fn has_incompatible_logical_types(&self) -> bool {
        let Some((_, first_type)) = self.binary_types.first() else {
            return false;
        };
        self.binary_types
            .iter()
            .any(|(_, binary_type)| binary_type != first_type)
    }
}

/// Sparse column metadata retained for span-level fallback.
pub struct TraceRetryColumn {
    /// Row indexes that contain this dynamic column.
    pub present_rows: Vec<usize>,
    /// Row-specific binary kinds needed to restore the logical schema.
    pub binary_types: Vec<(usize, TraceBinaryType)>,
}

/// Retry metadata keyed by dynamic trace column name.
pub type TraceRetryColumns = HashMap<String, TraceRetryColumn>;

/// Schema observations for dynamic columns in one converted trace chunk.
#[derive(Default)]
pub struct TraceBatchSchema {
    columns: HashMap<String, TraceBatchColumnSchema>,
}

impl TraceBatchSchema {
    /// Records a scalar value type and the row where the column is present.
    fn observe_value_type(&mut self, name: &str, row_index: usize, value_type: ColumnDataType) {
        let column = self.columns.entry(name.to_string()).or_default();
        column.observe_row(row_index);
        if !column.value_types.contains(&value_type) {
            column.value_types.push(value_type);
        }
    }

    /// Records the logical kind of a binary protobuf value for one row.
    fn observe_binary_type(&mut self, name: &str, row_index: usize, binary_type: TraceBinaryType) {
        let column = self.columns.entry(name.to_string()).or_default();
        column.observe_row(row_index);
        if !column.value_types.contains(&ColumnDataType::Binary) {
            column.value_types.push(ColumnDataType::Binary);
        }
        if let Some((last_row_index, last_type)) = column.binary_types.last_mut()
            && *last_row_index == row_index
        {
            *last_type = binary_type;
        } else {
            column.binary_types.push((row_index, binary_type));
        }
    }

    /// Records a sparse column occurrence that has no dynamic value type.
    fn observe_present_column(&mut self, name: &str, row_index: usize) {
        self.columns
            .entry(name.to_string())
            .or_default()
            .observe_row(row_index);
    }

    /// Returns the distinct value types in first-seen order for a column.
    pub fn value_types(&self, name: &str) -> Option<&[ColumnDataType]> {
        self.columns
            .get(name)
            .map(|column| column.value_types.as_slice())
    }

    /// Returns whether any column mixes raw binary and JSONB values.
    pub fn has_incompatible_logical_types(&self) -> bool {
        self.columns
            .values()
            .any(TraceBatchColumnSchema::has_incompatible_logical_types)
    }

    /// Returns whether the named column mixes raw binary and JSONB values.
    pub fn has_incompatible_logical_types_for(&self, name: &str) -> bool {
        self.columns
            .get(name)
            .is_some_and(TraceBatchColumnSchema::has_incompatible_logical_types)
    }

    /// Converts batch observations into sparse metadata for per-span retries.
    pub fn into_retry_columns(self) -> TraceRetryColumns {
        self.columns
            .into_iter()
            .filter(|(_, column)| !column.present_rows.is_empty())
            .map(|(name, column)| {
                let binary_types = if column.has_incompatible_logical_types()
                    || column
                        .value_types
                        .iter()
                        .any(|datatype| *datatype != ColumnDataType::Binary)
                {
                    column.binary_types
                } else {
                    Vec::new()
                };
                (
                    name,
                    TraceRetryColumn {
                        present_rows: column.present_rows,
                        binary_types,
                    },
                )
            })
            .collect()
    }
}

/// Converts trace spans into row insert requests for the main v1 trace table.
///
/// Auxiliary service and operation table writes are built separately so the
/// caller can update them only after the main span write succeeds.
pub fn v1_to_grpc_main_insert_requests(
    spans: &[TraceSpan],
    table_name: &str,
) -> Result<(RowInsertRequests, usize)> {
    let requests = v1_to_grpc_main_insert_requests_from_iter(spans.iter().cloned(), table_name)?;
    Ok((requests, spans.len()))
}

/// Converts owned spans into unpadded main-table rows and schema observations.
pub fn v1_to_main_table_data_with_schema(
    spans: Vec<TraceSpan>,
) -> Result<(TableData, TraceBatchSchema)> {
    build_trace_table_data_with_schema(spans.into_iter())
}

/// Builds the main-table request without collecting batch schema observations.
fn v1_to_grpc_main_insert_requests_from_iter(
    spans: impl ExactSizeIterator<Item = TraceSpan>,
    table_name: &str,
) -> Result<RowInsertRequests> {
    let mut multi_table_writer = MultiTableData::default();
    let trace_writer = build_trace_table_data_from_iter(spans, None)?;
    multi_table_writer.add_table_data(table_name, trace_writer);

    Ok(multi_table_writer.into_row_insert_requests().0)
}

/// Builds the row-oriented payload for the main v1 trace table.
pub fn build_trace_table_data(spans: &[TraceSpan]) -> Result<TableData> {
    build_trace_table_data_from_iter(spans.iter().cloned(), None)
}

/// Builds trace rows while collecting dynamic column observations.
fn build_trace_table_data_with_schema(
    spans: impl ExactSizeIterator<Item = TraceSpan>,
) -> Result<(TableData, TraceBatchSchema)> {
    let mut batch_schema = TraceBatchSchema::default();
    let trace_writer = build_trace_table_data_from_iter(spans, Some(&mut batch_schema))?;
    Ok((trace_writer, batch_schema))
}

/// Shared row builder with optional batch schema observation.
fn build_trace_table_data_from_iter(
    spans: impl ExactSizeIterator<Item = TraceSpan>,
    mut batch_schema: Option<&mut TraceBatchSchema>,
) -> Result<TableData> {
    let mut trace_writer = TableData::new(APPROXIMATE_COLUMN_COUNT, spans.len());
    if spans.len() == 0 {
        return Ok(trace_writer);
    }
    let fixed_columns = FixedTraceColumnIndexes::resolve(&mut trace_writer)?;
    for span in spans {
        let row_index = trace_writer.num_rows();
        write_span_to_row_inner(
            &mut trace_writer,
            span,
            row_index,
            &fixed_columns,
            batch_schema.as_deref_mut(),
        )?;
    }
    Ok(trace_writer)
}

/// Builds row insert requests for the v1 trace auxiliary tables.
pub fn build_aux_table_requests(
    aux_data: TraceAuxData,
    table_name: &str,
) -> Result<(RowInsertRequests, usize)> {
    let mut multi_table_writer = MultiTableData::default();
    let mut trace_services_writer = TableData::new(APPROXIMATE_COLUMN_COUNT, 1);
    let mut trace_operations_writer = TableData::new(APPROXIMATE_COLUMN_COUNT, 1);

    write_trace_services_to_row(&mut trace_services_writer, aux_data.services)?;
    write_trace_operations_to_row(&mut trace_operations_writer, aux_data.operations)?;

    multi_table_writer.add_table_data(trace_services_table_name(table_name), trace_services_writer);
    multi_table_writer.add_table_data(
        trace_operations_table_name(table_name),
        trace_operations_writer,
    );

    Ok(multi_table_writer.into_row_insert_requests())
}

pub fn write_span_to_row(writer: &mut TableData, span: TraceSpan) -> Result<()> {
    let row_index = writer.num_rows();
    let fixed_columns = FixedTraceColumnIndexes::resolve(writer)?;
    write_span_to_row_inner(writer, span, row_index, &fixed_columns, None)
}

/// Computes the span duration as the signed `duration_nano` value written by
/// the v1 data model.
///
/// A span whose end precedes its start carries no meaningful duration; it
/// clamps to 0 instead of wrapping — a negative `duration_nano` would fail
/// the checked Int64→UInt64 coercion into pre-existing unsigned tables and
/// would break unsigned readers such as the Jaeger query API. A duration
/// that does not fit `i64` saturates at `i64::MAX`. Clamping at the source
/// keeps new Int64 tables and existing UInt64 tables behaving identically:
/// the written value is always a non-negative, in-range `i64`.
pub(super) fn span_duration_nano(span: &TraceSpan) -> i64 {
    span.end_in_nanosecond
        .saturating_sub(span.start_in_nanosecond)
        .min(i64::MAX as u64) as i64
}

/// Writes one span and optionally records its dynamic columns for reconciliation.
fn write_span_to_row_inner(
    writer: &mut TableData,
    span: TraceSpan,
    row_index: usize,
    fixed_columns: &FixedTraceColumnIndexes,
    mut batch_schema: Option<&mut TraceBatchSchema>,
) -> Result<()> {
    let mut row = writer.alloc_one_row();

    for (index, value) in [
        (
            fixed_columns.timestamp,
            Some(ValueData::TimestampNanosecondValue(
                span.start_in_nanosecond as i64,
            )),
        ),
        (
            fixed_columns.timestamp_end,
            Some(ValueData::TimestampNanosecondValue(
                span.end_in_nanosecond as i64,
            )),
        ),
        (
            fixed_columns.duration_nano,
            Some(ValueData::I64Value(span_duration_nano(&span))),
        ),
        (
            fixed_columns.parent_span_id,
            span.parent_span_id.map(ValueData::StringValue),
        ),
        (
            fixed_columns.trace_id,
            Some(ValueData::StringValue(span.trace_id)),
        ),
        (
            fixed_columns.span_id,
            Some(ValueData::StringValue(span.span_id)),
        ),
        (
            fixed_columns.span_kind,
            Some(ValueData::StringValue(span.span_kind)),
        ),
        (
            fixed_columns.span_name,
            Some(ValueData::StringValue(span.span_name)),
        ),
        (
            fixed_columns.span_status_code,
            Some(ValueData::StringValue(span.span_status_code)),
        ),
        (
            fixed_columns.span_status_message,
            Some(ValueData::StringValue(span.span_status_message)),
        ),
        (
            fixed_columns.trace_state,
            Some(ValueData::StringValue(span.trace_state)),
        ),
        (
            fixed_columns.scope_name,
            Some(ValueData::StringValue(span.scope_name)),
        ),
        (
            fixed_columns.scope_version,
            Some(ValueData::StringValue(span.scope_version)),
        ),
    ] {
        row[index].value_data = value;
    }

    if let Some(service_name) = span.service_name {
        if let Some(batch_schema) = batch_schema.as_deref_mut() {
            batch_schema.observe_present_column(SERVICE_NAME_COLUMN, row_index);
        }
        row_writer::write_tags(
            writer,
            std::iter::once((SERVICE_NAME_COLUMN.to_string(), service_name)),
            &mut row,
        )?;
    }

    write_attributes_with_schema(
        writer,
        "span_attributes",
        span.span_attributes.take(),
        &mut row,
        row_index,
        batch_schema.as_deref_mut(),
    )?;
    write_shared_attributes_with_schema(
        writer,
        "scope_attributes",
        span.scope_attributes.as_ref(),
        &mut row,
        row_index,
        batch_schema.as_deref_mut(),
    )?;
    write_shared_attributes_with_schema(
        writer,
        "resource_attributes",
        span.resource_attributes.as_ref(),
        &mut row,
        row_index,
        batch_schema,
    )?;

    row_writer::write_json(
        writer,
        SPAN_EVENTS_COLUMN,
        span.span_events.into(),
        &mut row,
    )?;
    row_writer::write_json(writer, "span_links", span.span_links.into(), &mut row)?;

    writer.add_row(row);

    Ok(())
}

fn write_trace_services_to_row(writer: &mut TableData, services: HashSet<String>) -> Result<()> {
    for service_name in services {
        let mut row = writer.alloc_one_row();
        // Write the timestamp as 0.
        row_writer::write_ts_to_nanos(
            writer,
            TIMESTAMP_COLUMN,
            Some(MAX_TIMESTAMP),
            Precision::Nanosecond,
            &mut row,
        )?;

        // Write the `service_name` column.
        row_writer::write_tags(
            writer,
            std::iter::once((SERVICE_NAME_COLUMN.to_string(), service_name)),
            &mut row,
        )?;
        writer.add_row(row);
    }

    Ok(())
}

fn write_trace_operations_to_row(
    writer: &mut TableData,
    operations: HashSet<(String, String, String)>,
) -> Result<()> {
    for (service_name, span_name, span_kind) in operations {
        let mut row = writer.alloc_one_row();
        // Write the timestamp as 0.
        row_writer::write_ts_to_nanos(
            writer,
            TIMESTAMP_COLUMN,
            Some(MAX_TIMESTAMP),
            Precision::Nanosecond,
            &mut row,
        )?;

        // Write the `service_name`, `span_name`, and `span_kind` columns as tags.
        row_writer::write_tags(
            writer,
            vec![
                (SERVICE_NAME_COLUMN.to_string(), service_name),
                (SPAN_NAME_COLUMN.to_string(), span_name),
                (SPAN_KIND_COLUMN.to_string(), span_kind),
            ]
            .into_iter(),
            &mut row,
        )?;
        writer.add_row(row);
    }

    Ok(())
}

#[cfg(test)]
pub(crate) fn write_attributes(
    writer: &mut TableData,
    prefix: &str,
    attributes: Attributes,
    row: &mut Vec<Value>,
) -> Result<()> {
    let row_index = writer.num_rows();
    write_attributes_with_schema(writer, prefix, attributes.take(), row, row_index, None)
}

/// Skips `resource_attributes.service.name` because it is already copied to the
/// top level as `SERVICE_NAME_COLUMN`.
fn skipped_attribute(prefix: &str, key: &str) -> bool {
    prefix == "resource_attributes" && key == KEY_SERVICE_NAME
}

/// Writes flattened attributes owned by one span.
fn write_attributes_with_schema(
    writer: &mut TableData,
    prefix: &str,
    attributes: Vec<KeyValue>,
    row: &mut Vec<Value>,
    row_index: usize,
    mut batch_schema: Option<&mut TraceBatchSchema>,
) -> Result<()> {
    for KeyValue { key, value, .. } in attributes {
        if skipped_attribute(prefix, &key) {
            continue;
        }
        write_attribute_with_schema(
            writer,
            prefix,
            &key,
            value.and_then(|v| v.value),
            row,
            row_index,
            batch_schema.as_deref_mut(),
        );
    }

    Ok(())
}

/// Writes flattened attributes shared by every span of a resource or scope.
///
/// The column name is rebuilt from `prefix`, so keys are read in place and only
/// values that reach a row are copied.
fn write_shared_attributes_with_schema(
    writer: &mut TableData,
    prefix: &str,
    attributes: &Attributes,
    row: &mut Vec<Value>,
    row_index: usize,
    mut batch_schema: Option<&mut TraceBatchSchema>,
) -> Result<()> {
    for attr in attributes.get_ref() {
        if skipped_attribute(prefix, &attr.key) {
            continue;
        }
        write_attribute_with_schema(
            writer,
            prefix,
            &attr.key,
            attr.value.as_ref().and_then(|v| v.value.clone()),
            row,
            row_index,
            batch_schema.as_deref_mut(),
        );
    }

    Ok(())
}

/// Writes one flattened attribute without coercion and optionally records its actual type.
fn write_attribute_with_schema(
    writer: &mut TableData,
    prefix: &str,
    key_suffix: &str,
    value: Option<OtlpValue>,
    row: &mut Vec<Value>,
    row_index: usize,
    batch_schema: Option<&mut TraceBatchSchema>,
) {
    let key = format!("{}.{}", prefix, key_suffix);
    match value {
        Some(OtlpValue::StringValue(v)) => {
            if let Some(batch_schema) = batch_schema {
                batch_schema.observe_value_type(&key, row_index, ColumnDataType::String);
            }
            // Keep the raw request value here. Mixed trace types are reconciled later
            // in the frontend once we can also see the existing table schema.
            writer.write_field_unchecked(
                &key,
                ColumnDataType::String,
                Some(ValueData::StringValue(v)),
                row,
            );
        }
        Some(OtlpValue::BoolValue(v)) => {
            if let Some(batch_schema) = batch_schema {
                batch_schema.observe_value_type(&key, row_index, ColumnDataType::Boolean);
            }
            // Do not coerce or promote types while building the request-local rows.
            writer.write_field_unchecked(
                &key,
                ColumnDataType::Boolean,
                Some(ValueData::BoolValue(v)),
                row,
            );
        }
        Some(OtlpValue::IntValue(v)) => {
            if let Some(batch_schema) = batch_schema {
                batch_schema.observe_value_type(&key, row_index, ColumnDataType::Int64);
            }
            // Preserving the original value avoids order-dependent behavior inside one batch.
            writer.write_field_unchecked(
                &key,
                ColumnDataType::Int64,
                Some(ValueData::I64Value(v)),
                row,
            );
        }
        Some(OtlpValue::DoubleValue(v)) => {
            if let Some(batch_schema) = batch_schema {
                batch_schema.observe_value_type(&key, row_index, ColumnDataType::Float64);
            }
            writer.write_field_unchecked(
                &key,
                ColumnDataType::Float64,
                Some(ValueData::F64Value(v)),
                row,
            );
        }
        Some(OtlpValue::ArrayValue(v)) => {
            if let Some(batch_schema) = batch_schema {
                batch_schema.observe_binary_type(&key, row_index, TraceBinaryType::Json);
            }
            writer.write_column_unchecked(
                row_writer::build_json_column_schema(key),
                Some(ValueData::BinaryValue(
                    any_value_to_jsonb(OtlpValue::ArrayValue(v)).to_vec(),
                )),
                row,
            );
        }
        Some(OtlpValue::KvlistValue(v)) => {
            if let Some(batch_schema) = batch_schema {
                batch_schema.observe_binary_type(&key, row_index, TraceBinaryType::Json);
            }
            writer.write_column_unchecked(
                row_writer::build_json_column_schema(key),
                Some(ValueData::BinaryValue(
                    any_value_to_jsonb(OtlpValue::KvlistValue(v)).to_vec(),
                )),
                row,
            );
        }
        Some(OtlpValue::BytesValue(v)) => {
            if let Some(batch_schema) = batch_schema {
                batch_schema.observe_binary_type(&key, row_index, TraceBinaryType::Binary);
            }
            writer.write_field_unchecked(
                key,
                ColumnDataType::Binary,
                Some(ValueData::BinaryValue(v)),
                row,
            );
        }
        // `StringValueStrindex` is profiling-signal-only and references the
        // Profiling `ProfilesDictionary.string_table`, which is unavailable to
        // traces. Per the OTLP spec, non-Profiling receivers must treat it as a
        // non-fatal issue and process the value as if it were absent. Like the
        // `None` arm, no field is written for the attribute.
        Some(OtlpValue::StringValueStrindex(_)) => {}
        None => {}
    }
}

#[cfg(test)]
mod tests {
    use api::v1::value::ValueData;
    use opentelemetry_proto::tonic::common::v1::any_value::Value as OtlpValue;
    use opentelemetry_proto::tonic::common::v1::{AnyValue, ArrayValue, KeyValue};

    use super::*;
    use crate::otlp::trace::TraceAuxData;
    use crate::otlp::trace::attributes::Attributes;
    use crate::otlp::trace::span::{SpanEvents, SpanLinks};
    use crate::row_writer::TableData;

    fn make_kv(key: &str, value: OtlpValue) -> KeyValue {
        KeyValue {
            key: key.to_string(),
            value: Some(AnyValue { value: Some(value) }),
            ..Default::default()
        }
    }

    fn make_span(service_name: &str, trace_id: &str, span_id: &str) -> TraceSpan {
        TraceSpan {
            service_name: Some(service_name.to_string()),
            trace_id: trace_id.to_string(),
            span_id: span_id.to_string(),
            parent_span_id: None,
            resource_attributes: Attributes::from(vec![]).into(),
            scope_name: "scope".to_string(),
            scope_version: "v1".to_string(),
            scope_attributes: Attributes::from(vec![]).into(),
            trace_state: String::new(),
            span_name: "op".to_string(),
            span_kind: "SPAN_KIND_SERVER".to_string(),
            span_status_code: "STATUS_CODE_UNSET".to_string(),
            span_status_message: String::new(),
            span_attributes: Attributes::from(vec![]),
            span_events: SpanEvents::from(vec![]),
            span_links: SpanLinks::from(vec![]),
            start_in_nanosecond: 1,
            end_in_nanosecond: 2,
        }
    }

    #[test]
    fn test_span_end_before_start_records_zero_duration() {
        // A span whose end precedes its start carries no meaningful duration;
        // it clamps to 0 instead of wrapping. A negative value would fail the
        // checked Int64→UInt64 coercion into pre-existing unsigned tables and
        // break unsigned readers such as the Jaeger query API.
        let mut span = make_span("svc", "trace", "span");
        span.start_in_nanosecond = 200;
        span.end_in_nanosecond = 100;

        let (schema, rows) = build_trace_table_data(&[span])
            .unwrap()
            .into_schema_and_rows();

        let idx = schema
            .iter()
            .position(|c| c.column_name == DURATION_NANO_COLUMN)
            .unwrap();
        assert_eq!(rows[0].values[idx].value_data, Some(ValueData::I64Value(0)));
    }

    #[test]
    fn test_span_duration_above_i64_max_saturates() {
        // The companion clamp: a duration that fits u64 but not i64 saturates
        // at i64::MAX rather than wrapping to a negative number.
        let mut span = make_span("svc", "trace", "span");
        span.start_in_nanosecond = 0;
        span.end_in_nanosecond = u64::MAX;

        let (schema, rows) = build_trace_table_data(&[span])
            .unwrap()
            .into_schema_and_rows();

        let idx = schema
            .iter()
            .position(|c| c.column_name == DURATION_NANO_COLUMN)
            .unwrap();
        assert_eq!(
            rows[0].values[idx].value_data,
            Some(ValueData::I64Value(i64::MAX))
        );
    }

    #[test]
    fn test_fixed_trace_columns_keep_schema_and_values_aligned() {
        let table_data = build_trace_table_data(&[make_span("svc", "trace", "span")]).unwrap();
        let (schema, rows) = table_data.into_schema_and_rows();
        let expected = [
            (
                TIMESTAMP_COLUMN,
                Some(ValueData::TimestampNanosecondValue(1)),
            ),
            (
                TIMESTAMP_END_COLUMN,
                Some(ValueData::TimestampNanosecondValue(2)),
            ),
            (DURATION_NANO_COLUMN, Some(ValueData::I64Value(1))),
            (PARENT_SPAN_ID_COLUMN, None),
            (
                TRACE_ID_COLUMN,
                Some(ValueData::StringValue("trace".to_string())),
            ),
            (
                SPAN_ID_COLUMN,
                Some(ValueData::StringValue("span".to_string())),
            ),
            (
                SPAN_KIND_COLUMN,
                Some(ValueData::StringValue("SPAN_KIND_SERVER".to_string())),
            ),
            (
                SPAN_NAME_COLUMN,
                Some(ValueData::StringValue("op".to_string())),
            ),
            (
                SPAN_STATUS_CODE,
                Some(ValueData::StringValue("STATUS_CODE_UNSET".to_string())),
            ),
            (
                SPAN_STATUS_MESSAGE_COLUMN,
                Some(ValueData::StringValue(String::new())),
            ),
            (
                TRACE_STATE_COLUMN,
                Some(ValueData::StringValue(String::new())),
            ),
            (
                SCOPE_NAME_COLUMN,
                Some(ValueData::StringValue("scope".to_string())),
            ),
            (
                SCOPE_VERSION_COLUMN,
                Some(ValueData::StringValue("v1".to_string())),
            ),
        ];

        for (index, (name, value)) in expected.into_iter().enumerate() {
            assert_eq!(schema[index].column_name, name);
            assert_eq!(rows[0].values[index].value_data, value);
        }
    }

    #[test]
    fn test_batch_schema_preserves_column_order_and_attribute_types() {
        let mut span1 = make_span("svc-a", "trace-a", "span-a");
        span1.span_attributes = Attributes::from(vec![make_kv("val", OtlpValue::IntValue(10))]);
        let mut span2 = make_span("svc-a", "trace-a", "span-b");
        span2.span_attributes = Attributes::from(vec![make_kv("val", OtlpValue::DoubleValue(1.5))]);

        let (table_data, batch_schema) =
            build_trace_table_data_with_schema(vec![span1, span2].into_iter()).unwrap();

        assert_eq!(
            batch_schema.value_types("span_attributes.val").unwrap(),
            [ColumnDataType::Int64, ColumnDataType::Float64]
        );
        let timestamp_index = table_data
            .columns()
            .iter()
            .position(|column| column.column_name == TIMESTAMP_COLUMN)
            .unwrap();
        let attribute_index = table_data
            .columns()
            .iter()
            .position(|column| column.column_name == "span_attributes.val")
            .unwrap();
        assert!(timestamp_index < attribute_index);
    }

    #[test]
    fn test_optional_batch_schema_observation_preserves_rows() {
        let mut span = make_span("svc-a", "trace-a", "span-a");
        span.span_attributes = Attributes::from(vec![make_kv("val", OtlpValue::IntValue(10))]);

        let without_schema = build_trace_table_data(std::slice::from_ref(&span)).unwrap();
        let (with_schema, batch_schema) =
            build_trace_table_data_with_schema(vec![span].into_iter()).unwrap();

        assert_eq!(
            without_schema.into_schema_and_rows(),
            with_schema.into_schema_and_rows()
        );
        assert_eq!(
            batch_schema.value_types("span_attributes.val").unwrap(),
            [ColumnDataType::Int64]
        );
        let retry_columns = batch_schema.into_retry_columns();
        let retry_column = &retry_columns["span_attributes.val"];
        assert_eq!(retry_column.present_rows, [0]);
        assert!(retry_column.binary_types.is_empty());
    }

    #[test]
    fn test_batch_schema_tracks_binary_and_json_rows() {
        let mut binary_span = make_span("svc-a", "trace-a", "span-a");
        binary_span.span_attributes = Attributes::from(vec![make_kv(
            "val",
            OtlpValue::BytesValue(vec![1_u8, 2, 3]),
        )]);
        let mut json_span = make_span("svc-a", "trace-a", "span-b");
        json_span.span_attributes = Attributes::from(vec![make_kv(
            "val",
            OtlpValue::ArrayValue(ArrayValue {
                values: vec![AnyValue {
                    value: Some(OtlpValue::IntValue(1)),
                }],
            }),
        )]);

        let (_, batch_schema) =
            build_trace_table_data_with_schema(vec![binary_span, json_span].into_iter()).unwrap();

        assert!(batch_schema.has_incompatible_logical_types());
        assert!(batch_schema.has_incompatible_logical_types_for("span_attributes.val"));
        let retry_columns = batch_schema.into_retry_columns();
        let retry_column = &retry_columns["span_attributes.val"];
        assert_eq!(retry_column.present_rows, [0, 1]);
        assert_eq!(
            retry_column.binary_types,
            [(0, TraceBinaryType::Binary), (1, TraceBinaryType::Json)]
        );
    }

    #[test]
    fn test_batch_schema_preserves_scalar_then_binary_values() {
        let mut scalar_span = make_span("svc-a", "trace-a", "span-a");
        scalar_span.span_attributes = Attributes::from(vec![
            make_kv("bytes", OtlpValue::StringValue("text".to_string())),
            make_kv("json", OtlpValue::StringValue("text".to_string())),
        ]);
        let mut binary_span = make_span("svc-a", "trace-a", "span-b");
        binary_span.span_attributes = Attributes::from(vec![
            make_kv("bytes", OtlpValue::BytesValue(vec![1_u8, 2, 3])),
            make_kv(
                "json",
                OtlpValue::ArrayValue(ArrayValue {
                    values: vec![AnyValue {
                        value: Some(OtlpValue::IntValue(1)),
                    }],
                }),
            ),
        ]);

        let (_, batch_schema) =
            build_trace_table_data_with_schema(vec![scalar_span, binary_span].into_iter()).unwrap();

        assert_eq!(
            batch_schema.value_types("span_attributes.bytes").unwrap(),
            [ColumnDataType::String, ColumnDataType::Binary]
        );
        assert_eq!(
            batch_schema.value_types("span_attributes.json").unwrap(),
            [ColumnDataType::String, ColumnDataType::Binary]
        );
        let retry_columns = batch_schema.into_retry_columns();
        assert_eq!(
            retry_columns["span_attributes.bytes"].binary_types,
            [(1, TraceBinaryType::Binary)]
        );
        assert_eq!(
            retry_columns["span_attributes.json"].binary_types,
            [(1, TraceBinaryType::Json)]
        );
    }

    #[test]
    fn test_keep_mixed_numeric_values_until_frontend_reconciliation() {
        let mut writer = TableData::new(4, 2);

        let attrs1 = Attributes::from(vec![make_kv("val", OtlpValue::DoubleValue(1.5))]);
        let mut row1 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs1, &mut row1).unwrap();
        writer.add_row(row1);

        let attrs2 = Attributes::from(vec![make_kv("val", OtlpValue::IntValue(42))]);
        let mut row2 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs2, &mut row2).unwrap();
        writer.add_row(row2);

        let (schema, rows) = writer.into_schema_and_rows();

        let col_idx = schema
            .iter()
            .position(|c| c.column_name == "attr.val")
            .unwrap();
        assert_eq!(schema[col_idx].datatype, ColumnDataType::Float64 as i32);

        assert_eq!(
            rows[0].values[col_idx].value_data,
            Some(ValueData::F64Value(1.5))
        );
        assert_eq!(
            rows[1].values[col_idx].value_data,
            Some(ValueData::I64Value(42))
        );
    }

    #[test]
    fn test_keep_mixed_string_and_int_values_until_frontend_reconciliation() {
        let mut writer = TableData::new(4, 2);

        let attrs1 = Attributes::from(vec![make_kv("val", OtlpValue::IntValue(10))]);
        let mut row1 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs1, &mut row1).unwrap();
        writer.add_row(row1);

        let attrs2 = Attributes::from(vec![make_kv(
            "val",
            OtlpValue::StringValue("20".to_string()),
        )]);
        let mut row2 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs2, &mut row2).unwrap();
        writer.add_row(row2);

        let (schema, rows) = writer.into_schema_and_rows();
        let col_idx = schema
            .iter()
            .position(|c| c.column_name == "attr.val")
            .unwrap();
        assert_eq!(schema[col_idx].datatype, ColumnDataType::Int64 as i32);
        assert_eq!(
            rows[1].values[col_idx].value_data,
            Some(ValueData::StringValue("20".to_string()))
        );
    }

    #[test]
    fn test_keep_first_seen_schema_until_frontend_reconciliation() {
        let mut writer = TableData::new(4, 2);

        let attrs1 = Attributes::from(vec![make_kv(
            "val",
            OtlpValue::StringValue("10".to_string()),
        )]);
        let mut row1 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs1, &mut row1).unwrap();
        writer.add_row(row1);

        let attrs2 = Attributes::from(vec![make_kv("val", OtlpValue::IntValue(20))]);
        let mut row2 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs2, &mut row2).unwrap();
        writer.add_row(row2);

        let (schema, rows) = writer.into_schema_and_rows();
        let col_idx = schema
            .iter()
            .position(|c| c.column_name == "attr.val")
            .unwrap();
        assert_eq!(schema[col_idx].datatype, ColumnDataType::String as i32);
        assert_eq!(
            rows[0].values[col_idx].value_data,
            Some(ValueData::StringValue("10".to_string()))
        );
        assert_eq!(
            rows[1].values[col_idx].value_data,
            Some(ValueData::I64Value(20))
        );
    }

    #[test]
    fn test_keep_mixed_string_and_float_values_until_frontend_reconciliation() {
        let mut writer = TableData::new(4, 2);

        let attrs1 = Attributes::from(vec![make_kv("val", OtlpValue::DoubleValue(1.5))]);
        let mut row1 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs1, &mut row1).unwrap();
        writer.add_row(row1);

        let attrs2 = Attributes::from(vec![make_kv(
            "val",
            OtlpValue::StringValue("1.5".to_string()),
        )]);
        let mut row2 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs2, &mut row2).unwrap();
        writer.add_row(row2);

        let (schema, rows) = writer.into_schema_and_rows();
        let col_idx = schema
            .iter()
            .position(|c| c.column_name == "attr.val")
            .unwrap();
        assert_eq!(schema[col_idx].datatype, ColumnDataType::Float64 as i32);
        assert_eq!(
            rows[1].values[col_idx].value_data,
            Some(ValueData::StringValue("1.5".to_string()))
        );
    }

    #[test]
    fn test_keep_mixed_string_and_bool_values_until_frontend_reconciliation() {
        let mut writer = TableData::new(4, 2);

        let attrs1 = Attributes::from(vec![make_kv(
            "val",
            OtlpValue::StringValue("true".to_string()),
        )]);
        let mut row1 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs1, &mut row1).unwrap();
        writer.add_row(row1);

        let attrs2 = Attributes::from(vec![make_kv("val", OtlpValue::BoolValue(false))]);
        let mut row2 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs2, &mut row2).unwrap();
        writer.add_row(row2);

        let (schema, rows) = writer.into_schema_and_rows();
        let col_idx = schema
            .iter()
            .position(|c| c.column_name == "attr.val")
            .unwrap();
        assert_eq!(schema[col_idx].datatype, ColumnDataType::String as i32);
        assert_eq!(
            rows[0].values[col_idx].value_data,
            Some(ValueData::StringValue("true".to_string()))
        );
        assert_eq!(
            rows[1].values[col_idx].value_data,
            Some(ValueData::BoolValue(false))
        );
    }

    #[test]
    fn test_keep_mixed_binary_and_string_values_until_frontend_reconciliation() {
        let mut writer = TableData::new(4, 2);

        let attrs1 = Attributes::from(vec![make_kv(
            "val",
            OtlpValue::BytesValue(vec![1_u8, 2, 3]),
        )]);
        let mut row1 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs1, &mut row1).unwrap();
        writer.add_row(row1);

        let attrs2 = Attributes::from(vec![make_kv(
            "val",
            OtlpValue::StringValue("false".to_string()),
        )]);
        let mut row2 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs2, &mut row2).unwrap();
        writer.add_row(row2);

        let (schema, rows) = writer.into_schema_and_rows();
        let col_idx = schema
            .iter()
            .position(|c| c.column_name == "attr.val")
            .unwrap();
        assert_eq!(schema[col_idx].datatype, ColumnDataType::Binary as i32);
        assert_eq!(
            rows[0].values[col_idx].value_data,
            Some(ValueData::BinaryValue(vec![1_u8, 2, 3]))
        );
        assert_eq!(
            rows[1].values[col_idx].value_data,
            Some(ValueData::StringValue("false".to_string()))
        );
    }

    #[test]
    fn test_keep_mixed_binary_and_json_values_until_frontend_reconciliation() {
        let mut writer = TableData::new(4, 2);

        let attrs1 = Attributes::from(vec![make_kv(
            "val",
            OtlpValue::BytesValue(vec![1_u8, 2, 3]),
        )]);
        let mut row1 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs1, &mut row1).unwrap();
        writer.add_row(row1);

        let attrs2 = Attributes::from(vec![make_kv(
            "val",
            OtlpValue::ArrayValue(ArrayValue {
                values: vec![AnyValue {
                    value: Some(OtlpValue::IntValue(1)),
                }],
            }),
        )]);
        let mut row2 = writer.alloc_one_row();
        write_attributes(&mut writer, "attr", attrs2, &mut row2).unwrap();
        writer.add_row(row2);

        let (schema, rows) = writer.into_schema_and_rows();
        let col_idx = schema
            .iter()
            .position(|c| c.column_name == "attr.val")
            .unwrap();
        assert_eq!(schema[col_idx].datatype, ColumnDataType::Binary as i32);
        assert!(matches!(
            rows[1].values[col_idx].value_data.as_ref(),
            Some(ValueData::BinaryValue(_))
        ));
    }

    #[test]
    fn test_build_aux_table_requests_deduplicates_services_and_operations() {
        let spans = vec![
            make_span("svc-a", "trace-a", "span-a"),
            make_span("svc-a", "trace-b", "span-b"),
        ];
        let mut aux_data = TraceAuxData::default();
        for span in &spans {
            aux_data.observe_span(span);
        }

        let (requests, total_rows) =
            build_aux_table_requests(aux_data, "opentelemetry_traces").unwrap();
        assert_eq!(requests.inserts.len(), 2);
        assert_eq!(total_rows, 2);
    }
    // Conversion matrix coverage lives in the shared coercion helper tests.
}