Address new comments

Author: ajantha-bhat

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

@@ -108,19 +108,7 @@ private PartitionStatsHandler() {}
   @Deprecated
   public static Schema schema(StructType unifiedPartitionType) {
     Preconditions.checkState(!unifiedPartitionType.fields().isEmpty(), "Table must be partitioned");
-    return new Schema(
-        NestedField.required(PARTITION_FIELD_ID, PARTITION_FIELD_NAME, unifiedPartitionType),
-        SPEC_ID,
-        DATA_RECORD_COUNT,
-        DATA_FILE_COUNT,
-        TOTAL_DATA_FILE_SIZE_IN_BYTES,
-        POSITION_DELETE_RECORD_COUNT,
-        POSITION_DELETE_FILE_COUNT,
-        EQUALITY_DELETE_RECORD_COUNT,
-        EQUALITY_DELETE_FILE_COUNT,
-        TOTAL_RECORD_COUNT,
-        LAST_UPDATED_AT,
-        LAST_UPDATED_SNAPSHOT_ID);
+    return v2Schema(unifiedPartitionType);
   }
 
   /**
@@ -133,11 +121,35 @@ public static Schema schema(StructType unifiedPartitionType) {
    */
   public static Schema schema(StructType unifiedPartitionType, int formatVersion) {
     Preconditions.checkState(!unifiedPartitionType.fields().isEmpty(), "Table must be partitioned");
+    Preconditions.checkState(
+        formatVersion > 0 && formatVersion <= TableMetadata.SUPPORTED_TABLE_FORMAT_VERSION,
+        "Invalid format version: %d",
+        formatVersion);
 
     if (formatVersion <= 2) {
-      return schema(unifiedPartitionType);
+      return v2Schema(unifiedPartitionType);
     }
 
+    return v3Schema(unifiedPartitionType);
+  }
+
+  private static Schema v2Schema(StructType unifiedPartitionType) {
+    return new Schema(
+        NestedField.required(PARTITION_FIELD_ID, PARTITION_FIELD_NAME, unifiedPartitionType),
+        SPEC_ID,
+        DATA_RECORD_COUNT,
+        DATA_FILE_COUNT,
+        TOTAL_DATA_FILE_SIZE_IN_BYTES,
+        POSITION_DELETE_RECORD_COUNT,
+        POSITION_DELETE_FILE_COUNT,
+        EQUALITY_DELETE_RECORD_COUNT,
+        EQUALITY_DELETE_FILE_COUNT,
+        TOTAL_RECORD_COUNT,
+        LAST_UPDATED_AT,
+        LAST_UPDATED_SNAPSHOT_ID);
+  }
+
+  private static Schema v3Schema(StructType unifiedPartitionType) {
     return new Schema(
         NestedField.required(PARTITION_FIELD_ID, PARTITION_FIELD_NAME, unifiedPartitionType),
         SPEC_ID,

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

@@ -57,6 +57,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.types.Comparators;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ContentFileUtil;
 import org.assertj.core.groups.Tuple;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestTemplate;
@@ -414,9 +415,8 @@ public void testPartitionStats() throws Exception {
             0));
 
     DeleteFile posDeletes = null;
-    DeleteFile deleteVectors = null;
-    if (formatVersion == 3) {
-      deleteVectors = commitDVs(testTable, dataFile3);
+    if (formatVersion >= 3) {
+      posDeletes = commitDVs(testTable, dataFile3);
     } else if (formatVersion == 2) {
       posDeletes = commitPositionDeletes(testTable);
     }
@@ -463,14 +463,14 @@ public void testPartitionStats() throws Exception {
             3 * dataFile3.recordCount(),
             3,
             3 * dataFile3.fileSizeInBytes(),
-            formatVersion == 3 ? deleteVectors.recordCount() : posDeletes.recordCount(),
-            formatVersion == 3 ? 0 : 1, // pos delete file count
+            posDeletes.recordCount(),
+            ContentFileUtil.isDV(posDeletes) ? 0 : 1, // pos delete file count
             0L,
             0,
             null,
             snapshot2.timestampMillis(),
             snapshot2.snapshotId(),
-            formatVersion == 3 ? 1 : 0), // dv count
+            ContentFileUtil.isDV(posDeletes) ? 1 : 0), // dv count
         Tuple.tuple(
             partitionRecord(partitionType, "bar", "B"),
             0,