vishalmore90 opened a new issue, #41288:
URL: https://github.com/apache/beam/issues/41288
### What happened?
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, `normalize_combine` in
`sdks/python/apache_beam/yaml/yaml_combine.py` contains a language-check flaw:
```python
def normalize_agg(dest, agg):
if isinstance(agg, str):
agg = {'fn': agg}
if 'value' not in agg and config.get('language') != 'sql':
agg['value'] = dest
if isinstance(agg['fn'], str):
agg['fn'] = {'type': agg['fn']}
return agg
```
This causes two distinct query generation failures:
#### 1. `language: calcite` with SQL expressions produces `sum(col)(dest) as
dest`
According to the documentation for SQL-style aggregations, users specify SQL
snippets as the combine fn:
```yaml
type: Combine
config:
language: calcite
group_by: col1
combine:
total: "sum(col2)"
num_values: "count(*)"
```
Because `config.get('language') != 'sql'` evaluates to `True` for
`'calcite'`, `agg['value'] = dest` is mistakenly executed, setting
`agg['value'] = 'total'`.
Then in `_SqlCombineTransform`:
```python
def combine_col(dest, fn_spec):
if 'value' in fn_spec or 'config' in fn_spec['fn']:
expr = '%s(%s)' % (
fn_spec['fn']['type'],
', '.join([fn_spec['value']] +
list(fn_spec['fn'].get('config', {}).values())))
else:
expr = fn_spec['fn']['type']
return f'{expr} as {dest}'
```
Because `'value'` is present in `fn_spec`, it constructs:
- `expr = 'sum(col2)(total)'`
- `expr = 'count(*)(num_values)'`
Generating the invalid SQL query:
```sql
SELECT col1, sum(col2)(total) as total, count(*)(num_values) as num_values
FROM PCOLLECTION GROUP BY col1
```
Calcite's SQL parser rejects this with `SqlParseException: Encountered "("`.
#### 2. `language: sql` with shorthand column aggregations produces `sum as
col2`
According to the documentation, shorthand aggregation where destination is
the input column is supported (e.g. `col2: sum`).
When specified under `language: sql`:
```yaml
type: Combine
config:
language: sql
group_by: col1
combine:
col2: sum
```
Because `config.get('language') != 'sql'` evaluates to `False`,
`agg['value'] = dest` is skipped.
`_SqlCombineTransform` receives `fn_spec = {'fn': {'type': 'sum'}}` without
`'value'`. It falls back to `expr = fn_spec['fn']['type']` (`'sum'`),
generating:
```sql
SELECT col1, sum as col2 FROM PCOLLECTION GROUP BY col1
```
Which is also completely invalid SQL syntax (`sum as col2` instead of
`sum(col2) as col2`).
---
### Steps to Reproduce
Run the following minimal script against `master`:
```python
import apache_beam as beam
from apache_beam.yaml.yaml_combine import normalize_combine,
_SqlCombineTransform
pcoll = beam.Pipeline() | beam.Create([beam.Row(col1='a', col2=10)])
# Case 1: language: calcite with SQL expressions
spec1 = {
'type': 'Combine',
'config': {
'language': 'calcite',
'group_by': ['col1'],
'combine': {
'total': 'sum(col2)',
'num_values': 'count(*)'
}
}
}
queries1 = []
_ = pcoll | _SqlCombineTransform(
lambda q: queries1.append(q) or beam.Map(lambda x: x),
**normalize_combine(spec1)['config'])
print("Case 1 (calcite):", queries1[-1])
# Output: SELECT col1, sum(col2)(total) as total, count(*)(num_values) as
num_values FROM PCOLLECTION GROUP BY col1
# Case 2: language: sql with shorthand
spec2 = {
'type': 'Combine',
'config': {
'language': 'sql',
'group_by': ['col1'],
'combine': {
'col2': 'sum'
}
}
}
queries2 = []
_ = pcoll | _SqlCombineTransform(
lambda q: queries2.append(q) or beam.Map(lambda x: x),
**normalize_combine(spec2)['config'])
print("Case 2 (sql):", queries2[-1])
# Output: SELECT col1, sum as col2 FROM PCOLLECTION GROUP BY col1
```
---
### Proposed Solution
In `sdks/python/apache_beam/yaml/yaml_combine.py`:
1. Check whether the language is a SQL dialect (`language in ('sql',
'calcite')`).
2. Only suppress defaulting `agg['value'] = dest` when the provided
aggregation snippet is already a call expression (e.g., contains `(`):
```python
def normalize_agg(dest, agg):
if isinstance(agg, str):
agg = {'fn': agg}
is_sql = config.get('language') in ('sql', 'calcite')
fn_type = (agg.get('fn') if isinstance(agg.get('fn'), str)
else agg.get('fn', {}).get('type', ''))
if 'value' not in agg:
if not (is_sql and '(' in fn_type):
agg['value'] = dest
if isinstance(agg['fn'], str):
agg['fn'] = {'type': agg['fn']}
return agg
```
This properly generates:
- `SELECT col1, sum(col2) as total, count(*) as num_values FROM PCOLLECTION
GROUP BY col1` for both `calcite` and `sql` expressions.
- `SELECT col1, sum(col2) as col2 FROM PCOLLECTION GROUP BY col1` for
shorthand function names in both dialects.
---
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
---
### Issue Components
- [x] Component: Python SDK
- [x] Component: Beam YAML
--
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]
