feat: impl duration datatype and vectors (#2180)

* feat: impl datatype, vector traits for duration.

* feat: duration and grpc.

* test: add unit test cases.

* chore: style and test case.

* fix: update greptime-proto version and helper.rs

* chore: fix type name.

* Update src/datatypes/src/data_type.rs

Co-authored-by: Yingwen <realevenyag@gmail.com>

* chore: cr.

* chore: fix greptime-proto

---------

Co-authored-by: Yingwen <realevenyag@gmail.com>

Cargo.lock

@@ -4102,7 +4102,7 @@ checksum = "d2fabcfbdc87f4758337ca535fb41a6d701b65693ce38287d856d1674551ec9b"
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=637c54338f0cd0a333fd67bcb37d79dabb014453#637c54338f0cd0a333fd67bcb37d79dabb014453"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=9d3f28d07d29607d0e3c1823f4a4d2bc229d05b9#9d3f28d07d29607d0e3c1823f4a4d2bc229d05b9"
 dependencies = [
  "prost",
  "serde",

Cargo.toml

@@ -80,7 +80,7 @@ datafusion-substrait = { git = "https://github.com/waynexia/arrow-datafusion.git
 derive_builder = "0.12"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "637c54338f0cd0a333fd67bcb37d79dabb014453" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "9d3f28d07d29607d0e3c1823f4a4d2bc229d05b9" }
 humantime-serde = "1.1"
 itertools = "0.10"
 lazy_static = "1.4"

src/api/src/helper.rs

@@ -18,16 +18,17 @@ use common_base::BitVec;
 use common_time::interval::IntervalUnit;
 use common_time::time::Time;
 use common_time::timestamp::TimeUnit;
-use common_time::{Date, DateTime, Interval, Timestamp};
+use common_time::{Date, DateTime, Duration, Interval, Timestamp};
 use datatypes::prelude::{ConcreteDataType, ValueRef};
 use datatypes::scalars::ScalarVector;
 use datatypes::types::{
-    Int16Type, Int8Type, IntervalType, TimeType, TimestampType, UInt16Type, UInt8Type,
+    DurationType, Int16Type, Int8Type, IntervalType, TimeType, TimestampType, UInt16Type, UInt8Type,
 };
 use datatypes::value::{OrderedF32, OrderedF64, Value};
 use datatypes::vectors::{
-    BinaryVector, BooleanVector, DateTimeVector, DateVector, Float32Vector, Float64Vector,
-    Int32Vector, Int64Vector, IntervalDayTimeVector, IntervalMonthDayNanoVector,
+    BinaryVector, BooleanVector, DateTimeVector, DateVector, DurationMicrosecondVector,
+    DurationMillisecondVector, DurationNanosecondVector, DurationSecondVector, Float32Vector,
+    Float64Vector, Int32Vector, Int64Vector, IntervalDayTimeVector, IntervalMonthDayNanoVector,
     IntervalYearMonthVector, PrimitiveVector, StringVector, TimeMicrosecondVector,
     TimeMillisecondVector, TimeNanosecondVector, TimeSecondVector, TimestampMicrosecondVector,
     TimestampMillisecondVector, TimestampNanosecondVector, TimestampSecondVector, UInt32Vector,
@@ -100,6 +101,14 @@ impl From<ColumnDataTypeWrapper> for ConcreteDataType {
             ColumnDataType::IntervalMonthDayNano => {
                 ConcreteDataType::interval_month_day_nano_datatype()
             }
+            ColumnDataType::DurationSecond => ConcreteDataType::duration_second_datatype(),
+            ColumnDataType::DurationMillisecond => {
+                ConcreteDataType::duration_millisecond_datatype()
+            }
+            ColumnDataType::DurationMicrosecond => {
+                ConcreteDataType::duration_microsecond_datatype()
+            }
+            ColumnDataType::DurationNanosecond => ConcreteDataType::duration_nanosecond_datatype(),
         }
     }
 }
@@ -141,6 +150,12 @@ impl TryFrom<ConcreteDataType> for ColumnDataTypeWrapper {
                 IntervalType::DayTime(_) => ColumnDataType::IntervalDayTime,
                 IntervalType::MonthDayNano(_) => ColumnDataType::IntervalMonthDayNano,
             },
+            ConcreteDataType::Duration(d) => match d {
+                DurationType::Second(_) => ColumnDataType::DurationSecond,
+                DurationType::Millisecond(_) => ColumnDataType::DurationMillisecond,
+                DurationType::Microsecond(_) => ColumnDataType::DurationMicrosecond,
+                DurationType::Nanosecond(_) => ColumnDataType::DurationNanosecond,
+            },
             ConcreteDataType::Null(_)
             | ConcreteDataType::List(_)
             | ConcreteDataType::Dictionary(_) => {
@@ -257,6 +272,22 @@ pub fn values_with_capacity(datatype: ColumnDataType, capacity: usize) -> Values
             interval_month_day_nano_values: Vec::with_capacity(capacity),
             ..Default::default()
         },
+        ColumnDataType::DurationSecond => Values {
+            duration_second_values: Vec::with_capacity(capacity),
+            ..Default::default()
+        },
+        ColumnDataType::DurationMillisecond => Values {
+            duration_millisecond_values: Vec::with_capacity(capacity),
+            ..Default::default()
+        },
+        ColumnDataType::DurationMicrosecond => Values {
+            duration_microsecond_values: Vec::with_capacity(capacity),
+            ..Default::default()
+        },
+        ColumnDataType::DurationNanosecond => Values {
+            duration_nanosecond_values: Vec::with_capacity(capacity),
+            ..Default::default()
+        },
     }
 }
 
@@ -304,6 +335,12 @@ pub fn push_vals(column: &mut Column, origin_count: usize, vector: VectorRef) {
                 .interval_month_day_nano_values
                 .push(convert_i128_to_interval(val.to_i128())),
         },
+        Value::Duration(val) => match val.unit() {
+            TimeUnit::Second => values.duration_second_values.push(val.value()),
+            TimeUnit::Millisecond => values.duration_millisecond_values.push(val.value()),
+            TimeUnit::Microsecond => values.duration_microsecond_values.push(val.value()),
+            TimeUnit::Nanosecond => values.duration_nanosecond_values.push(val.value()),
+        },
         Value::List(_) => unreachable!(),
     });
     column.null_mask = null_mask.into_vec();
@@ -395,6 +432,10 @@ pub fn pb_value_to_value_ref(value: &v1::Value) -> ValueRef {
             let interval = Interval::from_month_day_nano(v.months, v.days, v.nanoseconds);
             ValueRef::Interval(interval)
         }
+        ValueData::DurationSecondValue(v) => ValueRef::Duration(Duration::new_second(*v)),
+        ValueData::DurationMillisecondValue(v) => ValueRef::Duration(Duration::new_millisecond(*v)),
+        ValueData::DurationMicrosecondValue(v) => ValueRef::Duration(Duration::new_microsecond(*v)),
+        ValueData::DurationNanosecondValue(v) => ValueRef::Duration(Duration::new_nanosecond(*v)),
     }
 }
 
@@ -467,6 +508,20 @@ pub fn pb_values_to_vector_ref(data_type: &ConcreteDataType, values: Values) ->
                 ))
             }
         },
+        ConcreteDataType::Duration(unit) => match unit {
+            DurationType::Second(_) => Arc::new(DurationSecondVector::from_vec(
+                values.duration_second_values,
+            )),
+            DurationType::Millisecond(_) => Arc::new(DurationMillisecondVector::from_vec(
+                values.duration_millisecond_values,
+            )),
+            DurationType::Microsecond(_) => Arc::new(DurationMicrosecondVector::from_vec(
+                values.duration_microsecond_values,
+            )),
+            DurationType::Nanosecond(_) => Arc::new(DurationNanosecondVector::from_vec(
+                values.duration_nanosecond_values,
+            )),
+        },
         ConcreteDataType::Null(_) | ConcreteDataType::List(_) | ConcreteDataType::Dictionary(_) => {
             unreachable!()
         }
@@ -617,6 +672,26 @@ pub fn pb_values_to_values(data_type: &ConcreteDataType, values: Values) -> Vec<
                 ))
             })
             .collect(),
+        ConcreteDataType::Duration(DurationType::Second(_)) => values
+            .duration_second_values
+            .into_iter()
+            .map(|v| Value::Duration(Duration::new_second(v)))
+            .collect(),
+        ConcreteDataType::Duration(DurationType::Millisecond(_)) => values
+            .duration_millisecond_values
+            .into_iter()
+            .map(|v| Value::Duration(Duration::new_millisecond(v)))
+            .collect(),
+        ConcreteDataType::Duration(DurationType::Microsecond(_)) => values
+            .duration_microsecond_values
+            .into_iter()
+            .map(|v| Value::Duration(Duration::new_microsecond(v)))
+            .collect(),
+        ConcreteDataType::Duration(DurationType::Nanosecond(_)) => values
+            .duration_nanosecond_values
+            .into_iter()
+            .map(|v| Value::Duration(Duration::new_nanosecond(v)))
+            .collect(),
         ConcreteDataType::Null(_) | ConcreteDataType::List(_) | ConcreteDataType::Dictionary(_) => {
             unreachable!()
         }
@@ -727,6 +802,20 @@ pub fn to_proto_value(value: Value) -> Option<v1::Value> {
                 )),
             },
         },
+        Value::Duration(v) => match v.unit() {
+            TimeUnit::Second => v1::Value {
+                value_data: Some(ValueData::DurationSecondValue(v.value())),
+            },
+            TimeUnit::Millisecond => v1::Value {
+                value_data: Some(ValueData::DurationMillisecondValue(v.value())),
+            },
+            TimeUnit::Microsecond => v1::Value {
+                value_data: Some(ValueData::DurationMicrosecondValue(v.value())),
+            },
+            TimeUnit::Nanosecond => v1::Value {
+                value_data: Some(ValueData::DurationNanosecondValue(v.value())),
+            },
+        },
         Value::List(_) => return None,
     };
 
