ahmedabu98 opened a new pull request, #41304:
URL: https://github.com/apache/beam/pull/41304
Today you have to name a metadata column twice to read it from
`CHANGES(...)`: once in the SELECT and again in `include_metadata_columns`.
With this PR, you name it once, in the query.
```sql
-- Before
SELECT id, name, _change_type, _commit_snapshot_id
FROM CHANGES(orders, 2, 4)
WITH ('include_metadata_columns' = ARRAY['_change_type',
'_commit_snapshot_id']);
-- After
SELECT id, name, _change_type, _commit_snapshot_id
FROM CHANGES(orders, 2, 4);
```
## What users see
- A metadata column works anywhere a column of the table would: the select
list, WHERE, GROUP BY, ORDER BY, a JOIN condition or a subquery.
- `SELECT *` returns the same columns as before: the table's columns, plus
any that `include_metadata_columns` lists. That's also how BigQuery's
pseudo-columns and Spark's metadata columns behave. A mirror can then take the
table's columns with `*` and add the metadata columns it needs:
```sql
INSERT INTO orders_mirror WITH ('mode' = 'merge-on-read',
'change_type_column' = '_change_type')
SELECT *, _change_type, _commit_snapshot_sequence_number FROM
CHANGES(orders);
```
- A real column of the table wins over a metadata column with the same name.
- If a SELECT reads two `CHANGES(...)` tables, qualify the name, as in
`a._change_type`.
- If you misspell a metadata column, the error lists the real ones:
```
Column '_chnage_type' not found in any table. CHANGES(orders) also has
these metadata columns:
_change_type, _commit_snapshot_id, _commit_snapshot_sequence_number,
_row_id, _last_updated_sequence_number
```
`include_metadata_columns` still works.
## How it works
### Connector API
`BeamSqlTable` gets a new method, and `forChanges` gets a new parameter:
```java
default List<String> getChangeMetadataColumns() { return
Collections.emptyList(); }
BeamSqlTable forChanges(ChangeRange changes, Map<String, Object> options,
List<String> metadataColumns);
```
`metadataColumns` holds the metadata columns that the query names, and the
table decides what to do with them. Iceberg tables return
`IcebergCdcMetadataColumns.SUPPORTED_COLUMNS`, and Delta Lake tables return
`DeltaIO.METADATA_COLUMNS`. Both read the named columns along with the ones in
`include_metadata_columns`.
I put the new method on `BeamSqlTable` instead of `TableProvider`. That
keeps it next to `forChanges`, and each table can have its own list.
`CHANGES(...)` hasn't been released yet, so I replaced the old `forChanges`
signature instead of keeping both.
### Finding the names
`MetadataColumnResolver` runs after parsing, before Calcite validates the
statement. For each `CHANGES(...)` read, it looks the table up the way the
validator will. That gives it the read's columns and the table's metadata
columns. Then it walks the names in each SELECT. A name goes to a read when
it's one of the read's metadata columns and isn't already one of its columns:
- `c._change_type` goes to the read aliased `c`.
- `_change_type` goes to the read in the same SELECT, or else to one in an
enclosing SELECT.
The resolver adds the names to the read's encoded table name, which becomes
``CHANGES(`orders`, start_version => 2, end_version => 4, metadata_columns =>
ARRAY['_change_type'])``. `BeamCalciteSchema` then passes them to `forChanges`.
Users can't write `metadata_columns =>` themselves.
### Keeping `*` the same
Calcite has no hidden columns, so a named metadata column would show up in
`SELECT *`. To keep it out, the resolver replaces `*` and `c.*` with the read's
own columns before validation.
### Where it runs
`CalciteQueryPlanner` runs the resolver for `BeamSqlEnv` and `SqlTransform`.
JDBC statements, which include everything typed in the shell, go through
Calcite's own prepare code instead. For those, `JdbcDriver` now creates a
`CalcitePrepareImpl` that runs the resolver from Calcite's `PARSE_TREE` hook.
### Errors
The resolver can't tell a typo from a select alias or from a column of
another table, so Calcite still decides that a column doesn't exist. When it
fails on a name the resolver couldn't place, Beam SQL appends the read's
metadata columns to the message.
## Limitations
- A name reaches a read only from the read's own SELECT or a subquery inside
it. `SELECT _change_type FROM (SELECT * FROM CHANGES(t)) s` fails. Name the
column in the inner SELECT instead.
- `*` can't leave a named metadata column out of a NATURAL JOIN, a JOIN ...
USING or a FROM item without an alias. The query fails, and the error suggests
`c.*` or a column list.
- The resolver only sees the columns of `CHANGES(...)` reads. If another
table in the same SELECT has a column named `_change_type`, an unqualified
`_change_type` becomes ambiguous. Qualify it.
## Tests
- `TableClauseTest`: the encoded name round-trips, and users can't write
`metadata_columns`.
- `TableClauseResolutionTest`, through both `BeamSqlEnv` and JDBC: names in
each clause, stars, joins, subqueries, a table column with a metadata column's
name, and the errors.
- `BeamSqlCliIcebergCdcTest`: the push-down tests and one mirror now name
their metadata columns instead of listing them, and a misspelled column fails
with Iceberg's metadata columns. `BeamSqlLineIcebergCdcTest` does the same
through the shell.
- `BeamSqlDeltaTest`: two Delta Lake queries now name their metadata columns
instead of listing them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
