use std::any::Any;
use std::fmt;
use std::hash::{Hash, Hasher};
use std::sync::Arc;
use crate::intervals::Interval;
use crate::physical_expr::down_cast_any_ref;
use crate::sort_properties::SortProperties;
use crate::PhysicalExpr;
use arrow::compute;
use arrow::compute::{kernels, CastOptions};
use arrow::datatypes::{DataType, Schema};
use arrow::record_batch::RecordBatch;
use compute::can_cast_types;
use datafusion_common::format::DEFAULT_FORMAT_OPTIONS;
use datafusion_common::{not_impl_err, DataFusionError, Result, ScalarValue};
use datafusion_expr::ColumnarValue;
const DEFAULT_CAST_OPTIONS: CastOptions<'static> = CastOptions {
safe: false,
format_options: DEFAULT_FORMAT_OPTIONS,
};
#[derive(Debug, Clone)]
pub struct CastExpr {
expr: Arc<dyn PhysicalExpr>,
cast_type: DataType,
cast_options: CastOptions<'static>,
}
impl CastExpr {
pub fn new(
expr: Arc<dyn PhysicalExpr>,
cast_type: DataType,
cast_options: Option<CastOptions<'static>>,
) -> Self {
Self {
expr,
cast_type,
cast_options: cast_options.unwrap_or(DEFAULT_CAST_OPTIONS),
}
}
pub fn expr(&self) -> &Arc<dyn PhysicalExpr> {
&self.expr
}
pub fn cast_type(&self) -> &DataType {
&self.cast_type
}
}
impl fmt::Display for CastExpr {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "CAST({} AS {:?})", self.expr, self.cast_type)
}
}
impl PhysicalExpr for CastExpr {
fn as_any(&self) -> &dyn Any {
self
}
fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
Ok(self.cast_type.clone())
}
fn nullable(&self, input_schema: &Schema) -> Result<bool> {
self.expr.nullable(input_schema)
}
fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
let value = self.expr.evaluate(batch)?;
cast_column(&value, &self.cast_type, Some(&self.cast_options))
}
fn children(&self) -> Vec<Arc<dyn PhysicalExpr>> {
vec![self.expr.clone()]
}
fn with_new_children(
self: Arc<Self>,
children: Vec<Arc<dyn PhysicalExpr>>,
) -> Result<Arc<dyn PhysicalExpr>> {
Ok(Arc::new(CastExpr::new(
children[0].clone(),
self.cast_type.clone(),
Some(self.cast_options.clone()),
)))
}
fn evaluate_bounds(&self, children: &[&Interval]) -> Result<Interval> {
children[0].cast_to(&self.cast_type, &self.cast_options)
}
fn propagate_constraints(
&self,
interval: &Interval,
children: &[&Interval],
) -> Result<Vec<Option<Interval>>> {
let child_interval = children[0];
let cast_type = child_interval.get_datatype()?;
Ok(vec![Some(
interval.cast_to(&cast_type, &self.cast_options)?,
)])
}
fn dyn_hash(&self, state: &mut dyn Hasher) {
let mut s = state;
self.expr.hash(&mut s);
self.cast_type.hash(&mut s);
}
fn get_ordering(&self, children: &[SortProperties]) -> SortProperties {
children[0]
}
}
impl PartialEq<dyn Any> for CastExpr {
fn eq(&self, other: &dyn Any) -> bool {
down_cast_any_ref(other)
.downcast_ref::<Self>()
.map(|x| {
self.expr.eq(&x.expr)
&& self.cast_type == x.cast_type
&& self.cast_options.safe == x.cast_options.safe
})
.unwrap_or(false)
}
}
pub fn cast_column(
value: &ColumnarValue,
cast_type: &DataType,
cast_options: Option<&CastOptions<'static>>,
) -> Result<ColumnarValue> {
let cast_options = cast_options.cloned().unwrap_or(DEFAULT_CAST_OPTIONS);
match value {
ColumnarValue::Array(array) => Ok(ColumnarValue::Array(
kernels::cast::cast_with_options(array, cast_type, &cast_options)?,
)),
ColumnarValue::Scalar(scalar) => {
let scalar_array = scalar.to_array();
let cast_array = kernels::cast::cast_with_options(
&scalar_array,
cast_type,
&cast_options,
)?;
let cast_scalar = ScalarValue::try_from_array(&cast_array, 0)?;
Ok(ColumnarValue::Scalar(cast_scalar))
}
}
}
pub fn cast_with_options(
expr: Arc<dyn PhysicalExpr>,
input_schema: &Schema,
cast_type: DataType,
cast_options: Option<CastOptions<'static>>,
) -> Result<Arc<dyn PhysicalExpr>> {
let expr_type = expr.data_type(input_schema)?;
if expr_type == cast_type {
Ok(expr.clone())
} else if can_cast_types(&expr_type, &cast_type) {
Ok(Arc::new(CastExpr::new(expr, cast_type, cast_options)))
} else {
not_impl_err!("Unsupported CAST from {expr_type:?} to {cast_type:?}")
}
}
pub fn cast(
expr: Arc<dyn PhysicalExpr>,
input_schema: &Schema,
cast_type: DataType,
) -> Result<Arc<dyn PhysicalExpr>> {
cast_with_options(expr, input_schema, cast_type, None)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::expressions::col;
use arrow::{
array::{
Array, Decimal128Array, Float32Array, Float64Array, Int16Array, Int32Array,
Int64Array, Int8Array, StringArray, Time64NanosecondArray,
TimestampNanosecondArray, UInt32Array,
},
datatypes::*,
};
use datafusion_common::Result;
macro_rules! generic_decimal_to_other_test_cast {
($DECIMAL_ARRAY:ident, $A_TYPE:expr, $TYPEARRAY:ident, $TYPE:expr, $VEC:expr,$CAST_OPTIONS:expr) => {{
let schema = Schema::new(vec![Field::new("a", $A_TYPE, true)]);
let batch = RecordBatch::try_new(
Arc::new(schema.clone()),
vec![Arc::new($DECIMAL_ARRAY)],
)?;
let expression =
cast_with_options(col("a", &schema)?, &schema, $TYPE, $CAST_OPTIONS)?;
assert_eq!(
format!("CAST(a@0 AS {:?})", $TYPE),
format!("{}", expression)
);
assert_eq!(expression.data_type(&schema)?, $TYPE);
let result = expression.evaluate(&batch)?.into_array(batch.num_rows());
assert_eq!(*result.data_type(), $TYPE);
let result = result
.as_any()
.downcast_ref::<$TYPEARRAY>()
.expect("failed to downcast");
for (i, x) in $VEC.iter().enumerate() {
match x {
Some(x) => assert_eq!(result.value(i), *x),
None => assert!(!result.is_valid(i)),
}
}
}};
}
macro_rules! generic_test_cast {
($A_ARRAY:ident, $A_TYPE:expr, $A_VEC:expr, $TYPEARRAY:ident, $TYPE:expr, $VEC:expr, $CAST_OPTIONS:expr) => {{
let schema = Schema::new(vec![Field::new("a", $A_TYPE, true)]);
let a_vec_len = $A_VEC.len();
let a = $A_ARRAY::from($A_VEC);
let batch =
RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?;
let expression =
cast_with_options(col("a", &schema)?, &schema, $TYPE, $CAST_OPTIONS)?;
assert_eq!(
format!("CAST(a@0 AS {:?})", $TYPE),
format!("{}", expression)
);
assert_eq!(expression.data_type(&schema)?, $TYPE);
let result = expression.evaluate(&batch)?.into_array(batch.num_rows());
assert_eq!(*result.data_type(), $TYPE);
assert_eq!(result.len(), a_vec_len);
let result = result
.as_any()
.downcast_ref::<$TYPEARRAY>()
.expect("failed to downcast");
for (i, x) in $VEC.iter().enumerate() {
match x {
Some(x) => assert_eq!(result.value(i), *x),
None => assert!(!result.is_valid(i)),
}
}
}};
}
#[test]
fn test_cast_decimal_to_decimal() -> Result<()> {
let array = vec![
Some(1234),
Some(2222),
Some(3),
Some(4000),
Some(5000),
None,
];
let decimal_array = array
.clone()
.into_iter()
.collect::<Decimal128Array>()
.with_precision_and_scale(10, 3)?;
generic_decimal_to_other_test_cast!(
decimal_array,
DataType::Decimal128(10, 3),
Decimal128Array,
DataType::Decimal128(20, 6),
[
Some(1_234_000),
Some(2_222_000),
Some(3_000),
Some(4_000_000),
Some(5_000_000),
None
],
None
);
let decimal_array = array
.into_iter()
.collect::<Decimal128Array>()
.with_precision_and_scale(10, 3)?;
generic_decimal_to_other_test_cast!(
decimal_array,
DataType::Decimal128(10, 3),
Decimal128Array,
DataType::Decimal128(10, 2),
[Some(123), Some(222), Some(0), Some(400), Some(500), None],
None
);
Ok(())
}
#[test]
fn test_cast_decimal_to_numeric() -> Result<()> {
let array = vec![Some(1), Some(2), Some(3), Some(4), Some(5), None];
let decimal_array = array
.clone()
.into_iter()
.collect::<Decimal128Array>()
.with_precision_and_scale(10, 0)?;
generic_decimal_to_other_test_cast!(
decimal_array,
DataType::Decimal128(10, 0),
Int8Array,
DataType::Int8,
[
Some(1_i8),
Some(2_i8),
Some(3_i8),
Some(4_i8),
Some(5_i8),
None
],
None
);
let decimal_array = array
.clone()
.into_iter()
.collect::<Decimal128Array>()
.with_precision_and_scale(10, 0)?;
generic_decimal_to_other_test_cast!(
decimal_array,
DataType::Decimal128(10, 0),
Int16Array,
DataType::Int16,
[
Some(1_i16),
Some(2_i16),
Some(3_i16),
Some(4_i16),
Some(5_i16),
None
],
None
);
let decimal_array = array
.clone()
.into_iter()
.collect::<Decimal128Array>()
.with_precision_and_scale(10, 0)?;
generic_decimal_to_other_test_cast!(
decimal_array,
DataType::Decimal128(10, 0),
Int32Array,
DataType::Int32,
[
Some(1_i32),
Some(2_i32),
Some(3_i32),
Some(4_i32),
Some(5_i32),
None
],
None
);
let decimal_array = array
.into_iter()
.collect::<Decimal128Array>()
.with_precision_and_scale(10, 0)?;
generic_decimal_to_other_test_cast!(
decimal_array,
DataType::Decimal128(10, 0),
Int64Array,
DataType::Int64,
[
Some(1_i64),
Some(2_i64),
Some(3_i64),
Some(4_i64),
Some(5_i64),
None
],
None
);
let array = vec![
Some(1234),
Some(2222),
Some(3),
Some(4000),
Some(5000),
None,
];
let decimal_array = array
.clone()
.into_iter()
.collect::<Decimal128Array>()
.with_precision_and_scale(10, 3)?;
generic_decimal_to_other_test_cast!(
decimal_array,
DataType::Decimal128(10, 3),
Float32Array,
DataType::Float32,
[
Some(1.234_f32),
Some(2.222_f32),
Some(0.003_f32),
Some(4.0_f32),
Some(5.0_f32),
None
],
None
);
let decimal_array = array
.into_iter()
.collect::<Decimal128Array>()
.with_precision_and_scale(20, 6)?;
generic_decimal_to_other_test_cast!(
decimal_array,
DataType::Decimal128(20, 6),
Float64Array,
DataType::Float64,
[
Some(0.001234_f64),
Some(0.002222_f64),
Some(0.000003_f64),
Some(0.004_f64),
Some(0.005_f64),
None
],
None
);
Ok(())
}
#[test]
fn test_cast_numeric_to_decimal() -> Result<()> {
generic_test_cast!(
Int8Array,
DataType::Int8,
vec![1, 2, 3, 4, 5],
Decimal128Array,
DataType::Decimal128(3, 0),
[Some(1), Some(2), Some(3), Some(4), Some(5)],
None
);
generic_test_cast!(
Int16Array,
DataType::Int16,
vec![1, 2, 3, 4, 5],
Decimal128Array,
DataType::Decimal128(5, 0),
[Some(1), Some(2), Some(3), Some(4), Some(5)],
None
);
generic_test_cast!(
Int32Array,
DataType::Int32,
vec![1, 2, 3, 4, 5],
Decimal128Array,
DataType::Decimal128(10, 0),
[Some(1), Some(2), Some(3), Some(4), Some(5)],
None
);
generic_test_cast!(
Int64Array,
DataType::Int64,
vec![1, 2, 3, 4, 5],
Decimal128Array,
DataType::Decimal128(20, 0),
[Some(1), Some(2), Some(3), Some(4), Some(5)],
None
);
generic_test_cast!(
Int64Array,
DataType::Int64,
vec![1, 2, 3, 4, 5],
Decimal128Array,
DataType::Decimal128(20, 2),
[Some(100), Some(200), Some(300), Some(400), Some(500)],
None
);
generic_test_cast!(
Float32Array,
DataType::Float32,
vec![1.5, 2.5, 3.0, 1.123_456_8, 5.50],
Decimal128Array,
DataType::Decimal128(10, 2),
[Some(150), Some(250), Some(300), Some(112), Some(550)],
None
);
generic_test_cast!(
Float64Array,
DataType::Float64,
vec![1.5, 2.5, 3.0, 1.123_456_8, 5.50],
Decimal128Array,
DataType::Decimal128(20, 4),
[
Some(15000),
Some(25000),
Some(30000),
Some(11235),
Some(55000)
],
None
);
Ok(())
}
#[test]
fn test_cast_i32_u32() -> Result<()> {
generic_test_cast!(
Int32Array,
DataType::Int32,
vec![1, 2, 3, 4, 5],
UInt32Array,
DataType::UInt32,
[
Some(1_u32),
Some(2_u32),
Some(3_u32),
Some(4_u32),
Some(5_u32)
],
None
);
Ok(())
}
#[test]
fn test_cast_i32_utf8() -> Result<()> {
generic_test_cast!(
Int32Array,
DataType::Int32,
vec![1, 2, 3, 4, 5],
StringArray,
DataType::Utf8,
[Some("1"), Some("2"), Some("3"), Some("4"), Some("5")],
None
);
Ok(())
}
#[test]
fn test_cast_i64_t64() -> Result<()> {
let original = vec![1, 2, 3, 4, 5];
let expected: Vec<Option<i64>> = original
.iter()
.map(|i| Some(Time64NanosecondArray::from(vec![*i]).value(0)))
.collect();
generic_test_cast!(
Int64Array,
DataType::Int64,
original,
TimestampNanosecondArray,
DataType::Timestamp(TimeUnit::Nanosecond, None),
expected,
None
);
Ok(())
}
#[test]
fn invalid_cast() {
let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
let result = cast(col("a", &schema).unwrap(), &schema, DataType::LargeBinary);
result.expect_err("expected Invalid CAST");
}
#[test]
fn invalid_cast_with_options_error() -> Result<()> {
let schema = Schema::new(vec![Field::new("a", DataType::Utf8, false)]);
let a = StringArray::from(vec!["9.1"]);
let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?;
let expression =
cast_with_options(col("a", &schema)?, &schema, DataType::Int32, None)?;
let result = expression.evaluate(&batch);
match result {
Ok(_) => panic!("expected error"),
Err(e) => {
assert!(e
.to_string()
.contains("Cannot cast string '9.1' to value of Int32 type"))
}
}
Ok(())
}
#[test]
#[ignore] fn test_cast_decimal() -> Result<()> {
let schema = Schema::new(vec![Field::new("a", DataType::Int64, false)]);
let a = Int64Array::from(vec![100]);
let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?;
let expression = cast_with_options(
col("a", &schema)?,
&schema,
DataType::Decimal128(38, 38),
None,
)?;
expression.evaluate(&batch)?;
Ok(())
}
}