diff --git a/src/api/src/helper.rs b/src/api/src/helper.rs index 40f9bdfb6d6..c23ae2587d5 100644 --- a/src/api/src/helper.rs +++ b/src/api/src/helper.rs @@ -87,6 +87,28 @@ impl ColumnDataTypeWrapper { } } +/// Returns the time unit if `datatype` is a timestamp type. +pub fn timestamp_unit(datatype: ColumnDataType) -> Option { + match datatype { + ColumnDataType::TimestampSecond => Some(TimeUnit::Second), + ColumnDataType::TimestampMillisecond => Some(TimeUnit::Millisecond), + ColumnDataType::TimestampMicrosecond => Some(TimeUnit::Microsecond), + ColumnDataType::TimestampNanosecond => Some(TimeUnit::Nanosecond), + _ => None, + } +} + +/// Returns the timestamp [ColumnDataType] for the given time unit. +/// This is the inverse of [timestamp_unit]. +pub fn timestamp_datatype(unit: TimeUnit) -> ColumnDataType { + match unit { + TimeUnit::Second => ColumnDataType::TimestampSecond, + TimeUnit::Millisecond => ColumnDataType::TimestampMillisecond, + TimeUnit::Microsecond => ColumnDataType::TimestampMicrosecond, + TimeUnit::Nanosecond => ColumnDataType::TimestampNanosecond, + } +} + impl From for ConcreteDataType { fn from(datatype_wrapper: ColumnDataTypeWrapper) -> Self { match datatype_wrapper.datatype { @@ -1215,6 +1237,21 @@ mod tests { use super::*; use crate::v1::Column; + #[test] + fn test_timestamp_unit_roundtrip() { + for unit in [ + TimeUnit::Second, + TimeUnit::Millisecond, + TimeUnit::Microsecond, + TimeUnit::Nanosecond, + ] { + assert_eq!(timestamp_unit(timestamp_datatype(unit)), Some(unit)); + } + // Non-timestamp types have no time unit. + assert_eq!(timestamp_unit(ColumnDataType::String), None); + assert_eq!(timestamp_unit(ColumnDataType::Datetime), None); + } + #[test] fn test_values_with_capacity() { let values = values_with_capacity(ColumnDataType::Int8, 2); diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index 9be1084f46e..e9392bdecc7 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -1451,17 +1451,23 @@ fn convert_rows_time_unit(rows: &mut Rows, target_unit: TimeUnit) -> Result<()> else { return Ok(()); }; - let Some(source_unit) = timestamp_unit_from_datatype(rows.schema[ts_index].datatype) else { + let Some(source_unit) = ColumnDataType::try_from(rows.schema[ts_index].datatype) + .ok() + .and_then(api::helper::timestamp_unit) + else { return Ok(()); }; if source_unit == target_unit { return Ok(()); } - rows.schema[ts_index].datatype = column_datatype_from_time_unit(target_unit) as i32; + rows.schema[ts_index].datatype = api::helper::timestamp_datatype(target_unit) as i32; // Timestamp columns never carry a datatype extension. rows.schema[ts_index].datatype_extension = None; + // Note: the schema is rewritten before the rows are converted, so an + // overflow error mid-batch leaves this request half-converted. That is + // harmless: the error aborts the whole insert request. for row in &mut rows.rows { let Some(value) = row.values.get_mut(ts_index) else { continue; @@ -1475,27 +1481,6 @@ fn convert_rows_time_unit(rows: &mut Rows, target_unit: TimeUnit) -> Result<()> Ok(()) } -fn timestamp_unit_from_datatype(datatype: i32) -> Option { - ColumnDataType::try_from(datatype) - .ok() - .and_then(|datatype| match datatype { - ColumnDataType::TimestampSecond => Some(TimeUnit::Second), - ColumnDataType::TimestampMillisecond => Some(TimeUnit::Millisecond), - ColumnDataType::TimestampMicrosecond => Some(TimeUnit::Microsecond), - ColumnDataType::TimestampNanosecond => Some(TimeUnit::Nanosecond), - _ => None, - }) -} - -fn column_datatype_from_time_unit(unit: TimeUnit) -> ColumnDataType { - match unit { - TimeUnit::Second => ColumnDataType::TimestampSecond, - TimeUnit::Millisecond => ColumnDataType::TimestampMillisecond, - TimeUnit::Microsecond => ColumnDataType::TimestampMicrosecond, - TimeUnit::Nanosecond => ColumnDataType::TimestampNanosecond, - } -} - fn convert_timestamp_value_data( value_data: ValueData, source_unit: TimeUnit, @@ -1516,12 +1501,7 @@ fn convert_timestamp_value_data( timestamp.value() ), })?; - Ok(Some(match target_unit { - TimeUnit::Second => ValueData::TimestampSecondValue(converted.value()), - TimeUnit::Millisecond => ValueData::TimestampMillisecondValue(converted.value()), - TimeUnit::Microsecond => ValueData::TimestampMicrosecondValue(converted.value()), - TimeUnit::Nanosecond => ValueData::TimestampNanosecondValue(converted.value()), - })) + Ok(api::helper::to_grpc_value(datatypes::value::Value::Timestamp(converted)).value_data) } fn table_is_native_histogram(table: &TableRef) -> bool { diff --git a/src/servers/src/otlp/logs.rs b/src/servers/src/otlp/logs.rs index f32816fbe36..53a2189044d 100644 --- a/src/servers/src/otlp/logs.rs +++ b/src/servers/src/otlp/logs.rs @@ -15,7 +15,7 @@ use std::collections::BTreeMap; use ahash::{HashMap, HashMapExt}; -use api::helper::ColumnDataTypeWrapper; +use api::helper::{ColumnDataTypeWrapper, timestamp_unit}; use api::v1::column_data_type_extension::TypeExt; use api::v1::column_def::options_from_column_schema; use api::v1::value::ValueData; @@ -898,16 +898,6 @@ fn is_timestamp_type(datatype: ColumnDataType) -> bool { timestamp_unit(datatype).is_some() } -fn timestamp_unit(datatype: ColumnDataType) -> Option { - match datatype { - ColumnDataType::TimestampSecond => Some(TimeUnit::Second), - ColumnDataType::TimestampMillisecond => Some(TimeUnit::Millisecond), - ColumnDataType::TimestampMicrosecond => Some(TimeUnit::Microsecond), - ColumnDataType::TimestampNanosecond => Some(TimeUnit::Nanosecond), - _ => None, - } -} - fn parse_export_logs_service_request_to_rows( request: ExportLogsServiceRequest, select_info: Box,