@Deprecated
  public static List<Types.NestedField> addPrimaryKey(
      List<Types.NestedField> projectedColumns, MixedTable table) {
    List<String> primaryKeys =
        table.isUnkeyedTable()
            ? Collections.EMPTY_LIST
            : table.asKeyedTable().primaryKeySpec().fields().stream()
                .map(PrimaryKeySpec.PrimaryKeyField::fieldName)
                .collect(Collectors.toList());

    List<Types.NestedField> columns = new ArrayList<>(projectedColumns);
    Set<String> projectedNames = new HashSet<>();

    projectedColumns.forEach(c -> projectedNames.add(c.name()));

    primaryKeys.forEach(
        pk -> {
          if (!projectedNames.contains(pk)) {
            columns.add(table.schema().findField(pk));
          }
        });

    LOG.info("Projected Columns after addPrimaryKey, columns:{}", columns);
    return columns;
  }
Primary keys are the required fields to guarantee that readers can read keyed table in right
order, due to the automatic scaling in/out of nodes. The required fields should be added even
though projection push down
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/FlinkSchemaUtil.java