OTF-4251: Code changes to ingore removed partitions during latest schema validation

Author: Ashok Vengala

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

@@ -131,6 +131,12 @@ public StructType partitionType() {
           for (PartitionField field : fields) {
             Type sourceType = schema.findType(field.sourceId());
             Type resultType = field.transform().getResultType(sourceType);
+
+            // When the source field has been dropped we cannot determine the type
+            if (sourceType == null) {
+              resultType = Types.UnknownType.get();
+            }
+
             structFields.add(Types.NestedField.optional(field.fieldId(), field.name(), resultType));
           }
 
@@ -613,8 +619,12 @@ Builder add(int sourceId, int fieldId, String name, Transform<?, ?> transform) {
     }
 
     public PartitionSpec build() {
+      return build(false);
+    }
+
+    public PartitionSpec build(boolean allowMissingFields) {
       PartitionSpec spec = buildUnchecked();
-      checkCompatibility(spec, schema);
+      checkCompatibility(spec, schema, allowMissingFields);
       return spec;
     }
 
@@ -624,9 +634,17 @@ PartitionSpec buildUnchecked() {
   }
 
   static void checkCompatibility(PartitionSpec spec, Schema schema) {
+    checkCompatibility(spec, schema, false);
+  }
+
+  static void checkCompatibility(PartitionSpec spec, Schema schema, boolean allowMissingFields) {
     for (PartitionField field : spec.fields) {
       Type sourceType = schema.findType(field.sourceId());
       Transform<?, ?> transform = field.transform();
+      // In the case the underlying field is dropped, we cannot check if they are compatible
+      if (allowMissingFields && sourceType == null) {
+        continue;
+      }
       // In the case of a Version 1 partition-spec field gets deleted,
       // it is replaced with a void transform, see:
       // https://iceberg.apache.org/spec/#partition-transforms

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

@@ -46,6 +46,10 @@ public PartitionSpec bind(Schema schema) {
     return copyToBuilder(schema).build();
   }
 
+  public PartitionSpec bind(Schema schema, boolean ignoreMissingFields) {
+    return copyToBuilder(schema).build(ignoreMissingFields);
+  }
+
   PartitionSpec bindUnchecked(Schema schema) {
     return copyToBuilder(schema).buildUnchecked();
   }

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

@@ -68,7 +68,7 @@ public static String toJson(UnboundPartitionSpec spec, boolean pretty) {
   }
 
   public static PartitionSpec fromJson(Schema schema, JsonNode json) {
-    return fromJson(json).bind(schema);
+    return fromJson(json).bind(schema, true);
   }
 
   public static UnboundPartitionSpec fromJson(JsonNode json) {

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

@@ -239,7 +239,8 @@ public static StructType groupingKeyType(Schema schema, Collection<PartitionSpec
    */
   public static StructType partitionType(Table table) {
     Collection<PartitionSpec> specs = table.specs().values();
-    return buildPartitionProjectionType("table partition", specs, allFieldIds(specs));
+    return buildPartitionProjectionType(
+        "table partition", specs, allActiveFieldIds(table.schema(), specs));
   }
 
   /**
@@ -346,10 +347,11 @@ private static boolean compatibleTransforms(Transform<?, ?> t1, Transform<?, ?>
         || t2.equals(Transforms.alwaysNull());
   }
 
-  // collects IDs of all partition field used across specs
-  private static Set<Integer> allFieldIds(Collection<PartitionSpec> specs) {
+  // collects IDs of all partition field used across specs that are in the current schema
+  private static Set<Integer> allActiveFieldIds(Schema schema, Collection<PartitionSpec> specs) {
     return FluentIterable.from(specs)
         .transformAndConcat(PartitionSpec::fields)
+        .filter(field -> schema.findField(field.sourceId()) != null)
         .transform(PartitionField::fieldId)
         .toSet();
   }

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

@@ -172,7 +172,7 @@ public void testPartitionTypeWithIncompatibleSpecEvolution() {
 
     PartitionSpec newSpec = PartitionSpec.builderFor(table.schema()).identity("category").build();
 
-    TableOperations ops = ((HasTableOperations) table).operations();
+    TableOperations ops = table.operations();
     TableMetadata current = ops.current();
     ops.commit(current, current.updatePartitionSpec(newSpec));
 
@@ -183,6 +183,34 @@ public void testPartitionTypeWithIncompatibleSpecEvolution() {
         .hasMessageStartingWith("Conflicting partition fields");
   }
 
+  @Test
+  public void testPartitionTypeIgnoreInactiveFields() {
+    TestTables.TestTable table =
+        TestTables.create(
+            tableDir, "test", SCHEMA, BY_DATA_CATEGORY_BUCKET_SPEC, V2_FORMAT_VERSION);
+
+    StructType actualType = Partitioning.partitionType(table);
+    assertThat(actualType)
+        .isEqualTo(
+            StructType.of(
+                NestedField.optional(1000, "data", Types.StringType.get()),
+                NestedField.optional(1001, "category_bucket", Types.IntegerType.get())));
+
+    // Create a new spec, and drop the field of the old spec
+    table.updateSpec().removeField("category_bucket").commit();
+    table.updateSchema().deleteColumn("category").commit();
+
+    actualType = Partitioning.partitionType(table);
+    assertThat(actualType)
+        .isEqualTo(StructType.of(NestedField.optional(1000, "data", Types.StringType.get())));
+
+    table.updateSpec().removeField("data").commit();
+    table.updateSchema().deleteColumn("data").commit();
+
+    actualType = Partitioning.partitionType(table);
+    assertThat(actualType).isEqualTo(StructType.of());
+  }
+
   @Test
   public void testGroupingKeyTypeWithSpecEvolutionInV1Tables() {
     TestTables.TestTable table =