public Builder upsert(boolean enabled) {
            this.upsert = enabled;
            writeOptions.put(FlinkWriteOptions.WRITE_UPSERT_ENABLED.key(), Boolean.toString(enabled));
            return this;
        }
All INSERT/UPDATE_AFTER events from input stream will be transformed to UPSERT events, which means it will
DELETE the old records and then INSERT the new records. In partitioned table, the partition fields should be
a subset of equality fields, otherwise the old row that located in partition-A could not be deleted by the
new row that located in partition-B.

@param enabled indicate whether it should transform all INSERT/UPDATE_AFTER events to UPSERT.
@return {@link Builder} to connect the iceberg table.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/sink/FlinkSink.java