[SPARK-31561][SQL] Add QUALIFY clause #39691

wangyum · 2023-01-21T16:14:13Z

What changes were proposed in this pull request?

The QUALIFY clause is used to filter the results of window functions. For example:

-- QUALIFY with window functions in the SELECT list.
SELECT
  city,
  car_model,
  RANK() OVER (PARTITION BY car_model ORDER BY quantity) AS rank
FROM dealer
QUALIFY rank = 1;

-- QUALIFY with window functions in the QUALIFY clause.
SELECT city, car_model
FROM dealer
QUALIFY RANK() OVER (PARTITION BY car_model ORDER BY quantity) = 1;

Why are the changes needed?

To easily migrate SQL queries to Spark SQL.

The following databases support this clause:
https://docs.databricks.com/sql/language-manual/sql-ref-syntax-qry-select-qualify.html
https://docs.snowflake.com/en/sql-reference/constructs/qualify.html
https://cloud.google.com/bigquery/docs/reference/standard-sql/query-syntax#qualify_clause
https://duckdb.org/docs/sql/query_syntax/qualify.html
https://docs.teradata.com/reader/2_MC9vCtAJRlKle2Rpb0mA/19NnI91neorAi7LX6SJXBw

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Unit test.

wangyum · 2023-01-21T16:15:46Z

In fact databricks also supports this clause.

wangyum · 2023-01-21T16:21:10Z

cc @xinrong-meng @MaxGekk @gengliangwang @cloud-fan

docs/sql-ref-syntax-qry-select-qualify.md

sql/core/src/test/resources/sql-tests/results/window.sql.out

gatorsmile · 2023-02-22T08:47:24Z

cc @allisonwang-db

wangyum · 2023-02-24T09:28:39Z

CALCITE added support for it too: https://issues.apache.org/jira/browse/CALCITE-5268.

allisonwang-db · 2023-03-08T09:40:54Z

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

@@ -1729,6 +1729,23 @@ class Analyzer(override val catalogManager: CatalogManager)
          resolveExpressionByPlanChildren(resolvedWithAgg, u, allowOuter = true)
        }

