Skip to main content

frontend/instance/otlp/
trace_types.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use api::v1::value::ValueData;
16use api::v1::{ColumnDataType, Row, Rows};
17use servers::error::{self, Result as ServerResult};
18use servers::otlp::coerce::{
19    coerce_value_data, is_supported_trace_coercion, resolve_new_trace_column_type,
20    trace_value_datatype,
21};
22
23use crate::instance::otlp::trace_semconv::trace_semconv_fixed_type;
24
25/// Attribute values are user data echoed back in the OTLP partial-success
26/// message and the server log, so diagnostics keep at most this many characters.
27const TRACE_VALUE_DIAGNOSTIC_LIMIT: usize = 16;
28
29/// Truncates to `limit` characters, marking the cut with `...`.
30///
31/// Diagnostics carry user-controlled text; slicing by byte offset would panic in
32/// the middle of a multi-byte sequence.
33pub(super) fn truncate_for_diagnostics(text: &str, limit: usize) -> String {
34    match text.char_indices().nth(limit) {
35        Some((offset, _)) => format!("{}...", &text[..offset]),
36        None => text.to_string(),
37    }
38}
39
40/// Renders a failing trace value as `Type(value)`, e.g. `String("")`.
41fn describe_trace_value(value: &ValueData, request_type: ColumnDataType) -> String {
42    let payload = match value {
43        ValueData::StringValue(string_value) => format!(
44            "{:?}",
45            truncate_for_diagnostics(string_value, TRACE_VALUE_DIAGNOSTIC_LIMIT)
46        ),
47        ValueData::BoolValue(bool_value) => bool_value.to_string(),
48        ValueData::I64Value(int_value) => int_value.to_string(),
49        ValueData::F64Value(float_value) => float_value.to_string(),
50        ValueData::BinaryValue(bytes) => format!("{} bytes", bytes.len()),
51        // Other value kinds never reach trace coercion, so report the type alone.
52        _ => return format!("{request_type:?}"),
53    };
54
55    format!("{request_type:?}({payload})")
56}
57
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59pub(super) enum TraceReconcileDecision {
60    UseExisting(ColumnDataType),
61    UseRequestLocal(ColumnDataType),
62    AlterExistingTo(ColumnDataType),
63}
64
65impl TraceReconcileDecision {
66    pub(super) fn target_type(self) -> ColumnDataType {
67        match self {
68            Self::UseExisting(target_type)
69            | Self::UseRequestLocal(target_type)
70            | Self::AlterExistingTo(target_type) => target_type,
71        }
72    }
73
74    pub(super) fn requires_alter(self) -> bool {
75        matches!(self, Self::AlterExistingTo(_))
76    }
77}
78
79/// Describes a column rewrite before its row values have been validated.
80#[derive(Debug)]
81pub(super) struct PendingTraceColumnRewrite {
82    pub(super) col_idx: usize,
83    pub(super) target_type: ColumnDataType,
84    pub(super) column_name: String,
85}
86
87/// Holds the schema and value rewrites prepared for atomic application.
88#[derive(Debug)]
89pub(super) struct PreparedTraceColumnRewrites {
90    columns: Vec<PreparedTraceColumnRewrite>,
91    values: Vec<PreparedTraceValueRewrite>,
92}
93
94/// Reports the column whose trace value could not be rewritten.
95#[derive(Debug)]
96pub(super) struct TraceColumnRewriteError {
97    pub(super) error: servers::error::Error,
98    pub(super) column_name: String,
99}
100
101/// Updates one request column to its reconciled datatype.
102#[derive(Debug)]
103struct PreparedTraceColumnRewrite {
104    col_idx: usize,
105    target_type: ColumnDataType,
106}
107
108/// Replaces one trace value with its precomputed coerced value.
109#[derive(Debug)]
110struct PreparedTraceValueRewrite {
111    row_idx: usize,
112    col_idx: usize,
113    value_data: Option<ValueData>,
114}
115
116impl PreparedTraceColumnRewrites {
117    pub(super) fn apply(self, rows: &mut Rows) {
118        for column in self.columns {
119            rows.schema[column.col_idx].datatype = column.target_type as i32;
120        }
121        for value in self.values {
122            rows.rows[value.row_idx].values[value.col_idx].value_data = value.value_data;
123        }
124    }
125}
126
127/// Picks the reconciliation action for one trace column.
128///
129/// Existing table schema is authoritative unless the only incompatible case is
130/// widening an existing Int64 column to Float64 for incoming Int64/Float64 data.
131pub(super) fn choose_trace_reconcile_decision(
132    column_name: &str,
133    observed_types: &[ColumnDataType],
134    existing_type: Option<ColumnDataType>,
135) -> ServerResult<Option<TraceReconcileDecision>> {
136    if let Some(fixed_type) = trace_semconv_fixed_type(column_name) {
137        return choose_fixed_trace_reconcile_decision(fixed_type, observed_types, existing_type);
138    }
139
140    let Some(existing_type) = existing_type else {
141        return resolve_new_trace_column_type(observed_types.iter().copied())
142            .map(|target_type| target_type.map(TraceReconcileDecision::UseRequestLocal))
143            .map_err(|_| {
144                error::InvalidParameterSnafu {
145                    reason: "unsupported trace type mix".to_string(),
146                }
147                .build()
148            });
149    };
150
151    if observed_types.iter().all(|&request_type| {
152        request_type == existing_type || is_supported_trace_coercion(request_type, existing_type)
153    }) {
154        return Ok(Some(TraceReconcileDecision::UseExisting(existing_type)));
155    }
156
157    if existing_type == ColumnDataType::Int64
158        && observed_types.contains(&ColumnDataType::Float64)
159        && observed_types.iter().all(|observed_type| {
160            matches!(
161                observed_type,
162                ColumnDataType::Int64 | ColumnDataType::Float64
163            )
164        })
165    {
166        return Ok(Some(TraceReconcileDecision::AlterExistingTo(
167            ColumnDataType::Float64,
168        )));
169    }
170
171    error::InvalidParameterSnafu {
172        reason: "unsupported trace type mix".to_string(),
173    }
174    .fail()
175}
176
177fn choose_fixed_trace_reconcile_decision(
178    fixed_type: ColumnDataType,
179    observed_types: &[ColumnDataType],
180    existing_type: Option<ColumnDataType>,
181) -> ServerResult<Option<TraceReconcileDecision>> {
182    let Some(existing_type) = existing_type else {
183        return Ok(Some(TraceReconcileDecision::UseRequestLocal(fixed_type)));
184    };
185
186    if existing_type == fixed_type {
187        return Ok(Some(TraceReconcileDecision::UseExisting(fixed_type)));
188    }
189
190    if fixed_type == ColumnDataType::Float64
191        && existing_type == ColumnDataType::Int64
192        && observed_types.iter().all(|observed_type| {
193            matches!(
194                observed_type,
195                ColumnDataType::Int64 | ColumnDataType::Float64
196            )
197        })
198    {
199        return Ok(Some(TraceReconcileDecision::AlterExistingTo(fixed_type)));
200    }
201
202    error::InvalidParameterSnafu {
203        reason: "unsupported trace type mix".to_string(),
204    }
205    .fail()
206}
207
208/// Prepares an atomic rewrite plan without mutating the input rows.
209///
210/// For each pending column rewrite, this precomputes every required value
211/// coercion. Missing, null, and already-correct values are skipped. If any
212/// coercion fails, it returns the failing column and leaves all rows unchanged.
213///
214/// Target types must already have been selected by `TraceRequestSchema`.
215/// Call [`PreparedTraceColumnRewrites::apply`] to update the schema and values.
216pub(super) fn prepare_trace_column_rewrites(
217    rows: &[Row],
218    pending_rewrites: Vec<PendingTraceColumnRewrite>,
219    table_name: &str,
220) -> Result<PreparedTraceColumnRewrites, TraceColumnRewriteError> {
221    let mut values = Vec::new();
222    for (row_idx, row) in rows.iter().enumerate() {
223        for pending_rewrite in &pending_rewrites {
224            let Some(value) = row.values.get(pending_rewrite.col_idx) else {
225                continue;
226            };
227            let Some(request_value) = value.value_data.as_ref() else {
228                continue;
229            };
230            let Some(request_type) = trace_value_datatype(request_value) else {
231                continue;
232            };
233            if request_type == pending_rewrite.target_type {
234                continue;
235            }
236
237            let value_data =
238                coerce_value_data(&value.value_data, pending_rewrite.target_type, request_type)
239                    .map_err(|_| TraceColumnRewriteError {
240                        error: error::InvalidParameterSnafu {
241                            reason: format!(
242                                "failed to coerce trace column '{}' in table '{}' from {} to {:?}",
243                                pending_rewrite.column_name,
244                                table_name,
245                                describe_trace_value(request_value, request_type),
246                                pending_rewrite.target_type
247                            ),
248                        }
249                        .build(),
250                        column_name: pending_rewrite.column_name.clone(),
251                    })?;
252            values.push(PreparedTraceValueRewrite {
253                row_idx,
254                col_idx: pending_rewrite.col_idx,
255                value_data,
256            });
257        }
258    }
259
260    let columns = pending_rewrites
261        .into_iter()
262        .map(|rewrite| PreparedTraceColumnRewrite {
263            col_idx: rewrite.col_idx,
264            target_type: rewrite.target_type,
265        })
266        .collect();
267    Ok(PreparedTraceColumnRewrites { columns, values })
268}
269
270pub(super) fn enrich_trace_reconcile_error(
271    table_name: &str,
272    column_name: &str,
273    observed_types: &[ColumnDataType],
274    existing_type: Option<ColumnDataType>,
275    fixed_type: Option<ColumnDataType>,
276) -> servers::error::Error {
277    let observed_types = observed_types
278        .iter()
279        .map(|datatype| format!("{datatype:?}"))
280        .collect::<Vec<_>>()
281        .join(", ");
282
283    error::InvalidParameterSnafu {
284        reason: match (existing_type, fixed_type) {
285            (Some(existing_type), Some(fixed_type)) => format!(
286                "failed to reconcile trace column '{}' in table '{}' with observed types [{}] against existing {:?} and fixed semconv {:?}",
287                column_name, table_name, observed_types, existing_type, fixed_type
288            ),
289            (Some(existing_type), None) => format!(
290                "failed to reconcile trace column '{}' in table '{}' with observed types [{}] against existing {:?}",
291                column_name, table_name, observed_types, existing_type
292            ),
293            (None, Some(fixed_type)) => format!(
294                "failed to reconcile trace column '{}' in table '{}' with observed types [{}] and fixed semconv {:?}",
295                column_name, table_name, observed_types, fixed_type
296            ),
297            (None, None) => format!(
298                "failed to reconcile trace column '{}' in table '{}' with observed types [{}]",
299                column_name, table_name, observed_types
300            ),
301        },
302    }
303    .build()
304}
305
306/// Only these trace scalar types participate in reconciliation. Other column kinds
307/// such as JSON and binary keep their original write path and schema checks.
308pub(super) fn is_trace_reconcile_candidate_type(datatype: ColumnDataType) -> bool {
309    matches!(
310        datatype,
311        ColumnDataType::String
312            | ColumnDataType::Boolean
313            | ColumnDataType::Int64
314            | ColumnDataType::Float64
315    )
316}
317
318/// Keeps the observed type list small without depending on enum ordering.
319pub(super) fn push_observed_trace_type(
320    observed_types: &mut Vec<ColumnDataType>,
321    datatype: ColumnDataType,
322) {
323    if !observed_types.contains(&datatype) {
324        observed_types.push(datatype);
325    }
326}
327
328#[cfg(test)]
329mod tests {
330    use api::v1::value::ValueData;
331    use api::v1::{ColumnDataType, ColumnSchema, Row, Rows, Value};
332    use common_error::ext::ErrorExt;
333    use common_error::status_code::StatusCode;
334
335    use super::{
336        PendingTraceColumnRewrite, TraceReconcileDecision, choose_trace_reconcile_decision,
337        describe_trace_value, enrich_trace_reconcile_error, is_trace_reconcile_candidate_type,
338        prepare_trace_column_rewrites, push_observed_trace_type, truncate_for_diagnostics,
339    };
340
341    #[test]
342    fn test_choose_trace_reconcile_decision_existing_int64_keeps_int64() {
343        assert_eq!(
344            choose_trace_reconcile_decision(
345                "span_attributes.attr_int",
346                &[ColumnDataType::Int64],
347                Some(ColumnDataType::Int64)
348            )
349            .unwrap(),
350            Some(TraceReconcileDecision::UseExisting(ColumnDataType::Int64))
351        );
352    }
353
354    #[test]
355    fn test_choose_trace_reconcile_decision_existing_uint64_keeps_uint64() {
356        // Backward-compat for the unsigned -> signed transition: an existing
357        // table whose `duration_nano` is still UInt64 must keep that type (no
358        // ALTER) when new signed (Int64) ingest arrives, coercing the value in
359        // place. This is the no-ALTER guarantee for the trace path; it relies on
360        // the Int64 -> UInt64 coercion arm added in Phase 0.
361        assert_eq!(
362            choose_trace_reconcile_decision(
363                "duration_nano",
364                &[ColumnDataType::Int64],
365                Some(ColumnDataType::Uint64)
366            )
367            .unwrap(),
368            Some(TraceReconcileDecision::UseExisting(ColumnDataType::Uint64))
369        );
370    }
371
372    #[test]
373    fn test_choose_trace_reconcile_decision_existing_int64_widens_to_float64() {
374        assert_eq!(
375            choose_trace_reconcile_decision(
376                "span_attributes.attr_double",
377                &[ColumnDataType::Int64, ColumnDataType::Float64],
378                Some(ColumnDataType::Int64)
379            )
380            .unwrap(),
381            Some(TraceReconcileDecision::AlterExistingTo(
382                ColumnDataType::Float64
383            ))
384        );
385    }
386
387    #[test]
388    fn test_choose_trace_reconcile_decision_existing_float64_stays_authoritative() {
389        assert_eq!(
390            choose_trace_reconcile_decision(
391                "span_attributes.attr_double",
392                &[ColumnDataType::Int64, ColumnDataType::Float64],
393                Some(ColumnDataType::Float64)
394            )
395            .unwrap(),
396            Some(TraceReconcileDecision::UseExisting(ColumnDataType::Float64))
397        );
398    }
399
400    #[test]
401    fn test_choose_trace_reconcile_decision_existing_int64_with_boolean_is_error() {
402        let err = choose_trace_reconcile_decision(
403            "span_attributes.attr_numeric",
404            &[ColumnDataType::Boolean, ColumnDataType::Int64],
405            Some(ColumnDataType::Int64),
406        )
407        .unwrap_err();
408        assert_eq!(err.status_code(), StatusCode::InvalidArguments);
409    }
410
411    #[test]
412    fn test_choose_trace_reconcile_decision_request_local_prefers_float64() {
413        assert_eq!(
414            choose_trace_reconcile_decision(
415                "span_attributes.attr_numeric",
416                &[ColumnDataType::Int64, ColumnDataType::Float64],
417                None
418            )
419            .unwrap(),
420            Some(TraceReconcileDecision::UseRequestLocal(
421                ColumnDataType::Float64
422            ))
423        );
424    }
425
426    #[test]
427    fn test_choose_trace_reconcile_decision_whitelisted_new_int64_column_uses_fixed_type() {
428        assert_eq!(
429            choose_trace_reconcile_decision(
430                "span_attributes.http.response.status_code",
431                &[ColumnDataType::String, ColumnDataType::Int64],
432                None
433            )
434            .unwrap(),
435            Some(TraceReconcileDecision::UseRequestLocal(
436                ColumnDataType::Int64
437            ))
438        );
439    }
440
441    #[test]
442    fn test_choose_trace_reconcile_decision_new_boolean_column_uses_dynamic_resolution() {
443        assert_eq!(
444            choose_trace_reconcile_decision(
445                "span_attributes.messaging.destination.temporary",
446                &[ColumnDataType::String, ColumnDataType::Boolean],
447                None
448            )
449            .unwrap(),
450            Some(TraceReconcileDecision::UseRequestLocal(
451                ColumnDataType::Boolean
452            ))
453        );
454    }
455
456    #[test]
457    fn test_choose_trace_reconcile_decision_whitelisted_existing_matching_type_uses_fixed_type() {
458        assert_eq!(
459            choose_trace_reconcile_decision(
460                "resource_attributes.service.name",
461                &[ColumnDataType::String],
462                Some(ColumnDataType::String)
463            )
464            .unwrap(),
465            Some(TraceReconcileDecision::UseExisting(ColumnDataType::String))
466        );
467    }
468
469    #[test]
470    fn test_choose_trace_reconcile_decision_whitelisted_existing_conflicting_type_is_error() {
471        let err = choose_trace_reconcile_decision(
472            "span_attributes.server.port",
473            &[ColumnDataType::Int64],
474            Some(ColumnDataType::String),
475        )
476        .unwrap_err();
477        assert_eq!(err.status_code(), StatusCode::InvalidArguments);
478    }
479
480    #[test]
481    fn test_choose_trace_reconcile_decision_non_whitelisted_retains_dynamic_behavior() {
482        assert_eq!(
483            choose_trace_reconcile_decision(
484                "span_attributes.attr_numeric",
485                &[ColumnDataType::Int64, ColumnDataType::Float64],
486                None
487            )
488            .unwrap(),
489            Some(TraceReconcileDecision::UseRequestLocal(
490                ColumnDataType::Float64
491            ))
492        );
493    }
494
495    #[test]
496    fn test_prepare_trace_column_rewrites_rejects_invalid_string_parse() {
497        let rows = vec![Row {
498            values: vec![Value {
499                value_data: Some(ValueData::StringValue("not_a_number".to_string())),
500            }],
501        }];
502        let pending_rewrites = vec![PendingTraceColumnRewrite {
503            col_idx: 0,
504            target_type: ColumnDataType::Int64,
505            column_name: "span_attributes.attr_int".to_string(),
506        }];
507
508        let err = prepare_trace_column_rewrites(&rows, pending_rewrites, "trace_type_atomicity")
509            .unwrap_err();
510        assert_eq!(err.error.status_code(), StatusCode::InvalidArguments);
511        assert_eq!(err.column_name, "span_attributes.attr_int");
512        assert!(
513            err.error.to_string().contains(
514                "failed to coerce trace column 'span_attributes.attr_int' in table \
515                 'trace_type_atomicity' from String(\"not_a_number\") to Int64"
516            ),
517            "unexpected error message: {}",
518            err.error
519        );
520    }
521
522    /// The PHP instrumentation case: `getHeaderLine('Content-Length')` sends ""
523    /// when the header is absent. The span must be kept with a null attribute.
524    #[test]
525    fn test_prepare_trace_column_rewrites_nulls_empty_string_value() {
526        let mut rows = Rows {
527            schema: vec![ColumnSchema {
528                column_name: "span_attributes.http.response.body.size".to_string(),
529                datatype: ColumnDataType::String as i32,
530                ..Default::default()
531            }],
532            rows: vec![Row {
533                values: vec![Value {
534                    value_data: Some(ValueData::StringValue(String::new())),
535                }],
536            }],
537        };
538        let pending_rewrites = vec![PendingTraceColumnRewrite {
539            col_idx: 0,
540            target_type: ColumnDataType::Int64,
541            column_name: "span_attributes.http.response.body.size".to_string(),
542        }];
543
544        prepare_trace_column_rewrites(&rows.rows, pending_rewrites, "opentelemetry_traces")
545            .unwrap()
546            .apply(&mut rows);
547
548        assert_eq!(rows.schema[0].datatype, ColumnDataType::Int64 as i32);
549        assert_eq!(rows.rows[0].values[0].value_data, None);
550    }
551
552    #[test]
553    fn test_describe_trace_value_bounds_and_escapes_strings() {
554        assert_eq!(
555            describe_trace_value(
556                &ValueData::StringValue(String::new()),
557                ColumnDataType::String
558            ),
559            r#"String("")"#
560        );
561        assert_eq!(
562            describe_trace_value(
563                &ValueData::StringValue("a\tb\"c".to_string()),
564                ColumnDataType::String
565            ),
566            r#"String("a\tb\"c")"#
567        );
568        assert_eq!(
569            describe_trace_value(
570                &ValueData::StringValue("0123456789abcdefghij".to_string()),
571                ColumnDataType::String
572            ),
573            r#"String("0123456789abcdef...")"#
574        );
575    }
576
577    #[test]
578    fn test_describe_trace_value_omits_binary_content() {
579        assert_eq!(
580            describe_trace_value(
581                &ValueData::BinaryValue(vec![1_u8, 2, 3]),
582                ColumnDataType::Binary
583            ),
584            "Binary(3 bytes)"
585        );
586        assert_eq!(
587            describe_trace_value(&ValueData::F64Value(1.5), ColumnDataType::Float64),
588            "Float64(1.5)"
589        );
590    }
591
592    /// Truncation runs over user-supplied text, so it must not split a
593    /// multi-byte character.
594    #[test]
595    fn test_truncate_for_diagnostics_cuts_on_char_boundary() {
596        assert_eq!(truncate_for_diagnostics("日本語テキスト", 3), "日本語...");
597        assert_eq!(truncate_for_diagnostics("short", 16), "short");
598        assert_eq!(truncate_for_diagnostics("exact", 5), "exact");
599    }
600
601    #[test]
602    fn test_prepare_trace_column_rewrites_applies_prepared_values() {
603        let mut rows = Rows {
604            schema: vec![ColumnSchema {
605                datatype: ColumnDataType::String as i32,
606                ..Default::default()
607            }],
608            rows: vec![Row {
609                values: vec![Value {
610                    value_data: Some(ValueData::StringValue("503".to_string())),
611                }],
612            }],
613        };
614        let pending_rewrites = vec![PendingTraceColumnRewrite {
615            col_idx: 0,
616            target_type: ColumnDataType::Int64,
617            column_name: "span_attributes.http.response.status_code".to_string(),
618        }];
619
620        let prepared =
621            prepare_trace_column_rewrites(&rows.rows, pending_rewrites, "trace_type_atomicity")
622                .unwrap();
623        assert_eq!(
624            rows.rows[0].values[0].value_data,
625            Some(ValueData::StringValue("503".to_string()))
626        );
627
628        prepared.apply(&mut rows);
629        assert_eq!(rows.schema[0].datatype, ColumnDataType::Int64 as i32);
630        assert_eq!(
631            rows.rows[0].values[0].value_data,
632            Some(ValueData::I64Value(503))
633        );
634    }
635
636    #[test]
637    fn test_prepare_trace_column_rewrites_coerces_int64_into_existing_uint64() {
638        // Existing-table backward-compat for the trace path: new signed ingest
639        // arrives as Int64, but the existing `duration_nano` column is UInt64, so
640        // the rewrite coerces the value into the existing type in place (no
641        // ALTER). Mirrors what happens for a table created before the
642        // unsigned -> signed flip.
643        let mut rows = Rows {
644            schema: vec![ColumnSchema {
645                datatype: ColumnDataType::Int64 as i32,
646                ..Default::default()
647            }],
648            rows: vec![Row {
649                values: vec![Value {
650                    value_data: Some(ValueData::I64Value(42)),
651                }],
652            }],
653        };
654        let pending_rewrites = vec![PendingTraceColumnRewrite {
655            col_idx: 0,
656            target_type: ColumnDataType::Uint64,
657            column_name: "duration_nano".to_string(),
658        }];
659
660        let prepared =
661            prepare_trace_column_rewrites(&rows.rows, pending_rewrites, "trace_type_atomicity")
662                .unwrap();
663
664        prepared.apply(&mut rows);
665        assert_eq!(rows.schema[0].datatype, ColumnDataType::Uint64 as i32);
666        assert_eq!(
667            rows.rows[0].values[0].value_data,
668            Some(ValueData::U64Value(42))
669        );
670    }
671
672    #[test]
673    fn test_prepare_trace_column_rewrites_boolean_rejects_invalid_string_parse() {
674        let rows = vec![Row {
675            values: vec![Value {
676                value_data: Some(ValueData::StringValue("not_a_bool".to_string())),
677            }],
678        }];
679        let pending_rewrites = vec![PendingTraceColumnRewrite {
680            col_idx: 0,
681            target_type: ColumnDataType::Boolean,
682            column_name: "span_attributes.messaging.destination.temporary".to_string(),
683        }];
684
685        let err = prepare_trace_column_rewrites(&rows, pending_rewrites, "trace_type_atomicity")
686            .unwrap_err();
687        assert_eq!(err.error.status_code(), StatusCode::InvalidArguments);
688        assert_eq!(
689            err.column_name,
690            "span_attributes.messaging.destination.temporary"
691        );
692    }
693
694    #[test]
695    fn test_enrich_trace_reconcile_error_includes_existing_type() {
696        let err = enrich_trace_reconcile_error(
697            "trace_type_atomicity",
698            "span_attributes.attr_int",
699            &[ColumnDataType::String, ColumnDataType::Int64],
700            Some(ColumnDataType::Boolean),
701            None,
702        );
703
704        assert_eq!(err.status_code(), StatusCode::InvalidArguments);
705        assert!(err.to_string().contains("span_attributes.attr_int"));
706        assert!(err.to_string().contains("Boolean"));
707    }
708
709    #[test]
710    fn test_enrich_trace_reconcile_error_includes_fixed_semconv_type() {
711        let err = enrich_trace_reconcile_error(
712            "trace_type_atomicity",
713            "span_attributes.server.port",
714            &[ColumnDataType::String, ColumnDataType::Int64],
715            Some(ColumnDataType::String),
716            Some(ColumnDataType::Int64),
717        );
718
719        assert_eq!(err.status_code(), StatusCode::InvalidArguments);
720        assert!(err.to_string().contains("span_attributes.server.port"));
721        assert!(err.to_string().contains("fixed semconv Int64"));
722    }
723
724    #[test]
725    fn test_is_trace_reconcile_candidate_type_filters_non_scalar_types() {
726        assert!(is_trace_reconcile_candidate_type(ColumnDataType::String));
727        assert!(is_trace_reconcile_candidate_type(ColumnDataType::Boolean));
728        assert!(!is_trace_reconcile_candidate_type(ColumnDataType::Binary));
729        assert!(!is_trace_reconcile_candidate_type(
730            ColumnDataType::TimestampMillisecond
731        ));
732    }
733
734    #[test]
735    fn test_push_observed_trace_type_deduplicates_types() {
736        let mut observed_types = Vec::new();
737
738        push_observed_trace_type(&mut observed_types, ColumnDataType::Int64);
739        push_observed_trace_type(&mut observed_types, ColumnDataType::Int64);
740        push_observed_trace_type(&mut observed_types, ColumnDataType::Float64);
741
742        assert_eq!(
743            observed_types,
744            vec![ColumnDataType::Int64, ColumnDataType::Float64]
745        );
746    }
747}