@@ -764,6 +853,10 @@ pub fn proto_value_type(value: &v1::Value) -> Option<ColumnDataType> {
         ValueData::IntervalYearMonthValues(_) => ColumnDataType::IntervalYearMonth,
         ValueData::IntervalDayTimeValues(_) => ColumnDataType::IntervalDayTime,
         ValueData::IntervalMonthDayNanoValues(_) => ColumnDataType::IntervalMonthDayNano,
+        ValueData::DurationSecondValue(_) => ColumnDataType::DurationSecond,
+        ValueData::DurationMillisecondValue(_) => ColumnDataType::DurationMillisecond,
+        ValueData::DurationMicrosecondValue(_) => ColumnDataType::DurationMicrosecond,
+        ValueData::DurationNanosecondValue(_) => ColumnDataType::DurationNanosecond,
     };
     Some(value_type)
 }
@@ -801,6 +894,7 @@ pub fn to_column_data_type(data_type: &ConcreteDataType) -> Option<ColumnDataTyp
         ConcreteDataType::Time(TimeType::Microsecond(_)) => ColumnDataType::TimeMicrosecond,
         ConcreteDataType::Time(TimeType::Nanosecond(_)) => ColumnDataType::TimeNanosecond,
         ConcreteDataType::Null(_)
+        | ConcreteDataType::Duration(_)
         | ConcreteDataType::Interval(_)
         | ConcreteDataType::List(_)
         | ConcreteDataType::Dictionary(_) => return None,
@@ -861,6 +955,12 @@ pub fn value_to_grpc_value(value: Value) -> GrpcValue {
                     ValueData::IntervalMonthDayNanoValues(convert_i128_to_interval(v.to_i128()))
                 }
             }),
+            Value::Duration(v) => Some(match v.unit() {
+                TimeUnit::Second => ValueData::DurationSecondValue(v.value()),
+                TimeUnit::Millisecond => ValueData::DurationMillisecondValue(v.value()),
+                TimeUnit::Microsecond => ValueData::DurationMicrosecondValue(v.value()),
+                TimeUnit::Nanosecond => ValueData::DurationNanosecondValue(v.value()),
+            }),
             Value::List(_) => unreachable!(),
         },
     }
@@ -880,15 +980,16 @@ mod tests {
     use std::sync::Arc;
 
     use datatypes::types::{
-        Int32Type, IntervalDayTimeType, IntervalMonthDayNanoType, IntervalYearMonthType,
-        TimeMillisecondType, TimeSecondType, TimestampMillisecondType, TimestampSecondType,
-        UInt32Type,
+        DurationMillisecondType, DurationSecondType, Int32Type, IntervalDayTimeType,
+        IntervalMonthDayNanoType, IntervalYearMonthType, TimeMillisecondType, TimeSecondType,
+        TimestampMillisecondType, TimestampSecondType, UInt32Type,
     };
     use datatypes::vectors::{
-        BooleanVector, IntervalDayTimeVector, IntervalMonthDayNanoVector, IntervalYearMonthVector,
-        TimeMicrosecondVector, TimeMillisecondVector, TimeNanosecondVector, TimeSecondVector,
-        TimestampMicrosecondVector, TimestampMillisecondVector, TimestampNanosecondVector,
-        TimestampSecondVector, Vector,
+        BooleanVector, DurationMicrosecondVector, DurationMillisecondVector,
+        DurationNanosecondVector, DurationSecondVector, IntervalDayTimeVector,
+        IntervalMonthDayNanoVector, IntervalYearMonthVector, TimeMicrosecondVector,
+        TimeMillisecondVector, TimeNanosecondVector, TimeSecondVector, TimestampMicrosecondVector,
+        TimestampMillisecondVector, TimestampNanosecondVector, TimestampSecondVector, Vector,
     };
     use paste::paste;
 
@@ -963,6 +1064,10 @@ mod tests {
         let values = values_with_capacity(ColumnDataType::IntervalMonthDayNano, 2);
         let values = values.interval_month_day_nano_values;
         assert_eq!(2, values.capacity());
+
+        let values = values_with_capacity(ColumnDataType::DurationMillisecond, 2);
+        let values = values.duration_millisecond_values;
+        assert_eq!(2, values.capacity());
     }
 
     #[test]
@@ -1047,6 +1152,10 @@ mod tests {
             ConcreteDataType::interval_datatype(IntervalUnit::MonthDayNano),
             ColumnDataTypeWrapper(ColumnDataType::IntervalMonthDayNano).into()
         );
+        assert_eq!(
+            ConcreteDataType::duration_millisecond_datatype(),
+            ColumnDataTypeWrapper(ColumnDataType::DurationMillisecond).into()
+        )
     }
 
     #[test]
@@ -1135,6 +1244,12 @@ mod tests {
                 .try_into()
                 .unwrap()
         );
+        assert_eq!(
+            ColumnDataTypeWrapper(ColumnDataType::DurationMillisecond),
+            ConcreteDataType::duration_millisecond_datatype()
+                .try_into()
+                .unwrap()
+        );
 
         let result: Result<ColumnDataTypeWrapper> = ConcreteDataType::null_datatype().try_into();
         assert!(result.is_err());
@@ -1278,6 +1393,47 @@ mod tests {
         });
     }
 
