[SPARK-44331][CONNECT][PYTHON] Add bitmap functions to Scala and Python

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

- bitmap_bucket_number
- bitmap_bit_position
- bitmap_construct_agg
- bitmap_count
- bitmap_or_agg

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 #41902 from panbingkun/SPARK-44331.

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

@@ -3575,6 +3575,51 @@ object functions {
    */
   def random(): Column = Column.fn("random")
 
+  /**
+   * Returns the bit position for the given input column.
+   *
+   * @group misc_funcs
+   * @since 3.5.0
+   */
+  def bitmap_bit_position(col: Column): Column =
+    Column.fn("bitmap_bit_position", col)
+
+  /**
+   * Returns the bucket number for the given input column.
+   *
+   * @group misc_funcs
+   * @since 3.5.0
+   */
+  def bitmap_bucket_number(col: Column): Column =
+    Column.fn("bitmap_bucket_number", col)
+
+  /**
+   * Returns a bitmap with the positions of the bits set from all the values from the input
+   * column. The input column will most likely be bitmap_bit_position().
+   *
+   * @group misc_funcs
+   * @since 3.5.0
+   */
+  def bitmap_construct_agg(col: Column): Column =
+    Column.fn("bitmap_construct_agg", col)
+
+  /**
+   * Returns the number of set bits in the input bitmap.
+   *
+   * @group misc_funcs
+   * @since 3.5.0
+   */
+  def bitmap_count(col: Column): Column = Column.fn("bitmap_count", col)
+
+  /**
+   * Returns a bitmap that is the bitwise OR of all of the bitmaps from the input column. The
+   * input column should be bitmaps created from bitmap_construct_agg().
+   *
+   * @group misc_funcs
+   * @since 3.5.0
+   */
+  def bitmap_or_agg(col: Column): Column = Column.fn("bitmap_or_agg", col)
+
   //////////////////////////////////////////////////////////////////////////////////////////////
   // String functions
   //////////////////////////////////////////////////////////////////////////////////////////////

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

@@ -1979,6 +1979,26 @@ class PlanGenerationTestSuite
     fn.row_number().over(Window.partitionBy(Column("a")).orderBy(Column("id")))
   }
 
+  functionTest("bitmap_bucket_number") {
+    fn.bitmap_bit_position(fn.col("id"))
+  }
+
+  functionTest("bitmap_bit_position") {
+    fn.bitmap_bit_position(fn.col("id"))
+  }
+
+  functionTest("bitmap_construct_agg") {
+    fn.bitmap_construct_agg(fn.col("id"))
+  }
+
+  test("function bitmap_count") {
+    binary.select(fn.bitmap_count(fn.col("bytes")))
+  }
+
+  test("function bitmap_or_agg") {
+    binary.select(fn.bitmap_or_agg(fn.col("bytes")))
+  }
+
   private def temporalFunctionTest(name: String)(f: => Column): Unit = {
     test("function " + name) {
       temporals.select(f)

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

@@ -0,0 +1,2 @@
+Project [staticinvoke(class org.apache.spark.sql.catalyst.expressions.BitmapExpressionUtils, LongType, bitmapBitPosition, id#0L, LongType, true, false, true) AS bitmap_bit_position(id)#0L]
++- 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_bitmap_bucket_number.explain

@@ -0,0 +1,2 @@
+Project [staticinvoke(class org.apache.spark.sql.catalyst.expressions.BitmapExpressionUtils, LongType, bitmapBitPosition, id#0L, LongType, true, false, true) AS bitmap_bit_position(id)#0L]
++- 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_bitmap_construct_agg.explain

@@ -0,0 +1,2 @@
+Aggregate [bitmap_construct_agg(id#0L, 0, 0) AS bitmap_construct_agg(id)#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_bitmap_count.explain

@@ -0,0 +1,2 @@
+Project [staticinvoke(class org.apache.spark.sql.catalyst.expressions.BitmapExpressionUtils, LongType, bitmapCount, bytes#0, BinaryType, true, false, true) AS bitmap_count(bytes)#0L]
++- LocalRelation <empty>, [id#0L, bytes#0]

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

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

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

@@ -0,0 +1,25 @@
+{
+  "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": "bitmap_bit_position",
+        "arguments": [{
+          "unresolvedAttribute": {
+            "unparsedIdentifier": "id"
+          }
+        }]
+      }
+    }]
+  }
+}

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

@@ -0,0 +1,25 @@
+{
+  "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": "bitmap_bit_position",
+        "arguments": [{
+          "unresolvedAttribute": {
+            "unparsedIdentifier": "id"
+          }
+        }]
+      }
+    }]
+  }
+}

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

@@ -0,0 +1,25 @@
+{
+  "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": "bitmap_construct_agg",
+        "arguments": [{
+          "unresolvedAttribute": {
+            "unparsedIdentifier": "id"
+          }
+        }]
+      }
+    }]
+  }
+}

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

