[SPARK-42169] Implement code generation for to_csv function (StructsToCsv)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/csvExpressions.scala

@@ -25,7 +25,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.DataTypeMismatch
 import org.apache.spark.sql.catalyst.csv._
-import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, CodegenFallback, ExprCode}
 import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase}
 import org.apache.spark.sql.internal.SQLConf
@@ -245,8 +245,8 @@ case class StructsToCsv(
      options: Map[String, String],
      child: Expression,
      timeZoneId: Option[String] = None)
-  extends UnaryExpression with TimeZoneAwareExpression with CodegenFallback with ExpectsInputTypes
-    with NullIntolerant {
+  extends UnaryExpression with TimeZoneAwareExpression with ExpectsInputTypes
+    with NullIntolerant with Serializable {
   override def nullable: Boolean = true
 
   def this(options: Map[String, String], child: Expression) = this(options, child, None)
@@ -293,4 +293,10 @@ case class StructsToCsv(
 
   override protected def withNewChildInternal(newChild: Expression): StructsToCsv =
     copy(child = newChild)
+
+  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    val structsToCsv = ctx.addReferenceObj("structsToCsv", this)
+    nullSafeCodeGen(ctx, ev,
+      eval => s"${ev.value} = (UTF8String) $structsToCsv.converter().apply($eval);")
+  }
 }

sql/core/src/test/scala/org/apache/spark/sql/CsvFunctionsSuite.scala

@@ -574,4 +574,36 @@ class CsvFunctionsSuite extends QueryTest with SharedSparkSession {
       $"csv", schema_of_csv("1,2\n2"), Map.empty[String, String].asJava))
     checkAnswer(actual, Row(Row(1, "2\n2")))
   }
+
+  test("SPARK-41049: make to_csv function deterministic") {
+    val df = sparkContext.parallelize(1 to 5).toDF("value")
+    val v1 = rand().*(lit(10000)).cast(IntegerType)
+    val v2 = to_csv(struct(v1.as("testCsvField")))
+    val withSingleCsvValue = df.select(v1, v1, v2, v2).collect()
+    withSingleCsvValue.foreach(row => {
+      val values = Seq(
+        row.getInt(0).toString,
+        row.getInt(1).toString,
+        row.getString(2),
+        row.getString(3))
+
+      assert(row.length == 4)
+      assert(values.size == 4)
+      assert(values.forall(_ == values.head))
+    })
+
+    val v3 = to_csv(struct(v1.as("testCsvField1"), v1.as("testCsvField2")))
+    val withMultipleCsvValues = df.select(v1, v1, v3, v3).collect()
+    withMultipleCsvValues.foreach(row => {
+      val values = Array(
+        row.getInt(0).toString,
+        row.getInt(1).toString) ++
+        row.getString(2).split(",") ++
+        row.getString(3).split(",")
+
+      assert(row.length == 4)
+      assert(values.length == 6)
+      assert(values.forall(_ == values.head))
+    })
+  }
 }