[SPARK-43009][SQL][3.4] Parameterized sql() with Any constants

[MaxGekk]

What changes were proposed in this pull request?

In the PR, I propose to change API of parameterized SQL, and replace type of argument values from string to Any in Scala/Java/Python and Expression.Literal in protobuf API. Language API can accept Any objects from which it is possible to construct literal expressions.

This is a backport of #40623

Scala/Java:

  def sql(sqlText: String, args: Map[String, Any]): DataFrame

values of the args map are wrapped by the lit() function which leaves Column as is and creates a literal from other Java/Scala objects (for more details see the Scala tab at https://spark.apache.org/docs/latest/sql-ref-datatypes.html).

Python:

def sql(self, sqlQuery: str, args: Optional[Dict[str, Any]] = None, **kwargs: Any) -> DataFrame:

Similarly to Scala/Java sql, Python's sql() accepts Python objects as values of the args dictionary (see more details about acceptable Python objects at https://spark.apache.org/docs/latest/sql-ref-datatypes.html). sql() converts dictionary values to Column literal expressions by lit().

Protobuf:

message SqlCommand {
  // (Required) SQL Query.
  string sql = 1;

  // (Optional) A map of parameter names to literal expressions.
  map<string, Expression.Literal> args = 2;
}

For example:

scala> val sqlText = """SELECT s FROM VALUES ('Jeff /*__*/ Green'), ('E\'Twaun Moore') AS t(s) WHERE s = :player_name"""
sqlText: String = SELECT s FROM VALUES ('Jeff /*__*/ Green'), ('E\'Twaun Moore') AS t(s) WHERE s = :player_name

scala> sql(sqlText, args = Map("player_name" -> lit("E'Twaun Moore"))).show(false)
+-------------+
|s            |
+-------------+
|E'Twaun Moore|
+-------------+

Why are the changes needed?

The current implementation the parameterized sql() requires arguments as string values parsed to SQL literal expressions that causes the following issues:

1. SQL comments are skipped while parsing, so, some fragments of input might be skipped. For example, 'Europe -- Amsterdam'. In this case, -- Amsterdam is excluded from the input.
2. Special chars in string values must be escaped, for instance 'E\'Twaun Moore'

Does this PR introduce any user-facing change?

No since the parameterized SQL feature #38864 hasn't been released yet.

How was this patch tested?

By running the affected tests:

$ build/sbt "test:testOnly *ParametersSuite"
$ python/run-tests --parallelism=1 --testnames 'pyspark.sql.tests.connect.test_connect_basic SparkConnectBasicTests.test_sql_with_args'
$ python/run-tests --parallelism=1 --testnames 'pyspark.sql.session SparkSession.sql'

Authored-by: Max Gekk max.gekk@gmail.com
(cherry picked from commit 156a12e)

[HyukjinKwon]

Merged to branch-3.4.