diff --git a/src/frontend/src/instance/otlp/trace_ingest.rs b/src/frontend/src/instance/otlp/trace_ingest.rs index 671f6c6f7c..5951728e6f 100644 --- a/src/frontend/src/instance/otlp/trace_ingest.rs +++ b/src/frontend/src/instance/otlp/trace_ingest.rs @@ -31,7 +31,7 @@ use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; use pipeline::{GreptimePipelineParams, PipelineWay}; use servers::error::{self, Result as ServerResult}; use servers::otlp; -use servers::otlp::coerce::{coerce_value_data, trace_value_datatype}; +use servers::otlp::coerce::{coerce_value_data, is_supported_trace_coercion, trace_value_datatype}; use servers::otlp::trace::span::{TraceSpan, TraceSpanGroup}; use servers::otlp::trace::v1::{TraceBatchSchema, TraceBinaryType, TraceRetryColumns}; use servers::otlp::trace::{SERVICE_NAME_COLUMN, TraceAuxData}; @@ -2025,10 +2025,23 @@ fn trace_logical_types_incompatible( right_datatype: ColumnDataType, right_concrete_type: &ConcreteDataType, ) -> bool { - left_concrete_type != right_concrete_type - && (left_datatype == right_datatype - || !is_trace_reconcile_candidate_type(left_datatype) - || !is_trace_reconcile_candidate_type(right_datatype)) + if left_concrete_type == right_concrete_type { + return false; + } + // A supported coercion in either direction means the two types can be + // reconciled, so they are not logically incompatible. This lets a signed + // request reach the coercion path instead of being excluded up front when + // the existing column is unsigned (e.g. trace `duration_nano` written as + // Int64 into an existing UInt64 column during the unsigned -> signed + // transition). + if is_supported_trace_coercion(left_datatype, right_datatype) + || is_supported_trace_coercion(right_datatype, left_datatype) + { + return false; + } + left_datatype == right_datatype + || !is_trace_reconcile_candidate_type(left_datatype) + || !is_trace_reconcile_candidate_type(right_datatype) } fn chunk_owned(items: Vec, chunk_size: usize) -> Vec> { diff --git a/src/frontend/src/instance/otlp/trace_ingest/tests.rs b/src/frontend/src/instance/otlp/trace_ingest/tests.rs index ec5bad8975..2445da95bb 100644 --- a/src/frontend/src/instance/otlp/trace_ingest/tests.rs +++ b/src/frontend/src/instance/otlp/trace_ingest/tests.rs @@ -659,6 +659,34 @@ fn test_trace_request_schema_isolates_non_candidate_batch() { ); } +#[test] +fn test_trace_request_schema_keeps_signed_request_into_existing_unsigned_column() { + // Regression for the unsigned -> signed transition: an existing + // `duration_nano` UInt64 column must NOT exclude a new signed (Int64) + // batch as "logically incompatible". The batch must reach the + // reconciliation path, which coerces Int64 -> UInt64 in place (no ALTER). + // Excluding it up front made ingestion error after the flip. + let column_name = "duration_nano"; + let int64_schema = field_schema(column_name, ColumnDataType::Int64); + let existing_uint64_schema = + DatatypesSchemaBuilder::try_from_columns(vec![DatatypesColumnSchema::new( + column_name, + ConcreteDataType::uint64_datatype(), + true, + )]) + .unwrap() + .build() + .unwrap(); + + let mut request_schema = TraceRequestSchema::default(); + request_schema.observe_trace_column(0, &int64_schema, Some(ColumnDataType::Int64)); + + assert_eq!( + request_schema.incompatible_schema_observations(Some(&existing_uint64_schema)), + HashMap::new() + ); +} + #[test] fn test_trace_request_schema_isolates_non_candidate_new_column_batch() { let column_name = "span_attributes.payload"; diff --git a/src/frontend/src/instance/otlp/trace_types.rs b/src/frontend/src/instance/otlp/trace_types.rs index 603cc66c0d..0295d12e68 100644 --- a/src/frontend/src/instance/otlp/trace_types.rs +++ b/src/frontend/src/instance/otlp/trace_types.rs @@ -351,6 +351,24 @@ mod tests { ); } + #[test] + fn test_choose_trace_reconcile_decision_existing_uint64_keeps_uint64() { + // Backward-compat for the unsigned -> signed transition: an existing + // table whose `duration_nano` is still UInt64 must keep that type (no + // ALTER) when new signed (Int64) ingest arrives, coercing the value in + // place. This is the no-ALTER guarantee for the trace path; it relies on + // the Int64 -> UInt64 coercion arm added in Phase 0. + assert_eq!( + choose_trace_reconcile_decision( + "duration_nano", + &[ColumnDataType::Int64], + Some(ColumnDataType::Uint64) + ) + .unwrap(), + Some(TraceReconcileDecision::UseExisting(ColumnDataType::Uint64)) + ); + } + #[test] fn test_choose_trace_reconcile_decision_existing_int64_widens_to_float64() { assert_eq!( @@ -612,6 +630,42 @@ mod tests { ); } + #[test] + fn test_prepare_trace_column_rewrites_coerces_int64_into_existing_uint64() { + // Existing-table backward-compat for the trace path: new signed ingest + // arrives as Int64, but the existing `duration_nano` column is UInt64, so + // the rewrite coerces the value into the existing type in place (no + // ALTER). Mirrors what happens for a table created before the + // unsigned -> signed flip. + let mut rows = Rows { + schema: vec![ColumnSchema { + datatype: ColumnDataType::Int64 as i32, + ..Default::default() + }], + rows: vec![Row { + values: vec![Value { + value_data: Some(ValueData::I64Value(42)), + }], + }], + }; + let pending_rewrites = vec![PendingTraceColumnRewrite { + col_idx: 0, + target_type: ColumnDataType::Uint64, + column_name: "duration_nano".to_string(), + }]; + + let prepared = + prepare_trace_column_rewrites(&rows.rows, pending_rewrites, "trace_type_atomicity") + .unwrap(); + + prepared.apply(&mut rows); + assert_eq!(rows.schema[0].datatype, ColumnDataType::Uint64 as i32); + assert_eq!( + rows.rows[0].values[0].value_data, + Some(ValueData::U64Value(42)) + ); + } + #[test] fn test_prepare_trace_column_rewrites_boolean_rejects_invalid_string_parse() { let rows = vec![Row { diff --git a/src/servers/src/otlp/coerce.rs b/src/servers/src/otlp/coerce.rs index 2d87744c5c..2e7dd36ed7 100644 --- a/src/servers/src/otlp/coerce.rs +++ b/src/servers/src/otlp/coerce.rs @@ -30,6 +30,37 @@ pub enum TraceCoerceError { // - String to Int64 // - String to Float64 // - String to Boolean +// +// Lossless signed-to-unsigned integer casts. These let the built-in data +// models move from unsigned to signed integers while existing unsigned tables +// keep accepting new signed ingest without an `ALTER TABLE`: an existing +// UInt64/UInt32 column coerces an incoming Int64/Int32 request into the +// existing type: +// - Int64 to UInt64 (e.g. trace `duration_nano` on the v1 path): checked, +// so negative values are rejected rather than silently wrapping. Counts and +// durations are non-negative by construction; a negative request is a +// malformed value (e.g. a span whose end precedes its start). +// - Int32 to UInt32 (e.g. log `trace_flags`): bit-preserving, because +// `trace_flags` is a bit field whose high bits may legitimately be set +// (e.g. the W3C sampled flag); bit patterns must round-trip exactly. + +/// The signed→unsigned integer coercions that let the built-in data models +/// move from unsigned to signed integers while existing unsigned tables keep +/// accepting new signed ingest without an `ALTER TABLE` (see the pair +/// descriptions in the module-level comment above). Kept as one predicate so +/// the trace and log ingest paths share the same supported pair set and +/// cannot drift. +pub fn is_supported_signed_to_unsigned_coercion( + request_type: ColumnDataType, + target_type: ColumnDataType, +) -> bool { + matches!( + (request_type, target_type), + (ColumnDataType::Int64, ColumnDataType::Uint64) + | (ColumnDataType::Int32, ColumnDataType::Uint32) + ) +} + pub fn is_supported_trace_coercion( request_type: ColumnDataType, target_type: ColumnDataType, @@ -43,7 +74,7 @@ pub fn is_supported_trace_coercion( | (ColumnDataType::String, ColumnDataType::Int64) | (ColumnDataType::String, ColumnDataType::Float64) | (ColumnDataType::String, ColumnDataType::Boolean) - ) + ) || is_supported_signed_to_unsigned_coercion(request_type, target_type) } pub fn coerce_value_data( @@ -88,6 +119,18 @@ pub fn coerce_non_null_value( (ColumnDataType::String, ColumnDataType::Boolean, ValueData::StringValue(s)) => { s.parse::().ok().map(ValueData::BoolValue) } + // Checked signed -> unsigned cast for built-in fields moving to signed + // types (durations, counts are always non-negative). Negative values + // are rejected instead of wrapping so the coercion stays lossless. + (ColumnDataType::Int64, ColumnDataType::Uint64, ValueData::I64Value(n)) => { + u64::try_from(*n).ok().map(ValueData::U64Value) + } + // Bit-preserving cast for the `trace_flags` bit field: high bits may + // legitimately be set, and the bit pattern must survive the round-trip + // into an existing UInt32 column unchanged. + (ColumnDataType::Int32, ColumnDataType::Uint32, ValueData::I32Value(n)) => { + Some(ValueData::U32Value(*n as u32)) + } _ => None, } } @@ -240,6 +283,55 @@ mod tests { assert_eq!(result, Err(TraceCoerceError::Unsupported)); } + #[test] + fn test_coerce_int64_to_uint64() { + // Non-negative durations coerce losslessly into an existing UInt64 + // column, so built-in fields moving to signed keep accepting writes. + let result = coerce_value_data( + &Some(ValueData::I64Value(123)), + ColumnDataType::Uint64, + ColumnDataType::Int64, + ); + assert_eq!(result, Ok(Some(ValueData::U64Value(123)))); + } + + #[test] + fn test_coerce_negative_int64_to_uint64_rejected() { + // Negative values must not wrap into the unsigned column: the + // signed -> unsigned coercion is only lossless for non-negative + // inputs, so anything else is rejected instead of silently wrapping + // to a huge number. + let result = coerce_value_data( + &Some(ValueData::I64Value(-1)), + ColumnDataType::Uint64, + ColumnDataType::Int64, + ); + assert_eq!(result, Err(TraceCoerceError::Unsupported)); + } + + #[test] + fn test_coerce_int32_to_uint32() { + let result = coerce_value_data( + &Some(ValueData::I32Value(7)), + ColumnDataType::Uint32, + ColumnDataType::Int32, + ); + assert_eq!(result, Ok(Some(ValueData::U32Value(7)))); + } + + #[test] + fn test_coerce_uint_to_int_not_supported() { + // Only the signed -> unsigned direction is supported (the direction + // the no-ALTER transition needs); the reverse would be lossy for + // values above the signed range and is intentionally rejected. + let result = coerce_value_data( + &Some(ValueData::U64Value(9)), + ColumnDataType::Int64, + ColumnDataType::Uint64, + ); + assert_eq!(result, Err(TraceCoerceError::Unsupported)); + } + #[test] fn test_coerce_none_value() { let result = coerce_value_data(&None, ColumnDataType::Float64, ColumnDataType::Int64); @@ -280,6 +372,20 @@ mod tests { ColumnDataType::Binary, ColumnDataType::Json )); + // Signed -> unsigned casts are supported (built-in no-ALTER transition). + assert!(is_supported_trace_coercion( + ColumnDataType::Int64, + ColumnDataType::Uint64 + )); + assert!(is_supported_trace_coercion( + ColumnDataType::Int32, + ColumnDataType::Uint32 + )); + // The reverse direction is intentionally not supported (lossy). + assert!(!is_supported_trace_coercion( + ColumnDataType::Uint64, + ColumnDataType::Int64 + )); } #[test] diff --git a/src/servers/src/otlp/logs.rs b/src/servers/src/otlp/logs.rs index 33d4f4eb5e..74f1a3d03a 100644 --- a/src/servers/src/otlp/logs.rs +++ b/src/servers/src/otlp/logs.rs @@ -44,7 +44,7 @@ use crate::error::{ UnsupportedJsonDataTypeForTagSnafu, }; use crate::http::event::PipelineIngestRequest; -use crate::otlp::coerce::coerce_value_data; +use crate::otlp::coerce::{coerce_value_data, is_supported_signed_to_unsigned_coercion}; use crate::otlp::trace::attributes::OtlpAnyValue; use crate::otlp::utils::{bytes_to_hex_string, key_value_to_jsonb}; use crate::pipeline::run_pipeline; @@ -257,7 +257,7 @@ fn build_otlp_logs_identity_schema() -> Vec { ), ( "trace_flags", - ColumnDataType::Uint32, + ColumnDataType::Int32, SemanticType::Field, None, None, @@ -440,7 +440,7 @@ fn build_otlp_build_in_row( value_data: Some(ValueData::BinaryValue(log_attr.to_vec())), }, GreptimeValue { - value_data: Some(ValueData::U32Value(log.flags)), + value_data: Some(ValueData::I32Value(log.flags as i32)), }, GreptimeValue { value_data: parse_ctx.scope_name.clone().map(ValueData::StringValue), @@ -549,14 +549,17 @@ fn decide_column_schema_and_convert_value( key: column_name, } .fail(), - JsonbNumber::UInt64(u) => Ok(Some(( - GreptimeValue { - value_data: Some(ValueData::U64Value(u)), - }, - ColumnDataType::Uint64, - SemanticType::Tag, - None, - ))), + JsonbNumber::UInt64(u) => { + let value = jsonb_uint64_to_log_value(u, column_name)?; + Ok(Some(( + GreptimeValue { + value_data: Some(ValueData::I64Value(value)), + }, + ColumnDataType::Int64, + SemanticType::Tag, + None, + ))) + } }, JsonbValue::Bool(b) => Ok(Some(( GreptimeValue { @@ -615,6 +618,21 @@ fn decide_existing_column_schema_and_convert_value( ))) } +/// Converts a JSON `UInt64` number into the signed value used by built-in +/// log columns. `JsonbNumber::UInt64` only arises for values that do not fit +/// in `i64` (see `decide_column_schema_and_convert_value`), so the conversion +/// is fallible: wrapping to a negative number would silently corrupt the +/// value (a counter reading back as negative), so out-of-range values are +/// rejected instead. +fn jsonb_uint64_to_log_value(u: u64, column_name: &str) -> Result { + i64::try_from(u).map_err(|_| InvalidParameterSnafu { + reason: format!( + "uint64 value {u} in column '{column_name}' exceeds the i64 range supported by built-in log columns" + ), + } + .build()) +} + fn jsonb_value_to_log_value_data( column_name: &str, value: JsonbValue, @@ -635,7 +653,10 @@ fn jsonb_value_to_log_value_data( key: column_name, } .fail(), - JsonbNumber::UInt64(u) => Ok(Some((ValueData::U64Value(u), ColumnDataType::Uint64))), + JsonbNumber::UInt64(u) => Ok(Some(( + ValueData::I64Value(jsonb_uint64_to_log_value(u, column_name)?), + ColumnDataType::Int64, + ))), }, JsonbValue::Bool(b) => Ok(Some((ValueData::BoolValue(b), ColumnDataType::Boolean))), JsonbValue::Array(_) | JsonbValue::Object(_) => UnsupportedJsonDataTypeForTagSnafu { @@ -718,6 +739,14 @@ fn coerce_log_value_data( return align_timestamp_value(value_data, target_unit, column_name, table_name).map(Some); } + // Lossless signed -> unsigned integer cast for built-in fields moving from + // unsigned to signed types (e.g. an existing UInt64/UInt32 log column + // receiving new Int64/Int32 ingest). Lets existing tables keep their + // unsigned columns without an `ALTER`. In the mutually-exclusive `else` of + // the String branch so the move stays local to non-String targets; the + // shared predicate keeps the supported pairs identical to the trace path + // (the generic String -> numeric coercions keep their pre-existing + // rejection on this path). if target_type == ColumnDataType::String { if let Ok(value_data) = coerce_value_data(&Some(value_data.clone()), target_type, request_type) @@ -727,6 +756,11 @@ fn coerce_log_value_data( if let Some(value_data) = stringify_scalar_value(value_data) { return Ok(Some(value_data)); } + } else if is_supported_signed_to_unsigned_coercion(request_type, target_type) + && let Ok(Some(value_data)) = + coerce_value_data(&Some(value_data), target_type, request_type) + { + return Ok(Some(value_data)); } InvalidParameterSnafu { @@ -1089,6 +1123,16 @@ mod tests { ExistingLogSchema::try_from_schema_parts(&columns, primary_key_indices).unwrap() } + fn existing_uint32_trace_flags_schema() -> ExistingLogSchema { + existing_schema( + vec![ + time_column(ConcreteDataType::timestamp_nanosecond_datatype()), + column("trace_flags", ConcreteDataType::uint32_datatype()), + ], + &[], + ) + } + fn kv(key: &str, value: OtlpValue) -> KeyValue { KeyValue { key: key.to_string(), @@ -1098,12 +1142,20 @@ mod tests { } fn request_with_log_attrs(attrs: Vec) -> ExportLogsServiceRequest { + request_with_log_attrs_and_flags(attrs, 0) + } + + fn request_with_log_attrs_and_flags( + attrs: Vec, + flags: u32, + ) -> ExportLogsServiceRequest { ExportLogsServiceRequest { resource_logs: vec![ResourceLogs { scope_logs: vec![ScopeLogs { log_records: vec![LogRecord { time_unix_nano: 1_234_000_000, trace_id: vec![1; 16], + flags, attributes: attrs, ..Default::default() }], @@ -1151,6 +1203,83 @@ mod tests { ); } + #[test] + fn test_fresh_table_trace_flags_is_int32() { + // Phase 1 of the unsigned -> signed transition: a fresh log table now + // creates `trace_flags` as Int32. Existing tables keep UInt32 (covered + // by the coercion test above); new tables start signed. + let rows = parse_with_select(request_with_log_attrs(vec![]), "", None).unwrap(); + let idx = column_index(&rows, "trace_flags"); + + assert_eq!(rows.schema[idx].datatype, ColumnDataType::Int32 as i32); + assert_eq!( + rows.rows[0].values[idx].value_data, + Some(ValueData::I32Value(0)) + ); + } + + #[test] + fn test_existing_uint32_trace_flags_keeps_type_and_coerces_int32_request() { + // An existing UInt32 `trace_flags` column keeps its type when new + // signed (Int32) ingest arrives, without an `ALTER`. + let existing = existing_uint32_trace_flags_schema(); + + let rows = parse_with_select(request_with_log_attrs(vec![]), "", Some(&existing)).unwrap(); + let idx = column_index(&rows, "trace_flags"); + + assert_eq!(rows.schema[idx].datatype, ColumnDataType::Uint32 as i32); + assert_eq!( + rows.rows[0].values[idx].value_data, + Some(ValueData::U32Value(0)) + ); + } + + #[test] + fn test_existing_uint32_trace_flags_preserves_nonzero_flags_value() { + // A non-zero trace_flags bit pattern (e.g. the W3C sampled flag as sent + // by common encoders) round-trips exactly through the Int32 -> UInt32 + // coercion into an existing unsigned column. Guards that the cast is + // bit-preserving for realistic values, not just the default 0. + let existing = existing_uint32_trace_flags_schema(); + + let rows = parse_with_select( + request_with_log_attrs_and_flags(vec![], 256), + "", + Some(&existing), + ) + .unwrap(); + let idx = column_index(&rows, "trace_flags"); + + assert_eq!(rows.schema[idx].datatype, ColumnDataType::Uint32 as i32); + assert_eq!( + rows.rows[0].values[idx].value_data, + Some(ValueData::U32Value(256)) + ); + } + + #[test] + fn test_existing_uint32_trace_flags_preserves_high_bit_flags_value() { + // The full u32 range round-trips: a value with the sign bit set proves + // the Int32 -> UInt32 coercion is bit-exact (`as i32` then `as u32` + // preserve the pattern), so no flags value is corrupted on the + // unsigned -> signed transition. + let existing = existing_uint32_trace_flags_schema(); + + let rows = parse_with_select( + request_with_log_attrs_and_flags(vec![], 0x8000_0000), + "", + Some(&existing), + ) + .unwrap(); + let idx = column_index(&rows, "trace_flags"); + + assert_eq!(rows.schema[idx].datatype, ColumnDataType::Uint32 as i32); + assert_eq!( + rows.rows[0].values[idx].value_data, + Some(ValueData::U32Value(0x8000_0000)) + ); + } + #[test] fn test_existing_primary_key_updates_builtin_column_semantic_type() { let existing = existing_schema( @@ -1326,4 +1455,82 @@ mod tests { SemanticType::Field as i32 ); } + + #[test] + fn test_existing_uint64_column_keeps_type_and_coerces_int64_request() { + // An existing UInt64 column keeps its type when new signed (Int64) + // ingest arrives, without an `ALTER`. The built-in models move to + // signed while existing unsigned tables stay byte-for-byte unchanged. + let existing = existing_schema( + vec![ + time_column(ConcreteDataType::timestamp_nanosecond_datatype()), + column("counter", ConcreteDataType::uint64_datatype()), + ], + &[], + ); + + let rows = parse_with_select( + request_with_log_attrs(vec![kv("counter", OtlpValue::IntValue(42))]), + "counter", + Some(&existing), + ) + .unwrap(); + let idx = column_index(&rows, "counter"); + + assert_eq!(rows.schema[idx].datatype, ColumnDataType::Uint64 as i32); + assert_eq!( + rows.rows[0].values[idx].value_data, + Some(ValueData::U64Value(42)) + ); + } + + #[test] + fn test_jsonb_uint64_exceeding_i64_range_rejected() { + // A JSON number that only fits in u64 must be rejected, not wrapped: + // storing it as a negative i64 would silently corrupt the value. The + // unsigned -> signed transition is only lossless for values that fit + // the signed range. + let err = jsonb_value_to_log_value_data( + "counter", + JsonbValue::Number(JsonbNumber::UInt64(u64::MAX)), + false, + ) + .unwrap_err(); + + assert!( + err.to_string() + .contains("exceeds the i64 range supported by built-in log columns") + ); + } + + #[test] + fn test_existing_int64_column_rejects_numeric_string_value() { + // Regression guard for the signed -> unsigned transition scope: even a + // parseable numeric string must keep its pre-existing rejection when + // targeting a non-String column. Only the signed -> unsigned integer + // pairs gained coercion on this path, not the generic String -> + // numeric ones. + let existing = existing_schema( + vec![ + time_column(ConcreteDataType::timestamp_nanosecond_datatype()), + column("counter", ConcreteDataType::int64_datatype()), + ], + &[], + ); + + let err = parse_with_select( + request_with_log_attrs(vec![kv( + "counter", + OtlpValue::StringValue("42".to_string()), + )]), + "counter", + Some(&existing), + ) + .unwrap_err(); + + assert!( + err.to_string() + .contains("failed to align log column 'counter'") + ); + } } diff --git a/src/servers/src/otlp/trace/v0.rs b/src/servers/src/otlp/trace/v0.rs index fa10dcc00f..1bc6752e90 100644 --- a/src/servers/src/otlp/trace/v0.rs +++ b/src/servers/src/otlp/trace/v0.rs @@ -108,6 +108,11 @@ pub fn write_span_to_row(writer: &mut TableData, span: TraceSpan) -> Result<()> span.end_in_nanosecond as i64, )), ), + // The v0 data model is frozen: `duration_nano` stays UInt64. The + // signed→unsigned ingest compatibility layer only runs on the v1 path + // (see `trace_ingest.rs`), so flipping v0 here would break writes into + // every pre-existing v0 table at mito's schema check. New v1 tables + // are signed; v0 tables keep the legacy unsigned column indefinitely. make_column_data( DURATION_NANO_COLUMN, ColumnDataType::Uint64, @@ -231,10 +236,13 @@ fn write_trace_operations_to_row( #[cfg(test)] mod tests { + use api::v1::ColumnDataType; + use api::v1::value::ValueData; + use super::{build_aux_table_requests, build_trace_table_data}; - use crate::otlp::trace::TraceAuxData; use crate::otlp::trace::attributes::Attributes; use crate::otlp::trace::span::{SpanEvents, SpanLinks, TraceSpan}; + use crate::otlp::trace::{DURATION_NANO_COLUMN, TraceAuxData}; fn make_span(service_name: &str, trace_id: &str, span_id: &str) -> TraceSpan { TraceSpan { @@ -271,6 +279,23 @@ mod tests { assert_eq!(rows.len(), 1); } + #[test] + fn test_v0_duration_nano_stays_uint64() { + // The v0 data model is frozen on the unsigned `duration_nano`. This + // pins the schema so the unsigned→signed transition of the built-in + // models (which flips only v1) cannot silently leak into v0 and break + // writes into pre-existing v0 tables. + let writer = build_trace_table_data(&[make_span("svc-a", "trace-a", "span-a")]).unwrap(); + let (schema, rows) = writer.into_schema_and_rows(); + + let idx = schema + .iter() + .position(|c| c.column_name == DURATION_NANO_COLUMN) + .unwrap(); + assert_eq!(schema[idx].datatype, ColumnDataType::Uint64 as i32); + assert_eq!(rows[0].values[idx].value_data, Some(ValueData::U64Value(1))); + } + #[test] fn test_build_aux_table_requests_deduplicates_services_and_operations() { let spans = vec![ diff --git a/src/servers/src/otlp/trace/v1.rs b/src/servers/src/otlp/trace/v1.rs index 8ddeac24f3..f55ab48251 100644 --- a/src/servers/src/otlp/trace/v1.rs +++ b/src/servers/src/otlp/trace/v1.rs @@ -79,7 +79,7 @@ impl FixedTraceColumnIndexes { Ok(Self { timestamp, timestamp_end: field("timestamp_end", ColumnDataType::TimestampNanosecond)?, - duration_nano: field(DURATION_NANO_COLUMN, ColumnDataType::Uint64)?, + 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)?, @@ -342,6 +342,22 @@ pub fn write_span_to_row(writer: &mut TableData, span: TraceSpan) -> Result<()> 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`. +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, @@ -367,9 +383,7 @@ fn write_span_to_row_inner( ), ( fixed_columns.duration_nano, - Some(ValueData::U64Value( - span.end_in_nanosecond - span.start_in_nanosecond, - )), + Some(ValueData::I64Value(span_duration_nano(&span))), ), ( fixed_columns.parent_span_id, @@ -689,6 +703,49 @@ mod tests { } } + #[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(); @@ -702,7 +759,7 @@ mod tests { "timestamp_end", Some(ValueData::TimestampNanosecondValue(2)), ), - (DURATION_NANO_COLUMN, Some(ValueData::U64Value(1))), + (DURATION_NANO_COLUMN, Some(ValueData::I64Value(1))), (PARENT_SPAN_ID_COLUMN, None), ( TRACE_ID_COLUMN, diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index 90cbb8c93f..f89d75e9ef 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -6860,6 +6860,19 @@ pub async fn test_otlp_traces_v0(store_type: StorageType) { ) .await; + // The v0 data model is frozen on the unsigned `duration_nano`: pin the + // schema so the unsigned→signed flip of the built-in models (which flips + // only v1) cannot leak into v0 and break writes into pre-existing v0 + // tables. + validate_data( + "otlp_traces_v0_schema", + &client, + "select column_name, lower(data_type) from information_schema.columns \ + where table_name = 'opentelemetry_traces' and column_name = 'duration_nano';", + r#"[["duration_nano","bigint unsigned"]]"#, + ) + .await; + // drop table let res = client .get("/v1/sql?sql=drop table opentelemetry_traces;") @@ -7075,7 +7088,7 @@ pub async fn test_otlp_traces_v1(store_type: StorageType) { ) .await; - let expected_ddl = r#"[["mytable","CREATE TABLE IF NOT EXISTS \"mytable\" (\n \"timestamp\" TIMESTAMP(9) NOT NULL,\n \"timestamp_end\" TIMESTAMP(9) NULL,\n \"duration_nano\" BIGINT UNSIGNED NULL,\n \"parent_span_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"trace_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_id\" STRING NULL,\n \"span_kind\" STRING NULL,\n \"span_name\" STRING NULL,\n \"span_status_code\" STRING NULL,\n \"span_status_message\" STRING NULL,\n \"trace_state\" STRING NULL,\n \"scope_name\" STRING NULL,\n \"scope_version\" STRING NULL,\n \"service_name\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_attributes.net.peer.ip\" STRING NULL,\n \"span_attributes.peer.service\" STRING NULL,\n \"span_events\" JSON NULL,\n \"span_links\" JSON NULL,\n TIME INDEX (\"timestamp\"),\n PRIMARY KEY (\"service_name\")\n)\nPARTITION ON COLUMNS (\"trace_id\") (\n trace_id < '1',\n trace_id >= '1' AND trace_id < '2',\n trace_id >= '2' AND trace_id < '3',\n trace_id >= '3' AND trace_id < '4',\n trace_id >= '4' AND trace_id < '5',\n trace_id >= '5' AND trace_id < '6',\n trace_id >= '6' AND trace_id < '7',\n trace_id >= '7' AND trace_id < '8',\n trace_id >= '8' AND trace_id < '9',\n trace_id >= '9' AND trace_id < 'a',\n trace_id >= 'a' AND trace_id < 'b',\n trace_id >= 'b' AND trace_id < 'c',\n trace_id >= 'c' AND trace_id < 'd',\n trace_id >= 'd' AND trace_id < 'e',\n trace_id >= 'e' AND trace_id < 'f',\n trace_id >= 'f'\n)\nENGINE=mito\nWITH(\n 'comment' = 'Created on insertion',\n append_mode = 'true',\n 'greptime.semantic.entity.service.id' = 'service_name',\n 'greptime.semantic.pipeline' = 'greptime_trace_v1',\n 'greptime.semantic.signal_type' = 'trace',\n 'greptime.semantic.source' = 'opentelemetry',\n 'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.4.0',\n table_data_model = 'greptime_trace_v1'\n)"]]"#; + let expected_ddl = r#"[["mytable","CREATE TABLE IF NOT EXISTS \"mytable\" (\n \"timestamp\" TIMESTAMP(9) NOT NULL,\n \"timestamp_end\" TIMESTAMP(9) NULL,\n \"duration_nano\" BIGINT NULL,\n \"parent_span_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"trace_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_id\" STRING NULL,\n \"span_kind\" STRING NULL,\n \"span_name\" STRING NULL,\n \"span_status_code\" STRING NULL,\n \"span_status_message\" STRING NULL,\n \"trace_state\" STRING NULL,\n \"scope_name\" STRING NULL,\n \"scope_version\" STRING NULL,\n \"service_name\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_attributes.net.peer.ip\" STRING NULL,\n \"span_attributes.peer.service\" STRING NULL,\n \"span_events\" JSON NULL,\n \"span_links\" JSON NULL,\n TIME INDEX (\"timestamp\"),\n PRIMARY KEY (\"service_name\")\n)\nPARTITION ON COLUMNS (\"trace_id\") (\n trace_id < '1',\n trace_id >= '1' AND trace_id < '2',\n trace_id >= '2' AND trace_id < '3',\n trace_id >= '3' AND trace_id < '4',\n trace_id >= '4' AND trace_id < '5',\n trace_id >= '5' AND trace_id < '6',\n trace_id >= '6' AND trace_id < '7',\n trace_id >= '7' AND trace_id < '8',\n trace_id >= '8' AND trace_id < '9',\n trace_id >= '9' AND trace_id < 'a',\n trace_id >= 'a' AND trace_id < 'b',\n trace_id >= 'b' AND trace_id < 'c',\n trace_id >= 'c' AND trace_id < 'd',\n trace_id >= 'd' AND trace_id < 'e',\n trace_id >= 'e' AND trace_id < 'f',\n trace_id >= 'f'\n)\nENGINE=mito\nWITH(\n 'comment' = 'Created on insertion',\n append_mode = 'true',\n 'greptime.semantic.entity.service.id' = 'service_name',\n 'greptime.semantic.pipeline' = 'greptime_trace_v1',\n 'greptime.semantic.signal_type' = 'trace',\n 'greptime.semantic.source' = 'opentelemetry',\n 'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.4.0',\n table_data_model = 'greptime_trace_v1'\n)"]]"#; validate_data( "otlp_traces", &client, @@ -7128,7 +7141,7 @@ pub async fn test_otlp_traces_v1(store_type: StorageType) { ) .await; assert_eq!(StatusCode::OK, res.status()); - let expected_ddl = r#"[["trace_table_part1","CREATE TABLE IF NOT EXISTS \"trace_table_part1\" (\n \"timestamp\" TIMESTAMP(9) NOT NULL,\n \"timestamp_end\" TIMESTAMP(9) NULL,\n \"duration_nano\" BIGINT UNSIGNED NULL,\n \"parent_span_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"trace_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_id\" STRING NULL,\n \"span_kind\" STRING NULL,\n \"span_name\" STRING NULL,\n \"span_status_code\" STRING NULL,\n \"span_status_message\" STRING NULL,\n \"trace_state\" STRING NULL,\n \"scope_name\" STRING NULL,\n \"scope_version\" STRING NULL,\n \"service_name\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_attributes.net.peer.ip\" STRING NULL,\n \"span_attributes.peer.service\" STRING NULL,\n \"span_events\" JSON NULL,\n \"span_links\" JSON NULL,\n TIME INDEX (\"timestamp\"),\n PRIMARY KEY (\"service_name\")\n)\n\nENGINE=mito\nWITH(\n 'comment' = 'Created on insertion',\n append_mode = 'true',\n 'greptime.semantic.entity.service.id' = 'service_name',\n 'greptime.semantic.pipeline' = 'greptime_trace_v1',\n 'greptime.semantic.signal_type' = 'trace',\n 'greptime.semantic.source' = 'opentelemetry',\n 'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.4.0',\n table_data_model = 'greptime_trace_v1'\n)"]]"#; + let expected_ddl = r#"[["trace_table_part1","CREATE TABLE IF NOT EXISTS \"trace_table_part1\" (\n \"timestamp\" TIMESTAMP(9) NOT NULL,\n \"timestamp_end\" TIMESTAMP(9) NULL,\n \"duration_nano\" BIGINT NULL,\n \"parent_span_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"trace_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_id\" STRING NULL,\n \"span_kind\" STRING NULL,\n \"span_name\" STRING NULL,\n \"span_status_code\" STRING NULL,\n \"span_status_message\" STRING NULL,\n \"trace_state\" STRING NULL,\n \"scope_name\" STRING NULL,\n \"scope_version\" STRING NULL,\n \"service_name\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_attributes.net.peer.ip\" STRING NULL,\n \"span_attributes.peer.service\" STRING NULL,\n \"span_events\" JSON NULL,\n \"span_links\" JSON NULL,\n TIME INDEX (\"timestamp\"),\n PRIMARY KEY (\"service_name\")\n)\n\nENGINE=mito\nWITH(\n 'comment' = 'Created on insertion',\n append_mode = 'true',\n 'greptime.semantic.entity.service.id' = 'service_name',\n 'greptime.semantic.pipeline' = 'greptime_trace_v1',\n 'greptime.semantic.signal_type' = 'trace',\n 'greptime.semantic.source' = 'opentelemetry',\n 'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.4.0',\n table_data_model = 'greptime_trace_v1'\n)"]]"#; validate_data( "otlp_traces", &client, @@ -7165,7 +7178,7 @@ pub async fn test_otlp_traces_v1(store_type: StorageType) { ) .await; assert_eq!(StatusCode::OK, res.status()); - let expected_ddl = r#"[["trace_table_part4","CREATE TABLE IF NOT EXISTS \"trace_table_part4\" (\n \"timestamp\" TIMESTAMP(9) NOT NULL,\n \"timestamp_end\" TIMESTAMP(9) NULL,\n \"duration_nano\" BIGINT UNSIGNED NULL,\n \"parent_span_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"trace_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_id\" STRING NULL,\n \"span_kind\" STRING NULL,\n \"span_name\" STRING NULL,\n \"span_status_code\" STRING NULL,\n \"span_status_message\" STRING NULL,\n \"trace_state\" STRING NULL,\n \"scope_name\" STRING NULL,\n \"scope_version\" STRING NULL,\n \"service_name\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_attributes.net.peer.ip\" STRING NULL,\n \"span_attributes.peer.service\" STRING NULL,\n \"span_events\" JSON NULL,\n \"span_links\" JSON NULL,\n TIME INDEX (\"timestamp\"),\n PRIMARY KEY (\"service_name\")\n)\nPARTITION ON COLUMNS (\"trace_id\") (\n trace_id < '4',\n trace_id >= '4' AND trace_id < '8',\n trace_id >= '8' AND trace_id < 'c',\n trace_id >= 'c'\n)\nENGINE=mito\nWITH(\n 'comment' = 'Created on insertion',\n append_mode = 'true',\n 'greptime.semantic.entity.service.id' = 'service_name',\n 'greptime.semantic.pipeline' = 'greptime_trace_v1',\n 'greptime.semantic.signal_type' = 'trace',\n 'greptime.semantic.source' = 'opentelemetry',\n 'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.4.0',\n table_data_model = 'greptime_trace_v1'\n)"]]"#; + let expected_ddl = r#"[["trace_table_part4","CREATE TABLE IF NOT EXISTS \"trace_table_part4\" (\n \"timestamp\" TIMESTAMP(9) NOT NULL,\n \"timestamp_end\" TIMESTAMP(9) NULL,\n \"duration_nano\" BIGINT NULL,\n \"parent_span_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"trace_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_id\" STRING NULL,\n \"span_kind\" STRING NULL,\n \"span_name\" STRING NULL,\n \"span_status_code\" STRING NULL,\n \"span_status_message\" STRING NULL,\n \"trace_state\" STRING NULL,\n \"scope_name\" STRING NULL,\n \"scope_version\" STRING NULL,\n \"service_name\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_attributes.net.peer.ip\" STRING NULL,\n \"span_attributes.peer.service\" STRING NULL,\n \"span_events\" JSON NULL,\n \"span_links\" JSON NULL,\n TIME INDEX (\"timestamp\"),\n PRIMARY KEY (\"service_name\")\n)\nPARTITION ON COLUMNS (\"trace_id\") (\n trace_id < '4',\n trace_id >= '4' AND trace_id < '8',\n trace_id >= '8' AND trace_id < 'c',\n trace_id >= 'c'\n)\nENGINE=mito\nWITH(\n 'comment' = 'Created on insertion',\n append_mode = 'true',\n 'greptime.semantic.entity.service.id' = 'service_name',\n 'greptime.semantic.pipeline' = 'greptime_trace_v1',\n 'greptime.semantic.signal_type' = 'trace',\n 'greptime.semantic.source' = 'opentelemetry',\n 'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.4.0',\n table_data_model = 'greptime_trace_v1'\n)"]]"#; validate_data( "otlp_traces", &client, @@ -7202,7 +7215,7 @@ pub async fn test_otlp_traces_v1(store_type: StorageType) { ) .await; assert_eq!(StatusCode::OK, res.status()); - let expected_ddl = r#"[["trace_table_part32","CREATE TABLE IF NOT EXISTS \"trace_table_part32\" (\n \"timestamp\" TIMESTAMP(9) NOT NULL,\n \"timestamp_end\" TIMESTAMP(9) NULL,\n \"duration_nano\" BIGINT UNSIGNED NULL,\n \"parent_span_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"trace_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_id\" STRING NULL,\n \"span_kind\" STRING NULL,\n \"span_name\" STRING NULL,\n \"span_status_code\" STRING NULL,\n \"span_status_message\" STRING NULL,\n \"trace_state\" STRING NULL,\n \"scope_name\" STRING NULL,\n \"scope_version\" STRING NULL,\n \"service_name\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_attributes.net.peer.ip\" STRING NULL,\n \"span_attributes.peer.service\" STRING NULL,\n \"span_events\" JSON NULL,\n \"span_links\" JSON NULL,\n TIME INDEX (\"timestamp\"),\n PRIMARY KEY (\"service_name\")\n)\nPARTITION ON COLUMNS (\"trace_id\") (\n trace_id < '08',\n trace_id >= '08' AND trace_id < '10',\n trace_id >= '10' AND trace_id < '18',\n trace_id >= '18' AND trace_id < '20',\n trace_id >= '20' AND trace_id < '28',\n trace_id >= '28' AND trace_id < '30',\n trace_id >= '30' AND trace_id < '38',\n trace_id >= '38' AND trace_id < '40',\n trace_id >= '40' AND trace_id < '48',\n trace_id >= '48' AND trace_id < '50',\n trace_id >= '50' AND trace_id < '58',\n trace_id >= '58' AND trace_id < '60',\n trace_id >= '60' AND trace_id < '68',\n trace_id >= '68' AND trace_id < '70',\n trace_id >= '70' AND trace_id < '78',\n trace_id >= '78' AND trace_id < '80',\n trace_id >= '80' AND trace_id < '88',\n trace_id >= '88' AND trace_id < '90',\n trace_id >= '90' AND trace_id < '98',\n trace_id >= '98' AND trace_id < 'a0',\n trace_id >= 'a0' AND trace_id < 'a8',\n trace_id >= 'a8' AND trace_id < 'b0',\n trace_id >= 'b0' AND trace_id < 'b8',\n trace_id >= 'b8' AND trace_id < 'c0',\n trace_id >= 'c0' AND trace_id < 'c8',\n trace_id >= 'c8' AND trace_id < 'd0',\n trace_id >= 'd0' AND trace_id < 'd8',\n trace_id >= 'd8' AND trace_id < 'e0',\n trace_id >= 'e0' AND trace_id < 'e8',\n trace_id >= 'e8' AND trace_id < 'f0',\n trace_id >= 'f0' AND trace_id < 'f8',\n trace_id >= 'f8'\n)\nENGINE=mito\nWITH(\n 'comment' = 'Created on insertion',\n append_mode = 'true',\n 'greptime.semantic.entity.service.id' = 'service_name',\n 'greptime.semantic.pipeline' = 'greptime_trace_v1',\n 'greptime.semantic.signal_type' = 'trace',\n 'greptime.semantic.source' = 'opentelemetry',\n 'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.4.0',\n table_data_model = 'greptime_trace_v1'\n)"]]"#; + let expected_ddl = r#"[["trace_table_part32","CREATE TABLE IF NOT EXISTS \"trace_table_part32\" (\n \"timestamp\" TIMESTAMP(9) NOT NULL,\n \"timestamp_end\" TIMESTAMP(9) NULL,\n \"duration_nano\" BIGINT NULL,\n \"parent_span_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"trace_id\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_id\" STRING NULL,\n \"span_kind\" STRING NULL,\n \"span_name\" STRING NULL,\n \"span_status_code\" STRING NULL,\n \"span_status_message\" STRING NULL,\n \"trace_state\" STRING NULL,\n \"scope_name\" STRING NULL,\n \"scope_version\" STRING NULL,\n \"service_name\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\n \"span_attributes.net.peer.ip\" STRING NULL,\n \"span_attributes.peer.service\" STRING NULL,\n \"span_events\" JSON NULL,\n \"span_links\" JSON NULL,\n TIME INDEX (\"timestamp\"),\n PRIMARY KEY (\"service_name\")\n)\nPARTITION ON COLUMNS (\"trace_id\") (\n trace_id < '08',\n trace_id >= '08' AND trace_id < '10',\n trace_id >= '10' AND trace_id < '18',\n trace_id >= '18' AND trace_id < '20',\n trace_id >= '20' AND trace_id < '28',\n trace_id >= '28' AND trace_id < '30',\n trace_id >= '30' AND trace_id < '38',\n trace_id >= '38' AND trace_id < '40',\n trace_id >= '40' AND trace_id < '48',\n trace_id >= '48' AND trace_id < '50',\n trace_id >= '50' AND trace_id < '58',\n trace_id >= '58' AND trace_id < '60',\n trace_id >= '60' AND trace_id < '68',\n trace_id >= '68' AND trace_id < '70',\n trace_id >= '70' AND trace_id < '78',\n trace_id >= '78' AND trace_id < '80',\n trace_id >= '80' AND trace_id < '88',\n trace_id >= '88' AND trace_id < '90',\n trace_id >= '90' AND trace_id < '98',\n trace_id >= '98' AND trace_id < 'a0',\n trace_id >= 'a0' AND trace_id < 'a8',\n trace_id >= 'a8' AND trace_id < 'b0',\n trace_id >= 'b0' AND trace_id < 'b8',\n trace_id >= 'b8' AND trace_id < 'c0',\n trace_id >= 'c0' AND trace_id < 'c8',\n trace_id >= 'c8' AND trace_id < 'd0',\n trace_id >= 'd0' AND trace_id < 'd8',\n trace_id >= 'd8' AND trace_id < 'e0',\n trace_id >= 'e0' AND trace_id < 'e8',\n trace_id >= 'e8' AND trace_id < 'f0',\n trace_id >= 'f0' AND trace_id < 'f8',\n trace_id >= 'f8'\n)\nENGINE=mito\nWITH(\n 'comment' = 'Created on insertion',\n append_mode = 'true',\n 'greptime.semantic.entity.service.id' = 'service_name',\n 'greptime.semantic.pipeline' = 'greptime_trace_v1',\n 'greptime.semantic.signal_type' = 'trace',\n 'greptime.semantic.source' = 'opentelemetry',\n 'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.4.0',\n table_data_model = 'greptime_trace_v1'\n)"]]"#; validate_data( "otlp_traces", &client, @@ -9601,7 +9614,7 @@ pub async fn test_jaeger_query_api_for_trace_v1(store_type: StorageType) { .await; assert_eq!(StatusCode::OK, res.status()); - let trace_table_sql = "[[\"mytable\",\"CREATE TABLE IF NOT EXISTS \\\"mytable\\\" (\\n \\\"timestamp\\\" TIMESTAMP(9) NOT NULL,\\n \\\"timestamp_end\\\" TIMESTAMP(9) NULL,\\n \\\"duration_nano\\\" BIGINT UNSIGNED NULL,\\n \\\"parent_span_id\\\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\\n \\\"trace_id\\\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\\n \\\"span_id\\\" STRING NULL,\\n \\\"span_kind\\\" STRING NULL,\\n \\\"span_name\\\" STRING NULL,\\n \\\"span_status_code\\\" STRING NULL,\\n \\\"span_status_message\\\" STRING NULL,\\n \\\"trace_state\\\" STRING NULL,\\n \\\"scope_name\\\" STRING NULL,\\n \\\"scope_version\\\" STRING NULL,\\n \\\"service_name\\\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\\n \\\"span_attributes.operation.type\\\" STRING NULL,\\n \\\"span_attributes.net.peer.ip\\\" STRING NULL,\\n \\\"span_attributes.peer.service\\\" STRING NULL,\\n \\\"span_events\\\" JSON NULL,\\n \\\"span_links\\\" JSON NULL,\\n TIME INDEX (\\\"timestamp\\\"),\\n PRIMARY KEY (\\\"service_name\\\")\\n)\\nPARTITION ON COLUMNS (\\\"trace_id\\\") (\\n trace_id < '1',\\n trace_id >= '1' AND trace_id < '2',\\n trace_id >= '2' AND trace_id < '3',\\n trace_id >= '3' AND trace_id < '4',\\n trace_id >= '4' AND trace_id < '5',\\n trace_id >= '5' AND trace_id < '6',\\n trace_id >= '6' AND trace_id < '7',\\n trace_id >= '7' AND trace_id < '8',\\n trace_id >= '8' AND trace_id < '9',\\n trace_id >= '9' AND trace_id < 'a',\\n trace_id >= 'a' AND trace_id < 'b',\\n trace_id >= 'b' AND trace_id < 'c',\\n trace_id >= 'c' AND trace_id < 'd',\\n trace_id >= 'd' AND trace_id < 'e',\\n trace_id >= 'e' AND trace_id < 'f',\\n trace_id >= 'f'\\n)\\nENGINE=mito\\nWITH(\\n 'comment' = 'Created on insertion',\\n append_mode = 'true',\\n 'greptime.semantic.entity.service.id' = 'service_name',\\n 'greptime.semantic.pipeline' = 'greptime_trace_v1',\\n 'greptime.semantic.signal_type' = 'trace',\\n 'greptime.semantic.source' = 'opentelemetry',\\n 'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.4.0',\\n table_data_model = 'greptime_trace_v1',\\n ttl = '7days'\\n)\"]]"; + let trace_table_sql = "[[\"mytable\",\"CREATE TABLE IF NOT EXISTS \\\"mytable\\\" (\\n \\\"timestamp\\\" TIMESTAMP(9) NOT NULL,\\n \\\"timestamp_end\\\" TIMESTAMP(9) NULL,\\n \\\"duration_nano\\\" BIGINT NULL,\\n \\\"parent_span_id\\\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\\n \\\"trace_id\\\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\\n \\\"span_id\\\" STRING NULL,\\n \\\"span_kind\\\" STRING NULL,\\n \\\"span_name\\\" STRING NULL,\\n \\\"span_status_code\\\" STRING NULL,\\n \\\"span_status_message\\\" STRING NULL,\\n \\\"trace_state\\\" STRING NULL,\\n \\\"scope_name\\\" STRING NULL,\\n \\\"scope_version\\\" STRING NULL,\\n \\\"service_name\\\" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'),\\n \\\"span_attributes.operation.type\\\" STRING NULL,\\n \\\"span_attributes.net.peer.ip\\\" STRING NULL,\\n \\\"span_attributes.peer.service\\\" STRING NULL,\\n \\\"span_events\\\" JSON NULL,\\n \\\"span_links\\\" JSON NULL,\\n TIME INDEX (\\\"timestamp\\\"),\\n PRIMARY KEY (\\\"service_name\\\")\\n)\\nPARTITION ON COLUMNS (\\\"trace_id\\\") (\\n trace_id < '1',\\n trace_id >= '1' AND trace_id < '2',\\n trace_id >= '2' AND trace_id < '3',\\n trace_id >= '3' AND trace_id < '4',\\n trace_id >= '4' AND trace_id < '5',\\n trace_id >= '5' AND trace_id < '6',\\n trace_id >= '6' AND trace_id < '7',\\n trace_id >= '7' AND trace_id < '8',\\n trace_id >= '8' AND trace_id < '9',\\n trace_id >= '9' AND trace_id < 'a',\\n trace_id >= 'a' AND trace_id < 'b',\\n trace_id >= 'b' AND trace_id < 'c',\\n trace_id >= 'c' AND trace_id < 'd',\\n trace_id >= 'd' AND trace_id < 'e',\\n trace_id >= 'e' AND trace_id < 'f',\\n trace_id >= 'f'\\n)\\nENGINE=mito\\nWITH(\\n 'comment' = 'Created on insertion',\\n append_mode = 'true',\\n 'greptime.semantic.entity.service.id' = 'service_name',\\n 'greptime.semantic.pipeline' = 'greptime_trace_v1',\\n 'greptime.semantic.signal_type' = 'trace',\\n 'greptime.semantic.source' = 'opentelemetry',\\n 'greptime.semantic.trace.conventions' = 'https://opentelemetry.io/schemas/1.4.0',\\n table_data_model = 'greptime_trace_v1',\\n ttl = '7days'\\n)\"]]"; validate_data( "trace_v1_create_table", &client,