API, Core, Parquet: Model the file type as its own nested type

api/src/main/java/org/apache/iceberg/Accessors.java

@@ -213,8 +213,12 @@ public Map<Integer, Accessor<StructLike>> schema(
     @Override
     public Map<Integer, Accessor<StructLike>> struct(
         Types.StructType struct, List<Map<Integer, Accessor<StructLike>>> fieldResults) {
+      return buildAccessors(struct.fields(), fieldResults);
+    }
+
+    private Map<Integer, Accessor<StructLike>> buildAccessors(
+        List<Types.NestedField> fields, List<Map<Integer, Accessor<StructLike>>> fieldResults) {
       Map<Integer, Accessor<StructLike>> accessors = Maps.newHashMap();
-      List<Types.NestedField> fields = struct.fields();
       for (int i = 0; i < fieldResults.size(); i += 1) {
         Types.NestedField field = fields.get(i);
         Map<Integer, Accessor<StructLike>> result = fieldResults.get(i);

api/src/main/java/org/apache/iceberg/Schema.java

@@ -67,7 +67,8 @@ public class Schema implements Serializable {
           Type.TypeID.VARIANT, 3,
           Type.TypeID.UNKNOWN, 3,
           Type.TypeID.GEOMETRY, 3,
-          Type.TypeID.GEOGRAPHY, 3);
+          Type.TypeID.GEOGRAPHY, 3,
+          Type.TypeID.FILE, 4);
 
   private final StructType struct;
   private final int schemaId;
@@ -148,6 +149,8 @@ public Schema(
     this.struct = StructType.of(finalColumns);
     this.aliasToId = aliases != null ? ImmutableBiMap.copyOf(aliases) : null;
 
+    validateFileFields(struct);
+
     // validate IdentifierField
     if (identifierFieldIds != null) {
       Map<Integer, Integer> idToParent = TypeUtil.indexParents(struct);
@@ -160,6 +163,58 @@ public Schema(
     this.highestFieldId = lazyIdToName().keySet().stream().mapToInt(i -> i).max().orElse(0);
   }
 
+  /**
+   * Validates that every file column holds the block of derived field IDs that its type produces.
+   */
+  private static void validateFileFields(StructType struct) {
+    Map<Integer, String> namesById = Maps.newHashMap();
+    Map<Integer, String> fileNamesById = Maps.newHashMap();
+    indexFileFields(struct, null, namesById, fileNamesById);
+
+    fileNamesById.forEach(
+        (enclosingId, fileName) -> {
+          for (int offset = 1; offset <= Types.FileType.NUM_NESTED_FIELDS; offset += 1) {
+            String conflictingName = namesById.get(enclosingId + offset);
+            Preconditions.checkArgument(
+                conflictingName == null,
+                "Invalid file column %s: derived field ID %s is already used by %s",
+                fileName,
+                enclosingId + offset,
+                conflictingName);
+          }
+        });
+  }
+
+  private static void indexFileFields(
+      Type type,
+      String prefix,
+      Map<Integer, String> namesById,
+      Map<Integer, String> fileNamesById) {
+    if (!type.isNestedType()) {
+      return;
+    }
+
+    for (NestedField field : type.asNestedType().fields()) {
+      String name = prefix == null ? field.name() : prefix + "." + field.name();
+      namesById.putIfAbsent(field.fieldId(), name);
+
+      // a file's derived fields are not indexed, so any ID found in its reserved block is a
+      // different column
+      if (field.type().isFileType()) {
+        Types.FileType file = field.type().asFileType();
+        Preconditions.checkArgument(
+            file.enclosingId() == field.fieldId(),
+            "Invalid file column %s: nested field IDs are derived from %s, not %s",
+            name,
+            field.fieldId(),
+            file.enclosingId());
+        fileNamesById.put(field.fieldId(), name);
+      } else {
+        indexFileFields(field.type(), name, namesById, fileNamesById);
+      }
+    }
+  }
+
   static void validateIdentifierField(
       int fieldId, Map<Integer, Types.NestedField> idToField, Map<Integer, Integer> idToParent) {
     Types.NestedField field = idToField.get(fieldId);
@@ -578,20 +633,33 @@ private List<NestedField> reassignIds(List<NestedField> columns, TypeUtil.GetID
     if (getID == null) {
       return columns;
     }
-    Type res =
-        TypeUtil.assignIds(
-            StructType.of(columns),
-            oldId -> {
-              int newId = getID.get(oldId);
-              if (newId != oldId) {
-                idsToReassigned.put(oldId, newId);
-                idsToOriginal.put(newId, oldId);
-              }
-              return newId;
-            });
+
+    TypeUtil.GetID tracked =
+        new TypeUtil.GetID() {
+          @Override
+          public int get(int oldId) {
+            return track(oldId, getID.get(oldId));
+          }
+
+          @Override
+          public int get(int oldId, int numReserved) {
+            return track(oldId, getID.get(oldId, numReserved));
+          }
+        };
+
+    Type res = TypeUtil.assignIds(StructType.of(columns), tracked);
     return res.asStructType().fields();
   }
 
+  private int track(int oldId, int newId) {
+    if (newId != oldId) {
+      idsToReassigned.put(oldId, newId);
+      idsToOriginal.put(newId, oldId);
+    }
+
+    return newId;
+  }
+
   /**
    * Check the compatibility of the schema with a format version.
    *

api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java

@@ -48,15 +48,22 @@ class AssignFreshIds extends TypeUtil.CustomOrderSchemaVisitor<Type> {
     this.nextId = nextId;
   }
 
-  private int idFor(String fullName) {
+  private int idFor(String fullName, Type type) {
+    Types.NestedField existingField = baseField(fullName);
+    // a base ID can only be reused for a file if the base field already reserved the derived IDs
+    if (existingField != null && (!type.isFileType() || existingField.type().isFileType())) {
+      return existingField.fieldId();
+    }
+
+    return type.isFileType() ? nextId.get(Types.FileType.NUM_NESTED_FIELDS) : nextId.get();
+  }
+
+  private Types.NestedField baseField(String fullName) {
     if (baseSchema != null && fullName != null) {
-      Types.NestedField field = baseSchema.findField(fullName);
-      if (field != null) {
-        return field.fieldId();
-      }
+      return baseSchema.findField(fullName);
     }
 
-    return nextId.get();
+    return null;
   }
 
   private String name(int id) {
@@ -80,15 +87,17 @@ public Type struct(Types.StructType struct, Iterable<Type> futures) {
     // assign IDs for this struct's fields first
     List<Integer> newIds = Lists.newArrayListWithExpectedSize(length);
     for (int i = 0; i < length; i += 1) {
-      newIds.add(idFor(name(fields.get(i).fieldId())));
+      Types.NestedField field = fields.get(i);
+      newIds.add(idFor(name(field.fieldId()), field.type()));
     }
 
     List<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(length);
     Iterator<Type> types = futures.iterator();
     for (int i = 0; i < length; i += 1) {
       Types.NestedField field = fields.get(i);
-      Type type = types.next();
-      newFields.add(Types.NestedField.from(field).withId(newIds.get(i)).ofType(type).build());
+      int newId = newIds.get(i);
+      Type type = TypeUtil.assignedType(field.type(), newId, types.next());
+      newFields.add(Types.NestedField.from(field).withId(newId).ofType(type).build());
     }
 
     return Types.StructType.of(newFields);
@@ -101,22 +110,25 @@ public Type field(Types.NestedField field, Supplier<Type> future) {
 
   @Override
   public Type list(Types.ListType list, Supplier<Type> future) {
-    int newId = idFor(name(list.elementId()));
+    int newId = idFor(name(list.elementId()), list.elementType());
+    Type elementType = TypeUtil.assignedType(list.elementType(), newId, future.get());
     if (list.isElementOptional()) {
-      return Types.ListType.ofOptional(newId, future.get());
+      return Types.ListType.ofOptional(newId, elementType);
     } else {
-      return Types.ListType.ofRequired(newId, future.get());
+      return Types.ListType.ofRequired(newId, elementType);
     }
   }
 
   @Override
   public Type map(Types.MapType map, Supplier<Type> keyFuture, Supplier<Type> valueFuture) {
-    int newKeyId = idFor(name(map.keyId()));
-    int newValueId = idFor(name(map.valueId()));
+    int newKeyId = idFor(name(map.keyId()), map.keyType());
+    int newValueId = idFor(name(map.valueId()), map.valueType());
+    Type keyType = TypeUtil.assignedType(map.keyType(), newKeyId, keyFuture.get());
+    Type valueType = TypeUtil.assignedType(map.valueType(), newValueId, valueFuture.get());
     if (map.isValueOptional()) {
-      return Types.MapType.ofOptional(newKeyId, newValueId, keyFuture.get(), valueFuture.get());
+      return Types.MapType.ofOptional(newKeyId, newValueId, keyType, valueType);
     } else {
-      return Types.MapType.ofRequired(newKeyId, newValueId, keyFuture.get(), valueFuture.get());
+      return Types.MapType.ofRequired(newKeyId, newValueId, keyType, valueType);
     }
   }
 
@@ -125,6 +137,12 @@ public Type variant(Types.VariantType variant) {
     return variant;
   }
 
+  @Override
+  public Type file(Types.FileType file, Iterable<Type> futures) {
+    // nested fields are rebuilt from the new id assigned to the field that holds this type
+    return file;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     return primitive;

api/src/main/java/org/apache/iceberg/types/AssignIds.java

@@ -31,8 +31,8 @@ class AssignIds extends TypeUtil.CustomOrderSchemaVisitor<Type> {
     this.getID = getID;
   }
 
-  private int idFor(int id) {
-    return getID.get(id);
+  private int idFor(int id, Type type) {
+    return type.isFileType() ? getID.get(id, Types.FileType.NUM_NESTED_FIELDS) : getID.get(id);
   }
 
   @Override
@@ -48,15 +48,16 @@ public Type struct(Types.StructType struct, Iterable<Type> futures) {
     // assign IDs for this struct's fields first
     List<Integer> newIds = Lists.newArrayListWithExpectedSize(length);
     for (Types.NestedField field : fields) {
-      newIds.add(idFor(field.fieldId()));
+      newIds.add(idFor(field.fieldId(), field.type()));
     }
 
     List<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(length);
     Iterator<Type> types = futures.iterator();
     for (int i = 0; i < length; i += 1) {
       Types.NestedField field = fields.get(i);
-      Type type = types.next();
-      newFields.add(Types.NestedField.from(field).withId(newIds.get(i)).ofType(type).build());
+      int newId = newIds.get(i);
+      Type type = TypeUtil.assignedType(field.type(), newId, types.next());
+      newFields.add(Types.NestedField.from(field).withId(newId).ofType(type).build());
     }
 
     return Types.StructType.of(newFields);
@@ -69,22 +70,25 @@ public Type field(Types.NestedField field, Supplier<Type> future) {
 
   @Override
   public Type list(Types.ListType list, Supplier<Type> future) {
-    int newId = idFor(list.elementId());
+    int newId = idFor(list.elementId(), list.elementType());
+    Type elementType = TypeUtil.assignedType(list.elementType(), newId, future.get());
     if (list.isElementOptional()) {
-      return Types.ListType.ofOptional(newId, future.get());
+      return Types.ListType.ofOptional(newId, elementType);
     } else {
-      return Types.ListType.ofRequired(newId, future.get());
+      return Types.ListType.ofRequired(newId, elementType);
     }
   }
 
   @Override
   public Type map(Types.MapType map, Supplier<Type> keyFuture, Supplier<Type> valueFuture) {
-    int newKeyId = idFor(map.keyId());
-    int newValueId = idFor(map.valueId());
+    int newKeyId = idFor(map.keyId(), map.keyType());
+    int newValueId = idFor(map.valueId(), map.valueType());
+    Type keyType = TypeUtil.assignedType(map.keyType(), newKeyId, keyFuture.get());
+    Type valueType = TypeUtil.assignedType(map.valueType(), newValueId, valueFuture.get());
     if (map.isValueOptional()) {
-      return Types.MapType.ofOptional(newKeyId, newValueId, keyFuture.get(), valueFuture.get());
+      return Types.MapType.ofOptional(newKeyId, newValueId, keyType, valueType);
     } else {
-      return Types.MapType.ofRequired(newKeyId, newValueId, keyFuture.get(), valueFuture.get());
+      return Types.MapType.ofRequired(newKeyId, newValueId, keyType, valueType);
     }
   }
 
@@ -93,6 +97,12 @@ public Type variant(Types.VariantType variant) {
     return variant;
   }
 
+  @Override
+  public Type file(Types.FileType file, Iterable<Type> futures) {
+    // nested fields are rebuilt from the new id assigned to the field that holds this type
+    return file;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     return primitive;

api/src/main/java/org/apache/iceberg/types/CheckCompatibility.java

@@ -260,6 +260,16 @@ public List<String> variant(Types.VariantType readVariant) {
     return ImmutableList.of(String.format(": %s cannot be read as a %s", currentType, readVariant));
   }
 
+  @Override
+  public List<String> file(Types.FileType readFile, Iterable<List<String>> fieldErrorLists) {
+    if (currentType.isFileType()) {
+      // the nested fields are derived from the enclosing id, so matching ids means matching fields
+      return NO_ERRORS;
+    }
+
+    return ImmutableList.of(String.format(": %s cannot be read as a %s", currentType, readFile));
+  }
+
   @Override
   public List<String> primitive(Type.PrimitiveType readPrimitive) {
     if (currentType.equals(readPrimitive)) {

api/src/main/java/org/apache/iceberg/types/Comparators.java

@@ -83,6 +83,8 @@ private static <T> Comparator<T> internal(Type type) {
       return forType(type.asPrimitiveType());
     } else if (type.isStructType()) {
       return (Comparator<T>) forType(type.asStructType());
+    } else if (type.isFileType()) {
+      return (Comparator<T>) forType(type.asFileType().asStruct());
     } else if (type.isListType()) {
       return (Comparator<T>) forType(type.asListType());
     } else if (type.isMapType()) {

api/src/main/java/org/apache/iceberg/types/FindTypeVisitor.java

@@ -85,6 +85,21 @@ public Type variant(Types.VariantType variant) {
     return null;
   }
 
+  @Override
+  public Type file(Types.FileType file, List<Type> fieldResults) {
+    if (predicate.test(file)) {
+      return file;
+    }
+
+    for (Type fieldType : fieldResults) {
+      if (fieldType != null) {
+        return fieldType;
+      }
+    }
+
+    return null;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     if (predicate.test(primitive)) {

api/src/main/java/org/apache/iceberg/types/GetProjectedIds.java

@@ -47,7 +47,7 @@ public Set<Integer> struct(Types.StructType struct, List<Set<Integer>> fieldResu
 
   @Override
   public Set<Integer> field(Types.NestedField field, Set<Integer> fieldResult) {
-    if ((includeStructIds && field.type().isStructType())
+    if ((includeStructIds && (field.type().isStructType() || field.type().isFileType()))
         || field.type().isPrimitiveType()
         || field.type().isVariantType()) {
       fieldIds.add(field.fieldId());