Core: Add v4 manifest reader

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

@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.types;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+class ReplaceTypeById extends TypeUtil.SchemaVisitor<Type> {
+  private final Map<Integer, Type> replacementsById;
+
+  ReplaceTypeById(Map<Integer, Type> replacementsById) {
+    this.replacementsById = replacementsById;
+  }
+
+  @Override
+  public Type schema(Schema schema, Type structResult) {
+    return structResult;
+  }
+
+  @Override
+  public Type struct(Types.StructType struct, List<Type> fieldResults) {
+    List<Types.NestedField> fields = struct.fields();
+    List<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(fields.size());
+    boolean hasChanged = false;
+
+    for (int i = 0; i < fields.size(); i += 1) {
+      Type fieldReplacement = fieldResults.get(i);
+      Types.NestedField field = fields.get(i);
+      if (field.type() != fieldReplacement) {
+        hasChanged = true;
+        newFields.add(Types.NestedField.from(field).ofType(fieldReplacement).build());
+      } else {
+        newFields.add(field);
+      }
+    }
+
+    if (hasChanged) {
+      return Types.StructType.of(newFields);
+    }
+
+    return struct;
+  }
+
+  @Override
+  public Type field(Types.NestedField field, Type fieldResult) {
+    return replacementsById.getOrDefault(field.fieldId(), fieldResult);
+  }
+
+  @Override
+  public Type list(Types.ListType list, Type elementResult) {
+    Type elementReplacement = replacementsById.getOrDefault(list.elementId(), elementResult);
+    if (list.elementType() != elementReplacement) {
+      if (list.isElementRequired()) {
+        return Types.ListType.ofRequired(list.elementId(), elementReplacement);
+      } else {
+        return Types.ListType.ofOptional(list.elementId(), elementReplacement);
+      }
+    }
+
+    return list;
+  }
+
+  @Override
+  public Type map(Types.MapType map, Type keyResult, Type valueResult) {
+    Type keyReplacement = replacementsById.getOrDefault(map.keyId(), keyResult);
+    Type valueReplacement = replacementsById.getOrDefault(map.valueId(), valueResult);
+    if (map.keyType() != keyReplacement || map.valueType() != valueReplacement) {
+      if (map.isValueRequired()) {
+        return Types.MapType.ofRequired(
+            map.keyId(), map.valueId(), keyReplacement, valueReplacement);
+      } else {
+        return Types.MapType.ofOptional(
+            map.keyId(), map.valueId(), keyReplacement, valueReplacement);
+      }
+    }
+
+    return map;
+  }
+
+  @Override
+  public Type variant(Types.VariantType variant) {
+    return variant;
+  }
+
+  @Override
+  public Type primitive(Type.PrimitiveType primitive) {
+    return primitive;
+  }
+}

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

@@ -157,6 +157,24 @@
     return project(schema, projectedIds);
   }
 
