Core: Fix drop partition field and schema field error

.palantir/revapi.yml

@@ -1145,6 +1145,10 @@ acceptedBreaks:
       new: "method org.apache.iceberg.BaseMetastoreOperations.CommitStatus org.apache.iceberg.BaseMetastoreTableOperations::checkCommitStatus(java.lang.String,\
         \ org.apache.iceberg.TableMetadata)"
       justification: "Removing deprecated code"
+    - code: "java.method.numberOfParametersChanged"
+      old: "method void org.apache.iceberg.UnboundPartitionSpec::<init>(int, java.util.List<org.apache.iceberg.UnboundPartitionSpec.UnboundPartitionField>)"
+      new: "method void org.apache.iceberg.UnboundPartitionSpec::<init>(int, int, java.util.List<org.apache.iceberg.UnboundPartitionSpec.UnboundPartitionField>)"
+      justification: "Add schema id in UnboundPartitionSpec"
   apache-iceberg-0.14.0:
     org.apache.iceberg:iceberg-api:
     - code: "java.class.defaultSerializationChanged"
@@ -1184,6 +1188,10 @@ acceptedBreaks:
     - code: "java.method.removed"
       old: "method org.apache.iceberg.RowDelta org.apache.iceberg.RowDelta::validateNoConflictingAppends(org.apache.iceberg.expressions.Expression)"
       justification: "Deprecations for 1.0 release"
+    - code: "java.method.numberOfParametersChanged"
+      old: "method void org.apache.iceberg.UnboundPartitionSpec::<init>(int, java.util.List<org.apache.iceberg.UnboundPartitionSpec.UnboundPartitionField>)"
+      new: "method void org.apache.iceberg.UnboundPartitionSpec::<init>(int, int, java.util.List<org.apache.iceberg.UnboundPartitionSpec.UnboundPartitionField>)"
+      justification: "Add schema id in UnboundPartitionSpec"
   release-base-0.13.0:
     org.apache.iceberg:iceberg-api:
     - code: "java.class.defaultSerializationChanged"

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