+      case u @ UnresolvedQualify(cond, child) if !u.resolved && child.resolved =>
+        if (!u.containsPattern(WINDOW_EXPRESSION)) {


What if we have a subquery that contains a window function but not the main query:

SELECT (SELECT window_func w FROM t QUALIFY t.c1 > 0) FROM t QUALIFY t.c1 > 0

CREATE TABLE dealer (id INT, city STRING, car_model STRING, quantity INT) using parquet; SELECT * from (SELECT *, ROW_NUMBER() OVER(PARTITION BY city ORDER BY id) AS rn FROM dealer QUALIFY id > 0) t QUALIFY t.id > 0;

The analyzed logical plan:

== Analyzed Logical Plan == id: int, city: string, car_model: string, quantity: int, rn: int Filter (id#2 > 0) +- Project [id#2, city#3, car_model#4, quantity#5, rn#0] +- SubqueryAlias t +- Filter (id#2 > 0) +- Project [id#2, city#3, car_model#4, quantity#5, rn#0] +- Project [id#2, city#3, car_model#4, quantity#5, rn#0, rn#0] +- Window [row_number() windowspecdefinition(city#3, id#2 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS rn#0], [city#3], [id#2 ASC NULLS FIRST] +- Project [id#2, city#3, car_model#4, quantity#5] +- SubqueryAlias spark_catalog.default.dealer +- Relation spark_catalog.default.dealer[id#2,city#3,car_model#4,quantity#5] parquet

allisonwang-db · 2023-03-08T09:46:37Z

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

+          throw QueryCompilationErrors.expressionWithoutWindowExpressionError(cond)
+        } else {
+          if (u.missingInput.nonEmpty) {
+            val (newCond, newChild) = resolveExprsAndAddMissingAttrs(Seq(cond), child)


If we treat the resolution order for QUALIFY the same as GROUP BY, then we need to resolve the attributes in the QUALIFY clause first using the columns in the FROM clause and then using columns and aliases in the SELECT clause.

For example: SELECT c1, window_func AS c2 FROM t QUALIFY c2 > 0; here c2 should be resolved as t.c2 instead of the window_func as c2.

Behavior looks consistent with aggregate:

CREATE TABLE dealer (id INT, city STRING, car_model STRING, quantity INT) using parquet; SELECT ROW_NUMBER() OVER(PARTITION BY city ORDER BY id) AS id FROM dealer QUALIFY id > 0; SELECT city, max(id) AS id FROM dealer GROUP BY city HAVING id > 0;

== Analyzed Logical Plan == id: int Filter (id#0 > 0) +- Project [id#0] +- Project [city#3, id#2, id#0, id#0] +- Window [row_number() windowspecdefinition(city#3, id#2 ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS id#0], [city#3], [id#2 ASC NULLS FIRST] +- Project [city#3, id#2] +- SubqueryAlias spark_catalog.default.dealer +- Relation spark_catalog.default.dealer[id#2,city#3,car_model#4,quantity#5] parquet

== Analyzed Logical Plan == city: string, id: int Filter (id#0 > 0) +- Aggregate [city#2], [city#2, max(id#1) AS id#0] +- SubqueryAlias spark_catalog.default.dealer +- Relation spark_catalog.default.dealer[id#1,city#2,car_model#3,quantity#4] parquet

allisonwang-db · 2023-03-08T09:54:42Z

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -571,7 +571,7 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase {
      messageParameters = Map.empty)
  }

-  def expressionWithoutWindowExpressionError(expr: NamedExpression): Throwable = {
+  def expressionWithoutWindowExpressionError(expr: Expression): Throwable = {


Can we use a new error for this? Currnetly the error message is not very user-friendly. E.g.:

SELECT * FROM t QUALIFY c1 > 0 ('c1 > 0) does not have any WindowExpression.

The new message:

[QUALIFY_EXPRESSION_MUST_CONTAIN_WINDOW_FUNCTION] QUALIFY expression '(id > 1)' must contain a window function.; line 1 pos 21

allisonwang-db · 2023-03-08T10:01:00Z

sql/core/src/test/resources/sql-tests/inputs/window.sql

@@ -465,3 +465,60 @@ SELECT
 	SUM(salary) OVER w sum_salary
 FROM
 	basic_pays;
+
+-- Test QUALIFY clause


We can split these tests into a new file.

gatorsmile · 2023-03-20T06:45:43Z

@cloud-fan @allisonwang-db take another look?

jaceklaskowski · 2023-03-25T15:03:18Z

core/src/main/resources/error/error-classes.json

@@ -1303,6 +1303,11 @@
      "Protobuf type not yet supported: <protobufType>."
    ]
  },
+  "QUALIFY_EXPRESSION_MUST_CONTAIN_WINDOW_FUNCTION" : {
+    "message" : [
+      "QUALIFY expression '<sqlExpr>' must contain a window function."


nit: QUALIFY is a clause so what this expression refers to?

jaceklaskowski · 2023-03-25T15:07:41Z

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

@@ -693,6 +693,19 @@ case class UnresolvedHaving(
    copy(child = newChild)
 }

+/**
+ * Represents unresolved qualify clause, It is turned by the analyzer into a Filter.


nit: QUALIFY (all uppercase) + "that becomes Filter after query analysis"

jaceklaskowski · 2023-03-25T15:08:34Z

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

+    qualifyCondition: Expression,
+    child: LogicalPlan)
+  extends UnaryNode {
+  override lazy val resolved: Boolean = false


lazy is not needed here, is it?

felicienveldema · 2023-05-26T13:16:24Z

Is this feature still on the planning?

jimmyzzxhlh · 2023-06-08T23:01:40Z

^ Same question -- Any plan to release this feature?

github-actions · 2023-09-17T00:18:48Z

We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!

github-actions bot added DOCS SQL labels Jan 21, 2023

itholic reviewed Jan 22, 2023

View reviewed changes

docs/sql-ref-syntax-qry-select-qualify.md Outdated Show resolved Hide resolved

sql/core/src/test/resources/sql-tests/results/window.sql.out Outdated Show resolved Hide resolved

allisonwang-db reviewed Mar 8, 2023

View reviewed changes

wangyum added 3 commits March 9, 2023 10:08

Add QUALIFY clause

7f874f8

address comment

1b63f9b

fix

d2e3e5d

wangyum force-pushed the SPARK-31561 branch from 551adc6 to d2e3e5d Compare March 9, 2023 14:47

github-actions bot added the CORE label Mar 9, 2023

jaceklaskowski reviewed Mar 25, 2023

View reviewed changes

github-actions bot added the Stale label Sep 17, 2023

github-actions bot closed this Sep 18, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-31561][SQL] Add QUALIFY clause #39691

[SPARK-31561][SQL] Add QUALIFY clause #39691

wangyum commented Jan 21, 2023 •

edited

wangyum commented Jan 21, 2023

wangyum commented Jan 21, 2023

gatorsmile commented Feb 22, 2023

wangyum commented Feb 24, 2023

allisonwang-db Mar 8, 2023

wangyum Mar 9, 2023

allisonwang-db Mar 8, 2023

wangyum Mar 9, 2023

allisonwang-db Mar 8, 2023

wangyum Mar 9, 2023

allisonwang-db Mar 8, 2023

wangyum Mar 9, 2023

gatorsmile commented Mar 20, 2023

jaceklaskowski Mar 25, 2023

jaceklaskowski Mar 25, 2023

jaceklaskowski Mar 25, 2023

felicienveldema commented May 26, 2023

jimmyzzxhlh commented Jun 8, 2023

github-actions bot commented Sep 17, 2023

[SPARK-31561][SQL] Add QUALIFY clause #39691

[SPARK-31561][SQL] Add QUALIFY clause #39691

Conversation

wangyum commented Jan 21, 2023 • edited

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

wangyum commented Jan 21, 2023

wangyum commented Jan 21, 2023

gatorsmile commented Feb 22, 2023

wangyum commented Feb 24, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gatorsmile commented Mar 20, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

felicienveldema commented May 26, 2023

jimmyzzxhlh commented Jun 8, 2023

github-actions bot commented Sep 17, 2023

wangyum commented Jan 21, 2023 •

edited