[SPARK-29545][SQL] Add support for bit_xor aggregate function

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala

@@ -527,6 +527,7 @@ object FunctionRegistry {
     expression[BitwiseCount]("bit_count"),
     expression[BitAndAgg]("bit_and"),
     expression[BitOrAgg]("bit_or"),
+    expression[BitXorAgg]("bit_xor"),
 
     // json
     expression[StructsToJson]("to_json"),

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/bitwiseAggregates.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.expressions.aggregate
 
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference, BitwiseAnd, BitwiseOr, ExpectsInputTypes, Expression, ExpressionDescription, If, IsNull, Literal}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, BitwiseAnd, BitwiseOr, BitwiseXor, ExpectsInputTypes, Expression, ExpressionDescription, If, IsNull, Literal}
 import org.apache.spark.sql.types.{AbstractDataType, DataType, IntegralType}
 
 @ExpressionDescription(
@@ -97,3 +97,42 @@ case class BitOrAgg(child: Expression) extends DeclarativeAggregate with Expects
 
   override lazy val evaluateExpression: AttributeReference = bitOr
 }
+
+@ExpressionDescription(
+  usage = "_FUNC_(expr) - Returns the bitwise XOR of all non-null input values, or null if none.",
+  examples = """
+    Examples:
+      > SELECT _FUNC_(col) FROM VALUES (3), (5) AS tab(col);
+       6
+  """,
+  since = "3.0.0")
+case class BitXorAgg(child: Expression) extends DeclarativeAggregate with ExpectsInputTypes {
+
+  override def nodeName: String = "bit_xor"
+
+  override def children: Seq[Expression] = child :: Nil
+
+  override def nullable: Boolean = true
+
+  override def dataType: DataType = child.dataType
+
+  override def inputTypes: Seq[AbstractDataType] = Seq(IntegralType)
+
+  private lazy val bitXOr = AttributeReference("bit_xor", child.dataType)()
+
+  override lazy val aggBufferAttributes: Seq[AttributeReference] = bitXOr :: Nil
+
+  override lazy val initialValues: Seq[Literal] = Literal.create(null, dataType) :: Nil
+
+  override lazy val updateExpressions: Seq[Expression] =
+    If(IsNull(bitXOr),
+      child,
+      If(IsNull(child), bitXOr, BitwiseXor(bitXOr, child))) :: Nil
+
+  override lazy val mergeExpressions: Seq[Expression] =
+    If(IsNull(bitXOr.left),
+      bitXOr.right,
+      If(IsNull(bitXOr.right), bitXOr.left, BitwiseXor(bitXOr.left, bitXOr.right))) :: Nil
+
+  override lazy val evaluateExpression: AttributeReference = bitXOr
+}

sql/core/src/test/resources/sql-tests/inputs/postgreSQL/aggregates_part2.sql

@@ -47,11 +47,12 @@ CREATE OR REPLACE TEMPORARY VIEW bitwise_test AS SELECT * FROM VALUES
   (7, 7, 7, 3L) AS bitwise_test(b1, b2, b3, b4);
 
 -- empty case
-SELECT BIT_AND(b1) AS n1, BIT_OR(b2)  AS n2 FROM bitwise_test where 1 = 0;
+SELECT BIT_AND(b1) AS n1, BIT_OR(b2) AS n2, BIT_XOR(b3) AS n3 FROM bitwise_test where 1 = 0;
 
 -- null case
-SELECT BIT_AND(b4) AS n1, BIT_OR(b4)  AS n2 FROM bitwise_test where b4 is null;
+SELECT BIT_AND(b4) AS n1, BIT_OR(b4)  AS n2, BIT_XOR(b4) AS n3 FROM bitwise_test where b4 is null;
 
+-- the suffix numbers show the expected answer
 SELECT
  BIT_AND(cast(b1 as tinyint)) AS a1,
  BIT_AND(cast(b2 as smallint)) AS b1,
@@ -60,18 +61,24 @@ SELECT
  BIT_OR(cast(b1 as tinyint))  AS e7,
  BIT_OR(cast(b2 as smallint))  AS f7,
  BIT_OR(b3)  AS g7,
- BIT_OR(b4)  AS h3
+ BIT_OR(b4)  AS h3,
+ BIT_XOR(cast(b1 as tinyint))  AS i5,
+ BIT_XOR(cast(b2 as smallint))  AS j5,
+ BIT_XOR(b3)  AS k5,
+ BIT_XOR(b4)  AS l2,
+ BIT_XOR(distinct b4) AS m2
 FROM bitwise_test;
 
 -- group by
-SELECT b1 , bit_and(b2), bit_or(b4) FROM bitwise_test GROUP BY b1;
+SELECT b1 , bit_and(b2), bit_or(b4), bit_xor(b3) FROM bitwise_test GROUP BY b1;
 
 --having
 SELECT b1, bit_and(b2) FROM bitwise_test GROUP BY b1 HAVING bit_and(b2) < 7;
 
 -- window
 SELECT b1, b2, bit_and(b2) OVER (PARTITION BY b1 ORDER BY b2) FROM bitwise_test;
 SELECT b1, b2, bit_or(b2) OVER (PARTITION BY b1 ORDER BY b2) FROM bitwise_test;
+SELECT b1, b2, bit_xor(b2) OVER (PARTITION BY b1 ORDER BY b2) FROM bitwise_test;
 
 --
 -- test boolean aggregates