V4 enable broad tests

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

@@ -29,6 +29,12 @@
 public interface ManifestFile {
   int PARTITION_SUMMARIES_ELEMENT_ID = 508;
 
+  /** Value of {@link #formatVersion()} for v4+ manifest files. */
+  int V4_FORMAT_VERSION = 4;
+
+  /** Value of {@link #formatVersion()} for pre-v4 manifest files (v1, v2, and v3). */
+  int LEGACY_FORMAT_VERSION = 0;
+
   Types.NestedField PATH =
       required(500, "manifest_path", Types.StringType.get(), "Location URI with FS scheme");
   Types.NestedField LENGTH =
@@ -186,6 +192,34 @@ default boolean hasDeletedFiles() {
   /** Returns the total number of rows in all files with status DELETED in the manifest file. */
   Long deletedRowsCount();
 
+  /**
+   * Returns the number of files with status REPLACED in the manifest file, or null if not tracked.
+   *
+   * <p>REPLACED files are the prior-state entries of v4+ REPLACED/MODIFIED pairs and are not live.
+   * Returns null for manifest files written by pre-v4 writers.
+   */
+  default Integer replacedFilesCount() {
+    return null;
+  }
+
+  /**
+   * Returns the total number of rows in all files with status REPLACED in the manifest file, or
+   * null if not tracked.
+   *
+   * <p>Returns null for manifest files written by pre-v4 writers.
+   */
+  default Long replacedRowsCount() {
+    return null;
+  }
+
+  /**
+   * Returns the format version of the manifest file, or {@link #LEGACY_FORMAT_VERSION} for pre-v4
+   * manifests.
+   */
+  default int formatVersion() {
+    return LEGACY_FORMAT_VERSION;
+  }
+
   /**
    * Returns a list of {@link PartitionFieldSummary partition field summaries}.
    *
@@ -210,6 +244,11 @@ default Long firstRowId() {
     return null;
   }
 
+  /** Returns the number of records in the manifest file, or {@code null} for pre-v4 manifests. */
+  default Long recordCount() {
+    return null;
+  }
+
   /**
    * Copies this {@link ManifestFile manifest file}. Readers can reuse manifest file instances; use
    * this method to make defensive copies.

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

@@ -170,6 +170,16 @@ default Iterable<DeleteFile> removedDeleteFiles(FileIO io) {
    */
   String manifestListLocation();
 
+  /**
+   * Returns the location of this snapshot's root manifest, or null if this snapshot uses a manifest
+   * list. Root manifests are introduced in format version 4 and replace manifest lists.
+   *
+   * @return the location of the root manifest for this snapshot, or null
+   */
+  default String rootManifestLocation() {
+    return null;
+  }
+
   /**
    * Return the id of the schema used when this snapshot was created, or null if this information is
    * not available.

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

@@ -144,9 +144,14 @@ protected CloseableIterable<FileScanTask> doPlanFiles() {
                       filter,
                       snap.snapshotId());
                 } else {
+                  // v4+ snapshots address the manifest tree via a root manifest; legacy snapshots
+                  // without a manifest list (synthetic) fall back to the table metadata file.
+                  String inputLocation =
+                      snap.rootManifestLocation() != null
+                          ? snap.rootManifestLocation()
+                          : ((BaseTable) table()).operations().current().metadataFileLocation();
                   return StaticDataTask.of(
-                      io.newInputFile(
-                          ((BaseTable) table()).operations().current().metadataFileLocation()),
+                      io.newInputFile(inputLocation),
                       MANIFEST_FILE_SCHEMA,
                       schema(),
                       snap.allManifests(io),

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

@@ -280,6 +280,10 @@ void setManifestLocation(String manifestLocation) {
     this.manifestLocation = manifestLocation;
   }
 
+  void setFileOrdinal(long ordinal) {
+    this.fileOrdinal = ordinal;
+  }
+
   @Override
   public Long fileSequenceNumber() {
     return fileSequenceNumber;
@@ -331,7 +335,7 @@ protected <T> void internalSet(int pos, T value) {
         return;
       case 4:
         // Preserve the constructor-initialized partitionData when the reader returns null
-        // (e.g., v4 Parquet manifests for unpartitioned tables omit the partition field).
+        // (e.g., v4+ Parquet manifests for unpartitioned tables omit the partition field).
         if (value != null) {
           this.partitionData = (PartitionData) value;
         }

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

@@ -33,6 +33,7 @@
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.encryption.EncryptingFileIO;
 import org.apache.iceberg.events.CreateSnapshotEvent;
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.exceptions.ValidationException;
@@ -254,7 +255,11 @@ private void performRewrite(List<ManifestFile> currentManifests) {
                 } else {
                   rewrittenManifests.add(manifest);
                   try (ManifestReader<DataFile> reader =
-                      ManifestFiles.read(manifest, ops().io(), ops().current().specsById())
+                      ManifestFiles.read(
+                              manifest,
+                              EncryptingFileIO.combine(ops().io(), ops().encryption()),
+                              ops().current().specsById(),
+                              true)
                           .select(Collections.singletonList("*"))) {
                     reader
                         .liveEntries()

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

@@ -32,13 +32,16 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.Pair;
 
 class BaseSnapshot implements Snapshot {
   private final long snapshotId;
   private final Long parentId;
   private final long sequenceNumber;
   private final long timestampMillis;
   private final String manifestListLocation;
+  private final String rootManifestLocation;
+  private final int formatVersion;
   private final String operation;
   private final Map<String, String> summary;
   private final Integer schemaId;
@@ -68,6 +71,41 @@ class BaseSnapshot implements Snapshot {
       Long firstRowId,
       Long addedRows,
       String keyId) {
+    this(
+        2,
+        sequenceNumber,
+        snapshotId,
+        parentId,
+        timestampMillis,
+        operation,
+        summary,
+        schemaId,
+        manifestList,
+        null,
+        firstRowId,
+        addedRows,
+        keyId);
+  }
+
+  BaseSnapshot(
+      int formatVersion,
+      long sequenceNumber,
+      long snapshotId,
+      Long parentId,
+      long timestampMillis,
+      String operation,
+      Map<String, String> summary,
+      Integer schemaId,
+      String manifestList,
+      String rootManifest,
+      Long firstRowId,
+      Long addedRows,
+      String keyId) {
+    Preconditions.checkArgument(
+        (manifestList == null) != (rootManifest == null),
+        "Invalid snapshot: must have exactly one of manifest-list (%s) or root-manifest (%s)",
+        manifestList,
+        rootManifest);
     Preconditions.checkArgument(
         firstRowId == null || firstRowId >= 0,
         "Invalid first-row-id (cannot be negative): %s",
@@ -79,6 +117,7 @@ class BaseSnapshot implements Snapshot {
     Preconditions.checkArgument(
         firstRowId == null || addedRows != null,
         "Invalid added-rows (required when first-row-id is set): null");
+    this.formatVersion = formatVersion;
     this.sequenceNumber = sequenceNumber;
     this.snapshotId = snapshotId;
     this.parentId = parentId;
@@ -87,6 +126,7 @@ class BaseSnapshot implements Snapshot {
     this.summary = summary;
     this.schemaId = schemaId;
     this.manifestListLocation = manifestList;
+    this.rootManifestLocation = rootManifest;
     this.v1ManifestLocations = null;
     this.firstRowId = firstRowId;
     this.addedRows = firstRowId != null ? addedRows : null;
@@ -102,6 +142,7 @@ class BaseSnapshot implements Snapshot {
       Map<String, String> summary,
       Integer schemaId,
       String[] v1ManifestLocations) {
+    this.formatVersion = 1;
     this.sequenceNumber = sequenceNumber;
     this.snapshotId = snapshotId;
     this.parentId = parentId;
@@ -110,6 +151,7 @@ class BaseSnapshot implements Snapshot {
     this.summary = summary;
     this.schemaId = schemaId;
     this.manifestListLocation = null;
+    this.rootManifestLocation = null;
     this.v1ManifestLocations = v1ManifestLocations;
     this.firstRowId = null;
     this.addedRows = null;
@@ -182,10 +224,16 @@ private void cacheManifests(FileIO fileIO) {
 
     if (allManifests == null) {
       // if manifests isn't set, then the snapshotFile is set and should be read to get the list
-      this.allManifests =
-          ManifestLists.read(
-              ManifestLists.newInputFile(
-                  fileIO, new BaseManifestListFile(manifestListLocation, keyId)));
+      if (formatVersion >= 4) {
+        this.allManifests =
+            RootManifests.read(
+                fileIO.newInputFile(new BaseManifestListFile(rootManifestLocation, keyId)));
+      } else {
+        this.allManifests =
+            ManifestLists.read(
+                ManifestLists.newInputFile(
+                    fileIO, new BaseManifestListFile(manifestListLocation, keyId)));
+      }
     }
 
     if (dataManifests == null || deleteManifests == null) {
@@ -261,6 +309,11 @@ public String manifestListLocation() {
     return manifestListLocation;
   }
 
+  @Override
+  public String rootManifestLocation() {
+    return rootManifestLocation;
+  }
+
   private void cacheDeleteFileChanges(FileIO fileIO) {
     Preconditions.checkArgument(fileIO != null, "Cannot cache delete file changes: FileIO is null");
 
@@ -291,6 +344,33 @@ private void cacheDeleteFileChanges(FileIO fileIO) {
       }
     }
 
+    // v4+ colocated DVs live on data manifests' ADDED/MODIFIED rows (newly-live DVs) and REPLACED
+    // rows (superseded DVs preserved by V4Writer.prepareReplaced). Scan data manifests written by
+    // this snapshot to surface those DV changes as delete-file deltas. Legacy data manifests
+    // produce an empty iterable from readColocatedDVChanges, so this scan is a no-op for v1-v3.
+    Iterable<ManifestFile> changedDataManifests =
+        Iterables.filter(
+            dataManifests(fileIO), manifest -> Objects.equal(manifest.snapshotId(), snapshotId));
+    for (ManifestFile manifest : changedDataManifests) {
+      try (CloseableIterable<Pair<ManifestEntry.Status, DeleteFile>> changes =
+          ManifestFiles.readColocatedDVChanges(manifest, fileIO, null)) {
+        for (Pair<ManifestEntry.Status, DeleteFile> change : changes) {
+          switch (change.first()) {
+            case ADDED:
+              adds.add(change.second());
+              break;
+            case DELETED:
+              deletes.add(change.second());
+              break;
+            default:
+              // No other statuses surface from readColocatedDVChanges.
+          }
+        }
+      } catch (IOException e) {
+        throw new UncheckedIOException("Failed to close manifest reader", e);
+      }
+    }
+
     this.addedDeleteFiles = adds.build();
     this.removedDeleteFiles = deletes.build();
   }
@@ -305,23 +385,48 @@ private void cacheDataFileChanges(FileIO fileIO) {
     Iterable<ManifestFile> changedManifests =
         Iterables.filter(
             dataManifests(fileIO), manifest -> Objects.equal(manifest.snapshotId(), snapshotId));
-    try (CloseableIterable<ManifestEntry<DataFile>> entries =
-        new ManifestGroup(fileIO, changedManifests).ignoreExisting().entries()) {
-      for (ManifestEntry<DataFile> entry : entries) {
-        switch (entry.status()) {
-          case ADDED:
-            adds.add(entry.file().copy());
-            break;
-          case DELETED:
-            deletes.add(entry.file().copyWithoutStats());
-            break;
-          default:
-            throw new IllegalStateException(
-                "Unexpected entry status, not added or deleted: " + entry);
+    for (ManifestFile manifest : changedManifests) {
+      // v4+ leaves (content_entry schema) carry REPLACED/MODIFIED rows that the legacy reader
+      // collapses to DELETED/EXISTING. Route v4+ manifests through readDataFileChanges so REPLACED
+      // (a DV-state transition) is not mis-classified as a data-file removal.
+      if (ManifestFiles.isV4ContentEntryManifest(manifest, fileIO)) {
+        try (CloseableIterable<Pair<ManifestEntry.Status, DataFile>> changes =
+            ManifestFiles.readDataFileChanges(manifest, fileIO, null)) {
+          for (Pair<ManifestEntry.Status, DataFile> change : changes) {
+            switch (change.first()) {
+              case ADDED:
+                adds.add(change.second().copy());
+                break;
+              case DELETED:
+                deletes.add(change.second().copyWithoutStats());
+                break;
+              default:
+                // No other statuses surface from readDataFileChanges.
+            }
+          }
+        } catch (IOException e) {
+          throw new RuntimeIOException(e, "Failed to close entries while caching changes");
+        }
+      } else {
+        try (CloseableIterable<ManifestEntry<DataFile>> entries =
+            new ManifestGroup(fileIO, ImmutableList.of(manifest)).ignoreExisting().entries()) {
+          for (ManifestEntry<DataFile> entry : entries) {
+            switch (entry.status()) {
+              case ADDED:
+                adds.add(entry.file().copy());
+                break;
+              case DELETED:
+                deletes.add(entry.file().copyWithoutStats());
+                break;
+              default:
+                throw new IllegalStateException(
+                    "Unexpected entry status, not added or deleted: " + entry);
+            }
+          }
+        } catch (IOException e) {
+          throw new RuntimeIOException(e, "Failed to close entries while caching changes");
         }
       }
-    } catch (IOException e) {
-      throw new RuntimeIOException(e, "Failed to close entries while caching changes");
     }
 
     this.addedDataFiles = adds.build();