+    #[test]
+    fn test_column_put_duration_values() {
+        let mut column = Column {
+            column_name: "test".to_string(),
+            semantic_type: 0,
+            values: Some(Values {
+                ..Default::default()
+            }),
+            null_mask: vec![],
+            datatype: 0,
+        };
+
+        let vector = Arc::new(DurationNanosecondVector::from_vec(vec![1, 2, 3]));
+        push_vals(&mut column, 3, vector);
+        assert_eq!(
+            vec![1, 2, 3],
+            column.values.as_ref().unwrap().duration_nanosecond_values
+        );
+
+        let vector = Arc::new(DurationMicrosecondVector::from_vec(vec![7, 8, 9]));
+        push_vals(&mut column, 3, vector);
+        assert_eq!(
+            vec![7, 8, 9],
+            column.values.as_ref().unwrap().duration_microsecond_values
+        );
+
+        let vector = Arc::new(DurationMillisecondVector::from_vec(vec![4, 5, 6]));
+        push_vals(&mut column, 3, vector);
+        assert_eq!(
+            vec![4, 5, 6],
+            column.values.as_ref().unwrap().duration_millisecond_values
+        );
+
+        let vector = Arc::new(DurationSecondVector::from_vec(vec![10, 11, 12]));
+        push_vals(&mut column, 3, vector);
+        assert_eq!(
+            vec![10, 11, 12],
+            column.values.as_ref().unwrap().duration_second_values
+        );
+    }
+
     #[test]
     fn test_column_put_vector() {
         use crate::v1::SemanticType;
@@ -1378,6 +1534,39 @@ mod tests {
         assert_eq!(expect, actual);
     }
 
+    #[test]
+    fn test_convert_duration_values() {
+        // second
+        let actual = pb_values_to_values(
+            &ConcreteDataType::Duration(DurationType::Second(DurationSecondType)),
+            Values {
+                duration_second_values: vec![1_i64, 2_i64, 3_i64],
+                ..Default::default()
+            },
+        );
+        let expect = vec![
+            Value::Duration(Duration::new_second(1_i64)),
+            Value::Duration(Duration::new_second(2_i64)),
+            Value::Duration(Duration::new_second(3_i64)),
+        ];
+        assert_eq!(expect, actual);
+
+        // millisecond
+        let actual = pb_values_to_values(
+            &ConcreteDataType::Duration(DurationType::Millisecond(DurationMillisecondType)),
+            Values {
+                duration_millisecond_values: vec![1_i64, 2_i64, 3_i64],
+                ..Default::default()
+            },
+        );
+        let expect = vec![
+            Value::Duration(Duration::new_millisecond(1_i64)),
+            Value::Duration(Duration::new_millisecond(2_i64)),
+            Value::Duration(Duration::new_millisecond(3_i64)),
+        ];
+        assert_eq!(expect, actual);
+    }
+
     #[test]
     fn test_convert_interval_values() {
         // year_month

src/common/grpc-expr/src/insert.rs

@@ -159,8 +159,8 @@ mod tests {
         );
 
         let column_defs = create_expr.column_defs;
-        assert_eq!(column_defs[5].name, create_expr.time_index);
-        assert_eq!(6, column_defs.len());
+        assert_eq!(column_defs[6].name, create_expr.time_index);
+        assert_eq!(7, column_defs.len());
 
         assert_eq!(
             ConcreteDataType::string_datatype(),
@@ -232,6 +232,20 @@ mod tests {
             )
         );
 
+        assert_eq!(
+            ConcreteDataType::duration_millisecond_datatype(),
+            ConcreteDataType::from(
+                ColumnDataTypeWrapper::try_new(
+                    column_defs
+                        .iter()
+                        .find(|c| c.name == "duration")
+                        .unwrap()
+                        .data_type
+                )
+                .unwrap()
+            )
+        );
+
         assert_eq!(
             ConcreteDataType::timestamp_millisecond_datatype(),
             ConcreteDataType::from(
@@ -265,7 +279,7 @@ mod tests {
 
         let add_columns = find_new_columns(&schema, &insert_batch.0).unwrap().unwrap();
 
-        assert_eq!(4, add_columns.add_columns.len());
+        assert_eq!(5, add_columns.add_columns.len());
         let host_column = &add_columns.add_columns[0];
         assert_eq!(
             ConcreteDataType::string_datatype(),
@@ -305,6 +319,18 @@ mod tests {
                 .unwrap()
             )
         );
+
+        let duration_column = &add_columns.add_columns[4];
+
+        assert_eq!(
+            ConcreteDataType::duration_millisecond_datatype(),
+            ConcreteDataType::from(
+                ColumnDataTypeWrapper::try_new(
+                    duration_column.column_def.as_ref().unwrap().data_type
+                )
+                .unwrap()
+            )
+        );
     }
 
     #[test]
@@ -394,6 +420,18 @@ mod tests {
             datatype: ColumnDataType::IntervalMonthDayNano as i32,
         };
 
+        let duration_vals = Values {
+            duration_millisecond_values: vec![100, 101],
+            ..Default::default()
+        };
+        let duration_column = Column {
+            column_name: "duration".to_string(),
+            semantic_type: SemanticType::Field as i32,
+            values: Some(duration_vals),
+            null_mask: vec![0],
+            datatype: ColumnDataType::DurationMillisecond as i32,
+        };
+
         let ts_vals = Values {
             timestamp_millisecond_values: vec![100, 101],
             ..Default::default()
@@ -413,6 +451,7 @@ mod tests {
                 mem_column,
                 time_column,
                 interval_column,
+                duration_column,
                 ts_column,
             ],
             row_count,

src/common/grpc/src/select.rs

@@ -15,10 +15,11 @@
 use api::helper::convert_i128_to_interval;
 use api::v1::column::Values;
 use common_base::BitVec;
-use datatypes::types::{IntervalType, TimeType, TimestampType, WrapperType};
+use datatypes::types::{DurationType, IntervalType, TimeType, TimestampType, WrapperType};
 use datatypes::vectors::{
-    BinaryVector, BooleanVector, DateTimeVector, DateVector, Float32Vector, Float64Vector,
-    Int16Vector, Int32Vector, Int64Vector, Int8Vector, IntervalDayTimeVector,
+    BinaryVector, BooleanVector, DateTimeVector, DateVector, DurationMicrosecondVector,
+    DurationMillisecondVector, DurationNanosecondVector, DurationSecondVector, Float32Vector,
+    Float64Vector, Int16Vector, Int32Vector, Int64Vector, Int8Vector, IntervalDayTimeVector,
     IntervalMonthDayNanoVector, IntervalYearMonthVector, StringVector, TimeMicrosecondVector,
     TimeMillisecondVector, TimeNanosecondVector, TimeSecondVector, TimestampMicrosecondVector,
     TimestampMillisecondVector, TimestampNanosecondVector, TimestampSecondVector, UInt16Vector,
@@ -212,6 +213,30 @@ pub fn values(arrays: &[VectorRef]) -> Result<Values> {
             IntervalMonthDayNanoVector,
             interval_month_day_nano_values,
             |x| { convert_i128_to_interval(x.into_native()) }
+        ),
+        (
+            ConcreteDataType::Duration(DurationType::Second(_)),
+            DurationSecondVector,
+            duration_second_values,
+            |x| { x.into_native() }
+        ),
+        (
+            ConcreteDataType::Duration(DurationType::Millisecond(_)),
+            DurationMillisecondVector,
+            duration_millisecond_values,
+            |x| { x.into_native() }
+        ),
+        (
+            ConcreteDataType::Duration(DurationType::Microsecond(_)),
+            DurationMicrosecondVector,
+            duration_microsecond_values,
+            |x| { x.into_native() }
+        ),
+        (
+            ConcreteDataType::Duration(DurationType::Nanosecond(_)),
+            DurationNanosecondVector,
+            duration_nanosecond_values,
+            |x| { x.into_native() }
         )
     )
 }
@@ -279,6 +304,16 @@ mod tests {
         })
     }
 
+    #[test]
+    fn test_convert_arrow_array_duration_second() {
+        let array = DurationSecondVector::from(vec![Some(1), Some(2), None, Some(3)]);
+        let array: VectorRef = Arc::new(array);
+
+        let values = values(&[array]).unwrap();
+
+        assert_eq!(vec![1, 2, 3], values.duration_second_values);
+    }
+
     #[test]
     fn test_convert_arrow_arrays_string() {
         let array = StringVector::from(vec![

src/common/time/src/duration.rs

@@ -87,6 +87,11 @@ impl Duration {
         let nsec = u32::try_from(sec_mod * nsec_mul).unwrap();
         (sec_div, nsec)
     }
+
+    /// Convert to std::time::Duration.
+    pub fn to_std_duration(self) -> std::time::Duration {
+        self.into()
+    }
 }
 
 /// Convert i64 to Duration Type.

src/datatypes/src/data_type.rs

@@ -27,11 +27,13 @@ use serde::{Deserialize, Serialize};
 use crate::error::{self, Error, Result};
 use crate::type_id::LogicalTypeId;
 use crate::types::{
-    BinaryType, BooleanType, DateTimeType, DateType, DictionaryType, Float32Type, Float64Type,
-    Int16Type, Int32Type, Int64Type, Int8Type, IntervalDayTimeType, IntervalMonthDayNanoType,
-    IntervalType, IntervalYearMonthType, ListType, NullType, StringType, TimeMillisecondType,
-    TimeType, TimestampMicrosecondType, TimestampMillisecondType, TimestampNanosecondType,
-    TimestampSecondType, TimestampType, UInt16Type, UInt32Type, UInt64Type, UInt8Type,
+    BinaryType, BooleanType, DateTimeType, DateType, DictionaryType, DurationMicrosecondType,
+    DurationMillisecondType, DurationNanosecondType, DurationSecondType, DurationType, Float32Type,
+    Float64Type, Int16Type, Int32Type, Int64Type, Int8Type, IntervalDayTimeType,
+    IntervalMonthDayNanoType, IntervalType, IntervalYearMonthType, ListType, NullType, StringType,
+    TimeMillisecondType, TimeType, TimestampMicrosecondType, TimestampMillisecondType,
+    TimestampNanosecondType, TimestampSecondType, TimestampType, UInt16Type, UInt32Type,
+    UInt64Type, UInt8Type,
 };
 use crate::value::Value;
 use crate::vectors::MutableVector;
@@ -64,7 +66,10 @@ pub enum ConcreteDataType {
     Timestamp(TimestampType),
     Time(TimeType),
 
-    // Interval types:
+    // Duration type:
+    Duration(DurationType),
+
+    // Interval type:
     Interval(IntervalType),
 
     // Compound types:
@@ -96,6 +101,7 @@ impl fmt::Display for ConcreteDataType {
             ConcreteDataType::List(_) => write!(f, "List"),
             ConcreteDataType::Dictionary(_) => write!(f, "Dictionary"),
             ConcreteDataType::Interval(_) => write!(f, "Interval"),
+            ConcreteDataType::Duration(_) => write!(f, "Duration"),
         }
     }
 }
@@ -123,6 +129,7 @@ impl ConcreteDataType {
                 | ConcreteDataType::Timestamp(_)
                 | ConcreteDataType::Time(_)
                 | ConcreteDataType::Interval(_)
+                | ConcreteDataType::Duration(_)
         )
     }
 
@@ -138,6 +145,7 @@ impl ConcreteDataType {
                 | ConcreteDataType::Timestamp(_)
                 | ConcreteDataType::Time(_)
                 | ConcreteDataType::Interval(_)
+                | ConcreteDataType::Duration(_)
         )
     }
 
@@ -211,6 +219,14 @@ impl ConcreteDataType {
         let array = arrow_array::new_empty_array(&self.as_arrow_type());
         arrow_array_cast(array.as_ref(), &to_type.as_arrow_type()).is_ok()
     }
+
+    /// Try to cast data type as a [`DurationType`].
+    pub fn as_duration(&self) -> Option<DurationType> {
+        match self {
+            ConcreteDataType::Duration(d) => Some(*d),
+            _ => None,
+        }
+    }
 }
 
 impl From<&ConcreteDataType> for ConcreteDataType {
@@ -252,6 +268,9 @@ impl TryFrom<&ArrowDataType> for ConcreteDataType {
             }
             ArrowDataType::Time32(u) => ConcreteDataType::Time(TimeType::from_unit(u.into())),
             ArrowDataType::Time64(u) => ConcreteDataType::Time(TimeType::from_unit(u.into())),
+            ArrowDataType::Duration(u) => {
+                ConcreteDataType::Duration(DurationType::from_unit(u.into()))
+            }
             _ => {
                 return error::UnsupportedArrowTypeSnafu {
                     arrow_type: dt.clone(),
@@ -325,14 +344,37 @@ impl ConcreteDataType {
         Self::time_datatype(TimeUnit::Nanosecond)
     }
 
+    /// Creates a [Duration(DurationSecondType)] datatype.
+    pub fn duration_second_datatype() -> Self {
+        ConcreteDataType::Duration(DurationType::Second(DurationSecondType))
+    }
+
+    /// Creates a [Duration(DurationMillisecondType)] datatype.
+    pub fn duration_millisecond_datatype() -> Self {
+        ConcreteDataType::Duration(DurationType::Millisecond(DurationMillisecondType))
+    }
+
+    /// Creates a [Duration(DurationMicrosecondType)] datatype.
+    pub fn duration_microsecond_datatype() -> Self {
+        ConcreteDataType::Duration(DurationType::Microsecond(DurationMicrosecondType))
+    }
+
+    /// Creates a [Duration(DurationNanosecondType)] datatype.
+    pub fn duration_nanosecond_datatype() -> Self {
+        ConcreteDataType::Duration(DurationType::Nanosecond(DurationNanosecondType))
+    }
+
+    /// Creates a [Interval(IntervalMonthDayNanoType)] datatype.
     pub fn interval_month_day_nano_datatype() -> Self {
         ConcreteDataType::Interval(IntervalType::MonthDayNano(IntervalMonthDayNanoType))
     }
 
+    /// Creates a [Interval(IntervalYearMonthType)] datatype.
     pub fn interval_year_month_datatype() -> Self {
         ConcreteDataType::Interval(IntervalType::YearMonth(IntervalYearMonthType))
     }
 
+    /// Creates a [Interval(IntervalDayTimeType)] datatype.
     pub fn interval_day_time_datatype() -> Self {
         ConcreteDataType::Interval(IntervalType::DayTime(IntervalDayTimeType))
     }
@@ -356,6 +398,15 @@ impl ConcreteDataType {
         }
     }
 
+    pub fn duration_datatype(unit: TimeUnit) -> Self {
+        match unit {
+            TimeUnit::Second => Self::duration_second_datatype(),
+            TimeUnit::Millisecond => Self::duration_millisecond_datatype(),
+            TimeUnit::Microsecond => Self::duration_microsecond_datatype(),
+            TimeUnit::Nanosecond => Self::duration_nanosecond_datatype(),
+        }
+    }
+
     pub fn interval_datatype(unit: IntervalUnit) -> Self {
         match unit {
             IntervalUnit::YearMonth => Self::interval_year_month_datatype(),
@@ -550,6 +601,10 @@ mod tests {
         assert!(!ConcreteDataType::time_millisecond_datatype().is_timestamp_compatible());
         assert!(!ConcreteDataType::time_microsecond_datatype().is_timestamp_compatible());
         assert!(!ConcreteDataType::time_nanosecond_datatype().is_timestamp_compatible());
+        assert!(!ConcreteDataType::duration_second_datatype().is_timestamp_compatible());
+        assert!(!ConcreteDataType::duration_millisecond_datatype().is_timestamp_compatible());
+        assert!(!ConcreteDataType::duration_microsecond_datatype().is_timestamp_compatible());
+        assert!(!ConcreteDataType::duration_nanosecond_datatype().is_timestamp_compatible());
     }
 
     #[test]
@@ -591,6 +646,11 @@ mod tests {
         assert!(ConcreteDataType::interval_year_month_datatype().is_stringifiable());
         assert!(ConcreteDataType::interval_day_time_datatype().is_stringifiable());
         assert!(ConcreteDataType::interval_month_day_nano_datatype().is_stringifiable());
+
+        assert!(ConcreteDataType::duration_second_datatype().is_stringifiable());
+        assert!(ConcreteDataType::duration_millisecond_datatype().is_stringifiable());
+        assert!(ConcreteDataType::duration_microsecond_datatype().is_stringifiable());
+        assert!(ConcreteDataType::duration_nanosecond_datatype().is_stringifiable());
     }
 
     #[test]
@@ -612,6 +672,10 @@ mod tests {
         assert!(ConcreteDataType::interval_year_month_datatype().is_signed());
         assert!(ConcreteDataType::interval_day_time_datatype().is_signed());
         assert!(ConcreteDataType::interval_month_day_nano_datatype().is_signed());
+        assert!(ConcreteDataType::duration_second_datatype().is_signed());
+        assert!(ConcreteDataType::duration_millisecond_datatype().is_signed());
+        assert!(ConcreteDataType::duration_microsecond_datatype().is_signed());
+        assert!(ConcreteDataType::duration_nanosecond_datatype().is_signed());
 
         assert!(!ConcreteDataType::uint8_datatype().is_signed());
         assert!(!ConcreteDataType::uint16_datatype().is_signed());
@@ -641,6 +705,10 @@ mod tests {
         assert!(!ConcreteDataType::interval_year_month_datatype().is_unsigned());
         assert!(!ConcreteDataType::interval_day_time_datatype().is_unsigned());
         assert!(!ConcreteDataType::interval_month_day_nano_datatype().is_unsigned());
+        assert!(!ConcreteDataType::duration_second_datatype().is_unsigned());
+        assert!(!ConcreteDataType::duration_millisecond_datatype().is_unsigned());
+        assert!(!ConcreteDataType::duration_microsecond_datatype().is_unsigned());
+        assert!(!ConcreteDataType::duration_nanosecond_datatype().is_unsigned());
 
         assert!(ConcreteDataType::uint8_datatype().is_unsigned());
         assert!(ConcreteDataType::uint16_datatype().is_unsigned());
@@ -749,6 +817,10 @@ mod tests {
             ))
             .to_string(),
             "Interval"
-        )
+        );
+        assert_eq!(
+            ConcreteDataType::duration_second_datatype().to_string(),
+            "Duration"
+        );
     }
 }

src/datatypes/src/duration.rs

@@ -0,0 +1,161 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use common_time::duration::Duration;
+use common_time::timestamp::TimeUnit;
+use paste::paste;
+use serde::{Deserialize, Serialize};
+
+use crate::prelude::{Scalar, Value, ValueRef};
+use crate::scalars::ScalarRef;
+use crate::types::{
+    DurationMicrosecondType, DurationMillisecondType, DurationNanosecondType, DurationSecondType,
+    WrapperType,
+};
+use crate::vectors::{
+    DurationMicrosecondVector, DurationMillisecondVector, DurationNanosecondVector,
+    DurationSecondVector,
+};
+
+macro_rules! define_duration_with_unit {
+    ($unit: ident) => {
+        paste! {
+            #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+            pub struct [<Duration $unit>](pub Duration);
+
+            impl [<Duration $unit>] {
+                pub fn new(val: i64) -> Self {
+                    Self(Duration::new(val, TimeUnit::$unit))
+                }
+            }
+
+            impl Default for [<Duration $unit>] {
+                fn default() -> Self {
+                    Self::new(0)
+                }
+            }
+
+            impl From<[<Duration $unit>]> for Value {
+                fn from(t: [<Duration $unit>]) -> Value {
+                    Value::Duration(t.0)
+                }
+            }
+
+            impl From<[<Duration $unit>]> for serde_json::Value {
+                fn from(t: [<Duration $unit>]) -> Self {
+                    t.0.into()
+                }
+            }
+
+            impl From<[<Duration $unit>]> for ValueRef<'static> {
+                fn from(t: [<Duration $unit>]) -> Self {
+                    ValueRef::Duration(t.0)
+                }
+            }
+
+            impl Scalar for [<Duration $unit>] {
+                type VectorType = [<Duration $unit Vector>];
+                type RefType<'a> = [<Duration $unit>];
+
+                fn as_scalar_ref(&self) -> Self::RefType<'_> {
+                    *self
+                }
+
+                fn upcast_gat<'short, 'long: 'short>(
+                    long: Self::RefType<'long>,
+                ) -> Self::RefType<'short> {
+                    long
+                }
+            }
+
+            impl<'a> ScalarRef<'a> for [<Duration $unit>] {
+                type ScalarType = [<Duration $unit>];
+
+                fn to_owned_scalar(&self) -> Self::ScalarType {
+                    *self
+                }
+            }
+
+            impl WrapperType for [<Duration $unit>] {
+                type LogicalType = [<Duration $unit Type>];
+                type Native = i64;
+
+                fn from_native(value: Self::Native) -> Self {
+                    Self::new(value)
+                }
+
+                fn into_native(self) -> Self::Native {
+                    self.0.into()
+                }
+            }
+
+            impl From<i64> for [<Duration $unit>] {
+                fn from(val: i64) -> Self {
+                    [<Duration $unit>]::from_native(val)
+                }
+            }
+
+            impl From<[<Duration $unit>]> for i64{
+                fn from(val: [<Duration $unit>]) -> Self {
+                    val.0.value()
+                }
+            }
+        }
+    };
+}
+
+define_duration_with_unit!(Second);
+define_duration_with_unit!(Millisecond);
+define_duration_with_unit!(Microsecond);
+define_duration_with_unit!(Nanosecond);
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_duration_scalar() {
+        let d = DurationSecond::new(456);
+        assert_eq!(d, d.as_scalar_ref());
+        assert_eq!(d, d.to_owned_scalar());
+        let d = DurationMillisecond::new(456);
+        assert_eq!(d, d.as_scalar_ref());
+        assert_eq!(d, d.to_owned_scalar());
+        let d = DurationMicrosecond::new(456);
+        assert_eq!(d, d.as_scalar_ref());
+        assert_eq!(d, d.to_owned_scalar());
+        let d = DurationNanosecond::new(456);
+        assert_eq!(d, d.as_scalar_ref());
+        assert_eq!(d, d.to_owned_scalar());
+    }
+
+    #[test]
+    fn test_duration_to_native_type() {
+        let duration = DurationSecond::new(456);
+        let native: i64 = duration.into_native();
+        assert_eq!(native, 456);
+
+        let duration = DurationMillisecond::new(456);
+        let native: i64 = duration.into_native();
+        assert_eq!(native, 456);
+
+        let duration = DurationMicrosecond::new(456);
+        let native: i64 = duration.into_native();
+        assert_eq!(native, 456);
+
+        let duration = DurationNanosecond::new(456);
+        let native: i64 = duration.into_native();
+        assert_eq!(native, 456);
+    }
+}

