mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-05-22 07:50:38 +00:00
feat: implement DateTime type (#198)
* feat: implement DateTime type * add some tests * Update src/common/time/src/datetime.rs Co-authored-by: Ning Sun <sunng@protonmail.com> * Update src/common/time/src/datetime.rs Co-authored-by: Ning Sun <sunng@protonmail.com>
This commit is contained in:
@@ -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)?,
|
||||
})
|
||||
|
||||
95
src/common/time/src/datetime.rs
Normal file
95
src/common/time/src/datetime.rs
Normal file
@@ -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());
|
||||
}
|
||||
}
|
||||
@@ -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>;
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
pub mod date;
|
||||
pub mod datetime;
|
||||
pub mod error;
|
||||
pub mod range;
|
||||
pub mod timestamp;
|
||||
|
||||
@@ -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(),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
#![feature(generic_associated_types)]
|
||||
#![feature(assert_matches)]
|
||||
|
||||
pub mod arrow_array;
|
||||
pub mod data_type;
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
|
||||
41
src/datatypes/src/types/datetime.rs
Normal file
41
src/datatypes/src/types/datetime.rs
Normal file
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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 =
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
243
src/datatypes/src/vectors/datetime.rs
Normal file
243
src/datatypes/src/vectors/datetime.rs
Normal file
@@ -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));
|
||||
}
|
||||
}
|
||||
@@ -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()),
|
||||
})
|
||||
|
||||
@@ -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!(
|
||||
|
||||
Reference in New Issue
Block a user