[SPARK-41205][SQL] Check that format is foldable in TryToBinary

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

@@ -20,6 +20,7 @@ package org.apache.spark.sql.catalyst.expressions
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, CodeGenerator, ExprCode}
 import org.apache.spark.sql.catalyst.expressions.codegen.Block._
+import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.types.{DataType, NumericType}
 
 case class TryEval(child: Expression) extends UnaryExpression with NullIntolerant {
@@ -227,7 +228,8 @@ case class TryToBinary(
 
   def this(expr: Expression, formatExpression: Expression) =
     this(expr, Some(formatExpression),
-      TryEval(ToBinary(expr, Some(formatExpression), nullOnInvalidFormat = true)))
+      TryEval(ToBinary(expr, Some(TryToBinary.checkFormat(formatExpression)),
+        nullOnInvalidFormat = true)))
 
   override def prettyName: String = "try_to_binary"
 
@@ -236,3 +238,13 @@ case class TryToBinary(
   override protected def withNewChildInternal(newChild: Expression): Expression =
     this.copy(replacement = newChild)
 }
+
+object TryToBinary {
+  def checkFormat(format: Expression): Expression = {
+    if (format.foldable) {
+      format
+    } else {
+      throw QueryCompilationErrors.requireLiteralParameter("try_to_binary", "format", "string")
+    }
+  }
+}

sql/core/src/test/resources/sql-tests/inputs/string-functions.sql

@@ -225,3 +225,7 @@ select to_binary(null, cast(null as string));
 -- invalid format
 select to_binary('abc', 1);
 select to_binary('abc', 'invalidFormat');
+-- format must be foldable
+SELECT to_binary(col1, col2) from values ('abc', 'utf-8') as data(col1, col2);
+-- non-foldable input string
+SELECT to_binary(col1, 'utf-8') from values ('abc') as data(col1);

sql/core/src/test/resources/sql-tests/inputs/try-string-functions.sql

@@ -45,4 +45,8 @@ select try_to_binary(null, null);
 select try_to_binary(null, cast(null as string));
 -- invalid format
 select try_to_binary('abc', 1);
-select try_to_binary('abc', 'invalidFormat');
+select try_to_binary('abc', 'invalidFormat');
+-- format must be foldable
+SELECT try_to_binary(col1, col2) from values ('abc', 'utf-8') as data(col1, col2);
+-- non-foldable input string
+SELECT try_to_binary(col1, 'utf-8') from values ('abc') as data(col1);

sql/core/src/test/resources/sql-tests/results/ansi/string-functions.sql.out

@@ -1597,3 +1597,20 @@ org.apache.spark.sql.AnalysisException
     "invalidValue" : "invalidformat"
   }
 }
+
+
+-- !query
+SELECT to_binary(col1, col2) from values ('abc', 'utf-8') as data(col1, col2)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.sql.AnalysisException
+The 'format' parameter of function 'to_binary' needs to be a string literal.; line 1 pos 7
+
+
+-- !query
+SELECT to_binary(col1, 'utf-8') from values ('abc') as data(col1)
+-- !query schema
+struct<to_binary(col1, utf-8):binary>
+-- !query output
+abc

sql/core/src/test/resources/sql-tests/results/string-functions.sql.out

@@ -1529,3 +1529,20 @@ org.apache.spark.sql.AnalysisException
     "invalidValue" : "invalidformat"
   }
 }
+
+
+-- !query
+SELECT to_binary(col1, col2) from values ('abc', 'utf-8') as data(col1, col2)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.sql.AnalysisException
+The 'format' parameter of function 'to_binary' needs to be a string literal.; line 1 pos 7
+
+
+-- !query
+SELECT to_binary(col1, 'utf-8') from values ('abc') as data(col1)
+-- !query schema
+struct<to_binary(col1, utf-8):binary>
+-- !query output
+abc

sql/core/src/test/resources/sql-tests/results/try-string-functions.sql.out

@@ -281,3 +281,20 @@ select try_to_binary('abc', 'invalidFormat')
 struct<try_to_binary(abc, invalidFormat):binary>
 -- !query output
 NULL
+
+
+-- !query
+SELECT try_to_binary(col1, col2) from values ('abc', 'utf-8') as data(col1, col2)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.sql.AnalysisException
+The 'format' parameter of function 'try_to_binary' needs to be a string literal.; line 1 pos 7
+
+
+-- !query
+SELECT try_to_binary(col1, 'utf-8') from values ('abc') as data(col1)
+-- !query schema
+struct<try_to_binary(col1, utf-8):binary>
+-- !query output
+abc