You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/21 02:34:34 UTC
(spark) branch branch-4.x updated: [SPARK-58831][SQL] Add bitmap scalar set operation functions
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new d20be69d9c07 [SPARK-58831][SQL] Add bitmap scalar set operation functions
d20be69d9c07 is described below
commit d20be69d9c07d0195288f73361db63c6d189701b
Author: StormSpirit <ji...@vip.qq.com>
AuthorDate: Fri Aug 21 10:33:58 2026 +0800
[SPARK-58831][SQL] Add bitmap scalar set operation functions
### What changes were proposed in this pull request?
This PR adds four scalar set operations for Spark's flat bitmap representation:
- `bitmap_and`
- `bitmap_or`
- `bitmap_andnot`
- `bitmap_xor`
Each function operates on two `BINARY` values from the same row. The implementation:
- accepts inputs up to 4096 bytes;
- treats missing trailing bytes as zero;
- returns a newly allocated, fixed-size 4096-byte bitmap;
- propagates `NULL` inputs;
- rejects oversized inputs with the structured `BITMAP_INPUT_TOO_LARGE` error;
- does not mutate either input.
The functions are implemented as Catalyst-native binary expressions with both interpreted and whole-stage codegen paths. They are registered in the SQL function registry and exposed through the Scala API, PySpark classic, and Spark Connect.
This PR also adds SQL documentation, Python documentation, a structured error definition, Connect plan and schema golden files, and tests for the new APIs.
### Why are the changes needed?
Spark provides functions for constructing, aggregating, and counting flat bitmaps, but it does not provide scalar set operations for combining two bitmap values from the same row.
Without native scalar functions, users need UDFs or external bitmap processing to compute intersection, union, difference, or symmetric difference between precomputed bitmap columns. Native Catalyst expressions provide consistent SQL semantics, structured validation, and code generation, and allow these operations to compose naturally with the existing bitmap aggregate and count functions.
### Does this PR introduce _any_ user-facing change?
Yes. It adds four new SQL, Scala, PySpark classic, and Spark Connect functions:
```sql
SELECT bitmap_and(left_bitmap, right_bitmap);
SELECT bitmap_or(left_bitmap, right_bitmap);
SELECT bitmap_andnot(left_bitmap, right_bitmap);
SELECT bitmap_xor(left_bitmap, right_bitmap);
```
`bitmap_andnot(left, right)` is directional and returns the bits present in `left` but not in `right`.
No existing function behavior is changed.
### How was this patch tested?
The following targeted suites and build checks passed:
- `build/sbt "catalyst/testOnly org.apache.spark.sql.catalyst.expressions.BitmapExpressionUtilsSuite"` (10 tests)
- `build/sbt "sql/testOnly org.apache.spark.sql.BitmapExpressionsQuerySuite"` (19 tests, including interpreted and codegen execution)
- `build/sbt "sql/testOnly org.apache.spark.sql.ExpressionsSchemaSuite"`
- `build/sbt "connect-client-jvm/testOnly org.apache.spark.sql.PlanGenerationTestSuite -- -z 'function bitmap'"` (9 tests)
- `build/sbt -Phive package`
- `python/run-tests --testnames "pyspark.sql.tests.test_functions FunctionsTests.test_bitmap_scalar_functions"`
- `python/run-tests --testnames "pyspark.sql.tests.connect.test_connect_plan SparkConnectPlanTests.test_bitmap_scalar_functions"`
- Targeted PySpark doctests for the four new functions (12 tests)
The tests cover variable-length and empty inputs, the 4096-byte boundary, input immutability, null propagation, invalid types, oversized inputs on either side, interpreted and codegen execution, aggregation composition, grouped queries, PySpark classic, and Spark Connect.
Scalafmt, Ruff, JSON validation, golden-file consistency checks, and `git diff --check` also passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex and Claude AI
Closes #58066 from jiangxt2/feat/bitmap-scalar-set-operations.
Authored-by: StormSpirit <ji...@vip.qq.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit cc75de63a45694f8e70c60cd3387ddf8a83cf80b)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../src/main/resources/error/error-conditions.json | 6 +
.../source/reference/pyspark.sql/functions.rst | 4 +
python/pyspark/sql/connect/functions/builtin.py | 28 +++
python/pyspark/sql/functions/__init__.py | 4 +
python/pyspark/sql/functions/builtin.py | 188 ++++++++++++++++++
.../pyspark/sql/tests/connect/test_connect_plan.py | 26 ++-
python/pyspark/sql/tests/test_functions.py | 12 ++
.../scala/org/apache/spark/sql/functions.scala | 85 ++++++++
.../expressions/BitmapExpressionUtils.java | 43 ++++
.../sql/catalyst/analysis/FunctionRegistry.scala | 4 +
.../catalyst/expressions/bitmapExpressions.scala | 217 +++++++++++++++++++++
.../spark/sql/errors/QueryExecutionErrors.scala | 9 +
.../expressions/BitmapExpressionUtilsSuite.scala | 91 +++++++++
.../apache/spark/sql/PlanGenerationTestSuite.scala | 16 ++
.../explain-results/function_bitmap_and.explain | 2 +
.../explain-results/function_bitmap_andnot.explain | 2 +
.../explain-results/function_bitmap_or.explain | 2 +
.../explain-results/function_bitmap_xor.explain | 2 +
.../query-tests/queries/function_bitmap_and.json | 81 ++++++++
.../queries/function_bitmap_and.proto.bin | Bin 0 -> 629 bytes
.../queries/function_bitmap_andnot.json | 81 ++++++++
.../queries/function_bitmap_andnot.proto.bin | Bin 0 -> 635 bytes
.../query-tests/queries/function_bitmap_or.json | 81 ++++++++
.../queries/function_bitmap_or.proto.bin | Bin 0 -> 627 bytes
.../query-tests/queries/function_bitmap_xor.json | 81 ++++++++
.../queries/function_bitmap_xor.proto.bin | Bin 0 -> 629 bytes
.../sql-functions/sql-expression-schema.md | 4 +
.../spark/sql/BitmapExpressionsQuerySuite.scala | 168 ++++++++++++++++
28 files changed, 1236 insertions(+), 1 deletion(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index b502818c85cc..6a052c63c988 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -489,6 +489,12 @@
],
"sqlState" : "0A000"
},
+ "BITMAP_INPUT_TOO_LARGE" : {
+ "message" : [
+ "The input bitmap has <inputNumBytes> bytes, which exceeds the maximum supported size of <maxNumBytes> bytes."
+ ],
+ "sqlState" : "22001"
+ },
"CALL_ON_STREAMING_DATASET_UNSUPPORTED" : {
"message" : [
"The method <methodName> can not be called on streaming Dataset/DataFrame."
diff --git a/python/docs/source/reference/pyspark.sql/functions.rst b/python/docs/source/reference/pyspark.sql/functions.rst
index 661fc086c4e5..031c48734c49 100644
--- a/python/docs/source/reference/pyspark.sql/functions.rst
+++ b/python/docs/source/reference/pyspark.sql/functions.rst
@@ -665,9 +665,13 @@ Misc Functions
aes_decrypt
aes_encrypt
assert_true
+ bitmap_and
+ bitmap_andnot
bitmap_bit_position
bitmap_bucket_number
bitmap_count
+ bitmap_or
+ bitmap_xor
current_catalog
current_database
current_path
diff --git a/python/pyspark/sql/connect/functions/builtin.py b/python/pyspark/sql/connect/functions/builtin.py
index 2e696f344b12..9d622925bf19 100644
--- a/python/pyspark/sql/connect/functions/builtin.py
+++ b/python/pyspark/sql/connect/functions/builtin.py
@@ -5663,6 +5663,34 @@ def bitmap_count(col: "ColumnOrName") -> Column:
bitmap_count.__doc__ = pysparkfuncs.bitmap_count.__doc__
+def bitmap_and(left: "ColumnOrName", right: "ColumnOrName") -> Column:
+ return _invoke_function_over_columns("bitmap_and", left, right)
+
+
+bitmap_and.__doc__ = pysparkfuncs.bitmap_and.__doc__
+
+
+def bitmap_or(left: "ColumnOrName", right: "ColumnOrName") -> Column:
+ return _invoke_function_over_columns("bitmap_or", left, right)
+
+
+bitmap_or.__doc__ = pysparkfuncs.bitmap_or.__doc__
+
+
+def bitmap_andnot(left: "ColumnOrName", right: "ColumnOrName") -> Column:
+ return _invoke_function_over_columns("bitmap_andnot", left, right)
+
+
+bitmap_andnot.__doc__ = pysparkfuncs.bitmap_andnot.__doc__
+
+
+def bitmap_xor(left: "ColumnOrName", right: "ColumnOrName") -> Column:
+ return _invoke_function_over_columns("bitmap_xor", left, right)
+
+
+bitmap_xor.__doc__ = pysparkfuncs.bitmap_xor.__doc__
+
+
def bitmap_or_agg(col: "ColumnOrName") -> Column:
return _invoke_function_over_columns("bitmap_or_agg", col)
diff --git a/python/pyspark/sql/functions/__init__.py b/python/pyspark/sql/functions/__init__.py
index a3c0b7016942..462509a5d7a2 100644
--- a/python/pyspark/sql/functions/__init__.py
+++ b/python/pyspark/sql/functions/__init__.py
@@ -526,9 +526,13 @@ __all__ = [ # noqa: F405
"aes_decrypt",
"aes_encrypt",
"assert_true",
+ "bitmap_and",
+ "bitmap_andnot",
"bitmap_bit_position",
"bitmap_bucket_number",
"bitmap_count",
+ "bitmap_or",
+ "bitmap_xor",
"current_catalog",
"current_database",
"current_path",
diff --git a/python/pyspark/sql/functions/builtin.py b/python/pyspark/sql/functions/builtin.py
index dbe06f837c3b..794bc9da206f 100644
--- a/python/pyspark/sql/functions/builtin.py
+++ b/python/pyspark/sql/functions/builtin.py
@@ -33043,6 +33043,194 @@ def bitmap_count(col: "ColumnOrName") -> Column:
return _invoke_function_over_columns("bitmap_count", col)
+@_try_remote_functions
+def bitmap_and(left: "ColumnOrName", right: "ColumnOrName") -> Column:
+ """
+ Returns a bitmap that is the bitwise AND of two input bitmaps.
+
+ .. versionadded:: 4.4.0
+
+ Parameters
+ ----------
+ left : :class:`~pyspark.sql.Column` or column name
+ The left input bitmap.
+ right : :class:`~pyspark.sql.Column` or column name
+ The right input bitmap.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.bitmap_andnot`
+ :meth:`pyspark.sql.functions.bitmap_or`
+ :meth:`pyspark.sql.functions.bitmap_xor`
+
+ Notes
+ -----
+ Inputs use Spark's Binary bitmap representation, not a RoaringBitmap serialization. Each
+ input may contain 0 to 4096 bytes; missing bytes are treated as zero. The result is always a
+ 4096-byte Binary value. NULL input returns NULL, and inputs longer than 4096 bytes raise
+ ``BITMAP_INPUT_TOO_LARGE``. Both inputs must use the same bit-position mapping. If they were
+ constructed by grouping ``bitmap_bit_position`` values by ``bitmap_bucket_number``, they must
+ represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ function combines two bitmaps from the same row; use ``bitmap_*_agg`` to combine bitmaps across
+ rows.
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([("F0", "70")], ["left", "right"])
+ >>> df.select(sf.bitmap_and(
+ ... sf.to_binary("left", sf.lit("hex")),
+ ... sf.to_binary("right", sf.lit("hex"))).alias("bitmap")).show()
+ +--------------------+
+ | bitmap|
+ +--------------------+
+ |[70 00 00 00 00 0...|
+ +--------------------+
+ """
+ return _invoke_function_over_columns("bitmap_and", left, right)
+
+
+@_try_remote_functions
+def bitmap_or(left: "ColumnOrName", right: "ColumnOrName") -> Column:
+ """
+ Returns a bitmap that is the bitwise OR of two input bitmaps.
+
+ .. versionadded:: 4.4.0
+
+ Parameters
+ ----------
+ left : :class:`~pyspark.sql.Column` or column name
+ The left input bitmap.
+ right : :class:`~pyspark.sql.Column` or column name
+ The right input bitmap.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.bitmap_and`
+ :meth:`pyspark.sql.functions.bitmap_andnot`
+ :meth:`pyspark.sql.functions.bitmap_xor`
+
+ Notes
+ -----
+ Inputs use Spark's Binary bitmap representation, not a RoaringBitmap serialization. Each
+ input may contain 0 to 4096 bytes; missing bytes are treated as zero. The result is always a
+ 4096-byte Binary value. NULL input returns NULL, and inputs longer than 4096 bytes raise
+ ``BITMAP_INPUT_TOO_LARGE``. Both inputs must use the same bit-position mapping. If they were
+ constructed by grouping ``bitmap_bit_position`` values by ``bitmap_bucket_number``, they must
+ represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ function combines two bitmaps from the same row; use ``bitmap_*_agg`` to combine bitmaps across
+ rows.
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([("10", "20")], ["left", "right"])
+ >>> df.select(sf.bitmap_or(
+ ... sf.to_binary("left", sf.lit("hex")),
+ ... sf.to_binary("right", sf.lit("hex"))).alias("bitmap")).show()
+ +--------------------+
+ | bitmap|
+ +--------------------+
+ |[30 00 00 00 00 0...|
+ +--------------------+
+ """
+ return _invoke_function_over_columns("bitmap_or", left, right)
+
+
+@_try_remote_functions
+def bitmap_andnot(left: "ColumnOrName", right: "ColumnOrName") -> Column:
+ """
+ Returns a bitmap that is the bitwise AND NOT of two input bitmaps.
+
+ .. versionadded:: 4.4.0
+
+ Parameters
+ ----------
+ left : :class:`~pyspark.sql.Column` or column name
+ The left input bitmap.
+ right : :class:`~pyspark.sql.Column` or column name
+ The right input bitmap.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.bitmap_and`
+ :meth:`pyspark.sql.functions.bitmap_or`
+ :meth:`pyspark.sql.functions.bitmap_xor`
+
+ Notes
+ -----
+ Inputs use Spark's Binary bitmap representation, not a RoaringBitmap serialization. Each
+ input may contain 0 to 4096 bytes; missing bytes are treated as zero. The result is always a
+ 4096-byte Binary value. NULL input returns NULL, and inputs longer than 4096 bytes raise
+ ``BITMAP_INPUT_TOO_LARGE``. Both inputs must use the same bit-position mapping. If they were
+ constructed by grouping ``bitmap_bit_position`` values by ``bitmap_bucket_number``, they must
+ represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ function combines two bitmaps from the same row; use ``bitmap_*_agg`` to combine bitmaps across
+ rows.
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([("F0", "70")], ["left", "right"])
+ >>> df.select(sf.bitmap_andnot(
+ ... sf.to_binary("left", sf.lit("hex")),
+ ... sf.to_binary("right", sf.lit("hex"))).alias("bitmap")).show()
+ +--------------------+
+ | bitmap|
+ +--------------------+
+ |[80 00 00 00 00 0...|
+ +--------------------+
+ """
+ return _invoke_function_over_columns("bitmap_andnot", left, right)
+
+
+@_try_remote_functions
+def bitmap_xor(left: "ColumnOrName", right: "ColumnOrName") -> Column:
+ """
+ Returns a bitmap that is the bitwise XOR of two input bitmaps.
+
+ .. versionadded:: 4.4.0
+
+ Parameters
+ ----------
+ left : :class:`~pyspark.sql.Column` or column name
+ The left input bitmap.
+ right : :class:`~pyspark.sql.Column` or column name
+ The right input bitmap.
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.bitmap_and`
+ :meth:`pyspark.sql.functions.bitmap_andnot`
+ :meth:`pyspark.sql.functions.bitmap_or`
+
+ Notes
+ -----
+ Inputs use Spark's Binary bitmap representation, not a RoaringBitmap serialization. Each
+ input may contain 0 to 4096 bytes; missing bytes are treated as zero. The result is always a
+ 4096-byte Binary value. NULL input returns NULL, and inputs longer than 4096 bytes raise
+ ``BITMAP_INPUT_TOO_LARGE``. Both inputs must use the same bit-position mapping. If they were
+ constructed by grouping ``bitmap_bit_position`` values by ``bitmap_bucket_number``, they must
+ represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ function combines two bitmaps from the same row; use ``bitmap_*_agg`` to combine bitmaps across
+ rows.
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([("F0", "70")], ["left", "right"])
+ >>> df.select(sf.bitmap_xor(
+ ... sf.to_binary("left", sf.lit("hex")),
+ ... sf.to_binary("right", sf.lit("hex"))).alias("bitmap")).show()
+ +--------------------+
+ | bitmap|
+ +--------------------+
+ |[80 00 00 00 00 0...|
+ +--------------------+
+ """
+ return _invoke_function_over_columns("bitmap_xor", left, right)
+
+
@_try_remote_functions
def bitmap_or_agg(col: "ColumnOrName") -> Column:
"""
diff --git a/python/pyspark/sql/tests/connect/test_connect_plan.py b/python/pyspark/sql/tests/connect/test_connect_plan.py
index 5105a58b400b..d1f09c37cf93 100644
--- a/python/pyspark/sql/tests/connect/test_connect_plan.py
+++ b/python/pyspark/sql/tests/connect/test_connect_plan.py
@@ -44,7 +44,17 @@ if should_test_connect:
from pyspark.sql.connect.observation import Observation
from pyspark.sql.connect.readwriter import DataFrameReader
from pyspark.sql.connect.expressions import LiteralExpression
- from pyspark.sql.connect.functions import col, lit, max, min, sum
+ from pyspark.sql.connect.functions import (
+ bitmap_and,
+ bitmap_andnot,
+ bitmap_or,
+ bitmap_xor,
+ col,
+ lit,
+ max,
+ min,
+ sum,
+ )
from pyspark.sql.connect.types import pyspark_types_to_proto_types
from pyspark.sql.types import (
StringType,
@@ -71,6 +81,20 @@ class SparkConnectPlanTests(PlanOnlyTestFixture):
self.assertIsNotNone(plan.root, "Root relation must be set")
self.assertIsNotNone(plan.root.read)
+ def test_bitmap_scalar_functions(self):
+ df = self.connect.readTable(table_name=self.tbl_name)
+ plan = df.select(
+ bitmap_and(col("bytes"), col("bytes")),
+ bitmap_or(col("bytes"), col("bytes")),
+ bitmap_andnot(col("bytes"), col("bytes")),
+ bitmap_xor(col("bytes"), col("bytes")),
+ )._plan.to_proto(self.connect)
+ function_names = [
+ expression.unresolved_function.function_name
+ for expression in plan.root.project.expressions
+ ]
+ self.assertEqual(function_names, ["bitmap_and", "bitmap_or", "bitmap_andnot", "bitmap_xor"])
+
def test_join_using_columns(self):
left_input = self.connect.readTable(table_name=self.tbl_name)
right_input = self.connect.readTable(table_name=self.tbl_name)
diff --git a/python/pyspark/sql/tests/test_functions.py b/python/pyspark/sql/tests/test_functions.py
index da8c13537ab8..cc7463d87a16 100644
--- a/python/pyspark/sql/tests/test_functions.py
+++ b/python/pyspark/sql/tests/test_functions.py
@@ -938,6 +938,18 @@ class FunctionsTestsMixin:
df.select(getattr(F, name)(F.col("name"))),
)
+ def test_bitmap_scalar_functions(self):
+ df = self.spark.createDataFrame([("F00F", "70")], ["left", "right"])
+ left = F.to_binary("left", F.lit("hex"))
+ right = F.to_binary("right", F.lit("hex"))
+ actual = df.select(
+ F.substring(F.hex(F.bitmap_and(left, right)), 0, 4),
+ F.substring(F.hex(F.bitmap_or(left, right)), 0, 4),
+ F.substring(F.hex(F.bitmap_andnot(left, right)), 0, 4),
+ F.substring(F.hex(F.bitmap_xor(left, right)), 0, 4),
+ )
+ assertDataFrameEqual([Row("7000", "F00F", "800F", "800F")], actual)
+
def test_collation(self):
df = self.spark.createDataFrame([("a",), ("b",)], ["name"])
actual = df.select(F.collation(F.collate("name", "UNICODE"))).distinct()
diff --git a/sql/api/src/main/scala/org/apache/spark/sql/functions.scala b/sql/api/src/main/scala/org/apache/spark/sql/functions.scala
index 8fac413bbd6c..d3d8a47b34d3 100644
--- a/sql/api/src/main/scala/org/apache/spark/sql/functions.scala
+++ b/sql/api/src/main/scala/org/apache/spark/sql/functions.scala
@@ -7338,6 +7338,91 @@ object functions {
*/
def bitmap_count(col: Column): Column = Column.fn("bitmap_count", col)
+ /**
+ * Returns a bitmap that is the bitwise AND of two input bitmaps. The result is always a
+ * 4096-byte Spark Binary bitmap. If either input is NULL, the result is NULL. Missing bytes in
+ * shorter inputs are treated as zero, and inputs longer than 4096 bytes raise
+ * `BITMAP_INPUT_TOO_LARGE`. Both inputs must use the same bit-position mapping. If they were
+ * constructed by grouping `bitmap_bit_position` values by `bitmap_bucket_number`, they must
+ * represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ * function operates on two bitmaps from the same row; use `bitmap_*_agg` to combine bitmaps
+ * across rows. The representation is not a RoaringBitmap serialization.
+ *
+ * @param left
+ * A column that evaluates to a binary bitmap.
+ * @param right
+ * A column that evaluates to a binary bitmap.
+ * @group misc_funcs
+ * @since 4.4.0
+ * @return
+ * Returns a column that evaluates to a binary bitmap.
+ */
+ def bitmap_and(left: Column, right: Column): Column = Column.fn("bitmap_and", left, right)
+
+ /**
+ * Returns a bitmap that is the bitwise OR of two input bitmaps. The result is always a
+ * 4096-byte Spark Binary bitmap. If either input is NULL, the result is NULL. Missing bytes in
+ * shorter inputs are treated as zero, and inputs longer than 4096 bytes raise
+ * `BITMAP_INPUT_TOO_LARGE`. Both inputs must use the same bit-position mapping. If they were
+ * constructed by grouping `bitmap_bit_position` values by `bitmap_bucket_number`, they must
+ * represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ * function operates on two bitmaps from the same row; use `bitmap_*_agg` to combine bitmaps
+ * across rows. The representation is not a RoaringBitmap serialization.
+ *
+ * @param left
+ * A column that evaluates to a binary bitmap.
+ * @param right
+ * A column that evaluates to a binary bitmap.
+ * @group misc_funcs
+ * @since 4.4.0
+ * @return
+ * Returns a column that evaluates to a binary bitmap.
+ */
+ def bitmap_or(left: Column, right: Column): Column = Column.fn("bitmap_or", left, right)
+
+ /**
+ * Returns a bitmap that is the bitwise AND NOT of two input bitmaps. The result is always a
+ * 4096-byte Spark Binary bitmap. If either input is NULL, the result is NULL. Missing bytes in
+ * shorter inputs are treated as zero, and inputs longer than 4096 bytes raise
+ * `BITMAP_INPUT_TOO_LARGE`. Both inputs must use the same bit-position mapping. If they were
+ * constructed by grouping `bitmap_bit_position` values by `bitmap_bucket_number`, they must
+ * represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ * function operates on two bitmaps from the same row; use `bitmap_*_agg` to combine bitmaps
+ * across rows. The representation is not a RoaringBitmap serialization.
+ *
+ * @param left
+ * A column that evaluates to a binary bitmap.
+ * @param right
+ * A column that evaluates to a binary bitmap.
+ * @group misc_funcs
+ * @since 4.4.0
+ * @return
+ * Returns a column that evaluates to a binary bitmap.
+ */
+ def bitmap_andnot(left: Column, right: Column): Column =
+ Column.fn("bitmap_andnot", left, right)
+
+ /**
+ * Returns a bitmap that is the bitwise XOR of two input bitmaps. The result is always a
+ * 4096-byte Spark Binary bitmap. If either input is NULL, the result is NULL. Missing bytes in
+ * shorter inputs are treated as zero, and inputs longer than 4096 bytes raise
+ * `BITMAP_INPUT_TOO_LARGE`. Both inputs must use the same bit-position mapping. If they were
+ * constructed by grouping `bitmap_bit_position` values by `bitmap_bucket_number`, they must
+ * represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ * function operates on two bitmaps from the same row; use `bitmap_*_agg` to combine bitmaps
+ * across rows. The representation is not a RoaringBitmap serialization.
+ *
+ * @param left
+ * A column that evaluates to a binary bitmap.
+ * @param right
+ * A column that evaluates to a binary bitmap.
+ * @group misc_funcs
+ * @since 4.4.0
+ * @return
+ * Returns a column that evaluates to a binary bitmap.
+ */
+ def bitmap_xor(left: Column, right: Column): Column = Column.fn("bitmap_xor", left, right)
+
/**
* 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().
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/BitmapExpressionUtils.java b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/BitmapExpressionUtils.java
index d03db7d952a5..714a3c28b476 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/BitmapExpressionUtils.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/BitmapExpressionUtils.java
@@ -78,4 +78,47 @@ public class BitmapExpressionUtils {
bitmap1[i] = 0;
}
}
+
+ /** Performs bitwise AND on both bitmaps and returns a new fixed-size bitmap. */
+ public static byte[] bitmapAnd(byte[] bitmap1, byte[] bitmap2) {
+ byte[] result = new byte[NUM_BYTES];
+ int numBytes = java.lang.Math.min(
+ NUM_BYTES, java.lang.Math.min(bitmap1.length, bitmap2.length));
+ for (int i = 0; i < numBytes; ++i) {
+ result[i] = (byte) ((bitmap1[i] & 0x0FF) & (bitmap2[i] & 0x0FF));
+ }
+ return result;
+ }
+
+ /** Performs bitwise OR on both bitmaps and returns a new fixed-size bitmap. */
+ public static byte[] bitmapOr(byte[] bitmap1, byte[] bitmap2) {
+ byte[] longer = bitmap1.length >= bitmap2.length ? bitmap1 : bitmap2;
+ byte[] shorter = bitmap1.length >= bitmap2.length ? bitmap2 : bitmap1;
+ byte[] result = java.util.Arrays.copyOf(longer, NUM_BYTES);
+ bitmapMerge(result, shorter);
+ return result;
+ }
+
+ /** Performs bitwise AND NOT on both bitmaps and returns a new fixed-size bitmap. */
+ public static byte[] bitmapAndNot(byte[] bitmap1, byte[] bitmap2) {
+ byte[] result = java.util.Arrays.copyOf(bitmap1, NUM_BYTES);
+ int numBytes = java.lang.Math.min(
+ NUM_BYTES, java.lang.Math.min(bitmap1.length, bitmap2.length));
+ for (int i = 0; i < numBytes; ++i) {
+ result[i] = (byte) ((result[i] & 0x0FF) & ~(bitmap2[i] & 0x0FF));
+ }
+ return result;
+ }
+
+ /** Performs bitwise XOR on both bitmaps and returns a new fixed-size bitmap. */
+ public static byte[] bitmapXor(byte[] bitmap1, byte[] bitmap2) {
+ byte[] longer = bitmap1.length >= bitmap2.length ? bitmap1 : bitmap2;
+ byte[] shorter = bitmap1.length >= bitmap2.length ? bitmap2 : bitmap1;
+ byte[] result = java.util.Arrays.copyOf(longer, NUM_BYTES);
+ int numBytes = java.lang.Math.min(NUM_BYTES, shorter.length);
+ for (int i = 0; i < numBytes; ++i) {
+ result[i] = (byte) ((result[i] & 0x0FF) ^ (shorter[i] & 0x0FF));
+ }
+ return result;
+ }
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
index a23e70188622..dd70ea9d177a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala
@@ -974,6 +974,10 @@ object FunctionRegistry {
expression[BitmapBitPosition]("bitmap_bit_position"),
expression[BitmapConstructAgg]("bitmap_construct_agg"),
expression[BitmapCount]("bitmap_count"),
+ expression[BitmapAnd]("bitmap_and"),
+ expression[BitmapOr]("bitmap_or"),
+ expression[BitmapAndNot]("bitmap_andnot"),
+ expression[BitmapXor]("bitmap_xor"),
expression[BitmapOrAgg]("bitmap_or_agg"),
expression[BitmapAndAgg]("bitmap_and_agg"),
expression[BitmapXorAgg]("bitmap_xor_agg"),
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/bitmapExpressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/bitmapExpressions.scala
index 02576f051979..806d8bf51010 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/bitmapExpressions.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/bitmapExpressions.scala
@@ -21,6 +21,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{DataTypeMismatch, TypeCheckSuccess}
import org.apache.spark.sql.catalyst.expressions.aggregate.ImperativeAggregate
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}
import org.apache.spark.sql.catalyst.expressions.objects.StaticInvoke
import org.apache.spark.sql.catalyst.trees.UnaryLike
import org.apache.spark.sql.catalyst.types.DataTypeUtils
@@ -158,6 +159,222 @@ case class BitmapCount(child: Expression)
copy(child = newChild)
}
+/** Base class for scalar bitmap binary operations. */
+abstract class BitmapBinaryExpression extends BinaryExpression with ExpectsInputTypes {
+
+ override def inputTypes: Seq[AbstractDataType] = Seq(BinaryType, BinaryType)
+
+ override def dataType: DataType = BinaryType
+
+ override def nullIntolerant: Boolean = true
+
+ protected def applyOperation(bitmap1: Array[Byte], bitmap2: Array[Byte]): Array[Byte]
+
+ protected def genCodeOperation(
+ bitmapUtils: String, bitmap1: String, bitmap2: String): String
+
+ private def checkBitmapLength(bitmap: Array[Byte]): Unit = {
+ if (bitmap.length > BitmapExpressionUtils.NUM_BYTES) {
+ throw QueryExecutionErrors.bitmapInputTooLargeError(
+ bitmap.length, BitmapExpressionUtils.NUM_BYTES)
+ }
+ }
+
+ override protected def nullSafeEval(input1: Any, input2: Any): Any = {
+ val bitmap1 = input1.asInstanceOf[Array[Byte]]
+ val bitmap2 = input2.asInstanceOf[Array[Byte]]
+ checkBitmapLength(bitmap1)
+ checkBitmapLength(bitmap2)
+ applyOperation(bitmap1, bitmap2)
+ }
+
+ override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+ val bitmapUtils = classOf[BitmapExpressionUtils].getName
+ val errors = QueryExecutionErrors.getClass.getName.stripSuffix("$")
+ nullSafeCodeGen(ctx, ev, (bitmap1, bitmap2) => {
+ s"""
+ |if ($bitmap1.length > ${BitmapExpressionUtils.NUM_BYTES}) {
+ | throw $errors.bitmapInputTooLargeError(
+ | $bitmap1.length, ${BitmapExpressionUtils.NUM_BYTES});
+ |}
+ |if ($bitmap2.length > ${BitmapExpressionUtils.NUM_BYTES}) {
+ | throw $errors.bitmapInputTooLargeError(
+ | $bitmap2.length, ${BitmapExpressionUtils.NUM_BYTES});
+ |}
+ |${ev.value} = ${genCodeOperation(bitmapUtils, bitmap1, bitmap2)};
+ |""".stripMargin
+ })
+ }
+}
+
+@ExpressionDescription(
+ usage = "_FUNC_(left, right) - Returns a bitmap that is the bitwise AND of two input bitmaps.",
+ arguments = """
+ Arguments:
+ * left - A binary bitmap.
+ * right - A binary bitmap.
+ """,
+ examples = """
+ Examples:
+ > SELECT substring(hex(_FUNC_(X 'F0', X '70')), 0, 2);
+ 70
+ """,
+ note = """
+ Inputs use Spark's Binary bitmap representation, not a RoaringBitmap serialization. Each
+ input may contain 0 to 4096 bytes; missing bytes are treated as zero. The result is always a
+ 4096-byte Binary value. NULL input returns NULL, and inputs longer than 4096 bytes raise
+ BITMAP_INPUT_TOO_LARGE. Both inputs must use the same bit-position mapping. If they were
+ constructed by grouping bitmap_bit_position values by bitmap_bucket_number, they must
+ represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ function combines two bitmaps from the same row; use bitmap_*_agg to combine bitmaps across
+ rows.
+ """,
+ since = "4.4.0",
+ group = "misc_funcs"
+)
+case class BitmapAnd(left: Expression, right: Expression) extends BitmapBinaryExpression {
+
+ override def prettyName: String = "bitmap_and"
+
+ override protected def applyOperation(
+ bitmap1: Array[Byte], bitmap2: Array[Byte]): Array[Byte] =
+ BitmapExpressionUtils.bitmapAnd(bitmap1, bitmap2)
+
+ override protected def genCodeOperation(
+ bitmapUtils: String, bitmap1: String, bitmap2: String): String =
+ s"$bitmapUtils.bitmapAnd($bitmap1, $bitmap2)"
+
+ override protected def withNewChildrenInternal(
+ newLeft: Expression, newRight: Expression): BitmapAnd =
+ copy(left = newLeft, right = newRight)
+}
+
+@ExpressionDescription(
+ usage = "_FUNC_(left, right) - Returns a bitmap that is the bitwise OR of two input bitmaps.",
+ arguments = """
+ Arguments:
+ * left - A binary bitmap.
+ * right - A binary bitmap.
+ """,
+ examples = """
+ Examples:
+ > SELECT substring(hex(_FUNC_(X '10', X '20')), 0, 2);
+ 30
+ """,
+ note = """
+ Inputs use Spark's Binary bitmap representation, not a RoaringBitmap serialization. Each
+ input may contain 0 to 4096 bytes; missing bytes are treated as zero. The result is always a
+ 4096-byte Binary value. NULL input returns NULL, and inputs longer than 4096 bytes raise
+ BITMAP_INPUT_TOO_LARGE. Both inputs must use the same bit-position mapping. If they were
+ constructed by grouping bitmap_bit_position values by bitmap_bucket_number, they must
+ represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ function combines two bitmaps from the same row; use bitmap_*_agg to combine bitmaps across
+ rows.
+ """,
+ since = "4.4.0",
+ group = "misc_funcs"
+)
+case class BitmapOr(left: Expression, right: Expression) extends BitmapBinaryExpression {
+
+ override def prettyName: String = "bitmap_or"
+
+ override protected def applyOperation(
+ bitmap1: Array[Byte], bitmap2: Array[Byte]): Array[Byte] =
+ BitmapExpressionUtils.bitmapOr(bitmap1, bitmap2)
+
+ override protected def genCodeOperation(
+ bitmapUtils: String, bitmap1: String, bitmap2: String): String =
+ s"$bitmapUtils.bitmapOr($bitmap1, $bitmap2)"
+
+ override protected def withNewChildrenInternal(
+ newLeft: Expression, newRight: Expression): BitmapOr =
+ copy(left = newLeft, right = newRight)
+}
+
+@ExpressionDescription(
+ usage = "_FUNC_(left, right) - Returns a bitmap that is the bitwise AND NOT of two bitmaps.",
+ arguments = """
+ Arguments:
+ * left - A binary bitmap.
+ * right - A binary bitmap.
+ """,
+ examples = """
+ Examples:
+ > SELECT substring(hex(_FUNC_(X 'F0', X '70')), 0, 2);
+ 80
+ """,
+ note = """
+ Inputs use Spark's Binary bitmap representation, not a RoaringBitmap serialization. Each
+ input may contain 0 to 4096 bytes; missing bytes are treated as zero. The result is always a
+ 4096-byte Binary value. NULL input returns NULL, and inputs longer than 4096 bytes raise
+ BITMAP_INPUT_TOO_LARGE. Both inputs must use the same bit-position mapping. If they were
+ constructed by grouping bitmap_bit_position values by bitmap_bucket_number, they must
+ represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ function combines two bitmaps from the same row; use bitmap_*_agg to combine bitmaps across
+ rows.
+ """,
+ since = "4.4.0",
+ group = "misc_funcs"
+)
+case class BitmapAndNot(left: Expression, right: Expression) extends BitmapBinaryExpression {
+
+ override def prettyName: String = "bitmap_andnot"
+
+ override protected def applyOperation(
+ bitmap1: Array[Byte], bitmap2: Array[Byte]): Array[Byte] =
+ BitmapExpressionUtils.bitmapAndNot(bitmap1, bitmap2)
+
+ override protected def genCodeOperation(
+ bitmapUtils: String, bitmap1: String, bitmap2: String): String =
+ s"$bitmapUtils.bitmapAndNot($bitmap1, $bitmap2)"
+
+ override protected def withNewChildrenInternal(
+ newLeft: Expression, newRight: Expression): BitmapAndNot =
+ copy(left = newLeft, right = newRight)
+}
+
+@ExpressionDescription(
+ usage = "_FUNC_(left, right) - Returns a bitmap that is the bitwise XOR of two input bitmaps.",
+ arguments = """
+ Arguments:
+ * left - A binary bitmap.
+ * right - A binary bitmap.
+ """,
+ examples = """
+ Examples:
+ > SELECT substring(hex(_FUNC_(X 'F0', X '70')), 0, 2);
+ 80
+ """,
+ note = """
+ Inputs use Spark's Binary bitmap representation, not a RoaringBitmap serialization. Each
+ input may contain 0 to 4096 bytes; missing bytes are treated as zero. The result is always a
+ 4096-byte Binary value. NULL input returns NULL, and inputs longer than 4096 bytes raise
+ BITMAP_INPUT_TOO_LARGE. Both inputs must use the same bit-position mapping. If they were
+ constructed by grouping bitmap_bit_position values by bitmap_bucket_number, they must
+ represent the same bucket because the bitmap bytes do not retain bucket metadata. This scalar
+ function combines two bitmaps from the same row; use bitmap_*_agg to combine bitmaps across
+ rows.
+ """,
+ since = "4.4.0",
+ group = "misc_funcs"
+)
+case class BitmapXor(left: Expression, right: Expression) extends BitmapBinaryExpression {
+
+ override def prettyName: String = "bitmap_xor"
+
+ override protected def applyOperation(
+ bitmap1: Array[Byte], bitmap2: Array[Byte]): Array[Byte] =
+ BitmapExpressionUtils.bitmapXor(bitmap1, bitmap2)
+
+ override protected def genCodeOperation(
+ bitmapUtils: String, bitmap1: String, bitmap2: String): String =
+ s"$bitmapUtils.bitmapXor($bitmap1, $bitmap2)"
+
+ override protected def withNewChildrenInternal(
+ newLeft: Expression, newRight: Expression): BitmapXor =
+ copy(left = newLeft, right = newRight)
+}
+
@ExpressionDescription(
usage = """
_FUNC_(child) - Returns a bitmap with the positions of the bits set from all the values from
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index 87205cd7bd15..6816bb681f32 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -269,6 +269,15 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
summary = "")
}
+ def bitmapInputTooLargeError(inputNumBytes: Int, maxNumBytes: Int): SparkRuntimeException = {
+ new SparkRuntimeException(
+ errorClass = "BITMAP_INPUT_TOO_LARGE",
+ messageParameters = Map(
+ "inputNumBytes" -> inputNumBytes.toString,
+ "maxNumBytes" -> maxNumBytes.toString),
+ cause = null)
+ }
+
def invalidFractionOfSecondError(secAndMicros: Double): DateTimeException = {
new SparkDateTimeException(
errorClass = "INVALID_FRACTION_OF_SECOND",
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/BitmapExpressionUtilsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/BitmapExpressionUtilsSuite.scala
index c33a14aa7b42..3cd4eea2e0ef 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/BitmapExpressionUtilsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/BitmapExpressionUtilsSuite.scala
@@ -93,6 +93,97 @@ class BitmapExpressionUtilsSuite extends SparkFunSuite {
assert(BitmapExpressionUtils.bitmapCount(bitmap) == 15L)
}
+ test("scalar bitmap binary operations") {
+ val bitmap1 = Array(0xf0.toByte, 0x0f.toByte)
+ val bitmap2 = Array(0x70.toByte)
+ val originalBitmap1 = bitmap1.clone()
+ val originalBitmap2 = bitmap2.clone()
+
+ val results = Seq(
+ BitmapExpressionUtils.bitmapAnd(bitmap1, bitmap2),
+ BitmapExpressionUtils.bitmapOr(bitmap1, bitmap2),
+ BitmapExpressionUtils.bitmapAndNot(bitmap1, bitmap2),
+ BitmapExpressionUtils.bitmapXor(bitmap1, bitmap2))
+ val expected = Seq(
+ Seq(0x70.toByte, 0x00.toByte),
+ Seq(0xf0.toByte, 0x0f.toByte),
+ Seq(0x80.toByte, 0x0f.toByte),
+ Seq(0x80.toByte, 0x0f.toByte))
+
+ results.zip(expected).foreach { case (result, expectedBytes) =>
+ assert(result.length == BitmapExpressionUtils.NUM_BYTES)
+ assert(result.take(expectedBytes.length).toSeq == expectedBytes)
+ assert(result.drop(expectedBytes.length).forall(_ == 0))
+ }
+ assert(bitmap1.sameElements(originalBitmap1))
+ assert(bitmap2.sameElements(originalBitmap2))
+ }
+
+ test("scalar bitmap binary operations with a longer right input") {
+ val bitmap1 = Array(0xf0.toByte)
+ val bitmap2 = Array(0x70.toByte, 0x0f.toByte)
+ val originalBitmap1 = bitmap1.clone()
+ val originalBitmap2 = bitmap2.clone()
+
+ val results = Seq(
+ BitmapExpressionUtils.bitmapAnd(bitmap1, bitmap2),
+ BitmapExpressionUtils.bitmapOr(bitmap1, bitmap2),
+ BitmapExpressionUtils.bitmapAndNot(bitmap1, bitmap2),
+ BitmapExpressionUtils.bitmapXor(bitmap1, bitmap2))
+ val expected = Seq(
+ Seq(0x70.toByte, 0x00.toByte),
+ Seq(0xf0.toByte, 0x0f.toByte),
+ Seq(0x80.toByte, 0x00.toByte),
+ Seq(0x80.toByte, 0x0f.toByte))
+
+ results.zip(expected).foreach { case (result, expectedBytes) =>
+ assert(result.length == BitmapExpressionUtils.NUM_BYTES)
+ assert(result.take(expectedBytes.length).toSeq == expectedBytes)
+ assert(result.drop(expectedBytes.length).forall(_ == 0))
+ }
+ assert(bitmap1.sameElements(originalBitmap1))
+ assert(bitmap2.sameElements(originalBitmap2))
+ }
+
+ test("scalar bitmap binary operations with boundary lengths") {
+ val expectedBytes = Seq(0x80.toByte, 0xff.toByte, 0x00.toByte, 0x7f.toByte)
+ Seq(0, 1, BitmapExpressionUtils.NUM_BYTES - 1, BitmapExpressionUtils.NUM_BYTES).foreach {
+ size =>
+ val bitmap1 = Array.fill[Byte](size)(0x80.toByte)
+ val bitmap2 = Array.fill[Byte](size)(0xff.toByte)
+ val results = Seq(
+ BitmapExpressionUtils.bitmapAnd(bitmap1, bitmap2),
+ BitmapExpressionUtils.bitmapOr(bitmap1, bitmap2),
+ BitmapExpressionUtils.bitmapAndNot(bitmap1, bitmap2),
+ BitmapExpressionUtils.bitmapXor(bitmap1, bitmap2))
+
+ results.zip(expectedBytes).foreach { case (result, expectedByte) =>
+ assert(result.length == BitmapExpressionUtils.NUM_BYTES)
+ assert(result.take(size).forall(_ == expectedByte))
+ assert(result.drop(size).forall(_ == 0))
+ }
+ }
+ }
+
+ test("scalar bitmap binary operations satisfy set operation properties") {
+ val bitmap = Array.fill[Byte](BitmapExpressionUtils.NUM_BYTES)(0x5a.toByte)
+ val other = Array.fill[Byte](BitmapExpressionUtils.NUM_BYTES)(0xa5.toByte)
+ val empty = Array.fill[Byte](BitmapExpressionUtils.NUM_BYTES)(0)
+
+ assert(BitmapExpressionUtils.bitmapAnd(bitmap, bitmap).sameElements(bitmap))
+ assert(BitmapExpressionUtils.bitmapOr(bitmap, bitmap).sameElements(bitmap))
+ assert(BitmapExpressionUtils.bitmapXor(bitmap, bitmap).sameElements(empty))
+ assert(BitmapExpressionUtils.bitmapAndNot(bitmap, bitmap).sameElements(empty))
+ assert(BitmapExpressionUtils.bitmapAndNot(bitmap, empty).sameElements(bitmap))
+ assert(BitmapExpressionUtils.bitmapAndNot(empty, bitmap).sameElements(empty))
+ assert(BitmapExpressionUtils.bitmapOr(bitmap, empty).sameElements(bitmap))
+ assert(BitmapExpressionUtils.bitmapXor(bitmap, empty).sameElements(bitmap))
+ assert(BitmapExpressionUtils.bitmapOr(bitmap, other).sameElements(
+ BitmapExpressionUtils.bitmapOr(other, bitmap)))
+ assert(BitmapExpressionUtils.bitmapXor(bitmap, other).sameElements(
+ BitmapExpressionUtils.bitmapXor(other, bitmap)))
+ }
+
test("bitmap_xor_merge equal length") {
val bitmap1 = Array[Byte](0x10, 0x30, 0x40)
val bitmap2 = Array[Byte](0x10, 0x20, 0x40)
diff --git a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala
index 0c6ff268aaf2..6346cf4b5cd8 100644
--- a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala
+++ b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/PlanGenerationTestSuite.scala
@@ -2266,6 +2266,22 @@ class PlanGenerationTestSuite extends ConnectFunSuite with Logging {
binary.select(fn.bitmap_or_agg(fn.col("bytes")))
}
+ test("function bitmap_and") {
+ binary.select(fn.bitmap_and(fn.col("bytes"), fn.col("bytes")))
+ }
+
+ test("function bitmap_or") {
+ binary.select(fn.bitmap_or(fn.col("bytes"), fn.col("bytes")))
+ }
+
+ test("function bitmap_andnot") {
+ binary.select(fn.bitmap_andnot(fn.col("bytes"), fn.col("bytes")))
+ }
+
+ test("function bitmap_xor") {
+ binary.select(fn.bitmap_xor(fn.col("bytes"), fn.col("bytes")))
+ }
+
test("function bitmap_xor_agg") {
binary.select(fn.bitmap_xor_agg(fn.col("bytes")))
}
diff --git a/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_and.explain b/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_and.explain
new file mode 100644
index 000000000000..d8892a7cfa82
--- /dev/null
+++ b/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_and.explain
@@ -0,0 +1,2 @@
+Project [bitmap_and(bytes#0, bytes#0) AS bitmap_and(bytes, bytes)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]
diff --git a/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_andnot.explain b/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_andnot.explain
new file mode 100644
index 000000000000..7a081fa44975
--- /dev/null
+++ b/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_andnot.explain
@@ -0,0 +1,2 @@
+Project [bitmap_andnot(bytes#0, bytes#0) AS bitmap_andnot(bytes, bytes)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]
diff --git a/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_or.explain b/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_or.explain
new file mode 100644
index 000000000000..f58db6eb6908
--- /dev/null
+++ b/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_or.explain
@@ -0,0 +1,2 @@
+Project [bitmap_or(bytes#0, bytes#0) AS bitmap_or(bytes, bytes)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]
diff --git a/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_xor.explain b/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_xor.explain
new file mode 100644
index 000000000000..a6f81bbc9a2b
--- /dev/null
+++ b/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_xor.explain
@@ -0,0 +1,2 @@
+Project [bitmap_xor(bytes#0, bytes#0) AS bitmap_xor(bytes, bytes)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_and.json b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_and.json
new file mode 100644
index 000000000000..1abf52ef603e
--- /dev/null
+++ b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_and.json
@@ -0,0 +1,81 @@
+{
+ "common": {
+ "planId": "1"
+ },
+ "project": {
+ "input": {
+ "common": {
+ "planId": "0"
+ },
+ "localRelation": {
+ "schema": "struct\u003cid:bigint,bytes:binary\u003e"
+ }
+ },
+ "expressions": [{
+ "unresolvedFunction": {
+ "functionName": "bitmap_and",
+ "arguments": [{
+ "unresolvedAttribute": {
+ "unparsedIdentifier": "bytes"
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "col",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }, {
+ "unresolvedAttribute": {
+ "unparsedIdentifier": "bytes"
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "col",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }],
+ "isInternal": false
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "bitmap_and",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }]
+ }
+}
\ No newline at end of file
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_and.proto.bin b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_and.proto.bin
new file mode 100644
index 000000000000..4ed1b96834a2
Binary files /dev/null and b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_and.proto.bin differ
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_andnot.json b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_andnot.json
new file mode 100644
index 000000000000..a3363b09d430
--- /dev/null
+++ b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_andnot.json
@@ -0,0 +1,81 @@
+{
+ "common": {
+ "planId": "1"
+ },
+ "project": {
+ "input": {
+ "common": {
+ "planId": "0"
+ },
+ "localRelation": {
+ "schema": "struct\u003cid:bigint,bytes:binary\u003e"
+ }
+ },
+ "expressions": [{
+ "unresolvedFunction": {
+ "functionName": "bitmap_andnot",
+ "arguments": [{
+ "unresolvedAttribute": {
+ "unparsedIdentifier": "bytes"
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "col",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }, {
+ "unresolvedAttribute": {
+ "unparsedIdentifier": "bytes"
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "col",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }],
+ "isInternal": false
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "bitmap_andnot",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }]
+ }
+}
\ No newline at end of file
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_andnot.proto.bin b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_andnot.proto.bin
new file mode 100644
index 000000000000..b0cdbd7ea9bd
Binary files /dev/null and b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_andnot.proto.bin differ
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_or.json b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_or.json
new file mode 100644
index 000000000000..7c0acff13b79
--- /dev/null
+++ b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_or.json
@@ -0,0 +1,81 @@
+{
+ "common": {
+ "planId": "1"
+ },
+ "project": {
+ "input": {
+ "common": {
+ "planId": "0"
+ },
+ "localRelation": {
+ "schema": "struct\u003cid:bigint,bytes:binary\u003e"
+ }
+ },
+ "expressions": [{
+ "unresolvedFunction": {
+ "functionName": "bitmap_or",
+ "arguments": [{
+ "unresolvedAttribute": {
+ "unparsedIdentifier": "bytes"
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "col",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }, {
+ "unresolvedAttribute": {
+ "unparsedIdentifier": "bytes"
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "col",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }],
+ "isInternal": false
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "bitmap_or",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }]
+ }
+}
\ No newline at end of file
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_or.proto.bin b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_or.proto.bin
new file mode 100644
index 000000000000..b8dbed026128
Binary files /dev/null and b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_or.proto.bin differ
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_xor.json b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_xor.json
new file mode 100644
index 000000000000..59c78fa16801
--- /dev/null
+++ b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_xor.json
@@ -0,0 +1,81 @@
+{
+ "common": {
+ "planId": "1"
+ },
+ "project": {
+ "input": {
+ "common": {
+ "planId": "0"
+ },
+ "localRelation": {
+ "schema": "struct\u003cid:bigint,bytes:binary\u003e"
+ }
+ },
+ "expressions": [{
+ "unresolvedFunction": {
+ "functionName": "bitmap_xor",
+ "arguments": [{
+ "unresolvedAttribute": {
+ "unparsedIdentifier": "bytes"
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "col",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }, {
+ "unresolvedAttribute": {
+ "unparsedIdentifier": "bytes"
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "col",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }],
+ "isInternal": false
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "bitmap_xor",
+ "fileName": "functions.scala"
+ }, {
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.PlanGenerationTestSuite",
+ "methodName": "~~trimmed~anonfun~~",
+ "fileName": "PlanGenerationTestSuite.scala"
+ }]
+ }
+ }
+ }
+ }]
+ }
+}
\ No newline at end of file
diff --git a/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_xor.proto.bin b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_xor.proto.bin
new file mode 100644
index 000000000000..dfc994306a53
Binary files /dev/null and b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_xor.proto.bin differ
diff --git a/sql/core/src/test/resources/sql-functions/sql-expression-schema.md b/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
index e0d9862bc22a..6f11ca0435ae 100644
--- a/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
+++ b/sql/core/src/test/resources/sql-functions/sql-expression-schema.md
@@ -49,12 +49,16 @@
| org.apache.spark.sql.catalyst.expressions.Between | between | SELECT 0.5 between 0.1 AND 1.0 | struct<between(0.5, 0.1, 1.0):boolean> |
| org.apache.spark.sql.catalyst.expressions.Bin | bin | SELECT bin(13) | struct<bin(13):string> |
| org.apache.spark.sql.catalyst.expressions.BitLength | bit_length | SELECT bit_length('Spark SQL') | struct<bit_length(Spark SQL):int> |
+| org.apache.spark.sql.catalyst.expressions.BitmapAnd | bitmap_and | SELECT substring(hex(bitmap_and(X 'F0', X '70')), 0, 2) | struct<substring(hex(bitmap_and(X'F0', X'70')), 0, 2):string> |
| org.apache.spark.sql.catalyst.expressions.BitmapAndAgg | bitmap_and_agg | SELECT substring(hex(bitmap_and_agg(col)), 0, 6) FROM VALUES (X 'F0'), (X '70'), (X '30') AS tab(col) | struct<substring(hex(bitmap_and_agg(col)), 0, 6):string> |
+| org.apache.spark.sql.catalyst.expressions.BitmapAndNot | bitmap_andnot | SELECT substring(hex(bitmap_andnot(X 'F0', X '70')), 0, 2) | struct<substring(hex(bitmap_andnot(X'F0', X'70')), 0, 2):string> |
| org.apache.spark.sql.catalyst.expressions.BitmapBitPosition | bitmap_bit_position | SELECT bitmap_bit_position(1) | struct<bitmap_bit_position(1):bigint> |
| org.apache.spark.sql.catalyst.expressions.BitmapBucketNumber | bitmap_bucket_number | SELECT bitmap_bucket_number(123) | struct<bitmap_bucket_number(123):bigint> |
| org.apache.spark.sql.catalyst.expressions.BitmapConstructAgg | bitmap_construct_agg | SELECT substring(hex(bitmap_construct_agg(bitmap_bit_position(col))), 0, 6) FROM VALUES (1), (2), (3) AS tab(col) | struct<substring(hex(bitmap_construct_agg(bitmap_bit_position(col))), 0, 6):string> |
| org.apache.spark.sql.catalyst.expressions.BitmapCount | bitmap_count | SELECT bitmap_count(X '1010') | struct<bitmap_count(X'1010'):bigint> |
+| org.apache.spark.sql.catalyst.expressions.BitmapOr | bitmap_or | SELECT substring(hex(bitmap_or(X '10', X '20')), 0, 2) | struct<substring(hex(bitmap_or(X'10', X'20')), 0, 2):string> |
| org.apache.spark.sql.catalyst.expressions.BitmapOrAgg | bitmap_or_agg | SELECT substring(hex(bitmap_or_agg(col)), 0, 6) FROM VALUES (X '10'), (X '20'), (X '40') AS tab(col) | struct<substring(hex(bitmap_or_agg(col)), 0, 6):string> |
+| org.apache.spark.sql.catalyst.expressions.BitmapXor | bitmap_xor | SELECT substring(hex(bitmap_xor(X 'F0', X '70')), 0, 2) | struct<substring(hex(bitmap_xor(X'F0', X'70')), 0, 2):string> |
| org.apache.spark.sql.catalyst.expressions.BitmapXorAgg | bitmap_xor_agg | SELECT substring(hex(bitmap_xor_agg(col)), 0, 6) FROM VALUES (X'10'), (X'30'), (X'40') AS tab(col) | struct<substring(hex(bitmap_xor_agg(col)), 0, 6):string> |
| org.apache.spark.sql.catalyst.expressions.BitwiseAnd | & | SELECT 3 & 5 | struct<(3 & 5):int> |
| org.apache.spark.sql.catalyst.expressions.BitwiseCount | bit_count | SELECT bit_count(0) | struct<bit_count(0):int> |
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/BitmapExpressionsQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/BitmapExpressionsQuerySuite.scala
index 8201e2c2f872..7d079fb894e7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/BitmapExpressionsQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/BitmapExpressionsQuerySuite.scala
@@ -17,20 +17,27 @@
package org.apache.spark.sql
+import org.apache.spark.SparkRuntimeException
import org.apache.spark.sql.functions.{
+ bitmap_and,
bitmap_and_agg,
+ bitmap_andnot,
bitmap_bit_position,
bitmap_bucket_number,
bitmap_construct_agg,
bitmap_count,
+ bitmap_or,
bitmap_or_agg,
+ bitmap_xor,
bitmap_xor_agg,
col,
expr,
hex,
lit,
substring,
+ sum,
to_binary}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
class BitmapExpressionsQuerySuite extends SharedSparkSession {
@@ -310,6 +317,167 @@ class BitmapExpressionsQuerySuite extends SharedSparkSession {
}
}
+ test("scalar bitmap binary operations") {
+ def checkResults(): Unit = {
+ val df = Seq(("F00F", "70")).toDF("left", "right")
+ checkAnswer(
+ df.selectExpr(
+ "substring(hex(bitmap_and(to_binary(left, 'hex'), to_binary(right, 'hex'))), 0, 4)",
+ "substring(hex(bitmap_or(to_binary(left, 'hex'), to_binary(right, 'hex'))), 0, 4)",
+ "substring(hex(bitmap_andnot(to_binary(left, 'hex'), to_binary(right, 'hex'))), 0, 4)",
+ "substring(hex(bitmap_xor(to_binary(left, 'hex'), to_binary(right, 'hex'))), 0, 4)"),
+ Seq(Row("7000", "F00F", "800F", "800F")))
+
+ val leftBitmap = to_binary(col("left"), lit("hex"))
+ val rightBitmap = to_binary(col("right"), lit("hex"))
+ checkAnswer(
+ df.select(
+ substring(hex(bitmap_and(leftBitmap, rightBitmap)), 0, 4),
+ substring(hex(bitmap_or(leftBitmap, rightBitmap)), 0, 4),
+ substring(hex(bitmap_andnot(leftBitmap, rightBitmap)), 0, 4),
+ substring(hex(bitmap_xor(leftBitmap, rightBitmap)), 0, 4)),
+ Seq(Row("7000", "F00F", "800F", "800F")))
+
+ val shortBitmap = Seq(("", "FF")).toDF("left", "right")
+ checkAnswer(
+ shortBitmap.selectExpr(
+ "length(bitmap_and(to_binary(left, 'hex'), to_binary(right, 'hex')))",
+ "length(bitmap_or(to_binary(left, 'hex'), to_binary(right, 'hex')))",
+ "length(bitmap_andnot(to_binary(left, 'hex'), to_binary(right, 'hex')))",
+ "length(bitmap_xor(to_binary(left, 'hex'), to_binary(right, 'hex')))"),
+ Seq(Row(4096, 4096, 4096, 4096)))
+
+ checkAnswer(
+ spark.sql("""
+ |SELECT bitmap_count(bitmap_and(X'', X'FF')),
+ | bitmap_count(bitmap_or(X'', X'FF')),
+ | bitmap_count(bitmap_andnot(X'', X'FF')),
+ | bitmap_count(bitmap_xor(X'', X'FF'))
+ |""".stripMargin),
+ Seq(Row(0, 8, 0, 8)))
+
+ val boundaryBitmap =
+ Seq((Array.fill[Byte](4096)(0), Array.fill[Byte](4096)(0))).toDF("left", "right")
+ checkAnswer(
+ boundaryBitmap.selectExpr(
+ "length(bitmap_and(left, right))",
+ "length(bitmap_or(left, right))",
+ "length(bitmap_andnot(left, right))",
+ "length(bitmap_xor(left, right))"),
+ Seq(Row(4096, 4096, 4096, 4096)))
+
+ checkAnswer(
+ spark.sql("SELECT bitmap_count(bitmap_or(bitmap_and(X 'F0', X '70'), X '0F'))"),
+ Seq(Row(7)))
+ }
+
+ Seq("CODEGEN_ONLY" -> "true", "NO_CODEGEN" -> "false").foreach {
+ case (codegenMode, wholeStageEnabled) =>
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStageEnabled,
+ SQLConf.CODEGEN_FACTORY_MODE.key -> codegenMode) {
+ checkResults()
+ }
+ }
+ }
+
+ test("scalar bitmap binary operations with aggregate bitmaps") {
+ val constructed = Seq((1, 2), (2, 3), (3, 4)).toDF("left", "right")
+ val leftBitmap = bitmap_construct_agg(bitmap_bit_position(col("left")))
+ val rightBitmap = bitmap_construct_agg(bitmap_bit_position(col("right")))
+ checkAnswer(
+ constructed.agg(bitmap_count(bitmap_and(leftBitmap, rightBitmap))),
+ Seq(Row(2)))
+
+ val precomputed = Seq("F0", "70").toDF("bitmap")
+ val orBitmap = bitmap_or_agg(to_binary(col("bitmap"), lit("hex")))
+ val andBitmap = bitmap_and_agg(to_binary(col("bitmap"), lit("hex")))
+ checkAnswer(
+ precomputed.agg(
+ bitmap_count(bitmap_and(orBitmap, andBitmap)),
+ bitmap_count(bitmap_or(orBitmap, andBitmap)),
+ bitmap_count(bitmap_andnot(orBitmap, andBitmap)),
+ bitmap_count(bitmap_xor(orBitmap, andBitmap))),
+ Seq(Row(3, 4, 1, 1)))
+ }
+
+ test("scalar bitmap binary operations in grouped query") {
+ val df = Seq((1, "F0", "70"), (1, "10", "20"), (2, "FF", "0F"))
+ .toDF("group_id", "left", "right")
+ checkAnswer(
+ df.selectExpr(
+ "group_id",
+ "bitmap_count(bitmap_and(to_binary(left, 'hex'), to_binary(right, 'hex'))) AS count")
+ .groupBy("group_id")
+ .agg(sum("count"))
+ .orderBy("group_id"),
+ Seq(Row(1, 3), Row(2, 4)))
+ }
+
+ test("scalar bitmap binary operations with nulls") {
+ val df = Seq[(String, String)](("F0", null), (null, "70"), (null, null)).toDF("left", "right")
+ checkAnswer(
+ df.selectExpr(
+ "bitmap_and(to_binary(left, 'hex'), to_binary(right, 'hex'))",
+ "bitmap_or(to_binary(left, 'hex'), to_binary(right, 'hex'))",
+ "bitmap_andnot(to_binary(left, 'hex'), to_binary(right, 'hex'))",
+ "bitmap_xor(to_binary(left, 'hex'), to_binary(right, 'hex'))"),
+ Seq(
+ Row(null, null, null, null),
+ Row(null, null, null, null),
+ Row(null, null, null, null)))
+ }
+
+ test("scalar bitmap binary operations reject oversized inputs") {
+ val oversizedInputs = Seq(
+ "left" -> Seq((Array.fill[Byte](4097)(0), Array[Byte](0))).toDF("left", "right"),
+ "right" -> Seq((Array[Byte](0), Array.fill[Byte](4097)(0))).toDF("left", "right"))
+
+ Seq("CODEGEN_ONLY" -> "true", "NO_CODEGEN" -> "false").foreach {
+ case (codegenMode, wholeStageEnabled) =>
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> wholeStageEnabled,
+ SQLConf.CODEGEN_FACTORY_MODE.key -> codegenMode) {
+ oversizedInputs.foreach { case (inputSide, oversized) =>
+ Seq("bitmap_and", "bitmap_or", "bitmap_andnot", "bitmap_xor").foreach { functionName =>
+ withClue(s"$functionName with oversized $inputSide input in $codegenMode: ") {
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ oversized.selectExpr(s"$functionName(left, right)").collect()
+ },
+ condition = "BITMAP_INPUT_TOO_LARGE",
+ parameters = Map("inputNumBytes" -> "4097", "maxNumBytes" -> "4096"))
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("scalar bitmap binary operations called with non-binary types") {
+ val invalidInputs = Seq(
+ ("first", "left", Seq((12, Array[Byte](0))).toDF("left", "right")),
+ ("second", "right", Seq((Array[Byte](0), 13)).toDF("left", "right")))
+
+ invalidInputs.foreach { case (paramIndex, inputColumn, df) =>
+ Seq("bitmap_and", "bitmap_or", "bitmap_andnot", "bitmap_xor").foreach { functionName =>
+ val sqlExpr = s"$functionName(left, right)"
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.selectExpr(sqlExpr)
+ },
+ condition = "DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE",
+ parameters = Map(
+ "sqlExpr" -> s""""$sqlExpr"""",
+ "paramIndex" -> paramIndex,
+ "requiredType" -> "\"BINARY\"",
+ "inputSql" -> s""""$inputColumn"""",
+ "inputType" -> "\"INT\""),
+ context = ExpectedContext(fragment = sqlExpr, start = 0, stop = sqlExpr.length - 1))
+ }
+ }
+ }
+
test("bitmap_count called with non-binary type") {
val df = Seq(12).toDF("a")
checkError(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org