[FLINK-39538][table] Support upsert output mode for FROM_CHANGELOG#28164
Merged
Conversation
Collaborator
fhueske
reviewed
May 18, 2026
Contributor
gustavodemorais
left a comment
gustavodemorais
left a comment
There was a problem hiding this comment.
The PR in general looks very good, @raminqaf!
I've added multiple comments and doc suggestions and only one change requested.
raminqaf
force-pushed
the
FLINK-39538
branch
5 times, most recently
from
7da043f to
41ce0e3
Compare
fhueske
approved these changes
May 18, 2026
github-actions
Bot
added
the
community-reviewed
Contributor
gustavodemorais
left a comment
gustavodemorais
left a comment
There was a problem hiding this comment.
Thanks for addressing the feedback, @raminqaf! Take a look
FROM_CHANGELOG now emits an upsert changelog (INSERT, UPDATE_AFTER, full DELETE) when the input table is partitioned (set semantics via PARTITION BY) and the active op_mapping maps to UPDATE_AFTER without UPDATE_BEFORE. The partition key acts as the upsert key. In all other cases the output remains a retract changelog. Submitting an op_mapping with UPDATE_AFTER but no UPDATE_BEFORE without PARTITION BY is rejected at validation time, because upsert mode requires a key. To enable the strategy to inspect the resolved op_mapping and the input table's partition keys, ChangelogFunction.ChangelogContext is extended with two default methods: getArgumentValue(int, Class) and getTableSemantics(int). Defaults return Optional.empty() to preserve source compatibility for existing implementations. The planner-side wrapper in FlinkChangelogModeInferenceProgram delegates the two new methods to the underlying CallContext. Upsert mode uses full deletes (ChangelogMode.upsert(false)) because the runtime forwards each input delete row with all fields populated; only the RowKind is rewritten. This matches the runtime's behavior and avoids forcing downstream operators to reconstruct rows from state. The upsert key derivation in FlinkRelMdUniqueKeys.getPtfUniqueKeys already returns the partition columns when a PTF emits upsert, so no metadata changes are needed.
Contributor
gustavodemorais
left a comment
gustavodemorais
left a comment
There was a problem hiding this comment.
Thanks for addressing the feedback, @raminqaf! Last two nits and we should be good to go
gustavodemorais
approved these changes
May 19, 2026
Contributor
gustavodemorais
left a comment
gustavodemorais
left a comment
There was a problem hiding this comment.
Thanks for addressing all the feedback. LGTM, @raminqaf
gustavodemorais
approved these changes
May 19, 2026
Contributor
|
Merging this PR |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
FROM_CHANGELOGnow emits an upsert changelog (INSERT,UPDATE_AFTER, fullDELETE) when the input table is partitioned (set semantics viaPARTITION BY) and the activeop_mappingmaps toUPDATE_AFTERwithoutUPDATE_BEFORE. The partition key acts as the upsert key. In all other cases the output remains a retract changelog.Submitting an
op_mappingwithUPDATE_AFTERbut noUPDATE_BEFOREwithoutPARTITION BYis rejected at validation time, because upsert mode requires a key.To enable the strategy to inspect the resolved op_mapping and the input table's partition keys, ChangelogFunction.ChangelogContext is extended with two default methods: getArgumentValue(int, Class) and getTableSemantics(int). Defaults return Optional.empty() to preserve source compatibility for existing implementations.
The planner-side wrapper in FlinkChangelogModeInferenceProgram delegates the two new methods to the underlying CallContext.
Upsert mode uses full deletes (ChangelogMode.upsert(false)) because the runtime forwards each input delete row with all fields populated; only the RowKind is rewritten. This matches the runtime's behavior and avoids forcing downstream operators to reconstruct rows from state.
The upsert key derivation in FlinkRelMdUniqueKeys.getPtfUniqueKeys already returns the partition columns when a PTF emits upsert, so no metadata changes are needed.
Verifying this change
Changes are backed by semantic tests and plan verification tests
FromChangelogTestProgramFromChangelogTestto verify generated plan and upsertKeysDoes this pull request potentially affect one of the following parts:
@Public(Evolving): (yes / no)Documentation
Was generative AI tooling used to co-author this PR?
Generated-by: Opus 4.7