Add UpdatePartitionSpec implementation

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

@@ -35,6 +35,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.transforms.Transforms;
 import org.apache.iceberg.transforms.UnknownTransform;
 import org.apache.iceberg.types.Type;
@@ -333,12 +334,12 @@ private void checkAndAddPartitionName(String name) {
       checkAndAddPartitionName(name, null);
     }
 
-    private void checkAndAddPartitionName(String name, Integer identitySourceColumnId) {
+    private void checkAndAddPartitionName(String name, Integer sourceColumnId) {
       Types.NestedField schemaField = schema.findField(name);
-      if (identitySourceColumnId != null) {
+      if (sourceColumnId != null) {
         // for identity transform case we allow  conflicts between partition and schema field name as
         //   long as they are sourced from the same schema field
-        Preconditions.checkArgument(schemaField == null || schemaField.fieldId() == identitySourceColumnId,
+        Preconditions.checkArgument(schemaField == null || schemaField.fieldId() == sourceColumnId,
             "Cannot create identity partition sourced from different field in schema: %s", name);
       } else {
         // for all other transforms we don't allow conflicts between partition name and schema field name
@@ -463,8 +464,8 @@ public Builder truncate(String sourceName, int width) {
     }
 
     public Builder alwaysNull(String sourceName, String targetName) {
-      checkAndAddPartitionName(targetName);
       Types.NestedField sourceColumn = findSourceColumn(sourceName);
+      checkAndAddPartitionName(targetName, sourceColumn.fieldId()); // can duplicate a source column name
       fields.add(new PartitionField(sourceColumn.fieldId(), nextFieldId(), targetName, Transforms.alwaysNull()));
       return this;
     }
@@ -480,9 +481,13 @@ Builder add(int sourceId, String name, String transform) {
 
     Builder add(int sourceId, int fieldId, String name, String transform) {
       Types.NestedField column = schema.findField(sourceId);
-      checkAndAddPartitionName(name, column.fieldId());
       Preconditions.checkNotNull(column, "Cannot find source column: %s", sourceId);
-      fields.add(new PartitionField(sourceId, fieldId, name, Transforms.fromString(column.type(), transform)));
+      return add(sourceId, fieldId, name, Transforms.fromString(column.type(), transform));
+    }
+
+    Builder add(int sourceId, int fieldId, String name, Transform<?, ?> transform) {
+      checkAndAddPartitionName(name, sourceId);
+      fields.add(new PartitionField(sourceId, fieldId, name, transform));
       lastAssignedFieldId.getAndAccumulate(fieldId, Math::max);
       return this;
     }

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

@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg;
+
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.expressions.Term;
+
+/**
+ * API for partition spec evolution.
+ * <p>
+ * When committing, these changes will be applied to the current table metadata. Commit conflicts
+ * will not be resolved and will result in a {@link CommitFailedException}.
+ */
+public interface UpdatePartitionSpec extends PendingUpdate<PartitionSpec> {
+  UpdatePartitionSpec addField(String sourceName);
+
+  UpdatePartitionSpec addField(Term term);
+
+  UpdatePartitionSpec addField(String name, Term term);
+
+  UpdatePartitionSpec removeField(String name);
+
+  UpdatePartitionSpec removeField(Term term);
+
+  UpdatePartitionSpec renameField(String name, String newName);
+}

api/src/main/java/org/apache/iceberg/transforms/PartitionSpecVisitor.java

@@ -113,37 +113,44 @@ static <R> List<R> visit(PartitionSpec spec, PartitionSpecVisitor<R> visitor) {
    * @deprecated this will be removed in 0.11.0; use {@link #visit(PartitionSpec, PartitionSpecVisitor)} instead.
    */
   @Deprecated
-  @SuppressWarnings("checkstyle:CyclomaticComplexity")
   static <R> List<R> visit(Schema schema, PartitionSpec spec, PartitionSpecVisitor<R> visitor) {
     List<R> results = Lists.newArrayListWithExpectedSize(spec.fields().size());
 
     for (PartitionField field : spec.fields()) {
-      String sourceName = schema.findColumnName(field.sourceId());
-      Transform<?, ?> transform = field.transform();
-
-      if (transform instanceof Identity) {
-        results.add(visitor.identity(field.fieldId(), sourceName, field.sourceId()));
-      } else if (transform instanceof Bucket) {
-        int numBuckets = ((Bucket<?>) transform).numBuckets();
-        results.add(visitor.bucket(field.fieldId(), sourceName, field.sourceId(), numBuckets));
-      } else if (transform instanceof Truncate) {
-        int width = ((Truncate<?>) transform).width();
-        results.add(visitor.truncate(field.fieldId(), sourceName, field.sourceId(), width));
-      } else if (transform == Dates.YEAR || transform == Timestamps.YEAR) {
-        results.add(visitor.year(field.fieldId(), sourceName, field.sourceId()));
-      } else if (transform == Dates.MONTH || transform == Timestamps.MONTH) {
-        results.add(visitor.month(field.fieldId(), sourceName, field.sourceId()));
-      } else if (transform == Dates.DAY || transform == Timestamps.DAY) {
-        results.add(visitor.day(field.fieldId(), sourceName, field.sourceId()));
-      } else if (transform == Timestamps.HOUR) {
-        results.add(visitor.hour(field.fieldId(), sourceName, field.sourceId()));
-      } else if (transform instanceof VoidTransform) {
-        results.add(visitor.alwaysNull(field.fieldId(), sourceName, field.sourceId()));
-      } else if (transform instanceof UnknownTransform) {
-        results.add(visitor.unknown(field.fieldId(), sourceName, field.sourceId(), transform.toString()));
-      }
+      results.add(visit(schema, field, visitor));
     }
 
     return results;
   }
+
+  @SuppressWarnings("checkstyle:CyclomaticComplexity")
+  static <R> R visit(Schema schema, PartitionField field, PartitionSpecVisitor<R> visitor) {
+    String sourceName = schema.findColumnName(field.sourceId());
+    Transform<?, ?> transform = field.transform();
+
+    if (transform instanceof Identity) {
+      return visitor.identity(field.fieldId(), sourceName, field.sourceId());
+    } else if (transform instanceof Bucket) {
+      int numBuckets = ((Bucket<?>) transform).numBuckets();
+      return visitor.bucket(field.fieldId(), sourceName, field.sourceId(), numBuckets);
+    } else if (transform instanceof Truncate) {
+      int width = ((Truncate<?>) transform).width();
+      return visitor.truncate(field.fieldId(), sourceName, field.sourceId(), width);
+    } else if (transform == Dates.YEAR || transform == Timestamps.YEAR) {
+      return visitor.year(field.fieldId(), sourceName, field.sourceId());
+    } else if (transform == Dates.MONTH || transform == Timestamps.MONTH) {
+      return visitor.month(field.fieldId(), sourceName, field.sourceId());
+    } else if (transform == Dates.DAY || transform == Timestamps.DAY) {
+      return visitor.day(field.fieldId(), sourceName, field.sourceId());
+    } else if (transform == Timestamps.HOUR) {
+      return visitor.hour(field.fieldId(), sourceName, field.sourceId());
+    } else if (transform instanceof VoidTransform) {
+      return visitor.alwaysNull(field.fieldId(), sourceName, field.sourceId());
+    } else if (transform instanceof UnknownTransform) {
+      return visitor.unknown(field.fieldId(), sourceName, field.sourceId(), transform.toString());
+    }
+
+    throw new UnsupportedOperationException(
+        String.format("Unknown transform class %s", field.transform().getClass().getName()));
+  }
 }