Add UpdatePartitionSpec operation for tables.

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

@@ -138,6 +138,13 @@ default String name() {
    */
   UpdateSchema updateSchema();
 
+  /**
+   * Create a new {@link UpdatePartitionSpec} to alter the partition spec of this table and commit the change.
+   *
+   * @return a new {@link UpdatePartitionSpec}
+   */
+  UpdatePartitionSpec updateSpec();
+
   /**
    * Create a new {@link UpdateProperties} to update table properties and commit the changes.
    *

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

@@ -40,6 +40,13 @@ public interface Transaction {
    */
   UpdateSchema updateSchema();
 
+  /**
+   * Create a new {@link UpdatePartitionSpec} to alter the partition spec of this table.
+   *
+   * @return a new {@link UpdatePartitionSpec}
+   */
+  UpdatePartitionSpec updateSpec();
+
   /**
    * Create a new {@link UpdateProperties} to update table properties.
    *

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

@@ -20,6 +20,7 @@
 package org.apache.iceberg;
 
 import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.expressions.Term;
 
 /**
@@ -29,15 +30,92 @@
  * will not be resolved and will result in a {@link CommitFailedException}.
  */
 public interface UpdatePartitionSpec extends PendingUpdate<PartitionSpec> {
+  /**
+   * Set whether column resolution in the source schema should be case sensitive.
+   *
+   * @param isCaseSensitive whether column resolution should be case sensitive
+   * @return this for method chaining
+   */
+  UpdatePartitionSpec caseSensitive(boolean isCaseSensitive);
+
+  /**
+   * Add a new partition field from a source column.
+   * <p>
+   * The partition field will be created as an identity partition field for the given source column, with the same name
+   * as the source column.
+   * <p>
+   * The source column is located using {@link Schema#findField(String)}.
+   *
+   * @param sourceName source column name in the table schema
+   * @return this for method chaining
+   * @throws IllegalArgumentException If the an identity partition field for the source already exists, or if this
+   *                                  change conflicts with other additions, removals, or renames.
+   */
   UpdatePartitionSpec addField(String sourceName);
 
+  /**
+   * Add a new partition field from an {@link Expressions expression term}.
+   * <p>
+   * The partition field will use the term's transform or the identity transform if the term is a reference.
+   * <p>
+   * The term's reference is used to locate the source column using {@link Schema#findField(String)}.
+   * <p>
+   * The new partition field will be named for the source column and the transform.
+   *
+   * @param term source column name in the table schema
+   * @return this for method chaining
+   * @throws IllegalArgumentException If the a partition field for the transform and source already exists, or if this
+   *                                  change conflicts with other additions, removals, or renames.
+   */
   UpdatePartitionSpec addField(Term term);
 
+  /**
+   * Add a new partition field from an {@link Expressions expression term}, with the given partition field name.
+   * <p>
+   * The partition field will use the term's transform or the identity transform if the term is a reference.
+   * <p>
+   * The term's reference is used to locate the source column using {@link Schema#findField(String)}.
+   *
+   * @param name name for the partition field
+   * @param term expression for the partition transform
+   * @return this for method chaining
+   * @throws IllegalArgumentException If the a partition field for the transform and source already exists, if a
+   *                                  partition field with the given name already exists, or if this change conflicts
+   *                                  with other additions, removals, or renames.
+   */
   UpdatePartitionSpec addField(String name, Term term);
 
+  /**
+   * Remove a partition field by name.
+   *
+   * @param name name of the partition field to remove
+   * @return this for method chaining
+   * @throws IllegalArgumentException If the a partition field with the given name does not exist, or if this change
+   *                                  conflicts with other additions, removals, or renames.
+   */
   UpdatePartitionSpec removeField(String name);
 
+  /**
+   * Remove a partition field by its transform {@link Expressions expression term}.
+   * <p>
+   * The partition field with the same transform and source reference will be removed. If the term is a reference and
+   * does not have a transform, the identity transform is used.
+   *
+   * @param term expression for the partition transform to remove
+   * @return this for method chaining
+   * @throws IllegalArgumentException If the a partition field with the given transform and source does not exist, or
+   *                                  if this change conflicts with other additions, removals, or renames.
+   */
   UpdatePartitionSpec removeField(Term term);
 
+  /**
+   * Rename a field in the partition spec.
+   *
+   * @param name name of the partition field to rename
+   * @param newName replacement name for the partition field
+   * @return this for method chaining
+   * @throws IllegalArgumentException If name doesn't identify a column in the schema or if this change conflicts with
+   *                                  other additions, removals, or renames.
+   */
   UpdatePartitionSpec renameField(String name, String newName);
 }

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

@@ -107,6 +107,11 @@ public UpdateSchema updateSchema() {
     throw new UnsupportedOperationException("Cannot update the schema of a metadata table");
   }
 
+  @Override
+  public UpdatePartitionSpec updateSpec() {
+    throw new UnsupportedOperationException("Cannot update the partition spec of a metadata table");
+  }
+
   @Override
   public UpdateProperties updateProperties() {
     throw new UnsupportedOperationException("Cannot update the properties of a metadata table");

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

@@ -119,6 +119,10 @@ public UpdateSchema updateSchema() {
     return new SchemaUpdate(ops);
   }
 
+  public UpdatePartitionSpec updateSpec() {
+    return new BaseUpdatePartitionSpec(ops);
+  }
+
   @Override
   public UpdateProperties updateProperties() {
     return new PropertiesUpdate(ops);

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

@@ -102,6 +102,14 @@ public UpdateSchema updateSchema() {
     return schemaChange;
   }
 
+  @Override
+  public UpdatePartitionSpec updateSpec() {
+    checkLastOperationCommitted("UpdateSpec");
+    UpdatePartitionSpec partitionSpecChange = new BaseUpdatePartitionSpec(transactionOps);
+    updates.add(partitionSpecChange);
+    return partitionSpecChange;
+  }
+
   @Override
   public UpdateProperties updateProperties() {
     checkLastOperationCommitted("UpdateProperties");
@@ -567,6 +575,11 @@ public UpdateSchema updateSchema() {
       return BaseTransaction.this.updateSchema();
     }
 
+    @Override
+    public UpdatePartitionSpec updateSpec() {
+      return BaseTransaction.this.updateSpec();
+    }
+
     @Override
     public UpdateProperties updateProperties() {
       return BaseTransaction.this.updateProperties();

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

@@ -43,7 +43,6 @@
 
 class BaseUpdatePartitionSpec implements UpdatePartitionSpec {
   private final TableOperations ops;
-  private final boolean caseSensitive;
   private final TableMetadata base;
   private final int formatVersion;
   private final PartitionSpec spec;
@@ -58,11 +57,12 @@ class BaseUpdatePartitionSpec implements UpdatePartitionSpec {
   private final Set<Object> deletes = Sets.newHashSet();
   private final Map<String, String> renames = Maps.newHashMap();
 
+  private boolean caseSensitive;
   private int lastAssignedPartitionId;
 
-  BaseUpdatePartitionSpec(TableOperations ops, boolean caseSensitive) {
+  BaseUpdatePartitionSpec(TableOperations ops) {
     this.ops = ops;
-    this.caseSensitive = caseSensitive;
+    this.caseSensitive = true;
     this.base = ops.current();
     this.formatVersion = base.formatVersion();
     this.spec = base.spec();
@@ -108,6 +108,12 @@ private int assignFieldId() {
     return lastAssignedPartitionId;
   }
 
+  @Override
+  public UpdatePartitionSpec caseSensitive(boolean isCaseSensitive) {
+    this.caseSensitive = isCaseSensitive;
+    return this;
+  }
+
   @Override
   public BaseUpdatePartitionSpec addField(String sourceName) {
     return addField(Expressions.ref(sourceName));

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

@@ -42,6 +42,11 @@ public UpdateSchema updateSchema() {
     return wrapped.updateSchema();
   }
 
+  @Override
+  public UpdatePartitionSpec updateSpec() {
+    return wrapped.updateSpec();
+  }
+
   @Override
   public UpdateProperties updateProperties() {
     return wrapped.updateProperties();

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

@@ -475,8 +475,10 @@ public TableMetadata updatePartitionSpec(PartitionSpec newPartitionSpec) {
       }
     }
 
-    Preconditions.checkArgument(defaultSpecId != newDefaultSpecId,
-        "Cannot set default partition spec to the current default");
+    if (defaultSpecId == newDefaultSpecId) {
+      // the new spec is already current and no change is needed
+      return this;
+    }
 
     ImmutableList.Builder<PartitionSpec> builder = ImmutableList.<PartitionSpec>builder()
         .addAll(specs);

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

@@ -501,7 +501,7 @@ public void testNewTableMetadataReassignmentAllIds() throws Exception {
 
     PartitionSpec spec = PartitionSpec.builderFor(schema).withSpecId(5)
         .add(3, 1005, "x_partition", "bucket[4]")
-        .add(5, 1005, "z_partition", "bucket[8]")
+        .add(5, 1003, "z_partition", "bucket[8]")
         .build();
     String location = "file://tmp/db/table";
     TableMetadata metadata = TableMetadata.newTableMetadata(schema, spec, location, ImmutableMap.of());