1use 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
25const TRACE_VALUE_DIAGNOSTIC_LIMIT: usize = 16;
28
29pub(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
40fn 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 _ => 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#[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#[derive(Debug)]
89pub(super) struct PreparedTraceColumnRewrites {
90 columns: Vec<PreparedTraceColumnRewrite>,
91 values: Vec<PreparedTraceValueRewrite>,
92}
93
94#[derive(Debug)]
96pub(super) struct TraceColumnRewriteError {
97 pub(super) error: servers::error::Error,
98 pub(super) column_name: String,
99}
100
101#[derive(Debug)]
103struct PreparedTraceColumnRewrite {
104 col_idx: usize,
105 target_type: ColumnDataType,
106}
107
108#[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
127pub(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
208pub(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
306pub(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
318pub(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 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 #[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 #[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 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}