dataengineervishal commented on code in PR #4536:
URL: https://github.com/apache/flink-cdc/pull/4536#discussion_r4043389377
##########
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:
@ivyanquan Nice suggestion! I was also thinking along similar lines. I was
considering checking whether the partition column value is the same in the
before and after values. If they are the same, we could send only the INSERT
otherwise, we would send both the DELETE for the old row and the INSERT for the
new row.
I’ll include the isPartitioned check as well, since it will avoid the
unnecessary overhead for non-partitioned tables.
I’ll take this up as a separate optimization task.
--
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]
