Rion Williams created FLINK-41260:
-------------------------------------
Summary: ARRAY_AGG with ORDER BY ignores the ordering and cannot
be serialized into a compiled plan
Key: FLINK-41260
URL: https://issues.apache.org/jira/browse/FLINK-41260
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Reporter: Rion Williams
`ARRAY_AGG(x ORDER BY y)` passes SQL validation even though Flink does not
support the ORDER BY clause in ARRAY_AGG, as the documentation says. What
happens next depends on how the query runs.
*Direct execution ignores the ordering.* The query runs, and the array holds
its elements in arrival order:
{code:sql}
-- rows (k, v, s): (1, 'c', 3), (1, 'a', 1), (1, 'b', 2)
SELECT k, ARRAY_AGG(v ORDER BY s DESC) FROM src GROUP BY k;
-- returns [c, a, b], expected [c, b, a]
{code}
*Compiled plans fail with an internal error.* In both streaming and batch mode:
{code:java}
tEnv.compilePlanSql("INSERT INTO snk SELECT k, ARRAY_AGG(v ORDER BY s) FROM src
GROUP BY k").asJsonString();
{code}
{noformat}
TableException: Symbol class 'org.apache.calcite.sql.SqlSyntax$3' has no
serializable representation.
(through reference chain:
JsonPlanGraph["nodes"]->ArrayList[2]->StreamExecGroupAggregate["aggCalls"]->AggregateCall[0])
{noformat}
In batch mode the chain goes through BatchExecSortAggregate. When assertions
are enabled (-ea), planning fails earlier, with a Calcite "type mismatch"
AssertionError.
*Root cause*
# The parser always creates the call with Calcite's
SqlLibraryOperators.ARRAY_AGG, whose syntax is ORDERED_FUNCTION.
# During validation, SqlBasicAggFunction#deriveType strips the ORDER BY with
ReturnTypes#stripOrderBy, which creates a copy of the call.
SqlFunction#deriveType resolves Flink's own ARRAY_AGG
(FlinkSqlOperatorTable#ARRAY_AGG, syntax FUNCTION) but swaps the operator only
on that copy.
# Without ORDER BY, the copy is the original call, so plain ARRAY_AGG gets
Flink's operator. With ORDER BY, the original call keeps Calcite's operator,
and SqlToRelConverter moves the ORDER BY into the AggregateCall collation.
# ArrayAggFunction ignores the collation, so the ordering is lost. SymbolUtil
has no mapping for SqlSyntax.ORDERED_FUNCTION, so the compiled plan cannot be
serialized.
*Proposed fix*
Reject ORDER BY inside ARRAY_AGG during validation with a ValidationException.
This enforces the documented behaviour. Adding only a SymbolUtil mapping, or
only fixing the operator swap, would make a plan that silently drops the
ordering serializable.
The other aggregates that Calcite's parser rule covers (STRING_AGG,
GROUP_CONCAT, ARRAY_CONCAT_AGG) are already rejected by Flink's parser, so
ARRAY_AGG is the only one affected.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
