You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/09/03 19:29:53 UTC
(spark-connect-swift) branch main updated: [SPARK-59215] Support general aggregate functions
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-connect-swift.git
The following commit(s) were added to refs/heads/main by this push:
new 9e01045 [SPARK-59215] Support general aggregate functions
9e01045 is described below
commit 9e01045f757b9a77e436d772a92cf193fb7761d2
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Sep 3 12:29:45 2026 -0700
[SPARK-59215] Support general aggregate functions
### What changes were proposed in this pull request?
This PR aims to support 21 general aggregate functions in the Swift Spark Connect
client by adding them to `Sources/SparkConnect/AggregateFunctions.swift`, keeping
the file's alphabetical order:
| Function | Since | Notes |
| --- | --- | --- |
| `approx_percentile(col, percentage, accuracy = 10000)` | 3.5.0 | `percentage` is a `Double` or `[Double]` |
| `array_agg(col)` | 3.5.0 | Alias of `collect_list` |
| `bit_and(col)` | 3.5.0 | Bitwise `AND` of all non-null input values |
| `bit_or(col)` | 3.5.0 | Bitwise `OR` of all non-null input values |
| `bit_xor(col)` | 3.5.0 | Bitwise `XOR` of all non-null input values |
| `collect_union(col)` | 4.3.0 | Distinct union of array elements across rows |
| `count_min_sketch(col, eps, confidence, seed?)` | 3.5.0 | Returns a binary sketch |
| `every(col)` | 3.5.0 | Alias of `bool_and` |
| `first_value(col, ignoreNulls?)` | 3.5.0 | |
| `histogram_numeric(col, nBins)` | 3.5.0 | |
| `last_value(col, ignoreNulls?)` | 3.5.0 | |
| `listagg(col, delimiter?)` | 4.0.0 | |
| `listagg_distinct(col, delimiter?)` | 4.0.0 | |
| `percentile(col, percentage, frequency = 1)` | 3.5.0 | `percentage` is a `Double` or `[Double]` |
| `product(col)` | 3.2.0 | |
| `some(col)` | 3.5.0 | Alias of `bool_or` |
| `std(col)` | 3.5.0 | Alias of `stddev_samp` |
| `string_agg(col, delimiter?)` | 4.0.0 | Alias of `listagg` |
| `string_agg_distinct(col, delimiter?)` | 4.0.0 | Alias of `listagg_distinct` |
| `try_avg(col)` | 3.5.0 | `null` on overflow |
| `try_sum(col)` | 3.5.0 | `null` on overflow |
All of these belong to the upstream "Aggregate Functions" documentation group,
including `bit_and`/`bit_or`/`bit_xor` (which are aggregates, unlike the scalar
bitwise functions in `BitwiseFunctions.swift`) and `try_avg`/`try_sum`.
Three details are worth calling out, since the naive mapping does not work:
1. `listagg_distinct` and `string_agg_distinct` are **not** separate function
names on the server. `FunctionRegistry` registers only `listagg` and
`string_agg`, and both `sql/api` `functions.scala`
(`Column.fn("listagg", isDistinct = true, e)`) and the Python Spark Connect
client (`UnresolvedFunction("listagg", _exprs, is_distinct=True)`) send the
base name with the `is_distinct` flag set. This PR follows the same pattern
already used by `count_distinct` and `sum_distinct`. Sending
`listagg_distinct` as a function name fails with `UNRESOLVED_ROUTINE`.
2. `count_min_sketch` always requires four arguments:
`CountMinSketchAggExpressionBuilder` declares `seed` without a default, so a
three-argument call fails with `REQUIRED_PARAMETER_NOT_FOUND`. The overload
without a seed therefore generates a random `Int64` seed on the client, which
is what `functions.count_min_sketch(e, eps, confidence)` does in Scala with
`lit(SparkClassUtils.random.nextLong)`.
3. `approx_percentile` is a distinct registered function from the existing
`percentile_approx`, so it is added rather than aliased.
For the array form of `percentage`, the `[Double]` overloads build an `array(...)`
of literals, which stays foldable and satisfies the analyzer's requirement that
the percentage expression be a constant.
### Why are the changes needed?
To improve API coverage and feature parity with PySpark and Spark SQL. These
aggregate functions were entirely missing from the Swift client, including the
SQL-standard aliases (`array_agg`, `std`, `every`, `some`, `first_value`,
`last_value`) that users coming from other SQL engines are most likely to reach
for first.
### Does this PR introduce _any_ user-facing change?
No, this is an additive change that introduces new APIs.
```swift
let df = try await spark.sql("SELECT * FROM VALUES (1), (2), (3) T(v)")
try await df.select(bit_and(col("v")), bit_or(col("v")), bit_xor(col("v"))).show()
try await df.select(percentile(col("v"), 0.5), approx_percentile(col("v"), [0.0, 1.0])).show()
try await df.select(try_avg(col("v")), try_sum(col("v")), product(col("v"))).show()
```
Non-`Column` arguments such as `accuracy`, `frequency`, `nBins`, `delimiter`, and
`seed` are taken as Swift primitives and wrapped with `lit(...)`, following the
existing convention (e.g. `approx_count_distinct(_:_:)` and
`regexp_extract(_:_:_:)`). Optional arguments are expressed as overloads, except
`accuracy` and `frequency`, which use their upstream default values.
### How was this patch tested?
Pass the CIs with new test cases added to `AggregateFunctionsTests`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
Closes #521 from dongjoon-hyun/SPARK-59215.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
Sources/SparkConnect/AggregateFunctions.swift | 286 +++++++++++++++++++++
.../AggregateFunctionsTests.swift | 203 +++++++++++++++
2 files changed, 489 insertions(+)
diff --git a/Sources/SparkConnect/AggregateFunctions.swift b/Sources/SparkConnect/AggregateFunctions.swift
index d065a48..e8770c6 100644
--- a/Sources/SparkConnect/AggregateFunctions.swift
+++ b/Sources/SparkConnect/AggregateFunctions.swift
@@ -50,6 +50,65 @@ public func approx_count_distinct(_ col: Column, _ rsd: Double) -> Column {
return fn("approx_count_distinct", col, lit(rsd))
}
+/// Returns the approximate `percentile` of the numeric column `col` which is the smallest value
+/// in the ordered `col` values (sorted from least to greatest) such that no more than `percentage`
+/// of `col` values is less than the value or equal to that value.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - percentage: A percentage. Must be between 0.0 and 1.0.
+/// - accuracy: A positive numeric literal that controls approximation accuracy
+/// at the cost of memory (default = 10000).
+/// - Returns: A ``Column``.
+public func approx_percentile(_ col: Column, _ percentage: Double, _ accuracy: Int32 = 10000)
+ -> Column
+{
+ return fn("approx_percentile", col, lit(percentage), lit(accuracy))
+}
+
+/// Returns the approximate `percentile`s of the numeric column `col` which are the smallest values
+/// in the ordered `col` values (sorted from least to greatest) such that no more than each
+/// `percentage` of `col` values is less than the value or equal to that value.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - percentage: An array of percentages. Each value must be between 0.0 and 1.0.
+/// - accuracy: A positive numeric literal that controls approximation accuracy
+/// at the cost of memory (default = 10000).
+/// - Returns: A ``Column``.
+public func approx_percentile(_ col: Column, _ percentage: [Double], _ accuracy: Int32 = 10000)
+ -> Column
+{
+ return fn("approx_percentile", col, fn("array", percentage.map { lit($0) }), lit(accuracy))
+}
+
+/// Returns a list of objects with duplicates.
+/// This is an alias of ``collect_list(_:)``.
+/// - Parameter col: A ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func array_agg(_ col: Column) -> Column {
+ return fn("array_agg", col)
+}
+
+/// Returns the bitwise `AND` of all non-null input values, or null if none.
+/// - Parameter col: A ``Column`` that evaluates to an integral.
+/// - Returns: A ``Column``.
+public func bit_and(_ col: Column) -> Column {
+ return fn("bit_and", col)
+}
+
+/// Returns the bitwise `OR` of all non-null input values, or null if none.
+/// - Parameter col: A ``Column`` that evaluates to an integral.
+/// - Returns: A ``Column``.
+public func bit_or(_ col: Column) -> Column {
+ return fn("bit_or", col)
+}
+
+/// Returns the bitwise `XOR` of all non-null input values, or null if none.
+/// - Parameter col: A ``Column`` that evaluates to an integral.
+/// - Returns: A ``Column``.
+public func bit_xor(_ col: Column) -> Column {
+ return fn("bit_xor", col)
+}
+
/// Returns true if all values of the column are true.
/// - Parameter col: A ``Column`` that evaluates to a boolean.
/// - Returns: A ``Column``.
@@ -78,6 +137,17 @@ public func collect_set(_ col: Column) -> Column {
return fn("collect_set", col)
}
+/// Given an array-typed column, collects the distinct union of the elements of the arrays
+/// across rows and returns it as an array.
+///
+/// Null elements are dropped by default, matching ``collect_set(_:)``.
+/// This requires Apache Spark 4.3.0 or later.
+/// - Parameter col: An array ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func collect_union(_ col: Column) -> Column {
+ return fn("collect_union", col)
+}
+
/// Returns the Pearson Correlation Coefficient for two columns.
/// - Parameters:
/// - column1: A ``Column``.
@@ -113,6 +183,32 @@ public func count_if(_ col: Column) -> Column {
return fn("count_if", col)
}
+/// Returns a count-min sketch of the column with the given `eps` and `confidence`,
+/// using a randomly generated seed.
+/// The result is an array of bytes, which can be deserialized to a `CountMinSketch` before usage.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - eps: The relative error. Must be positive.
+/// - confidence: The confidence. Must be positive and less than 1.0.
+/// - Returns: A ``Column``.
+public func count_min_sketch(_ col: Column, _ eps: Double, _ confidence: Double) -> Column {
+ return count_min_sketch(col, eps, confidence, Int64.random(in: Int64.min...Int64.max))
+}
+
+/// Returns a count-min sketch of the column with the given `eps`, `confidence` and `seed`.
+/// The result is an array of bytes, which can be deserialized to a `CountMinSketch` before usage.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - eps: The relative error. Must be positive.
+/// - confidence: The confidence. Must be positive and less than 1.0.
+/// - seed: The random seed.
+/// - Returns: A ``Column``.
+public func count_min_sketch(_ col: Column, _ eps: Double, _ confidence: Double, _ seed: Int64)
+ -> Column
+{
+ return fn("count_min_sketch", col, lit(eps), lit(confidence), lit(seed))
+}
+
/// Returns the population covariance for two columns.
/// - Parameters:
/// - column1: A ``Column``.
@@ -131,6 +227,14 @@ public func covar_samp(_ column1: Column, _ column2: Column) -> Column {
return fn("covar_samp", column1, column2)
}
+/// Returns true if all values of the column are true.
+/// This is an alias of ``bool_and(_:)``.
+/// - Parameter col: A ``Column`` that evaluates to a boolean.
+/// - Returns: A ``Column``.
+public func every(_ col: Column) -> Column {
+ return fn("every", col)
+}
+
/// Returns the first value in a group.
///
/// The function by default returns the first values it sees. It will return the first non-null
@@ -153,6 +257,27 @@ public func first(_ col: Column, _ ignoreNulls: Bool) -> Column {
return fn("first", col, lit(ignoreNulls))
}
+/// Returns the first value in a group.
+/// This is similar to ``first(_:)``.
+/// - Parameter col: A ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func first_value(_ col: Column) -> Column {
+ return fn("first_value", col)
+}
+
+/// Returns the first value in a group.
+///
+/// The function by default returns the first values it sees. It will return the first non-null
+/// value it sees when `ignoreNulls` is set to true. If all values are null, then null is returned.
+/// This is similar to ``first(_:_:)``.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - ignoreNulls: Whether to skip null values.
+/// - Returns: A ``Column``.
+public func first_value(_ col: Column, _ ignoreNulls: Bool) -> Column {
+ return fn("first_value", col, lit(ignoreNulls))
+}
+
/// Indicates whether a specified column in a GROUP BY list is aggregated or not,
/// returns 1 for aggregated or 0 for not aggregated in the result set.
/// - Parameter col: A ``Column`` to check.
@@ -168,6 +293,15 @@ public func grouping_id(_ cols: Column...) -> Column {
return fn("grouping_id", cols)
}
+/// Computes a histogram on numeric `col` using `nBins` bins.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - nBins: The number of bins. Must be greater than 1.
+/// - Returns: A ``Column``.
+public func histogram_numeric(_ col: Column, _ nBins: Int32) -> Column {
+ return fn("histogram_numeric", col, lit(nBins))
+}
+
/// Returns the kurtosis of the values in a group.
/// - Parameter col: A ``Column`` to aggregate.
/// - Returns: A ``Column``.
@@ -197,6 +331,63 @@ public func last(_ col: Column, _ ignoreNulls: Bool) -> Column {
return fn("last", col, lit(ignoreNulls))
}
+/// Returns the last value in a group.
+/// This is similar to ``last(_:)``.
+/// - Parameter col: A ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func last_value(_ col: Column) -> Column {
+ return fn("last_value", col)
+}
+
+/// Returns the last value in a group.
+///
+/// The function by default returns the last values it sees. It will return the last non-null
+/// value it sees when `ignoreNulls` is set to true. If all values are null, then null is returned.
+/// This is similar to ``last(_:_:)``.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - ignoreNulls: Whether to skip null values.
+/// - Returns: A ``Column``.
+public func last_value(_ col: Column, _ ignoreNulls: Bool) -> Column {
+ return fn("last_value", col, lit(ignoreNulls))
+}
+
+/// Returns the concatenation of the non-null input values.
+/// This requires Apache Spark 4.0.0 or later.
+/// - Parameter col: A ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func listagg(_ col: Column) -> Column {
+ return fn("listagg", col)
+}
+
+/// Returns the concatenation of the non-null input values, separated by `delimiter`.
+/// This requires Apache Spark 4.0.0 or later.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - delimiter: A delimiter to separate the values.
+/// - Returns: A ``Column``.
+public func listagg(_ col: Column, _ delimiter: String) -> Column {
+ return fn("listagg", col, lit(delimiter))
+}
+
+/// Returns the concatenation of the distinct non-null input values.
+/// This requires Apache Spark 4.0.0 or later.
+/// - Parameter col: A ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func listagg_distinct(_ col: Column) -> Column {
+ return fn("listagg", [col], isDistinct: true)
+}
+
+/// Returns the concatenation of the distinct non-null input values, separated by `delimiter`.
+/// This requires Apache Spark 4.0.0 or later.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - delimiter: A delimiter to separate the values.
+/// - Returns: A ``Column``.
+public func listagg_distinct(_ col: Column, _ delimiter: String) -> Column {
+ return fn("listagg", [col, lit(delimiter)], isDistinct: true)
+}
+
/// Returns the value associated with the maximum value of `ord`.
/// - Parameters:
/// - col: A ``Column`` to return the value from.
@@ -239,6 +430,28 @@ public func mode(_ col: Column, _ deterministic: Bool) -> Column {
return fn("mode", col, lit(deterministic))
}
+/// Returns the exact `percentile` of the numeric column `col` at the given `percentage`.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - percentage: A percentage. Must be between 0.0 and 1.0.
+/// - frequency: A positive numeric literal for the number of times `col` should be counted
+/// (default = 1).
+/// - Returns: A ``Column``.
+public func percentile(_ col: Column, _ percentage: Double, _ frequency: Int32 = 1) -> Column {
+ return fn("percentile", col, lit(percentage), lit(frequency))
+}
+
+/// Returns the exact `percentile`s of the numeric column `col` at the given `percentage`s.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - percentage: An array of percentages. Each value must be between 0.0 and 1.0.
+/// - frequency: A positive numeric literal for the number of times `col` should be counted
+/// (default = 1).
+/// - Returns: A ``Column``.
+public func percentile(_ col: Column, _ percentage: [Double], _ frequency: Int32 = 1) -> Column {
+ return fn("percentile", col, fn("array", percentage.map { lit($0) }), lit(frequency))
+}
+
/// Returns the approximate `percentile` of the numeric column `col` which is the smallest value
/// in the ordered `col` values (sorted from least to greatest) such that no more than `percentage`
/// of `col` values is less than the value or equal to that value.
@@ -252,6 +465,13 @@ public func percentile_approx(_ col: Column, _ percentage: Column, _ accuracy: C
return fn("percentile_approx", col, percentage, accuracy)
}
+/// Returns the product of the values in a group.
+/// - Parameter col: A ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func product(_ col: Column) -> Column {
+ return fn("product", col)
+}
+
/// Returns the skewness of the values in a group.
/// - Parameter col: A ``Column`` to aggregate.
/// - Returns: A ``Column``.
@@ -259,6 +479,22 @@ public func skewness(_ col: Column) -> Column {
return fn("skewness", col)
}
+/// Returns true if at least one value of the column is true.
+/// This is an alias of ``bool_or(_:)``.
+/// - Parameter col: A ``Column`` that evaluates to a boolean.
+/// - Returns: A ``Column``.
+public func some(_ col: Column) -> Column {
+ return fn("some", col)
+}
+
+/// Returns the sample standard deviation of the expression in a group.
+/// This is an alias of ``stddev_samp(_:)``.
+/// - Parameter col: A ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func std(_ col: Column) -> Column {
+ return fn("std", col)
+}
+
/// Returns the sample standard deviation of the expression in a group.
/// This is an alias of ``stddev_samp(_:)``.
/// - Parameter col: A ``Column`` to aggregate.
@@ -281,6 +517,42 @@ public func stddev_samp(_ col: Column) -> Column {
return fn("stddev_samp", col)
}
+/// Returns the concatenation of the non-null input values.
+/// This is an alias of ``listagg(_:)`` and requires Apache Spark 4.0.0 or later.
+/// - Parameter col: A ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func string_agg(_ col: Column) -> Column {
+ return fn("string_agg", col)
+}
+
+/// Returns the concatenation of the non-null input values, separated by `delimiter`.
+/// This is an alias of ``listagg(_:_:)`` and requires Apache Spark 4.0.0 or later.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - delimiter: A delimiter to separate the values.
+/// - Returns: A ``Column``.
+public func string_agg(_ col: Column, _ delimiter: String) -> Column {
+ return fn("string_agg", col, lit(delimiter))
+}
+
+/// Returns the concatenation of the distinct non-null input values.
+/// This is an alias of ``listagg_distinct(_:)`` and requires Apache Spark 4.0.0 or later.
+/// - Parameter col: A ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func string_agg_distinct(_ col: Column) -> Column {
+ return fn("string_agg", [col], isDistinct: true)
+}
+
+/// Returns the concatenation of the distinct non-null input values, separated by `delimiter`.
+/// This is an alias of ``listagg_distinct(_:_:)`` and requires Apache Spark 4.0.0 or later.
+/// - Parameters:
+/// - col: A ``Column`` to aggregate.
+/// - delimiter: A delimiter to separate the values.
+/// - Returns: A ``Column``.
+public func string_agg_distinct(_ col: Column, _ delimiter: String) -> Column {
+ return fn("string_agg", [col, lit(delimiter)], isDistinct: true)
+}
+
/// Returns the sum of distinct values in the expression.
/// This is an alias of ``sum_distinct(_:)``.
/// - Parameter col: A ``Column`` to aggregate.
@@ -296,6 +568,20 @@ public func sum_distinct(_ col: Column) -> Column {
return fn("sum", [col], isDistinct: true)
}
+/// Returns the mean calculated from values of a group and `null` on overflow.
+/// - Parameter col: A ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func try_avg(_ col: Column) -> Column {
+ return fn("try_avg", col)
+}
+
+/// Returns the sum calculated from values of a group and `null` on overflow.
+/// - Parameter col: A ``Column`` to aggregate.
+/// - Returns: A ``Column``.
+public func try_sum(_ col: Column) -> Column {
+ return fn("try_sum", col)
+}
+
/// Returns the population variance of the values in a group.
/// - Parameter col: A ``Column`` to aggregate.
/// - Returns: A ``Column``.
diff --git a/Tests/SparkConnectTests/AggregateFunctionsTests.swift b/Tests/SparkConnectTests/AggregateFunctionsTests.swift
index 3ac536a..0cfb3ad 100644
--- a/Tests/SparkConnectTests/AggregateFunctionsTests.swift
+++ b/Tests/SparkConnectTests/AggregateFunctionsTests.swift
@@ -30,20 +30,35 @@ struct AggregateFunctionsTests {
for (column, name) in [
(any_value(col("a")), "any_value"),
(approx_count_distinct(col("a")), "approx_count_distinct"),
+ (array_agg(col("a")), "array_agg"),
+ (bit_and(col("a")), "bit_and"),
+ (bit_or(col("a")), "bit_or"),
+ (bit_xor(col("a")), "bit_xor"),
(bool_and(col("a")), "bool_and"),
(bool_or(col("a")), "bool_or"),
(collect_list(col("a")), "collect_list"),
(collect_set(col("a")), "collect_set"),
+ (collect_union(col("a")), "collect_union"),
(count_if(col("a")), "count_if"),
+ (every(col("a")), "every"),
+ (first_value(col("a")), "first_value"),
(grouping(col("a")), "grouping"),
(grouping_id(col("a")), "grouping_id"),
(kurtosis(col("a")), "kurtosis"),
+ (last_value(col("a")), "last_value"),
+ (listagg(col("a")), "listagg"),
(median(col("a")), "median"),
(mode(col("a")), "mode"),
+ (product(col("a")), "product"),
(skewness(col("a")), "skewness"),
+ (some(col("a")), "some"),
+ (std(col("a")), "std"),
(stddev(col("a")), "stddev"),
(stddev_pop(col("a")), "stddev_pop"),
(stddev_samp(col("a")), "stddev_samp"),
+ (string_agg(col("a")), "string_agg"),
+ (try_avg(col("a")), "try_avg"),
+ (try_sum(col("a")), "try_sum"),
(var_pop(col("a")), "var_pop"),
(var_samp(col("a")), "var_samp"),
(variance(col("a")), "variance"),
@@ -84,6 +99,66 @@ struct AggregateFunctionsTests {
#expect(percentile.unresolvedFunction.arguments.count == 3)
#expect(grouping_id(col("a"), col("b")).expr.unresolvedFunction.arguments.count == 2)
+
+ let bins = histogram_numeric(col("a"), 5).expr
+ #expect(bins.unresolvedFunction.functionName == "histogram_numeric")
+ #expect(bins.unresolvedFunction.arguments[1].literal.integer == 5)
+
+ let sketch = count_min_sketch(col("a"), 3.0, 0.1).expr
+ #expect(sketch.unresolvedFunction.functionName == "count_min_sketch")
+ #expect(sketch.unresolvedFunction.arguments.count == 4)
+ #expect(sketch.unresolvedFunction.arguments[1].literal.double == 3.0)
+ #expect(sketch.unresolvedFunction.arguments[2].literal.double == 0.1)
+
+ let seeded = count_min_sketch(col("a"), 3.0, 0.1, 1).expr
+ #expect(seeded.unresolvedFunction.functionName == "count_min_sketch")
+ #expect(seeded.unresolvedFunction.arguments.count == 4)
+ #expect(seeded.unresolvedFunction.arguments[3].literal.long == 1)
+
+ for (column, name, isDistinct) in [
+ (listagg(col("a"), ","), "listagg", false),
+ (listagg_distinct(col("a"), ","), "listagg", true),
+ (string_agg(col("a"), ","), "string_agg", false),
+ (string_agg_distinct(col("a"), ","), "string_agg", true),
+ ] {
+ let expr = column.expr
+ #expect(expr.unresolvedFunction.functionName == name)
+ #expect(expr.unresolvedFunction.isDistinct == isDistinct)
+ #expect(expr.unresolvedFunction.arguments.count == 2)
+ #expect(expr.unresolvedFunction.arguments[1].literal.string == ",")
+ }
+ }
+
+ @Test
+ func percentileFunctions() throws {
+ for (column, name) in [
+ (approx_percentile(col("a"), 0.5), "approx_percentile"),
+ (percentile(col("a"), 0.5), "percentile"),
+ ] {
+ let expr = column.expr
+ #expect(expr.unresolvedFunction.functionName == name)
+ #expect(expr.unresolvedFunction.arguments.count == 3)
+ #expect(expr.unresolvedFunction.arguments[0].unresolvedAttribute.unparsedIdentifier == "a")
+ #expect(expr.unresolvedFunction.arguments[1].literal.double == 0.5)
+ }
+
+ #expect(approx_percentile(col("a"), 0.5).expr.unresolvedFunction.arguments[2].literal.integer
+ == 10000)
+ #expect(approx_percentile(col("a"), 0.5, 100).expr.unresolvedFunction.arguments[2].literal
+ .integer == 100)
+ #expect(percentile(col("a"), 0.5).expr.unresolvedFunction.arguments[2].literal.integer == 1)
+ #expect(percentile(col("a"), 0.5, 2).expr.unresolvedFunction.arguments[2].literal.integer == 2)
+
+ for column in [
+ approx_percentile(col("a"), [0.25, 0.75]),
+ percentile(col("a"), [0.25, 0.75]),
+ ] {
+ let percentages = column.expr.unresolvedFunction.arguments[1].unresolvedFunction
+ #expect(percentages.functionName == "array")
+ #expect(percentages.arguments.count == 2)
+ #expect(percentages.arguments[0].literal.double == 0.25)
+ #expect(percentages.arguments[1].literal.double == 0.75)
+ }
}
@Test
@@ -93,6 +168,10 @@ struct AggregateFunctionsTests {
(first(col("a"), true), "first", true),
(last(col("a")), "last", false),
(last(col("a"), true), "last", true),
+ (first_value(col("a"), false), "first_value", false),
+ (first_value(col("a"), true), "first_value", true),
+ (last_value(col("a"), false), "last_value", false),
+ (last_value(col("a"), true), "last_value", true),
] {
let expr = column.expr
#expect(expr.unresolvedFunction.functionName == name)
@@ -111,6 +190,10 @@ struct AggregateFunctionsTests {
(count_distinct(col("a"), col("b")), "count", 2),
(sumDistinct(col("a")), "sum", 1),
(sum_distinct(col("a")), "sum", 1),
+ (listagg_distinct(col("a")), "listagg", 1),
+ (listagg_distinct(col("a"), ","), "listagg", 2),
+ (string_agg_distinct(col("a")), "string_agg", 1),
+ (string_agg_distinct(col("a"), ","), "string_agg", 2),
] {
let expr = column.expr
#expect(expr.unresolvedFunction.functionName == name)
@@ -196,6 +279,126 @@ struct AggregateFunctionsTests {
await spark.stop()
}
+ @Test
+ func selectAliasAggregateFunctions() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let df = try await spark.sql(
+ "SELECT * FROM VALUES (true, double(1.0)), (false, double(3.0)) T(b, v)")
+ let rows = try await df.select(
+ every(col("b")), some(col("b")), std(col("v")),
+ sort_array(array_agg(col("v"))).cast("string")
+ ).collect()
+ #expect(rows == [Row(false, true, 1.4142135623730951, "[1.0, 3.0]")])
+
+ let nulls = try await spark.sql("SELECT * FROM VALUES (NULL), (1), (2), (NULL) T(v)")
+ .select(
+ first_value(col("v")), first_value(col("v"), true),
+ last_value(col("v")), last_value(col("v"), true)
+ ).collect()
+ #expect(nulls == [Row(nil, 1, nil, 2)])
+ await spark.stop()
+ }
+
+ @Test
+ func selectBitAggregateFunctions() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let df = try await spark.sql("SELECT * FROM VALUES (6), (3), (NULL) T(v)")
+ let rows = try await df.select(
+ bit_and(col("v")), bit_or(col("v")), bit_xor(col("v"))
+ ).collect()
+ #expect(rows == [Row(2, 7, 5)])
+ await spark.stop()
+ }
+
+ @Test
+ func selectPercentileFunctions() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let df = try await spark.sql(
+ "SELECT * FROM VALUES (double(0.0)), (double(1.0)), (double(2.0)), (double(10.0)) T(v)")
+ let rows = try await df.select(
+ percentile(col("v"), 0.5), approx_percentile(col("v"), 0.5),
+ percentile(col("v"), [0.0, 1.0]).cast("string"),
+ approx_percentile(col("v"), [0.0, 1.0], 100).cast("string")
+ ).collect()
+ #expect(rows == [Row(1.5, 1.0, "[0.0, 10.0]", "[0.0, 10.0]")])
+
+ let weighted = try await spark.sql("SELECT * FROM VALUES (double(1.0)), (double(3.0)) T(v)")
+ .select(percentile(col("v"), 0.5, 2)).collect()
+ #expect(weighted == [Row(2.0)])
+ await spark.stop()
+ }
+
+ @Test
+ func selectTryAggregateFunctions() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let df = try await spark.sql("SELECT * FROM VALUES (1), (2) T(v)")
+ #expect(try await df.select(try_avg(col("v")), try_sum(col("v"))).collect() == [Row(1.5, 3)])
+
+ let overflow = try await spark.sql(
+ "SELECT * FROM VALUES (9223372036854775807L), (1L) T(v)")
+ .select(try_avg(col("v")), try_sum(col("v"))).collect()
+ #expect(overflow == [Row(4611686018427387904.0, nil)])
+ await spark.stop()
+ }
+
+ @Test
+ func selectMiscAggregateFunctions() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let df = try await spark.sql(
+ "SELECT * FROM VALUES (double(1.0)), (double(2.0)), (double(3.0)) T(v)")
+ #expect(try await df.select(product(col("v"))).collect() == [Row(6.0)])
+
+ let histogram = try await df.select(histogram_numeric(col("v"), 2).cast("string")).collect()
+ #expect(histogram == [Row("[{1.0, 1.0}, {2.5, 2.0}]")])
+
+ // `count_min_sketch` requires an integral, string or binary input column.
+ let sketch = try await spark.sql("SELECT * FROM VALUES (1), (2), (3) T(v)").select(
+ length(hex(count_min_sketch(col("v"), 3.0, 0.1))),
+ hex(count_min_sketch(col("v"), 3.0, 0.1, 1))
+ ).collect()
+ #expect(
+ sketch == [
+ Row(72, "0000000100000000000000030000000100000001000000005D8D6AB90000000000000003")
+ ])
+ await spark.stop()
+ }
+
+ @Test
+ func selectStringAggregateFunctions() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ let df = try await spark.sql("SELECT * FROM VALUES ('a'), ('b'), ('a') T(v)")
+ if await isSparkVersionAtLeast(spark.version, "4.0") {
+ // `listagg` is non-deterministic in the absence of an ordering, so the results are matched
+ // against patterns. The `_distinct` variants collapse the duplicated 'a' to a single one.
+ let rows = try await df.select(
+ listagg(col("v")).rlike("^[ab]{3}$"),
+ listagg(col("v"), ",").rlike("^[ab](,[ab]){2}$"),
+ listagg_distinct(col("v")).rlike("^ab|ba$"),
+ listagg_distinct(col("v"), ",").rlike("^[ab],[ab]$"),
+ string_agg(col("v")).rlike("^[ab]{3}$"),
+ string_agg(col("v"), ",").rlike("^[ab](,[ab]){2}$"),
+ string_agg_distinct(col("v")).rlike("^ab|ba$"),
+ string_agg_distinct(col("v"), ",").rlike("^[ab],[ab]$")
+ ).collect()
+ #expect(rows == [Row(true, true, true, true, true, true, true, true)])
+ }
+ await spark.stop()
+ }
+
+ @Test
+ func selectCollectUnion() async throws {
+ let spark = try await SparkSession.builder.getOrCreate()
+ if await isSparkVersionAtLeast(spark.version, "4.3") {
+ let df = try await spark.sql(
+ "SELECT * FROM VALUES (array(1, 2)), (array(2, 3)), (array(1)) T(v)")
+ let rows = try await df.select(
+ sort_array(collect_union(col("v"))).cast("string")
+ ).collect()
+ #expect(rows == [Row("[1, 2, 3]")])
+ }
+ await spark.stop()
+ }
+
@Test
func selectGroupingFunctions() async throws {
let spark = try await SparkSession.builder.getOrCreate()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org