feat: otlp duration_nano and trace_flag signed integer coercion (#8816)

* feat(servers): add signed→unsigned int coercion to OTLP ingest path

Phase 0 of transitioning built-in data models from unsigned to signed
integers (#8793): add lossless Int64→UInt64 and Int32→UInt32 coercion
arms so existing UInt64/UInt32 columns (e.g. trace `duration_nano`, log
`trace_flags`) keep accepting new signed ingest without an ALTER TABLE.

The OTLP ingest path already reconciles every incoming column against the
existing table schema and treats it as authoritative. With these arms,
`choose_trace_reconcile_decision` returns `UseExisting(UInt64/Uint32)`
for an existing unsigned column receiving signed data: the table keeps
its type byte-for-byte and the request value is coerced. No persisted
format is mutated; existing data stays readable as-is. This is the safety
net that makes the actual schema flip (Phase 1) safe.

Only the signed→unsigned direction is supported — the reverse would be
lossy for values above the signed range and is intentionally rejected.

Tests cover both new arms plus an end-to-end log test proving an existing
UInt64 column coerces an incoming Int64 value while keeping its type.

Signed-off-by: Ning Sun <sunning@greptime.com>

* feat: add compatibility layer for uint trace/log fields

Signed-off-by: Ning Sun <sunning@greptime.com>

* fix: jaeger test

Signed-off-by: Ning Sun <sunning@greptime.com>

* fix: keep trace v0 unsigned, reject negative span durations

- Keep the frozen v0 data model on UInt64 duration_nano: the signed
  ingest compatibility layer only runs on the v1 path, so flipping v0
  would break writes into every pre-existing v0 table at mito's schema
  check. Pin the schema with unit and integration tests.
- Reject spans whose end precedes their start (or whose duration does
  not fit i64) on the v1 path instead of wrapping: new Int64 tables and
  existing UInt64 tables now fail identically, rather than storing
  negative durations that break the Jaeger query API.
- Extract is_supported_signed_to_unsigned_coercion so the trace and log
  ingest paths share one supported-pair predicate and cannot drift.

Signed-off-by: Ning Sun <sunning@greptime.com>

* fix: clamp negative span durations to zero, revert semantic_graph comment

- Record duration 0 for spans whose end precedes their start instead of
  erroring: a malformed span no longer fails the request, and the value
  written is always a non-negative, in-range i64 so new Int64 tables and
  existing UInt64 tables (via the checked coercion) behave identically.
  Durations above i64::MAX saturate rather than wrap.
- Revert the doc-comment tweak on the semantic_graph test fixture; the
  file is untouched by this PR again.

Signed-off-by: Ning Sun <sunning@greptime.com>

---------

Signed-off-by: Ning Sun <sunning@greptime.com>

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<T>(items: Vec<T>, chunk_size: usize) -> Vec<Vec<T>> {

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";

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 {

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::<bool>().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]

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<ColumnSchema> {
         ),
         (
             "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> {
+    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<KeyValue>) -> ExportLogsServiceRequest {
+        request_with_log_attrs_and_flags(attrs, 0)
+    }
+
+    fn request_with_log_attrs_and_flags(
+        attrs: Vec<KeyValue>,
+        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'")
+        );
+    }
 }

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![

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,