src/datatypes/src/lib.rs

@@ -16,6 +16,7 @@
 
 pub mod arrow_array;
 pub mod data_type;
+pub mod duration;
 pub mod error;
 pub mod interval;
 pub mod macros;

src/datatypes/src/type_id.rs

@@ -51,6 +51,12 @@ pub enum LogicalTypeId {
     TimeMillisecond,
     TimeMicrosecond,
     TimeNanosecond,
+    /// A 64-bit duration representing the elapsed time in either seconds,
+    /// milliseconds, microseconds or nanoseconds.
+    DurationSecond,
+    DurationMillisecond,
+    DurationMicrosecond,
+    DurationNanosecond,
     /// A 32-bit interval representing the elapsed time in months.
     IntervalYearMonth,
     /// A 64-bit interval representing the elapsed time in days and milliseconds.
@@ -113,6 +119,10 @@ impl LogicalTypeId {
             LogicalTypeId::IntervalMonthDayNano => {
                 ConcreteDataType::interval_month_day_nano_datatype()
             }
+            LogicalTypeId::DurationSecond => ConcreteDataType::duration_second_datatype(),
+            LogicalTypeId::DurationMillisecond => ConcreteDataType::duration_millisecond_datatype(),
+            LogicalTypeId::DurationMicrosecond => ConcreteDataType::duration_microsecond_datatype(),
+            LogicalTypeId::DurationNanosecond => ConcreteDataType::duration_nanosecond_datatype(),
         }
     }
 }

src/datatypes/src/types.rs

@@ -17,6 +17,7 @@ mod boolean_type;
 mod date_type;
 mod datetime_type;
 mod dictionary_type;
+mod duration_type;
 mod interval_type;
 mod list_type;
 mod null_type;
