refactor: share timestamp unit conversions in api helper

Address review feedback on the time index unit changeset:

- Add shared timestamp_unit/timestamp_datatype helpers to api::helper
  (the only crate that sees both proto ColumnDataType and TimeUnit due
  to layering; common-time and datatypes have no greptime-proto dep).
  This removes the ColumnDataType -> TimeUnit match duplicated between
  operator's insert path and the OTLP logs path.
- Collapse the two TimeUnit <-> ValueData matches in
  convert_timestamp_value_data by reusing api::helper::to_grpc_value
  for the construction side.
- Note that convert_rows_time_unit rewrites the schema before the
  values, so an overflow mid-batch leaves the request half-converted;
  harmless because the error aborts the whole insert request.

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

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<TimeUnit> {
+    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<ColumnDataTypeWrapper> 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);

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<TimeUnit> {
-    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 {

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<TimeUnit> {
-    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<SelectInfo>,