feat(query): support timezone

common/datavalues/src/types/data_type.rs

@@ -15,6 +15,7 @@
 use std::any::Any;
 use std::collections::BTreeMap;
 
+use chrono_tz::Tz;
 use common_arrow::arrow::datatypes::DataType as ArrowType;
 use common_arrow::arrow::datatypes::Field as ArrowField;
 use common_exception::Result;
@@ -125,7 +126,15 @@ pub trait DataType: std::fmt::Debug + Sync + Send + DynClone {
 
     fn create_mutable(&self, capacity: usize) -> Box<dyn MutableColumn>;
     fn create_serializer(&self) -> TypeSerializerImpl;
+    /// work only for timestamp serializer
+    fn create_serializer_with_tz(&self, _tz: Tz) -> TypeSerializerImpl {
+        unimplemented!()
+    }
     fn create_deserializer(&self, capacity: usize) -> TypeDeserializerImpl;
+    /// work only for timestamp deserializer
+    fn create_deserializer_with_tz(&self, _capacity: usize, _tz: Tz) -> TypeDeserializerImpl {
+        unimplemented!()
+    }
 }
 
 pub fn from_arrow_type(dt: &ArrowType) -> DataTypeImpl {

common/datavalues/src/types/serializations/array.rs

@@ -14,6 +14,7 @@
 
 use common_exception::ErrorCode;
 use common_exception::Result;
+use common_io::prelude::FormatSettings;
 use serde_json::Value;
 
 use crate::prelude::*;
@@ -25,7 +26,7 @@ pub struct ArraySerializer {
 }
 
 impl TypeSerializer for ArraySerializer {
-    fn serialize_value(&self, value: &DataValue) -> Result<String> {
+    fn serialize_value(&self, value: &DataValue, format: &FormatSettings) -> Result<String> {
         if let DataValue::Array(vals) = value {
             let mut res = String::new();
             res.push('[');
@@ -37,7 +38,7 @@ impl TypeSerializer for ArraySerializer {
                 }
                 first = false;
 
-                let s = self.inner.serialize_value(val)?;
+                let s = self.inner.serialize_value(val, format)?;
                 if quoted {
                     res.push_str(&format!("'{}'", s));
                 } else {
@@ -51,24 +52,25 @@ impl TypeSerializer for ArraySerializer {
         }
     }
 
-    fn serialize_column(&self, column: &ColumnRef) -> Result<Vec<String>> {
+    fn serialize_column(&self, column: &ColumnRef, format: &FormatSettings) -> Result<Vec<String>> {
         let column: &ArrayColumn = Series::check_get(column)?;
         let mut result = Vec::with_capacity(column.len());
         for i in 0..column.len() {
             let val = column.get(i);
-            let s = self.serialize_value(&val)?;
+            let s = self.serialize_value(&val, format)?;
             result.push(s);
         }
         Ok(result)
     }
 
-    fn serialize_json(&self, _column: &ColumnRef) -> Result<Vec<Value>> {
+    fn serialize_json(&self, _column: &ColumnRef, _format: &FormatSettings) -> Result<Vec<Value>> {
         todo!()
     }
 
     fn serialize_clickhouse_format(
         &self,
         _column: &ColumnRef,
+        _format: &FormatSettings,
     ) -> Result<opensrv_clickhouse::types::column::ArcColumnData> {
         todo!()
     }

common/datavalues/src/types/serializations/boolean.rs

@@ -15,6 +15,7 @@
 use common_arrow::arrow::bitmap::Bitmap;
 use common_exception::ErrorCode;
 use common_exception::Result;
+use common_io::prelude::FormatSettings;
 use opensrv_clickhouse::types::column::ArcColumnWrapper;
 use opensrv_clickhouse::types::column::ColumnFrom;
 use serde_json::Value;
@@ -28,7 +29,7 @@ const TRUE_STR: &str = "1";
 const FALSE_STR: &str = "0";
 
 impl TypeSerializer for BooleanSerializer {
-    fn serialize_value(&self, value: &DataValue) -> Result<String> {
+    fn serialize_value(&self, value: &DataValue, _format: &FormatSettings) -> Result<String> {
         if let DataValue::Boolean(x) = value {
             if *x {
                 Ok(TRUE_STR.to_owned())
@@ -40,7 +41,11 @@ impl TypeSerializer for BooleanSerializer {
         }
     }
 
-    fn serialize_column(&self, column: &ColumnRef) -> Result<Vec<String>> {
+    fn serialize_column(
+        &self,
+        column: &ColumnRef,
+        _format: &FormatSettings,
+    ) -> Result<Vec<String>> {
         let array: &BooleanColumn = Series::check_get(column)?;
 
         let result: Vec<String> = array
@@ -56,7 +61,7 @@ impl TypeSerializer for BooleanSerializer {
         Ok(result)
     }
 
-    fn serialize_json(&self, column: &ColumnRef) -> Result<Vec<Value>> {
+    fn serialize_json(&self, column: &ColumnRef, _format: &FormatSettings) -> Result<Vec<Value>> {
         let array: &BooleanColumn = Series::check_get(column)?;
         let result: Vec<Value> = array
             .iter()
@@ -68,6 +73,7 @@ impl TypeSerializer for BooleanSerializer {
     fn serialize_clickhouse_format(
         &self,
         column: &ColumnRef,
+        _format: &FormatSettings,
     ) -> Result<opensrv_clickhouse::types::column::ArcColumnData> {
         let col: &BooleanColumn = Series::check_get(column)?;
         let values: Vec<u8> = col.iter().map(|c| c as u8).collect();
@@ -78,13 +84,15 @@ impl TypeSerializer for BooleanSerializer {
         &self,
         column: &ColumnRef,
         _valids: Option<&Bitmap>,
+        format: &FormatSettings,
     ) -> Result<Vec<Value>> {
-        self.serialize_json(column)
+        self.serialize_json(column, format)
     }
 
     fn serialize_json_object_suppress_error(
         &self,
         column: &ColumnRef,
+        _format: &FormatSettings,
     ) -> Result<Vec<Option<Value>>> {
         let column: &BooleanColumn = Series::check_get(column)?;
         let result: Vec<Option<Value>> = column

common/datavalues/src/types/serializations/date.rs

@@ -11,7 +11,6 @@
 // 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 std::marker::PhantomData;
 use std::ops::AddAssign;
 
@@ -20,6 +19,7 @@ use chrono::Duration;
 use chrono::NaiveDate;
 use chrono_tz::Tz;
 use common_exception::*;
+use common_io::prelude::FormatSettings;
 use num::cast::AsPrimitive;
 use opensrv_clickhouse::types::column::ArcColumnWrapper;
 use opensrv_clickhouse::types::column::ColumnFrom;
@@ -43,14 +43,18 @@ impl<T: PrimitiveType + AsPrimitive<i64>> Default for DateSerializer<T> {
 const DATE_FMT: &str = "%Y-%m-%d";
 
 impl<T: PrimitiveType + AsPrimitive<i64>> TypeSerializer for DateSerializer<T> {
-    fn serialize_value(&self, value: &DataValue) -> Result<String> {
+    fn serialize_value(&self, value: &DataValue, _format: &FormatSettings) -> Result<String> {
         let mut date = NaiveDate::from_ymd(1970, 1, 1);
         let d = Duration::days(value.as_i64()?);
         date.add_assign(d);
         Ok(date.format(DATE_FMT).to_string())
     }
 
-    fn serialize_column(&self, column: &ColumnRef) -> Result<Vec<String>> {
+    fn serialize_column(
+        &self,
+        column: &ColumnRef,
+        _format: &FormatSettings,
+    ) -> Result<Vec<String>> {
         let column: &PrimitiveColumn<T> = Series::check_get(column)?;
 
         let result: Vec<String> = column
@@ -65,7 +69,7 @@ impl<T: PrimitiveType + AsPrimitive<i64>> TypeSerializer for DateSerializer<T> {
         Ok(result)
     }
 
-    fn serialize_json(&self, column: &ColumnRef) -> Result<Vec<Value>> {
+    fn serialize_json(&self, column: &ColumnRef, _format: &FormatSettings) -> Result<Vec<Value>> {
         let array: &PrimitiveColumn<T> = Series::check_get(column)?;
         let result: Vec<Value> = array
             .iter()
@@ -83,6 +87,7 @@ impl<T: PrimitiveType + AsPrimitive<i64>> TypeSerializer for DateSerializer<T> {
     fn serialize_clickhouse_format(
         &self,
         column: &ColumnRef,
+        _format: &FormatSettings,
     ) -> Result<opensrv_clickhouse::types::column::ArcColumnData> {
         let array: &PrimitiveColumn<T> = Series::check_get(column)?;
         let tz: Tz = "UTC".parse().unwrap();

common/datavalues/src/types/serializations/mod.rs

@@ -15,6 +15,7 @@
 use common_arrow::arrow::bitmap::Bitmap;
 use common_exception::ErrorCode;
 use common_exception::Result;
+use common_io::prelude::FormatSettings;
 use enum_dispatch::enum_dispatch;
 use opensrv_clickhouse::types::column::ArcColumnData;
 use serde_json::Value;
@@ -44,15 +45,20 @@ pub use variant::*;
 
 #[enum_dispatch]
 pub trait TypeSerializer: Send + Sync {
-    fn serialize_value(&self, value: &DataValue) -> Result<String>;
-    fn serialize_json(&self, column: &ColumnRef) -> Result<Vec<Value>>;
-    fn serialize_column(&self, column: &ColumnRef) -> Result<Vec<String>>;
-    fn serialize_clickhouse_format(&self, column: &ColumnRef) -> Result<ArcColumnData>;
+    fn serialize_value(&self, value: &DataValue, format: &FormatSettings) -> Result<String>;
+    fn serialize_json(&self, column: &ColumnRef, format: &FormatSettings) -> Result<Vec<Value>>;
+    fn serialize_column(&self, column: &ColumnRef, format: &FormatSettings) -> Result<Vec<String>>;
+    fn serialize_clickhouse_format(
+        &self,
+        column: &ColumnRef,
+        _format: &FormatSettings,
+    ) -> Result<ArcColumnData>;
 
     fn serialize_json_object(
         &self,
         _column: &ColumnRef,
         _valids: Option<&Bitmap>,
+        _format: &FormatSettings,
     ) -> Result<Vec<Value>> {
         Err(ErrorCode::BadDataValueType(
             "Error parsing JSON: unsupported data type",
@@ -62,6 +68,7 @@ pub trait TypeSerializer: Send + Sync {
     fn serialize_json_object_suppress_error(
         &self,
         _column: &ColumnRef,
+        _format: &FormatSettings,
     ) -> Result<Vec<Option<Value>>> {
         Err(ErrorCode::BadDataValueType(
             "Error parsing JSON: unsupported data type",

common/datavalues/src/types/serializations/null.rs

@@ -15,6 +15,7 @@
 use std::sync::Arc;
 
 use common_exception::Result;
+use common_io::prelude::FormatSettings;
 use opensrv_clickhouse::types::column::ArcColumnWrapper;
 use opensrv_clickhouse::types::column::ColumnFrom;
 use opensrv_clickhouse::types::column::NullableColumnData;
@@ -30,16 +31,20 @@ pub struct NullSerializer {}
 const NULL_STR: &str = "NULL";
 
 impl TypeSerializer for NullSerializer {
-    fn serialize_value(&self, _value: &DataValue) -> Result<String> {
+    fn serialize_value(&self, _value: &DataValue, _format: &FormatSettings) -> Result<String> {
         Ok(NULL_STR.to_owned())
     }
 
-    fn serialize_column(&self, column: &ColumnRef) -> Result<Vec<String>> {
+    fn serialize_column(
+        &self,
+        column: &ColumnRef,
+        _format: &FormatSettings,
+    ) -> Result<Vec<String>> {
         let result: Vec<String> = vec![NULL_STR.to_owned(); column.len()];
         Ok(result)
     }
 
-    fn serialize_json(&self, column: &ColumnRef) -> Result<Vec<Value>> {
+    fn serialize_json(&self, column: &ColumnRef, _format: &FormatSettings) -> Result<Vec<Value>> {
         let null = Value::Null;
         let result: Vec<Value> = vec![null; column.len()];
         Ok(result)
@@ -48,6 +53,7 @@ impl TypeSerializer for NullSerializer {
     fn serialize_clickhouse_format(
         &self,
         column: &ColumnRef,
+        _format: &FormatSettings,
     ) -> Result<opensrv_clickhouse::types::column::ArcColumnData> {
         let nulls = vec![1u8; column.len()];
         let inner = Vec::column_from::<ArcColumnWrapper>(vec![1u8; column.len()]);

common/datavalues/src/types/serializations/nullable.rs

@@ -15,6 +15,7 @@
 use std::sync::Arc;
 
 use common_exception::Result;
+use common_io::prelude::FormatSettings;
 use opensrv_clickhouse::types::column::NullableColumnData;
 use serde_json::Value;
 
@@ -32,18 +33,18 @@ pub struct NullableSerializer {
 }
 
 impl TypeSerializer for NullableSerializer {
-    fn serialize_value(&self, value: &DataValue) -> Result<String> {
+    fn serialize_value(&self, value: &DataValue, format: &FormatSettings) -> Result<String> {
         if value.is_null() {
             Ok("NULL".to_owned())
         } else {
-            self.inner.serialize_value(value)
+            self.inner.serialize_value(value, format)
         }
     }
 
-    fn serialize_column(&self, column: &ColumnRef) -> Result<Vec<String>> {
+    fn serialize_column(&self, column: &ColumnRef, format: &FormatSettings) -> Result<Vec<String>> {
         let column: &NullableColumn = Series::check_get(column)?;
         let rows = column.len();
-        let mut res = self.inner.serialize_column(column.inner())?;
+        let mut res = self.inner.serialize_column(column.inner(), format)?;
 
         (0..rows).for_each(|row| {
             if column.null_at(row) {
@@ -53,10 +54,10 @@ impl TypeSerializer for NullableSerializer {
         Ok(res)
     }
 
-    fn serialize_json(&self, column: &ColumnRef) -> Result<Vec<Value>> {
+    fn serialize_json(&self, column: &ColumnRef, format: &FormatSettings) -> Result<Vec<Value>> {
         let column: &NullableColumn = Series::check_get(column)?;
         let rows = column.len();
-        let mut res = self.inner.serialize_json(column.inner())?;
+        let mut res = self.inner.serialize_json(column.inner(), format)?;
 
         (0..rows).for_each(|row| {
             if column.null_at(row) {
@@ -69,9 +70,12 @@ impl TypeSerializer for NullableSerializer {
     fn serialize_clickhouse_format(
         &self,
         column: &ColumnRef,
+        format: &FormatSettings,
     ) -> Result<opensrv_clickhouse::types::column::ArcColumnData> {
         let column: &NullableColumn = Series::check_get(column)?;
-        let inner = self.inner.serialize_clickhouse_format(column.inner())?;
+        let inner = self
+            .inner
+            .serialize_clickhouse_format(column.inner(), format)?;
         let nulls = column.ensure_validity().iter().map(|v| !v as u8).collect();
         let data = NullableColumnData { nulls, inner };