Fix

Author: maropu

sql/core/src/test/resources/sql-tests/inputs/transform.sql

@@ -9,16 +9,6 @@ SELECT TRANSFORM(a)
 USING 'cat' AS (a)
 FROM t;
 
--- with non-exist command
-SELECT TRANSFORM(a)
-USING 'some_non_existent_command' AS (a)
-FROM t;
-
--- with non-exist file
-SELECT TRANSFORM(a)
-USING 'python some_non_existent_file' AS (a)
-FROM t;
-
 -- common supported data types between no serde and serde transform
 SELECT a, b, decode(c, 'UTF-8'), d, e, f, g, h, i, j, k, l FROM (
   SELECT TRANSFORM(a, b, c, d, e, f, g, h, i, j, k, l)

sql/core/src/test/resources/sql-tests/results/transform.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 18
+-- Number of queries: 16
 
 
 -- !query
@@ -26,28 +26,6 @@ struct<a:string>
 3
 
 
--- !query
-SELECT TRANSFORM(a)
-USING 'some_non_existent_command' AS (a)
-FROM t
--- !query schema
-struct<>
--- !query output
-org.apache.spark.SparkException
-Subprocess exited with status 127. Error: /bin/bash: some_non_existent_command: command not found
-
-
--- !query
-SELECT TRANSFORM(a)
-USING 'python some_non_existent_file' AS (a)
-FROM t
--- !query schema
-struct<>
--- !query output
-org.apache.spark.SparkException
-Subprocess exited with status 2. Error: python: can't open file 'some_non_existent_file': [Errno 2] No such file or directory
-
-
 -- !query
 SELECT a, b, decode(c, 'UTF-8'), d, e, f, g, h, i, j, k, l FROM (
   SELECT TRANSFORM(a, b, c, d, e, f, g, h, i, j, k, l)

sql/core/src/test/scala/org/apache/spark/sql/execution/BaseScriptTransformationSuite.scala

@@ -420,6 +420,28 @@ abstract class BaseScriptTransformationSuite extends SparkPlanTest with SQLTestU
         'b.cast("string").as("b"),
         lit(null), lit(null)).collect())
   }
+
+  test("SPARK-32106: TRANSFORM with non-existent command/file") {
+      val e1 = intercept[SparkException] {
+        sql(
+          """
+            |SELECT TRANSFORM(a)
+            |USING 'some_non_existent_command' AS (a)
+            |FROM VALUES (1) t(a)
+          """.stripMargin).collect()
+      }.getMessage
+      assert(e1.contains("Subprocess exited"))
+
+      val e2 = intercept[SparkException] {
+        sql(
+          """
+            |SELECT TRANSFORM(a)
+            |USING 'python some_non_existent_file' AS (a)
+            |FROM VALUES (1) t(a)
+          """.stripMargin).collect()
+      }.getMessage
+      assert(e2.contains("Subprocess exited"))
+  }
 }
 
 case class ExceptionInjectingOperator(child: SparkPlan) extends UnaryExecNode {