[
https://issues.apache.org/jira/browse/FLINK-41371?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Gustavo de Morais updated FLINK-41371:
--------------------------------------
Description:
PTFs with row semantics always receive retractions for upsert inputs, because
{{FlinkChangelogModeInferenceProgram}} requires UPDATE_BEFORE whenever there is
no PARTITION BY. This adds an unnecessary {{ChangelogNormalize}} and hash
exchange, e.g. for {{TO_CHANGELOG}} with an {{op_mapping}} without
UPDATE_BEFORE.
We should keep upsert mode when the input has an upsert key and the PTF is
insert-only, so the PTF can be chained to its input.
was:
PTFs with row semantics always receive retractions for upsert inputs, because
{{FlinkChangelogModeInferenceProgram}} requires UPDATE_BEFORE whenever there is
no PARTITION BY. This adds an unnecessary {{ChangelogNormalize}} and hash
exchange, e.g. for {{TO_CHANGELOG}} with an {{op_mapping}} without
UPDATE_BEFORE.
We should keep upsert mode when the input has an upsert key and the PTF is
insert-only or passes all columns through, so the PTF can be chained to its
input.
> Row-semantic PTFs force retract mode on upsert inputs
> -----------------------------------------------------
>
> Key: FLINK-41371
> URL: https://issues.apache.org/jira/browse/FLINK-41371
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.3.0, 2.4.0
> Reporter: Gustavo de Morais
> Assignee: Gustavo de Morais
> Priority: Major
> Labels: pull-request-available
>
> PTFs with row semantics always receive retractions for upsert inputs, because
> {{FlinkChangelogModeInferenceProgram}} requires UPDATE_BEFORE whenever there
> is no PARTITION BY. This adds an unnecessary {{ChangelogNormalize}} and hash
> exchange, e.g. for {{TO_CHANGELOG}} with an {{op_mapping}} without
> UPDATE_BEFORE.
> We should keep upsert mode when the input has an upsert key and the PTF is
> insert-only, so the PTF can be chained to its input.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