@@ -0,0 +1,25 @@
+{
+  "common": {
+    "planId": "1"
+  },
+  "project": {
+    "input": {
+      "common": {
+        "planId": "0"
+      },
+      "localRelation": {
+        "schema": "struct\u003cid:bigint,bytes:binary\u003e"
+      }
+    },
+    "expressions": [{
+      "unresolvedFunction": {
+        "functionName": "bitmap_count",
+        "arguments": [{
+          "unresolvedAttribute": {
+            "unparsedIdentifier": "bytes"
+          }
+        }]
+      }
+    }]
+  }
+}

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

@@ -0,0 +1,25 @@
+{
+  "common": {
+    "planId": "1"
+  },
+  "project": {
+    "input": {
+      "common": {
+        "planId": "0"
+      },
+      "localRelation": {
+        "schema": "struct\u003cid:bigint,bytes:binary\u003e"
+      }
+    },
+    "expressions": [{
+      "unresolvedFunction": {
+        "functionName": "bitmap_or_agg",
+        "arguments": [{
+          "unresolvedAttribute": {
+            "unparsedIdentifier": "bytes"
+          }
+        }]
+      }
+    }]
+  }
+}

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

@@ -477,6 +477,11 @@ Misc Functions
 
     aes_decrypt
     aes_encrypt
+    bitmap_bit_position
+    bitmap_bucket_number
+    bitmap_construct_agg
+    bitmap_count
+    bitmap_or_agg
     current_catalog
     current_database
     current_schema

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

@@ -3788,6 +3788,41 @@ def random(
 random.__doc__ = pysparkfuncs.random.__doc__
 
 
+def bitmap_bit_position(col: "ColumnOrName") -> Column:
+    return _invoke_function_over_columns("bitmap_bit_position", col)
+
+
+bitmap_bit_position.__doc__ = pysparkfuncs.bitmap_bit_position.__doc__
+
+
+def bitmap_bucket_number(col: "ColumnOrName") -> Column:
+    return _invoke_function_over_columns("bitmap_bucket_number", col)
+
+
+bitmap_bucket_number.__doc__ = pysparkfuncs.bitmap_bucket_number.__doc__
+
+
+def bitmap_construct_agg(col: "ColumnOrName") -> Column:
+    return _invoke_function_over_columns("bitmap_construct_agg", col)
+
+
+bitmap_construct_agg.__doc__ = pysparkfuncs.bitmap_construct_agg.__doc__
+
+
+def bitmap_count(col: "ColumnOrName") -> Column:
+    return _invoke_function_over_columns("bitmap_count", col)
+
+
+bitmap_count.__doc__ = pysparkfuncs.bitmap_count.__doc__
+
+
+def bitmap_or_agg(col: "ColumnOrName") -> Column:
+    return _invoke_function_over_columns("bitmap_or_agg", col)
+
+
+bitmap_or_agg.__doc__ = pysparkfuncs.bitmap_or_agg.__doc__
+
+
 # User Defined Function