[
https://issues.apache.org/jira/browse/FLINK-40368?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18123825#comment-18123825
]
Gustavo de Morais commented on FLINK-40368:
-------------------------------------------
Yes, I've opened PRs to fix this for 2.1, 2.2 and 2.3. I’d like to fix all
affected versions here, since the bug was introduced in 2.1 and 2.0 and earlier
plan these queries correctly. If only 2.2 and 2.3 are fixed, 2.1 would be the
only version still producing an inconsistent and wrong plan.
> Deduplicate output should only be insert-only when ordered by time attribute
> ----------------------------------------------------------------------------
>
> Key: FLINK-40368
> URL: https://issues.apache.org/jira/browse/FLINK-40368
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.3.0, 2.2.1, 2.1.3
> Reporter: David Anderson
> Assignee: Gustavo de Morais
> Priority: Major
> Labels: pull-request-available
> Fix For: 2.4.0
>
>
> FLINK-37005 went too far, and the planner now incorrectly treats
> deduplication with ordering ASC on an ordering column that not a time
> attribute as producing an insert-only result.
> Claude suggests this approach for fixing this:
> Make the inference guard match the translation guard — add the
> {{sortOnTimeAttributeOnly}} condition at
> FlinkChangelogModeInferenceProgram.scala:264 so non-time top-1 ranks fall
> into the existing {{ALL_CHANGES}} branch at line 286. One wrinkle: you can't
> simply call {{canConvertToDeduplicate}} there, because it calls
> {{{}ChangelogPlanUtils.inputInsertOnly{}}}, which reads a trait that isn't
> set yet during inference — the same trap documented at
> FlinkRelMdModifiedMonotonicity.scala:252. So expose
> {{RankUtil.sortOnTimeAttributeOnly}} (currently private) and combine it with
> the locally computed {{insertOnly}} from {{{}children{}}}.
> Test gap worth closing at the same time: {{DeduplicateTest}} only covers
> {{proctime/rowtime/PROCTIME()}} ordering — there is no case with a plain
> column. Relatedly, {{StreamPhysicalRank.getDeduplicateDescription}} hardcodes
> {{{}order=[PROCTIME|ROWTIME]{}}}, which is another sign the dedup path
> assumes a time attribute throughout.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
