Core: Reject v1 manifests in RewriteManifests.addManifest for v2+ tables

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

@@ -140,6 +140,17 @@ public RewriteManifests addManifest(ManifestFile manifest) {
     Preconditions.checkArgument(
         manifest.sequenceNumber() == -1, "Sequence must be assigned during commit");
 
+    int tableFormatVersion = ops().current().formatVersion();
+    if (tableFormatVersion > 1) {
+      int manifestFormatVersion = ManifestReader.formatVersion(ops().io().newInputFile(manifest));
+      Preconditions.checkArgument(
+          manifestFormatVersion > 1,
+          "Cannot add v%s manifest to a v%s table: %s",
+          manifestFormatVersion,
+          tableFormatVersion,
+          manifest.path());
+    }
+
     if (canInheritSnapshotId() && manifest.snapshotId() == null) {
       addedManifests.add(manifest);
     } else {

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

@@ -204,6 +204,16 @@ private static <T extends ContentFile<T>> Map<String, String> readMetadata(Input
     return metadata;
   }
 
+  static int formatVersion(InputFile inputFile) {
+    if (FileFormat.fromFileName(inputFile.location()) != FileFormat.AVRO) {
+      // v1 manifests are always Avro, so a non-Avro manifest cannot be v1
+      return Integer.MAX_VALUE;
+    }
+
+    String version = readMetadata(inputFile).get("format-version");
+    return version == null ? 1 : Integer.parseInt(version);
+  }
+
   public boolean isDeleteManifestReader() {
     return content == FileType.DELETE_FILES;
   }

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

@@ -39,6 +39,7 @@
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.junit.jupiter.api.TestTemplate;
@@ -942,6 +943,64 @@ public void testManifestReplacementCombinedWithRewriteConcurrentDelete() throws
         statuses(ManifestEntry.Status.EXISTING));
   }
 
+  @TestTemplate
+  public void testAddV1ManifestToV2PlusTable() throws IOException {
+    assumeThat(formatVersion).isGreaterThan(1);
+
+    table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
+    Snapshot snapshot = table.currentSnapshot();
+    ManifestFile original = Iterables.getOnlyElement(snapshot.allManifests(table.io()));
+
+    OutputFile outputFile =
+        table.ops().io().newOutputFile(temp.resolve("v1-manifest.avro").toString());
+    ManifestWriter<DataFile> writer = ManifestFiles.write(1, table.spec(), outputFile, null);
+    try {
+      writer.existing(
+          FILE_A, snapshot.snapshotId(), snapshot.sequenceNumber(), snapshot.sequenceNumber());
+      writer.existing(
+          FILE_B, snapshot.snapshotId(), snapshot.sequenceNumber(), snapshot.sequenceNumber());
+    } finally {
+      writer.close();
+    }
+
+    ManifestFile v1Manifest = writer.toManifestFile();
+
+    assertThatThrownBy(
+            () ->
+                table.rewriteManifests().deleteManifest(original).addManifest(v1Manifest).commit())
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageStartingWith("Cannot add v1 manifest to a v");
+  }
+
+  @TestTemplate
+  public void testAddV1CopiedManifestToV2PlusTable() throws IOException {
+    assumeThat(formatVersion).isGreaterThan(1);
+
+    table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit();
+    Snapshot snapshot = table.currentSnapshot();
+    ManifestFile original = Iterables.getOnlyElement(snapshot.allManifests(table.io()));
+
+    OutputFile outputFile =
+        table.ops().io().newOutputFile(temp.resolve("v1-copied-manifest.avro").toString());
+    ManifestWriter<DataFile> writer = ManifestFiles.write(1, table.spec(), outputFile, -1L);
+    try {
+      writer.existing(
+          FILE_A, snapshot.snapshotId(), snapshot.sequenceNumber(), snapshot.sequenceNumber());
+      writer.existing(
+          FILE_B, snapshot.snapshotId(), snapshot.sequenceNumber(), snapshot.sequenceNumber());
+    } finally {
+      writer.close();
+    }
+
+    ManifestFile v1Manifest = writer.toManifestFile();
+
+    assertThatThrownBy(
+            () ->
+                table.rewriteManifests().deleteManifest(original).addManifest(v1Manifest).commit())
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageStartingWith("Cannot add v1 manifest to a v");
+  }
+
   @TestTemplate
   public void testInvalidUsage() throws IOException {
     assertThat(table.currentSnapshot()).isNull();