diff --git a/src/common/query/src/logical_plan/accumulator.rs b/src/common/query/src/logical_plan/accumulator.rs
index ee411efd5a..58eb2ab9d9 100644
--- a/src/common/query/src/logical_plan/accumulator.rs
+++ b/src/common/query/src/logical_plan/accumulator.rs
@@ -181,7 +181,7 @@ fn try_into_scalar_value(value: Value, datatype: &ConcreteDataType) -> Result<Sc
         Value::String(v) => ScalarValue::LargeUtf8(Some(v.as_utf8().to_string())),
         Value::Binary(v) => ScalarValue::LargeBinary(Some(v.to_vec())),
         Value::Date(v) => ScalarValue::Date32(Some(v.val())),
-        Value::DateTime(v) => ScalarValue::Date64(Some(v)),
+        Value::DateTime(v) => ScalarValue::Date64(Some(v.val())),
         Value::Null => try_convert_null_value(datatype)?,
         Value::List(list) => try_convert_list_value(list)?,
     })
diff --git a/src/common/time/src/datetime.rs b/src/common/time/src/datetime.rs
new file mode 100644
index 0000000000..d927a4b04e
--- /dev/null
+++ b/src/common/time/src/datetime.rs
@@ -0,0 +1,95 @@
+use std::fmt::{Display, Formatter};
+use std::str::FromStr;
+
+use chrono::NaiveDateTime;
+use serde::{Deserialize, Serialize};
+use snafu::{ensure, ResultExt};
+
+use crate::error::{DateTimeOverflowSnafu, Error, ParseDateStrSnafu, Result};
+
+const DATETIME_FORMAT: &str = "%F %T";
+
+/// [DateTime] represents the **seconds elapsed since "1970-01-01 00:00:00 UTC" (UNIX Epoch)**.  
+///
+/// Valid [DateTime] value ranges from "1000-01-01 00:00:00" to "9999-12-31 23:59:59".
+#[derive(
+    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default, Serialize, Deserialize,
+)]
+pub struct DateTime(i64);
+
+impl Display for DateTime {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let abs_time = NaiveDateTime::from_timestamp(self.0, 0);
+        write!(f, "{}", abs_time.format(DATETIME_FORMAT))
+    }
+}
+
+impl From<DateTime> for serde_json::Value {
+    fn from(d: DateTime) -> Self {
+        serde_json::Value::String(d.to_string())
+    }
+}
+
+impl FromStr for DateTime {
+    type Err = Error;
+
+    fn from_str(s: &str) -> Result<Self> {
+        let datetime = NaiveDateTime::parse_from_str(s, DATETIME_FORMAT)
+            .context(ParseDateStrSnafu { raw: s })?;
+        Ok(Self(datetime.timestamp()))
+    }
+}
+
+impl DateTime {
+    pub fn try_new(val: i64) -> Result<Self> {
+        ensure!(
+            val >= Self::MIN.0 && val <= Self::MAX.0,
+            DateTimeOverflowSnafu { value: val }
+        );
+
+        Ok(Self(val))
+    }
+
+    pub fn val(&self) -> i64 {
+        self.0
+    }
+
+    /// Max valid DateTime value: 9999-12-31 23:59:59
+    pub const MAX: DateTime = DateTime(253402300799);
+    /// Min valid DateTime value: 0000-01-01 00:00:00
+    pub const MIN: DateTime = DateTime(-30610224000);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    pub fn test_new_date_time() {
+        assert_eq!(
+            "1970-01-01 00:00:00",
+            DateTime::try_new(0).unwrap().to_string()
+        );
+        assert_eq!(
+            "1970-01-01 00:00:01",
+            DateTime::try_new(1).unwrap().to_string()
+        );
+        assert_eq!(
+            "1969-12-31 23:59:59",
+            DateTime::try_new(-1).unwrap().to_string()
+        );
+    }
+
+    #[test]
+    pub fn test_max_min() {
+        assert_eq!("9999-12-31 23:59:59", DateTime::MAX.to_string());
+        assert_eq!("1000-01-01 00:00:00", DateTime::MIN.to_string());
+    }
+
+    #[test]
+    pub fn test_parse_from_string() {
+        let time = "1970-01-01 00:00:00";
+        let dt = DateTime::from_str(time).unwrap();
+        assert_eq!(time, &dt.to_string());
+    }
+}
diff --git a/src/common/time/src/error.rs b/src/common/time/src/error.rs
index 371d6aa76d..b9c9fa28b6 100644
--- a/src/common/time/src/error.rs
+++ b/src/common/time/src/error.rs
@@ -1,5 +1,5 @@
 use chrono::ParseError;
