feat: implement native empty2null spark inner function

native/spark-expr/src/comet_scalar_funcs.rs

@@ -27,8 +27,8 @@ use crate::{
     spark_isnan, spark_lpad, spark_make_decimal, spark_month_name, spark_read_side_padding,
     spark_round, spark_rpad, spark_to_time, spark_unhex, spark_unscaled_value, EvalMode,
     SparkArrayCompact, SparkArrayPositionFunc, SparkArraySlice, SparkArraysOverlap, SparkContains,
-    SparkDateDiff, SparkDateFromUnixDate, SparkDateTrunc, SparkMakeDate, SparkMakeTime,
-    SparkNextDay, SparkSecondsToTimestamp, SparkSizeFunc,
+    SparkDateDiff, SparkDateFromUnixDate, SparkDateTrunc, SparkEmpty2Null, SparkMakeDate,
+    SparkMakeTime, SparkNextDay, SparkSecondsToTimestamp, SparkSizeFunc,
 };
 use arrow::datatypes::DataType;
 use datafusion::common::{DataFusionError, Result as DataFusionResult};
@@ -249,6 +249,7 @@ fn all_scalar_functions() -> Vec<Arc<ScalarUDF>> {
         Arc::new(ScalarUDF::new_from_impl(SparkSecondsToTimestamp::default())),
         Arc::new(ScalarUDF::new_from_impl(SparkSizeFunc::default())),
         Arc::new(ScalarUDF::new_from_impl(JsonArrayLength::default())),
+        Arc::new(ScalarUDF::new_from_impl(SparkEmpty2Null::default())),
     ]
 }
 

native/spark-expr/src/string_funcs/empty_to_null.rs

@@ -0,0 +1,151 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, 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 arrow::array::{ArrayRef, OffsetSizeTrait, StringArray};
+use arrow::datatypes::DataType;
+use datafusion::common::cast::{as_generic_string_array, as_string_view_array};
+use datafusion::common::{exec_err, Result, ScalarValue};
+use datafusion::logical_expr::{
+    ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
+};
+use std::any::Any;
+use std::sync::Arc;
+
+/// Spark `Empty2Null`: returns NULL if the string is NULL or empty (`""`),
+/// otherwise returns the string itself. Used in partitioned writes to ensure
+/// that an empty string ends up in `__HIVE_DEFAULT_PARTITION__`, just like NULL.
+#[derive(Debug, PartialEq, Eq, Hash)]
+pub struct SparkEmpty2Null {
+    signature: Signature,
+}
+
+impl Default for SparkEmpty2Null {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl SparkEmpty2Null {
+    pub fn new() -> Self {
+        Self {
+            signature: Signature::variadic_any(Volatility::Immutable),
+        }
+    }
+}
+
+impl ScalarUDFImpl for SparkEmpty2Null {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn name(&self) -> &str {
+        "empty2Null"
+    }
+
+    fn signature(&self) -> &Signature {
+        &self.signature
+    }
+
+    fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
+        Ok(arg_types[0].clone())
+    }
+
+    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
+        spark_empty_to_null(&args.args)
+    }
+}
+
+fn spark_empty_to_null(args: &[ColumnarValue]) -> Result<ColumnarValue> {
+    if args.len() != 1 {
+        return exec_err!("spark empty2Null takes exactly one argument");
+    }
+    match &args[0] {
+        ColumnarValue::Array(array) => {
+            let result = spark_empty_to_null_array(&array)?;
+            Ok(ColumnarValue::Array(result))
+        }
+        ColumnarValue::Scalar(scalar) => {
+            let result = spark_empty_to_null_scalar(&scalar)?;
+            Ok(ColumnarValue::Scalar(result))
+        }
+    }
+}
+
+fn spark_empty_to_null_array(array: &ArrayRef) -> Result<ArrayRef> {
+    match array.data_type() {
+        DataType::Utf8 => spark_empty_to_null_string_array::<i32>(array),
+        DataType::LargeUtf8 => spark_empty_to_null_string_array::<i64>(array),
+        DataType::Utf8View => spark_empty_to_null_string_view(array),
+        other => {
+            exec_err!("unsupported data type {other:?} for function `empty2Null`")
+        }
+    }
+}
+
+fn spark_empty_to_null_scalar(scalar: &ScalarValue) -> Result<ScalarValue> {
+    match scalar {
+        ScalarValue::Utf8(v) => Ok(ScalarValue::Utf8(v.clone().filter(|s| !s.is_empty()))),
+        ScalarValue::LargeUtf8(v) => {
+            Ok(ScalarValue::LargeUtf8(v.clone().filter(|s| !s.is_empty())))
+        }
+        ScalarValue::Utf8View(v) => Ok(ScalarValue::Utf8View(v.clone().filter(|s| !s.is_empty()))),
+        other => {
+            exec_err!("unsupported data type {other:?} for function `empty2Null`")
+        }
+    }
+}
+
+fn spark_empty_to_null_string_array<O: OffsetSizeTrait>(array: &ArrayRef) -> Result<ArrayRef> {
+    let str_array = as_generic_string_array::<O>(array)?;
+    let result = str_array
+        .iter()
+        .map(|opt| opt.filter(|s| !s.is_empty()))
+        .collect::<StringArray>();
+    Ok(Arc::new(result))
+}
+
+fn spark_empty_to_null_string_view(str_view: &ArrayRef) -> Result<ArrayRef> {
+    let str_array = as_string_view_array(str_view)?;
+    let result = str_array
+        .iter()
+        .map(|opt| opt.filter(|s| !s.is_empty()))
+        .collect::<StringArray>();
+    Ok(Arc::new(result))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow::array::{Array, StringArray};
+
+    #[test]
+    fn test_empty_to_null_utf8() {
+        let input = Arc::new(StringArray::from(vec![
+            Some("a"),
+            Some(""),
+            None,
+            Some("b"),
+        ])) as ArrayRef;
+        let result = spark_empty_to_null_array(&input).unwrap();
+        let result = result.as_any().downcast_ref::<StringArray>().unwrap();
+        assert_eq!(result.value(0), "a");
+        assert!(result.is_null(1)); // "" -> null
+        assert!(result.is_null(2)); // null -> null
+        assert_eq!(result.value(3), "b");
+    }
+}
+

