@Override
  public void upsert(Iterator<RowData> dataStream) throws IOException {
    while (dataStream.hasNext()) {
      RowData value = dataStream.next();
      if (filter(value)) {
        continue;
      }
      RowData key = new KeyRowData(uniqueKeyIndexMapping, value);

      if (value.getRowKind() == RowKind.INSERT || value.getRowKind() == RowKind.UPDATE_AFTER) {
        recordState.put(key, value);
      } else {
        recordState.delete(key);
      }
    }
  }
Use a unique index to lookup. Working for the situation where the join keys include the
mixed-format table's primary keys.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/lookup/UniqueIndexTable.java