private Schema fileProjection(
      Schema tableSchema,
      Schema requestedSchema,
      Collection<PrimaryKeyedFile> eqDeletes,
      Collection<DeleteFile> posDeletes) {
    if (eqDeletes.isEmpty() && posDeletes.isEmpty()) {
      return requestedSchema;
    }

    Set<Integer> requiredIds = Sets.newLinkedHashSet();
    if (!posDeletes.isEmpty()) {
      requiredIds.add(org.apache.iceberg.MetadataColumns.FILE_PATH.fieldId());
      requiredIds.add(org.apache.iceberg.MetadataColumns.ROW_POSITION.fieldId());
    }

    if (!eqDeletes.isEmpty()) {
      requiredIds.addAll(primaryKeyId);
      requiredIds.add(MetadataColumns.TRANSACTION_ID_FILED.fieldId());
      requiredIds.add(MetadataColumns.FILE_OFFSET_FILED.fieldId());
    }

    Set<Integer> missingIds =
        Sets.newLinkedHashSet(
            Sets.difference(requiredIds, TypeUtil.getProjectedIds(requestedSchema)));

    // TODO: support adding nested columns. this will currently fail when finding nested columns to
    // add
    List<Types.NestedField> columns = Lists.newArrayList(requestedSchema.columns());
    for (int fieldId : missingIds) {
      if (fieldId == org.apache.iceberg.MetadataColumns.ROW_POSITION.fieldId()
          || fieldId == org.apache.iceberg.MetadataColumns.FILE_PATH.fieldId()
          || fieldId == MetadataColumns.TRANSACTION_ID_FILED.fieldId()
          || fieldId == MetadataColumns.FILE_OFFSET_FILED.fieldId()) {
        continue;
      }

      Types.NestedField field = tableSchema.asStruct().field(fieldId);
      Preconditions.checkArgument(field != null, "Cannot find required field for ID %s", fieldId);

      columns.add(field);
    }

    if (missingIds.contains(org.apache.iceberg.MetadataColumns.FILE_PATH.fieldId())) {
      columns.add(org.apache.iceberg.MetadataColumns.FILE_PATH);
    }

    if (missingIds.contains(org.apache.iceberg.MetadataColumns.ROW_POSITION.fieldId())) {
      columns.add(org.apache.iceberg.MetadataColumns.ROW_POSITION);
    }

    // add lsn
    if (missingIds.contains(MetadataColumns.TRANSACTION_ID_FILED.fieldId())) {
      columns.add(MetadataColumns.TRANSACTION_ID_FILED);
    }
    if (missingIds.contains(MetadataColumns.FILE_OFFSET_FILED.fieldId())) {
      columns.add(MetadataColumns.FILE_OFFSET_FILED);
    }

    return new Schema(columns);
  }
@return The data in equity delete file
amoro-format-iceberg/src/main/java/org/apache/amoro/io/reader/MixedDeleteFilter.java