[SPARK-49563][SQL] Add SQL pipe syntax for the WINDOW operator

[Angryrou]

What changes were proposed in this pull request?

This PR adds SQL pipe syntax support for the WINDOW clause within the pipe SELECT operator |> SELECT.

For example

CREATE TEMPORARY VIEW t 
AS SELECT * from VALUES
  (1, 'apple', 1),
  (2, 'banana', 2),
  (3, 'apple', 3),
  (4, 'banana', 4),
AS t(id, name, amount);

TABLE t
|> SELECT id, name, amount, SUM(amount) OVER w
   WINDOW w AS (PARTITION BY name ORDER BY id);

1, apple, 1, 3
3, apple, 2, 3
2, banana, 3, 7
4, banana, 4, 7

Notes:

1. |> WHERE is not extended to use the WINDOW clause because |> WHERE does not support window functions in its expressions.

Why are the changes needed?

The SQL pipe operator syntax will let users compose queries in a more flexible fashion.

Does this PR introduce any user-facing change?

Yes, see above.

How was this patch tested?

Yes

Was this patch authored or co-authored using generative AI tooling?

No

[dtenedor]

cc @cloud-fan @gengliangwang :)

[dtenedor]

Thanks for working on this!! I tried to think of testing ideas.

[Angryrou]

Thank you @dtenedor for the consideration suggestions! I have updated the code accordingly and added more test cases.

[Angryrou]

kindly ping @dtenedor

[dtenedor]

The extra test coverage helps a lot, thanks for adding it!
(Apologies for delay in review, will try to respond faster next time.)

[dtenedor]

Thanks, this approach looks right! Just a final suggestion for a better error message in one location and an update to the SQL golden file tests to make them deterministic, then this PR should be complete.

[Angryrou]

Thanks again for the review! @dtenedor

Please feel free to let me know any other comments!

[cloud-fan]

I'm a bit confused, aren't w1 and w2 both defined?

[Angryrou]

Similar to the classic SQL, when multiple WINDOW clauses are presented, the SQL parser will only see the first window definition, and hence miss the second window definition w2.

[cloud-fan]

hmm, is it a bug? @srielau

[dtenedor]

If it is, it seems like the same bug for both regular SQL syntax and pipe operators, and the behavior would work the same for both (now, and if we were to change it later).

[cloud-fan]

Is this an expected behavior in the SQL pipe syntax spec? cc @dtenedor @srielau

[dtenedor]

@Angryrou the query in this test case should fail now, since the window w is not yet defined in the |> select operator. Do we need to re-generate the golden files?

[Angryrou]

@cloud-fan @dtenedor

Fixed! I added logic to ensure that each window name used in UnresolvedWindowExpression is properly defined within the same |> SELECT clause.

In the previous logic, UnresolvedWindowExpression was analyzed only after scanning all pipe operators. As a result, in this test case, the window w in the first |> SELECT clause could incorrectly reference the definition in the second |> SELECT.

With this change, an error will be raised with INVALID_SQL_SYNTAX.WINDOW_REFERENCE_NOT_FOUND if a window name is used but no window clause exists, or INVALID_SQL_SYNTAX.UNRESOLVED_WINDOW_REFERENCE if a window name in UnresolvedWindowExpression is not defined within the same |> SELECT clause.

[dtenedor]

Generally LGTM, just have a question about one remaining test case.

[dtenedor]

@Angryrou the query in this test case should fail now, since the window w is not yet defined in the |> select operator. Do we need to re-generate the golden files?

[dtenedor]

If it is, it seems like the same bug for both regular SQL syntax and pipe operators, and the behavior would work the same for both (now, and if we were to change it later).

[cloud-fan]

It's better to not do analysis work in the parser, as we may miss something when re-implementing the analysis work. E.g. shall we consider case sensitivity here?

[cloud-fan]

