[
https://issues.apache.org/jira/browse/FLINK-40368?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18124459#comment-18124459
]
Gustavo de Morais commented on FLINK-40368:
-------------------------------------------
Backported to
* release-2.3:
[fa2ec19c464|https://github.com/apache/flink/commit/fa2ec19c46454d00778146618985e37e245e951e]
* release-2.2:
[768a0379c63|https://github.com/apache/flink/commit/768a0379c633a1d27e6ad317a88b567dbc4b1e86]
release-2.1 is no longer supported by the community, so the 2.1 backport was
closed.
> 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)
