[
https://issues.apache.org/jira/browse/FLINK-40935?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18125549#comment-18125549
]
Donam Kim edited comment on FLINK-40935 at 10/9/26 2:14 PM:
------------------------------------------------------------
Hi, thank you for the detailed report and the clear reproduction.
If no one else is working on it yet, I would like to work on this issue.
>From my investigation, the root cause seems to be the return type inference
>rather than the computation itself. For a grouped aggregation over a NOT NULL
>input, VAR_SAMP and VARIANCE are inferred as NOT NULL. As a result, the NULL
>produced by the rewrite in AggregateReduceFunctionsRule is cast to a NOT NULL
>type, and the generated code returns the default value -1 (-1.0 for DOUBLE).
Calcite has fixed this in CALCITE-7656, which is planned for Calcite 1.43.0 and
is not included in 1.42.0. STDDEV_SAMP has already been nullable since Calcite
1.38 (CALCITE-6547), so on master the SQL STDDEV_SAMP returns NULL correctly.
However, the Table API still infers NOT NULL for both functions, and on master
stddevSamp() even fails with an EnforcerException for such a group.
I also checked the released versions with the same query. 2.0.2, 2.1.3, 2.2.1
and 2.3.0 all return -1 for VAR_SAMP, STDDEV_SAMP, VARIANCE and STDDEV in SQL,
as well as for varSamp() and stddevSamp() in the Table API, in both batch and
streaming mode. So the affected versions might also include 2.0.x to 2.2.x. In
streaming mode, every group emits -1 while it contains only one row, so groups
that later receive more rows also produce a wrong intermediate result.
My proposal would be to define VAR_SAMP and VARIANCE in FlinkSqlOperatorTable
with an always nullable return type until Flink upgrades to a Calcite version
that includes CALCITE-7656, and to align the output type strategies of VAR_SAMP
and STDDEV_SAMP in BuiltInFunctionDefinitions accordingly. I already have a fix
with tests prepared locally.
Could a committer please assign this ticket to me? Any feedback on the approach
would be very much appreciated. Thank you!
was (Author: JIRAUSER314930):
Hi, thank you for the detailed report and the clear reproduction.
If no one else is working on it yet, I would like to work on this issue.
>From my investigation, the root cause seems to be the return type inference
>rather than the computation itself. For a grouped aggregation over a NOT NULL
>input, VAR_SAMP and VARIANCE are inferred as NOT NULL. As a result, the NULL
>produced by the rewrite in AggregateReduceFunctionsRule is cast to a NOT NULL
>type, and the generated code returns the default value -1 (-1.0 for DOUBLE).
Calcite has fixed this in CALCITE-7656, which is planned for Calcite 1.43.0 and
is not included in 1.42.0. STDDEV_SAMP has already been nullable since Calcite
1.38 (CALCITE-6547), so on master the SQL STDDEV_SAMP returns NULL correctly.
However, the Table API still infers NOT NULL for both functions, and on master
stddevSamp() even fails with an EnforcerException for such a group.
I also checked the released versions with the same query. 2.0.2, 2.1.3, 2.2.1
and 2.3.0 all return -1 for VAR_SAMP, STDDEV_SAMP, VARIANCE and STDDEV in SQL,
as well as for varSamp() and stddevSamp() in the Table API, in both batch and
streaming mode. So the affected versions might also include 2.0.x to 2.2.x. In
streaming mode, every group emits -1 while it contains only one row, so groups
that later receive more rows also produce a wrong intermediate result.
My proposal would be to define VAR_SAMP and VARIANCE in FlinkSqlOperatorTable
with an always nullable return type until Flink upgrades to a Calcite version
that includes CALCITE-7656, and to align the output type strategies of VAR_SAMP
and STDDEV_SAMP in BuiltInFunctionDefinitions accordingly. I already have a fix
with tests prepared locally.
Could a committer please assign this ticket to me? Any feedback on the approach
would be very much appreciated. Thank you!
> VAR_SAMP / STDDEV_SAMP return -1 for a group with a single row
> --------------------------------------------------------------
>
> Key: FLINK-40935
> URL: https://issues.apache.org/jira/browse/FLINK-40935
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.3.0, 2.4.0
> Reporter: Yaoxuan Wu
> Priority: Major
>
> {code:java}
> SELECT k, VAR_SAMP(c), STDDEV_SAMP(c) FROM (VALUES (1, 10), (2, 20)) AS v(k,
> c) GROUP BY k;
> -- Flink: (1, -1, -1), (2, -1, -1) expected: (1, NULL, NULL), (2, NULL,
> NULL)
> SELECT VAR_SAMP(c) FROM (VALUES (10)) AS v(c);
> -- NULL (correct, no GROUP BY)
> SELECT k, VAR_SAMP(c) FROM (VALUES (1, CAST(NULL AS INT)), (1, 10)) AS v(k,
> c) GROUP BY k;
> -- NULL (correct, nullable input) {code}
> The sample variance of a single value is NULL, and a variance can never be
> negative. With GROUP BY and a NOT NULL input, the result type is inferred as
> {{{}INT NOT NULL{}}}, although the value is NULL for a group with one row:
> {code:java}
> SELECT VAR_SAMP(c) AS r FROM (VALUES (1, 10)) AS v(k, c) GROUP BY k -- r:
> INT NOT NULL, returns -1
> SELECT VAR_SAMP(c) AS r FROM (VALUES (10)) AS v(c) -- r:
> INT, returns NULL {code}
> The optimized plan computes CAST(/($f1, CASE(=($f2, 1), null:BIGINT, -($f2,
> 1))) AS INTEGER), so the NULL produced by the CASE is lost and -1 (the
> default value of a non-nullable INT) is returned. DOUBLE inputs return -1.0.
> Batch and streaming mode behave the same.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
