1use std::fmt;
19use std::hash::Hash;
20use std::sync::Arc;
21
22use crate::physical_expr::PhysicalExpr;
23
24use arrow::compute::{CastOptions, can_cast_types};
25use arrow::datatypes::{DataType, DataType::*, FieldRef, Schema};
26use arrow::record_batch::RecordBatch;
27use datafusion_common::datatype::DataTypeExt;
28use datafusion_common::format::DEFAULT_FORMAT_OPTIONS;
29use datafusion_common::nested_struct::{
30 requires_nested_struct_cast, validate_data_type_compatibility,
31};
32use datafusion_common::{Result, not_impl_err};
33use datafusion_expr_common::columnar_value::ColumnarValue;
34use datafusion_expr_common::interval_arithmetic::Interval;
35use datafusion_expr_common::sort_properties::ExprProperties;
36
37const DEFAULT_CAST_OPTIONS: CastOptions<'static> = CastOptions {
38 safe: false,
39 format_options: DEFAULT_FORMAT_OPTIONS,
40};
41
42const DEFAULT_SAFE_CAST_OPTIONS: CastOptions<'static> = CastOptions {
43 safe: true,
44 format_options: DEFAULT_FORMAT_OPTIONS,
45};
46
47fn can_cast_named_struct_types(source: &DataType, target: &DataType) -> bool {
54 validate_data_type_compatibility("", source, target).is_ok()
55}
56
57#[derive(Debug, Clone, Eq)]
59pub struct CastExpr {
60 pub expr: Arc<dyn PhysicalExpr>,
62 target_field: FieldRef,
64 cast_options: CastOptions<'static>,
66}
67
68impl PartialEq for CastExpr {
70 fn eq(&self, other: &Self) -> bool {
71 self.expr.eq(&other.expr)
72 && self.target_field.eq(&other.target_field)
73 && self.cast_options.eq(&other.cast_options)
74 }
75}
76
77impl Hash for CastExpr {
78 fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
79 self.expr.hash(state);
80 self.target_field.hash(state);
81 self.cast_options.hash(state);
82 }
83}
84
85impl CastExpr {
86 pub fn new(
101 expr: Arc<dyn PhysicalExpr>,
102 cast_type: DataType,
103 cast_options: Option<CastOptions<'static>>,
104 ) -> Self {
105 Self::new_with_target_field(
106 expr,
107 cast_type.into_nullable_field_ref(),
108 cast_options,
109 )
110 }
111
112 pub fn new_with_target_field(
122 expr: Arc<dyn PhysicalExpr>,
123 target_field: FieldRef,
124 cast_options: Option<CastOptions<'static>>,
125 ) -> Self {
126 Self {
127 expr,
128 target_field,
129 cast_options: cast_options.unwrap_or(DEFAULT_CAST_OPTIONS),
130 }
131 }
132
133 pub fn expr(&self) -> &Arc<dyn PhysicalExpr> {
135 &self.expr
136 }
137
138 pub fn cast_type(&self) -> &DataType {
140 self.target_field.data_type()
141 }
142
143 pub fn target_field(&self) -> &FieldRef {
145 &self.target_field
146 }
147
148 pub fn cast_options(&self) -> &CastOptions<'static> {
150 &self.cast_options
151 }
152
153 fn resolved_target_field(&self, input_schema: &Schema) -> Result<FieldRef> {
154 if is_default_target_field(&self.target_field) {
155 self.expr.return_field(input_schema).map(|field| {
156 Arc::new(
157 field
158 .as_ref()
159 .clone()
160 .with_data_type(self.cast_type().clone()),
161 )
162 })
163 } else {
164 Ok(Arc::clone(&self.target_field))
165 }
166 }
167
168 pub fn check_bigger_cast(cast_type: &DataType, src: &DataType) -> bool {
171 if cast_type.eq(src) {
172 return true;
173 }
174 matches!(
175 (src, cast_type),
176 (Int8, Int16 | Int32 | Int64)
177 | (Int16, Int32 | Int64)
178 | (Int32, Int64)
179 | (UInt8, UInt16 | UInt32 | UInt64)
180 | (UInt16, UInt32 | UInt64)
181 | (UInt32, UInt64)
182 | (Int8 | Int16 | UInt8 | UInt16, Float32)
183 | (Int8 | Int16 | Int32 | UInt8 | UInt16 | UInt32, Float64)
184 | (Utf8, LargeUtf8)
185 )
186 }
187
188 pub fn is_bigger_cast(&self, src: &DataType) -> bool {
190 Self::check_bigger_cast(self.cast_type(), src)
191 }
192}
193
194fn is_default_target_field(target_field: &FieldRef) -> bool {
195 target_field.name().is_empty()
196 && target_field.is_nullable()
197 && target_field.metadata().is_empty()
198}
199
200pub(crate) fn is_order_preserving_cast_family(
201 source_type: &DataType,
202 target_type: &DataType,
203) -> bool {
204 (source_type.is_numeric() || *source_type == Boolean) && target_type.is_numeric()
205 || source_type.is_temporal() && target_type.is_temporal()
206 || source_type.eq(target_type)
207}
208
209pub(crate) fn cast_expr_properties(
210 child: &ExprProperties,
211 target_type: &DataType,
212) -> Result<ExprProperties> {
213 let unbounded = Interval::make_unbounded(target_type)?;
214 let source_type = child.range.data_type();
215 let bigger_cast = CastExpr::check_bigger_cast(target_type, &source_type);
219 if is_order_preserving_cast_family(&source_type, target_type) || bigger_cast {
220 Ok(child
221 .clone()
222 .with_range(unbounded)
223 .with_strictly_order_preserving(
224 child.strictly_order_preserving && bigger_cast,
225 ))
226 } else {
227 Ok(ExprProperties::new_unknown().with_range(unbounded))
228 }
229}
230
231impl fmt::Display for CastExpr {
232 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
233 write!(f, "CAST({} AS {})", self.expr, self.cast_type())
234 }
235}
236
237impl PhysicalExpr for CastExpr {
238 fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
239 Ok(self.cast_type().clone())
240 }
241
242 fn nullable(&self, input_schema: &Schema) -> Result<bool> {
243 let child_nullable = self.expr.nullable(input_schema)?;
249 let target_nullable = self.resolved_target_field(input_schema)?.is_nullable();
250 Ok(child_nullable || target_nullable)
251 }
252
253 fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
254 let value = self.expr.evaluate(batch)?;
255 value.cast_to(self.cast_type(), Some(&self.cast_options))
256 }
257
258 fn return_field(&self, input_schema: &Schema) -> Result<FieldRef> {
259 self.resolved_target_field(input_schema)
260 }
261
262 fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
263 vec![&self.expr]
264 }
265
266 fn with_new_children(
267 self: Arc<Self>,
268 children: Vec<Arc<dyn PhysicalExpr>>,
269 ) -> Result<Arc<dyn PhysicalExpr>> {
270 Ok(Arc::new(CastExpr::new_with_target_field(
271 Arc::clone(&children[0]),
272 Arc::clone(&self.target_field),
273 Some(self.cast_options.clone()),
274 )))
275 }
276
277 fn evaluate_bounds(&self, children: &[&Interval]) -> Result<Interval> {
278 children[0].cast_to(self.cast_type(), &self.cast_options)
280 }
281
282 fn propagate_constraints(
283 &self,
284 interval: &Interval,
285 children: &[&Interval],
286 ) -> Result<Option<Vec<Interval>>> {
287 let child_interval = children[0];
288 let cast_type = child_interval.data_type();
290 Ok(Some(vec![
291 interval.cast_to(&cast_type, &DEFAULT_SAFE_CAST_OPTIONS)?,
292 ]))
293 }
294
295 fn get_properties(&self, children: &[ExprProperties]) -> Result<ExprProperties> {
298 cast_expr_properties(&children[0], self.cast_type())
299 }
300
301 fn fmt_sql(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
302 write!(f, "CAST(")?;
303 self.expr.fmt_sql(f)?;
304 write!(f, " AS {:?}", self.cast_type())?;
305
306 write!(f, ")")
307 }
308
309 #[cfg(feature = "proto")]
310 fn try_to_proto(
311 &self,
312 ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>,
313 ) -> Result<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
314 use datafusion_proto_models::protobuf;
315
316 Ok(Some(protobuf::PhysicalExprNode {
317 expr_id: None,
318 expr_type: Some(protobuf::physical_expr_node::ExprType::Cast(Box::new(
319 protobuf::PhysicalCastNode {
320 expr: Some(Box::new(ctx.encode_child(self.expr())?)),
321 arrow_type: Some(self.cast_type().try_into()?),
322 },
323 ))),
324 }))
325 }
326}
327
328#[cfg(feature = "proto")]
329impl CastExpr {
330 pub fn try_from_proto(
338 node: &datafusion_proto_models::protobuf::PhysicalExprNode,
339 ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
340 ) -> Result<Arc<dyn PhysicalExpr>> {
341 use datafusion_common::internal_datafusion_err;
342 use datafusion_common::internal_err;
343 use datafusion_proto_models::protobuf;
344
345 let cast_expr = match &node.expr_type {
346 Some(protobuf::physical_expr_node::ExprType::Cast(cast_expr)) => {
347 cast_expr.as_ref()
348 }
349 _ => return internal_err!("PhysicalExprNode is not a CastExpr"),
350 };
351
352 let expr = ctx.decode_required_expression(
353 cast_expr.expr.as_deref(),
354 "CastExpr",
355 "expr",
356 )?;
357 let arrow_type = cast_expr.arrow_type.as_ref().ok_or_else(|| {
358 internal_datafusion_err!("CastExpr is missing required field 'arrow_type'")
359 })?;
360
361 Ok(Arc::new(CastExpr::new(expr, arrow_type.try_into()?, None)))
362 }
363}
364
365pub fn cast_with_options(
370 expr: Arc<dyn PhysicalExpr>,
371 input_schema: &Schema,
372 cast_type: DataType,
373 cast_options: Option<CastOptions<'static>>,
374) -> Result<Arc<dyn PhysicalExpr>> {
375 cast_with_target_field(
376 expr,
377 input_schema,
378 cast_type.into_nullable_field_ref(),
379 cast_options,
380 )
381}
382
383pub fn cast_with_target_field(
392 expr: Arc<dyn PhysicalExpr>,
393 input_schema: &Schema,
394 target_field: FieldRef,
395 cast_options: Option<CastOptions<'static>>,
396) -> Result<Arc<dyn PhysicalExpr>> {
397 let expr_type = expr.data_type(input_schema)?;
398 let cast_type = target_field.data_type();
399 if expr_type == *cast_type && is_default_target_field(&target_field) {
400 return Ok(Arc::clone(&expr));
401 }
402
403 let can_build_cast = if requires_nested_struct_cast(&expr_type, cast_type) {
404 can_cast_named_struct_types(&expr_type, cast_type)
410 } else {
411 can_cast_types(&expr_type, cast_type)
412 };
413
414 if !can_build_cast {
415 return not_impl_err!("Unsupported CAST from {expr_type} to {cast_type}");
416 }
417
418 Ok(Arc::new(CastExpr::new_with_target_field(
419 expr,
420 target_field,
421 cast_options,
422 )))
423}
424
425pub fn cast(
430 expr: Arc<dyn PhysicalExpr>,
431 input_schema: &Schema,
432 cast_type: DataType,
433) -> Result<Arc<dyn PhysicalExpr>> {
434 cast_with_options(expr, input_schema, cast_type, None)
435}
436
437#[cfg(test)]
438mod tests {
439 use super::*;
440
441 use crate::expressions::column::col;
442
443 use arrow::{
444 array::{
445 Array, ArrayRef, Decimal128Array, Float32Array, Float64Array, Int8Array,
446 Int16Array, Int32Array, Int64Array, StringArray, StructArray,
447 Time64NanosecondArray, TimestampNanosecondArray, UInt32Array,
448 },
449 datatypes::*,
450 };
451 use datafusion_common::ScalarValue;
452 use datafusion_common::cast::{
453 as_boolean_array, as_int64_array, as_string_array, as_struct_array,
454 as_uint8_array,
455 };
456 use datafusion_physical_expr_common::physical_expr::fmt_sql;
457 use insta::assert_snapshot;
458 use std::collections::HashMap;
459
460 fn make_struct_array(fields: Fields, arrays: Vec<ArrayRef>) -> StructArray {
461 StructArray::new(fields, arrays, None)
462 }
463
464 fn cast_struct_array(
465 column: &str,
466 input_field: Field,
467 target_field: Field,
468 input_array: StructArray,
469 ) -> Result<StructArray> {
470 let schema = Arc::new(Schema::new(vec![input_field]));
471 let batch = RecordBatch::try_new(
472 Arc::clone(&schema),
473 vec![Arc::new(input_array) as ArrayRef],
474 )?;
475 let expr = CastExpr::new_with_target_field(
476 col(column, schema.as_ref())?,
477 Arc::new(target_field),
478 None,
479 );
480
481 let result = expr.evaluate(&batch)?.into_array(batch.num_rows())?;
482 Ok(as_struct_array(result.as_ref())?.clone())
483 }
484
485 macro_rules! generic_decimal_to_other_test_cast {
492 ($DECIMAL_ARRAY:ident, $A_TYPE:expr, $TYPEARRAY:ident, $TYPE:expr, $VEC:expr,$CAST_OPTIONS:expr) => {{
493 let schema = Schema::new(vec![Field::new("a", $A_TYPE, true)]);
494 let batch = RecordBatch::try_new(
495 Arc::new(schema.clone()),
496 vec![Arc::new($DECIMAL_ARRAY)],
497 )?;
498 let expression =
500 cast_with_options(col("a", &schema)?, &schema, $TYPE, $CAST_OPTIONS)?;
501
502 assert_eq!(format!("CAST(a@0 AS {})", $TYPE), format!("{}", expression));
504
505 assert_eq!(expression.data_type(&schema)?, $TYPE);
507
508 let result = expression
510 .evaluate(&batch)?
511 .into_array(batch.num_rows())
512 .expect("Failed to convert to array");
513
514 assert_eq!(*result.data_type(), $TYPE);
516
517 let result = result
519 .as_any()
520 .downcast_ref::<$TYPEARRAY>()
521 .expect("failed to downcast");
522
523 for (i, x) in $VEC.iter().enumerate() {
525 match x {
526 Some(x) => assert_eq!(result.value(i), *x),
527 None => assert!(result.is_null(i)),
528 }
529 }
530 }};
531 }
532
533 macro_rules! generic_test_cast {
540 ($A_ARRAY:ident, $A_TYPE:expr, $A_VEC:expr, $TYPEARRAY:ident, $TYPE:expr, $VEC:expr, $CAST_OPTIONS:expr) => {{
541 let schema = Schema::new(vec![Field::new("a", $A_TYPE, true)]);
542 let a_vec_len = $A_VEC.len();
543 let a = $A_ARRAY::from($A_VEC);
544 let batch =
545 RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?;
546
547 let expression =
549 cast_with_options(col("a", &schema)?, &schema, $TYPE, $CAST_OPTIONS)?;
550
551 assert_eq!(format!("CAST(a@0 AS {})", $TYPE), format!("{}", expression));
553
554 assert_eq!(expression.data_type(&schema)?, $TYPE);
556
557 let result = expression
559 .evaluate(&batch)?
560 .into_array(batch.num_rows())
561 .expect("Failed to convert to array");
562
563 assert_eq!(*result.data_type(), $TYPE);
565
566 assert_eq!(result.len(), a_vec_len);
568
569 let result = result
571 .as_any()
572 .downcast_ref::<$TYPEARRAY>()
573 .expect("failed to downcast");
574
575 for (i, x) in $VEC.iter().enumerate() {
577 match x {
578 Some(x) => assert_eq!(result.value(i), *x),
579 None => assert!(result.is_null(i)),
580 }
581 }
582 }};
583 }
584
585 #[test]
586 fn test_cast_decimal_to_decimal() -> Result<()> {
587 let array = vec![
588 Some(1234),
589 Some(2222),
590 Some(3),
591 Some(4000),
592 Some(5000),
593 None,
594 ];
595
596 let decimal_array = array
597 .clone()
598 .into_iter()
599 .collect::<Decimal128Array>()
600 .with_precision_and_scale(10, 3)?;
601
602 generic_decimal_to_other_test_cast!(
603 decimal_array,
604 Decimal128(10, 3),
605 Decimal128Array,
606 Decimal128(20, 6),
607 [
608 Some(1_234_000),
609 Some(2_222_000),
610 Some(3_000),
611 Some(4_000_000),
612 Some(5_000_000),
613 None
614 ],
615 None
616 );
617
618 let decimal_array = array
619 .into_iter()
620 .collect::<Decimal128Array>()
621 .with_precision_and_scale(10, 3)?;
622
623 generic_decimal_to_other_test_cast!(
624 decimal_array,
625 Decimal128(10, 3),
626 Decimal128Array,
627 Decimal128(10, 2),
628 [Some(123), Some(222), Some(0), Some(400), Some(500), None],
629 None
630 );
631
632 Ok(())
633 }
634
635 #[test]
636 fn test_cast_decimal_to_decimal_overflow() -> Result<()> {
637 let array = vec![Some(123456789)];
638
639 let decimal_array = array
640 .clone()
641 .into_iter()
642 .collect::<Decimal128Array>()
643 .with_precision_and_scale(10, 3)?;
644
645 let schema = Schema::new(vec![Field::new("a", Decimal128(10, 3), false)]);
646 let batch = RecordBatch::try_new(
647 Arc::new(schema.clone()),
648 vec![Arc::new(decimal_array)],
649 )?;
650 let expression =
651 cast_with_options(col("a", &schema)?, &schema, Decimal128(6, 2), None)?;
652 let e = expression.evaluate(&batch).unwrap_err().strip_backtrace(); assert_snapshot!(e, @"Arrow error: Invalid argument error: 123456.79 is too large to store in a Decimal128 of precision 6. Max is 9999.99");
654 let expression_safe = cast_with_options(
656 col("a", &schema)?,
657 &schema,
658 Decimal128(6, 2),
659 Some(DEFAULT_SAFE_CAST_OPTIONS),
660 )?;
661 let result_safe = expression_safe
662 .evaluate(&batch)?
663 .into_array(batch.num_rows())
664 .expect("failed to convert to array");
665
666 assert!(result_safe.is_null(0));
667
668 Ok(())
669 }
670
671 #[test]
672 fn test_cast_decimal_to_numeric() -> Result<()> {
673 let array = vec![Some(1), Some(2), Some(3), Some(4), Some(5), None];
674 let decimal_array = array
676 .clone()
677 .into_iter()
678 .collect::<Decimal128Array>()
679 .with_precision_and_scale(10, 0)?;
680 generic_decimal_to_other_test_cast!(
681 decimal_array,
682 Decimal128(10, 0),
683 Int8Array,
684 Int8,
685 [
686 Some(1_i8),
687 Some(2_i8),
688 Some(3_i8),
689 Some(4_i8),
690 Some(5_i8),
691 None
692 ],
693 None
694 );
695
696 let decimal_array = array
698 .clone()
699 .into_iter()
700 .collect::<Decimal128Array>()
701 .with_precision_and_scale(10, 0)?;
702 generic_decimal_to_other_test_cast!(
703 decimal_array,
704 Decimal128(10, 0),
705 Int16Array,
706 Int16,
707 [
708 Some(1_i16),
709 Some(2_i16),
710 Some(3_i16),
711 Some(4_i16),
712 Some(5_i16),
713 None
714 ],
715 None
716 );
717
718 let decimal_array = array
720 .clone()
721 .into_iter()
722 .collect::<Decimal128Array>()
723 .with_precision_and_scale(10, 0)?;
724 generic_decimal_to_other_test_cast!(
725 decimal_array,
726 Decimal128(10, 0),
727 Int32Array,
728 Int32,
729 [
730 Some(1_i32),
731 Some(2_i32),
732 Some(3_i32),
733 Some(4_i32),
734 Some(5_i32),
735 None
736 ],
737 None
738 );
739
740 let decimal_array = array
742 .into_iter()
743 .collect::<Decimal128Array>()
744 .with_precision_and_scale(10, 0)?;
745 generic_decimal_to_other_test_cast!(
746 decimal_array,
747 Decimal128(10, 0),
748 Int64Array,
749 Int64,
750 [
751 Some(1_i64),
752 Some(2_i64),
753 Some(3_i64),
754 Some(4_i64),
755 Some(5_i64),
756 None
757 ],
758 None
759 );
760
761 let array = vec![
763 Some(1234),
764 Some(2222),
765 Some(3),
766 Some(4000),
767 Some(5000),
768 None,
769 ];
770 let decimal_array = array
771 .clone()
772 .into_iter()
773 .collect::<Decimal128Array>()
774 .with_precision_and_scale(10, 3)?;
775 generic_decimal_to_other_test_cast!(
776 decimal_array,
777 Decimal128(10, 3),
778 Float32Array,
779 Float32,
780 [
781 Some(1.234_f32),
782 Some(2.222_f32),
783 Some(0.003_f32),
784 Some(4.0_f32),
785 Some(5.0_f32),
786 None
787 ],
788 None
789 );
790
791 let decimal_array = array
793 .into_iter()
794 .collect::<Decimal128Array>()
795 .with_precision_and_scale(20, 6)?;
796 generic_decimal_to_other_test_cast!(
797 decimal_array,
798 Decimal128(20, 6),
799 Float64Array,
800 Float64,
801 [
802 Some(0.001234_f64),
803 Some(0.002222_f64),
804 Some(0.000003_f64),
805 Some(0.004_f64),
806 Some(0.005_f64),
807 None
808 ],
809 None
810 );
811 Ok(())
812 }
813
814 #[test]
815 fn test_cast_numeric_to_decimal() -> Result<()> {
816 generic_test_cast!(
818 Int8Array,
819 Int8,
820 vec![1, 2, 3, 4, 5],
821 Decimal128Array,
822 Decimal128(3, 0),
823 [Some(1), Some(2), Some(3), Some(4), Some(5)],
824 None
825 );
826
827 generic_test_cast!(
829 Int16Array,
830 Int16,
831 vec![1, 2, 3, 4, 5],
832 Decimal128Array,
833 Decimal128(5, 0),
834 [Some(1), Some(2), Some(3), Some(4), Some(5)],
835 None
836 );
837
838 generic_test_cast!(
840 Int32Array,
841 Int32,
842 vec![1, 2, 3, 4, 5],
843 Decimal128Array,
844 Decimal128(10, 0),
845 [Some(1), Some(2), Some(3), Some(4), Some(5)],
846 None
847 );
848
849 generic_test_cast!(
851 Int64Array,
852 Int64,
853 vec![1, 2, 3, 4, 5],
854 Decimal128Array,
855 Decimal128(20, 0),
856 [Some(1), Some(2), Some(3), Some(4), Some(5)],
857 None
858 );
859
860 generic_test_cast!(
862 Int64Array,
863 Int64,
864 vec![1, 2, 3, 4, 5],
865 Decimal128Array,
866 Decimal128(20, 2),
867 [Some(100), Some(200), Some(300), Some(400), Some(500)],
868 None
869 );
870
871 generic_test_cast!(
873 Float32Array,
874 Float32,
875 vec![1.5, 2.5, 3.0, 1.123_456_8, 5.50],
876 Decimal128Array,
877 Decimal128(10, 2),
878 [Some(150), Some(250), Some(300), Some(112), Some(550)],
879 None
880 );
881
882 generic_test_cast!(
884 Float64Array,
885 Float64,
886 vec![1.5, 2.5, 3.0, 1.123_456_8, 5.50],
887 Decimal128Array,
888 Decimal128(20, 4),
889 [
890 Some(15000),
891 Some(25000),
892 Some(30000),
893 Some(11235),
894 Some(55000)
895 ],
896 None
897 );
898 Ok(())
899 }
900
901 #[test]
902 fn test_cast_i32_u32() -> Result<()> {
903 generic_test_cast!(
904 Int32Array,
905 Int32,
906 vec![1, 2, 3, 4, 5],
907 UInt32Array,
908 UInt32,
909 [
910 Some(1_u32),
911 Some(2_u32),
912 Some(3_u32),
913 Some(4_u32),
914 Some(5_u32)
915 ],
916 None
917 );
918 Ok(())
919 }
920
921 #[test]
922 fn test_cast_i32_utf8() -> Result<()> {
923 generic_test_cast!(
924 Int32Array,
925 Int32,
926 vec![1, 2, 3, 4, 5],
927 StringArray,
928 Utf8,
929 [Some("1"), Some("2"), Some("3"), Some("4"), Some("5")],
930 None
931 );
932 Ok(())
933 }
934
935 #[test]
936 fn test_cast_i64_t64() -> Result<()> {
937 let original = vec![1, 2, 3, 4, 5];
938 let expected: Vec<Option<i64>> = original
939 .iter()
940 .map(|i| Some(Time64NanosecondArray::from(vec![*i]).value(0)))
941 .collect();
942 generic_test_cast!(
943 Int64Array,
944 Int64,
945 original,
946 TimestampNanosecondArray,
947 Timestamp(TimeUnit::Nanosecond, None),
948 expected,
949 None
950 );
951 Ok(())
952 }
953
954 #[test]
958 fn invalid_cast() {
959 let schema = Schema::new(vec![Field::new("a", Int32, false)]);
961
962 let result = cast(
963 col("a", &schema).unwrap(),
964 &schema,
965 Interval(IntervalUnit::MonthDayNano),
966 );
967 result.expect_err("expected Invalid CAST");
968 }
969
970 #[test]
971 fn invalid_cast_with_options_error() -> Result<()> {
972 let schema = Schema::new(vec![Field::new("a", Utf8, false)]);
974 let a = StringArray::from(vec!["9.1"]);
975 let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?;
976 let expression = cast_with_options(col("a", &schema)?, &schema, Int32, None)?;
977 let result = expression.evaluate(&batch);
978
979 match result {
980 Ok(_) => panic!("expected error"),
981 Err(e) => {
982 assert!(
983 e.to_string()
984 .contains("Cannot cast string '9.1' to value of Int32 type")
985 )
986 }
987 }
988 Ok(())
989 }
990
991 #[test]
992 fn field_aware_cast_preserves_target_field_semantics() -> Result<()> {
993 let metadata = HashMap::from([("target_meta".to_string(), "1".to_string())]);
994
995 for (child_nullable, target_nullable) in [(true, false), (false, true)] {
996 let schema = Schema::new(vec![Field::new("a", Int32, child_nullable)]);
997 let expr = CastExpr::new_with_target_field(
998 col("a", &schema)?,
999 Arc::new(
1000 Field::new("cast_target", Int64, target_nullable)
1001 .with_metadata(metadata.clone()),
1002 ),
1003 None,
1004 );
1005
1006 let field = expr.return_field(&schema)?;
1007 assert_eq!(field.name(), "cast_target");
1008 assert_eq!(field.data_type(), &Int64);
1009 assert_eq!(field.is_nullable(), target_nullable);
1010 assert_eq!(
1011 field.metadata().get("target_meta").map(String::as_str),
1012 Some("1")
1013 );
1014 assert_eq!(expr.nullable(&schema)?, child_nullable || target_nullable);
1015 }
1016
1017 Ok(())
1018 }
1019
1020 #[test]
1021 fn type_only_cast_preserves_legacy_field_name_and_nullability() -> Result<()> {
1022 let schema = Schema::new(vec![Field::new("a", Int32, false)]);
1023 let expr = CastExpr::new(col("a", &schema)?, Int64, None);
1024
1025 let field = expr.return_field(&schema)?;
1026
1027 assert_eq!(field.name(), "a");
1028 assert_eq!(field.data_type(), &Int64);
1029 assert!(!field.is_nullable());
1030 assert!(!expr.nullable(&schema)?);
1031
1032 Ok(())
1033 }
1034
1035 #[test]
1036 fn struct_cast_validation_uses_nested_target_fields() -> Result<()> {
1037 let source_type = Struct(Fields::from(vec![
1038 Arc::new(Field::new("x", Int32, true)),
1039 Arc::new(Field::new("y", Utf8, true)),
1040 ]));
1041 let schema = Schema::new(vec![Field::new("a", source_type.clone(), true)]);
1042
1043 let valid_target = Struct(Fields::from(vec![
1044 Arc::new(Field::new("y", Utf8, true)),
1045 Arc::new(Field::new("x", Int64, true)),
1046 ]));
1047 cast_with_options(col("a", &schema)?, &schema, valid_target, None)?;
1048
1049 let invalid_target = Struct(Fields::from(vec![
1050 Arc::new(Field::new("y", Utf8, true)),
1051 Arc::new(Field::new("missing", Int64, false)),
1052 ]));
1053 let err = cast_with_options(col("a", &schema)?, &schema, invalid_target, None)
1054 .expect_err("missing required struct field should fail");
1055
1056 assert!(err.to_string().contains("Unsupported CAST"));
1057
1058 Ok(())
1059 }
1060
1061 #[test]
1062 fn field_aware_cast_struct_array_missing_child() -> Result<()> {
1063 let source_a = Field::new("a", Int32, true);
1064 let source_b = Field::new("b", Utf8, true);
1065 let target_field = Field::new(
1066 "s",
1067 Struct(
1068 vec![
1069 Arc::new(Field::new("a", Int64, true)),
1070 Arc::new(Field::new("c", Utf8, true)),
1071 ]
1072 .into(),
1073 ),
1074 true,
1075 );
1076
1077 let struct_array = cast_struct_array(
1078 "s",
1079 Field::new(
1080 "s",
1081 Struct(
1082 vec![Arc::new(source_a.clone()), Arc::new(source_b.clone())].into(),
1083 ),
1084 true,
1085 ),
1086 target_field,
1087 make_struct_array(
1088 vec![Arc::new(source_a), Arc::new(source_b)].into(),
1089 vec![
1090 Arc::new(Int32Array::from(vec![Some(1), None])) as ArrayRef,
1091 Arc::new(StringArray::from(vec![Some("alpha"), Some("beta")]))
1092 as ArrayRef,
1093 ],
1094 ),
1095 )?;
1096 let cast_a = as_int64_array(struct_array.column_by_name("a").unwrap().as_ref())?;
1097 assert_eq!(cast_a.value(0), 1);
1098 assert!(cast_a.is_null(1));
1099
1100 let cast_c = as_string_array(struct_array.column_by_name("c").unwrap().as_ref())?;
1101 assert!(cast_c.is_null(0));
1102 assert!(cast_c.is_null(1));
1103 Ok(())
1104 }
1105
1106 #[test]
1107 fn field_aware_cast_nested_struct_array() -> Result<()> {
1108 let inner_source = Field::new(
1109 "inner",
1110 Struct(vec![Arc::new(Field::new("x", Int32, true))].into()),
1111 true,
1112 );
1113 let inner_target = Field::new(
1114 "inner",
1115 Struct(
1116 vec![
1117 Arc::new(Field::new("x", Int64, true)),
1118 Arc::new(Field::new("y", Boolean, true)),
1119 ]
1120 .into(),
1121 ),
1122 true,
1123 );
1124 let target_field =
1125 Field::new("root", Struct(vec![Arc::new(inner_target)].into()), true);
1126
1127 let inner_struct = make_struct_array(
1128 vec![Arc::new(Field::new("x", Int32, true))].into(),
1129 vec![Arc::new(Int32Array::from(vec![Some(7), None])) as ArrayRef],
1130 );
1131 let outer_struct = make_struct_array(
1132 vec![Arc::new(inner_source.clone())].into(),
1133 vec![Arc::new(inner_struct) as ArrayRef],
1134 );
1135 let struct_array = cast_struct_array(
1136 "root",
1137 Field::new("root", Struct(vec![Arc::new(inner_source)].into()), true),
1138 target_field,
1139 outer_struct,
1140 )?;
1141 let inner =
1142 as_struct_array(struct_array.column_by_name("inner").unwrap().as_ref())?;
1143 let x = as_int64_array(inner.column_by_name("x").unwrap().as_ref())?;
1144 assert_eq!(x.value(0), 7);
1145 assert!(x.is_null(1));
1146 let y = as_boolean_array(inner.column_by_name("y").unwrap().as_ref())?;
1147 assert!(y.is_null(0));
1148 assert!(y.is_null(1));
1149 Ok(())
1150 }
1151
1152 #[test]
1153 fn field_aware_cast_struct_scalar() -> Result<()> {
1154 let source_field = Field::new("a", Int32, true);
1155 let target_field = Field::new(
1156 "s",
1157 Struct(vec![Arc::new(Field::new("a", UInt8, true))].into()),
1158 true,
1159 );
1160
1161 let schema = Arc::new(Schema::new(vec![Field::new(
1162 "s",
1163 Struct(vec![Arc::new(source_field.clone())].into()),
1164 true,
1165 )]));
1166 let scalar_struct = make_struct_array(
1167 vec![Arc::new(source_field)].into(),
1168 vec![Arc::new(Int32Array::from(vec![Some(9)])) as ArrayRef],
1169 );
1170 let literal = Arc::new(crate::expressions::Literal::new(ScalarValue::Struct(
1171 Arc::new(scalar_struct),
1172 )));
1173 let expr = CastExpr::new_with_target_field(literal, Arc::new(target_field), None);
1174
1175 let batch = RecordBatch::new_empty(schema);
1176 let result = expr.evaluate(&batch)?;
1177 let ColumnarValue::Scalar(ScalarValue::Struct(array)) = result else {
1178 panic!("expected struct scalar");
1179 };
1180 let casted = as_uint8_array(array.column_by_name("a").unwrap().as_ref())?;
1181 assert_eq!(casted.value(0), 9);
1182 Ok(())
1183 }
1184
1185 #[test]
1186 #[ignore] fn test_cast_decimal() -> Result<()> {
1188 let schema = Schema::new(vec![Field::new("a", Int64, false)]);
1189 let a = Int64Array::from(vec![100]);
1190 let batch = RecordBatch::try_new(Arc::new(schema.clone()), vec![Arc::new(a)])?;
1191 let expression =
1192 cast_with_options(col("a", &schema)?, &schema, Decimal128(38, 38), None)?;
1193 expression.evaluate(&batch)?;
1194 Ok(())
1195 }
1196
1197 #[test]
1198 fn test_fmt_sql() -> Result<()> {
1199 let schema = Schema::new(vec![Field::new("a", Int32, true)]);
1200
1201 let expr = cast(col("a", &schema)?, &schema, Int64)?;
1203 let display_string = expr.to_string();
1204 assert_eq!(display_string, "CAST(a@0 AS Int64)");
1205 let sql_string = fmt_sql(expr.as_ref()).to_string();
1206 assert_eq!(sql_string, "CAST(a AS Int64)");
1207
1208 let schema = Schema::new(vec![Field::new("b", Utf8, true)]);
1210 let expr = cast(col("b", &schema)?, &schema, Int32)?;
1211 let display_string = expr.to_string();
1212 assert_eq!(display_string, "CAST(b@0 AS Int32)");
1213 let sql_string = fmt_sql(expr.as_ref()).to_string();
1214 assert_eq!(sql_string, "CAST(b AS Int32)");
1215
1216 Ok(())
1217 }
1218
1219 #[test]
1220 fn test_check_bigger_cast_precision_loss() {
1221 use DataType::*;
1222
1223 assert!(CastExpr::check_bigger_cast(&Int16, &Int8));
1225 assert!(CastExpr::check_bigger_cast(&Int64, &Int32));
1226 assert!(CastExpr::check_bigger_cast(&Float32, &Int16));
1227 assert!(CastExpr::check_bigger_cast(&Float32, &UInt16));
1228 assert!(CastExpr::check_bigger_cast(&Float64, &Int32));
1229 assert!(CastExpr::check_bigger_cast(&Float64, &UInt32));
1230 assert!(CastExpr::check_bigger_cast(&LargeUtf8, &Utf8));
1231
1232 assert!(!CastExpr::check_bigger_cast(&Float32, &Int32));
1234 assert!(!CastExpr::check_bigger_cast(&Float32, &UInt32));
1235 assert!(!CastExpr::check_bigger_cast(&Float64, &Int64));
1236 assert!(!CastExpr::check_bigger_cast(&Float64, &UInt64));
1237
1238 assert!(!CastExpr::check_bigger_cast(&UInt16, &Int8));
1240 assert!(!CastExpr::check_bigger_cast(&UInt32, &Int16));
1241 assert!(!CastExpr::check_bigger_cast(&Int16, &UInt8));
1242 }
1243}
1244
1245#[cfg(all(test, feature = "proto"))]
1247mod proto_tests {
1248 use super::*;
1249 use crate::expressions::{Column, col};
1250 use crate::proto_test_util::{
1251 StubDecoder, StubEncoder, UnreachableDecoder, column_node,
1252 };
1253 use arrow::datatypes::Field;
1254 use datafusion_common::DataFusionError;
1255 use datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx;
1256 use datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx;
1257 use datafusion_proto_models::datafusion_common::ArrowType;
1258 use datafusion_proto_models::protobuf::{
1259 PhysicalCastNode, PhysicalExprNode, physical_expr_node,
1260 };
1261
1262 fn proto_cast_fixture() -> CastExpr {
1264 let schema = Schema::new(vec![Field::new("a", Int32, false)]);
1265 CastExpr::new(col("a", &schema).unwrap(), Int64, None)
1266 }
1267
1268 fn proto_int64_arrow_type() -> ArrowType {
1269 (&Int64).try_into().unwrap()
1270 }
1271
1272 fn proto_cast_node(
1274 expr: Option<Box<PhysicalExprNode>>,
1275 arrow_type: Option<ArrowType>,
1276 ) -> PhysicalExprNode {
1277 PhysicalExprNode {
1278 expr_id: None,
1279 expr_type: Some(physical_expr_node::ExprType::Cast(Box::new(
1280 PhysicalCastNode { expr, arrow_type },
1281 ))),
1282 }
1283 }
1284
1285 #[test]
1286 fn try_to_proto_encodes_cast_expr() {
1287 let cast = proto_cast_fixture();
1288 let encoder = StubEncoder::ok();
1289 let ctx = PhysicalExprEncodeCtx::new(&encoder);
1290
1291 let node = cast
1292 .try_to_proto(&ctx)
1293 .unwrap()
1294 .expect("CastExpr should encode to Some(node)");
1295
1296 assert!(node.expr_id.is_none());
1297 let cast_node = match node.expr_type {
1298 Some(physical_expr_node::ExprType::Cast(cast_node)) => *cast_node,
1299 other => panic!("expected a Cast node, got {other:?}"),
1300 };
1301 assert!(cast_node.expr.is_some());
1302
1303 let arrow_type = cast_node
1304 .arrow_type
1305 .as_ref()
1306 .expect("cast type should be encoded");
1307 let data_type: DataType = arrow_type.try_into().unwrap();
1308 assert_eq!(data_type, Int64);
1309 }
1310
1311 #[test]
1312 fn try_to_proto_propagates_child_encode_error() {
1313 let cast = proto_cast_fixture();
1314 let encoder = StubEncoder::failing_on(1);
1315 let ctx = PhysicalExprEncodeCtx::new(&encoder);
1316
1317 let err = cast.try_to_proto(&ctx).unwrap_err();
1318 assert!(matches!(
1319 err,
1320 DataFusionError::Internal(msg) if msg.contains("call 1")
1321 ));
1322 }
1323
1324 #[test]
1325 fn try_from_proto_decodes_cast_expr() {
1326 let node = proto_cast_node(
1327 Some(Box::new(column_node("a"))),
1328 Some(proto_int64_arrow_type()),
1329 );
1330 let schema = Schema::empty();
1331 let decoder = StubDecoder::ok();
1332 let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
1333
1334 let decoded = CastExpr::try_from_proto(&node, &ctx).unwrap();
1335 let cast = decoded
1336 .downcast_ref::<CastExpr>()
1337 .expect("decoded expr should be a CastExpr");
1338
1339 assert_eq!(cast.cast_type(), &Int64);
1340 assert!(cast.expr().downcast_ref::<Column>().is_some());
1341 }
1342
1343 #[test]
1344 fn try_from_proto_rejects_non_cast_node() {
1345 let node = column_node("a");
1346 let schema = Schema::empty();
1347 let decoder = UnreachableDecoder;
1348 let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
1349
1350 let err = CastExpr::try_from_proto(&node, &ctx).unwrap_err();
1351 assert!(matches!(
1352 err,
1353 DataFusionError::Internal(msg)
1354 if msg.contains("PhysicalExprNode is not a CastExpr")
1355 ));
1356 }
1357
1358 #[test]
1359 fn try_from_proto_rejects_missing_expr() {
1360 let node = proto_cast_node(None, Some(proto_int64_arrow_type()));
1361 let schema = Schema::empty();
1362 let decoder = UnreachableDecoder;
1363 let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
1364
1365 let err = CastExpr::try_from_proto(&node, &ctx).unwrap_err();
1366 assert!(matches!(
1367 err,
1368 DataFusionError::Internal(msg)
1369 if msg.contains("CastExpr is missing required field 'expr'")
1370 ));
1371 }
1372
1373 #[test]
1374 fn try_from_proto_rejects_missing_arrow_type() {
1375 let node = proto_cast_node(Some(Box::new(column_node("a"))), None);
1376 let schema = Schema::empty();
1377 let decoder = StubDecoder::ok();
1378 let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
1379
1380 let err = CastExpr::try_from_proto(&node, &ctx).unwrap_err();
1381 assert!(matches!(
1382 err,
1383 DataFusionError::Internal(msg)
1384 if msg.contains("CastExpr is missing required field 'arrow_type'")
1385 ));
1386 }
1387
1388 #[test]
1389 fn try_from_proto_propagates_child_decode_error() {
1390 let node = proto_cast_node(
1391 Some(Box::new(column_node("a"))),
1392 Some(proto_int64_arrow_type()),
1393 );
1394 let schema = Schema::empty();
1395 let decoder = StubDecoder::failing_on(1);
1396 let ctx = PhysicalExprDecodeCtx::new(&schema, &decoder);
1397
1398 let err = CastExpr::try_from_proto(&node, &ctx).unwrap_err();
1399 assert!(matches!(
1400 err,
1401 DataFusionError::Internal(msg) if msg.contains("call 1")
1402 ));
1403 }
1404}