Core: Add v4 manifest reader

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("table 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,7 +96,13 @@ interface TrackedFile {
           Types.ListType.ofRequired(136, Types.IntegerType.get()),
           "Field ids used to determine row equality in equality delete files");
 
-  static Types.StructType schemaWithContentStats(
+  /**
+   * 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 Types.StructType schema(
       Types.StructType partitionType, Types.StructType contentStatsType) {
     return Types.StructType.of(
         TRACKING,
@@ -106,9 +113,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 +128,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());
     }
   }
 

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

@@ -0,0 +1,272 @@
+/*
+ * 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;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.expressions.Evaluator;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Projections;
+import org.apache.iceberg.io.CloseableGroup;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.metrics.ScanMetrics;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructProjection;
+
+/** Reader that reads a v4+ manifest file as {@link TrackedFile}s. */
+class V4ManifestReader extends CloseableGroup implements CloseableIterable<TrackedFile> {
+  // tracking fields read on the scan path; row_position backs Tracking.manifestPos
+  private static final Types.StructType SCAN_TRACKING =
+      Types.StructType.of(
+          Tracking.STATUS,
+          Tracking.SNAPSHOT_ID,
+          Tracking.SEQUENCE_NUMBER,
+          Tracking.FILE_SEQUENCE_NUMBER,
+          Tracking.FIRST_ROW_ID,
+          MetadataColumns.ROW_POSITION);
+
+  private final InputFile file;
+  private final Schema readSchema;
+  private final boolean onlyLive;
+  private final ScanMetrics scanMetrics;
+
+  // partition pruning state, keyed by spec ID
+  private final Map<Integer, Evaluator> partitionEvaluators;
+  private final Map<Integer, StructProjection> partitionProjections;
+
+  private V4ManifestReader(
+      InputFile file,
+      Schema readSchema,
+      Map<Integer, Evaluator> partitionEvaluators,
+      Map<Integer, StructProjection> partitionProjections,
+      boolean onlyLive,
+      ScanMetrics scanMetrics) {
+    this.file = file;
+    this.readSchema = readSchema;
+    this.partitionEvaluators = partitionEvaluators;
+    this.partitionProjections = partitionProjections;
+    this.onlyLive = onlyLive;
+    this.scanMetrics = scanMetrics;
+  }
+
+  static Builder builder(InputFile file, Map<Integer, PartitionSpec> specsById) {
+    return new Builder(file, specsById);
+  }
+
+  /** Returns copies of the tracked files that match this reader's configured filters. */
+  @Override
+  public CloseableIterator<TrackedFile> iterator() {
+    CloseableIterable<TrackedFile> entries = CloseableIterable.transform(open(), this::prepare);
+    if (!partitionEvaluators.isEmpty()) {
+      // manifest references are expanded later and are not pruned by the partition filter
+      entries =
+          CloseableIterable.filter(entries, entry -> isManifest(entry) || matchesPartition(entry));
+    }
+
+    if (onlyLive) {
+      entries = CloseableIterable.filter(entries, entry -> entry.tracking().isLive());
+    }
+
+    return CloseableIterable.transform(entries, TrackedFile::copy).iterator();
+  }
+
+  private static boolean isManifest(TrackedFile trackedFile) {
+    FileContent content = trackedFile.contentType();
+    return content == FileContent.DATA_MANIFEST || content == FileContent.DELETE_MANIFEST;
+  }
+
+  private boolean matchesPartition(TrackedFile trackedFile) {
+    Integer specId = trackedFile.specId();
+    Evaluator evaluator = specId != null ? partitionEvaluators.get(specId) : null;
+    StructProjection projection = specId != null ? partitionProjections.get(specId) : null;
+    Preconditions.checkState(
+        evaluator != null && projection != null,
+        "Cannot apply partition filter: spec ID %s is not one of the known specs %s in manifest %s",
+        specId,
+        partitionEvaluators.keySet(),
+        file.location());
+
+    boolean matches = evaluator.eval(projection.wrap(trackedFile.partition()));
+    if (!matches) {
+      incrementSkipCount(trackedFile.contentType());
+    }
+
+    return matches;
+  }
+
+  private void incrementSkipCount(FileContent content) {
+    switch (content) {
+      case DATA:
+        scanMetrics.skippedDataFiles().increment();
+        break;
+      case EQUALITY_DELETES:
+        scanMetrics.skippedDeleteFiles().increment();
+        break;
+      case DATA_MANIFEST:
+        scanMetrics.skippedDataManifests().increment();
+        break;
+      case DELETE_MANIFEST:
+        scanMetrics.skippedDeleteManifests().increment();
+        break;
+      default:
+        throw new UnsupportedOperationException("Unsupported content type: " + content);
+    }
+  }
+
+  private CloseableIterable<TrackedFile> open() {
+    FileFormat format = FileFormat.fromFileName(file.location());
+    Preconditions.checkArgument(
+        format != null, "Cannot determine format of manifest: %s", file.location());
+
+    CloseableIterable<TrackedFile> reader =
+        InternalData.read(format, file)
+            .project(readSchema)
+            .setRootType(TrackedFileStruct.class)
+            .setCustomType(TrackedFile.TRACKING.fieldId(), TrackingStruct.class)
+            .setCustomType(TrackedFile.DELETION_VECTOR.fieldId(), DeletionVectorStruct.class)
+            .setCustomType(TrackedFile.MANIFEST_INFO.fieldId(), ManifestInfoStruct.class)
+            .setCustomType(TrackedFile.PARTITION_ID, PartitionData.class)
+            .reuseContainers()
+            .build();
+    addCloseable(reader);
+    return reader;
+  }
+
+  private TrackedFile prepare(TrackedFile trackedFile) {
+    Tracking tracking = trackedFile.tracking();
+    // manifestLocation is not stored in the manifest; the reader fills it in
+    if (tracking instanceof TrackingStruct) {
+      ((TrackingStruct) tracking).setManifestLocation(file.location());
+    }
+
+    return trackedFile;
+  }
+
+  static class Builder {
+    private final InputFile file;
+    private final Types.StructType partitionType;
+    private final Map<Integer, PartitionSpec> specsById;
+    private Expression rowFilter = Expressions.alwaysTrue();
+    private boolean caseSensitive = true;
+    private boolean onlyLive = false;
+    private Schema fileProjection = null;
+    private ScanMetrics scanMetrics = ScanMetrics.noop();
+
+    private Builder(InputFile file, Map<Integer, PartitionSpec> specsById) {
+      this.file = file;
+      this.partitionType = Partitioning.unionPartitionTypes(specsById.values());
+      this.specsById = specsById;
+    }
+
+    /** Sets a row filter; files that cannot match the expression are skipped. */
+    Builder filterRows(Expression expr) {
+      Preconditions.checkArgument(expr != null, "Invalid row filter: null");
+      this.rowFilter = expr;
+      return this;
+    }
+
+    Builder caseSensitive(boolean isCaseSensitive) {
+      this.caseSensitive = isCaseSensitive;
+      return this;
+    }
+
+    /** Returns only files whose tracking {@link Tracking#isLive() is live}. */
+    Builder liveOnly() {
+      this.onlyLive = true;
+      return this;
+    }
+
+    Builder project(Schema newFileProjection) {
+      this.fileProjection = newFileProjection;
+      return this;
+    }
+
+    Builder scanMetrics(ScanMetrics newScanMetrics) {
+      Preconditions.checkArgument(newScanMetrics != null, "Invalid scan metrics: null");
+      this.scanMetrics = newScanMetrics;
+      return this;
+    }
+
+    V4ManifestReader build() {
+      Map<Integer, Evaluator> partitionEvaluators = Maps.newHashMap();
+      Map<Integer, StructProjection> partitionProjections = Maps.newHashMap();
+      if (hasPartitionFilter()) {
+        for (PartitionSpec spec : specsById.values()) {
+          Expression partFilter = Projections.inclusive(spec, caseSensitive).project(rowFilter);
+          partitionEvaluators.put(
+              spec.specId(), new Evaluator(spec.partitionType(), partFilter, caseSensitive));
+          partitionProjections.put(
+              spec.specId(), StructProjection.create(partitionType, spec.partitionType()));
+        }
+      }
+
+      return new V4ManifestReader(
+          file, readSchema(), partitionEvaluators, partitionProjections, onlyLive, scanMetrics);
+    }
+
+    private boolean hasPartitionFilter() {
+      return rowFilter != Expressions.alwaysTrue() && !partitionType.fields().isEmpty();
+    }
+
+    private Schema readSchema() {
+      Types.StructType fullType = TrackedFile.schema(partitionType, Types.StructType.of());
+
+      // replace tracking with the subset of fields read on the scan path
+      List<Types.NestedField> fields = Lists.newArrayList();
+      for (Types.NestedField field : fullType.fields()) {
+        if (field.fieldId() == TrackedFile.TRACKING.fieldId()) {
+          fields.add(
+              Types.NestedField.required(
+                  field.fieldId(), field.name(), SCAN_TRACKING, field.doc()));
+        } else {
+          fields.add(field);
+        }
+      }
+
+      Schema fullSchema = new Schema(fields);
+      if (fileProjection == null) {
+        return fullSchema;
+      }
+
+      Set<Integer> projectedIds = Sets.newHashSet();
+      for (Types.NestedField field : fileProjection.asStruct().fields()) {
+        projectedIds.add(field.fieldId());
+      }
+
+      // status drives live-file filtering and content type distinguishes entry kinds
+      projectedIds.add(TrackedFile.TRACKING.fieldId());
+      projectedIds.add(TrackedFile.CONTENT_TYPE.fieldId());
+      if (hasPartitionFilter()) {
+        projectedIds.add(TrackedFile.SPEC_ID.fieldId());
+        projectedIds.add(TrackedFile.PARTITION_ID);
+      }
+
+      return TypeUtil.select(fullSchema, projectedIds);
+    }
+  }
+}

core/src/test/java/org/apache/iceberg/TestPartitioning.java

@@ -211,6 +211,25 @@ public void testPartitionTypeIgnoreInactiveFields() {
     assertThat(actualType).isEqualTo(StructType.of());
   }
 
+  @Test
+  public void testUnionPartitionTypesRetainsDroppedSourceFields() {
+    TestTables.TestTable table =
+        TestTables.create(
+            tableDir, "test", SCHEMA, BY_DATA_CATEGORY_BUCKET_SPEC, V2_FORMAT_VERSION);
+
+    table.updateSpec().removeField("category_bucket").commit();
+    table.updateSchema().deleteColumn("category").commit();
+
+    // fields with dropped source columns are retained to preserve partition tuple equality;
+    // their type is unknown because it cannot be determined without the source column
+    StructType actualType = Partitioning.unionPartitionTypes(table.specs().values());
+    assertThat(actualType)
+        .isEqualTo(
+            StructType.of(
+                NestedField.optional(1000, "data", Types.StringType.get()),
+                NestedField.optional(1001, "category_bucket", Types.UnknownType.get())));
+  }
+
   @Test
   public void testGroupingKeyTypeWithSpecEvolutionInV1Tables() {
     TestTables.TestTable table =