@Override
  public ChangelogMode getChangelogMode(ChangelogMode changelogMode) {
    ChangelogMode.Builder builder = ChangelogMode.newBuilder().addContainedKind(RowKind.INSERT);
    if (primaryKeyExisted) {
      builder
          .addContainedKind(RowKind.UPDATE_BEFORE)
          .addContainedKind(RowKind.UPDATE_AFTER)
          .addContainedKind(RowKind.DELETE);
    }
    return builder.build();
  }
Flink table api that generates sink operators.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/MixedFormatDynamicSink.java