API: Add the file type

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;

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

@@ -48,15 +48,31 @@ class AssignFreshIds extends TypeUtil.CustomOrderSchemaVisitor<Type> {
     this.nextId = nextId;
   }
 
-  private int idFor(String fullName) {
+  private int idFor(int currentId, Type type) {
+    Types.NestedField existingField = baseField(name(currentId));
+    if (canReuseBaseId(existingField, type)) {
+      return existingField.fieldId();
+    }
+
+    return type.isFileType() ? nextId.get(Types.FileType.NUM_NESTED_FIELDS) : nextId.get();
+  }
+
+  private static boolean canReuseBaseId(Types.NestedField baseField, Type type) {
+    if (baseField == null) {
+      return false;
+    }
+
+    // a file derives its nested field IDs from its own ID, so reusing a base ID is only safe when
+    // the base field was already a file and had reserved that block
+    return !type.isFileType() || baseField.type().isFileType();
+  }
+
+  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 +96,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(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.applyAssignedId(field.type(), newId, types.next());
+      newFields.add(Types.NestedField.from(field).withId(newId).ofType(type).build());
     }
 
     return Types.StructType.of(newFields);
@@ -101,22 +119,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(list.elementId(), list.elementType());
+    Type elementType = TypeUtil.applyAssignedId(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(map.keyId(), map.keyType());
+    int newValueId = idFor(map.valueId(), map.valueType());
+    Type keyType = TypeUtil.applyAssignedId(map.keyType(), newKeyId, keyFuture.get());
+    Type valueType = TypeUtil.applyAssignedId(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 +146,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.applyAssignedId(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.applyAssignedId(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.applyAssignedId(map.keyType(), newKeyId, keyFuture.get());
+    Type valueType = TypeUtil.applyAssignedId(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/ReassignDoc.java

@@ -96,6 +96,12 @@ public Type variant(Types.VariantType variant) {
     return variant;
   }
 
+  @Override
+  public Type file(Types.FileType file, Iterable<Type> fieldTypes) {
+    // the nested fields of a file cannot carry docs
+    return file;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     return primitive;

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

@@ -50,7 +50,7 @@ public Type schema(Schema schema, Supplier<Type> future) {
     }
   }
 
-  private int id(Types.StructType sourceStruct, String name) {
+  private int id(Types.StructType sourceStruct, String name, Type type) {
     Types.NestedField sourceField =
         caseSensitive ? sourceStruct.field(name) : sourceStruct.caseInsensitiveField(name);
 
@@ -59,7 +59,7 @@ private int id(Types.StructType sourceStruct, String name) {
     }
 
     if (assignId != null) {
-      return assignId.get();
+      return type.isFileType() ? assignId.get(Types.FileType.NUM_NESTED_FIELDS) : assignId.get();
     }
 
     throw new IllegalArgumentException("Field " + name + " not found in source schema");
@@ -78,8 +78,9 @@ public Type struct(Types.StructType struct, Iterable<Type> fieldTypes) {
     List<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(length);
     for (int i = 0; i < length; i += 1) {
       Types.NestedField field = fields.get(i);
-      int fieldId = id(sourceStruct, field.name());
-      newFields.add(Types.NestedField.from(field).withId(fieldId).ofType(types.get(i)).build());
+      int fieldId = id(sourceStruct, field.name(), field.type());
+      Type type = TypeUtil.applyAssignedId(field.type(), fieldId, types.get(i));
+      newFields.add(Types.NestedField.from(field).withId(fieldId).ofType(type).build());
     }
 
     return Types.StructType.of(newFields);
@@ -120,10 +121,12 @@ public Type list(Types.ListType list, Supplier<Type> elementTypeFuture) {
 
     this.sourceType = sourceList.elementType();
     try {
+      Type elementType =
+          TypeUtil.applyAssignedId(list.elementType(), sourceElementId, elementTypeFuture.get());
       if (list.isElementOptional()) {
-        return Types.ListType.ofOptional(sourceElementId, elementTypeFuture.get());
+        return Types.ListType.ofOptional(sourceElementId, elementType);
       } else {
-        return Types.ListType.ofRequired(sourceElementId, elementTypeFuture.get());
+        return Types.ListType.ofRequired(sourceElementId, elementType);
       }
 
     } finally {
@@ -141,10 +144,11 @@ public Type map(Types.MapType map, Supplier<Type> keyTypeFuture, Supplier<Type>
 
     try {
       this.sourceType = sourceMap.keyType();
-      Type keyType = keyTypeFuture.get();
+      Type keyType = TypeUtil.applyAssignedId(map.keyType(), sourceKeyId, keyTypeFuture.get());
 
       this.sourceType = sourceMap.valueType();
-      Type valueType = valueTypeFuture.get();
+      Type valueType =
+          TypeUtil.applyAssignedId(map.valueType(), sourceValueId, valueTypeFuture.get());
 
       if (map.isValueOptional()) {
         return Types.MapType.ofOptional(sourceKeyId, sourceValueId, keyType, valueType);
@@ -162,6 +166,12 @@ public Type variant(Types.VariantType variant) {
     return variant;
   }
 
+  @Override
+  public Type file(Types.FileType file, Iterable<Type> fieldTypes) {
+    // nested fields are rebuilt from the id assigned to the field that holds this type
+    return file;
+  }
+
   @Override
   public Type primitive(Type.PrimitiveType primitive) {
     return primitive; // nothing to reassign

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

@@ -37,7 +37,16 @@ public Type schema(Schema schema, Type structResult) {
 
   @Override
   public Type struct(Types.StructType struct, List<Type> fieldResults) {
-    List<Types.NestedField> fields = struct.fields();
+    return replaceFieldTypes(struct.fields(), fieldResults, struct);
+  }
+
+  @Override
+  public Type file(Types.FileType file, List<Type> fieldResults) {
+    return replaceFieldTypes(file.fields(), fieldResults, file);
+  }
+
+  private Type replaceFieldTypes(
+      List<Types.NestedField> fields, List<Type> fieldResults, Type unchangedResult) {
     List<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(fields.size());
     boolean hasChanged = false;
 
@@ -56,7 +65,7 @@ public Type struct(Types.StructType struct, List<Type> fieldResults) {
       return Types.StructType.of(newFields);
     }
 
-    return struct;
+    return unchangedResult;
   }
 
   @Override

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

@@ -50,6 +50,7 @@ enum TypeID {
     LIST(List.class),
     MAP(Map.class),
     VARIANT(Variant.class),
+    FILE(StructLike.class),
     UNKNOWN(Object.class);
 
     private final Class<?> javaClass;
@@ -89,6 +90,10 @@ default Types.VariantType asVariantType() {
     throw new IllegalArgumentException("Not a variant type: " + this);
   }
 
+  default Types.FileType asFileType() {
+    throw new IllegalArgumentException("Not a file type: " + this);
+  }
+
   default boolean isNestedType() {
     return false;
   }
@@ -109,6 +114,10 @@ default boolean isVariantType() {
     return false;
   }
 
+  default boolean isFileType() {
+    return false;
+  }
+
   default NestedType asNestedType() {
     throw new IllegalArgumentException("Not a nested type: " + this);
   }