Core: Validate transforms while building partition type

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

@@ -31,6 +31,7 @@
 import org.apache.iceberg.transforms.PartitionSpecVisitor;
 import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.transforms.UnknownTransform;
 import org.apache.iceberg.types.Types.NestedField;
 import org.apache.iceberg.types.Types.StructType;
 
@@ -199,6 +200,11 @@ public Void alwaysNull(int fieldId, String sourceName, int sourceId) {
    * @return the constructed common partition type
    */
   public static StructType partitionType(Table table) {
+    // we currently don't know the output type of unknown transforms
+    List<Transform<?, ?>> unknownTransforms = collectUnknownTransforms(table);
+    ValidationException.check(unknownTransforms.isEmpty(),
+        "Cannot build table partition type, unknown transforms: %s", unknownTransforms);
+
     if (table.specs().size() == 1) {
       return table.spec().partitionType();
     }
@@ -237,6 +243,19 @@ public static StructType partitionType(Table table) {
     return StructType.of(sortedStructFields);
   }
 
+  private static List<Transform<?, ?>> collectUnknownTransforms(Table table) {
+    List<Transform<?, ?>> unknownTransforms = Lists.newArrayList();
+
+    table.specs().values().forEach(spec -> {
+      spec.fields().stream()
+          .map(PartitionField::transform)
+          .filter(transform -> transform instanceof UnknownTransform)
+          .forEach(unknownTransforms::add);
+    });
+
+    return unknownTransforms;
+  }
+
   private static boolean equivalentIgnoringNames(PartitionField field, PartitionField anotherField) {
     return field.fieldId() == anotherField.fieldId() &&
         field.sourceId() == anotherField.sourceId() &&

spark3/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java

@@ -24,9 +24,16 @@
 import java.util.Map;
 import java.util.Random;
 import java.util.concurrent.ThreadLocalRandom;
+import org.apache.iceberg.AssertHelpers;
 import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -254,6 +261,32 @@ public void testEntriesMetadataTable() throws ParseException {
     }
   }
 
+  @Test
+  public void testMetadataTablesWithUnknownTransforms() {
+    sql("CREATE TABLE %s (id bigint NOT NULL, category string, data string) USING iceberg", tableName);
+    initTable();
+
+    sql("INSERT INTO TABLE %s VALUES (1, 'a1', 'b1')", tableName);
+
+    Table table = validationCatalog.loadTable(tableIdent);
+
+    PartitionSpec unknownSpec = PartitionSpecParser.fromJson(table.schema(),
+        "{ \"spec-id\": 1, \"fields\": [ { \"name\": \"id_zero\", \"transform\": \"zero\", \"source-id\": 1 } ] }");
+
+    // replace the table spec to include an unknown transform
+    TableOperations ops = ((HasTableOperations) table).operations();
+    TableMetadata base = ops.current();
+    ops.commit(base, base.updatePartitionSpec(unknownSpec));
+
+    sql("REFRESH TABLE %s", tableName);
+
+    for (MetadataTableType tableType : Arrays.asList(FILES, ALL_DATA_FILES, ENTRIES, ALL_ENTRIES)) {
+      AssertHelpers.assertThrows("Should complain about the partition type",
+          ValidationException.class, "Cannot build table partition type, unknown transforms",
+          () -> loadMetadataTable(tableType));
+    }
+  }
+
   private void assertPartitions(List<Object[]> expectedPartitions, String expectedTypeAsString,
                                 MetadataTableType tableType) throws ParseException {
     Dataset<Row> df = loadMetadataTable(tableType);