-use snafu::Snafu;
+use snafu::{Backtrace, Snafu};
 
 #[derive(Debug, Snafu)]
 #[snafu(visibility(pub))]
@@ -7,8 +7,11 @@ pub enum Error {
     #[snafu(display("Failed to parse string to date, raw: {}, source: {}", raw, source))]
     ParseDateStr { raw: String, source: ParseError },
 
-    #[snafu(display("Failed to parse i32 value to date: {}", value))]
-    DateOverflow { value: i32 },
+    #[snafu(display("Failed to parse i32 value to Date: {}", value))]
+    DateOverflow { value: i32, backtrace: Backtrace },
+
+    #[snafu(display("Failed to parse i64 value to DateTime: {}", value))]
+    DateTimeOverflow { value: i64, backtrace: Backtrace },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
diff --git a/src/common/time/src/lib.rs b/src/common/time/src/lib.rs
index 43371f84d2..e8bf43975e 100644
--- a/src/common/time/src/lib.rs
+++ b/src/common/time/src/lib.rs
@@ -1,4 +1,5 @@
 pub mod date;
+pub mod datetime;
 pub mod error;
 pub mod range;
 pub mod timestamp;
diff --git a/src/datanode/src/sql/create.rs b/src/datanode/src/sql/create.rs
index c99a7d1de1..f933a7cbe4 100644
--- a/src/datanode/src/sql/create.rs
+++ b/src/datanode/src/sql/create.rs
@@ -5,6 +5,7 @@ use catalog::RegisterTableRequest;
 use common_telemetry::tracing::info;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::schema::{ColumnSchema, SchemaBuilder};
+use datatypes::types::DateTimeType;
 use query::query_engine::Output;
 use snafu::{OptionExt, ResultExt};
 use sql::ast::{ColumnDef, ColumnOption, DataType as SqlDataType, ObjectName, TableConstraint};
@@ -201,7 +202,16 @@ fn sql_data_type_to_concrete_data_type(t: &SqlDataType) -> Result<ConcreteDataTy
         SqlDataType::Double => Ok(ConcreteDataType::float64_datatype()),
         SqlDataType::Boolean => Ok(ConcreteDataType::boolean_datatype()),
         SqlDataType::Date => Ok(ConcreteDataType::date_datatype()),
-        // TODO(hl): DateTime not supported
+        SqlDataType::Custom(obj_name) => match &obj_name.0[..] {
+            [type_name] => {
+                if type_name.value.eq_ignore_ascii_case(DateTimeType::name()) {
+                    Ok(ConcreteDataType::datetime_datatype())
+                } else {
+                    SqlTypeNotSupportedSnafu { t: t.clone() }.fail()
+                }
+            }
+            _ => SqlTypeNotSupportedSnafu { t: t.clone() }.fail(),
+        },
         _ => SqlTypeNotSupportedSnafu { t: t.clone() }.fail(),
     }
 }
@@ -210,6 +220,7 @@ fn sql_data_type_to_concrete_data_type(t: &SqlDataType) -> Result<ConcreteDataTy
 mod tests {
     use std::assert_matches::assert_matches;
 
+    use sql::ast::Ident;
     use sql::dialect::GenericDialect;
     use sql::parser::ParserContext;
     use sql::statements::statement::Statement;
@@ -387,5 +398,9 @@ mod tests {
         check_type(SqlDataType::Double, ConcreteDataType::float64_datatype());
         check_type(SqlDataType::Boolean, ConcreteDataType::boolean_datatype());
         check_type(SqlDataType::Date, ConcreteDataType::date_datatype());
+        check_type(
+            SqlDataType::Custom(ObjectName(vec![Ident::new("datetime")])),
+            ConcreteDataType::datetime_datatype(),
+        );
     }
 }
