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

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

@@ -213,8 +213,18 @@ 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);
+    }
+
+    @Override
+    public Map<Integer, Accessor<StructLike>> file(
+        Types.FileType file, List<Map<Integer, Accessor<StructLike>>> fieldResults) {
+      return buildAccessors(file.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;
@@ -578,20 +579,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,24 @@ class AssignFreshIds extends TypeUtil.CustomOrderSchemaVisitor<Type> {
     this.nextId = nextId;
   }
 
-  private int idFor(String fullName) {
+  private int idFor(String fullName, Type type) {
+    Integer existingId = baseId(fullName);
+    if (existingId != null) {
+      return existingId;
+    }
+
+    return type.isFileType() ? nextId.get(Types.FileType.NUM_NESTED_FIELDS) : nextId.get();
+  }
+
+  private Integer baseId(String fullName) {
     if (baseSchema != null && fullName != null) {
       Types.NestedField field = baseSchema.findField(fullName);
       if (field != null) {
         return field.fieldId();
       }
     }
 
-    return nextId.get();
+    return null;
   }
 
   private String name(int id) {
@@ -80,15 +89,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 +112,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 +139,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

@@ -45,9 +45,14 @@ public Set<Integer> struct(Types.StructType struct, List<Set<Integer>> fieldResu
     return fieldIds;
   }
 
+  @Override
+  public Set<Integer> file(Types.FileType file, List<Set<Integer>> fieldResults) {
+    return fieldIds;
+  }
+
   @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());

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

@@ -38,6 +38,12 @@ public Map<Integer, Types.NestedField> struct(
     return index;
   }
 
+  @Override
+  public Map<Integer, Types.NestedField> file(
+      Types.FileType file, List<Map<Integer, Types.NestedField>> fieldResults) {
+    return index;
+  }
+
   @Override
   public Map<Integer, Types.NestedField> field(
       Types.NestedField field, Map<Integer, Types.NestedField> fieldResult) {

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

@@ -113,7 +113,7 @@ public void beforeListElement(Types.NestedField elementField) {
     // only add "element" to the short name if the element is not a struct, so that names are more
     // natural
     // for example, locations.latitude instead of locations.element.latitude
-    if (!elementField.type().isStructType()) {
+    if (!hasNestedFields(elementField)) {
       shortFieldNames.push(elementField.name());
     }
   }
@@ -123,7 +123,7 @@ public void afterListElement(Types.NestedField elementField) {
     fieldNames.pop();
 
     // only remove "element" if it was added
-    if (!elementField.type().isStructType()) {
+    if (!hasNestedFields(elementField)) {
       shortFieldNames.pop();
     }
   }
@@ -143,7 +143,7 @@ public void beforeMapValue(Types.NestedField valueField) {
     fieldNames.push(valueField.name());
 
     // only add "value" to the name if the value is not a struct, so that names are more natural
-    if (!valueField.type().isStructType()) {
+    if (!hasNestedFields(valueField)) {
       shortFieldNames.push(valueField.name());
     }
   }
@@ -153,11 +153,15 @@ public void afterMapValue(Types.NestedField valueField) {
     fieldNames.pop();
 
     // only remove "value" if it was added
-    if (!valueField.type().isStructType()) {
+    if (!hasNestedFields(valueField)) {
       shortFieldNames.pop();
     }
   }
 
+  private static boolean hasNestedFields(Types.NestedField field) {
+    return field.type().isStructType() || field.type().isFileType();
+  }
+
   @Override
   public Map<String, Integer> schema(Schema schema, Map<String, Integer> structResult) {
     return nameToId;
@@ -194,6 +198,11 @@ public Map<String, Integer> variant(Types.VariantType variant) {
     return nameToId;
   }
 
+  @Override
+  public Map<String, Integer> file(Types.FileType file, List<Map<String, Integer>> fieldResults) {
+    return nameToId;
+  }
+
   @Override
   public Map<String, Integer> primitive(Type.PrimitiveType primitive) {
     return nameToId;

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

@@ -47,7 +47,16 @@ public Map<Integer, Integer> schema(Schema schema, Map<Integer, Integer> structR
   @Override
   public Map<Integer, Integer> struct(
       Types.StructType struct, List<Map<Integer, Integer>> fieldResults) {
-    for (Types.NestedField field : struct.fields()) {
+    return indexFields(struct.fields());
+  }
+
+  @Override
+  public Map<Integer, Integer> file(Types.FileType file, List<Map<Integer, Integer>> fieldResults) {
+    return indexFields(file.fields());
+  }
+
+  private Map<Integer, Integer> indexFields(List<Types.NestedField> fields) {
+    for (Types.NestedField field : fields) {
       Integer parentId = idStack.peek();
       if (parentId != null) {
         // fields in the root struct are not added