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 79e801558a2c [SPARK-57924][SQL] Add bitmap_xor_agg aggregation function
79e801558a2c is described below
commit 79e801558a2c76f97594e2d12b676f5df3cb0893
Author: StormSpirit <[email protected]>
AuthorDate: Thu Aug 20 21:59:22 2026 +0800
[SPARK-57924][SQL] Add bitmap_xor_agg aggregation function
### What changes were proposed in this pull request?
Currently, Spark has three bitmap aggregation functions:
`bitmap_construct_agg`, `bitmap_or_agg` (since 3.5.0), and `bitmap_and_agg`
(since 4.1.0) for constructing, union, and intersection of sets represented by
bitmaps. However, the symmetric difference operation (bitwise XOR) is missing,
which is needed for use cases such as finding elements belonging to exactly one
of two groups and detecting changes between snapshots.
This PR adds a new aggregation function `bitmap_xor_agg` that performs
bitwise XOR on binary bitmap inputs, following the same `ImperativeAggregate`
pattern as `BitmapAndAgg`.
#### Design Decisions
- **Result on empty input is the XOR identity element (all zeros)**: An
ungrouped empty input, or an existing group with no matching rows after FILTER,
returns the all-zero bitmap as the XOR identity element (X ^ 0 = X).
- **Missing bytes handling**: For XOR operations, bytes beyond the shorter
input's length are left unchanged, which is equivalent to XOR with 0 (absent
bits). This is consistent with the OR operation and requires no explicit
trailing-byte zeroing unlike AND.
- **Input validation**: Only BINARY input is accepted, with the same
`UNEXPECTED_INPUT_TYPE` error class as the other bitmap aggregates.
### Why are the changes needed?
Symmetric difference is a fundamental set operation alongside union and
intersection. Other analytical engines provide this functionality. Adding
`bitmap_xor_agg` completes the bitmap aggregation function family.
Example:
```sql
-- Symmetric difference: 0x10 ^ 0x30 ^ 0x40 = 0x60
SELECT substring(hex(bitmap_xor_agg(col)), 0, 6)
FROM VALUES (X'10'), (X'30'), (X'40') AS tab(col);
-- 600000
-- Same values cancel out: 0x10 ^ 0x10 = 0x00
SELECT substring(hex(bitmap_xor_agg(col)), 0, 6)
FROM VALUES (X'10'), (X'10') AS tab(col);
-- 000000
-- Count set bits in the XOR result: 0x10 ^ 0x30 ^ 0x40 = 0x60
SELECT bitmap_count(bitmap_xor_agg(col)) AS symmetric_diff_count
FROM VALUES (X'10'), (X'30'), (X'40') AS tab(col);
-- 2
```
### Does this PR introduce _any_ user-facing change?
Yes. A new `bitmap_xor_agg` function is available in SQL, the Scala
DataFrame API, and PySpark (classic and Connect).
### How was this patch tested?
- **`BitmapExpressionUtilsSuite`**: 5 new unit tests covering equal-length
XOR, different-length XOR, all-zeros identity, self-XOR equals zero, and
sign-bit bytes.
- **`BitmapExpressionsQuerySuite`**: 3 new tests covering basic XOR,
same-value cancellation, zero identity, different-length bitmaps, empty input,
FILTER clause, GROUP BY, complex `bitmap_construct_agg` composition, and type
mismatch error handling.
- **`PlanGenerationTestSuite`**: Spark Connect plan generation test with
regenerated golden files.
- **`ProtoToParsedPlanTestSuite`**: Parsed-plan golden test validating the
Connect explain artifact.
- **`ExpressionsSchemaSuite`**: Regenerated the SQL expression schema
golden file.
- **`ExpressionInfoSuite`**: Verified the new expression examples.
- **PySpark doctest smoke**: Targeted doctests for the `bitmap_xor_agg`
wrapper in classic PySpark.
- **`spark-pr-precheck.py`**: Pre-push static checks (format, license,
golden, version consistency) all passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex and Claude AI
Closes #58025 from jiangxt2/feat/bitmap-xor-agg.
Authored-by: StormSpirit <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 7e9c387cf9ad5ac46801cea45d4ce21f5d109488)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../source/reference/pyspark.sql/functions.rst | 1 +
python/pyspark/sql/connect/functions/builtin.py | 7 ++
python/pyspark/sql/functions/__init__.py | 1 +
python/pyspark/sql/functions/builtin.py | 36 +++++++
.../scala/org/apache/spark/sql/functions.scala | 14 +++
.../expressions/BitmapExpressionUtils.java | 9 ++
.../sql/catalyst/analysis/FunctionRegistry.scala | 1 +
.../catalyst/expressions/bitmapExpressions.scala | 98 ++++++++++++++++++
.../expressions/BitmapExpressionUtilsSuite.scala | 53 ++++++++++
.../apache/spark/sql/PlanGenerationTestSuite.scala | 4 +
.../function_bitmap_xor_agg.explain | 2 +
.../queries/function_bitmap_xor_agg.json | 60 +++++++++++
.../queries/function_bitmap_xor_agg.proto.bin | Bin 0 -> 448 bytes
.../sql-functions/sql-expression-schema.md | 1 +
.../spark/sql/BitmapExpressionsQuerySuite.scala | 111 ++++++++++++++++++++-
15 files changed, 397 insertions(+), 1 deletion(-)
diff --git a/python/docs/source/reference/pyspark.sql/functions.rst
b/python/docs/source/reference/pyspark.sql/functions.rst
index 229fcaf6c381..9ac05bd3b0ed 100644
--- a/python/docs/source/reference/pyspark.sql/functions.rst
+++ b/python/docs/source/reference/pyspark.sql/functions.rst
@@ -455,6 +455,7 @@ Aggregate Functions
bit_xor
bitmap_construct_agg
bitmap_or_agg
+ bitmap_xor_agg
bool_and
bool_or
collect_list
diff --git a/python/pyspark/sql/connect/functions/builtin.py
b/python/pyspark/sql/connect/functions/builtin.py
index 9ad1bf43cb63..91fadcf056ab 100644
--- a/python/pyspark/sql/connect/functions/builtin.py
+++ b/python/pyspark/sql/connect/functions/builtin.py
@@ -5655,6 +5655,13 @@ def bitmap_and_agg(col: "ColumnOrName") -> Column:
bitmap_and_agg.__doc__ = pysparkfuncs.bitmap_and_agg.__doc__
+def bitmap_xor_agg(col: "ColumnOrName") -> Column:
+ return _invoke_function_over_columns("bitmap_xor_agg", col)
+
+
+bitmap_xor_agg.__doc__ = pysparkfuncs.bitmap_xor_agg.__doc__
+
+
# Geospatial ST Functions
diff --git a/python/pyspark/sql/functions/__init__.py
b/python/pyspark/sql/functions/__init__.py
index 001a7d4784ca..a5a95196c1ad 100644
--- a/python/pyspark/sql/functions/__init__.py
+++ b/python/pyspark/sql/functions/__init__.py
@@ -369,6 +369,7 @@ __all__ = [ # noqa: F405
"bitmap_and_agg",
"bitmap_construct_agg",
"bitmap_or_agg",
+ "bitmap_xor_agg",
"bool_and",
"bool_or",
"collect_list",
diff --git a/python/pyspark/sql/functions/builtin.py
b/python/pyspark/sql/functions/builtin.py
index 5662ec0487bc..44491a2f6faf 100644
--- a/python/pyspark/sql/functions/builtin.py
+++ b/python/pyspark/sql/functions/builtin.py
@@ -33057,6 +33057,42 @@ def bitmap_and_agg(col: "ColumnOrName") -> Column:
return _invoke_function_over_columns("bitmap_and_agg", col)
+@_try_remote_functions
+def bitmap_xor_agg(col: "ColumnOrName") -> Column:
+ """
+ Returns a bitmap that is the bitwise XOR of all of the bitmaps from the
input column.
+ The input column should be bitmaps created from bitmap_construct_agg().
+
+ .. versionadded:: 4.4.0
+
+ See Also
+ --------
+ :meth:`pyspark.sql.functions.bitmap_bit_position`
+ :meth:`pyspark.sql.functions.bitmap_bucket_number`
+ :meth:`pyspark.sql.functions.bitmap_construct_agg`
+ :meth:`pyspark.sql.functions.bitmap_count`
+ :meth:`pyspark.sql.functions.bitmap_or_agg`
+ :meth:`pyspark.sql.functions.bitmap_and_agg`
+
+ Parameters
+ ----------
+ col : :class:`~pyspark.sql.Column` or column name
+ The input column should be bitmaps created from bitmap_construct_agg().
+
+ Examples
+ --------
+ >>> from pyspark.sql import functions as sf
+ >>> df = spark.createDataFrame([("10",), ("30",), ("40",)], ["a"])
+ >>> df.select(sf.bitmap_xor_agg(sf.to_binary(df.a, sf.lit("hex")))).show()
+ +---------------------------------+
+ |bitmap_xor_agg(to_binary(a, hex))|
+ +---------------------------------+
+ | [60 00 00 00 00 0...|
+ +---------------------------------+
+ """
+ return _invoke_function_over_columns("bitmap_xor_agg", col)
+
+
# ---------------------------- User Defined Function
----------------------------------
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 053e0281fa5b..a66216bac9a3 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
@@ -7366,6 +7366,20 @@ object functions {
*/
def bitmap_and_agg(col: Column): Column = Column.fn("bitmap_and_agg", col)
+ /**
+ * Returns a bitmap that is the bitwise XOR of all of the bitmaps from the
input column. The
+ * input column should be bitmaps created from bitmap_construct_agg().
+ *
+ * @param col
+ * A column containing bitmaps created by bitmap_construct_agg() and
evaluating to binary
+ * data.
+ * @group agg_funcs
+ * @since 4.4.0
+ * @return
+ * Returns a column that evaluates to a binary.
+ */
+ def bitmap_xor_agg(col: Column): Column = Column.fn("bitmap_xor_agg", col)
+
//////////////////////////////////////////////////////////////////////////////////////////////
// String functions
//////////////////////////////////////////////////////////////////////////////////////////////
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 a505b8d7d756..d03db7d952a5 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
@@ -57,6 +57,15 @@ public class BitmapExpressionUtils {
}
}
+ /** Performs bitwise XOR on both bitmaps and writes the result into bitmap1.
*/
+ public static void bitmapXorMerge(byte[] bitmap1, byte[] bitmap2) {
+ // For XOR operation, bytes beyond the shorter bitmap's length are
unchanged
+ // since XOR with 0 (absent bits) is identity: X ^ 0 = X.
+ for (int i = 0; i < java.lang.Math.min(bitmap1.length, bitmap2.length);
++i) {
+ bitmap1[i] = (byte) ((bitmap1[i] & 0x0FF) ^ (bitmap2[i] & 0x0FF));
+ }
+ }
+
/** Performs bitwise AND on both bitmaps and writes the result into bitmap1.
*/
public static void bitmapAndMerge(byte[] bitmap1, byte[] bitmap2) {
int minLen = java.lang.Math.min(bitmap1.length, bitmap2.length);
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 0f091af61671..30b65c7ee608 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
@@ -975,6 +975,7 @@ object FunctionRegistry {
expression[BitmapCount]("bitmap_count"),
expression[BitmapOrAgg]("bitmap_or_agg"),
expression[BitmapAndAgg]("bitmap_and_agg"),
+ expression[BitmapXorAgg]("bitmap_xor_agg"),
// json
expression[StructsToJson]("to_json"),
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 87c04d229e91..02576f051979 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
@@ -445,3 +445,101 @@ case class BitmapAndAgg(
buffer.getBinary(mutableAggBufferOffset)
}
}
+
+@ExpressionDescription(
+ usage = """
+ _FUNC_(child) - Returns a bitmap that is the bitwise XOR of all of the
bitmaps from the child
+ expression. The input should be bitmaps created from
bitmap_construct_agg().
+ """,
+ arguments = """
+ Arguments:
+ * child - The expression whose bitmap values are combined with a bitwise
XOR.
+ An expression that evaluates to a binary bitmap created from
bitmap_construct_agg().
+ """,
+ // scalastyle:off line.size.limit
+ examples = """
+ Examples:
+ > SELECT substring(hex(_FUNC_(col)), 0, 6) FROM VALUES (X'10'), (X'30'),
(X'40') AS tab(col);
+ 600000
+ > SELECT substring(hex(_FUNC_(col)), 0, 6) FROM VALUES (X'10'), (X'10')
AS tab(col);
+ 000000
+ """,
+ // scalastyle:on line.size.limit
+ since = "4.4.0",
+ group = "agg_funcs")
+case class BitmapXorAgg(
+ child: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0)
+ extends ImperativeAggregate
+ with UnaryLike[Expression] {
+
+ def this(child: Expression) = {
+ this(child = child, mutableAggBufferOffset = 0, inputAggBufferOffset = 0)
+ }
+
+ override def checkInputDataTypes(): TypeCheckResult = {
+ if (child.dataType != BinaryType) {
+ DataTypeMismatch(
+ errorSubClass = "UNEXPECTED_INPUT_TYPE",
+ messageParameters = Map(
+ "paramIndex" -> ordinalNumber(0),
+ "requiredType" -> toSQLType(BinaryType),
+ "inputSql" -> toSQLExpr(child),
+ "inputType" -> toSQLType(child.dataType)))
+ } else {
+ TypeCheckSuccess
+ }
+ }
+
+ override def dataType: DataType = BinaryType
+
+ override def prettyName: String = "bitmap_xor_agg"
+
+ override protected def withNewChildInternal(newChild: Expression):
BitmapXorAgg =
+ copy(child = newChild)
+
+ override def withNewMutableAggBufferOffset(
+ newMutableAggBufferOffset: Int): ImperativeAggregate =
+ copy(mutableAggBufferOffset = newMutableAggBufferOffset)
+
+ override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int):
ImperativeAggregate =
+ copy(inputAggBufferOffset = newInputAggBufferOffset)
+
+ override def nullable: Boolean = false
+
+ override def aggBufferSchema: StructType =
DataTypeUtils.fromAttributes(aggBufferAttributes)
+
+ // The aggregation buffer is a fixed size binary.
+ private val bitmapAttr = AttributeReference("bitmap", BinaryType, false)()
+
+ override def aggBufferAttributes: Seq[AttributeReference] = bitmapAttr :: Nil
+
+ override def defaultResult: Option[Literal] =
+ Option(Literal(Array.fill[Byte](BitmapExpressionUtils.NUM_BYTES)(0)))
+
+ override val inputAggBufferAttributes: Seq[AttributeReference] =
+ aggBufferAttributes.map(_.newInstance())
+
+ override def initialize(buffer: InternalRow): Unit = {
+ buffer.update(mutableAggBufferOffset,
Array.fill[Byte](BitmapExpressionUtils.NUM_BYTES)(0))
+ }
+
+ override def update(buffer: InternalRow, input: InternalRow): Unit = {
+ val input_bitmap = child.eval(input).asInstanceOf[Array[Byte]]
+ if (input_bitmap != null) {
+ val bitmap = buffer.getBinary(mutableAggBufferOffset)
+ BitmapExpressionUtils.bitmapXorMerge(bitmap, input_bitmap)
+ }
+ }
+
+ override def merge(buffer1: InternalRow, buffer2: InternalRow): Unit = {
+ val bitmap1 = buffer1.getBinary(mutableAggBufferOffset)
+ val bitmap2 = buffer2.getBinary(inputAggBufferOffset)
+ BitmapExpressionUtils.bitmapXorMerge(bitmap1, bitmap2)
+ }
+
+ override def eval(buffer: InternalRow): Any = {
+ buffer.getBinary(mutableAggBufferOffset)
+ }
+}
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 53935c66c613..c33a14aa7b42 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
@@ -92,4 +92,57 @@ class BitmapExpressionUtilsSuite extends SparkFunSuite {
setBitmapBits(bitmap, bitmap.length - 1, 0x67)
assert(BitmapExpressionUtils.bitmapCount(bitmap) == 15L)
}
+
+ test("bitmap_xor_merge equal length") {
+ val bitmap1 = Array[Byte](0x10, 0x30, 0x40)
+ val bitmap2 = Array[Byte](0x10, 0x20, 0x40)
+ // 0x10 ^ 0x10 = 0x00, 0x30 ^ 0x20 = 0x10, 0x40 ^ 0x40 = 0x00
+ val expected = Array[Byte](0x00, 0x10, 0x00)
+ BitmapExpressionUtils.bitmapXorMerge(bitmap1, bitmap2)
+ for (i <- expected.indices) {
+ assert(bitmap1(i) == expected(i), s"bitmap1($i) should be
${expected(i)}")
+ }
+ }
+
+ test("bitmap_xor_merge different lengths") {
+ val bitmap1 = Array[Byte](0x0A, 0x0B, 0x0C)
+ val bitmap2 = Array[Byte](0x0A)
+ // 0x0A ^ 0x0A = 0x00, remaining bytes unchanged because XOR 0 = X
+ val expected = Array[Byte](0x00, 0x0B, 0x0C)
+ BitmapExpressionUtils.bitmapXorMerge(bitmap1, bitmap2)
+ for (i <- expected.indices) {
+ assert(bitmap1(i) == expected(i), s"bitmap1($i) should be
${expected(i)}")
+ }
+ }
+
+ test("bitmap_xor_merge all zeros") {
+ val bitmap1 = Array[Byte](0x10, 0x20)
+ val bitmap2 = Array[Byte](0x00, 0x00)
+ val expected = Array[Byte](0x10, 0x20)
+ BitmapExpressionUtils.bitmapXorMerge(bitmap1, bitmap2)
+ for (i <- expected.indices) {
+ assert(bitmap1(i) == expected(i), s"bitmap1($i) should be
${expected(i)}")
+ }
+ }
+
+ test("bitmap_xor_merge self xor equals zero") {
+ val bitmap1 = Array[Byte](0x10, 0x30, 0x40)
+ val bitmap2 = Array[Byte](0x10, 0x30, 0x40)
+ val expected = Array[Byte](0x00, 0x00, 0x00)
+ BitmapExpressionUtils.bitmapXorMerge(bitmap1, bitmap2)
+ for (i <- expected.indices) {
+ assert(bitmap1(i) == expected(i), s"bitmap1($i) should be
${expected(i)}")
+ }
+ }
+
+ test("bitmap_xor_merge with bytes containing sign bits") {
+ val bitmap1 = Array[Byte](0xFF.toByte, 0x80.toByte)
+ val bitmap2 = Array[Byte](0xF0.toByte, 0x0F.toByte)
+ // 0xFF ^ 0xF0 = 0x0F, 0x80 ^ 0x0F = 0x8F
+ val expected = Array[Byte](0x0F.toByte, 0x8F.toByte)
+ BitmapExpressionUtils.bitmapXorMerge(bitmap1, bitmap2)
+ for (i <- expected.indices) {
+ assert(bitmap1(i) == expected(i), s"bitmap1($i) should be
${expected(i)}")
+ }
+ }
}
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 de37f913016d..88f887d302a5 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,10 @@ class PlanGenerationTestSuite extends ConnectFunSuite
with Logging {
binary.select(fn.bitmap_or_agg(fn.col("bytes")))
}
+ test("function bitmap_xor_agg") {
+ binary.select(fn.bitmap_xor_agg(fn.col("bytes")))
+ }
+
private def temporalFunctionTest(name: String)(f: => Column): Unit = {
test("function " + name) {
temporals.select(f)
diff --git
a/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_xor_agg.explain
b/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_xor_agg.explain
new file mode 100644
index 000000000000..ee4bb977934c
--- /dev/null
+++
b/sql/connect/common/src/test/resources/query-tests/explain-results/function_bitmap_xor_agg.explain
@@ -0,0 +1,2 @@
+Aggregate [bitmap_xor_agg(bytes#0, 0, 0) AS bitmap_xor_agg(bytes)#0]
++- LocalRelation <empty>, [id#0L, bytes#0]
diff --git
a/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_xor_agg.json
b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_xor_agg.json
new file mode 100644
index 000000000000..12f2878b6b97
--- /dev/null
+++
b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_xor_agg.json
@@ -0,0 +1,60 @@
+{
+ "common": {
+ "planId": "1"
+ },
+ "project": {
+ "input": {
+ "common": {
+ "planId": "0"
+ },
+ "localRelation": {
+ "schema": "struct\u003cid:bigint,bytes:binary\u003e"
+ }
+ },
+ "expressions": [{
+ "unresolvedFunction": {
+ "functionName": "bitmap_xor_agg",
+ "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"
+ }]
+ }
+ }
+ }
+ }],
+ "isInternal": false
+ },
+ "common": {
+ "origin": {
+ "jvmOrigin": {
+ "stackTrace": [{
+ "classLoaderName": "app",
+ "declaringClass": "org.apache.spark.sql.functions$",
+ "methodName": "bitmap_xor_agg",
+ "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_agg.proto.bin
b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_xor_agg.proto.bin
new file mode 100644
index 000000000000..27ca4a6c3880
Binary files /dev/null and
b/sql/connect/common/src/test/resources/query-tests/queries/function_bitmap_xor_agg.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 8a4d226fd94a..2db516884258 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
@@ -55,6 +55,7 @@
| 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.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.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> |
| org.apache.spark.sql.catalyst.expressions.BitwiseGet | bit_get | SELECT
bit_get(11, 0) | struct<bit_get(11, 0):tinyint> |
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 458688de3fbd..8201e2c2f872 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,7 +17,20 @@
package org.apache.spark.sql
-import org.apache.spark.sql.functions.{bitmap_and_agg, bitmap_bit_position,
bitmap_bucket_number, bitmap_construct_agg, bitmap_count, bitmap_or_agg, col,
expr, hex, lit, substring, to_binary}
+import org.apache.spark.sql.functions.{
+ bitmap_and_agg,
+ bitmap_bit_position,
+ bitmap_bucket_number,
+ bitmap_construct_agg,
+ bitmap_count,
+ bitmap_or_agg,
+ bitmap_xor_agg,
+ col,
+ expr,
+ hex,
+ lit,
+ substring,
+ to_binary}
import org.apache.spark.sql.test.SharedSparkSession
class BitmapExpressionsQuerySuite extends SharedSparkSession {
@@ -356,4 +369,100 @@ class BitmapExpressionsQuerySuite extends
SharedSparkSession {
"inputType" -> "\"INT\""),
context = ExpectedContext(fragment = "bitmap_and_agg(a)", start = 0,
stop = 16))
}
+
+ test("bitmap_xor_agg") {
+ // Test basic XOR functionality: 10 ^ 30 ^ 40 = 60
+ val df = Seq("10", "30", "40").toDF("a")
+ checkAnswer(
+ df.selectExpr("substring(hex(bitmap_xor_agg(to_binary(a, 'hex'))), 0,
6)"),
+ Seq(Row("600000")))
+ checkAnswer(
+ df.select(substring(hex(bitmap_xor_agg(to_binary(col("a"),
lit("hex")))), 0, 6)),
+ Seq(Row("600000")))
+
+ // Test with same values - XOR should result in zero: 10 ^ 10 = 00
+ val df2 = Seq("10", "10").toDF("a")
+ checkAnswer(
+ df2.selectExpr("substring(hex(bitmap_xor_agg(to_binary(a, 'hex'))), 0,
6)"),
+ Seq(Row("000000")))
+
+ // Test with zero - X XOR 0 = X: A0 ^ 00 = A0
+ val df3 = Seq("A0", "00").toDF("a")
+ checkAnswer(
+ df3.selectExpr("substring(hex(bitmap_xor_agg(to_binary(a, 'hex'))), 0,
6)"),
+ Seq(Row("A00000")))
+
+ // Test with binary values of different lengths: 0A0B ^ 0A = 000B
+ val df4 = Seq("0A0B", "0A").toDF("a")
+ checkAnswer(
+ df4.selectExpr("substring(hex(bitmap_xor_agg(to_binary(a, 'hex'))), 0,
6)"),
+ Seq(Row("000B00")))
+
+ // Test empty result (no rows) - should return all zeros as XOR identity
+ val emptyDf = Seq.empty[String].toDF("a")
+ checkAnswer(
+ emptyDf.selectExpr("substring(hex(bitmap_xor_agg(to_binary(a, 'hex'))),
0, 6)"),
+ Seq(Row("000000")))
+
+ val emptyDf2 = Seq.empty[(String, Int)].toDF("a", "b")
+ checkAnswer(
+ emptyDf2
+ .selectExpr("bitmap_xor_agg(to_binary(a, 'hex')) " +
+ "filter (where b = 1) as xor_agg")
+ .select(substring(hex(col("xor_agg")), 0, 6)),
+ Seq(Row("000000")))
+
+ // Test empty result (no rows) with GROUP BY - should return empty
DataFrame
+ val emptyDf3 = Seq.empty[(String, Int, Int)].toDF("a", "b", "c")
+ checkAnswer(
+ emptyDf3
+ .groupBy("c")
+ .agg(expr("bitmap_xor_agg(to_binary(a, 'hex')) " +
+ "filter (where b = 1) as xor_agg").alias("xor_agg"))
+ .select(substring(hex(col("xor_agg")), 0, 6)),
+ Seq())
+ }
+
+ test("bitmap_xor_agg with complex bitmaps from bitmap_construct_agg") {
+ val table = "bitmap_xor_test_table"
+ withTable(table) {
+ // Create test data: Group 1 has bits 1,2,4; Group 2 has bits 1,2,3.
+ spark.sql(s"""
+ | CREATE TABLE $table (group_id INT, bit_pos LONG)
+ | """.stripMargin)
+ spark.sql(s"""
+ | INSERT INTO $table VALUES
+ | (1, 1), (1, 2), (1, 4),
+ | (2, 1), (2, 2), (2, 3)
+ | """.stripMargin)
+ // XOR of the group bitmaps cancels bits 1 and 2, leaving bits 3 and 4.
+ val xorResult = spark.sql(s"""
+ | SELECT bitmap_count(
+ | bitmap_xor_agg(group_bitmap)
+ | ) as xor_count
+ | FROM (
+ | SELECT bitmap_construct_agg(bitmap_bit_position(bit_pos)) as
group_bitmap
+ | FROM $table
+ | GROUP BY group_id
+ | )
+ | """.stripMargin)
+ checkAnswer(xorResult, Seq(Row(2)))
+ }
+ }
+
+ test("bitmap_xor_agg called with non-binary type") {
+ val df = Seq(12).toDF("a")
+ checkError(
+ exception = intercept[AnalysisException] {
+ df.selectExpr("bitmap_xor_agg(a)")
+ },
+ condition = "DATATYPE_MISMATCH.UNEXPECTED_INPUT_TYPE",
+ parameters = Map(
+ "sqlExpr" -> "\"bitmap_xor_agg(a)\"",
+ "paramIndex" -> "first",
+ "requiredType" -> "\"BINARY\"",
+ "inputSql" -> "\"a\"",
+ "inputType" -> "\"INT\""),
+ context = ExpectedContext(fragment = "bitmap_xor_agg(a)", start = 0,
stop = 16))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
