Repository: apache/iceberg (language: python)

Issue:
Create V3 Metadata Classes
In preparation for Iceberg Table V3 we need to start creating the classes to represent V3 Tables in the Iceberg Core Repository.


Discussion / hints:
Trying to start out with just a plain copy of all the V2 Classes

Things i'm covering in the initial commit: 
Metadata
ManifestList Writer
Manifest Writer
Everything required to run basic tests (parameterized tests with V (1, 2) in core

I'll split out into more tickets for folks to cover
[]Any named tests touching V2
[]Any support for specific engines (Spark, Flink)
[]All V3 Specific functionality

Fix patch:
diff --git a/core/src/main/java/org/apache/iceberg/BaseUpdatePartitionSpec.java b/core/src/main/java/org/apache/iceberg/BaseUpdatePartitionSpec.java
index b59292c397a1..2e1c9199174c 100644
--- a/core/src/main/java/org/apache/iceberg/BaseUpdatePartitionSpec.java
+++ b/core/src/main/java/org/apache/iceberg/BaseUpdatePartitionSpec.java
@@ -118,7 +118,7 @@ private int assignFieldId() {
    */
   private PartitionField recycleOrCreatePartitionField(
       Pair<Integer, Transform<?, ?>> sourceTransform, String name) {
-    if (formatVersion == 2 && base != null) {
+    if (formatVersion >= 2 && base != null) {
       int sourceId = sourceTransform.first();
       Transform<?, ?> transform = sourceTransform.second();
 
diff --git a/core/src/main/java/org/apache/iceberg/ManifestFiles.java b/core/src/main/java/org/apache/iceberg/ManifestFiles.java
index 840c90bebdde..f630bb3eb743 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestFiles.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestFiles.java
@@ -182,6 +182,8 @@ public static ManifestWriter<DataFile> write(
         return new ManifestWriter.V1Writer(spec, encryptedOutputFile, snapshotId);
       case 2:
         return new ManifestWriter.V2Writer(spec, encryptedOutputFile, snapshotId);
+      case 3:
+        return new ManifestWriter.V3Writer(spec, encryptedOutputFile, snapshotId);
     }
     throw new UnsupportedOperationException(
         "Cannot write manifest for table version: " + formatVersion);
@@ -238,6 +240,8 @@ public static ManifestWriter<DeleteFile> writeDeleteManifest(
         throw new IllegalArgumentException("Cannot write delete files in a v1 table");
       case 2:
         return new ManifestWriter.V2DeleteWriter(spec, outputFile, snapshotId);
+      case 3:
+        return new ManifestWriter.V3DeleteWriter(spec, outputFile, snapshotId);
     }
     throw new UnsupportedOperationException(
         "Cannot write manifest for table version: " + formatVersion);
diff --git a/core/src/main/java/org/apache/iceberg/ManifestListWriter.java b/core/src/main/java/org/apache/iceberg/ManifestListWriter.java
index 3f7f20d4df6c..b17eedad18af 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestListWriter.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestListWriter.java
@@ -70,6 +70,41 @@ public long length() {
     return writer.length();
   }
 
+  static class V3Writer extends ManifestListWriter {
+    private final V3Metadata.IndexedManifestFile wrapper;
+
+    V3Writer(OutputFile snapshotFile, long snapshotId, Long parentSnapshotId, long sequenceNumber) {
+      super(
+          snapshotFile,
+          ImmutableMap.of(
+              "snapshot-id", String.valueOf(snapshotId),
+              "parent-snapshot-id", String.valueOf(parentSnapshotId),
+              "sequence-number", String.valueOf(sequenceNumber),
+              "format-version", "3"));
+      this.wrapper = new V3Metadata.IndexedManifestFile(snapshotId, sequenceNumber);
+    }
+
+    @Override
+    protected ManifestFile prepare(ManifestFile manifest) {
+      return wrapper.wrap(manifest);
+    }
+
+    @Override
+    protected FileAppender<ManifestFile> newAppender(OutputFile file, Map<String, String> meta) {
+      try {
+        return Avro.write(file)
+            .schema(V3Metadata.MANIFEST_LIST_SCHEMA)
+            .named("manifest_file")
+            .meta(meta)
+            .overwrite()
+            .build();
+
+      } catch (IOException e) {
+        throw new RuntimeIOException(e, "Failed to create snapshot list writer for path: %s", file);
+      }
+    }
+  }
+
   static class V2Writer extends ManifestListWriter {
     private final V2Metadata.IndexedManifestFile wrapper;
 
diff --git a/core/src/main/java/org/apache/iceberg/ManifestLists.java b/core/src/main/java/org/apache/iceberg/ManifestLists.java
index c7b3e5fee5a9..f20a481cf25a 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestLists.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestLists.java
@@ -66,6 +66,9 @@ static ManifestListWriter write(
       case 2:
         return new ManifestListWriter.V2Writer(
             manifestListFile, snapshotId, parentSnapshotId, sequenceNumber);
+      case 3:
+        return new ManifestListWriter.V3Writer(
+            manifestListFile, snapshotId, parentSnapshotId, sequenceNumber);
     }
     throw new UnsupportedOperationException(
         "Cannot write manifest list for table version: " + formatVersion);
diff --git a/core/src/main/java/org/apache/iceberg/ManifestWriter.java b/core/src/main/java/org/apache/iceberg/ManifestWriter.java
index cea907ddaca1..88587a1ebc89 100644
--- a/core/src/main/java/org/apache/iceberg/ManifestWriter.java
+++ b/core/src/main/java/org/apache/iceberg/ManifestWriter.java
@@ -217,6 +217,79 @@ public void close() throws IOException {
     writer.close();
   }
 
+  static class V3Writer extends ManifestWriter<DataFile> {
+    private final V3Metadata.IndexedManifestEntry<DataFile> entryWrapper;
+
+    V3Writer(PartitionSpec spec, EncryptedOutputFile file, Long snapshotId) {
+      super(spec, file, snapshotId);
+      this.entryWrapper = new V3Metadata.IndexedManifestEntry<>(snapshotId, spec.partitionType());
+    }
+
+    @Override
+    protected ManifestEntry<DataFile> prepare(ManifestEntry<DataFile> entry) {
+      return entryWrapper.wrap(entry);
+    }
+
+    @Override
+    protected FileAppender<ManifestEntry<DataFile>> newAppender(
+        PartitionSpec spec, OutputFile file) {
+      Schema manifestSchema = V3Metadata.entrySchema(spec.partitionType());
+      try {
+        return Avro.write(file)
+            .schema(manifestSchema)
+            .named("manifest_entry")
+            .meta("schema", SchemaParser.toJson(spec.schema()))
+            .meta("partition-spec", PartitionSpecParser.toJsonFields(spec))
+            .meta("partition-spec-id", String.valueOf(spec.specId()))
+            .meta("format-version", "3")
+            .meta("content", "data")
+            .overwrite()
+            .build();
+      } catch (IOException e) {
+        throw new RuntimeIOException(e, "Failed to create manifest writer for path: %s", file);
+      }
+    }
+  }
+
+  static class V3DeleteWriter extends ManifestWriter<DeleteFile> {
+    private final V3Metadata.IndexedManifestEntry<DeleteFile> entryWrapper;
+
+    V3DeleteWriter(PartitionSpec spec, EncryptedOutputFile file, Long snapshotId) {
+      super(spec, file, snapshotId);
+      this.entryWrapper = new V3Metadata.IndexedManifestEntry<>(snapshotId, spec.partitionType());
+    }
+
+    @Override
+    protected ManifestEntry<DeleteFile> prepare(ManifestEntry<DeleteFile> entry) {
+      return entryWrapper.wrap(entry);
+    }
+
+    @Override
+    protected FileAppender<ManifestEntry<DeleteFile>> newAppender(
+        PartitionSpec spec, OutputFile file) {
+      Schema manifestSchema = V3Metadata.entrySchema(spec.partitionType());
+      try {
+        return Avro.write(file)
+            .schema(manifestSchema)
+            .named("manifest_entry")
+            .meta("schema", SchemaParser.toJson(spec.schema()))
+            .meta("partition-spec", PartitionSpecParser.toJsonFields(spec))
+            .meta("partition-spec-id", String.valueOf(spec.specId()))
+            .meta("format-version", "3")
+            .meta("content", "deletes")
+            .overwrite()
+            .build();
+      } catch (IOException e) {
+        throw new RuntimeIOException(e, "Failed to create manifest writer for path: %s", file);
+      }
+    }
+
+    @Override
+    protected ManifestContent content() {
+      return ManifestContent.DELETES;
+    }
+  }
+
   static class V2Writer extends ManifestWriter<DataFile> {
     private final V2Metadata.IndexedManifestEntry<DataFile> entryWrapper;
 
diff --git a/core/src/main/java/org/apache/iceberg/TableMetadata.java b/core/src/main/java/org/apache/iceberg/TableMetadata.java
index 74b8ad0bbddc..bd1c8a1a0371 100644
--- a/core/src/main/java/org/apache/iceberg/TableMetadata.java
+++ b/core/src/main/java/org/apache/iceberg/TableMetadata.java
@@ -51,7 +51,7 @@ public class TableMetadata implements Serializable {
   static final long INITIAL_SEQUENCE_NUMBER = 0;
   static final long INVALID_SEQUENCE_NUMBER = -1;
   static final int DEFAULT_TABLE_FORMAT_VERSION = 2;
-  static final int SUPPORTED_TABLE_FORMAT_VERSION = 2;
+  static final int SUPPORTED_TABLE_FORMAT_VERSION = 3;
   static final int INITIAL_SPEC_ID = 0;
   static final int INITIAL_SORT_ORDER_ID = 1;
   static final int INITIAL_SCHEMA_ID = 0;
diff --git a/core/src/main/java/org/apache/iceberg/V3Metadata.java b/core/src/main/java/org/apache/iceberg/V3Metadata.java
new file mode 100644
index 000000000000..94e20ea99858
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/V3Metadata.java
@@ -0,0 +1,575 @@
+/*
+ * 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 static org.apache.iceberg.types.Types.NestedField.required;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.types.Types;
+
+class V3Metadata {
+  private V3Metadata() {}
+
+  static final Schema MANIFEST_LIST_SCHEMA =
+      new Schema(
+          ManifestFile.PATH,
+          ManifestFile.LENGTH,
+          ManifestFile.SPEC_ID,
+          ManifestFile.MANIFEST_CONTENT.asRequired(),
+          ManifestFile.SEQUENCE_NUMBER.asRequired(),
+          ManifestFile.MIN_SEQUENCE_NUMBER.asRequired(),
+          ManifestFile.SNAPSHOT_ID.asRequired(),
+          ManifestFile.ADDED_FILES_COUNT.asRequired(),
+          ManifestFile.EXISTING_FILES_COUNT.asRequired(),
+          ManifestFile.DELETED_FILES_COUNT.asRequired(),
+          ManifestFile.ADDED_ROWS_COUNT.asRequired(),
+          ManifestFile.EXISTING_ROWS_COUNT.asRequired(),
+          ManifestFile.DELETED_ROWS_COUNT.asRequired(),
+          ManifestFile.PARTITION_SUMMARIES,
+          ManifestFile.KEY_METADATA);
+
+  /**
+   * A wrapper class to write any ManifestFile implementation to Avro using the v3 write schema.
+   *
+   * <p>This is used to maintain compatibility with v3 by writing manifest list files with the old
+   * schema, instead of writing a sequence number into metadata files in v3 tables.
+   */
+  static class IndexedManifestFile implements ManifestFile, IndexedRecord {
+    private static final org.apache.avro.Schema AVRO_SCHEMA =
+        AvroSchemaUtil.convert(MANIFEST_LIST_SCHEMA, "manifest_file");
+
+    private final long commitSnapshotId;
+    private final long sequenceNumber;
+    private ManifestFile wrapped = null;
+
+    IndexedManifestFile(long commitSnapshotId, long sequenceNumber) {
+      this.commitSnapshotId = commitSnapshotId;
+      this.sequenceNumber = sequenceNumber;
+    }
+
+    public ManifestFile wrap(ManifestFile file) {
+      this.wrapped = file;
+      return this;
+    }
+
+    @Override
+    public org.apache.avro.Schema getSchema() {
+      return AVRO_SCHEMA;
+    }
+
+    @Override
+    public void put(int i, Object v) {
+      throw new UnsupportedOperationException("Cannot modify IndexedManifestFile wrapper via put");
+    }
+
+    @Override
+    public Object get(int pos) {
+      switch (pos) {
+        case 0:
+          return wrapped.path();
+        case 1:
+          return wrapped.length();
+        case 2:
+          return wrapped.partitionSpecId();
+        case 3:
+          return wrapped.content().id();
+        case 4:
+          if (wrapped.sequenceNumber() == ManifestWriter.UNASSIGNED_SEQ) {
+            // if the sequence number is being assigned here, then the manifest must be created by
+            // the current
+            // operation. to validate this, check that the snapshot id matches the current commit
+            Preconditions.checkState(
+                commitSnapshotId == wrapped.snapshotId(),
+                "Found unassigned sequence number for a manifest from snapshot: %s",
+                wrapped.snapshotId());
+            return sequenceNumber;
+          } else {
+            return wrapped.sequenceNumber();
+          }
+        case 5:
+          if (wrapped.minSequenceNumber() == ManifestWriter.UNASSIGNED_SEQ) {
+            // same sanity check as above
+            Preconditions.checkState(
+                commitSnapshotId == wrapped.snapshotId(),
+                "Found unassigned sequence number for a manifest from snapshot: %s",
+                wrapped.snapshotId());
+            // if the min sequence number is not determined, then there was no assigned sequence
+            // number for any file
+            // written to the wrapped manifest. replace the unassigned sequence number with the one
+            // for this commit
+            return sequenceNumber;
+          } else {
+            return wrapped.minSequenceNumber();
+          }
+        case 6:
+          return wrapped.snapshotId();
+        case 7:
+          return wrapped.addedFilesCount();
+        case 8:
+          return wrapped.existingFilesCount();
+        case 9:
+          return wrapped.deletedFilesCount();
+        case 10:
+          return wrapped.addedRowsCount();
+        case 11:
+          return wrapped.existingRowsCount();
+        case 12:
+          return wrapped.deletedRowsCount();
+        case 13:
+          return wrapped.partitions();
+        case 14:
+          return wrapped.keyMetadata();
+        default:
+          throw new UnsupportedOperationException("Unknown field ordinal: " + pos);
+      }
+    }
+
+    @Override
+    public String path() {
+      return wrapped.path();
+    }
+
+    @Override
+    public long length() {
+      return wrapped.length();
+    }
+
+    @Override
+    public int partitionSpecId() {
+      return wrapped.partitionSpecId();
+    }
+
+    @Override
+    public ManifestContent content() {
+      return wrapped.content();
+    }
+
+    @Override
+    public long sequenceNumber() {
+      return wrapped.sequenceNumber();
+    }
+
+    @Override
+    public long minSequenceNumber() {
+      return wrapped.minSequenceNumber();
+    }
+
+    @Override
+    public Long snapshotId() {
+      return wrapped.snapshotId();
+    }
+
+    @Override
+    public boolean hasAddedFiles() {
+      return wrapped.hasAddedFiles();
+    }
+
+    @Override
+    public Integer addedFilesCount() {
+      return wrapped.addedFilesCount();
+    }
+
+    @Override
+    public Long addedRowsCount() {
+      return wrapped.addedRowsCount();
+    }
+
+    @Override
+    public boolean hasExistingFiles() {
+      return wrapped.hasExistingFiles();
+    }
+
+    @Override
+    public Integer existingFilesCount() {
+      return wrapped.existingFilesCount();
+    }
+
+    @Override
+    public Long existingRowsCount() {
+      return wrapped.existingRowsCount();
+    }
+
+    @Override
+    public boolean hasDeletedFiles() {
+      return wrapped.hasDeletedFiles();
+    }
+
+    @Override
+    public Integer deletedFilesCount() {
+      return wrapped.deletedFilesCount();
+    }
+
+    @Override
+    public Long deletedRowsCount() {
+      return wrapped.deletedRowsCount();
+    }
+
+    @Override
+    public List<PartitionFieldSummary> partitions() {
+      return wrapped.partitions();
+    }
+
+    @Override
+    public ByteBuffer keyMetadata() {
+      return wrapped.keyMetadata();
+    }
+
+    @Override
+    public ManifestFile copy() {
+      return wrapped.copy();
+    }
+  }
+
+  static Schema entrySchema(Types.StructType partitionType) {
+    return wrapFileSchema(fileType(partitionType));
+  }
+
+  static Schema wrapFileSchema(Types.StructType fileSchema) {
+    // this is used to build projection schemas
+    return new Schema(
+        ManifestEntry.STATUS,
+        ManifestEntry.SNAPSHOT_ID,
+        ManifestEntry.SEQUENCE_NUMBER,
+        ManifestEntry.FILE_SEQUENCE_NUMBER,
+        required(ManifestEntry.DATA_FILE_ID, "data_file", fileSchema));
+  }
+
+  static Types.StructType fileType(Types.StructType partitionType) {
+    return Types.StructType.of(
+        DataFile.CONTENT.asRequired(),
+        DataFile.FILE_PATH,
+        DataFile.FILE_FORMAT,
+        required(
+            DataFile.PARTITION_ID, DataFile.PARTITION_NAME, partitionType, DataFile.PARTITION_DOC),
+        DataFile.RECORD_COUNT,
+        DataFile.FILE_SIZE,
+        DataFile.COLUMN_SIZES,
+        DataFile.VALUE_COUNTS,
+        DataFile.NULL_VALUE_COUNTS,
+        DataFile.NAN_VALUE_COUNTS,
+        DataFile.LOWER_BOUNDS,
+        DataFile.UPPER_BOUNDS,
+        DataFile.KEY_METADATA,
+        DataFile.SPLIT_OFFSETS,
+        DataFile.EQUALITY_IDS,
+        DataFile.SORT_ORDER_ID);
+  }
+
+  static class IndexedManifestEntry<F extends ContentFile<F>>
+      implements ManifestEntry<F>, IndexedRecord {
+    private final org.apache.avro.Schema avroSchema;
+    private final Long commitSnapshotId;
+    private final IndexedDataFile<?> fileWrapper;
+    private ManifestEntry<F> wrapped = null;
+
+    IndexedManifestEntry(Long commitSnapshotId, Types.StructType partitionType) {
+      this.avroSchema = AvroSchemaUtil.convert(entrySchema(partitionType), "manifest_entry");
+      this.commitSnapshotId = commitSnapshotId;
+      this.fileWrapper = new IndexedDataFile<>(partitionType);
+    }
+
+    public IndexedManifestEntry<F> wrap(ManifestEntry<F> entry) {
+      this.wrapped = entry;
+      return this;
+    }
+
+    @Override
+    public org.apache.avro.Schema getSchema() {
+      return avroSchema;
+    }
+
+    @Override
+    public void put(int i, Object v) {
+      throw new UnsupportedOperationException("Cannot modify IndexedManifestEntry wrapper via put");
+    }
+
+    @Override
+    public Object get(int i) {
+      switch (i) {
+        case 0:
+          return wrapped.status().id();
+        case 1:
+          return wrapped.snapshotId();
+        case 2:
+          if (wrapped.dataSequenceNumber() == null) {
+            // if the entry's data sequence number is null,
+            // then it will inherit the sequence number of the current commit.
+            // to validate that this is correct, check that the snapshot id is either null (will
+            // also be inherited) or that it matches the id of the current commit.
+            Preconditions.checkState(
+                wrapped.snapshotId() == null || wrapped.snapshotId().equals(commitSnapshotId),
+                "Found unassigned sequence number for an entry from snapshot: %s",
+                wrapped.snapshotId());
+
+            // inheritance should work only for ADDED entries
+            Preconditions.checkState(
+                wrapped.status() == Status.ADDED,
+                "Only entries with status ADDED can have null sequence number");
+
+            return null;
+          }
+          return wrapped.dataSequenceNumber();
+        case 3:
+          return wrapped.fileSequenceNumber();
+        case 4:
+          return fileWrapper.wrap(wrapped.file());
+        default:
+          throw new UnsupportedOperationException("Unknown field ordinal: " + i);
+      }
+    }
+
+    @Override
+    public Status status() {
+      return wrapped.status();
+    }
+
+    @Override
+    public Long snapshotId() {
+      return wrapped.snapshotId();
+    }
+
+    @Override
+    public void setSnapshotId(long snapshotId) {
+      wrapped.setSnapshotId(snapshotId);
+    }
+
+    @Override
+    public Long dataSequenceNumber() {
+      return wrapped.dataSequenceNumber();
+    }
+
+    @Override
+    public void setDataSequenceNumber(long dataSequenceNumber) {
+      wrapped.setDataSequenceNumber(dataSequenceNumber);
+    }
+
+    @Override
+    public Long fileSequenceNumber() {
+      return wrapped.fileSequenceNumber();
+    }
+
+    @Override
+    public void setFileSequenceNumber(long fileSequenceNumber) {
+      wrapped.setFileSequenceNumber(fileSequenceNumber);
+    }
+
+    @Override
+    public F file() {
+      return wrapped.file();
+    }
+
+    @Override
+    public ManifestEntry<F> copy() {
+      return wrapped.copy();
+    }
+
+    @Override
+    public ManifestEntry<F> copyWithoutStats() {
+      return wrapped.copyWithoutStats();
+    }
+  }
+
+  /** Wrapper used to write DataFile or DeleteFile to v3 metadata. */
+  static class IndexedDataFile<F> implements ContentFile<F>, IndexedRecord {
+    private final org.apache.avro.Schema avroSchema;
+    private final IndexedStructLike partitionWrapper;
+    private ContentFile<F> wrapped = null;
+
+    IndexedDataFile(Types.StructType partitionType) {
+      this.avroSchema = AvroSchemaUtil.convert(fileType(partitionType), "data_file");
+      this.partitionWrapper = new IndexedStructLike(avroSchema.getField("partition").schema());
+    }
+
+    @SuppressWarnings("unchecked")
+    IndexedDataFile<F> wrap(ContentFile<?> file) {
+      this.wrapped = (ContentFile<F>) file;
+      return this;
+    }
+
+    @Override
+    public org.apache.avro.Schema getSchema() {
+      return avroSchema;
+    }
+
+    @Override
+    public Object get(int pos) {
+      switch (pos) {
+        case 0:
+          return wrapped.content().id();
+        case 1:
+          return wrapped.path().toString();
+        case 2:
+          return wrapped.format() != null ? wrapped.format().toString() : null;
+        case 3:
+          return partitionWrapper.wrap(wrapped.partition());
+        case 4:
+          return wrapped.recordCount();
+        case 5:
+          return wrapped.fileSizeInBytes();
+        case 6:
+          return wrapped.columnSizes();
+        case 7:
+          return wrapped.valueCounts();
+        case 8:
+          return wrapped.nullValueCounts();
+        case 9:
+          return wrapped.nanValueCounts();
+        case 10:
+          return wrapped.lowerBounds();
+        case 11:
+          return wrapped.upperBounds();
+        case 12:
+          return wrapped.keyMetadata();
+        case 13:
+          return wrapped.splitOffsets();
+        case 14:
+          return wrapped.equalityFieldIds();
+        case 15:
+          return wrapped.sortOrderId();
+      }
+      throw new IllegalArgumentException("Unknown field ordinal: " + pos);
+    }
+
+    @Override
+    public void put(int i, Object v) {
+      throw new UnsupportedOperationException("Cannot modify IndexedDataFile wrapper via put");
+    }
+
+    @Override
+    public Long pos() {
+      return null;
+    }
+
+    @Override
+    public int specId() {
+      return wrapped.specId();
+    }
+
+    @Override
+    public FileContent content() {
+      return wrapped.content();
+    }
+
+    @Override
+    public CharSequence path() {
+      return wrapped.path();
+    }
+
+    @Override
+    public FileFormat format() {
+      return wrapped.format();
+    }
+
+    @Override
+    public StructLike partition() {
+      return wrapped.partition();
+    }
+
+    @Override
+    public long recordCount() {
+      return wrapped.recordCount();
+    }
+
+    @Override
+    public long fileSizeInBytes() {
+      return wrapped.fileSizeInBytes();
+    }
+
+    @Override
+    public Map<Integer, Long> columnSizes() {
+      return wrapped.columnSizes();
+    }
+
+    @Override
+    public Map<Integer, Long> valueCounts() {
+      return wrapped.valueCounts();
+    }
+
+    @Override
+    public Map<Integer, Long> nullValueCounts() {
+      return wrapped.nullValueCounts();
+    }
+
+    @Override
+    public Map<Integer, Long> nanValueCounts() {
+      return wrapped.nanValueCounts();
+    }
+
+    @Override
+    public Map<Integer, ByteBuffer> lowerBounds() {
+      return wrapped.lowerBounds();
+    }
+
+    @Override
+    public Map<Integer, ByteBuffer> upperBounds() {
+      return wrapped.upperBounds();
+    }
+
+    @Override
+    public ByteBuffer keyMetadata() {
+      return wrapped.keyMetadata();
+    }
+
+    @Override
+    public List<Long> splitOffsets() {
+      return wrapped.splitOffsets();
+    }
+
+    @Override
+    public List<Integer> equalityFieldIds() {
+      return wrapped.equalityFieldIds();
+    }
+
+    @Override
+    public Integer sortOrderId() {
+      return wrapped.sortOrderId();
+    }
+
+    @Override
+    public Long dataSequenceNumber() {
+      return wrapped.dataSequenceNumber();
+    }
+
+    @Override
+    public Long fileSequenceNumber() {
+      return wrapped.fileSequenceNumber();
+    }
+
+    @Override
+    public F copy() {
+      throw new UnsupportedOperationException("Cannot copy IndexedDataFile wrapper");
+    }
+
+    @Override
+    public F copyWithStats(Set<Integer> requestedColumnIds) {
+      throw new UnsupportedOperationException("Cannot copy IndexedDataFile wrapper");
+    }
+
+    @Override
+    public F copyWithoutStats() {
+      throw new UnsupportedOperationException("Cannot copy IndexedDataFile wrapper");
+    }
+  }
+}


Test patch:
diff --git a/core/src/test/java/org/apache/iceberg/DeleteFileIndexTestBase.java b/core/src/test/java/org/apache/iceberg/DeleteFileIndexTestBase.java
index 229650566ca8..836a1ddd80f5 100644
--- a/core/src/test/java/org/apache/iceberg/DeleteFileIndexTestBase.java
+++ b/core/src/test/java/org/apache/iceberg/DeleteFileIndexTestBase.java
@@ -46,7 +46,7 @@ public abstract class DeleteFileIndexTestBase<
 
   @Parameters(name = "formatVersion = {0}")
   public static List<Object> parameters() {
-    return Arrays.asList(2);
+    return Arrays.asList(2, 3);
   }
 
   static final DeleteFile FILE_A_POS_1 =
diff --git a/core/src/test/java/org/apache/iceberg/MetadataTableScanTestBase.java b/core/src/test/java/org/apache/iceberg/MetadataTableScanTestBase.java
index a4e964b017ba..98d2d8f38af6 100644
--- a/core/src/test/java/org/apache/iceberg/MetadataTableScanTestBase.java
+++ b/core/src/test/java/org/apache/iceberg/MetadataTableScanTestBase.java
@@ -40,7 +40,7 @@ public abstract class MetadataTableScanTestBase extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   protected Set<String> scannedPaths(TableScan scan) {
diff --git a/core/src/test/java/org/apache/iceberg/ScanPlanningAndReportingTestBase.java b/core/src/test/java/org/apache/iceberg/ScanPlanningAndReportingTestBase.java
index 3ba74c3b4b71..13e96869b454 100644
--- a/core/src/test/java/org/apache/iceberg/ScanPlanningAndReportingTestBase.java
+++ b/core/src/test/java/org/apache/iceberg/ScanPlanningAndReportingTestBase.java
@@ -47,7 +47,7 @@ public abstract class ScanPlanningAndReportingTestBase<
 
   @Parameters(name = "formatVersion = {0}")
   public static List<Object> parameters() {
-    return Arrays.asList(2);
+    return Arrays.asList(2, 3);
   }
 
   protected abstract ScanT newScan(Table table);
diff --git a/core/src/test/java/org/apache/iceberg/TestBase.java b/core/src/test/java/org/apache/iceberg/TestBase.java
index 2322062dad85..e03a1efd5156 100644
--- a/core/src/test/java/org/apache/iceberg/TestBase.java
+++ b/core/src/test/java/org/apache/iceberg/TestBase.java
@@ -174,7 +174,7 @@ public class TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @Parameter protected int formatVersion;
diff --git a/core/src/test/java/org/apache/iceberg/TestBatchScans.java b/core/src/test/java/org/apache/iceberg/TestBatchScans.java
index c7210486e05e..1597f44f6338 100644
--- a/core/src/test/java/org/apache/iceberg/TestBatchScans.java
+++ b/core/src/test/java/org/apache/iceberg/TestBatchScans.java
@@ -34,7 +34,7 @@ public class TestBatchScans extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestCommitReporting.java b/core/src/test/java/org/apache/iceberg/TestCommitReporting.java
index bbba192fab7f..41b301668722 100644
--- a/core/src/test/java/org/apache/iceberg/TestCommitReporting.java
+++ b/core/src/test/java/org/apache/iceberg/TestCommitReporting.java
@@ -37,7 +37,7 @@ public class TestCommitReporting extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(2);
+    return Arrays.asList(2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestCreateTransaction.java b/core/src/test/java/org/apache/iceberg/TestCreateTransaction.java
index 0c6b50b37792..766dd85a0655 100644
--- a/core/src/test/java/org/apache/iceberg/TestCreateTransaction.java
+++ b/core/src/test/java/org/apache/iceberg/TestCreateTransaction.java
@@ -40,7 +40,7 @@
 public class TestCreateTransaction extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestEntriesMetadataTable.java b/core/src/test/java/org/apache/iceberg/TestEntriesMetadataTable.java
index 60e79aad91ce..9bce4e60a4f3 100644
--- a/core/src/test/java/org/apache/iceberg/TestEntriesMetadataTable.java
+++ b/core/src/test/java/org/apache/iceberg/TestEntriesMetadataTable.java
@@ -34,7 +34,7 @@ public class TestEntriesMetadataTable extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestFastAppend.java b/core/src/test/java/org/apache/iceberg/TestFastAppend.java
index 9dd479ecf0fc..b281536ab0fa 100644
--- a/core/src/test/java/org/apache/iceberg/TestFastAppend.java
+++ b/core/src/test/java/org/apache/iceberg/TestFastAppend.java
@@ -39,7 +39,7 @@
 public class TestFastAppend extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestFindFiles.java b/core/src/test/java/org/apache/iceberg/TestFindFiles.java
index 68d757de5cdb..191dbd384197 100644
--- a/core/src/test/java/org/apache/iceberg/TestFindFiles.java
+++ b/core/src/test/java/org/apache/iceberg/TestFindFiles.java
@@ -36,7 +36,7 @@
 public class TestFindFiles extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestFormatVersions.java b/core/src/test/java/org/apache/iceberg/TestFormatVersions.java
index 2f6a01c6e603..4a9f18581d09 100644
--- a/core/src/test/java/org/apache/iceberg/TestFormatVersions.java
+++ b/core/src/test/java/org/apache/iceberg/TestFormatVersions.java
@@ -71,7 +71,7 @@ public void testFormatVersionUpgradeNotSupported() {
                     base,
                     base.upgradeToFormatVersion(TableMetadata.SUPPORTED_TABLE_FORMAT_VERSION + 1)))
         .isInstanceOf(IllegalArgumentException.class)
-        .hasMessage("Cannot upgrade table to unsupported format version: v3 (supported: v2)");
+        .hasMessage("Cannot upgrade table to unsupported format version: v4 (supported: v3)");
 
     assertThat(ops.current().formatVersion()).isEqualTo(1);
   }
diff --git a/core/src/test/java/org/apache/iceberg/TestIncrementalDataTableScan.java b/core/src/test/java/org/apache/iceberg/TestIncrementalDataTableScan.java
index fe6b9b0c763c..ecd6a14ffefb 100644
--- a/core/src/test/java/org/apache/iceberg/TestIncrementalDataTableScan.java
+++ b/core/src/test/java/org/apache/iceberg/TestIncrementalDataTableScan.java
@@ -43,7 +43,7 @@
 public class TestIncrementalDataTableScan extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @BeforeEach
diff --git a/core/src/test/java/org/apache/iceberg/TestLocalFilterFiles.java b/core/src/test/java/org/apache/iceberg/TestLocalFilterFiles.java
index 275b89df3695..8a4f5db256f0 100644
--- a/core/src/test/java/org/apache/iceberg/TestLocalFilterFiles.java
+++ b/core/src/test/java/org/apache/iceberg/TestLocalFilterFiles.java
@@ -28,7 +28,7 @@ public class TestLocalFilterFiles
 
   @Parameters(name = "formatVersion = {0}")
   public static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @Override
diff --git a/core/src/test/java/org/apache/iceberg/TestLocationProvider.java b/core/src/test/java/org/apache/iceberg/TestLocationProvider.java
index 93dd1a8555ce..7afb69483490 100644
--- a/core/src/test/java/org/apache/iceberg/TestLocationProvider.java
+++ b/core/src/test/java/org/apache/iceberg/TestLocationProvider.java
@@ -34,7 +34,7 @@
 public class TestLocationProvider extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   // publicly visible for testing to be dynamically loaded
diff --git a/core/src/test/java/org/apache/iceberg/TestManifestCleanup.java b/core/src/test/java/org/apache/iceberg/TestManifestCleanup.java
index b5f6d05cc6a0..37ccbb8bb845 100644
--- a/core/src/test/java/org/apache/iceberg/TestManifestCleanup.java
+++ b/core/src/test/java/org/apache/iceberg/TestManifestCleanup.java
@@ -30,7 +30,7 @@
 public class TestManifestCleanup extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestManifestReaderStats.java b/core/src/test/java/org/apache/iceberg/TestManifestReaderStats.java
index 4c60a943f76c..175178e48167 100644
--- a/core/src/test/java/org/apache/iceberg/TestManifestReaderStats.java
+++ b/core/src/test/java/org/apache/iceberg/TestManifestReaderStats.java
@@ -39,7 +39,7 @@
 public class TestManifestReaderStats extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   private static final Map<Integer, Long> VALUE_COUNT = ImmutableMap.of(3, 3L);
diff --git a/core/src/test/java/org/apache/iceberg/TestManifestWriter.java b/core/src/test/java/org/apache/iceberg/TestManifestWriter.java
index eb7910a79fc5..7dd3ea2d1ce7 100644
--- a/core/src/test/java/org/apache/iceberg/TestManifestWriter.java
+++ b/core/src/test/java/org/apache/iceberg/TestManifestWriter.java
@@ -38,7 +38,7 @@
 public class TestManifestWriter extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   private static final int FILE_SIZE_CHECK_ROWS_DIVISOR = 250;
diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
index 0a3040939c04..80b71208a869 100644
--- a/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
+++ b/core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java
@@ -55,7 +55,7 @@ public class TestMetadataTableScans extends MetadataTableScanTestBase {
   private void preparePartitionedTable(boolean transactional) {
     preparePartitionedTableData(transactional);
 
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       if (transactional) {
         table
             .newRowDelta()
@@ -485,7 +485,7 @@ public void testPartitionsTableScanNoFilter() {
 
     CloseableIterable<ManifestEntry<?>> entries =
         PartitionsTable.planEntries((StaticTableScan) scanNoFilter);
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       assertThat(entries).hasSize(8);
     } else {
       assertThat(entries).hasSize(4);
@@ -510,7 +510,7 @@ public void testPartitionsTableScanWithProjection() {
     assertThat(scanWithProjection.schema().asStruct()).isEqualTo(expected);
     CloseableIterable<ManifestEntry<?>> entries =
         PartitionsTable.planEntries((StaticTableScan) scanWithProjection);
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       assertThat(entries).hasSize(8);
     } else {
       assertThat(entries).hasSize(4);
@@ -552,7 +552,7 @@ public void testPartitionsTableScanAndFilter() {
     TableScan scanAndEq = partitionsTable.newScan().filter(andEquals);
     CloseableIterable<ManifestEntry<?>> entries =
         PartitionsTable.planEntries((StaticTableScan) scanAndEq);
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       assertThat(entries).hasSize(2);
     } else {
       assertThat(entries).hasSize(1);
@@ -574,7 +574,7 @@ public void testPartitionsTableScanLtFilter() {
     TableScan scanLtAnd = partitionsTable.newScan().filter(ltAnd);
     CloseableIterable<ManifestEntry<?>> entries =
         PartitionsTable.planEntries((StaticTableScan) scanLtAnd);
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       assertThat(entries).hasSize(4);
     } else {
       assertThat(entries).hasSize(2);
@@ -598,7 +598,7 @@ public void testPartitionsTableScanOrFilter() {
 
     CloseableIterable<ManifestEntry<?>> entries =
         PartitionsTable.planEntries((StaticTableScan) scanOr);
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       assertThat(entries).hasSize(8);
     } else {
       assertThat(entries).hasSize(4);
@@ -619,7 +619,7 @@ public void testPartitionsScanNotFilter() {
     TableScan scanNot = partitionsTable.newScan().filter(not);
     CloseableIterable<ManifestEntry<?>> entries =
         PartitionsTable.planEntries((StaticTableScan) scanNot);
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       assertThat(entries).hasSize(4);
     } else {
       assertThat(entries).hasSize(2);
@@ -639,7 +639,7 @@ public void testPartitionsTableScanInFilter() {
     TableScan scanSet = partitionsTable.newScan().filter(set);
     CloseableIterable<ManifestEntry<?>> entries =
         PartitionsTable.planEntries((StaticTableScan) scanSet);
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       assertThat(entries).hasSize(4);
     } else {
       assertThat(entries).hasSize(2);
@@ -659,7 +659,7 @@ public void testPartitionsTableScanNotNullFilter() {
     TableScan scanUnary = partitionsTable.newScan().filter(unary);
     CloseableIterable<ManifestEntry<?>> entries =
         PartitionsTable.planEntries((StaticTableScan) scanUnary);
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       assertThat(entries).hasSize(8);
     } else {
       assertThat(entries).hasSize(4);
@@ -719,8 +719,7 @@ public void testFilesTableScanWithDroppedPartition() throws IOException {
 
   @TestTemplate
   public void testDeleteFilesTableSelection() throws IOException {
-    assumeThat(formatVersion).as("Only V2 Tables Support Deletes").isGreaterThanOrEqualTo(2);
-
+    assumeThat(formatVersion).as("Position deletes are not supported by V1 Tables").isNotEqualTo(1);
     table.newFastAppend().appendFile(FILE_A).commit();
 
     table.newRowDelta().addDeletes(FILE_A_DELETES).addDeletes(FILE_A2_DELETES).commit();
@@ -960,7 +959,7 @@ public void testPartitionSpecEvolutionAdditive() {
     TableScan scan = metadataTable.newScan().filter(filter);
     CloseableIterable<ManifestEntry<?>> entries =
         PartitionsTable.planEntries((StaticTableScan) scan);
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       // Four data files and delete files of old spec, one new data file of new spec
       assertThat(entries).hasSize(9);
     } else {
@@ -975,7 +974,7 @@ public void testPartitionSpecEvolutionAdditive() {
     scan = metadataTable.newScan().filter(filter);
     entries = PartitionsTable.planEntries((StaticTableScan) scan);
 
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       // 1 original data file and delete file written by old spec, plus 1 new data file written by
       // new spec
       assertThat(entries).hasSize(3);
@@ -1026,7 +1025,7 @@ public void testPartitionSpecEvolutionRemoval() {
     CloseableIterable<ManifestEntry<?>> entries =
         PartitionsTable.planEntries((StaticTableScan) scan);
 
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       // Four data and delete files of original spec, one data file written by new spec
       assertThat(entries).hasSize(9);
     } else {
@@ -1187,7 +1186,7 @@ public void testPartitionsTableScanWithPlanExecutor() {
                     }));
     CloseableIterable<ManifestEntry<?>> entries =
         PartitionsTable.planEntries((StaticTableScan) scan);
-    if (formatVersion == 2) {
+    if (formatVersion >= 2) {
       assertThat(entries).hasSize(8);
     } else {
       assertThat(entries).hasSize(4);
@@ -1366,7 +1365,7 @@ public void testAllManifestsTableSnapshotNot() {
 
   @TestTemplate
   public void testPositionDeletesWithFilter() {
-    assumeThat(formatVersion).as("Position deletes supported only for v2 tables").isEqualTo(2);
+    assumeThat(formatVersion).as("Position deletes are not supported by V1 Tables").isNotEqualTo(1);
     preparePartitionedTable();
 
     PositionDeletesTable positionDeletesTable = new PositionDeletesTable(table);
@@ -1429,7 +1428,7 @@ public void testPositionDeletesBaseTableFilterEntriesLevel() {
   }
 
   private void testPositionDeletesBaseTableFilter(boolean transactional) {
-    assumeThat(formatVersion).as("Position deletes supported only for v2 tables").isEqualTo(2);
+    assumeThat(formatVersion).as("Position deletes are not supported by V1 Tables").isNotEqualTo(1);
     preparePartitionedTable(transactional);
 
     PositionDeletesTable positionDeletesTable = new PositionDeletesTable(table);
@@ -1490,9 +1489,9 @@ private void testPositionDeletesBaseTableFilter(boolean transactional) {
 
   @TestTemplate
   public void testPositionDeletesWithBaseTableFilterNot() {
-    assumeThat(formatVersion).as("Position deletes supported only for v2 tables").isEqualTo(2);
-
-    // use identity rather than bucket partition spec,
+    assumeThat(formatVersion)
+        .as("Position deletes are not supported by V1 Tables")
+        .isNotEqualTo(1); // use identity rather than bucket partition spec,
     // as bucket.project does not support projecting notEq
     table.updateSpec().removeField("data_bucket").addField("id").commit();
     PartitionSpec spec = table.spec();
@@ -1574,7 +1573,7 @@ public void testPositionDeletesWithBaseTableFilterNot() {
 
   @TestTemplate
   public void testPositionDeletesResiduals() {
-    assumeThat(formatVersion).as("Position deletes supported only for v2 tables").isEqualTo(2);
+    assumeThat(formatVersion).as("Position deletes are not supported by V1 Tables").isNotEqualTo(1);
     preparePartitionedTable();
 
     PositionDeletesTable positionDeletesTable = new PositionDeletesTable(table);
@@ -1603,7 +1602,7 @@ public void testPositionDeletesResiduals() {
 
   @TestTemplate
   public void testPositionDeletesUnpartitioned() {
-    assumeThat(formatVersion).as("Position deletes supported only for v2 tables").isEqualTo(2);
+    assumeThat(formatVersion).as("Position deletes are not supported by V1 Tables").isNotEqualTo(1);
     table.updateSpec().removeField(Expressions.bucket("data", BUCKETS_NUMBER)).commit();
 
     assertThat(table.spec().fields()).as("Table should now be unpartitioned").hasSize(0);
@@ -1694,8 +1693,7 @@ public void testPositionDeletesUnpartitioned() {
 
   @TestTemplate
   public void testPositionDeletesManyColumns() {
-    assumeThat(formatVersion).as("Position deletes supported only for v2 tables").isEqualTo(2);
-
+    assumeThat(formatVersion).as("Position deletes are not supported by V1 Tables").isNotEqualTo(1);
     UpdateSchema updateSchema = table.updateSchema();
     for (int i = 0; i <= 2000; i++) {
       updateSchema.addColumn(String.valueOf(i), Types.IntegerType.get());
diff --git a/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java b/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java
index a2e5386d29df..ac96642319a3 100644
--- a/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java
+++ b/core/src/test/java/org/apache/iceberg/TestMetadataTableScansWithPartitionEvolution.java
@@ -163,8 +163,7 @@ public void testPartitionsTableScanWithAddPartitionOnNestedField() {
 
   @TestTemplate
   public void testPositionDeletesPartitionSpecRemoval() {
-    assumeThat(formatVersion).as("Position deletes supported only for v2 tables").isEqualTo(2);
-
+    assumeThat(formatVersion).as("Position deletes are not supported by V1 Tables").isNotEqualTo(1);
     table.updateSpec().removeField("id").commit();
 
     DeleteFile deleteFile = newDeleteFile(table.ops().current().spec().specId(), "nested.id=1");
diff --git a/core/src/test/java/org/apache/iceberg/TestMetrics.java b/core/src/test/java/org/apache/iceberg/TestMetrics.java
index b95b92979f91..2c4849135f64 100644
--- a/core/src/test/java/org/apache/iceberg/TestMetrics.java
+++ b/core/src/test/java/org/apache/iceberg/TestMetrics.java
@@ -69,7 +69,7 @@ public abstract class TestMetrics {
 
   @Parameters(name = "formatVersion = {0}")
   public static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TempDir public Path temp;
diff --git a/core/src/test/java/org/apache/iceberg/TestMetricsModes.java b/core/src/test/java/org/apache/iceberg/TestMetricsModes.java
index 31ae459df506..00e961097c34 100644
--- a/core/src/test/java/org/apache/iceberg/TestMetricsModes.java
+++ b/core/src/test/java/org/apache/iceberg/TestMetricsModes.java
@@ -47,7 +47,7 @@ public class TestMetricsModes {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TempDir private Path temp;
diff --git a/core/src/test/java/org/apache/iceberg/TestMicroBatchBuilder.java b/core/src/test/java/org/apache/iceberg/TestMicroBatchBuilder.java
index 733bb0bb38fd..8c6f18619ac0 100644
--- a/core/src/test/java/org/apache/iceberg/TestMicroBatchBuilder.java
+++ b/core/src/test/java/org/apache/iceberg/TestMicroBatchBuilder.java
@@ -34,7 +34,7 @@
 public class TestMicroBatchBuilder extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @BeforeEach
diff --git a/core/src/test/java/org/apache/iceberg/TestPartitionSpecInfo.java b/core/src/test/java/org/apache/iceberg/TestPartitionSpecInfo.java
index ff54929504ee..a324b8af2e70 100644
--- a/core/src/test/java/org/apache/iceberg/TestPartitionSpecInfo.java
+++ b/core/src/test/java/org/apache/iceberg/TestPartitionSpecInfo.java
@@ -47,7 +47,7 @@ public class TestPartitionSpecInfo {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @Parameter private int formatVersion;
diff --git a/core/src/test/java/org/apache/iceberg/TestReplaceTransaction.java b/core/src/test/java/org/apache/iceberg/TestReplaceTransaction.java
index 3a6d2017eb82..b1b481dd5305 100644
--- a/core/src/test/java/org/apache/iceberg/TestReplaceTransaction.java
+++ b/core/src/test/java/org/apache/iceberg/TestReplaceTransaction.java
@@ -50,7 +50,7 @@
 public class TestReplaceTransaction extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestRewriteManifests.java b/core/src/test/java/org/apache/iceberg/TestRewriteManifests.java
index 176f61079fc3..f1d23de32a42 100644
--- a/core/src/test/java/org/apache/iceberg/TestRewriteManifests.java
+++ b/core/src/test/java/org/apache/iceberg/TestRewriteManifests.java
@@ -48,7 +48,7 @@
 public class TestRewriteManifests extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestScanSummary.java b/core/src/test/java/org/apache/iceberg/TestScanSummary.java
index b326274842a3..10c22b759988 100644
--- a/core/src/test/java/org/apache/iceberg/TestScanSummary.java
+++ b/core/src/test/java/org/apache/iceberg/TestScanSummary.java
@@ -39,7 +39,7 @@
 public class TestScanSummary extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestScansAndSchemaEvolution.java b/core/src/test/java/org/apache/iceberg/TestScansAndSchemaEvolution.java
index 0d5b1bc7066b..3fc691ce942a 100644
--- a/core/src/test/java/org/apache/iceberg/TestScansAndSchemaEvolution.java
+++ b/core/src/test/java/org/apache/iceberg/TestScansAndSchemaEvolution.java
@@ -55,7 +55,7 @@ public class TestScansAndSchemaEvolution {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @Parameter private int formatVersion;
diff --git a/core/src/test/java/org/apache/iceberg/TestSchemaAndMappingUpdate.java b/core/src/test/java/org/apache/iceberg/TestSchemaAndMappingUpdate.java
index 1bd1761ffc60..111693792f78 100644
--- a/core/src/test/java/org/apache/iceberg/TestSchemaAndMappingUpdate.java
+++ b/core/src/test/java/org/apache/iceberg/TestSchemaAndMappingUpdate.java
@@ -43,7 +43,7 @@
 public class TestSchemaAndMappingUpdate extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestSchemaID.java b/core/src/test/java/org/apache/iceberg/TestSchemaID.java
index 33ac84d20801..e3c58af1cbe3 100644
--- a/core/src/test/java/org/apache/iceberg/TestSchemaID.java
+++ b/core/src/test/java/org/apache/iceberg/TestSchemaID.java
@@ -36,7 +36,7 @@ public class TestSchemaID extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestSetPartitionStatistics.java b/core/src/test/java/org/apache/iceberg/TestSetPartitionStatistics.java
index 3ebe047e90b7..9504f78f5345 100644
--- a/core/src/test/java/org/apache/iceberg/TestSetPartitionStatistics.java
+++ b/core/src/test/java/org/apache/iceberg/TestSetPartitionStatistics.java
@@ -29,7 +29,7 @@
 public class TestSetPartitionStatistics extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestSetStatistics.java b/core/src/test/java/org/apache/iceberg/TestSetStatistics.java
index 41941e3c6630..03c644117a95 100644
--- a/core/src/test/java/org/apache/iceberg/TestSetStatistics.java
+++ b/core/src/test/java/org/apache/iceberg/TestSetStatistics.java
@@ -31,7 +31,7 @@
 public class TestSetStatistics extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestSnapshot.java b/core/src/test/java/org/apache/iceberg/TestSnapshot.java
index 2ec6abd4e428..8a30036f3242 100644
--- a/core/src/test/java/org/apache/iceberg/TestSnapshot.java
+++ b/core/src/test/java/org/apache/iceberg/TestSnapshot.java
@@ -33,7 +33,7 @@
 public class TestSnapshot extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestSnapshotLoading.java b/core/src/test/java/org/apache/iceberg/TestSnapshotLoading.java
index 89312201265d..96dde0f48eb7 100644
--- a/core/src/test/java/org/apache/iceberg/TestSnapshotLoading.java
+++ b/core/src/test/java/org/apache/iceberg/TestSnapshotLoading.java
@@ -43,7 +43,7 @@ public class TestSnapshotLoading extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   private Snapshot currentSnapshot;
diff --git a/core/src/test/java/org/apache/iceberg/TestSnapshotManager.java b/core/src/test/java/org/apache/iceberg/TestSnapshotManager.java
index 88233dd99097..61dd082701c2 100644
--- a/core/src/test/java/org/apache/iceberg/TestSnapshotManager.java
+++ b/core/src/test/java/org/apache/iceberg/TestSnapshotManager.java
@@ -51,7 +51,7 @@ public class TestSnapshotManager extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestSnapshotSelection.java b/core/src/test/java/org/apache/iceberg/TestSnapshotSelection.java
index 7ce59e9df1c9..6d339c0cbeaf 100644
--- a/core/src/test/java/org/apache/iceberg/TestSnapshotSelection.java
+++ b/core/src/test/java/org/apache/iceberg/TestSnapshotSelection.java
@@ -33,7 +33,7 @@
 public class TestSnapshotSelection extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestSnapshotSummary.java b/core/src/test/java/org/apache/iceberg/TestSnapshotSummary.java
index 053a9c374178..529e0cc614f6 100644
--- a/core/src/test/java/org/apache/iceberg/TestSnapshotSummary.java
+++ b/core/src/test/java/org/apache/iceberg/TestSnapshotSummary.java
@@ -32,7 +32,7 @@ public class TestSnapshotSummary extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestSortOrder.java b/core/src/test/java/org/apache/iceberg/TestSortOrder.java
index a3ba69a808b3..ad773192b417 100644
--- a/core/src/test/java/org/apache/iceberg/TestSortOrder.java
+++ b/core/src/test/java/org/apache/iceberg/TestSortOrder.java
@@ -76,7 +76,7 @@ public class TestSortOrder {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @Parameter private int formatVersion;
diff --git a/core/src/test/java/org/apache/iceberg/TestSplitPlanning.java b/core/src/test/java/org/apache/iceberg/TestSplitPlanning.java
index 04bb2ae215d8..ea2142982382 100644
--- a/core/src/test/java/org/apache/iceberg/TestSplitPlanning.java
+++ b/core/src/test/java/org/apache/iceberg/TestSplitPlanning.java
@@ -58,7 +58,7 @@ public class TestSplitPlanning extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @Override
diff --git a/core/src/test/java/org/apache/iceberg/TestTableMetadataSerialization.java b/core/src/test/java/org/apache/iceberg/TestTableMetadataSerialization.java
index f47968434bfe..94a3d35b35a6 100644
--- a/core/src/test/java/org/apache/iceberg/TestTableMetadataSerialization.java
+++ b/core/src/test/java/org/apache/iceberg/TestTableMetadataSerialization.java
@@ -35,7 +35,7 @@
 public class TestTableMetadataSerialization extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestTableUpdatePartitionSpec.java b/core/src/test/java/org/apache/iceberg/TestTableUpdatePartitionSpec.java
index a4e587068e74..482514c40093 100644
--- a/core/src/test/java/org/apache/iceberg/TestTableUpdatePartitionSpec.java
+++ b/core/src/test/java/org/apache/iceberg/TestTableUpdatePartitionSpec.java
@@ -36,7 +36,7 @@ public class TestTableUpdatePartitionSpec extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @BeforeEach
diff --git a/core/src/test/java/org/apache/iceberg/TestTimestampPartitions.java b/core/src/test/java/org/apache/iceberg/TestTimestampPartitions.java
index 08714dec01f1..656dd782cfe4 100644
--- a/core/src/test/java/org/apache/iceberg/TestTimestampPartitions.java
+++ b/core/src/test/java/org/apache/iceberg/TestTimestampPartitions.java
@@ -35,7 +35,7 @@
 public class TestTimestampPartitions extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestTransaction.java b/core/src/test/java/org/apache/iceberg/TestTransaction.java
index 393494da1283..8fed7134fae1 100644
--- a/core/src/test/java/org/apache/iceberg/TestTransaction.java
+++ b/core/src/test/java/org/apache/iceberg/TestTransaction.java
@@ -42,7 +42,7 @@
 public class TestTransaction extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestUpdatePartitionSpec.java b/core/src/test/java/org/apache/iceberg/TestUpdatePartitionSpec.java
index 33b003cfd8c5..97f695315e16 100644
--- a/core/src/test/java/org/apache/iceberg/TestUpdatePartitionSpec.java
+++ b/core/src/test/java/org/apache/iceberg/TestUpdatePartitionSpec.java
@@ -54,7 +54,7 @@ public class TestUpdatePartitionSpec extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/TestWapWorkflow.java b/core/src/test/java/org/apache/iceberg/TestWapWorkflow.java
index 58b83f43f399..b1e8e57850e8 100644
--- a/core/src/test/java/org/apache/iceberg/TestWapWorkflow.java
+++ b/core/src/test/java/org/apache/iceberg/TestWapWorkflow.java
@@ -35,7 +35,7 @@
 public class TestWapWorkflow extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @BeforeEach
diff --git a/core/src/test/java/org/apache/iceberg/actions/TestSizeBasedRewriter.java b/core/src/test/java/org/apache/iceberg/actions/TestSizeBasedRewriter.java
index a9a23d587ac9..77d16d3bc821 100644
--- a/core/src/test/java/org/apache/iceberg/actions/TestSizeBasedRewriter.java
+++ b/core/src/test/java/org/apache/iceberg/actions/TestSizeBasedRewriter.java
@@ -41,7 +41,7 @@ public class TestSizeBasedRewriter extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/java/org/apache/iceberg/io/TestOutputFileFactory.java b/core/src/test/java/org/apache/iceberg/io/TestOutputFileFactory.java
index ceffeb3749da..bb98925f504c 100644
--- a/core/src/test/java/org/apache/iceberg/io/TestOutputFileFactory.java
+++ b/core/src/test/java/org/apache/iceberg/io/TestOutputFileFactory.java
@@ -40,7 +40,7 @@ public class TestOutputFileFactory extends TestBase {
 
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   private static final int PARTITION_ID = 1;
diff --git a/core/src/test/java/org/apache/iceberg/mapping/TestMappingUpdates.java b/core/src/test/java/org/apache/iceberg/mapping/TestMappingUpdates.java
index b9ae9dc273f2..2161b40a60b5 100644
--- a/core/src/test/java/org/apache/iceberg/mapping/TestMappingUpdates.java
+++ b/core/src/test/java/org/apache/iceberg/mapping/TestMappingUpdates.java
@@ -36,7 +36,7 @@
 public class TestMappingUpdates extends TestBase {
   @Parameters(name = "formatVersion = {0}")
   protected static List<Object> parameters() {
-    return Arrays.asList(1, 2);
+    return Arrays.asList(1, 2, 3);
   }
 
   @TestTemplate
diff --git a/core/src/test/resources/TableMetadataUnsupportedVersion.json b/core/src/test/resources/TableMetadataUnsupportedVersion.json
index 0633a71d24d5..c40a0c9cd5ae 100644
--- a/core/src/test/resources/TableMetadataUnsupportedVersion.json
+++ b/core/src/test/resources/TableMetadataUnsupportedVersion.json
@@ -1,5 +1,5 @@
 {
-  "format-version": 3,
+  "format-version": 4,
   "table-uuid": "d20125c8-7284-442c-9aea-15fee620737c",
   "location": "s3://bucket/test/location",
   "last-updated-ms": 1602638573874,
@@ -33,4 +33,4 @@
   "properties": {},
   "current-snapshot-id": -1,
   "snapshots": []
-}
\ No newline at end of file
+}