@@ -101,7 +101,8 @@ int lastAssignedFieldId() {
   }
 
   public UnboundPartitionSpec toUnbound() {
-    UnboundPartitionSpec.Builder builder = UnboundPartitionSpec.builder().withSpecId(specId);
+    UnboundPartitionSpec.Builder builder =
+        UnboundPartitionSpec.builder().withSpecId(specId).withSchemaId(schema.schemaId());
 
     for (PartitionField field : fields) {
       builder.addField(
@@ -630,21 +631,19 @@ static void checkCompatibility(PartitionSpec spec, Schema schema) {
       // 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
-      // We don't care about the source type since a VoidTransform is always compatible and skip the
-      // checks
-      if (!transform.equals(Transforms.alwaysNull())) {
-        ValidationException.check(
-            sourceType != null, "Cannot find source column for partition field: %s", field);
-        ValidationException.check(
-            sourceType.isPrimitiveType(),
-            "Cannot partition by non-primitive source field: %s",
-            sourceType);
-        ValidationException.check(
-            transform.canTransform(sourceType),
-            "Invalid source type %s for transform: %s",
-            sourceType,
-            transform);
-      }
+      // We don't care about the source type, but we should check partition field could
+      // be found in schema
+      ValidationException.check(
+          sourceType != null, "Cannot find source column for partition field: %s", field);
+      ValidationException.check(
+          sourceType.isPrimitiveType(),
+          "Cannot partition by non-primitive source field: %s",
+          sourceType);
+      ValidationException.check(
+          transform.canTransform(sourceType),
+          "Invalid source type %s for transform: %s",
+          sourceType,
+          transform);
     }
   }
 

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

@@ -27,17 +27,23 @@
 public class UnboundPartitionSpec {
 
   private final int specId;
+  private final int schmeId;
   private final List<UnboundPartitionField> fields;
 
-  public UnboundPartitionSpec(int specId, List<UnboundPartitionField> fields) {
+  public UnboundPartitionSpec(int specId, int schmeId, List<UnboundPartitionField> fields) {
     this.specId = specId;
     this.fields = fields;
+    this.schmeId = schmeId;
   }
 
   public int specId() {
     return specId;
   }
 
+  public int schemaId() {
+    return schmeId;
+  }
+
   public List<UnboundPartitionField> fields() {
     return fields;
   }
@@ -78,6 +84,7 @@ static Builder builder() {
   static class Builder {
     private final List<UnboundPartitionField> fields;
     private int specId = 0;
+    private int schemaId = -1;
 
     private Builder() {
       this.fields = Lists.newArrayList();
@@ -88,6 +95,11 @@ Builder withSpecId(int newSpecId) {
       return this;
     }
 
+    Builder withSchemaId(int newSchemaId) {
+      this.schemaId = newSchemaId;
+      return this;
+    }
+
     Builder addField(String transformAsString, int sourceId, int partitionId, String name) {
       fields.add(new UnboundPartitionField(transformAsString, sourceId, partitionId, name));
       return this;
@@ -99,7 +111,7 @@ Builder addField(String transformAsString, int sourceId, String name) {
     }
 
     UnboundPartitionSpec build() {
-      return new UnboundPartitionSpec(specId, fields);
+      return new UnboundPartitionSpec(specId, schemaId, fields);
     }
   }
 

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

@@ -19,6 +19,7 @@
 package org.apache.iceberg;
 
 import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
@@ -33,6 +34,7 @@ public class PartitionSpecParser {
   private PartitionSpecParser() {}
 
   private static final String SPEC_ID = "spec-id";
+  private static final String SCHEMA_ID = "schema-id";
   private static final String FIELDS = "fields";
   private static final String SOURCE_ID = "source-id";
   private static final String FIELD_ID = "field-id";
@@ -54,6 +56,7 @@ public static String toJson(PartitionSpec spec, boolean pretty) {
   public static void toJson(UnboundPartitionSpec spec, JsonGenerator generator) throws IOException {
     generator.writeStartObject();
     generator.writeNumberField(SPEC_ID, spec.specId());
+    generator.writeNumberField(SCHEMA_ID, spec.schemaId());
     generator.writeFieldName(FIELDS);
     toJsonFields(spec, generator);
     generator.writeEndObject();
@@ -75,10 +78,22 @@ public static UnboundPartitionSpec fromJson(JsonNode json) {
     Preconditions.checkArgument(json.isObject(), "Cannot parse spec from non-object: %s", json);
     int specId = JsonUtil.getInt(SPEC_ID, json);
     UnboundPartitionSpec.Builder builder = UnboundPartitionSpec.builder().withSpecId(specId);
+    Integer schemaId = JsonUtil.getIntOrNull(SCHEMA_ID, json);
+    if (schemaId != null) {
+      builder.withSchemaId(schemaId);
+    }
     buildFromJsonFields(builder, JsonUtil.get(FIELDS, json));
     return builder.build();
   }
 
+  public static UnboundPartitionSpec fromJson(String json) {
+    try {
+      return fromJson(JsonUtil.mapper().readValue(json, JsonNode.class));
+    } catch (JsonProcessingException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
   private static final Cache<Pair<Types.StructType, String>, PartitionSpec> SPEC_CACHE =
       Caffeine.newBuilder().weakValues().build();
 

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

@@ -193,7 +193,15 @@ public Map<Integer, PartitionSpec> specs() {
           Map<Integer, PartitionSpec> specs = Maps.newHashMapWithExpectedSize(specAsJsonMap.size());
           specAsJsonMap.forEach(
               (specId, specAsJson) -> {
-                specs.put(specId, PartitionSpecParser.fromJson(schema(), specAsJson));
+                int schemaId = PartitionSpecParser.fromJson(specAsJson).schemaId();
+                // In some unit tests, there may be situations that
+                // metadata location is not passed in.
+                // In this case, schemas cannot be read and latest schema is used by default.
+                Schema schema =
+                    schemaId != -1 && metadataFileLocation != null
+                        ? schemas().get(schemaId)
+                        : schema();
+                specs.put(specId, PartitionSpecParser.fromJson(schema, specAsJson));
               });
           this.lazySpecs = specs;
         } else if (lazySpecs == null) {

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

@@ -1052,12 +1052,6 @@ public Builder setCurrentSchema(int schemaId) {
       Preconditions.checkArgument(
           schema != null, "Cannot set current schema to unknown schema: %s", schemaId);
 
-      // rebuild all the partition specs and sort orders for the new current schema
-      this.specs =
-          Lists.newArrayList(Iterables.transform(specs, spec -> updateSpecSchema(schema, spec)));
-      specsById.clear();
-      specsById.putAll(PartitionUtil.indexSpecs(specs));
-
       this.sortOrders =
           Lists.newArrayList(
               Iterables.transform(sortOrders, order -> updateSortOrderSchema(schema, order)));
@@ -1537,6 +1531,17 @@ && changes(MetadataUpdate.AddSchema.class)
 
       this.lastAddedSchemaId = newSchemaId;
 
+      PartitionSpec currentSpec = specsById.get(defaultSpecId);
+      if (currentSpec != null) {
+        PartitionSpec newCurrentSpec = updateSpecSchema(newSchema, currentSpec);
+        specsById.put(defaultSpecId, newCurrentSpec);
+        for (int i = 0; i < specs.size(); i++) {
+          if (specs.get(i).specId() == defaultSpecId) {
+            specs.set(i, newCurrentSpec);
+          }
+        }
+      }
+
       return newSchemaId;
     }
 

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

@@ -387,8 +387,9 @@ public static TableMetadata fromJson(String metadataLocation, JsonNode node) {
       ImmutableList.Builder<PartitionSpec> builder = ImmutableList.builder();
       for (JsonNode spec : specArray) {
         UnboundPartitionSpec unboundSpec = PartitionSpecParser.fromJson(spec);
-        if (unboundSpec.specId() == defaultSpecId) {
-          builder.add(unboundSpec.bind(schema));
+        int schemaId = unboundSpec.schemaId();
+        if (schemaId != -1) {
+          builder.add(unboundSpec.bind(schemas.get(schemaId)));
         } else {
           builder.add(unboundSpec.bindUnchecked(schema));
         }

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

@@ -122,7 +122,7 @@ private String taskJson() {
         + "{\"id\":12,\"name\":\"lower_bound\",\"required\":false,\"type\":\"string\"},"
         + "{\"id\":13,\"name\":\"upper_bound\",\"required\":false,\"type\":\"string\"}]},\"element-required\":true}},"
         + "{\"id\":18,\"name\":\"reference_snapshot_id\",\"required\":true,\"type\":\"long\"}]},"
-        + "\"partition-specs\":[{\"spec-id\":0,\"fields\":[{\"name\":\"data_bucket\","
+        + "\"partition-specs\":[{\"spec-id\":0,\"schema-id\":0,\"fields\":[{\"name\":\"data_bucket\","
         + "\"transform\":\"bucket[16]\",\"source-id\":4,\"field-id\":1000}]}],"
         + "\"manifest-list-Location\":\"/path/manifest-list-file.avro\","
         + "\"residual-filter\":{\"type\":\"eq\",\"term\":\"id\",\"value\":1},"

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

@@ -102,7 +102,7 @@ private String fileScanTaskJson() {
         + "\"schema\":{\"type\":\"struct\",\"schema-id\":0,\"fields\":["
         + "{\"id\":3,\"name\":\"id\",\"required\":true,\"type\":\"int\"},"
         + "{\"id\":4,\"name\":\"data\",\"required\":true,\"type\":\"string\"}]},"
-        + "\"spec\":{\"spec-id\":0,\"fields\":[{\"name\":\"data_bucket\","
+        + "\"spec\":{\"spec-id\":0,\"schema-id\":0,\"fields\":[{\"name\":\"data_bucket\","
         + "\"transform\":\"bucket[16]\",\"source-id\":4,\"field-id\":1000}]},"
         + "\"data-file\":{\"spec-id\":0,\"content\":\"DATA\",\"file-path\":\"/path/to/data-a.parquet\","
         + "\"file-format\":\"PARQUET\",\"partition\":{\"1000\":0},"

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

@@ -100,7 +100,7 @@ private String taskJson() {
         + "{\"id\":4,\"name\":\"data\",\"required\":true,\"type\":\"string\"}]},"
         + "\"file-io\":{\"io-impl\":\"org.apache.iceberg.hadoop.HadoopFileIO\","
         + "\"properties\":{\"k1\":\"v1\",\"k2\":\"v2\"}},"
-        + "\"partition-specs\":[{\"spec-id\":0,\"fields\":[{"
+        + "\"partition-specs\":[{\"spec-id\":0,\"schema-id\":0,\"fields\":[{"
         + "\"name\":\"data_bucket\",\"transform\":\"bucket[16]\",\"source-id\":4,\"field-id\":1000}]}],"
         + "\"residual-filter\":{\"type\":\"eq\",\"term\":\"id\",\"value\":1},"
         + "\"manifest-file\":{\"path\":\"/path/input.m0.avro\","

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

@@ -22,6 +22,7 @@
 
 import java.util.Arrays;
 import java.util.List;
+import org.apache.iceberg.types.Types;
 import org.junit.jupiter.api.TestTemplate;
 import org.junit.jupiter.api.extension.ExtendWith;
 
@@ -37,6 +38,7 @@ public void testToJsonForV1Table() {
     String expected =
         "{\n"
             + "  \"spec-id\" : 0,\n"
+            + "  \"schema-id\" : 0,\n"
             + "  \"fields\" : [ {\n"
             + "    \"name\" : \"data_bucket\",\n"
             + "    \"transform\" : \"bucket[16]\",\n"
@@ -54,6 +56,7 @@ public void testToJsonForV1Table() {
     expected =
         "{\n"
             + "  \"spec-id\" : 1,\n"
+            + "  \"schema-id\" : 0,\n"
             + "  \"fields\" : [ {\n"
             + "    \"name\" : \"id_bucket\",\n"
             + "    \"transform\" : \"bucket[8]\",\n"
@@ -74,6 +77,7 @@ public void testFromJsonWithFieldId() {
     String specString =
         "{\n"
             + "  \"spec-id\" : 1,\n"
+            + "  \"schema-id\" : 0,\n"
             + "  \"fields\" : [ {\n"
             + "    \"name\" : \"id_bucket\",\n"
             + "    \"transform\" : \"bucket[8]\",\n"
@@ -100,6 +104,7 @@ public void testFromJsonWithoutFieldId() {
     String specString =
         "{\n"
             + "  \"spec-id\" : 1,\n"
+            + "  \"schema-id\" : 0,\n"
             + "  \"fields\" : [ {\n"
             + "    \"name\" : \"id_bucket\",\n"
             + "    \"transform\" : \"bucket[8]\",\n"
@@ -120,10 +125,78 @@ public void testFromJsonWithoutFieldId() {
   }
 
   @TestTemplate
-  public void testTransforms() {
-    for (PartitionSpec spec : PartitionSpecTestBase.SPECS) {
-      assertThat(roundTripJSON(spec)).isEqualTo(spec);
-    }
+  public void testFromJsonWithoutSchemaId() {
+    String specString =
+        "{\n"
+            + "  \"spec-id\" : 1,\n"
+            + "  \"fields\" : [ {\n"
+            + "    \"name\" : \"id_bucket\",\n"
+            + "    \"transform\" : \"bucket[8]\",\n"
+            + "    \"source-id\" : 1,\n"
+            + "    \"field-id\" : 1001\n"
+            + "  }, {\n"
+            + "    \"name\" : \"data_bucket\",\n"
+            + "    \"transform\" : \"bucket[16]\",\n"
+            + "    \"source-id\" : 2,\n"
+            + "    \"field-id\" : 1000\n"
+            + "  } ]\n"
+            + "}";
+
+    PartitionSpec spec = PartitionSpecParser.fromJson(table.schema(), specString);
+
+    assertThat(spec.fields().size()).isEqualTo(2);
+    // should be the field ids in the JSON
+    assertThat(spec.fields().get(0).fieldId()).isEqualTo(1001);
+    assertThat(spec.fields().get(1).fieldId()).isEqualTo(1000);
+  }
+
+  @TestTemplate
+  public void testForUpdateSchema() {
+    String expected =
+        "{\n"
+            + "  \"spec-id\" : 0,\n"
+            + "  \"schema-id\" : 0,\n"
+            + "  \"fields\" : [ {\n"
+            + "    \"name\" : \"data_bucket\",\n"
+            + "    \"transform\" : \"bucket[16]\",\n"
+            + "    \"source-id\" : 2,\n"
+            + "    \"field-id\" : 1000\n"
+            + "  } ]\n"
+            + "}";
+    assertThat(PartitionSpecParser.toJson(table.spec(), true)).isEqualTo(expected);
+
+    PartitionSpec spec =
+        PartitionSpec.builderFor(table.schema()).bucket("id", 8).bucket("data", 16).build();
+
+    table.ops().commit(table.ops().current(), table.ops().current().updatePartitionSpec(spec));
+    Schema newSchema =
+        new Schema(
+            Types.NestedField.required(1, "id", Types.IntegerType.get()),
+            Types.NestedField.required(2, "data_renamed", Types.StringType.get()));
+
+    table
+        .ops()
+        .commit(
+            table.ops().current(),
+            table.ops().current().updateSchema(newSchema, table.schema().highestFieldId()));
+
+    expected =
+        "{\n"
+            + "  \"spec-id\" : 1,\n"
+            + "  \"schema-id\" : 1,\n"
+            + "  \"fields\" : [ {\n"
+            + "    \"name\" : \"id_bucket\",\n"
+            + "    \"transform\" : \"bucket[8]\",\n"
+            + "    \"source-id\" : 1,\n"
+            + "    \"field-id\" : 1000\n"
+            + "  }, {\n"
+            + "    \"name\" : \"data_bucket\",\n"
+            + "    \"transform\" : \"bucket[16]\",\n"
+            + "    \"source-id\" : 2,\n"
+            + "    \"field-id\" : 1001\n"
+            + "  } ]\n"
+            + "}";
+    assertThat(PartitionSpecParser.toJson(table.spec(), true)).isEqualTo(expected);
   }
 
   private static PartitionSpec roundTripJSON(PartitionSpec spec) {