@Override
    public void write(RowData row) throws IOException {
        RowDataDeltaWriter writer = route(row);

        switch (row.getRowKind()) {
            case INSERT:
            case UPDATE_AFTER:
                if (upsert) {
                    writer.deleteKey(keyProjection.wrap(row));
                }
                writer.write(row);
                break;

            case UPDATE_BEFORE:
                if (upsert) {
                    break; // UPDATE_BEFORE is not necessary for UPDATE, we do nothing to prevent delete one row twice
                }
                writer.delete(row);
                break;
            case DELETE:
                if (upsert) { // https://github.com/apache/iceberg/pull/6753/files
                    writer.deleteKey(keyProjection.wrap(row));
                } else {
                    writer.delete(row);
                }
                break;

            default:
                throw new UnsupportedOperationException("Unknown row kind: " + row.getRowKind());
        }
    }
Copy from iceberg-flink:iceberg-flink-1.13:0.13.2
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/BaseDeltaTaskWriter.java