diff --git a/api/src/main/java/org/apache/iceberg/PartitionSpec.java b/api/src/main/java/org/apache/iceberg/PartitionSpec.java
index 9b74893f1831..80985dcb79a9 100644
--- a/api/src/main/java/org/apache/iceberg/PartitionSpec.java
+++ b/api/src/main/java/org/apache/iceberg/PartitionSpec.java
@@ -40,6 +40,7 @@
 import org.apache.iceberg.transforms.Transforms;
 import org.apache.iceberg.transforms.UnknownTransform;
 import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.StructType;
 
@@ -624,6 +625,7 @@ PartitionSpec buildUnchecked() {
   }
 
   static void checkCompatibility(PartitionSpec spec, Schema schema) {
+    final Map<Integer, Integer> parents = TypeUtil.indexParents(schema.asStruct());
     for (PartitionField field : spec.fields) {
       Type sourceType = schema.findType(field.sourceId());
       Transform<?, ?> transform = field.transform();
@@ -644,6 +646,15 @@ static void checkCompatibility(PartitionSpec spec, Schema schema) {
             "Invalid source type %s for transform: %s",
             sourceType,
             transform);
+        // The only valid parent types for a PartitionField are StructTypes. This must be checked
+        // recursively.
+        Integer parentId = parents.get(field.sourceId());
+        while (parentId != null) {
+          Type parentType = schema.findType(parentId);
+          ValidationException.check(
+              parentType.isStructType(), "Invalid partition field parent: %s", parentType);
+          parentId = parents.get(parentId);
+        }
       }
     }
   }
diff --git a/api/src/test/java/org/apache/iceberg/TestPartitionSpecValidation.java b/api/src/test/java/org/apache/iceberg/TestPartitionSpecValidation.java
index ee71d39bb2db..b8e16a9ee45e 100644
--- a/api/src/test/java/org/apache/iceberg/TestPartitionSpecValidation.java
+++ b/api/src/test/java/org/apache/iceberg/TestPartitionSpecValidation.java
@@ -340,4 +340,105 @@ private static Object[][] unsupportedFieldsProvider() {
       {10, "unknown_partition1", "Invalid source type unknown for transform: bucket[5]"}
     };
   }
+
+  @Test
+  void testSourceIdNotFound() {
+    assertThatThrownBy(
+            () ->
+                PartitionSpec.builderFor(SCHEMA)
+                    .add(99, 1000, "Test", Transforms.identity())
+                    .build())
+        .isInstanceOf(ValidationException.class)
+        .hasMessage("Cannot find source column for partition field: 1000: Test: identity(99)");
+  }
+
+  @Test
+  void testPartitionFieldInStruct() {
+    final Schema schema =
+        new Schema(
+            NestedField.required(SCHEMA.highestFieldId() + 1, "MyStruct", SCHEMA.asStruct()));
+    PartitionSpec.builderFor(schema).identity("MyStruct.id").build();
+  }
+
+  @Test
+  void testPartitionFieldInStructInStruct() {
+    final Schema schema =
+        new Schema(
+            NestedField.optional(
+                SCHEMA.highestFieldId() + 2,
+                "Outer",
+                Types.StructType.of(
+                    NestedField.required(
+                        SCHEMA.highestFieldId() + 1, "Inner", SCHEMA.asStruct()))));
+    PartitionSpec.builderFor(schema).identity("Outer.Inner.id").build();
+  }
+
+  @Test
+  void testPartitionFieldInList() {
+    final Schema schema =
+        new Schema(
+            NestedField.required(
+                2, "MyList", Types.ListType.ofRequired(1, Types.IntegerType.get())));
+    assertThatThrownBy(() -> PartitionSpec.builderFor(schema).identity("MyList.element").build())
+        .isInstanceOf(ValidationException.class)
+        .hasMessage("Invalid partition field parent: list<int>");
+  }
+
+  @Test
+  void testPartitionFieldInStructInList() {
+    final Schema schema =
+        new Schema(
+            NestedField.required(
+                3,
+                "MyList",
+                Types.ListType.ofRequired(
+                    2,
+                    Types.StructType.of(NestedField.optional(1, "Foo", Types.IntegerType.get())))));
+    assertThatThrownBy(
+            () -> PartitionSpec.builderFor(schema).identity("MyList.element.Foo").build())
+        .isInstanceOf(ValidationException.class)
+        .hasMessage("Invalid partition field parent: list<struct<1: Foo: optional int>>");
+  }
+
+  @Test
+  void testPartitionFieldInMap() {
+    final Schema schema =
+        new Schema(
+            NestedField.required(
+                3,
+                "MyMap",
+                Types.MapType.ofRequired(1, 2, Types.IntegerType.get(), Types.IntegerType.get())));
+
+    assertThatThrownBy(() -> PartitionSpec.builderFor(schema).identity("MyMap.key").build())
+        .isInstanceOf(ValidationException.class)
+        .hasMessage("Invalid partition field parent: map<int, int>");
+
+    assertThatThrownBy(() -> PartitionSpec.builderFor(schema).identity("MyMap.value").build())
+        .isInstanceOf(ValidationException.class)
+        .hasMessage("Invalid partition field parent: map<int, int>");
+  }
+
+  @Test
+  void testPartitionFieldInStructInMap() {
+    final Schema schema =
+        new Schema(
+            NestedField.required(
+                5,
+                "MyMap",
+                Types.MapType.ofRequired(
+                    3,
+                    4,
+                    Types.StructType.of(NestedField.optional(1, "Foo", Types.IntegerType.get())),
+                    Types.StructType.of(NestedField.optional(2, "Bar", Types.IntegerType.get())))));
+
+    assertThatThrownBy(() -> PartitionSpec.builderFor(schema).identity("MyMap.key.Foo").build())
+        .isInstanceOf(ValidationException.class)
+        .hasMessage(
+            "Invalid partition field parent: map<struct<1: Foo: optional int>, struct<2: Bar: optional int>>");
+
+    assertThatThrownBy(() -> PartitionSpec.builderFor(schema).identity("MyMap.value.Bar").build())
+        .isInstanceOf(ValidationException.class)
+        .hasMessage(
+            "Invalid partition field parent: map<struct<1: Foo: optional int>, struct<2: Bar: optional int>>");
+  }
 }