[SPARK-49566][SQL] Add SQL pipe syntax for the EXTEND operator

[dtenedor]

What changes were proposed in this pull request?

This PR adds SQL pipe syntax support for the EXTEND operator.

This operator preserves the existing input table and adds one or more new computed columns whose values are equal to evaluating the specified expressions. This is equivalent to SELECT *, <newExpressions> in the SQL compiler. It is provided as a convenience feature and some functionality overlap exists with lateral column aliases.

For example:

CREATE TABLE t(x INT, y STRING) USING CSV;
INSERT INTO t VALUES (0, 'abc'), (1, 'def');

TABLE t
|> EXTEND x + LENGTH(y) AS z;

+----+-----+-----+
|  x |   y |   z |
+----+-----+-----+
|  0 | abc |   3 |
|  1 | def |   4 |
+----+-----+-----+

Like the |> SELECT operator, aggregate functions are not allowed in these expressions. During the course of developing reasonable error messages for this, I found that the SQL pipe syntax research paper also specified that the |> AGGREGATE operator should require that each non-grouping expression contains at least one aggregate function; I added a check and reasonable error message for this case as well.

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?

This PR adds a few unit test cases, but mostly relies on golden file test coverage. I did this to make sure the answers are correct as this feature is implemented and also so we can look at the analyzer output plans to ensure they look right as well.

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

No

[dtenedor]

The implementation for this PR ended up getting too big. I can split this into separate PRs.

[dtenedor]

This PR now just covers the EXTEND operator.

[dtenedor]

cc @gengliangwang @cloud-fan this one is ready for review at your convenience.

[cloud-fan]

nit:

val hasAggFunc = this.exists(_.isInstanceOf[AggregateFunction])
if (isAggregate && !hasAggFunc) error ...
else if (!isAggregate && hasAggFunc) error...
...

[dtenedor]

Thanks, this is simpler. (I had to keep the recursive method in order to stop traversing through window expressions, but this logic is better.)

[cloud-fan]

unrelated to this PR, but do we support SELECT * in pipe?

[dtenedor]

👍 yes we do! https://github.com/apache/spark/blob/2e1c3dc8004b4f003cde8dfae6857f5bef4bb170/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out#L404C1-L404C12

[dtenedor]

Thanks @cloud-fan for your reviews!!

[dtenedor]

👍 yes we do! https://github.com/apache/spark/blob/2e1c3dc8004b4f003cde8dfae6857f5bef4bb170/sql/core/src/test/resources/sql-tests/results/pipe-operators.sql.out#L404C1-L404C12

[dtenedor]

Thanks, this is simpler. (I had to keep the recursive method in order to stop traversing through window expressions, but this logic is better.)

[cloud-fan]

thanks, merging to master!