+  /**
+   * Returns a copy of the schema with the type of each field in {@code replacementsById} replaced
+   * by its mapped type. Fields not in the map are unchanged.
+   *
+   * @param schema a schema
+   * @param replacementsById a map from field ID to the type that should replace the field's type
+   * @return a schema with the replaced field types
+   */
+  public static Schema replaceFieldTypes(Schema schema, Map<Integer, Type> replacementsById) {
+    Types.StructType struct = visit(schema, new ReplaceTypeById(replacementsById)).asStructType();
+    if (struct.equals(schema.asStruct())) {
+      return schema;
+    }
+
+    return new Schema(
+        schema.schemaId(), struct.fields(), schema.getAliases(), schema.identifierFieldIds());
+  }
+
   public static Schema join(Schema left, Schema right) {
     List<Types.NestedField> joinedColumns = Lists.newArrayList(left.columns());
     for (Types.NestedField rightColumn : right.columns()) {
@@ -457,7 +475,7 @@
       return true;
     }
 
     switch (from.typeId()) {
       case INTEGER:
         return to.typeId() == Type.TypeID.LONG;
 
@@ -561,7 +579,7 @@
   }
 
   private static int estimateSize(Type type) {
     switch (type.typeId()) {
       case BOOLEAN:
         // the size of a boolean variable is virtual machine dependent
         // it is common to believe booleans occupy 1 byte in most JVMs
@@ -745,7 +763,7 @@
   }
 
   public static <T> T visit(Type type, SchemaVisitor<T> visitor) {
     switch (type.typeId()) {
       case STRUCT:
         Types.StructType struct = type.asNestedType().asStructType();
         List<T> results = Lists.newArrayListWithExpectedSize(struct.fields().size());
@@ -885,7 +903,7 @@
    * @return the result of traversing the given type with the visitor
    */
   public static <T> T visit(Type type, CustomOrderSchemaVisitor<T> visitor) {
     switch (type.typeId()) {
       case STRUCT:
         Types.StructType struct = type.asNestedType().asStructType();
         List<VisitFieldFuture<T>> results =

api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java

@@ -29,6 +29,7 @@
 import java.util.stream.Stream;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Types.IntegerType;
@@ -1037,4 +1038,133 @@ public void testIndexStatsNames() {
         .containsEntry(24, "addresses_value") // the leaf takes precedence
         .hasSize(22);
   }
+
+  @Test
+  public void testReplaceFieldTypes() {
+    Types.StructType replacement = Types.StructType.of(required(10, "x", IntegerType.get()));
+    Schema schema =
+        new Schema(
+            required(1, "id", IntegerType.get()),
+            required(2, "s", Types.StructType.of(required(3, "a", Types.LongType.get()))));
+
+    Schema result = TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(2, replacement));
+
+    assertThat(result.findField(1).type()).isEqualTo(IntegerType.get());
+    assertThat(result.findField(2).type()).isEqualTo(replacement);
+  }
+
+  @Test
+  public void testReplaceFieldTypesPrimitive() {
+    Schema schema =
+        new Schema(required(1, "id", IntegerType.get()), required(2, "count", IntegerType.get()));
+
+    Schema result = TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(2, Types.LongType.get()));
+
+    assertThat(result.findField(1).type()).isEqualTo(IntegerType.get());
+    assertThat(result.findField(2).type()).isEqualTo(Types.LongType.get());
+  }
+
+  @Test
+  public void testReplaceFieldTypesListElement() {
+    Schema schema =
+        new Schema(required(1, "list", Types.ListType.ofRequired(2, Types.StructType.of())));
+
+    Schema result =
+        TypeUtil.replaceFieldTypes(
+            schema, ImmutableMap.of(2, Types.StructType.of(required(3, "x", IntegerType.get()))));
+
+    Types.ListType list = (Types.ListType) result.findField(1).type();
+    assertThat(list.elementType().asStructType().field(3).name()).isEqualTo("x");
+    assertThat(list.isElementRequired()).isTrue();
+  }
+
+  @Test
+  public void testReplaceFieldTypesMapKeyAndValue() {
+    Schema schema =
+        new Schema(
+            required(1, "m", Types.MapType.ofRequired(2, 3, IntegerType.get(), IntegerType.get())));
+
+    Schema result =
+        TypeUtil.replaceFieldTypes(
+            schema, ImmutableMap.of(2, Types.LongType.get(), 3, Types.StringType.get()));
+
+    Types.MapType map = (Types.MapType) result.findField(1).type();
+    assertThat(map.keyType()).isEqualTo(Types.LongType.get());
+    assertThat(map.valueType()).isEqualTo(Types.StringType.get());
+    assertThat(map.isValueRequired()).isTrue();
+  }
+
+  @Test
+  public void testReplaceFieldTypesPreservesDefaultsOnCompatibleType() {
+    Types.NestedField field =
+        Types.NestedField.optional("count")
+            .withId(1)
+            .ofType(IntegerType.get())
+            .withInitialDefault(Literal.of(5))
+            .withWriteDefault(Literal.of(7))
+            .build();
+    Schema schema = new Schema(field);
+
+    Schema result = TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(1, Types.LongType.get()));
+
+    Types.NestedField replaced = result.findField(1);
+    assertThat(replaced.type()).isEqualTo(Types.LongType.get());
+    assertThat(replaced.initialDefault()).isEqualTo(5L);
+    assertThat(replaced.writeDefault()).isEqualTo(7L);
+  }
+
+  @Test
+  public void testReplaceFieldTypesRejectsDefaultIncompatibleWithType() {
+    Types.NestedField field =
+        Types.NestedField.optional("count")
+            .withId(1)
+            .ofType(IntegerType.get())
+            .withInitialDefault(Literal.of(5))
+            .build();
+    Schema schema = new Schema(field);
+
+    assertThatThrownBy(
+            () -> TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(1, Types.StringType.get())))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Cannot cast default value to string");
+  }
+
+  @Test
+  public void testReplaceFieldTypesPreservesIdentifierField() {
+    Schema schema =
+        new Schema(
+            Lists.newArrayList(
+                required(1, "id", IntegerType.get()), required(2, "data", Types.StringType.get())),
+            Sets.newHashSet(1));
+
+    Schema result = TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(1, Types.LongType.get()));
+
+    assertThat(result.findField(1).type()).isEqualTo(Types.LongType.get());
+    assertThat(result.identifierFieldIds()).containsExactly(1);
+  }
+
+  @Test
+  public void testReplaceFieldTypesRejectsNonPrimitiveIdentifierField() {
+    Schema schema =
+        new Schema(
+            Lists.newArrayList(
+                required(1, "id", IntegerType.get()), required(2, "data", Types.StringType.get())),
+            Sets.newHashSet(1));
+
+    // an identifier field must be primitive, so replacing it with a struct is rejected
+    assertThatThrownBy(
+            () ->
+                TypeUtil.replaceFieldTypes(
+                    schema,
+                    ImmutableMap.of(1, Types.StructType.of(required(3, "x", IntegerType.get())))))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("not a primitive type field");
+  }
+
+  @Test
+  public void testReplaceFieldTypesNoMatchReturnsSameSchema() {
+    Schema schema = new Schema(required(1, "id", IntegerType.get()));
+    Schema result = TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(99, Types.LongType.get()));
+    assertThat(result).isSameAs(schema);
+  }
 }

