[SPARK-44329][CONNECT][PYTHON] Add hll_sketch_agg, hll_union_agg, to_varchar, try_aes_decrypt to Scala and Python

### What changes were proposed in this pull request?
Add following functions:

- hll_sketch_agg
- hll_union_agg
- to_varchar
- try_aes_decrypt

to:

- Scala API
- Python API
- Spark Connect Scala Client
- Spark Connect Python Client

### Why are the changes needed?
for parity

### Does this PR introduce _any_ user-facing change?
Yes, new functions.

### How was this patch tested?
- Add New UT.

Closes #41907 from panbingkun/SPARK-44329.

Authored-by: panbingkun <[email protected]>
Signed-off-by: Ruifeng Zheng <[email protected]>

Author: panbingkun

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/functions.scala

@@ -607,6 +607,16 @@ object functions {
   def grouping_id(colName: String, colNames: String*): Column =
     grouping_id((Seq(colName) ++ colNames).map(n => Column(n)): _*)
 
+  /**
+   * Aggregate function: returns the updatable binary representation of the Datasketches HllSketch
+   * configured with lgConfigK arg.
+   *
+   * @group agg_funcs
+   * @since 3.5.0
+   */
+  def hll_sketch_agg(e: Column, lgConfigK: Column): Column =
+    Column.fn("hll_sketch_agg", e, lgConfigK)
+
   /**
    * Aggregate function: returns the updatable binary representation of the Datasketches HllSketch
    * configured with lgConfigK arg.
@@ -646,6 +656,18 @@ object functions {
    */
   def hll_sketch_agg(columnName: String): Column = hll_sketch_agg(Column(columnName))
 
+  /**
+   * Aggregate function: returns the updatable binary representation of the Datasketches
+   * HllSketch, generated by merging previously created Datasketches HllSketch instances via a
+   * Datasketches Union instance. Throws an exception if sketches have different lgConfigK values
+   * and allowDifferentLgConfigK is set to false.
+   *
+   * @group agg_funcs
+   * @since 3.5.0
+   */
+  def hll_union_agg(e: Column, allowDifferentLgConfigK: Column): Column =
+    Column.fn("hll_union_agg", e, allowDifferentLgConfigK)
+
   /**
    * Aggregate function: returns the updatable binary representation of the Datasketches
    * HllSketch, generated by merging previously created Datasketches HllSketch instances via a
@@ -3491,6 +3513,72 @@ object functions {
   def aes_decrypt(input: Column, key: Column): Column =
     Column.fn("aes_encrypt", input, key)
 
+  /**
+   * This is a special version of `aes_decrypt` that performs the same operation, but returns a
+   * NULL value instead of raising an error if the decryption cannot be performed.
+   *
+   * @param input
+   *   The binary value to decrypt.
+   * @param key
+   *   The passphrase to use to decrypt the data.
+   * @param mode
+   *   Specifies which block cipher mode should be used to decrypt messages. Valid modes: ECB,
+   *   GCM, CBC.
+   * @param padding
+   *   Specifies how to pad messages whose length is not a multiple of the block size. Valid
+   *   values: PKCS, NONE, DEFAULT. The DEFAULT padding means PKCS for ECB, NONE for GCM and PKCS
+   *   for CBC.
+   * @param aad
+   *   Optional additional authenticated data. Only supported for GCM mode. This can be any
+   *   free-form input and must be provided for both encryption and decryption.
+   *
+   * @group misc_funcs
+   * @since 3.5.0
+   */
+  def try_aes_decrypt(
+      input: Column,
+      key: Column,
+      mode: Column,
+      padding: Column,
+      aad: Column): Column =
+    Column.fn("try_aes_decrypt", input, key, mode, padding, aad)
+
+  /**
+   * Returns a decrypted value of `input`.
+   *
+   * @see
+   *   `org.apache.spark.sql.functions.try_aes_decrypt(Column, Column, Column, Column, Column)`
+   *
+   * @group misc_funcs
+   * @since 3.5.0
+   */
+  def try_aes_decrypt(input: Column, key: Column, mode: Column, padding: Column): Column =
+    Column.fn("try_aes_decrypt", input, key, mode, padding)
+
+  /**
+   * Returns a decrypted value of `input`.
+   *
+   * @see
+   *   `org.apache.spark.sql.functions.try_aes_decrypt(Column, Column, Column, Column, Column)`
+   *
+   * @group misc_funcs
+   * @since 3.5.0
+   */
+  def try_aes_decrypt(input: Column, key: Column, mode: Column): Column =
+    Column.fn("try_aes_decrypt", input, key, mode)
+
+  /**
+   * Returns a decrypted value of `input`.
+   *
+   * @see
+   *   `org.apache.spark.sql.functions.try_aes_decrypt(Column, Column, Column, Column, Column)`
+   *
+   * @group misc_funcs
+   * @since 3.5.0
+   */
+  def try_aes_decrypt(input: Column, key: Column): Column =
+    Column.fn("try_aes_decrypt", input, key)
+
   /**
    * Returns a sha1 hash value as a hex string of the `col`.
    *
@@ -4192,6 +4280,33 @@ object functions {
    */
   def to_char(e: Column, format: Column): Column = Column.fn("to_char", e, format)
 
+  /**
+   * Convert `e` to a string based on the `format`. Throws an exception if the conversion fails.
+   *
+   * @param e
+   *   A column of number to be converted
+   * @param format
+   *   The format can consist of the following characters, case insensitive: <ul> <li> '0' or '9':
+   *   Specifies an expected digit between 0 and 9. A sequence of 0 or 9 in the format string
+   *   matches a sequence of digits in the input value, generating a result string of the same
+   *   length as the corresponding sequence in the format string. The result string is left-padded
+   *   with zeros if the 0/9 sequence comprises more digits than the matching part of the decimal
+   *   value, starts with 0, and is before the decimal point. Otherwise, it is padded with
+   *   spaces.</li> <li>'.' or 'D': Specifies the position of the decimal point (optional, only
+   *   allowed once).</li> <li>',' or 'G': Specifies the position of the grouping (thousands)
+   *   separator (,). There must be a 0 or 9 to the left and right of each grouping
+   *   separator.</li> <li>'$': Specifies the location of the $ currency sign. This character may
+   *   only be specified once.</li> <li>'S' or 'MI': Specifies the position of a '-' or '+' sign
+   *   (optional, only allowed once at the beginning or end of the format string). Note that 'S'
+   *   prints '+' for positive values but 'MI' prints a space.</li> <li>'PR': Only allowed at the
+   *   end of the format string; specifies that the result string will be wrapped by angle
+   *   brackets if the input value is negative.</li> </ul>
+   *
+   * @group string_funcs
+   * @since 3.5.0
+   */
+  def to_varchar(e: Column, format: Column): Column = Column.fn("to_varchar", e, format)
+
   /**
    * Convert string 'e' to a number based on the string format 'format'. Throws an exception if
    * the conversion fails.

connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala

@@ -2550,6 +2550,10 @@ class PlanGenerationTestSuite
     fn.to_char(fn.col("b"), lit("$99.99"))
   }
 
+  functionTest("to_varchar") {
+    fn.to_varchar(fn.col("b"), lit("$99.99"))
+  }
+
   functionTest("to_number") {
     fn.to_number(fn.col("g"), lit("$99.99"))
   }
@@ -2821,6 +2825,22 @@ class PlanGenerationTestSuite
     fn.aes_decrypt(fn.col("g"), fn.col("g"))
   }
 
+  functionTest("try_aes_decrypt with mode padding aad") {
+    fn.try_aes_decrypt(fn.col("g"), fn.col("g"), fn.col("g"), fn.col("g"), fn.col("g"))
+  }
+
+  functionTest("try_aes_decrypt with mode padding") {
+    fn.try_aes_decrypt(fn.col("g"), fn.col("g"), fn.col("g"), fn.col("g"))
+  }
+
+  functionTest("try_aes_decrypt with mode") {
+    fn.try_aes_decrypt(fn.col("g"), fn.col("g"), fn.col("g"))
+  }
+
+  functionTest("try_aes_decrypt") {
+    fn.try_aes_decrypt(fn.col("g"), fn.col("g"))
+  }
+
   functionTest("sha") {
     fn.sha(fn.col("g"))
   }
@@ -2853,6 +2873,46 @@ class PlanGenerationTestSuite
     fn.random(lit(1))
   }
 
+  test("hll_sketch_agg with column lgConfigK") {
+    binary.select(fn.hll_sketch_agg(fn.col("bytes"), lit(0)))
+  }
+
+  test("hll_sketch_agg with column lgConfigK_int") {
+    binary.select(fn.hll_sketch_agg(fn.col("bytes"), 0))
+  }
+
+  test("hll_sketch_agg with columnName lgConfigK_int") {
+    binary.select(fn.hll_sketch_agg("bytes", 0))
+  }
+
+  test("hll_sketch_agg") {
+    binary.select(fn.hll_sketch_agg(fn.col("bytes")))
+  }
+
+  test("hll_sketch_agg with columnName") {
+    binary.select(fn.hll_sketch_agg("bytes"))
+  }
+
+  test("hll_union_agg with column allowDifferentLgConfigK") {
+    binary.select(fn.hll_union_agg(fn.col("bytes"), lit(false)))
+  }
+
+  test("hll_union_agg with column allowDifferentLgConfigK_boolean") {
+    binary.select(fn.hll_union_agg(fn.col("bytes"), false))
+  }
+
+  test("hll_union_agg with columnName allowDifferentLgConfigK_boolean") {
+    binary.select(fn.hll_union_agg("bytes", false))
+  }
+
+  test("hll_union_agg") {
+    binary.select(fn.hll_union_agg(fn.col("bytes")))
+  }
+
+  test("hll_union_agg with columnName") {
+    binary.select(fn.hll_union_agg("bytes"))
+  }
+
   test("groupby agg") {
     simple
       .groupBy(Column("id"))

connector/connect/common/src/test/resources/query-tests/explain-results/function_to_varchar.explain

@@ -0,0 +1,2 @@
+Project [to_char(cast(b#0 as decimal(30,15)), $99.99) AS to_char(b, $99.99)#0]
++- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]

connector/connect/common/src/test/resources/query-tests/explain-results/function_try_aes_decrypt.explain

@@ -0,0 +1,2 @@
+Project [tryeval(staticinvoke(class org.apache.spark.sql.catalyst.expressions.ExpressionImplUtils, BinaryType, aesDecrypt, cast(g#0 as binary), cast(g#0 as binary), GCM, DEFAULT, cast( as binary), BinaryType, BinaryType, StringType, StringType, BinaryType, true, true, true)) AS try_aes_decrypt(g, g, GCM, DEFAULT, )#0]
++- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]

connector/connect/common/src/test/resources/query-tests/explain-results/function_try_aes_decrypt_with_mode.explain

@@ -0,0 +1,2 @@
+Project [tryeval(staticinvoke(class org.apache.spark.sql.catalyst.expressions.ExpressionImplUtils, BinaryType, aesDecrypt, cast(g#0 as binary), cast(g#0 as binary), g#0, DEFAULT, cast( as binary), BinaryType, BinaryType, StringType, StringType, BinaryType, true, true, true)) AS try_aes_decrypt(g, g, g, DEFAULT, )#0]
++- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]

connector/connect/common/src/test/resources/query-tests/explain-results/function_try_aes_decrypt_with_mode_padding.explain

@@ -0,0 +1,2 @@
+Project [tryeval(staticinvoke(class org.apache.spark.sql.catalyst.expressions.ExpressionImplUtils, BinaryType, aesDecrypt, cast(g#0 as binary), cast(g#0 as binary), g#0, g#0, cast( as binary), BinaryType, BinaryType, StringType, StringType, BinaryType, true, true, true)) AS try_aes_decrypt(g, g, g, g, )#0]
++- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]

connector/connect/common/src/test/resources/query-tests/explain-results/function_try_aes_decrypt_with_mode_padding_aad.explain

@@ -0,0 +1,2 @@
+Project [tryeval(staticinvoke(class org.apache.spark.sql.catalyst.expressions.ExpressionImplUtils, BinaryType, aesDecrypt, cast(g#0 as binary), cast(g#0 as binary), g#0, g#0, cast(g#0 as binary), BinaryType, BinaryType, StringType, StringType, BinaryType, true, true, true)) AS try_aes_decrypt(g, g, g, g, g)#0]
++- LocalRelation <empty>, [id#0L, a#0, b#0, d#0, e#0, f#0, g#0]

connector/connect/common/src/test/resources/query-tests/explain-results/hll_sketch_agg.explain

@@ -0,0 +1,2 @@
+Aggregate [hll_sketch_agg(bytes#0, 12, 0, 0) AS hll_sketch_agg(bytes, 12)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]

connector/connect/common/src/test/resources/query-tests/explain-results/hll_sketch_agg_with_columnName.explain

@@ -0,0 +1,2 @@
+Aggregate [hll_sketch_agg(bytes#0, 12, 0, 0) AS hll_sketch_agg(bytes, 12)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]

connector/connect/common/src/test/resources/query-tests/explain-results/hll_sketch_agg_with_columnName_lgConfigK_int.explain

@@ -0,0 +1,2 @@
+Aggregate [hll_sketch_agg(bytes#0, 0, 0, 0) AS hll_sketch_agg(bytes, 0)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]

connector/connect/common/src/test/resources/query-tests/explain-results/hll_sketch_agg_with_column_lgConfigK.explain

@@ -0,0 +1,2 @@
+Aggregate [hll_sketch_agg(bytes#0, 0, 0, 0) AS hll_sketch_agg(bytes, 0)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]

connector/connect/common/src/test/resources/query-tests/explain-results/hll_sketch_agg_with_column_lgConfigK_int.explain

@@ -0,0 +1,2 @@
+Aggregate [hll_sketch_agg(bytes#0, 0, 0, 0) AS hll_sketch_agg(bytes, 0)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]

connector/connect/common/src/test/resources/query-tests/explain-results/hll_union_agg.explain

@@ -0,0 +1,2 @@
+Aggregate [hll_union_agg(bytes#0, false, 0, 0) AS hll_union_agg(bytes, false)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]

connector/connect/common/src/test/resources/query-tests/explain-results/hll_union_agg_with_columnName.explain

@@ -0,0 +1,2 @@
+Aggregate [hll_union_agg(bytes#0, false, 0, 0) AS hll_union_agg(bytes, false)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]

connector/connect/common/src/test/resources/query-tests/explain-results/hll_union_agg_with_columnName_allowDifferentLgConfigK_boolean.explain

@@ -0,0 +1,2 @@
+Aggregate [hll_union_agg(bytes#0, false, 0, 0) AS hll_union_agg(bytes, false)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]

connector/connect/common/src/test/resources/query-tests/explain-results/hll_union_agg_with_column_allowDifferentLgConfigK.explain

@@ -0,0 +1,2 @@
+Aggregate [hll_union_agg(bytes#0, false, 0, 0) AS hll_union_agg(bytes, false)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]

connector/connect/common/src/test/resources/query-tests/explain-results/hll_union_agg_with_column_allowDifferentLgConfigK_boolean.explain

@@ -0,0 +1,2 @@
+Aggregate [hll_union_agg(bytes#0, false, 0, 0) AS hll_union_agg(bytes, false)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]

connector/connect/common/src/test/resources/query-tests/queries/function_to_varchar.json

@@ -0,0 +1,29 @@
+{
+  "common": {
+    "planId": "1"
+  },
+  "project": {
+    "input": {
+      "common": {
+        "planId": "0"
+      },
+      "localRelation": {
+        "schema": "struct\u003cid:bigint,a:int,b:double,d:struct\u003cid:bigint,a:int,b:double\u003e,e:array\u003cint\u003e,f:map\u003cstring,struct\u003cid:bigint,a:int,b:double\u003e\u003e,g:string\u003e"
+      }
+    },
+    "expressions": [{
+      "unresolvedFunction": {
+        "functionName": "to_varchar",
+        "arguments": [{
+          "unresolvedAttribute": {
+            "unparsedIdentifier": "b"
+          }
+        }, {
+          "literal": {
+            "string": "$99.99"
+          }
+        }]
+      }
+    }]
+  }
+}

connector/connect/common/src/test/resources/query-tests/queries/function_to_varchar.proto.bin

@@ -0,0 +1,29 @@
+{
+  "common": {
+    "planId": "1"
+  },
+  "project": {
+    "input": {
+      "common": {
+        "planId": "0"
+      },
+      "localRelation": {
+        "schema": "struct\u003cid:bigint,a:int,b:double,d:struct\u003cid:bigint,a:int,b:double\u003e,e:array\u003cint\u003e,f:map\u003cstring,struct\u003cid:bigint,a:int,b:double\u003e\u003e,g:string\u003e"
+      }
+    },
+    "expressions": [{
+      "unresolvedFunction": {
+        "functionName": "try_aes_decrypt",
+        "arguments": [{
+          "unresolvedAttribute": {
+            "unparsedIdentifier": "g"
+          }
+        }, {
+          "unresolvedAttribute": {
+            "unparsedIdentifier": "g"
+          }
+        }]
+      }
+    }]
+  }
+}

connector/connect/common/src/test/resources/query-tests/queries/function_try_aes_decrypt.json

@@ -0,0 +1,33 @@
+{
+  "common": {
+    "planId": "1"
+  },
+  "project": {
+    "input": {
+      "common": {
+        "planId": "0"
+      },
+      "localRelation": {
+        "schema": "struct\u003cid:bigint,a:int,b:double,d:struct\u003cid:bigint,a:int,b:double\u003e,e:array\u003cint\u003e,f:map\u003cstring,struct\u003cid:bigint,a:int,b:double\u003e\u003e,g:string\u003e"
+      }
+    },
+    "expressions": [{
+      "unresolvedFunction": {
+        "functionName": "try_aes_decrypt",
+        "arguments": [{
+          "unresolvedAttribute": {
+            "unparsedIdentifier": "g"
+          }
+        }, {
+          "unresolvedAttribute": {
+            "unparsedIdentifier": "g"
+          }
+        }, {
+          "unresolvedAttribute": {
+            "unparsedIdentifier": "g"
+          }
+        }]
+      }
+    }]
+  }
+}