private static Schema fileProjection(
      Schema tableSchema, Schema requestedSchema, boolean hasPosDelete, Set<Integer> eqDeleteIds) {
    if (!hasPosDelete && eqDeleteIds == null) {
      return requestedSchema;
    }
    List<Integer> requiredEqDeleteIds =
        TypeUtil.select(tableSchema, eqDeleteIds).columns().stream()
            .map(Types.NestedField::fieldId)
            .collect(Collectors.toList());
    Set<Integer> requiredIds = Sets.newLinkedHashSet();
    if (hasPosDelete) {
      requiredIds.add(MetadataColumns.FILE_PATH.fieldId());
      requiredIds.add(MetadataColumns.ROW_POSITION.fieldId());
    }

    requiredIds.addAll(requiredEqDeleteIds);
    requiredIds.add(org.apache.amoro.table.MetadataColumns.TRANSACTION_ID_FILED.fieldId());

    requiredIds.add(MetadataColumns.IS_DELETED.fieldId());

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

    if (missingIds.isEmpty()) {
      return requestedSchema;
    }

    List<Types.NestedField> columns = Lists.newArrayList(requestedSchema.columns());
    for (int fieldId : missingIds) {
      if (fieldId == MetadataColumns.ROW_POSITION.fieldId()
          || fieldId == MetadataColumns.IS_DELETED.fieldId()
          || fieldId == org.apache.amoro.table.MetadataColumns.TRANSACTION_ID_FILED.fieldId()
          || fieldId == MetadataColumns.FILE_PATH.fieldId()) {
        continue; // add _pos and _deleted at the end
      }

      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(MetadataColumns.FILE_PATH.fieldId())) {
      columns.add(MetadataColumns.FILE_PATH);
    }

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

    if (missingIds.contains(
        org.apache.amoro.table.MetadataColumns.TRANSACTION_ID_FILED.fieldId())) {
      columns.add(org.apache.amoro.table.MetadataColumns.TRANSACTION_ID_FILED);
    }

    if (missingIds.contains(MetadataColumns.IS_DELETED.fieldId())) {
      columns.add(MetadataColumns.IS_DELETED);
    }

    return new Schema(columns);
  }
Read data by {@link CombinedIcebergScanTask} for optimizer of native iceberg. During the
execution of readData and readDeleteData, delete data is read only once, thereby improving
performance.
amoro-format-iceberg/src/main/java/org/apache/amoro/io/reader/GenericCombinedIcebergDataReader.java