feat: mapping sql Char/Text/String default to Utf8View (#16290)

* feat: mapping sql Char/Text/String default to Utf8View

* Add support utf8view for sort merge join

* fix binary utf8view union with int32

* fix slt order

* fix

* fmt

* Fix test

* fix

* fix test

* fix

* clean

* Address comments

* Fix test

* support md5 for ut8view

Author: zhuqi-lucas

datafusion/common/src/config.rs

@@ -259,10 +259,10 @@ config_namespace! {
         /// string length and thus DataFusion can not enforce such limits.
         pub support_varchar_with_length: bool, default = true
 
-       /// If true, `VARCHAR` is mapped to `Utf8View` during SQL planning.
-       /// If false, `VARCHAR` is mapped to `Utf8`  during SQL planning.
-       /// Default is false.
-        pub map_varchar_to_utf8view: bool, default = true
+        /// If true, string types (VARCHAR, CHAR, Text, and String) are mapped to `Utf8View` during SQL planning.
+        /// If false, they are mapped to `Utf8`.
+        /// Default is true.
+        pub map_string_types_to_utf8view: bool, default = true
 
         /// When set to true, the source locations relative to the original SQL
         /// query (i.e. [`Span`](https://docs.rs/sqlparser/latest/sqlparser/tokenizer/struct.Span.html)) will be collected

datafusion/core/src/execution/session_state.rs

@@ -494,7 +494,7 @@ impl SessionState {
             enable_options_value_normalization: sql_parser_options
                 .enable_options_value_normalization,
             support_varchar_with_length: sql_parser_options.support_varchar_with_length,
-            map_varchar_to_utf8view: sql_parser_options.map_varchar_to_utf8view,
+            map_string_types_to_utf8view: sql_parser_options.map_string_types_to_utf8view,
             collect_spans: sql_parser_options.collect_spans,
         }
     }

datafusion/core/tests/sql/create_drop.rs

@@ -61,7 +61,7 @@ async fn create_external_table_with_ddl() -> Result<()> {
     assert_eq!(3, table_schema.fields().len());
 
     assert_eq!(&DataType::Int32, table_schema.field(0).data_type());
-    assert_eq!(&DataType::Utf8, table_schema.field(1).data_type());
+    assert_eq!(&DataType::Utf8View, table_schema.field(1).data_type());
     assert_eq!(&DataType::Boolean, table_schema.field(2).data_type());
 
     Ok(())

datafusion/core/tests/user_defined/user_defined_scalar_functions.rs

@@ -1181,7 +1181,7 @@ async fn create_scalar_function_from_sql_statement_postgres_syntax() -> Result<(
                 quote_style: None,
                 span: Span::empty(),
             }),
-            data_type: DataType::Utf8,
+            data_type: DataType::Utf8View,
             default_expr: None,
         }]),
         return_type: Some(DataType::Int32),

datafusion/expr-common/src/type_coercion/binary.rs