@Angryrou can you highlight the logical plan differences between classic and pipe SQL syntaxes? I'd like to understand why we can reference previously defined window frame in the pipe syntax and see how we can fix it in the analyzer.

[Angryrou]

Differences @cloud-fan

1. with the pipe syntax, the unresolvedwindowexpression is wrapped by a pipeselect expression
2. the classic syntax has an auto-generated subquery name for the subquery

In the pipe syntax, the WINDOW clause in the 2nd |> SELECT leads to a WithWindowDefinition for w at the top of the plan tree. The 1st |> SELECT seems to be able to get the information as well.

In the classic syntax, the outer query defines w and gets a WithWindowDefinition at the top of the tree as well. So the inner subquery is able to get the window definition.

SQLs

-- pipe syntax
table windowTestData
|> select cate, val, first_value(cate) over w as first_val
|> select cate, val, sum(val) over w as sum_val
   window w as (order by val);

-- classic syntax
select cate, val, sum(val) over w as sum_val
from (
  select cate, val, first_value(cate) over w as first_val
  from windowTestData
)
window w as (order by val);

Parsed Logical Plans (parser.parsePlan(...))

-- from pipe syntax
'WithWindowDefinition [w=windowspecdefinition('val ASC NULLS FIRST, unspecifiedframe$())]
+- 'Project ['cate, 'val, pipeselect(unresolvedwindowexpression('sum('val), WindowSpecReference(w))) AS sum_val#3]
   +- 'Project ['cate, 'val, pipeselect(unresolvedwindowexpression('first_value('cate), WindowSpecReference(w))) AS first_val#2]
      +- 'UnresolvedRelation [windowTestData], [], false

-- from classic syntax
'WithWindowDefinition [w=windowspecdefinition('val ASC NULLS FIRST, unspecifiedframe$())]
+- 'Project ['cate, 'val, unresolvedwindowexpression('sum('val), WindowSpecReference(w)) AS sum_val#1]
   +- 'SubqueryAlias __auto_generated_subquery_name
      +- 'Project ['cate, 'val, unresolvedwindowexpression('first_value('cate), WindowSpecReference(w)) AS first_val#0]
         +- 'UnresolvedRelation [windowTestData], [], false

Analyzed Logical Plans (from the golden file)
https://docs.google.com/document/d/1qa1jUAoWa0aS5037oazydJRYQSAFFklUlPmnwLxOZDM/edit?usp=sharing

[cloud-fan]

@Angryrou thanks for the detailed explanation!

Looking at the analyzer rule WindowsSubstitution, we actually allow all the child nodes to reference the window frame definitions inside WithWindowDefinition. This makes sense for classic SQL, as the WINDOW clause is for the entire query, but it's pretty weird for the pipe SQL.

It looks like we want a variant of WithWindowDefinition that only works for its direct child node. How about we add a flag forPipeSQL in WithWindowDefinition, and only resolve the window frames in the direct child.

  object WindowsSubstitution extends Rule[LogicalPlan] {
    def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
      _.containsAnyPattern(WITH_WINDOW_DEFINITION, UNRESOLVED_WINDOW_EXPRESSION), ruleId) {
      // Lookup WindowSpecDefinitions. This rule works with unresolved children.
      case WithWindowDefinition(windowDefinitions, child) => child.resolveExpressions {
        case UnresolvedWindowExpression(c, WindowSpecReference(windowName)) =>
          val windowSpecDefinition = windowDefinitions.getOrElse(windowName,
            throw QueryCompilationErrors.windowSpecificationNotDefinedError(windowName))
          WindowExpression(c, windowSpecDefinition)
      }
    }
  }

child.resolveExpressions actually traverses down the entire plan tree, and we should call child.transformExpressions to only resolve expressions in the current node, when the forPipeSQL flag is true.

[Angryrou]

@cloud-fan Thanks for the guidance! I have changed it accordingly.

[dtenedor]

This solution generally LGTM, if Wenchen agrees :)

[cloud-fan]

thanks, merging to master!