native/spark-expr/src/string_funcs/mod.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 mod contains;
+mod empty_to_null;
 mod get_json_object;
 mod regexp_extract;
 mod regexp_extract_all;
@@ -24,6 +25,7 @@ mod split;
 mod substring;
 
 pub use contains::SparkContains;
+pub use empty_to_null::SparkEmpty2Null;
 pub use get_json_object::spark_get_json_object;
 pub use regexp_extract::spark_regexp_extract;
 pub use regexp_extract_all::spark_regexp_extract_all;

spark/src/main/scala/org/apache/comet/serde/QueryPlanSerde.scala

@@ -256,7 +256,8 @@ object QueryPlanSerde extends Logging with CometExprShim with CometTypeShim {
       classOf[ToCharacter] -> CometToCharacter,
       classOf[ToNumber] -> CometToNumber,
       classOf[TryToNumber] -> CometTryToNumber,
-      classOf[Mask] -> CometMask)
+      classOf[Mask] -> CometMask,
+      classOf[Empty2Null] -> CometEmpty2Null)
     base ++ sparkVersionSpecificStringExpressions
   }
 

spark/src/main/scala/org/apache/comet/serde/strings.scala

@@ -19,7 +19,7 @@
 
 package org.apache.comet.serde
 
-import org.apache.spark.sql.catalyst.expressions.{Attribute, BitLength, Cast, Concat, ConcatWs, Elt, Expression, FindInSet, FormatNumber, FormatString, GetJsonObject, If, InitCap, IsNull, Left, Length, Levenshtein, Like, Literal, Lower, Mask, OctetLength, Overlay, RegExpExtract, RegExpExtractAll, RegExpInStr, RegExpReplace, Right, RLike, SoundEx, StringLocate, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, StringTranslate, Substring, SubstringIndex, ToCharacter, ToNumber, TryToNumber, UnBase64, Upper}
+import org.apache.spark.sql.catalyst.expressions.{Attribute, BitLength, Cast, Concat, ConcatWs, Elt, Empty2Null, Expression, FindInSet, FormatNumber, FormatString, GetJsonObject, If, InitCap, IsNull, Left, Length, Levenshtein, Like, Literal, Lower, Mask, OctetLength, Overlay, RegExpExtract, RegExpExtractAll, RegExpInStr, RegExpReplace, Right, RLike, SoundEx, StringLocate, StringLPad, StringRepeat, StringReplace, StringRPad, StringSplit, StringTranslate, Substring, SubstringIndex, ToCharacter, ToNumber, TryToNumber, UnBase64, Upper}
 import org.apache.spark.sql.types.{BinaryType, DataTypes, LongType, StringType}
 import org.apache.spark.unsafe.types.UTF8String
 
@@ -620,6 +620,10 @@ object CometGetJsonObject extends CometCodegenDispatch[GetJsonObject] {
     }
 }
 
+// A internal function that converts the empty string to null for partition values.
+// This function should be only used in V1Writes.
+object CometEmpty2Null extends CometScalarFunction[Empty2Null]("empty2Null")
+
 // Expressions routed through the JVM codegen dispatcher: no native implementation, so Spark's own
 // doGenCode runs inside the Comet pipeline, matching Spark exactly.
 object CometLevenshtein extends CometCodegenDispatch[Levenshtein]