Replace dropped PartitonField with AnyType

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

@@ -130,7 +130,14 @@ public StructType partitionType() {
 
           for (PartitionField field : fields) {
             Type sourceType = schema.findType(field.sourceId());
-            Type resultType = field.transform().getResultType(sourceType);
+
+            final Type resultType;
+            if (sourceType == null) {
+              resultType = Types.AnyType.get();
+            } else {
+              resultType = field.transform().getResultType(sourceType);
+            }
+
             structFields.add(Types.NestedField.optional(field.fieldId(), field.name(), resultType));
           }
 

api/src/main/java/org/apache/iceberg/types/Type.java

@@ -45,7 +45,8 @@ enum TypeID {
     DECIMAL(BigDecimal.class),
     STRUCT(StructLike.class),
     LIST(List.class),
-    MAP(Map.class);
+    MAP(Map.class),
+    ANY(Object.class);
 
     private final Class<?> javaClass;
 

api/src/main/java/org/apache/iceberg/types/Types.java

@@ -55,6 +55,7 @@ private Types() {}
           .put(StringType.get().toString(), StringType.get())
           .put(UUIDType.get().toString(), UUIDType.get())
           .put(BinaryType.get().toString(), BinaryType.get())
+          .put(AnyType.get().toString(), AnyType.get())
           .buildOrThrow();
 
   private static final Pattern FIXED = Pattern.compile("fixed\\[\\s*(\\d+)\\s*\\]");
@@ -412,6 +413,24 @@ public String toString() {
     }
   }
 
+  public static class AnyType extends PrimitiveType {
+    private static final AnyType INSTANCE = new AnyType();
+
+    public static AnyType get() {
+      return INSTANCE;
+    }
+
+    @Override
+    public TypeID typeId() {
+      return TypeID.ANY;
+    }
+
+    @Override
+    public String toString() {
+      return "any";
+    }
+  }
+
   public static class DecimalType extends PrimitiveType {
     public static DecimalType of(int precision, int scale) {
       return new DecimalType(precision, scale);

core/src/main/java/org/apache/iceberg/avro/AvroSchemaUtil.java

@@ -188,6 +188,8 @@ static Schema toOption(Schema schema) {
       Preconditions.checkArgument(
           isOptionSchema(schema), "Union schemas are not supported: %s", schema);
       return schema;
+    } else if (schema.getType() == Schema.Type.NULL) {
+      return schema;
     } else {
       return Schema.createUnion(NULL, schema);
     }

core/src/main/java/org/apache/iceberg/avro/TypeToSchema.java

@@ -49,6 +49,7 @@ abstract class TypeToSchema extends TypeUtil.SchemaVisitor<Schema> {
   private static final Schema UUID_SCHEMA =
       LogicalTypes.uuid().addToSchema(Schema.createFixed("uuid_fixed", null, null, 16));
   private static final Schema BINARY_SCHEMA = Schema.create(Schema.Type.BYTES);
+  private static final Schema NULL_SCHEMA = Schema.create(Schema.Type.NULL);
 
   static {
     TIMESTAMP_SCHEMA.addProp(AvroSchemaUtil.ADJUST_TO_UTC_PROP, false);
@@ -243,6 +244,9 @@ public Schema primitive(Type.PrimitiveType primitive) {
                         null,
                         TypeUtil.decimalRequiredBytes(decimal.precision())));
         break;
+      case ANY:
+        primitiveSchema = NULL_SCHEMA;
+        break;
       default:
         throw new UnsupportedOperationException("Unsupported type ID: " + primitive.typeId());
     }

spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestAlterTablePartitionFields.java

@@ -436,6 +436,20 @@ public void testReplacePartitionAndRename() {
         .isEqualTo(expected);
   }
 
+  @TestTemplate
+  public void testDropPartitionAndUnderlyingField() {
+    sql(
+        "CREATE TABLE %s (col0 BIGINT, col1 BIGINT, col2 BIGINT) USING ICEBERG TBLPROPERTIES ('format-version' = %d, 'write.delete.mode' = 'merge-on-read')",
+        tableName, formatVersion);
+    sql("INSERT INTO %s VALUES (1, 11, 21)", tableName);
+    sql("ALTER TABLE %s ADD PARTITION FIELD col2", tableName);
+    sql("INSERT INTO %s VALUES (2, 12, 22)", tableName);
+    sql("ALTER TABLE %s DROP PARTITION FIELD col2", tableName);
+    sql("INSERT INTO %s VALUES (3, 13, 23)", tableName);
+    sql("ALTER TABLE %s DROP COLUMN col2", tableName);
+    sql("SELECT * FROM %s", tableName);
+  }
+
   @TestTemplate
   public void testReplaceNamedPartition() {
     createTable("id bigint NOT NULL, category string, ts timestamp, data string");