Support Empty2Null in 332db

sql-plugin/src/main/spark332db/scala/com/nvidia/spark/rapids/shims/SparkShims.scala

@@ -29,9 +29,11 @@ import org.apache.spark.sql.execution.{ColumnarToRowTransition, SparkPlan}
 import org.apache.spark.sql.execution.adaptive.ShuffleQueryStageExec
 import org.apache.spark.sql.execution.command.{CreateDataSourceTableAsSelectCommand, DataWritingCommand, RunnableCommand}
 import org.apache.spark.sql.execution.datasources._
+import org.apache.spark.sql.execution.datasources.V1WritesUtils.Empty2Null
 import org.apache.spark.sql.execution.datasources.parquet.ParquetFilters
 import org.apache.spark.sql.execution.exchange.{EXECUTOR_BROADCAST, ShuffleExchangeExec, ShuffleExchangeLike}
 import org.apache.spark.sql.rapids.GpuElementAtMeta
+import org.apache.spark.sql.rapids.GpuV1WriteUtils.GpuEmpty2Null
 import org.apache.spark.sql.rapids.execution.{GpuBroadcastHashJoinExec, GpuBroadcastNestedLoopJoinExec}
 
 object SparkShimImpl extends Spark321PlusDBShims {
@@ -56,6 +58,16 @@ object SparkShimImpl extends Spark321PlusDBShims {
 
   override def getExprs: Map[Class[_ <: Expression], ExprRule[_ <: Expression]] = {
     val shimExprs: Map[Class[_ <: Expression], ExprRule[_ <: Expression]] = Seq(
+      // Empty2Null is pulled out of FileFormatWriter by default since Spark 3.4.0,
+      // so it is visible in the overriding stage.
+      GpuOverrides.expr[Empty2Null](
+        "Converts the empty string to null for writing data",
+        ExprChecks.unaryProjectInputMatchesOutput(
+          TypeSig.STRING, TypeSig.STRING),
+        (a, conf, p, r) => new UnaryExprMeta[Empty2Null](a, conf, p, r) {
+          override def convertToGpu(child: Expression): GpuExpression = GpuEmpty2Null(child)
+        }
+      ),
       GpuOverrides.expr[KnownNullable](
         "Tags the expression as being nullable",
         ExprChecks.unaryProjectInputMatchesOutput(