SQL Physical Operator Fixes and Enhancements

[damccorm]

Fixes critical planning and execution bugs in Beam SQL, specifically addressing Volcano planner convergence failures and correlated scalar subquery execution.

1. Volcano Planner Convergence in Windowing & Calculations

• The Problem: When planning complex queries with nested window functions (OVER clauses) or subqueries, the Volcano planner would often fail to find a physical plan, throwing CannotPlanException due to "infinite cost". This happened because BeamWindowRel and BeamCalcRel cost estimation relied on BeamSqlRelUtils.getNodeStats(input). During the Volcano search, the input is often an unoptimized RelSubset. Calling getNodeStats on these subsets returned UNKNOWN (infinite cost), creating a recursion loop and preventing the planner from converging on a valid physical path.
• The Fix: Modified BeamWindowRel.computeSelfCost and BeamCalcRel.computeSelfCost to use mq.getRowCount(this.input) instead of getNodeStats when estimating row counts during planning. This is Volcano-safe and allows the planner to successfully estimate costs and converge.

2. Correlated Subquery Execution (SINGLE_VALUE Aggregation)

• The Problem: Correlated scalar subqueries (subqueries that return a single value and reference the outer query) are translated by Calcite using a SINGLE_VALUE aggregation to guarantee at most one row is returned. Beam SQL's physical execution layer was missing the mapping for the SINGLE_VALUE operator, leading to runtime translation failures.
• The Fix: Registered and implemented the SINGLE_VALUE aggregation operator in the physical layer, mapping it to a combiner that asserts at most one element is present and returns it.
• Tests: Added a dedicated integration test in BeamSqlDslAggregationTest verifying correlated scalar subqueries execute correctly.