public static Schema fillUpIdentifierFields(
      Schema baseSchema, Schema fromSchema, PrimaryKeySpec primaryKeySpec) {
    int schemaId = fromSchema.schemaId();
    Types.StructType struct = fromSchema.asStruct();
    List<Types.NestedField> fields = Lists.newArrayList(struct.fields());
    Set<Integer> identifierFieldIds = Sets.newHashSet(baseSchema.identifierFieldIds());
    primaryKeySpec.fields().stream()
        .map(PrimaryKeySpec.PrimaryKeyField::fieldName)
        .forEach(p -> identifierFieldIds.add(baseSchema.findField(p).fieldId()));

    identifierFieldIds.forEach(
        fieldId -> {
          if (struct.field(fieldId) == null) {
            fields.add(baseSchema.findField(fieldId));
          }
        });

    return new Schema(schemaId, fields, identifierFieldIds);
  }
Convert an Iceberg Schema {@link Schema} to a {@link Schema} based on the given schema.

<p>This fill-up does not assign new ids; it uses ids from the base schema.

<p>If the fromSchema does contain the identifierFields of the based schema, it will fill-up the
identifierFields to a new schema.

@param baseSchema a Schema on which loading is based
@param fromSchema a Schema on which compared to
@return a new Schema on which contain the identifier fields of the base Schema and column
    fields of the fromSchema
amoro-format-iceberg/src/main/java/org/apache/amoro/utils/SchemaUtil.java