[
https://issues.apache.org/jira/browse/FLINK-41260?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rion Williams updated FLINK-41260:
----------------------------------
Summary: ARRAY_AGG/JSON_ARRAY_AGG with ORDER BY ignores the ordering and
cannot be serialized into a compiled plan (was: ARRAY_AGG with ORDER BY
ignores the ordering and cannot be serialized into a compiled plan)
> ARRAY_AGG/JSON_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
> Priority: Major
> Labels: pull-request-available
>
> `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)
