Repository: tobymao/sqlglot
Pull request: feat(optimizer): Support for UNION BY NAME
Fixes #3222 

Introduce support for `UNION [ALL] BY NAME` in `pushdown_projections` rule, which matches referenced columns by name instead of by position:

```
>>> optimize(sqlglot.parse_one("select a from (select 1 a, 2 c union all by name select 3 c, 4 a)")).sql()
'WITH "_q_0" AS (SELECT 1 AS "a" UNION ALL BY NAME SELECT 4 AS "a") SELECT "_q_0"."a" AS "a" FROM "_q_0" AS "_q_0"'
```

```
-- UNION (by position)
D select a from (select 1 a, 2 c union all select 3 c, 4 a);
┌───────┐
│   a   │
│ int32 │
├───────┤
│     1 │
│     3 │
└───────┘


-- UNION BY NAME
D select a from (select 1 a, 2 c union all by name select 3 c, 4 a);
┌───────┐
│   a   │
│ int32 │
├───────┤
│     1 │
│     4 │
└───────┘


-- Optimized UNION BY NAME
D WITH "_q_0" AS (SELECT 1 AS "a" UNION ALL BY NAME SELECT 4 AS "a") SELECT "_q_0"."a" AS "a" FROM "_q_0" AS "_q_0";
┌───────┐
│   a   │
│ int32 │
├───────┤
│     1 │
│     4 │
└───────┘
```

Docs
----------
- [DuckDB UNION BY NAME](https://duckdb.org/docs/sql/query_syntax/setops.html#union-all-by-name)
- [StackOverflow: UNION ALL ignoring column names](https://stackoverflow.com/questions/53076981/sql-union-doesnt-check-for-matching-column-names-is-it-up-to-me)

Problem statement:
Incorrect optimized SQL.
```
import sqlglot
from sqlglot.optimizer import optimize
optimize(sqlglot.parse_one("select a from (select 1 a, 2 c union all by name select 3 c, 4 a)", dialect="duckdb")).sql(dialect="duckdb")
```
```
WITH "_q_0" AS (SELECT 1 AS "a" UNION ALL BY NAME SELECT 3 AS "c") SELECT "_q_0"."a" AS "a" FROM "_q_0" AS "_q_0"
```
The result of optimized SQL query is different from original SQL query.
```
select a from (select 1 a, 2 c union all by name select 3 c, 4 a);
┌───────┐
│   a   │
│ int32 │
├───────┤
│     1 │
│     4 │
└───────┘
-- optimized
WITH "_q_0" AS (SELECT 1 AS "a" UNION ALL BY NAME SELECT 3 AS "c") SELECT "_q_0"."a" AS "a" FROM "_q_0" AS "_q_0";
┌───────┐
│   a   │
│ int32 │
├───────┤
│     1 │
│       │
└───────┘
```

Hints from issue discussion:

Commit message: feat(optimizer): Support for UNION BY NAME

Patch to review:
diff --git a/sqlglot/optimizer/pushdown_projections.py b/sqlglot/optimizer/pushdown_projections.py
index 53490bfce3..dac7314d30 100644
--- a/sqlglot/optimizer/pushdown_projections.py
+++ b/sqlglot/optimizer/pushdown_projections.py
@@ -54,11 +54,17 @@ def pushdown_projections(expression, schema=None, remove_unused_selections=True)
             if any(select.is_star for select in right.expression.selects):
                 referenced_columns[right] = parent_selections
             elif not any(select.is_star for select in left.expression.selects):
-                referenced_columns[right] = [
-                    right.expression.selects[i].alias_or_name
-                    for i, select in enumerate(left.expression.selects)
-                    if SELECT_ALL in parent_selections or select.alias_or_name in parent_selections
-                ]
+                union_by_name = scope.expression.args.get("by_name")
+
+                if union_by_name:
+                    referenced_columns[right] = referenced_columns[left]
+                else:
+                    referenced_columns[right] = [
+                        right.expression.selects[i].alias_or_name
+                        for i, select in enumerate(left.expression.selects)
+                        if SELECT_ALL in parent_selections
+                        or select.alias_or_name in parent_selections
+                    ]
 
         if isinstance(scope.expression, exp.Select):
             if remove_unused_selections:
diff --git a/tests/fixtures/optimizer/pushdown_projections.sql b/tests/fixtures/optimizer/pushdown_projections.sql
index b7103ef914..32f5f97e7d 100644
--- a/tests/fixtures/optimizer/pushdown_projections.sql
+++ b/tests/fixtures/optimizer/pushdown_projections.sql
@@ -106,3 +106,6 @@ WITH cte1 AS (SELECT tb.cola AS cola FROM tb AS tb UNION ALL SELECT tb2.colc AS
 
 SELECT * FROM ((SELECT c FROM t1) JOIN t2);
 SELECT * FROM ((SELECT t1.c AS c FROM t1 AS t1) AS _q_0, t2 AS t2);
+
+SELECT a, d FROM (SELECT 1 a, 2 c, 3 d, 4 e UNION ALL BY NAME SELECT 5 b, 6 c, 7 d, 8 a, 9 e)
+SELECT a, d FROM (SELECT 1 a, 3 d, UNION ALL BY NAME SELECT 7 d, 8 a)


Code review comments:
diff_hunk: @@ -54,11 +54,17 @@ def pushdown_projections(expression, schema=None, remove_unused_selections=True)
             if any(select.is_star for select in right.expression.selects):
                 referenced_columns[right] = parent_selections
             elif not any(select.is_star for select in left.expression.selects):
-                referenced_columns[right] = [
-                    right.expression.selects[i].alias_or_name
-                    for i, select in enumerate(left.expression.selects)
-                    if SELECT_ALL in parent_selections or select.alias_or_name in parent_selections
-                ]
+                union_by_name = scope.expression.args.get("by_name")
+
+                if union_by_name:
path: sqlglot/optimizer/pushdown_projections.py
text: @user1:
```suggestion
                if scope.expression.args.get("by_name"):
```