[
https://issues.apache.org/jira/browse/FLINK-41260?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rion Williams updated FLINK-41260:
----------------------------------
Description:
`ARRAY_AGG(x ORDER BY y)` and its sibling `JSON_ARRAY_AGG(...)` both pass SQL
validation even though Flink does not support the ORDER BY clause in
ARRAY_AGG/JSON_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/JSON_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.
was:
`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.
> 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)` and its sibling `JSON_ARRAY_AGG(...)` both pass SQL
> validation even though Flink does not support the ORDER BY clause in
> ARRAY_AGG/JSON_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/JSON_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)
