lvyanquan commented on code in PR #4536:
URL: https://github.com/apache/flink-cdc/pull/4536#discussion_r4034834614
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-iceberg/src/main/java/org/apache/flink/cdc/connectors/iceberg/sink/utils/RowDataUtils.java:
##########
@@ -23,34 +23,43 @@
import org.apache.flink.table.data.RowData;
import org.apache.flink.types.RowKind;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
/** Utils for convertion of {@link RowData} and {@link DataChangeEvent}. */
public class RowDataUtils {
- /** Convert {@link DataChangeEvent} to {@link RowData}. */
- public static RowData convertDataChangeEventToRowData(
+ /**
+ * Convert a {@link DataChangeEvent} to one or two {@link RowData}.
+ *
+ * <p>An {@code UPDATE} is split into a {@code DELETE} built from the
before-image followed by
+ * an {@code INSERT} built from the after-image, rather than a single
after-image row. Iceberg
+ * derives both the equality-delete key and the delete's target partition
from the row passed to
+ * the writer; collapsing an UPDATE to just the after-image (as before)
means a partition key
+ * change is never deleted from its old partition, leaving a stale
duplicate behind.
+ */
+ public static List<RowData> convertDataChangeEventToRowData(
DataChangeEvent dataChangeEvent, List<RecordData.FieldGetter>
fieldGetters) {
- RecordData recordData;
- RowKind kind;
switch (dataChangeEvent.op()) {
case INSERT:
- case UPDATE:
case REPLACE:
- {
- recordData = dataChangeEvent.after();
- kind = RowKind.INSERT;
- break;
- }
+ return Collections.singletonList(
+ toRowData(dataChangeEvent.after(), fieldGetters,
RowKind.INSERT));
case DELETE:
- {
- recordData = dataChangeEvent.before();
- kind = RowKind.DELETE;
- break;
- }
+ return Collections.singletonList(
+ toRowData(dataChangeEvent.before(), fieldGetters,
RowKind.DELETE));
+ case UPDATE:
+ return Arrays.asList(
+ toRowData(dataChangeEvent.before(), fieldGetters,
RowKind.DELETE),
+ toRowData(dataChangeEvent.after(), fieldGetters,
RowKind.INSERT));
Review Comment:
Great fix!
One optimization suggestion:
Since the Iceberg upsert writer automatically calls `deleteKey()` on every
`INSERT` row, the explicit `DELETE` row is **redundant for non-partitioned
tables** — the auto-generated equality-delete already covers the old row in the
same partition. The bug only affects partitioned tables where the partition
value changes.
Suggest only splitting when the table is partitioned. `TableSchemaWrapper`
already holds the `Schema`, so adding a cached `isPartitioned` flag keeps
per-record overhead to a single field read:
```java
// TableSchemaWrapper
private final boolean partitioned;
public TableSchemaWrapper(Schema schema, ZoneId zoneId) {
...
this.partitioned = !schema.partitionKeys().isEmpty();
}
public boolean isPartitioned() {
return partitioned;
}
```
```java
// RowDataUtils — UPDATE case
if (isPartitioned) {
return Arrays.asList(
toRowData(before, fieldGetters, RowKind.DELETE),
toRowData(after, fieldGetters, RowKind.INSERT));
} else {
return Collections.singletonList(
toRowData(after, fieldGetters, RowKind.INSERT));
}
```
Overall LGTM, the optimization is non-blocking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
