cloud-fan commented on code in PR #57592:
URL: https://github.com/apache/spark/pull/57592#discussion_r3675457447
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala:
##########
@@ -316,6 +316,175 @@ case class CollectSet(
copy(child = newChild)
}
+/**
+ * Collect the distinct union of the elements of an array-typed input across
rows.
+ *
+ * Unlike collect_set, whose input is a scalar and whose output is the set of
those scalars,
+ * collect_union's input is itself an array and its output is the set of the
array's
+ * *elements* unioned across all rows. The aggregation buffer holds only the
distinct
+ * elements (a set), so its size is bounded by the element universe rather
than by the
+ * number of input rows.
+ *
+ * Null handling mirrors collect_set: by default (IGNORE NULLS) null elements
are dropped.
+ * With RESPECT NULLS, one null element is kept, in which case collect_union
is equivalent to
+ * `array_distinct(flatten(collect_list(arr)))`.
+ *
+ * @param ignoreNulls when true (IGNORE NULLS, the default), null elements are
excluded from
+ * the result array. When false (RESPECT NULLS), a single
null element is
+ * kept.
+ */
+@ExpressionDescription(
+ usage =
+ "_FUNC_(expr) - Collects and returns the distinct union of the elements of
array `expr`.",
+ examples = """
+ Examples:
+ > SELECT _FUNC_(col) FROM VALUES (array(1, 2)), (array(2, 3)),
(array(1)) AS tab(col);
+ [1,2,3]
+ """,
+ note = """
+ The function is non-deterministic because the order of collected results
depends
+ on the order of the rows which may be non-deterministic after a shuffle.
+ """,
+ group = "agg_funcs",
+ since = "4.3.0")
+case class CollectUnion(
+ child: Expression,
+ mutableAggBufferOffset: Int = 0,
+ inputAggBufferOffset: Int = 0,
+ ignoreNulls: Boolean = true)
+ extends Collect[mutable.HashSet[Any]] with QueryErrorsBase with
UnaryLike[Expression] {
+
+ def this(child: Expression) = this(child, 0, 0, true)
+
+ // The input is guarded by checkInputDataTypes to be an ArrayType; this is
its element type.
+ private lazy val elementType: DataType = child.dataType match {
+ case ArrayType(et, _) => et
+ case other => other
+ }
+
+ // The result array contains a null only when null elements are respected
(RESPECT NULLS).
+ override protected def bufferContainsNull: Boolean = !ignoreNulls
+
+ // Result is array<elementType>; nullable iff null elements are kept.
Review Comment:
`nullable` describes whether the array value itself may be null, while this
field controls whether its elements may be null.
```suggestion
// Result is array<elementType>; containsNull is true iff null elements
are kept.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
