[SPARK-49557][SQL] Add SQL pipe syntax for the WHERE operator

[dtenedor]

What changes were proposed in this pull request?

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

For example:

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

CREATE TABLE other(a INT, b INT) USING JSON;
INSERT INTO other VALUES (1, 1), (1, 2), (2, 4);

TABLE t
|> WHERE x + LENGTH(y) < 4;

0	abc

TABLE t
|> WHERE (SELECT ANY_VALUE(a) FROM other WHERE x = a LIMIT 1) = 1

1       def

TABLE t
|> WHERE SUM(x) = 1

Error: aggregate functions are not allowed in the pipe operator |> WHERE clause

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]

cc @cloud-fan @gengliangwang here is the WHERE operator, the next one. The implementation is relatively simple, I tried to think of as many test cases as possible.

[cloud-fan]

This is great! This skips the tricky aggregate function pushdown stuff from Filter/Sort which complicates the analyzer quite a bit. We also don't need this with pipe syntax, as it's quite easy for users to filter on the aggregated query.

[cloud-fan]

that being said, seems like we don't need to add subquery alias if the child plan is UnresolvedRelation. We don't need to isolate the table scan node here.

[dtenedor]

We talked offline and found that updating the UnresolvedRelation pattern match to this fixes the problem:

        case u: UnresolvedRelation =>
          u

In this way we don't add another redundant SubqueryAlias when ResolveRelations will already add one. Looking at the commit that performs this update, we see the analyzer plans improve accordingly.

[cloud-fan]

ah, it also fixes a regression. We can add a test for table t |> where spark_catalog.default.t.x = 1, which didn't work before this fix.

[dtenedor]

Sounds good, this is done.

[gengliangwang]

Thanks, merging to master

[allisonwang-db]

Awesome feature!

[allisonwang-db]

it also supports correlated subqueries!