vishalmore90 opened a new pull request, #41289:
URL: https://github.com/apache/beam/pull/41289
<!--
Thank you for your contribution! Follow this checklist to help us
incorporate your contribution quickly and easily:
- [ ] If this contribution is large, please file an Apache [Individual
Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
- [x] Replace these instructions with a description of your change and why
you have made it.
- [x] Mention the appropriate issue in your description (for example:
`addresses #123` or `fixes #456`).
- [x] Update `CHANGES.md` with noteworthy changes.
See [CONTRIBUTING.md]() for more details.
-->
fixes #41288
### Description
In Beam YAML, SQL-backed aggregations are supported via both `language: sql`
and `language: calcite` (as registered by `SqlBackedProvider` in
`apache_beam/yaml/yaml_combine.py` and documented in
`documentation/sdks/yaml-combine.md`).
However, two related query generation bugs occurred:
1. When using `language: calcite` with SQL expressions (e.g. `total:
"sum(col2)"`), `normalize_combine` checked `config.get('language') != 'sql'`,
which evaluated to `True`. This mistakenly injected `agg['value'] = dest`
('total'). In `_SqlCombineTransform`, having a `value` caused `combine_col` to
format it as `sum(col2)(total) as total`, which failed in Calcite with
`SqlParseException`.
2. When using `language: sql` with shorthand aggregations (e.g. `col2:
sum`), `config.get('language') != 'sql'` evaluated to `False`, skipping
`agg['value'] = dest`. In `_SqlCombineTransform`, having no `value` caused
`combine_col` to treat `'sum'` as a complete expression, formatting it as `sum
as col2` (instead of `sum(col2) as col2`).
3. If `type` was specified with an explicit provider suffix like `type:
Combine-sql` or `Combine-calcite`, `normalize_combine` did not run because it
strictly checked `spec['type'] == 'Combine'`.
### Changes Made
1. **`normalize_combine`**:
- Matches `spec['type'] == 'Combine' or
spec['type'].startswith('Combine-')`.
- Recognizes both `'sql'` and `'calcite'` as SQL dialects.
- For SQL dialects, only skips injecting `agg['value'] = dest` if the
aggregation is already an expression containing arguments (`'(' in fn_name`).
Shorthand aggregations (e.g. `col2: sum`) properly receive `agg['value'] =
'col2'`.
2. **`_SqlCombineTransform`**:
- In `combine_col`, checks if `fn_type` already contains `(`. If so, uses
it directly as the expression without attempting to append parentheses or
arguments.
- If `fn_type` does not contain `(` and no `value` was specified,
defaults the argument to `dest` (`f'{fn_type}({dest})'`).
3. **Tests & Docs**:
- Added unit tests in `yaml_combine_test.py` verifying Calcite
expressions (`sum(c)`, `count(*)`), SQL shorthand (`c: sum`), Calcite
shorthand, and `Combine-calcite` type suffixes.
- Updated `CHANGES.md`.
--
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]