@@ -462,7 +462,7 @@ pub fn type_union_resolution(data_types: &[DataType]) -> Option<DataType> {
 
     // If all the data_types are null, return string
     if data_types.iter().all(|t| t == &DataType::Null) {
-        return Some(DataType::Utf8);
+        return Some(DataType::Utf8View);
     }
 
     // Ignore Nulls, if any data_type category is not the same, return None
@@ -1202,7 +1202,8 @@ pub fn string_coercion(lhs_type: &DataType, rhs_type: &DataType) -> Option<DataT
 fn numeric_string_coercion(lhs_type: &DataType, rhs_type: &DataType) -> Option<DataType> {
     use arrow::datatypes::DataType::*;
     match (lhs_type, rhs_type) {
-        (Utf8 | LargeUtf8, other_type) | (other_type, Utf8 | LargeUtf8)
+        (Utf8 | LargeUtf8 | Utf8View, other_type)
+        | (other_type, Utf8 | LargeUtf8 | Utf8View)
             if other_type.is_numeric() =>
         {
             Some(other_type.clone())

datafusion/functions/src/crypto/basic.rs

@@ -21,7 +21,7 @@ use arrow::array::{
     Array, ArrayRef, BinaryArray, BinaryArrayType, BinaryViewArray, GenericBinaryArray,
     OffsetSizeTrait,
 };
-use arrow::array::{AsArray, GenericStringArray, StringArray, StringViewArray};
+use arrow::array::{AsArray, GenericStringArray, StringViewArray};
 use arrow::datatypes::DataType;
 use blake2::{Blake2b512, Blake2s256, Digest};
 use blake3::Hasher as Blake3;
@@ -169,18 +169,18 @@ pub fn md5(args: &[ColumnarValue]) -> Result<ColumnarValue> {
     let [data] = take_function_args("md5", args)?;
     let value = digest_process(data, DigestAlgorithm::Md5)?;
 
-    // md5 requires special handling because of its unique utf8 return type
+    // md5 requires special handling because of its unique utf8view return type
     Ok(match value {
         ColumnarValue::Array(array) => {
             let binary_array = as_binary_array(&array)?;
-            let string_array: StringArray = binary_array
+            let string_array: StringViewArray = binary_array
                 .iter()
                 .map(|opt| opt.map(hex_encode::<_>))
                 .collect();
             ColumnarValue::Array(Arc::new(string_array))
         }
         ColumnarValue::Scalar(ScalarValue::Binary(opt)) => {
-            ColumnarValue::Scalar(ScalarValue::Utf8(opt.map(hex_encode::<_>)))
+            ColumnarValue::Scalar(ScalarValue::Utf8View(opt.map(hex_encode::<_>)))
         }
         _ => return exec_err!("Impossibly got invalid results from digest"),
     })

datafusion/functions/src/crypto/md5.rs

@@ -92,12 +92,12 @@ impl ScalarUDFImpl for Md5Func {
     fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
         use DataType::*;
         Ok(match &arg_types[0] {
-            LargeUtf8 | LargeBinary => Utf8,
-            Utf8View | Utf8 | Binary | BinaryView => Utf8,
+            LargeUtf8 | LargeBinary => Utf8View,
+            Utf8View | Utf8 | Binary | BinaryView => Utf8View,
             Null => Null,
             Dictionary(_, t) => match **t {
-                LargeUtf8 | LargeBinary => Utf8,
-                Utf8 | Binary | BinaryView => Utf8,
+                LargeUtf8 | LargeBinary => Utf8View,
+                Utf8 | Binary | BinaryView => Utf8View,
                 Null => Null,
                 _ => {
                     return plan_err!(

datafusion/physical-plan/src/joins/sort_merge_join.rs

@@ -2492,6 +2492,7 @@ fn compare_join_arrays(
             DataType::Float32 => compare_value!(Float32Array),
             DataType::Float64 => compare_value!(Float64Array),
             DataType::Utf8 => compare_value!(StringArray),
+            DataType::Utf8View => compare_value!(StringViewArray),
             DataType::LargeUtf8 => compare_value!(LargeStringArray),
             DataType::Decimal128(..) => compare_value!(Decimal128Array),
             DataType::Timestamp(time_unit, None) => match time_unit {
@@ -2559,6 +2560,7 @@ fn is_join_arrays_equal(
             DataType::Float32 => compare_value!(Float32Array),
             DataType::Float64 => compare_value!(Float64Array),
             DataType::Utf8 => compare_value!(StringArray),
+            DataType::Utf8View => compare_value!(StringViewArray),
             DataType::LargeUtf8 => compare_value!(LargeStringArray),
             DataType::Decimal128(..) => compare_value!(Decimal128Array),
             DataType::Timestamp(time_unit, None) => match time_unit {

datafusion/sql/src/planner.rs

@@ -52,8 +52,8 @@ pub struct ParserOptions {
     pub enable_options_value_normalization: bool,
     /// Whether to collect spans
     pub collect_spans: bool,
-    /// Whether `VARCHAR` is mapped to `Utf8View` during SQL planning.
-    pub map_varchar_to_utf8view: bool,
+    /// Whether string types (VARCHAR, CHAR, Text, and String) are mapped to `Utf8View` during SQL planning.
+    pub map_string_types_to_utf8view: bool,
 }
 
 impl ParserOptions {
@@ -72,7 +72,7 @@ impl ParserOptions {
             parse_float_as_decimal: false,
             enable_ident_normalization: true,
             support_varchar_with_length: true,
-            map_varchar_to_utf8view: true,
+            map_string_types_to_utf8view: true,
             enable_options_value_normalization: false,
             collect_spans: false,
         }
@@ -112,9 +112,9 @@ impl ParserOptions {
         self
     }
 
-    /// Sets the `map_varchar_to_utf8view` option.
-    pub fn with_map_varchar_to_utf8view(mut self, value: bool) -> Self {
-        self.map_varchar_to_utf8view = value;
+    /// Sets the `map_string_types_to_utf8view` option.
+    pub fn with_map_string_types_to_utf8view(mut self, value: bool) -> Self {
+        self.map_string_types_to_utf8view = value;
         self
     }
 
@@ -143,7 +143,7 @@ impl From<&SqlParserOptions> for ParserOptions {
             parse_float_as_decimal: options.parse_float_as_decimal,
             enable_ident_normalization: options.enable_ident_normalization,
             support_varchar_with_length: options.support_varchar_with_length,
-            map_varchar_to_utf8view: options.map_varchar_to_utf8view,
+            map_string_types_to_utf8view: options.map_string_types_to_utf8view,
             enable_options_value_normalization: options
                 .enable_options_value_normalization,
             collect_spans: options.collect_spans,
@@ -577,7 +577,7 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
                     please set `support_varchar_with_length` to be true"
                     ),
                     _ => {
-                        if self.options.map_varchar_to_utf8view {
+                        if self.options.map_string_types_to_utf8view {
                             Ok(DataType::Utf8View)
                         } else {
                             Ok(DataType::Utf8)
@@ -601,7 +601,11 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
                 )
             }
             SQLDataType::Char(_) | SQLDataType::Text | SQLDataType::String(_) => {
-                Ok(DataType::Utf8)
+                if self.options.map_string_types_to_utf8view {
+                    Ok(DataType::Utf8View)
+                } else {
+                    Ok(DataType::Utf8)
+                }
             }
             SQLDataType::Timestamp(precision, tz_info)
                 if precision.is_none() || [0, 3, 6, 9].contains(&precision.unwrap()) =>

datafusion/sql/tests/cases/params.rs

@@ -746,31 +746,31 @@ fn test_prepare_statement_to_plan_multi_params() {
     assert_snapshot!(
         plan,
         @r#"
-    Prepare: "my_plan" [Int32, Utf8, Float64, Int32, Float64, Utf8]
+    Prepare: "my_plan" [Int32, Utf8View, Float64, Int32, Float64, Utf8View]
       Projection: person.id, person.age, $6
         Filter: person.age IN ([$1, $4]) AND person.salary > $3 AND person.salary < $5 OR person.first_name < $2
           TableScan: person
     "#
     );
-    assert_snapshot!(dt, @r#"[Int32, Utf8, Float64, Int32, Float64, Utf8]"#);
+    assert_snapshot!(dt, @r#"[Int32, Utf8View, Float64, Int32, Float64, Utf8View]"#);
 
     ///////////////////
     // replace params with values
     let param_values = vec![
         ScalarValue::Int32(Some(10)),
-        ScalarValue::from("abc"),
+        ScalarValue::Utf8View(Some("abc".into())),
         ScalarValue::Float64(Some(100.0)),
         ScalarValue::Int32(Some(20)),
         ScalarValue::Float64(Some(200.0)),
-        ScalarValue::from("xyz"),
+        ScalarValue::Utf8View(Some("xyz".into())),
     ];
 
     let plan_with_params = plan.with_param_values(param_values).unwrap();
     assert_snapshot!(
         plan_with_params,
         @r#"
-    Projection: person.id, person.age, Utf8("xyz") AS $6
-      Filter: person.age IN ([Int32(10), Int32(20)]) AND person.salary > Float64(100) AND person.salary < Float64(200) OR person.first_name < Utf8("abc")
+    Projection: person.id, person.age, Utf8View("xyz") AS $6
+      Filter: person.age IN ([Int32(10), Int32(20)]) AND person.salary > Float64(100) AND person.salary < Float64(200) OR person.first_name < Utf8View("abc")
         TableScan: person
     "#
     );