diff --git a/src/datanode/src/sql/insert.rs b/src/datanode/src/sql/insert.rs
index 107fdcb74d..949a0b19d3 100644
--- a/src/datanode/src/sql/insert.rs
+++ b/src/datanode/src/sql/insert.rs
@@ -175,6 +175,16 @@ fn parse_string_to_value(s: String, data_type: &ConcreteDataType) -> Result<Valu
                 .fail()
             }
         }
+        ConcreteDataType::DateTime(_) => {
+            if let Ok(datetime) = common_time::datetime::DateTime::from_str(&s) {
+                Ok(Value::DateTime(datetime))
+            } else {
+                ParseSqlValueSnafu {
+                    msg: format!("Failed to parse {} to DateTime value", s),
+                }
+                .fail()
+            }
+        }
         _ => {
             unreachable!()
         }
@@ -295,4 +305,30 @@ mod tests {
             unreachable!()
         }
     }
+
+    #[test]
+    pub fn test_parse_datetime_literal() {
+        let value = parse_sql_value(
+            "datetime_col",
+            &ConcreteDataType::datetime_datatype(),
+            &SqlValue::DoubleQuotedString("2022-02-22 00:01:03".to_string()),
+        )
+        .unwrap();
+        assert_eq!(ConcreteDataType::date_datatype(), value.data_type());
+        if let Value::DateTime(d) = value {
+            assert_eq!("2022-02-22 00:01:03", d.to_string());
+        } else {
+            unreachable!()
+        }
+    }
+
+    #[test]
+    pub fn test_parse_illegal_datetime_literal() {
+        assert!(parse_sql_value(
+            "datetime_col",
+            &ConcreteDataType::datetime_datatype(),
+            &SqlValue::DoubleQuotedString("2022-02-22 00:01:61".to_string()),
+        )
+        .is_err());
+    }
 }
diff --git a/src/datatypes/src/data_type.rs b/src/datatypes/src/data_type.rs
index bbd7382473..605d08ba89 100644
--- a/src/datatypes/src/data_type.rs
+++ b/src/datatypes/src/data_type.rs
@@ -6,6 +6,7 @@ use serde::{Deserialize, Serialize};
 
 use crate::error::{self, Error, Result};
 use crate::type_id::LogicalTypeId;
