Repository navigation
Flink: Preserve RowKind in DynamicIcebergSink's DataConverter - #18101
Merged
pvary merged 1 commit intoSep 14, 2026
Merged
Conversation
RowDataConverter.convert() rebuilds rows whose schema differs from the target table (reordered/missing/widened columns) via `new GenericRowData(fieldGetters.length)`, which defaults to RowKind.INSERT. Nothing restores the original kind afterwards, so a converted DELETE/UPDATE_BEFORE row reaches BaseDeltaTaskWriter as an INSERT: it is written as data instead of a delete, or NPEs if a required non-key column is missing from the source row. Carry sourceData.getRowKind() onto the converted row. Closes apache#18100
pvary
approved these changes
Sep 14, 2026
Contributor
|
Merged to main. |
nickwu241
added a commit
to Affirm/iceberg
that referenced
this pull request
Sep 25, 2026
…-pick apache#18101) (#13) * Flink: Preserve RowKind in DynamicIcebergSink's DataConverter RowDataConverter.convert() rebuilt a row as new GenericRowData(fieldGetters.length) whenever its schema didn't structurally match the target table's, which defaults RowKind to INSERT and silently drops the source row's actual kind. A CDC delete/UPDATE_BEFORE routed through this rebuild path was coerced into an INSERT. Cherry-picked from apache#18101 (closes apache#18100). * Flink: Backport RowKind-preservation fix to v1.20 and v2.0 v1.20/v2.1 are covered by apache#18196; v2.0 was never covered upstream (the flink/v2.0 module was dropped from apache/iceberg main by apache#17849 before apache#18196 landed), so this is a manual port rather than a real cherry-pick.
1 task done
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
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
RowDataConverter.convert()rebuilds a record whenever its schema is not byte-identical to the target table's — a column-order difference, a missing optional column, a widening type. It builds the replacement as:GenericRowData(int)defaultsRowKindtoINSERT, and nothing downstream restores the original kind —RowKindis never referenced anywhere else in thesink/dynamicpackage.Why
BaseDeltaTaskWriterdispatches purely onRowKind. So a CDC delete/UPDATE_BEFOREthat happens to go through this conversion path (e.g. a delete envelope carrying only key columns) is treated as anINSERT:NullPointerExceptiondownstreamEverything else the converter does is already correct — it maps fields by name, null-fills missing optional columns, and widens types — so carrying the kind through is the whole fix.
Fix
Added
TestRowDataConverter#testPreservesRowKind, which fails on the old code (expected: DELETE but was: INSERT) and passes with the fix.Applied to
flink/v2.3only, following the convention used for the original RowData-evolution feature (#13340, backported to v1.19/v1.20 separately in #13401) — happy to backport to v1.20/v2.1/v2.2 in a follow-up if maintainers want it there too.Closes #18100