This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new cc75de63a456 [SPARK-58831][SQL] Add bitmap scalar set operation
functions
cc75de63a456 is described below
commit cc75de63a45694f8e70c60cd3387ddf8a83cf80b
Author: StormSpirit <[email protected]>
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 <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../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 59da38fdc8dd..df1a1f3a5b8f 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 1d2c6355e9e2..5acb3301be4d 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 3ac36551b0ed..9c79e1117332 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 a17b1ca7e9bb..63ab10b69422 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: [email protected]
For additional commands, e-mail: [email protected]