core/src/main/java/org/apache/iceberg/Partitioning.java

@@ -243,6 +243,17 @@ public static StructType partitionType(Table table) {
         "table partition", specs, allActiveFieldIds(table.schema(), specs));
   }
 
+  /**
+   * Builds a unified partition type containing all partition fields from the given specs, including
+   * fields whose source columns are no longer present in the table schema.
+   *
+   * @param specs the partition specs to unify
+   * @return the constructed unified partition type
+   */
+  static StructType unionPartitionTypes(Collection<PartitionSpec> specs) {
+    return buildPartitionProjectionType("union partition", specs, allFieldIds(specs));
+  }
+
   /**
    * Checks if any of the specs in a table is partitioned.
    *
@@ -347,6 +358,14 @@ private static boolean compatibleTransforms(Transform<?, ?> t1, Transform<?, ?>
         || t2.equals(Transforms.alwaysNull());
   }
 
+  // collects IDs of all partition fields used across specs
+  private static Set<Integer> allFieldIds(Collection<PartitionSpec> specs) {
+    return FluentIterable.from(specs)
+        .transformAndConcat(PartitionSpec::fields)
+        .transform(PartitionField::fieldId)
+        .toSet();
+  }
+
   // collects IDs of all partition field used across specs that are in the current schema
   private static Set<Integer> allActiveFieldIds(Schema schema, Collection<PartitionSpec> specs) {
     return FluentIterable.from(specs)

core/src/main/java/org/apache/iceberg/TrackedFile.java

@@ -22,6 +22,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
+import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 
 /** A file tracked by a manifest. */