+use crate::types::DateTimeType;
 use crate::types::{
     BinaryType, BooleanType, DateType, Float32Type, Float64Type, Int16Type, Int32Type, Int64Type,
     Int8Type, ListType, NullType, StringType, UInt16Type, UInt32Type, UInt64Type, UInt8Type,
@@ -35,6 +36,7 @@ pub enum ConcreteDataType {
     String(StringType),
 
     Date(DateType),
+    DateTime(DateTimeType),
 
     List(ListType),
 }
@@ -54,7 +56,7 @@ impl ConcreteDataType {
     pub fn is_string(&self) -> bool {
         matches!(
             self,
-            ConcreteDataType::String(_) | ConcreteDataType::Date(_)
+            ConcreteDataType::String(_) | ConcreteDataType::Date(_) | ConcreteDataType::DateTime(_)
         )
     }
 
@@ -66,6 +68,7 @@ impl ConcreteDataType {
                 | ConcreteDataType::Int32(_)
                 | ConcreteDataType::Int64(_)
                 | ConcreteDataType::Date(_)
+                | ConcreteDataType::DateTime(_)
         )
     }
 
@@ -125,6 +128,7 @@ impl TryFrom<&ArrowDataType> for ConcreteDataType {
             ArrowDataType::Float32 => Self::float32_datatype(),
             ArrowDataType::Float64 => Self::float64_datatype(),
             ArrowDataType::Date32 => Self::date_datatype(),
+            ArrowDataType::Date64 => Self::datetime_datatype(),
             ArrowDataType::Binary | ArrowDataType::LargeBinary => Self::binary_datatype(),
             ArrowDataType::Utf8 | ArrowDataType::LargeUtf8 => Self::string_datatype(),
             ArrowDataType::List(field) => Self::List(ListType::new(
@@ -158,7 +162,7 @@ macro_rules! impl_new_concrete_type_functions {
 
 impl_new_concrete_type_functions!(
     Null, Boolean, UInt8, UInt16, UInt32, UInt64, Int8, Int16, Int32, Int64, Float32, Float64,
-    Binary, String, Date
+    Binary, String, Date, DateTime
 );
 
 impl ConcreteDataType {
diff --git a/src/datatypes/src/lib.rs b/src/datatypes/src/lib.rs
index 19f0fdfc18..1a3a36f686 100644
--- a/src/datatypes/src/lib.rs
+++ b/src/datatypes/src/lib.rs
@@ -1,4 +1,5 @@
 #![feature(generic_associated_types)]
+#![feature(assert_matches)]
 
 pub mod arrow_array;
 pub mod data_type;
diff --git a/src/datatypes/src/scalars.rs b/src/datatypes/src/scalars.rs
index 987f4419cc..6f8cd9fbbf 100644
--- a/src/datatypes/src/scalars.rs
+++ b/src/datatypes/src/scalars.rs
@@ -2,6 +2,7 @@ use std::any::Any;
 
 use crate::prelude::*;
 use crate::vectors::date::DateVector;
+use crate::vectors::datetime::DateTimeVector;
 use crate::vectors::*;
 
 pub mod common;
@@ -263,6 +264,28 @@ impl<'a> ScalarRef<'a> for common_time::date::Date {
     }
 }
 
+impl Scalar for common_time::datetime::DateTime {
+    type VectorType = DateTimeVector;
+    type RefType<'a> = common_time::datetime::DateTime;
+
+    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 common_time::datetime::DateTime {
+    type VectorType = DateTimeVector;
+    type ScalarType = common_time::datetime::DateTime;
+
+    fn to_owned_scalar(&self) -> Self::ScalarType {
+        *self
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use common_time::date::Date;
diff --git a/src/datatypes/src/types.rs b/src/datatypes/src/types.rs
index eb53902f78..764d7e963f 100644
--- a/src/datatypes/src/types.rs
+++ b/src/datatypes/src/types.rs
@@ -1,6 +1,7 @@
 mod binary_type;
 mod boolean_type;
 mod date;
+mod datetime;
 mod list_type;
 mod null_type;
 mod primitive_traits;
@@ -10,6 +11,7 @@ mod string_type;
 pub use binary_type::BinaryType;
 pub use boolean_type::BooleanType;
 pub use date::DateType;
+pub use datetime::DateTimeType;
 pub use list_type::ListType;
 pub use null_type::NullType;
 pub use primitive_traits::Primitive;
diff --git a/src/datatypes/src/types/datetime.rs b/src/datatypes/src/types/datetime.rs
new file mode 100644
index 0000000000..31a0f8dec1
--- /dev/null
+++ b/src/datatypes/src/types/datetime.rs
@@ -0,0 +1,41 @@
+use std::sync::Arc;
+
+use arrow::datatypes::DataType as ArrowDataType;
+use serde::{Deserialize, Serialize};
+
+use crate::data_type::{DataType, DataTypeRef};
+use crate::prelude::{LogicalTypeId, Value};
+
+#[derive(Debug, Default, Clone, PartialEq, Serialize, Deserialize)]
+pub struct DateTimeType;
+
+const DATE_TIME_TYPE_NAME: &str = "DateTime";
+
+/// [DateTimeType] represents the seconds elapsed since UNIX EPOCH.
+impl DataType for DateTimeType {
+    fn name(&self) -> &str {
+        DATE_TIME_TYPE_NAME
+    }
+
+    fn logical_type_id(&self) -> LogicalTypeId {
+        LogicalTypeId::DateTime
+    }
+
+    fn default_value(&self) -> Value {
+        Value::DateTime(Default::default())
+    }
+
+    fn as_arrow_type(&self) -> ArrowDataType {
+        ArrowDataType::Date64
+    }
+}
+
+impl DateTimeType {
+    pub fn arc() -> DataTypeRef {
+        Arc::new(Self)
+    }
+
+    pub fn name() -> &'static str {
+        DATE_TIME_TYPE_NAME
+    }
+}
diff --git a/src/datatypes/src/value.rs b/src/datatypes/src/value.rs
index d76acb18b0..36b44d921e 100644
--- a/src/datatypes/src/value.rs
+++ b/src/datatypes/src/value.rs
@@ -37,7 +37,7 @@ pub enum Value {
 
     // Date & Time types:
     Date(common_time::date::Date),
-    DateTime(i64),
+    DateTime(common_time::datetime::DateTime),
 
     List(ListValue),
 }
@@ -65,9 +65,7 @@ impl Value {
             Value::Binary(_) => ConcreteDataType::binary_datatype(),
             Value::List(list) => ConcreteDataType::list_datatype(list.datatype().clone()),
             Value::Date(_) => ConcreteDataType::date_datatype(),
-            Value::DateTime(_) => {
-                unimplemented!("Unsupported data type of value {:?}", self)
-            }
+            Value::DateTime(_) => ConcreteDataType::date_datatype(),
         }
     }
 
@@ -153,7 +151,7 @@ impl TryFrom<Value> for serde_json::Value {
             Value::String(bytes) => serde_json::Value::String(bytes.as_utf8().to_string()),
             Value::Binary(bytes) => serde_json::to_value(bytes)?,
             Value::Date(v) => serde_json::Value::Number(v.val().into()),
-            Value::DateTime(v) => serde_json::Value::Number(v.into()),
+            Value::DateTime(v) => serde_json::Value::Number(v.val().into()),
             Value::List(v) => serde_json::to_value(v)?,
         };
 
@@ -205,6 +203,8 @@ impl Ord for ListValue {
 
 #[cfg(test)]
 mod tests {
+    use common_time::datetime::DateTime;
+
     use super::*;
 
     #[test]
@@ -410,7 +410,7 @@ mod tests {
         );
         assert_eq!(
             serde_json::Value::Number(5000i64.into()),
-            to_json(Value::DateTime(5000))
+            to_json(Value::DateTime(DateTime::try_new(5000).unwrap()))
         );
 
         let json_value: serde_json::Value =
diff --git a/src/datatypes/src/vectors.rs b/src/datatypes/src/vectors.rs
index d451ccf4e2..660151b45e 100644
--- a/src/datatypes/src/vectors.rs
+++ b/src/datatypes/src/vectors.rs
@@ -3,6 +3,7 @@ pub mod boolean;
 mod builder;
 pub mod constant;
 pub mod date;
+pub mod datetime;
 mod helper;
 mod list;
 pub mod mutable;
diff --git a/src/datatypes/src/vectors/builder.rs b/src/datatypes/src/vectors/builder.rs
index c4983f9149..2f379ee3aa 100644
--- a/src/datatypes/src/vectors/builder.rs
+++ b/src/datatypes/src/vectors/builder.rs
@@ -1,11 +1,13 @@
 use std::sync::Arc;
 
 use common_time::date::Date;
+use common_time::datetime::DateTime;
 
 use crate::data_type::ConcreteDataType;
 use crate::scalars::ScalarVectorBuilder;
 use crate::value::Value;
 use crate::vectors::date::DateVectorBuilder;
+use crate::vectors::datetime::DateTimeVectorBuilder;
 use crate::vectors::{
     BinaryVectorBuilder, BooleanVectorBuilder, Float32VectorBuilder, Float64VectorBuilder,
     Int16VectorBuilder, Int32VectorBuilder, Int64VectorBuilder, Int8VectorBuilder, MutableVector,
@@ -34,6 +36,7 @@ pub enum VectorBuilder {
     Binary(BinaryVectorBuilder),
 
     Date(DateVectorBuilder),
+    DateTime(DateTimeVectorBuilder),
 }
 
 impl VectorBuilder {
@@ -86,6 +89,9 @@ impl VectorBuilder {
             ConcreteDataType::Date(_) => {
                 VectorBuilder::Date(DateVectorBuilder::with_capacity(capacity))
             }
+            ConcreteDataType::DateTime(_) => {
+                VectorBuilder::DateTime(DateTimeVectorBuilder::with_capacity(capacity))
+            }
             _ => unimplemented!(),
         }
     }
@@ -107,6 +113,7 @@ impl VectorBuilder {
             VectorBuilder::String(b) => b.data_type(),
             VectorBuilder::Binary(b) => b.data_type(),
             VectorBuilder::Date(b) => b.data_type(),
+            VectorBuilder::DateTime(b) => b.data_type(),
         }
     }
 
@@ -132,6 +139,10 @@ impl VectorBuilder {
             (VectorBuilder::Binary(b), Value::Binary(v)) => b.push(Some(v)),
             (VectorBuilder::Date(b), Value::Date(v)) => b.push(Some(*v)),
             (VectorBuilder::Date(b), Value::Int32(v)) => b.push(Some(Date::try_new(*v).unwrap())),
+            (VectorBuilder::DateTime(b), Value::DateTime(v)) => b.push(Some(*v)),
+            (VectorBuilder::DateTime(b), Value::Int64(v)) => {
+                b.push(Some(DateTime::try_new(*v).unwrap()))
+            }
             _ => panic!(
                 "Value {:?} does not match builder type {:?}",
                 value,
@@ -157,6 +168,7 @@ impl VectorBuilder {
             VectorBuilder::String(b) => b.push(None),
             VectorBuilder::Binary(b) => b.push(None),
             VectorBuilder::Date(b) => b.push(None),
+            VectorBuilder::DateTime(b) => b.push(None),
         }
     }
 
@@ -177,6 +189,7 @@ impl VectorBuilder {
             VectorBuilder::String(b) => Arc::new(b.finish()),
             VectorBuilder::Binary(b) => Arc::new(b.finish()),
             VectorBuilder::Date(b) => Arc::new(b.finish()),
+            VectorBuilder::DateTime(b) => Arc::new(b.finish()),
         }
     }
 }
@@ -188,6 +201,7 @@ mod tests {
     use super::*;
     use crate::prelude::Vector;
     use crate::vectors::date::DateVector;
+    use crate::vectors::datetime::DateTimeVector;
 
     macro_rules! impl_integer_builder_test {
         ($Type: ident, $datatype: ident) => {
@@ -288,4 +302,20 @@ mod tests {
             v.to_arrow_array().data_type()
         );
     }
+
+    #[test]
+    pub fn test_datetime_vector_builder() {
+        let mut builder = VectorBuilder::with_capacity(ConcreteDataType::datetime_datatype(), 3);
+        assert_eq!(ConcreteDataType::datetime_datatype(), builder.data_type());
+        builder.push_null();
+        builder.push(&Value::DateTime(DateTime::try_new(123).unwrap()));
+        let v = builder.finish();
+        let v = v.as_any().downcast_ref::<DateTimeVector>().unwrap();
+        assert_eq!(Value::Null, v.get(0));
+        assert_eq!(Value::DateTime(DateTime::try_new(123).unwrap()), v.get(1));
+        assert_eq!(
+            &arrow::datatypes::DataType::Date64,
+            v.to_arrow_array().data_type()
+        );
+    }
 }
diff --git a/src/datatypes/src/vectors/datetime.rs b/src/datatypes/src/vectors/datetime.rs
new file mode 100644
index 0000000000..1f51775c5e
--- /dev/null
+++ b/src/datatypes/src/vectors/datetime.rs
@@ -0,0 +1,243 @@
+use std::any::Any;
+use std::sync::Arc;
+
+use arrow::array::{Array, ArrayRef, PrimitiveArray};
+use common_time::datetime::DateTime;
+use snafu::OptionExt;
+
+use crate::data_type::ConcreteDataType;
+use crate::error::ConversionSnafu;
+use crate::prelude::{
+    MutableVector, ScalarVector, ScalarVectorBuilder, Validity, Value, Vector, VectorRef,
+};
+use crate::serialize::Serializable;
+use crate::vectors::{PrimitiveIter, PrimitiveVector, PrimitiveVectorBuilder};
+
+#[derive(Debug, Clone)]
+pub struct DateTimeVector {
+    array: PrimitiveVector<i64>,
+}
+
+impl DateTimeVector {
+    pub fn new(array: PrimitiveArray<i64>) -> Self {
+        Self {
+            array: PrimitiveVector { array },
+        }
+    }
+
+    pub fn try_from_arrow_array(array: impl AsRef<dyn Array>) -> crate::error::Result<Self> {
+        Ok(Self::new(
+            array
+                .as_ref()
+                .as_any()
+                .downcast_ref::<PrimitiveArray<i64>>()
+                .with_context(|| ConversionSnafu {
+                    from: format!("{:?}", array.as_ref().data_type()),
+                })?
+                .clone(),
+        ))
+    }
+}
+
+impl Vector for DateTimeVector {
+    fn data_type(&self) -> ConcreteDataType {
+        ConcreteDataType::datetime_datatype()
+    }
+
+    fn vector_type_name(&self) -> String {
+        "DateTimeVector".to_string()
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn len(&self) -> usize {
+        self.array.len()
+    }
+
+    fn to_arrow_array(&self) -> ArrayRef {
+        let validity = self.array.array.validity().cloned();
+        let buffer = self.array.array.values().clone();
+        Arc::new(PrimitiveArray::new(
+            arrow::datatypes::DataType::Date64,
+            buffer,
+            validity,
+        ))
+    }
+
+    fn validity(&self) -> Validity {
+        self.array.validity()
+    }
+
+    fn memory_size(&self) -> usize {
+        self.array.memory_size()
+    }
+
+    fn is_null(&self, row: usize) -> bool {
+        self.array.is_null(row)
+    }
+
+    fn slice(&self, offset: usize, length: usize) -> VectorRef {
+        self.array.slice(offset, length)
+    }
+
+    fn get(&self, index: usize) -> Value {
+        match self.array.get(index) {
+            Value::Int64(v) => {
+                Value::DateTime(DateTime::try_new(v).expect("Not expected to overflow here"))
+            }
+            Value::Null => Value::Null,
+            _ => {
+                unreachable!()
+            }
+        }
+    }
+
+    fn replicate(&self, offsets: &[usize]) -> VectorRef {
+        self.array.replicate(offsets)
+    }
+}
+
+impl Serializable for DateTimeVector {
+    fn serialize_to_json(&self) -> crate::Result<Vec<serde_json::Value>> {
+        Ok(self
+            .array
+            .iter_data()
+            .map(|v| v.map(|d| DateTime::try_new(d).unwrap()))
+            .map(|v| match v {
+                None => serde_json::Value::Null,
+                Some(v) => v.into(),
+            })
+            .collect::<Vec<_>>())
+    }
+}
+
+pub struct DateTimeVectorBuilder {
+    buffer: PrimitiveVectorBuilder<i64>,
+}
+
+impl ScalarVectorBuilder for DateTimeVectorBuilder {
+    type VectorType = DateTimeVector;
+
+    fn with_capacity(capacity: usize) -> Self {
+        Self {
+            buffer: PrimitiveVectorBuilder::with_capacity(capacity),
+        }
+    }
+
+    fn push(&mut self, value: Option<<Self::VectorType as ScalarVector>::RefItem<'_>>) {
+        self.buffer.push(value.map(|d| d.val()))
+    }
+
+    fn finish(&mut self) -> Self::VectorType {
+        Self::VectorType {
+            array: self.buffer.finish(),
+        }
+    }
+}
+
+impl MutableVector for DateTimeVectorBuilder {
+    fn data_type(&self) -> ConcreteDataType {
+        ConcreteDataType::datetime_datatype()
+    }
+
+    fn len(&self) -> usize {
+        self.buffer.len()
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn as_mut_any(&mut self) -> &mut dyn Any {
+        self
+    }
+
+    fn to_vector(&mut self) -> VectorRef {
+        Arc::new(self.finish())
+    }
+}
+
+pub struct DateTimeIter<'a> {
+    iter: PrimitiveIter<'a, i64>,
+}
+
+impl<'a> Iterator for DateTimeIter<'a> {
+    type Item = Option<DateTime>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.iter
+            .next()
+            .map(|v| v.map(|v| DateTime::try_new(v).unwrap()))
+    }
+}
+
+impl ScalarVector for DateTimeVector {
+    type OwnedItem = DateTime;
+    type RefItem<'a> = DateTime;
+    type Iter<'a> = DateTimeIter<'a>;
+    type Builder = DateTimeVectorBuilder;
+
+    fn get_data(&self, idx: usize) -> Option<Self::RefItem<'_>> {
+        self.array
+            .get_data(idx)
+            .map(|v| DateTime::try_new(v).unwrap())
+    }
+
+    fn iter_data(&self) -> Self::Iter<'_> {
+        DateTimeIter {
+            iter: self.array.iter_data(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::assert_matches::assert_matches;
+
+    use super::*;
+
+    #[test]
+    pub fn test_datetime_vector() {
+        let v = DateTimeVector::new(PrimitiveArray::from_vec(vec![1, 2, 3]));
+        assert_eq!(ConcreteDataType::datetime_datatype(), v.data_type());
+        assert_eq!(3, v.len());
+        assert_eq!("DateTimeVector", v.vector_type_name());
+        assert_eq!(
+            &arrow::datatypes::DataType::Date64,
+            v.to_arrow_array().data_type()
+        );
+        let mut iter = v.iter_data();
+        assert_eq!(Some(DateTime::try_new(1).unwrap()), iter.next().unwrap());
+        assert_eq!(Some(DateTime::try_new(2).unwrap()), iter.next().unwrap());
+        assert_eq!(Some(DateTime::try_new(3).unwrap()), iter.next().unwrap());
+        assert!(!v.is_null(0));
+        assert_eq!(24, v.memory_size()); // size of i64 * 3
+
+        assert_matches!(v.validity(), Validity::AllValid);
+        if let Value::DateTime(d) = v.get(0) {
+            assert_eq!(1, d.val());
+        } else {
+            unreachable!()
+        }
+        assert_eq!(
+            "[\"1970-01-01 00:00:01\",\"1970-01-01 00:00:02\",\"1970-01-01 00:00:03\"]",
+            serde_json::to_string(&v.serialize_to_json().unwrap()).unwrap()
+        );
+    }
+
+    #[test]
+    pub fn test_datetime_vector_builder() {
+        let mut builder = DateTimeVectorBuilder::with_capacity(3);
+        builder.push(Some(DateTime::try_new(1).unwrap()));
+        builder.push(None);
+        builder.push(Some(DateTime::try_new(-1).unwrap()));
+
+        let v = builder.finish();
+        assert_eq!(ConcreteDataType::datetime_datatype(), v.data_type());
+        assert_eq!(Value::DateTime(DateTime::try_new(1).unwrap()), v.get(0));
+        assert_eq!(Value::Null, v.get(1));
+        assert_eq!(Value::DateTime(DateTime::try_new(-1).unwrap()), v.get(2));
+    }
+}
diff --git a/src/datatypes/src/vectors/helper.rs b/src/datatypes/src/vectors/helper.rs
index 191f138ef6..cc2bcd790e 100644
--- a/src/datatypes/src/vectors/helper.rs
+++ b/src/datatypes/src/vectors/helper.rs
@@ -11,6 +11,7 @@ use snafu::OptionExt;
 use crate::error::{ConversionSnafu, Result, UnknownVectorSnafu};
 use crate::scalars::*;
 use crate::vectors::date::DateVector;
+use crate::vectors::datetime::DateTimeVector;
 use crate::vectors::*;
 
 pub struct Helper;
@@ -177,6 +178,7 @@ impl Helper {
                 Arc::new(StringVector::try_from_arrow_array(array)?)
             }
             ArrowDataType::Date32 => Arc::new(DateVector::try_from_arrow_array(array)?),
+            ArrowDataType::Date64 => Arc::new(DateTimeVector::try_from_arrow_array(array)?),
             ArrowDataType::List(_) => Arc::new(ListVector::try_from_arrow_array(array)?),
             _ => unimplemented!("Arrow array datatype: {:?}", array.as_ref().data_type()),
         })
diff --git a/src/servers/src/mysql/writer.rs b/src/servers/src/mysql/writer.rs
index cc973d1735..4e365286ac 100644
--- a/src/servers/src/mysql/writer.rs
+++ b/src/servers/src/mysql/writer.rs
@@ -102,7 +102,7 @@ impl<'a, W: io::Write> MysqlResultWriter<'a, W> {
                     Value::String(v) => row_writer.write_col(v.as_utf8())?,
                     Value::Binary(v) => row_writer.write_col(v.deref())?,
                     Value::Date(v) => row_writer.write_col(v.val())?,
-                    Value::DateTime(v) => row_writer.write_col(v)?,
+                    Value::DateTime(v) => row_writer.write_col(v.val())?,
                     Value::List(_) => {
                         return Err(Error::Internal {
                             err_msg: format!(