@@ -30,6 +31,10 @@ pub use boolean_type::BooleanType;
 pub use date_type::DateType;
 pub use datetime_type::DateTimeType;
 pub use dictionary_type::DictionaryType;
+pub use duration_type::{
+    DurationMicrosecondType, DurationMillisecondType, DurationNanosecondType, DurationSecondType,
+    DurationType,
+};
 pub use interval_type::{
     IntervalDayTimeType, IntervalMonthDayNanoType, IntervalType, IntervalYearMonthType,
 };

src/datatypes/src/types/duration_type.rs

@@ -0,0 +1,212 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use arrow::datatypes::{
+    DataType as ArrowDataType, DurationMicrosecondType as ArrowDurationMicrosecondType,
+    DurationMillisecondType as ArrowDurationMillisecondType,
+    DurationNanosecondType as ArrowDurationNanosecondType,
+    DurationSecondType as ArrowDurationSecondType, TimeUnit as ArrowTimeUnit,
+};
+use common_time::duration::Duration;
+use common_time::timestamp::TimeUnit;
+use enum_dispatch::enum_dispatch;
+use paste::paste;
+use serde::{Deserialize, Serialize};
+use snafu::OptionExt;
+
+use super::LogicalPrimitiveType;
+use crate::data_type::DataType;
+use crate::duration::{
+    DurationMicrosecond, DurationMillisecond, DurationNanosecond, DurationSecond,
+};
+use crate::error;
+use crate::prelude::{
+    ConcreteDataType, LogicalTypeId, MutableVector, ScalarVectorBuilder, Value, ValueRef, Vector,
+};
+use crate::vectors::{
+    DurationMicrosecondVector, DurationMicrosecondVectorBuilder, DurationMillisecondVector,
+    DurationMillisecondVectorBuilder, DurationNanosecondVector, DurationNanosecondVectorBuilder,
+    DurationSecondVector, DurationSecondVectorBuilder, PrimitiveVector,
+};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
+#[enum_dispatch(DataType)]
+pub enum DurationType {
+    Second(DurationSecondType),
+    Millisecond(DurationMillisecondType),
+    Microsecond(DurationMicrosecondType),
+    Nanosecond(DurationNanosecondType),
+}
+
+impl DurationType {
+    /// Creates time type from `TimeUnit`.
+    pub fn from_unit(unit: TimeUnit) -> Self {
+        match unit {
+            TimeUnit::Second => Self::Second(DurationSecondType),
+            TimeUnit::Millisecond => Self::Millisecond(DurationMillisecondType),
+            TimeUnit::Microsecond => Self::Microsecond(DurationMicrosecondType),
+            TimeUnit::Nanosecond => Self::Nanosecond(DurationNanosecondType),
+        }
+    }
+
+    /// Returns the [`TimeUnit`] of this type.
+    pub fn unit(&self) -> TimeUnit {
+        match self {
+            DurationType::Second(_) => TimeUnit::Second,
+            DurationType::Millisecond(_) => TimeUnit::Millisecond,
+            DurationType::Microsecond(_) => TimeUnit::Microsecond,
+            DurationType::Nanosecond(_) => TimeUnit::Nanosecond,
+        }
+    }
+}
+
+macro_rules! impl_data_type_for_duration {
+    ($unit: ident) => {
+        paste! {
+            #[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
+            pub struct [<Duration $unit Type>];
+
+            impl DataType for [<Duration $unit Type>] {
+                fn name(&self) -> &str {
+                    stringify!([<Duration $unit>])
+                }
+
+                fn logical_type_id(&self) -> LogicalTypeId {
+                    LogicalTypeId::[<Duration $unit>]
+                }
+
+                fn default_value(&self) -> Value {
+                    Value::Duration(Duration::new(0, TimeUnit::$unit))
+                }
+
+                fn as_arrow_type(&self) -> ArrowDataType {
+                    ArrowDataType::Duration(ArrowTimeUnit::$unit)
+                }
+
+                fn create_mutable_vector(&self, capacity: usize) -> Box<dyn MutableVector> {
+                    Box::new([<Duration $unit Vector Builder>]::with_capacity(capacity))
+                }
+
+                fn is_timestamp_compatible(&self) -> bool {
+                    false
+                }
+            }
+
+            impl LogicalPrimitiveType for [<Duration $unit Type>] {
+                type ArrowPrimitive = [<Arrow Duration $unit Type>];
+                type Native = i64;
+                type Wrapper = [<Duration $unit>];
+                type LargestType = Self;
+
+                fn build_data_type() -> ConcreteDataType {
+                    ConcreteDataType::Duration(DurationType::$unit(
+                        [<Duration $unit Type>]::default(),
+                    ))
+                }
+
+                fn type_name() -> &'static str {
+                    stringify!([<Duration $unit Type>])
+                }
+
+                fn cast_vector(vector: &dyn Vector) -> crate::Result<&PrimitiveVector<Self>> {
+                    vector
+                        .as_any()
+                        .downcast_ref::<[<Duration $unit Vector>]>()
+                        .with_context(|| error::CastTypeSnafu {
+                            msg: format!(
+                                "Failed to cast {} to {}",
+                                vector.vector_type_name(), stringify!([<Duration $unit Vector>])
+                            ),
+                        })
+                }
+
+                fn cast_value_ref(value: ValueRef) -> crate::Result<Option<Self::Wrapper>> {
+                    match value {
+                        ValueRef::Null => Ok(None),
+                        ValueRef::Duration(t) => match t.unit() {
+                            TimeUnit::$unit => Ok(Some([<Duration $unit>](t))),
+                            other => error::CastTypeSnafu {
+                                msg: format!(
+                                    "Failed to cast Duration value with different unit {:?} to {}",
+                                    other, stringify!([<Duration $unit>])
+                                ),
+                            }
+                            .fail(),
+                        },
+                        other => error::CastTypeSnafu {
+                            msg: format!("Failed to cast value {:?} to {}", other, stringify!([<Duration $unit>])),
+                        }
+                        .fail(),
+                    }
+                }
+            }
+        }
+    }
+}
+
+impl_data_type_for_duration!(Second);
+impl_data_type_for_duration!(Millisecond);
+impl_data_type_for_duration!(Microsecond);
+impl_data_type_for_duration!(Nanosecond);
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_duration_type_unit() {
+        assert_eq!(
+            TimeUnit::Second,
+            DurationType::Second(DurationSecondType).unit()
+        );
+
+        assert_eq!(
+            TimeUnit::Millisecond,
+            DurationType::Millisecond(DurationMillisecondType).unit()
+        );
+
+        assert_eq!(
+            TimeUnit::Microsecond,
+            DurationType::Microsecond(DurationMicrosecondType).unit()
+        );
+
+        assert_eq!(
+            TimeUnit::Nanosecond,
+            DurationType::Nanosecond(DurationNanosecondType).unit()
+        );
+    }
+
+    #[test]
+    fn test_from_unit() {
+        assert_eq!(
+            DurationType::Second(DurationSecondType),
+            DurationType::from_unit(TimeUnit::Second)
+        );
+
+        assert_eq!(
+            DurationType::Millisecond(DurationMillisecondType),
+            DurationType::from_unit(TimeUnit::Millisecond)
+        );
+
+        assert_eq!(
+            DurationType::Microsecond(DurationMicrosecondType),
+            DurationType::from_unit(TimeUnit::Microsecond)
+        );
+
+        assert_eq!(
+            DurationType::Nanosecond(DurationNanosecondType),
+            DurationType::from_unit(TimeUnit::Nanosecond)
+        );
+    }
+}

src/datatypes/src/value.rs

@@ -25,7 +25,7 @@ use common_time::datetime::DateTime;
 use common_time::interval::IntervalUnit;
 use common_time::time::Time;
 use common_time::timestamp::{TimeUnit, Timestamp};
-use common_time::Interval;
+use common_time::{Duration, Interval};
 use datafusion_common::ScalarValue;
 pub use ordered_float::OrderedFloat;
 use serde::{Deserialize, Serialize};
@@ -70,6 +70,7 @@ pub enum Value {
     DateTime(DateTime),
     Timestamp(Timestamp),
     Time(Time),
+    Duration(Duration),
     Interval(Interval),
 
     List(ListValue),
@@ -104,6 +105,7 @@ impl Display for Value {
             Value::Timestamp(v) => write!(f, "{}", v.to_iso8601_string()),
             Value::Time(t) => write!(f, "{}", t.to_iso8601_string()),
             Value::Interval(v) => write!(f, "{}", v.to_iso8601_string()),
+            Value::Duration(d) => write!(f, "{d}"),
             Value::List(v) => {
                 let default = Box::<Vec<Value>>::default();
                 let items = v.items().as_ref().unwrap_or(&default);
@@ -145,6 +147,7 @@ impl Value {
             Value::Timestamp(v) => ConcreteDataType::timestamp_datatype(v.unit()),
             Value::Interval(v) => ConcreteDataType::interval_datatype(v.unit()),
             Value::List(list) => ConcreteDataType::list_datatype(list.datatype().clone()),
+            Value::Duration(d) => ConcreteDataType::duration_datatype(d.unit()),
         }
     }
 
@@ -188,6 +191,7 @@ impl Value {
             Value::Timestamp(v) => ValueRef::Timestamp(*v),
             Value::Time(v) => ValueRef::Time(*v),
             Value::Interval(v) => ValueRef::Interval(*v),
+            Value::Duration(v) => ValueRef::Duration(*v),
         }
     }
 
@@ -246,6 +250,12 @@ impl Value {
                 IntervalUnit::DayTime => LogicalTypeId::IntervalDayTime,
                 IntervalUnit::MonthDayNano => LogicalTypeId::IntervalMonthDayNano,
             },
+            Value::Duration(d) => match d.unit() {
+                TimeUnit::Second => LogicalTypeId::DurationSecond,
+                TimeUnit::Millisecond => LogicalTypeId::DurationMillisecond,
+                TimeUnit::Microsecond => LogicalTypeId::DurationMicrosecond,
+                TimeUnit::Nanosecond => LogicalTypeId::DurationNanosecond,
+            },
         }
     }
 