@@ -95,9 +96,14 @@ interface TrackedFile {
           Types.ListType.ofRequired(136, Types.IntegerType.get()),
           "Field ids used to determine row equality in equality delete files");
 
-  static Types.StructType schemaWithContentStats(
-      Types.StructType partitionType, Types.StructType contentStatsType) {
-    return Types.StructType.of(
+  /**
+   * Returns the schema for the given partition and content stats types.
+   *
+   * <p>The partition and content stats fields use {@link Types.UnknownType} when their types have
+   * no fields, so that they are not stored in manifest files.
+   */
+  static Schema schema(Types.StructType partitionType, Types.StructType contentStatsType) {
+    return new Schema(
         TRACKING,
         CONTENT_TYPE,
         FORMAT_VERSION,
@@ -106,9 +112,13 @@ static Types.StructType schemaWithContentStats(
         RECORD_COUNT,
         FILE_SIZE_IN_BYTES,
         SPEC_ID,
-        Types.NestedField.optional(PARTITION_ID, PARTITION_NAME, partitionType, PARTITION_DOC),
         Types.NestedField.optional(
-            CONTENT_STATS_ID, CONTENT_STATS_NAME, contentStatsType, CONTENT_STATS_DOC),
+            PARTITION_ID, PARTITION_NAME, typeOrUnknown(partitionType), PARTITION_DOC),
+        Types.NestedField.optional(
+            CONTENT_STATS_ID,
+            CONTENT_STATS_NAME,
+            typeOrUnknown(contentStatsType),
+            CONTENT_STATS_DOC),
         SORT_ORDER_ID,
         DELETION_VECTOR,
         MANIFEST_INFO,
@@ -117,6 +127,10 @@ static Types.StructType schemaWithContentStats(
         EQUALITY_IDS);
   }
 
+  private static Type typeOrUnknown(Types.StructType structType) {
+    return structType.fields().isEmpty() ? Types.UnknownType.get() : structType;
+  }
+
   /** Returns the tracking information for this entry. */
   Tracking tracking();
 

core/src/main/java/org/apache/iceberg/TrackedFileStruct.java

@@ -83,10 +83,11 @@ class TrackedFileStruct extends SupportsIndexProjection implements TrackedFile,
   /** Used by internal readers to instantiate this class with a projection schema. */
   TrackedFileStruct(Types.StructType projection) {
     super(BASE_TYPE, projection);
-    // partition type may be null if the field was not projected
+    // partition type may be null if the field was not projected, or unknown for unpartitioned
+    // manifests
     Type partType = projection.fieldType("partition");
-    if (partType != null) {
-      this.partitionData = new PartitionData(partType.asNestedType().asStructType());
+    if (partType != null && partType.isStructType()) {
+      this.partitionData = new PartitionData(partType.asStructType());
     }
   }
 
@@ -101,10 +102,10 @@ class TrackedFileStruct extends SupportsIndexProjection implements TrackedFile,
       int formatVersion,
       String location,
       FileFormat fileFormat,
-      PartitionData partition,
       long recordCount,
       long fileSizeInBytes,
       Integer specId,
+      PartitionData partition,
       ContentStats contentStats,
       Integer sortOrderId,
       DeletionVector deletionVector,
@@ -120,9 +121,8 @@ class TrackedFileStruct extends SupportsIndexProjection implements TrackedFile,
     this.fileFormat = fileFormat;
     this.recordCount = recordCount;
     this.fileSizeInBytes = fileSizeInBytes;
-    this.partitionData = partition;
-
     this.specId = specId;
+    this.partitionData = partition;
     this.contentStats = contentStats;
     this.sortOrderId = sortOrderId;
     this.deletionVector = deletionVector;