You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2023/07/19 01:13:31 UTC
[spark] branch branch-3.5 updated: [SPARK-44448][SQL] Fix wrong results bug from DenseRankLimitIterator and InferWindowGroupLimit
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-3.5
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-3.5 by this push:
new cf52568af1e [SPARK-44448][SQL] Fix wrong results bug from DenseRankLimitIterator and InferWindowGroupLimit
cf52568af1e is described below
commit cf52568af1ef0a17a68f2ea57fb35442547482b5
Author: Jack Chen <ja...@databricks.com>
AuthorDate: Wed Jul 19 09:12:54 2023 +0800
[SPARK-44448][SQL] Fix wrong results bug from DenseRankLimitIterator and InferWindowGroupLimit
### What changes were proposed in this pull request?
Top-k filters on a dense_rank() window function return wrong results, due to a bug in optimization InferWindowGroupLimit, specifically in the code for DenseRankLimitIterator, introduced in https://issues.apache.org/jira/browse/SPARK-37099.
The bug is in DenseRankLimitIterator, it fails to reset state properly when transitioning from one window partition to the next. reset only resets rank = 0, what it is missing is to reset currentRankRow = null. This means that when processing the second and later window partitions, the rank incorrectly gets incremented based on comparing the ordering of the last row of the previous partition to the first row of the new partition.
This means that a dense_rank window func that has more than one window partition and more than one row with dense_rank = 1 in the second or later partitions can give wrong results when optimized.
RankLimitIterator narrowly avoids this bug by happenstance, the first row in the new partition will try to increment rank, but increment it by the value of count which is 0, so it happens to work by accident. This PR also fixes the reset function in RankLimitIterator to make it more robust.
Example repro:
```
create or replace temp view t1 (p, o) as values (1, 1), (1, 1), (1, 2), (2, 1), (2, 1), (2, 2);
select * from (select *, dense_rank() over (partition by p order by o) as rnk from t1) where rnk = 1;
```
Spark result:
```
[1,1,1]
[1,1,1]
[2,1,1]
```
Correct result:
```
[1,1,1]
[1,1,1]
[2,1,1]
[2,1,1]
```
### Why are the changes needed?
Fix wrong results bug.
### Does this PR introduce _any_ user-facing change?
Yes, fixes wrong results.
### How was this patch tested?
Add sql tests and unit tests.
Unfortunately, the previous tests for the optimization only had a single row per rank, so did not catch the bug as the bug requires multiple rows per rank. This PR strengthens the tests with data that contains multiple rows per rank.
Closes #42026 from jchen5/dense-rank-limit.
Authored-by: Jack Chen <ja...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit f35c814f9b3cf73b94297f8086a117df4c46f39a)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../execution/window/WindowGroupLimitExec.scala | 2 +
.../sql-tests/analyzer-results/window.sql.out | 131 +++++++++++++++++++++
.../src/test/resources/sql-tests/inputs/window.sql | 13 ++
.../resources/sql-tests/results/window.sql.out | 90 ++++++++++++++
.../spark/sql/DataFrameWindowFunctionsSuite.scala | 52 ++++++--
5 files changed, 276 insertions(+), 12 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowGroupLimitExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowGroupLimitExec.scala
index b1f375f4151..98969f60c2b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowGroupLimitExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowGroupLimitExec.scala
@@ -170,6 +170,7 @@ case class RankLimitIterator(
override def reset(): Unit = {
rank = 0
count = 0
+ currentRankRow = null
}
}
@@ -193,6 +194,7 @@ case class DenseRankLimitIterator(
override def reset(): Unit = {
rank = 0
+ currentRankRow = null
}
}
diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/window.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/window.sql.out
index ad75b97ceb1..cc2cc8bfc0e 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/window.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/window.sql.out
@@ -1279,3 +1279,134 @@ org.apache.spark.sql.AnalysisException
"windowName" : "w"
}
}
+
+
+-- !query
+create or replace temp view t1 (p, o) as values (1, 1), (1, 1), (1, 2), (2, 1), (2, 1), (2, 2)
+-- !query analysis
+CreateViewCommand `t1`, [(p,None), (o,None)], values (1, 1), (1, 1), (1, 2), (2, 1), (2, 1), (2, 2), false, true, LocalTempView, true
+ +- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+select * from (select *, dense_rank() over (partition by p order by o) as rnk from t1) where rnk = 1
+-- !query analysis
+Project [p#x, o#x, rnk#x]
++- Filter (rnk#x = 1)
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [p#x, o#x, rnk#x]
+ +- Project [p#x, o#x, rnk#x, rnk#x]
+ +- Window [dense_rank(o#x) windowspecdefinition(p#x, o#x ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS rnk#x], [p#x], [o#x ASC NULLS FIRST]
+ +- Project [p#x, o#x]
+ +- SubqueryAlias t1
+ +- View (`t1`, [p#x,o#x])
+ +- Project [cast(col1#x as int) AS p#x, cast(col2#x as int) AS o#x]
+ +- LocalRelation [col1#x, col2#x]
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r = 1
+-- !query analysis
+Project [cate#x, val#x, r#x]
++- Filter (r#x = 1)
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [cate#x, val#x, r#x]
+ +- Project [cate#x, val#x, r#x, r#x]
+ +- Window [rank(val#x) windowspecdefinition(cate#x, val#x ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS r#x], [cate#x], [val#x ASC NULLS FIRST]
+ +- Project [cate#x, val#x]
+ +- SubqueryAlias testdata
+ +- View (`testData`, [val#x,val_long#xL,val_double#x,val_date#x,val_timestamp#x,cate#x])
+ +- Project [cast(val#x as int) AS val#x, cast(val_long#xL as bigint) AS val_long#xL, cast(val_double#x as double) AS val_double#x, cast(val_date#x as date) AS val_date#x, cast(val_timestamp#x as timestamp) AS val_timestamp#x, cast(cate#x as string) AS cate#x]
+ +- Project [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
+ +- SubqueryAlias testData
+ +- LocalRelation [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r <= 2
+-- !query analysis
+Project [cate#x, val#x, r#x]
++- Filter (r#x <= 2)
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [cate#x, val#x, r#x]
+ +- Project [cate#x, val#x, r#x, r#x]
+ +- Window [rank(val#x) windowspecdefinition(cate#x, val#x ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS r#x], [cate#x], [val#x ASC NULLS FIRST]
+ +- Project [cate#x, val#x]
+ +- SubqueryAlias testdata
+ +- View (`testData`, [val#x,val_long#xL,val_double#x,val_date#x,val_timestamp#x,cate#x])
+ +- Project [cast(val#x as int) AS val#x, cast(val_long#xL as bigint) AS val_long#xL, cast(val_double#x as double) AS val_double#x, cast(val_date#x as date) AS val_date#x, cast(val_timestamp#x as timestamp) AS val_timestamp#x, cast(cate#x as string) AS cate#x]
+ +- Project [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
+ +- SubqueryAlias testData
+ +- LocalRelation [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, dense_rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r = 1
+-- !query analysis
+Project [cate#x, val#x, r#x]
++- Filter (r#x = 1)
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [cate#x, val#x, r#x]
+ +- Project [cate#x, val#x, r#x, r#x]
+ +- Window [dense_rank(val#x) windowspecdefinition(cate#x, val#x ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS r#x], [cate#x], [val#x ASC NULLS FIRST]
+ +- Project [cate#x, val#x]
+ +- SubqueryAlias testdata
+ +- View (`testData`, [val#x,val_long#xL,val_double#x,val_date#x,val_timestamp#x,cate#x])
+ +- Project [cast(val#x as int) AS val#x, cast(val_long#xL as bigint) AS val_long#xL, cast(val_double#x as double) AS val_double#x, cast(val_date#x as date) AS val_date#x, cast(val_timestamp#x as timestamp) AS val_timestamp#x, cast(cate#x as string) AS cate#x]
+ +- Project [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
+ +- SubqueryAlias testData
+ +- LocalRelation [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, dense_rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r <= 2
+-- !query analysis
+Project [cate#x, val#x, r#x]
++- Filter (r#x <= 2)
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [cate#x, val#x, r#x]
+ +- Project [cate#x, val#x, r#x, r#x]
+ +- Window [dense_rank(val#x) windowspecdefinition(cate#x, val#x ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS r#x], [cate#x], [val#x ASC NULLS FIRST]
+ +- Project [cate#x, val#x]
+ +- SubqueryAlias testdata
+ +- View (`testData`, [val#x,val_long#xL,val_double#x,val_date#x,val_timestamp#x,cate#x])
+ +- Project [cast(val#x as int) AS val#x, cast(val_long#xL as bigint) AS val_long#xL, cast(val_double#x as double) AS val_double#x, cast(val_date#x as date) AS val_date#x, cast(val_timestamp#x as timestamp) AS val_timestamp#x, cast(cate#x as string) AS cate#x]
+ +- Project [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
+ +- SubqueryAlias testData
+ +- LocalRelation [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, row_number() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r = 1
+-- !query analysis
+Project [cate#x, val#x, r#x]
++- Filter (r#x = 1)
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [cate#x, val#x, r#x]
+ +- Project [cate#x, val#x, r#x, r#x]
+ +- Window [row_number() windowspecdefinition(cate#x, val#x ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS r#x], [cate#x], [val#x ASC NULLS FIRST]
+ +- Project [cate#x, val#x]
+ +- SubqueryAlias testdata
+ +- View (`testData`, [val#x,val_long#xL,val_double#x,val_date#x,val_timestamp#x,cate#x])
+ +- Project [cast(val#x as int) AS val#x, cast(val_long#xL as bigint) AS val_long#xL, cast(val_double#x as double) AS val_double#x, cast(val_date#x as date) AS val_date#x, cast(val_timestamp#x as timestamp) AS val_timestamp#x, cast(cate#x as string) AS cate#x]
+ +- Project [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
+ +- SubqueryAlias testData
+ +- LocalRelation [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, row_number() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r <= 2
+-- !query analysis
+Project [cate#x, val#x, r#x]
++- Filter (r#x <= 2)
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [cate#x, val#x, r#x]
+ +- Project [cate#x, val#x, r#x, r#x]
+ +- Window [row_number() windowspecdefinition(cate#x, val#x ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS r#x], [cate#x], [val#x ASC NULLS FIRST]
+ +- Project [cate#x, val#x]
+ +- SubqueryAlias testdata
+ +- View (`testData`, [val#x,val_long#xL,val_double#x,val_date#x,val_timestamp#x,cate#x])
+ +- Project [cast(val#x as int) AS val#x, cast(val_long#xL as bigint) AS val_long#xL, cast(val_double#x as double) AS val_double#x, cast(val_date#x as date) AS val_date#x, cast(val_timestamp#x as timestamp) AS val_timestamp#x, cast(cate#x as string) AS cate#x]
+ +- Project [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
+ +- SubqueryAlias testData
+ +- LocalRelation [val#x, val_long#xL, val_double#x, val_date#x, val_timestamp#x, cate#x]
diff --git a/sql/core/src/test/resources/sql-tests/inputs/window.sql b/sql/core/src/test/resources/sql-tests/inputs/window.sql
index 8f8963c5558..f94ff0f0a68 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/window.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/window.sql
@@ -2,6 +2,8 @@
--CONFIG_DIM1 spark.sql.codegen.wholeStage=true
--CONFIG_DIM1 spark.sql.codegen.wholeStage=false,spark.sql.codegen.factoryMode=CODEGEN_ONLY
--CONFIG_DIM1 spark.sql.codegen.wholeStage=false,spark.sql.codegen.factoryMode=NO_CODEGEN
+--CONFIG_DIM2 spark.sql.optimizer.windowGroupLimitThreshold=-1
+--CONFIG_DIM2 spark.sql.optimizer.windowGroupLimitThreshold=1000
-- Test data.
CREATE OR REPLACE TEMPORARY VIEW testData AS SELECT * FROM VALUES
@@ -465,3 +467,14 @@ SELECT
SUM(salary) OVER w sum_salary
FROM
basic_pays;
+
+-- Test cases for InferWindowGroupLimit
+create or replace temp view t1 (p, o) as values (1, 1), (1, 1), (1, 2), (2, 1), (2, 1), (2, 2);
+select * from (select *, dense_rank() over (partition by p order by o) as rnk from t1) where rnk = 1;
+
+SELECT * FROM (SELECT cate, val, rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r = 1;
+SELECT * FROM (SELECT cate, val, rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r <= 2;
+SELECT * FROM (SELECT cate, val, dense_rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r = 1;
+SELECT * FROM (SELECT cate, val, dense_rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r <= 2;
+SELECT * FROM (SELECT cate, val, row_number() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r = 1;
+SELECT * FROM (SELECT cate, val, row_number() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r <= 2;
diff --git a/sql/core/src/test/resources/sql-tests/results/window.sql.out b/sql/core/src/test/resources/sql-tests/results/window.sql.out
index 13e2f5209dd..7566f9eb20f 100644
--- a/sql/core/src/test/resources/sql-tests/results/window.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/window.sql.out
@@ -1342,3 +1342,93 @@ org.apache.spark.sql.AnalysisException
"windowName" : "w"
}
}
+
+
+-- !query
+create or replace temp view t1 (p, o) as values (1, 1), (1, 1), (1, 2), (2, 1), (2, 1), (2, 2)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+select * from (select *, dense_rank() over (partition by p order by o) as rnk from t1) where rnk = 1
+-- !query schema
+struct<p:int,o:int,rnk:int>
+-- !query output
+1 1 1
+1 1 1
+2 1 1
+2 1 1
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r = 1
+-- !query schema
+struct<cate:string,val:int,r:int>
+-- !query output
+NULL NULL 1
+a NULL 1
+b 1 1
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r <= 2
+-- !query schema
+struct<cate:string,val:int,r:int>
+-- !query output
+NULL 3 2
+NULL NULL 1
+a 1 2
+a 1 2
+a NULL 1
+b 1 1
+b 2 2
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, dense_rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r = 1
+-- !query schema
+struct<cate:string,val:int,r:int>
+-- !query output
+NULL NULL 1
+a NULL 1
+b 1 1
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, dense_rank() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r <= 2
+-- !query schema
+struct<cate:string,val:int,r:int>
+-- !query output
+NULL 3 2
+NULL NULL 1
+a 1 2
+a 1 2
+a NULL 1
+b 1 1
+b 2 2
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, row_number() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r = 1
+-- !query schema
+struct<cate:string,val:int,r:int>
+-- !query output
+NULL NULL 1
+a NULL 1
+b 1 1
+
+
+-- !query
+SELECT * FROM (SELECT cate, val, row_number() OVER(PARTITION BY cate ORDER BY val) as r FROM testData) where r <= 2
+-- !query schema
+struct<cate:string,val:int,r:int>
+-- !query output
+NULL 3 2
+NULL NULL 1
+a 1 2
+a NULL 1
+b 1 1
+b 2 2
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
index f2f645b126c..a57e927ba84 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameWindowFunctionsSuite.scala
@@ -1283,7 +1283,13 @@ class DataFrameWindowFunctionsSuite extends QueryTest
("b", 1, "n", Double.PositiveInfinity),
("c", 1, "z", -2.0),
("c", 1, "a", -4.0),
- ("c", 2, nullStr, 5.0)).toDF("key", "value", "order", "value2")
+ ("c", 2, nullStr, 5.0),
+ ("d", 0, "1", 1.0),
+ ("d", 1, "1", 2.0),
+ ("d", 2, "2", 3.0),
+ ("d", 3, "2", -1.0),
+ ("d", 4, "2", 2.0),
+ ("d", 4, "3", 2.0)).toDF("key", "value", "order", "value2")
val window = Window.partitionBy($"key").orderBy($"order".asc_nulls_first)
val window2 = Window.partitionBy($"key").orderBy($"order".desc_nulls_first)
@@ -1304,7 +1310,8 @@ class DataFrameWindowFunctionsSuite extends QueryTest
Seq(
Row("a", 4, "", 2.0, 1),
Row("b", 1, "h", Double.NaN, 1),
- Row("c", 2, null, 5.0, 1)
+ Row("c", 2, null, 5.0, 1),
+ Row("d", 0, "1", 1.0, 1)
)
)
@@ -1313,7 +1320,9 @@ class DataFrameWindowFunctionsSuite extends QueryTest
Row("a", 4, "", 2.0, 1),
Row("a", 4, "", 2.0, 1),
Row("b", 1, "h", Double.NaN, 1),
- Row("c", 2, null, 5.0, 1)
+ Row("c", 2, null, 5.0, 1),
+ Row("d", 0, "1", 1.0, 1),
+ Row("d", 1, "1", 2.0, 1)
)
)
@@ -1322,7 +1331,9 @@ class DataFrameWindowFunctionsSuite extends QueryTest
Row("a", 4, "", 2.0, 1),
Row("a", 4, "", 2.0, 1),
Row("b", 1, "h", Double.NaN, 1),
- Row("c", 2, null, 5.0, 1)
+ Row("c", 2, null, 5.0, 1),
+ Row("d", 0, "1", 1.0, 1),
+ Row("d", 1, "1", 2.0, 1)
)
)
@@ -1353,7 +1364,9 @@ class DataFrameWindowFunctionsSuite extends QueryTest
Row("b", 1, "h", Double.NaN, 1),
Row("b", 1, "n", Double.PositiveInfinity, 2),
Row("c", 1, "a", -4.0, 2),
- Row("c", 2, null, 5.0, 1)
+ Row("c", 2, null, 5.0, 1),
+ Row("d", 0, "1", 1.0, 1),
+ Row("d", 1, "1", 2.0, 2)
)
)
@@ -1364,7 +1377,9 @@ class DataFrameWindowFunctionsSuite extends QueryTest
Row("b", 1, "h", Double.NaN, 1),
Row("b", 1, "n", Double.PositiveInfinity, 2),
Row("c", 1, "a", -4.0, 2),
- Row("c", 2, null, 5.0, 1)
+ Row("c", 2, null, 5.0, 1),
+ Row("d", 0, "1", 1.0, 1),
+ Row("d", 1, "1", 2.0, 1)
)
)
@@ -1376,7 +1391,12 @@ class DataFrameWindowFunctionsSuite extends QueryTest
Row("b", 1, "h", Double.NaN, 1),
Row("b", 1, "n", Double.PositiveInfinity, 2),
Row("c", 1, "a", -4.0, 2),
- Row("c", 2, null, 5.0, 1)
+ Row("c", 2, null, 5.0, 1),
+ Row("d", 0, "1", 1.0, 1),
+ Row("d", 1, "1", 2.0, 1),
+ Row("d", 2, "2", 3.0, 2),
+ Row("d", 3, "2", -1.0, 2),
+ Row("d", 4, "2", 2.0, 2)
)
)
@@ -1408,7 +1428,8 @@ class DataFrameWindowFunctionsSuite extends QueryTest
checkAnswer(df.withColumn("rn", row_number().over(window)).where(condition),
Seq(
Row("a", 4, "", 2.0, 2),
- Row("b", 1, "n", Double.PositiveInfinity, 2)
+ Row("b", 1, "n", Double.PositiveInfinity, 2),
+ Row("d", 1, "1", 2.0, 2)
)
)
@@ -1421,7 +1442,9 @@ class DataFrameWindowFunctionsSuite extends QueryTest
checkAnswer(df.withColumn("rn", dense_rank().over(window)).where(condition),
Seq(
Row("a", 0, "c", 1.0, 2),
- Row("b", 1, "n", Double.PositiveInfinity, 2)
+ Row("b", 1, "n", Double.PositiveInfinity, 2),
+ Row("d", 2, "2", 3.0, 2),
+ Row("d", 4, "2", 2.0, 2)
)
)
@@ -1433,7 +1456,8 @@ class DataFrameWindowFunctionsSuite extends QueryTest
Seq(
Row("a", 4, "", 2.0, 1, 1),
Row("b", 1, "h", Double.NaN, 1, 1),
- Row("c", 2, null, 5.0, 1, 1)
+ Row("c", 2, null, 5.0, 1, 1),
+ Row("d", 0, "1", 1.0, 1, 1)
)
)
@@ -1446,7 +1470,9 @@ class DataFrameWindowFunctionsSuite extends QueryTest
Row("a", 4, "", 2.0, 1, 1),
Row("a", 4, "", 2.0, 1, 1),
Row("b", 1, "h", Double.NaN, 1, 1),
- Row("c", 2, null, 5.0, 1, 1)
+ Row("c", 2, null, 5.0, 1, 1),
+ Row("d", 0, "1", 1.0, 1, 1),
+ Row("d", 1, "1", 2.0, 1, 1)
)
)
@@ -1459,7 +1485,9 @@ class DataFrameWindowFunctionsSuite extends QueryTest
Row("a", 4, "", 2.0, 1, 1),
Row("a", 4, "", 2.0, 1, 1),
Row("b", 1, "h", Double.NaN, 1, 1),
- Row("c", 2, null, 5.0, 1, 1)
+ Row("c", 2, null, 5.0, 1, 1),
+ Row("d", 0, "1", 1.0, 1, 1),
+ Row("d", 1, "1", 2.0, 1, 1)
)
)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org