vishalmore90 opened a new issue, #41047: URL: https://github.com/apache/beam/issues/41047
### What happened? In `sdks/python/apache_beam/yaml/yaml_combine.py`, `_SqlCombineTransform` generates a SQL query string to perform aggregations with Calcite / SQL-backed transforms. However, when performing a global aggregation without grouping keys (i.e. `group_by: []` or omitting `group_by`), `_SqlCombineTransform` unconditionally appends `GROUP BY %s` formatted with `', '.join(group_by)`: https://github.com/apache/beam/blob/master/sdks/python/apache_beam/yaml/yaml_combine.py#L181-L208 ```python @beam.ptransform.ptransform_fn def _SqlCombineTransform( pcoll, sql_transform_constructor, group_by, combine, language=None): all_fields = [ x for x, _ in named_fields_from_element_type(pcoll.element_type) ] unknown_keys = set(group_by) - set(all_fields) if unknown_keys: raise ValueError(f'Unknown grouping columns: {list(unknown_keys)}') 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}' return pcoll | sql_transform_constructor( 'SELECT %s FROM PCOLLECTION GROUP BY %s' % ( ', '.join( list(group_by) + [combine_col(dest, fn_spec) for dest, fn_spec in combine.items()]), ', '.join(group_by), )) ``` #### Problem 1: Invalid SQL Syntax (`GROUP BY `) with `group_by: []` When `group_by` is empty (`group_by: []` or `group_by: ()`), `', '.join(group_by)` evaluates to `""`. The resulting query string constructed on line 202 is: ```sql SELECT sum(c) as c FROM PCOLLECTION GROUP BY ``` Notice the dangling `GROUP BY ` with nothing after it. When passed to Calcite (or ZetaSQL), this fails immediately with a syntax error at EOF: ``` org.apache.beam.vendor.calcite.v1_28_0.org.apache.calcite.sql.parser.SqlParseException: Encountered "<EOF>" at line 1, column 45. Was expecting one of: ... ``` In standard SQL and Calcite, a query with no grouping columns must omit the `GROUP BY` clause entirely: ```sql SELECT sum(c) as c FROM PCOLLECTION ``` #### Problem 2: `TypeError: 'NoneType' object is not iterable` when `group_by` is omitted In Beam YAML, global aggregations can naturally omit `group_by` entirely: ```yaml type: Combine config: language: sql combine: c: sum ``` However, in `yaml_combine.py:normalize_combine`: ```python if isinstance(config.get('group_by'), str): config['group_by'] = [config['group_by']] ``` If `group_by` is not provided, `config.get('group_by')` returns `None` and is left as `None`. When passed to `_SqlCombineTransform`: ```python unknown_keys = set(group_by) - set(all_fields) ``` Python raises: ```python TypeError: 'NoneType' object is not iterable ``` Similarly, `PyJsYamlCombine.__init__(self, group_by: Iterable[str], ...)` lacks a default parameter for `group_by`, raising `TypeError: PyJsYamlCombine.__init__() missing 1 required positional argument: 'group_by'`. ### Code Pointers / Steps to Reproduce 1. Define a pipeline with a global combine using SQL: ```yaml pipeline: transforms: - type: Create config: elements: - {a: 'x', b: 1, c: 101} - {a: 'x', b: 1, c: 102} - type: Combine config: language: sql group_by: [] combine: total_c: fn: sum value: c ``` 2. Run the pipeline. 3. Observe that `_SqlCombineTransform` generates: `SELECT sum(c) as total_c FROM PCOLLECTION GROUP BY ` and fails with Calcite `SqlParseException`. 4. Existing testing gap: In `sdks/python/apache_beam/yaml/yaml_combine_test.py:YamlCombineTest`: `test_no_keys` (lines 75–92) only tests `group_by: []` with the default in-memory Python provider (`PyJsYamlCombine`). There are no tests for global aggregation with `_SqlCombineTransform` / `language: sql`. ### Impact - Any Beam YAML pipeline using Calcite/SQL-backed `Combine` to compute global aggregates (e.g., sums, counts, min/max across the entire dataset) fails to parse and cannot execute. - Omitting `group_by` in YAML configuration causes a crash during pipeline construction (`TypeError: 'NoneType' object is not iterable`). ### Proposed Solution 1. In `_SqlCombineTransform`, default `group_by` to an empty collection if `None`, and conditionally omit `GROUP BY`: ```python @beam.ptransform.ptransform_fn def _SqlCombineTransform( pcoll, sql_transform_constructor, group_by=(), combine=None, language=None): group_by = list(group_by or []) all_fields = [ x for x, _ in named_fields_from_element_type(pcoll.element_type) ] unknown_keys = set(group_by) - set(all_fields) if unknown_keys: raise ValueError(f'Unknown grouping columns: {list(unknown_keys)}') 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}' select_exprs = group_by + [ combine_col(dest, fn_spec) for dest, fn_spec in combine.items() ] if group_by: query = 'SELECT %s FROM PCOLLECTION GROUP BY %s' % ( ', '.join(select_exprs), ', '.join(group_by)) else: query = 'SELECT %s FROM PCOLLECTION' % ', '.join(select_exprs) return pcoll | sql_transform_constructor(query) ``` 2. In `normalize_combine(spec)` in `sdks/python/apache_beam/yaml/yaml_combine.py`: ```python group_by = config.get('group_by') if group_by is None: config['group_by'] = [] elif isinstance(group_by, str): config['group_by'] = [group_by] ``` 3. In `PyJsYamlCombine.__init__`: ```python def __init__( self, group_by: Iterable[str] = (), combine: Mapping[str, Mapping[str, Any]] = None, language: Optional[str] = None): self._group_by = tuple(group_by or ()) self._combine = combine self._language = language ``` 4. Add a unit test in `yaml_combine_test.py` testing SQL combine with `group_by: []` and omitted `group_by`. ### Issue Priority Priority: 2 (default / most bugs should be filed as P2) ### Issue Components - [ ] Component: Python SDK - [ ] Component: Java SDK - [ ] Component: Go SDK - [ ] Component: Typescript SDK - [ ] Component: IO connector - [x] Component: Beam YAML - [ ] Component: Beam examples - [ ] Component: Beam playground - [ ] Component: Beam katas - [ ] Component: Website - [ ] Component: Infrastructure - [ ] Component: Spark Runner - [ ] Component: Flink Runner - [ ] Component: Prism Runner - [ ] Component: Hazelcast Jet Runner - [ ] Component: Google Cloud Dataflow Runner -- 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]
