feat: support_decimal_types_bool_cast_native_impl (#2490)

Author: coderfender

docs/source/user-guide/latest/compatibility.md

@@ -193,6 +193,7 @@ The following cast operations are generally compatible with Spark except for the
 | double | long |  |
 | double | float |  |
 | double | string | There can be differences in precision. For example, the input "1.4E-45" will produce 1.0E-45 instead of 1.4E-45 |
+| decimal | boolean |  |
 | decimal | byte |  |
 | decimal | short |  |
 | decimal | integer |  |

native/spark-expr/src/conversion_funcs/cast.rs

@@ -20,7 +20,8 @@ use crate::{timezone, BinaryOutputStyle};
 use crate::{EvalMode, SparkError, SparkResult};
 use arrow::array::builder::StringBuilder;
 use arrow::array::{
-    Decimal128Builder, DictionaryArray, GenericByteArray, ListArray, StringArray, StructArray,
+    BooleanBuilder, Decimal128Builder, DictionaryArray, GenericByteArray, ListArray, StringArray,
+    StructArray,
 };
 use arrow::compute::can_cast_types;
 use arrow::datatypes::{
@@ -52,7 +53,7 @@ use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_plan::ColumnarValue;
 use num::{
     cast::AsPrimitive, integer::div_floor, traits::CheckedNeg, CheckedSub, Integer, Num,
-    ToPrimitive,
+    ToPrimitive, Zero,
 };
 use regex::Regex;
 use std::str::FromStr;
@@ -1020,6 +1021,7 @@ fn cast_array(
         {
             spark_cast_nonintegral_numeric_to_integral(&array, eval_mode, from_type, to_type)
         }
+        (Decimal128(_p, _s), Boolean) => spark_cast_decimal_to_boolean(&array),
         (Utf8View, Utf8) => Ok(cast_with_options(&array, to_type, &CAST_OPTIONS)?),
         (Struct(_), Utf8) => Ok(casts_struct_to_string(array.as_struct(), cast_options)?),
         (Struct(_), Struct(_)) => Ok(cast_struct_to_struct(
@@ -1678,6 +1680,19 @@ where
     Ok(Arc::new(output_array))
 }
 
+fn spark_cast_decimal_to_boolean(array: &dyn Array) -> SparkResult<ArrayRef> {
+    let decimal_array = array.as_primitive::<Decimal128Type>();
+    let mut result = BooleanBuilder::with_capacity(decimal_array.len());
+    for i in 0..decimal_array.len() {
+        if decimal_array.is_null(i) {
+            result.append_null()
+        } else {
+            result.append_value(!decimal_array.value(i).is_zero());
+        }
+    }
+    Ok(Arc::new(result.finish()))
+}
+
 fn spark_cast_nonintegral_numeric_to_integral(
     array: &dyn Array,
     eval_mode: EvalMode,

spark/src/main/scala/org/apache/comet/expressions/CometCast.scala

@@ -338,7 +338,7 @@ object CometCast extends CometExpressionSerde[Cast] with CometExprShim {
 
   private def canCastFromDecimal(toType: DataType): SupportLevel = toType match {
     case DataTypes.FloatType | DataTypes.DoubleType | DataTypes.ByteType | DataTypes.ShortType |
-        DataTypes.IntegerType | DataTypes.LongType =>
+        DataTypes.IntegerType | DataTypes.LongType | DataTypes.BooleanType =>
       Compatible()
     case _ => Unsupported(Some(s"Cast from DecimalType to $toType is not supported"))
   }

spark/src/test/scala/org/apache/comet/CometCastSuite.scala

@@ -519,8 +519,7 @@ class CometCastSuite extends CometTestBase with AdaptiveSparkPlanHelper {
 
   // CAST from DecimalType(10,2)
 
-  ignore("cast DecimalType(10,2) to BooleanType") {
-    // Arrow error: Cast error: Casting from Decimal128(38, 18) to Boolean not supported
+  test("cast DecimalType(10,2) to BooleanType") {
     castTest(generateDecimalsPrecision10Scale2(), DataTypes.BooleanType)
   }