@@ -292,6 +302,7 @@ impl Value {
                 IntervalUnit::DayTime => ScalarValue::IntervalDayTime(Some(v.to_i64())),
                 IntervalUnit::MonthDayNano => ScalarValue::IntervalMonthDayNano(Some(v.to_i128())),
             },
+            Value::Duration(d) => duration_to_scalar_value(d.unit(), Some(d.value())),
         };
 
         Ok(scalar_value)
@@ -330,6 +341,7 @@ pub fn to_null_scalar_value(output_type: &ConcreteDataType) -> Result<ScalarValu
             Box::new(to_null_scalar_value(dict.value_type())?),
         ),
         ConcreteDataType::Time(t) => time_to_scalar_value(t.unit(), None)?,
+        ConcreteDataType::Duration(d) => duration_to_scalar_value(d.unit(), None),
     })
 }
 
@@ -362,6 +374,16 @@ pub fn time_to_scalar_value(unit: TimeUnit, val: Option<i64>) -> Result<ScalarVa
     })
 }
 
+/// Cast the 64-bit duration into the arrow ScalarValue with time unit.
+pub fn duration_to_scalar_value(unit: TimeUnit, val: Option<i64>) -> ScalarValue {
+    match unit {
+        TimeUnit::Second => ScalarValue::DurationSecond(val),
+        TimeUnit::Millisecond => ScalarValue::DurationMillisecond(val),
+        TimeUnit::Microsecond => ScalarValue::DurationMicrosecond(val),
+        TimeUnit::Nanosecond => ScalarValue::DurationNanosecond(val),
+    }
+}
+
 /// Convert [ScalarValue] to [Timestamp].
 /// Return `None` if given scalar value cannot be converted to a valid timestamp.
 pub fn scalar_value_to_timestamp(scalar: &ScalarValue) -> Option<Timestamp> {
@@ -419,6 +441,7 @@ macro_rules! impl_ord_for_value_like {
                 ($Type::Timestamp(v1), $Type::Timestamp(v2)) => v1.cmp(v2),
                 ($Type::Time(v1), $Type::Time(v2)) => v1.cmp(v2),
                 ($Type::Interval(v1), $Type::Interval(v2)) => v1.cmp(v2),
+                ($Type::Duration(v1), $Type::Duration(v2)) => v1.cmp(v2),
                 ($Type::List(v1), $Type::List(v2)) => v1.cmp(v2),
                 _ => panic!(
                     "Cannot compare different values {:?} and {:?}",
@@ -536,6 +559,7 @@ impl_value_from!(Time, Time);
 impl_value_from!(DateTime, DateTime);
 impl_value_from!(Timestamp, Timestamp);
 impl_value_from!(Interval, Interval);
+impl_value_from!(Duration, Duration);
 impl_value_from!(String, String);
 
 impl From<&str> for Value {
@@ -581,6 +605,7 @@ impl TryFrom<Value> for serde_json::Value {
             Value::Timestamp(v) => serde_json::to_value(v.value())?,
             Value::Time(v) => serde_json::to_value(v.value())?,
             Value::Interval(v) => serde_json::to_value(v.to_i128())?,
+            Value::Duration(v) => serde_json::to_value(v.value())?,
         };
 
         Ok(json_value)
@@ -740,11 +765,19 @@ impl TryFrom<ScalarValue> for Value {
             ScalarValue::IntervalMonthDayNano(t) => t
                 .map(|x| Value::Interval(Interval::from_i128(x)))
                 .unwrap_or(Value::Null),
+            ScalarValue::DurationSecond(d) => d
+                .map(|x| Value::Duration(Duration::new(x, TimeUnit::Second)))
+                .unwrap_or(Value::Null),
+            ScalarValue::DurationMillisecond(d) => d
+                .map(|x| Value::Duration(Duration::new(x, TimeUnit::Millisecond)))
+                .unwrap_or(Value::Null),
+            ScalarValue::DurationMicrosecond(d) => d
+                .map(|x| Value::Duration(Duration::new(x, TimeUnit::Microsecond)))
+                .unwrap_or(Value::Null),
+            ScalarValue::DurationNanosecond(d) => d
+                .map(|x| Value::Duration(Duration::new(x, TimeUnit::Nanosecond)))
+                .unwrap_or(Value::Null),
             ScalarValue::Decimal128(_, _, _)
-            | ScalarValue::DurationSecond(_)
-            | ScalarValue::DurationMillisecond(_)
-            | ScalarValue::DurationMicrosecond(_)
-            | ScalarValue::DurationNanosecond(_)
             | ScalarValue::Struct(_, _)
             | ScalarValue::Dictionary(_, _) => {
                 return error::UnsupportedArrowTypeSnafu {
@@ -779,6 +812,7 @@ impl From<ValueRef<'_>> for Value {
             ValueRef::Timestamp(v) => Value::Timestamp(v),
             ValueRef::Time(v) => Value::Time(v),
             ValueRef::Interval(v) => Value::Interval(v),
+            ValueRef::Duration(v) => Value::Duration(v),
             ValueRef::List(v) => v.to_value(),
         }
     }
@@ -811,6 +845,7 @@ pub enum ValueRef<'a> {
     DateTime(DateTime),
     Timestamp(Timestamp),
     Time(Time),
+    Duration(Duration),
     Interval(Interval),
 
     // Compound types:
@@ -929,6 +964,10 @@ impl<'a> ValueRef<'a> {
         impl_as_for_value_ref!(self, Time)
     }
 
+    pub fn as_duration(&self) -> Result<Option<Duration>> {
+        impl_as_for_value_ref!(self, Duration)
+    }
+
     /// Cast itself to [Interval].
     pub fn as_interval(&self) -> Result<Option<Interval>> {
         impl_as_for_value_ref!(self, Interval)
@@ -987,6 +1026,7 @@ impl_value_ref_from!(DateTime, DateTime);
 impl_value_ref_from!(Timestamp, Timestamp);
 impl_value_ref_from!(Time, Time);
 impl_value_ref_from!(Interval, Interval);
+impl_value_ref_from!(Duration, Duration);
 
 impl<'a> From<&'a str> for ValueRef<'a> {
     fn from(string: &'a str) -> ValueRef<'a> {
@@ -1298,6 +1338,46 @@ mod tests {
             ScalarValue::Time64Nanosecond(None).try_into().unwrap()
         );
 
+        assert_eq!(
+            Value::Duration(Duration::new_second(1)),
+            ScalarValue::DurationSecond(Some(1)).try_into().unwrap()
+        );
+        assert_eq!(
+            Value::Null,
+            ScalarValue::DurationSecond(None).try_into().unwrap()
+        );
+
+        assert_eq!(
+            Value::Duration(Duration::new_millisecond(1)),
+            ScalarValue::DurationMillisecond(Some(1))
+                .try_into()
+                .unwrap()
+        );
+        assert_eq!(
+            Value::Null,
+            ScalarValue::DurationMillisecond(None).try_into().unwrap()
+        );
+
+        assert_eq!(
+            Value::Duration(Duration::new_microsecond(1)),
+            ScalarValue::DurationMicrosecond(Some(1))
+                .try_into()
+                .unwrap()
+        );
+        assert_eq!(
+            Value::Null,
+            ScalarValue::DurationMicrosecond(None).try_into().unwrap()
+        );
+
+        assert_eq!(
+            Value::Duration(Duration::new_nanosecond(1)),
+            ScalarValue::DurationNanosecond(Some(1)).try_into().unwrap()
+        );
+        assert_eq!(
+            Value::Null,
+            ScalarValue::DurationNanosecond(None).try_into().unwrap()
+        );
+
         let result: Result<Value> = ScalarValue::Decimal128(Some(1), 0, 0).try_into();
         assert!(result
             .unwrap_err()
@@ -1450,6 +1530,22 @@ mod tests {
             &ConcreteDataType::interval_month_day_nano_datatype(),
             &Value::Interval(Interval::from_month_day_nano(1, 2, 3)),
         );
+        check_type_and_value(
+            &ConcreteDataType::duration_second_datatype(),
+            &Value::Duration(Duration::new_second(1)),
+        );
+        check_type_and_value(
+            &ConcreteDataType::duration_millisecond_datatype(),
+            &Value::Duration(Duration::new_millisecond(1)),
+        );
+        check_type_and_value(
+            &ConcreteDataType::duration_microsecond_datatype(),
+            &Value::Duration(Duration::new_microsecond(1)),
+        );
+        check_type_and_value(
+            &ConcreteDataType::duration_nanosecond_datatype(),
+            &Value::Duration(Duration::new_nanosecond(1)),
+        );
     }
 
     #[test]
@@ -1549,6 +1645,10 @@ mod tests {
             serde_json::Value::Number(1.into()),
             to_json(Value::Time(Time::new_millisecond(1)))
         );
+        assert_eq!(
+            serde_json::Value::Number(1.into()),
+            to_json(Value::Duration(Duration::new_millisecond(1)))
+        );
 
         let json_value: serde_json::Value =
             serde_json::from_str(r#"{"items":[{"Int32":123}],"datatype":{"Int32":{}}}"#).unwrap();
@@ -1608,6 +1708,7 @@ mod tests {
         check_as_value_ref!(Timestamp, Timestamp::new_millisecond(1));
         check_as_value_ref!(Time, Time::new_millisecond(1));
         check_as_value_ref!(Interval, Interval::from_month_day_nano(1, 2, 3));
+        check_as_value_ref!(Duration, Duration::new_millisecond(1));
 
         assert_eq!(
             ValueRef::String("hello"),
@@ -1658,6 +1759,7 @@ mod tests {
         check_as_correct!(Date::new(123), Date, as_date);
         check_as_correct!(DateTime::new(12), DateTime, as_datetime);
         check_as_correct!(Time::new_second(12), Time, as_time);
+        check_as_correct!(Duration::new_second(12), Duration, as_duration);
         let list = ListValue {
             items: None,
             datatype: ConcreteDataType::int32_datatype(),
@@ -1709,6 +1811,10 @@ mod tests {
             Value::Time(Time::new(1000, TimeUnit::Millisecond)).to_string(),
             "08:00:01+0800"
         );
+        assert_eq!(
+            Value::Duration(Duration::new_millisecond(1000)).to_string(),
+            "1000ms"
+        );
         assert_eq!(
             Value::List(ListValue::new(
                 Some(Box::new(vec![Value::Int8(1), Value::Int8(2)])),
@@ -1944,6 +2050,31 @@ mod tests {
                 .try_to_scalar_value(&ConcreteDataType::time_nanosecond_datatype())
                 .unwrap()
         );
+
+        assert_eq!(
+            ScalarValue::DurationSecond(None),
+            Value::Null
+                .try_to_scalar_value(&ConcreteDataType::duration_second_datatype())
+                .unwrap()
+        );
+        assert_eq!(
+            ScalarValue::DurationMillisecond(None),
+            Value::Null
+                .try_to_scalar_value(&ConcreteDataType::duration_millisecond_datatype())
+                .unwrap()
+        );
+        assert_eq!(
+            ScalarValue::DurationMicrosecond(None),
+            Value::Null
+                .try_to_scalar_value(&ConcreteDataType::duration_microsecond_datatype())
+                .unwrap()
+        );
+        assert_eq!(
+            ScalarValue::DurationNanosecond(None),
+            Value::Null
+                .try_to_scalar_value(&ConcreteDataType::duration_nanosecond_datatype())
+                .unwrap()
+        );
     }
 
     #[test]
@@ -2009,4 +2140,24 @@ mod tests {
             time_to_scalar_value(TimeUnit::Nanosecond, Some(1)).unwrap()
         );
     }
+
+    #[test]
+    fn test_duration_to_scalar_value() {
+        assert_eq!(
+            ScalarValue::DurationSecond(Some(1)),
+            duration_to_scalar_value(TimeUnit::Second, Some(1))
+        );
+        assert_eq!(
+            ScalarValue::DurationMillisecond(Some(1)),
+            duration_to_scalar_value(TimeUnit::Millisecond, Some(1))
+        );
+        assert_eq!(
+            ScalarValue::DurationMicrosecond(Some(1)),
+            duration_to_scalar_value(TimeUnit::Microsecond, Some(1))
+        );
+        assert_eq!(
+            ScalarValue::DurationNanosecond(Some(1)),
+            duration_to_scalar_value(TimeUnit::Nanosecond, Some(1))
+        );
+    }
 }

src/datatypes/src/vectors.rs

@@ -30,6 +30,7 @@ mod boolean;
 mod constant;
 mod date;
 mod datetime;
+mod duration;
 mod eq;
 mod helper;
 mod interval;
@@ -47,6 +48,11 @@ pub use boolean::{BooleanVector, BooleanVectorBuilder};
 pub use constant::ConstantVector;
 pub use date::{DateVector, DateVectorBuilder};
 pub use datetime::{DateTimeVector, DateTimeVectorBuilder};
+pub use duration::{
+    DurationMicrosecondVector, DurationMicrosecondVectorBuilder, DurationMillisecondVector,
+    DurationMillisecondVectorBuilder, DurationNanosecondVector, DurationNanosecondVectorBuilder,
+    DurationSecondVector, DurationSecondVectorBuilder,
+};
 pub use helper::Helper;
 pub use interval::{
     IntervalDayTimeVector, IntervalDayTimeVectorBuilder, IntervalMonthDayNanoVector,

src/datatypes/src/vectors/duration.rs

@@ -0,0 +1,30 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::types::{
+    DurationMicrosecondType, DurationMillisecondType, DurationNanosecondType, DurationSecondType,
+};
+use crate::vectors::{PrimitiveVector, PrimitiveVectorBuilder};
+
+pub type DurationSecondVector = PrimitiveVector<DurationSecondType>;
+pub type DurationSecondVectorBuilder = PrimitiveVectorBuilder<DurationSecondType>;
+
+pub type DurationMillisecondVector = PrimitiveVector<DurationMillisecondType>;
+pub type DurationMillisecondVectorBuilder = PrimitiveVectorBuilder<DurationMillisecondType>;
+
+pub type DurationMicrosecondVector = PrimitiveVector<DurationMicrosecondType>;
+pub type DurationMicrosecondVectorBuilder = PrimitiveVectorBuilder<DurationMicrosecondType>;
+
+pub type DurationNanosecondVector = PrimitiveVector<DurationNanosecondType>;
+pub type DurationNanosecondVectorBuilder = PrimitiveVectorBuilder<DurationNanosecondType>;

src/datatypes/src/vectors/eq.rs

@@ -17,14 +17,15 @@ use std::sync::Arc;
 use common_time::interval::IntervalUnit;
 
 use crate::data_type::DataType;
-use crate::types::{TimeType, TimestampType};
+use crate::types::{DurationType, TimeType, TimestampType};
 use crate::vectors::constant::ConstantVector;
 use crate::vectors::{
-    BinaryVector, BooleanVector, DateTimeVector, DateVector, IntervalDayTimeVector,
-    IntervalMonthDayNanoVector, IntervalYearMonthVector, ListVector, PrimitiveVector, StringVector,
-    TimeMicrosecondVector, TimeMillisecondVector, TimeNanosecondVector, TimeSecondVector,
-    TimestampMicrosecondVector, TimestampMillisecondVector, TimestampNanosecondVector,
-    TimestampSecondVector, Vector,
+    BinaryVector, BooleanVector, DateTimeVector, DateVector, DurationMicrosecondVector,
+    DurationMillisecondVector, DurationNanosecondVector, DurationSecondVector,
+    IntervalDayTimeVector, IntervalMonthDayNanoVector, IntervalYearMonthVector, ListVector,
+    PrimitiveVector, StringVector, TimeMicrosecondVector, TimeMillisecondVector,
+    TimeNanosecondVector, TimeSecondVector, TimestampMicrosecondVector, TimestampMillisecondVector,
+    TimestampNanosecondVector, TimestampSecondVector, Vector,
 };
 use crate::with_match_primitive_type_id;
 
@@ -136,6 +137,20 @@ fn equal(lhs: &dyn Vector, rhs: &dyn Vector) -> bool {
                 is_vector_eq!(TimeNanosecondVector, lhs, rhs)
             }
         },
+        Duration(d) => match d {
+            DurationType::Second(_) => {
+                is_vector_eq!(DurationSecondVector, lhs, rhs)
+            }
+            DurationType::Millisecond(_) => {
+                is_vector_eq!(DurationMillisecondVector, lhs, rhs)
+            }
+            DurationType::Microsecond(_) => {
+                is_vector_eq!(DurationMicrosecondVector, lhs, rhs)
+            }
+            DurationType::Nanosecond(_) => {
+                is_vector_eq!(DurationNanosecondVector, lhs, rhs)
+            }
+        },
     }
 }
 
@@ -143,8 +158,9 @@ fn equal(lhs: &dyn Vector, rhs: &dyn Vector) -> bool {
 mod tests {
     use super::*;
     use crate::vectors::{
-        list, Float32Vector, Float64Vector, Int16Vector, Int32Vector, Int64Vector, Int8Vector,
-        NullVector, UInt16Vector, UInt32Vector, UInt64Vector, UInt8Vector, VectorRef,
+        list, DurationMicrosecondVector, DurationMillisecondVector, DurationNanosecondVector,
+        DurationSecondVector, Float32Vector, Float64Vector, Int16Vector, Int32Vector, Int64Vector,
+        Int8Vector, NullVector, UInt16Vector, UInt32Vector, UInt64Vector, UInt8Vector, VectorRef,
     };
 
     fn assert_vector_ref_eq(vector: VectorRef) {
@@ -222,6 +238,10 @@ mod tests {
         assert_vector_ref_eq(Arc::new(IntervalMonthDayNanoVector::from_values([
             1000, 2000, 3000, 4000,
         ])));
+        assert_vector_ref_eq(Arc::new(DurationSecondVector::from_values([300, 310])));
+        assert_vector_ref_eq(Arc::new(DurationMillisecondVector::from_values([300, 310])));
+        assert_vector_ref_eq(Arc::new(DurationMicrosecondVector::from_values([300, 310])));
+        assert_vector_ref_eq(Arc::new(DurationNanosecondVector::from_values([300, 310])));
     }
 
     #[test]
@@ -287,5 +307,10 @@ mod tests {
             Arc::new(IntervalYearMonthVector::from_values([1000, 2000])),
             Arc::new(IntervalYearMonthVector::from_values([2100, 1200])),
         );
+
+        assert_vector_ref_ne(
+            Arc::new(DurationSecondVector::from_values([300, 310])),
+            Arc::new(DurationSecondVector::from_values([300, 320])),
+        );
     }
 }

src/datatypes/src/vectors/primitive.rs

@@ -23,7 +23,10 @@ use arrow::array::{
     TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray,
     TimestampSecondArray,
 };
-use arrow_array::{IntervalDayTimeArray, IntervalMonthDayNanoArray, IntervalYearMonthArray};
+use arrow_array::{
+    DurationMicrosecondArray, DurationMillisecondArray, DurationNanosecondArray,
+    DurationSecondArray, IntervalDayTimeArray, IntervalMonthDayNanoArray, IntervalYearMonthArray,
+};
 use arrow_schema::DataType;
 use serde_json::Value as JsonValue;
 use snafu::OptionExt;
@@ -198,6 +201,45 @@ impl<T: LogicalPrimitiveType> PrimitiveVector<T> {
         Ok(Self::new(concrete_array))
     }
 
+    pub fn try_from_arrow_duration_array(array: impl AsRef<dyn Array>) -> Result<Self> {
+        let array = array.as_ref();
+        let array_data = match array.data_type() {
+            DataType::Duration(unit) => match unit {
+                arrow_schema::TimeUnit::Second => array
+                    .as_any()
+                    .downcast_ref::<DurationSecondArray>()
+                    .unwrap()
+                    .to_data(),
+                arrow_schema::TimeUnit::Millisecond => array
+                    .as_any()
+                    .downcast_ref::<DurationMillisecondArray>()
+                    .unwrap()
+                    .to_data(),
+                arrow_schema::TimeUnit::Microsecond => array
+                    .as_any()
+                    .downcast_ref::<DurationMicrosecondArray>()
+                    .unwrap()
+                    .to_data(),
+                arrow_schema::TimeUnit::Nanosecond => array
+                    .as_any()
+                    .downcast_ref::<DurationNanosecondArray>()
+                    .unwrap()
+                    .to_data(),
+            },
+            arrow_type => {
+                return CastTypeSnafu {
+                    msg: format!(
+                        "Failed to cast arrow array {:?} to interval vector",
+                        arrow_type,
+                    ),
+                }
+                .fail()?;
+            }
+        };
+        let concrete_array = PrimitiveArray::<T::ArrowPrimitive>::from(array_data);
+        Ok(Self::new(concrete_array))
+    }
+
     pub fn from_slice<P: AsRef<[T::Native]>>(slice: P) -> Self {
         let iter = slice.as_ref().iter().copied();
         Self {
@@ -538,7 +580,7 @@ mod tests {
         Time64NanosecondArray,
     };
     use arrow::datatypes::DataType as ArrowDataType;
-    use arrow_array::{IntervalDayTimeArray, IntervalYearMonthArray};
+    use arrow_array::{DurationSecondArray, IntervalDayTimeArray, IntervalYearMonthArray};
     use serde_json;
 
     use super::*;
@@ -546,9 +588,11 @@ mod tests {
     use crate::serialize::Serializable;
     use crate::types::Int64Type;
     use crate::vectors::{
-        IntervalDayTimeVector, IntervalYearMonthVector, TimeMicrosecondVector,
-        TimeMillisecondVector, TimeNanosecondVector, TimeSecondVector, TimestampMicrosecondVector,
-        TimestampMillisecondVector, TimestampNanosecondVector, TimestampSecondVector,
+        DurationMicrosecondVector, DurationMillisecondVector, DurationNanosecondVector,
+        DurationSecondVector, IntervalDayTimeVector, IntervalYearMonthVector,
+        TimeMicrosecondVector, TimeMillisecondVector, TimeNanosecondVector, TimeSecondVector,
+        TimestampMicrosecondVector, TimestampMillisecondVector, TimestampNanosecondVector,
+        TimestampSecondVector,
     };
 
     fn check_vec(v: Int32Vector) {
@@ -779,4 +823,35 @@ mod tests {
             vector
         );
     }
+
+    #[test]
+    fn test_try_from_arrow_duration_array() {
+        let array: ArrayRef = Arc::new(DurationSecondArray::from(vec![1000, 2000, 3000]));
+        let vector = DurationSecondVector::try_from_arrow_duration_array(array).unwrap();
+        assert_eq!(
+            DurationSecondVector::from_values(vec![1000, 2000, 3000]),
+            vector
+        );
+
+        let array: ArrayRef = Arc::new(DurationMillisecondArray::from(vec![1000, 2000, 3000]));
+        let vector = DurationMillisecondVector::try_from_arrow_duration_array(array).unwrap();
+        assert_eq!(
+            DurationMillisecondVector::from_values(vec![1000, 2000, 3000]),
+            vector
+        );
+
+        let array: ArrayRef = Arc::new(DurationMicrosecondArray::from(vec![1000, 2000, 3000]));
+        let vector = DurationMicrosecondVector::try_from_arrow_duration_array(array).unwrap();
+        assert_eq!(
+            DurationMicrosecondVector::from_values(vec![1000, 2000, 3000]),
+            vector
+        );
+
+        let array: ArrayRef = Arc::new(DurationNanosecondArray::from(vec![1000, 2000, 3000]));
+        let vector = DurationNanosecondVector::try_from_arrow_duration_array(array).unwrap();
+        assert_eq!(
+            DurationNanosecondVector::from_values(vec![1000, 2000, 3000]),
+            vector
+        );
+    }
 }

src/mito2/src/row_converter.rs

@@ -15,7 +15,7 @@
 use bytes::Buf;
 use common_base::bytes::Bytes;
 use common_time::time::Time;
-use common_time::{Date, Interval};
+use common_time::{Date, Duration, Interval};
 use datatypes::data_type::ConcreteDataType;
 use datatypes::prelude::Value;
 use datatypes::value::ValueRef;
@@ -72,6 +72,7 @@ impl SortField {
             ConcreteDataType::DateTime(_) => 9,
             ConcreteDataType::Timestamp(_) => 10,
             ConcreteDataType::Time(_) => 10,
+            ConcreteDataType::Duration(_) => 10,
             ConcreteDataType::Interval(_) => 18,
             ConcreteDataType::Null(_)
             | ConcreteDataType::List(_)
@@ -136,7 +137,8 @@ impl SortField {
             Date, date,
             DateTime, datetime,
             Time, time,
-            Interval, interval
+            Interval, interval,
+            Duration, duration
         );
 
         Ok(())
@@ -201,7 +203,8 @@ impl SortField {
             Date, Date,
             Time, Time,
             DateTime, DateTime,
-            Interval, Interval
+            Interval, Interval,
+            Duration, Duration
         )
     }
 }
@@ -311,6 +314,20 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_memcmp_duration() {
+        check_encode_and_decode(
+            &[
+                ConcreteDataType::duration_millisecond_datatype(),
+                ConcreteDataType::int64_datatype(),
+            ],
+            vec![
+                Value::Duration(Duration::new_millisecond(44)),
+                Value::Int64(45),
+            ],
+        )
+    }
+
     #[test]
     fn test_memcmp_binary() {
         check_encode_and_decode(

src/operator/src/req_convert/common.rs

@@ -161,6 +161,22 @@ fn push_column_to_rows(column: Column, rows: &mut [Row]) -> Result<()> {
             IntervalMonthDayNanoValues,
             interval_month_day_nano_values
         ),
+        (DurationSecond, DurationSecondValue, duration_second_values),
+        (
+            DurationMillisecond,
+            DurationMillisecondValue,
+            duration_millisecond_values
+        ),
+        (
+            DurationMicrosecond,
+            DurationMicrosecondValue,
+            duration_microsecond_values
+        ),
+        (
+            DurationNanosecond,
+            DurationNanosecondValue,
+            duration_nanosecond_values
+        ),
     );
 
     Ok(())

src/servers/src/mysql/writer.rs

@@ -186,6 +186,7 @@ impl<'a, W: AsyncWrite + Unpin> MysqlResultWriter<'a, W> {
                     Value::Timestamp(v) => row_writer
                         .write_col(v.to_timezone_aware_string(query_context.time_zone()))?,
                     Value::Interval(v) => row_writer.write_col(v.to_iso8601_string())?,
+                    Value::Duration(v) => row_writer.write_col(v.to_std_duration())?,
                     Value::List(_) => {
                         return Err(Error::Internal {
                             err_msg: format!(
@@ -241,6 +242,7 @@ pub(crate) fn create_mysql_column(
         ConcreteDataType::Date(_) => Ok(ColumnType::MYSQL_TYPE_DATE),
         ConcreteDataType::DateTime(_) => Ok(ColumnType::MYSQL_TYPE_DATETIME),
         ConcreteDataType::Interval(_) => Ok(ColumnType::MYSQL_TYPE_VARCHAR),
+        ConcreteDataType::Duration(_) => Ok(ColumnType::MYSQL_TYPE_TIME),
         _ => error::InternalSnafu {
             err_msg: format!("not implemented for column datatype {:?}", data_type),
         }

src/servers/src/postgres/types.rs

@@ -103,12 +103,14 @@ pub(super) fn encode_value(value: &Value, builder: &mut DataRowEncoder) -> PgWir
             }
         }
         Value::Interval(v) => builder.encode_field(&PgInterval::from(*v)),
-        Value::List(_) => Err(PgWireError::ApiError(Box::new(Error::Internal {
-            err_msg: format!(
-                "cannot write value {:?} in postgres protocol: unimplemented",
-                &value
-            ),
-        }))),
+        Value::List(_) | Value::Duration(_) => {
+            Err(PgWireError::ApiError(Box::new(Error::Internal {
+                err_msg: format!(
+                    "cannot write value {:?} in postgres protocol: unimplemented",
+                    &value
+                ),
+            })))
+        }
     }
 }
 
@@ -129,7 +131,9 @@ pub(super) fn type_gt_to_pg(origin: &ConcreteDataType) -> Result<Type> {
         &ConcreteDataType::Timestamp(_) => Ok(Type::TIMESTAMP),
         &ConcreteDataType::Time(_) => Ok(Type::TIME),
         &ConcreteDataType::Interval(_) => Ok(Type::INTERVAL),
-        &ConcreteDataType::List(_) | &ConcreteDataType::Dictionary(_) => error::InternalSnafu {
+        &ConcreteDataType::Duration(_)
+        | &ConcreteDataType::List(_)
+        | &ConcreteDataType::Dictionary(_) => error::InternalSnafu {
             err_msg: format!("not implemented for column datatype {origin:?}"),
         }
         .fail(),

src/sql/src/statements.rs

@@ -419,11 +419,14 @@ pub fn concrete_data_type_to_sql_data_type(data_type: &ConcreteDataType) -> Resu
             Some(time_type.precision()),
             TimezoneInfo::None,
         )),
+        ConcreteDataType::Interval(_) => Ok(SqlDataType::Interval),
         ConcreteDataType::Binary(_) => Ok(SqlDataType::Varbinary(None)),
-        ConcreteDataType::Null(_) | ConcreteDataType::List(_) | ConcreteDataType::Dictionary(_) => {
+        ConcreteDataType::Duration(_)
+        | ConcreteDataType::Null(_)
+        | ConcreteDataType::List(_)
+        | ConcreteDataType::Dictionary(_) => {
             unreachable!()
         }
-        ConcreteDataType::Interval(_) => Ok(SqlDataType::Interval),
     }
 }