[vishalmore90]

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.

[github-actions]

📋 Review Summary

This PR cleanly resolves SQL query generation bugs in Beam YAML's Combine transform when using calcite or sql dialects with full SQL expressions, shorthand column aggregations, or explicit Combine-<language> transform types. The implementation in normalize_combine and _SqlCombineTransform is well-structured and backed by unit tests covering the newly fixed scenarios.

🔍 General Feedback

• The normalization logic and fallback handling in _SqlCombineTransform properly handle both parameterized expressions (containing () and shorthand function names across sql and calcite configurations.
• Unit tests directly validate the generated SQL queries without requiring an external Java/Calcite expansion service, keeping the test suite fast and deterministic.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

Assigning reviewers:

R: @chamikaramj for label yaml.
Backup reviewer: derrickaw (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).