Repository navigation
[Bug]: Beam YAML SQL-backed Combine generates invalid syntax for language: calcite expressions and language: sql shorthand aggregations #41288
Description
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:
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 aggThis 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:
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:
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:
SELECT col1, sum(col2)(total) as total, count(*)(num_values) as num_values FROM PCOLLECTION GROUP BY col1Calcite'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:
type: Combine
config:
language: sql
group_by: col1
combine:
col2: sumBecause 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:
SELECT col1, sum as col2 FROM PCOLLECTION GROUP BY col1Which 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:
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 col1Proposed Solution
In sdks/python/apache_beam/yaml/yaml_combine.py:
- Check whether the language is a SQL dialect (
language in ('sql', 'calcite')). - Only suppress defaulting
agg['value'] = destwhen the provided aggregation snippet is already a call expression (e.g., contains():
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 aggThis properly generates:
SELECT col1, sum(col2) as total, count(*) as num_values FROM PCOLLECTION GROUP BY col1for bothcalciteandsqlexpressions.SELECT col1, sum(col2) as col2 FROM PCOLLECTION GROUP BY col1for shorthand function names in both dialects.
Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
Issue Components
- Component: Python SDK
- Component: Beam YAML
Activity
vishalmore90 commentedon Nov 13, 2026
.take-issue