zml1206 commented on code in PR #57815:
URL: https://github.com/apache/spark/pull/57815#discussion_r3763010971
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/SQLWindowFunctionSuite.scala:
##########
@@ -196,11 +197,380 @@ class SQLWindowFunctionSuite extends SharedSparkSession {
val e = intercept[AnalysisException] {
sql(
"""
- |select month, area, product, sum(distinct product + 1) over
(partition by 1 order by 2)
+ |select month, area, product, sum(distinct product + 1) over (
+ | partition by 1 order by 2 rows between current row and current
row)
|from windowData
""".stripMargin)
}
- assert(e.getMessage.contains("Distinct window functions are not
supported"))
+ assert(e.getMessage.contains("Unsupported DISTINCT window function"))
+ }
+ }
+
+ test("window function: distinct rejects unorderable inputs") {
+ val e = intercept[AnalysisException] {
+ sql("SELECT count(DISTINCT map('key', id)) OVER () FROM range(1)")
+ }
+ assert(e.getCondition === "DISTINCT_WINDOW_FUNCTION_UNSUPPORTED")
+ }
+
+ test("window function: distinct aggregates with an unbounded preceding
frame") {
+ val data = Seq(
+ (1, 0, 10, "a", 10),
+ (1, 1, 20, "a", 10),
+ (1, 2, 20, "b", 20),
+ (1, 3, 20, null.asInstanceOf[String], 30),
+ (1, 4, 30, "c", 30),
+ (2, 5, 5, "b", 5),
+ (2, 6, 5, "b", 5),
+ (2, 7, 6, "a", 6)
+ ).toDF("k", "id", "v", "x", "amount")
+
+ withTempView("distinctWindowData") {
+ data.createOrReplaceTempView("distinctWindowData")
+
+ checkAnswer(
+ sql(
+ """
+ |SELECT k, id,
+ | count(DISTINCT x) OVER (PARTITION BY k ORDER BY v) AS
range_count,
+ | count(DISTINCT x) OVER (
+ | PARTITION BY k ORDER BY v, id
+ | ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS
rows_count,
+ | count(DISTINCT x) OVER (
+ | PARTITION BY k ORDER BY v, id
+ | ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING) AS
preceding_count,
+ | count(DISTINCT x) OVER (
+ | PARTITION BY k ORDER BY v, id
+ | ROWS BETWEEN UNBOUNDED PRECEDING AND 1 FOLLOWING) AS
following_count,
+ | count(DISTINCT x) OVER (PARTITION BY k) AS partition_count,
+ | sum(DISTINCT amount) OVER (PARTITION BY k ORDER BY v) AS
range_sum,
+ | avg(DISTINCT amount) OVER (PARTITION BY k ORDER BY v) AS
range_avg,
+ | sort_array(collect_list(DISTINCT amount) OVER (
+ | PARTITION BY k ORDER BY v)) AS range_values
+ |FROM distinctWindowData
+ """.stripMargin),
+ Seq(
+ Row(1, 0, 1L, 1L, 0L, 1L, 3L, 10L, 10.0, Seq(10)),
+ Row(1, 1, 2L, 1L, 1L, 2L, 3L, 60L, 20.0, Seq(10, 20, 30)),
+ Row(1, 2, 2L, 2L, 1L, 2L, 3L, 60L, 20.0, Seq(10, 20, 30)),
+ Row(1, 3, 2L, 2L, 2L, 3L, 3L, 60L, 20.0, Seq(10, 20, 30)),
+ Row(1, 4, 3L, 3L, 2L, 3L, 3L, 60L, 20.0, Seq(10, 20, 30)),
+ Row(2, 5, 1L, 1L, 0L, 1L, 2L, 5L, 5.0, Seq(5)),
+ Row(2, 6, 1L, 1L, 1L, 2L, 2L, 5L, 5.0, Seq(5)),
+ Row(2, 7, 2L, 2L, 1L, 2L, 2L, 11L, 5.5, Seq(5, 6))
+ ))
+ }
+ }
+
+ test("window function: count distinct with a range offset, filter, and
multiple columns") {
Review Comment:
Added WINDOW_EXEC_BUFFER_IN_MEMORY_THRESHOLD = 1 to the existing
RANGE-offset/filter/multiple-column test. It now exercises the real non-empty
DISTINCT keys through an UnsafeExternalSorter-backed window input buffer while
the RANGE frame uses its concurrent iterators.
--
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]
