API, Core: Add Schema#withUpdatedDoc and View#updateColumnDoc APIs

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

@@ -401,6 +401,37 @@ public String idToAlias(Integer fieldId) {
     return null;
   }
 
+  /**
+   * Returns a schema with an updated field doc for the given field.
+   *
+   * @param name field name
+   * @param doc column doc
+   * @return a Schema with the updated field doc
+   * @throws IllegalArgumentException if no field with the given name is found
+   */
+  public Schema withColumnDoc(String name, String doc) {
+    NestedField fieldToUpdate = findField(name);
+    Preconditions.checkArgument(fieldToUpdate != null, "Field %s does not exist", name);
+    NestedField updatedField =
+        NestedField.of(
+            fieldToUpdate.fieldId(),
+            fieldToUpdate.isOptional(),
+            fieldToUpdate.name(),
+            fieldToUpdate.type(),
+            doc);
+
+    List<NestedField> newFields = Lists.newArrayList();
+    newFields.add(updatedField);
+
+    for (NestedField field : columns()) {
+      if (field.fieldId() != updatedField.fieldId()) {
+        newFields.add(field);
+      }
+    }
+
+    return new Schema(newFields, getAliases(), identifierFieldIds());
+  }
+
   /**
    * Returns an accessor for retrieving the data from {@link StructLike}.
    *

api/src/main/java/org/apache/iceberg/view/View.java

@@ -132,4 +132,15 @@ default SQLViewRepresentation sqlFor(String dialect) {
     throw new UnsupportedOperationException(
         "Resolving a sql with a given dialect is not supported");
   }
+
+  /**
+   * Updates the column documentation for the field with the given name
+   *
+   * @param name field name
+   * @param doc doc to update
+   * @return ReplaceViewVersion to replace the current view's version
+   */
+  default ReplaceViewVersion updateColumnDoc(String name, String doc) {
+    throw new UnsupportedOperationException("Updating column documentation is not supported");
+  }
 }

core/src/main/java/org/apache/iceberg/view/BaseView.java

@@ -128,4 +128,23 @@ public SQLViewRepresentation sqlFor(String dialect) {
 
     return closest;
   }
+
+  @Override
+  @SuppressWarnings("checkstyle:HiddenField")
+  public ReplaceViewVersion updateColumnDoc(String name, String doc) {
+    ReplaceViewVersion replace =
+        replaceVersion()
+            .withSchema(schema().withColumnDoc(name, doc))
+            .withDefaultCatalog(currentVersion().defaultCatalog())
+            .withDefaultNamespace(currentVersion().defaultNamespace());
+
+    for (ViewRepresentation representation : currentVersion().representations()) {
+      if (representation instanceof SQLViewRepresentation) {
+        SQLViewRepresentation sqlViewRepresentation = (SQLViewRepresentation) representation;
+        replace.withQuery(sqlViewRepresentation.dialect(), sqlViewRepresentation.sql());
+      }
+    }
+
+    return replace;
+  }
 }

core/src/test/java/org/apache/iceberg/view/ViewCatalogTests.java

@@ -1739,4 +1739,49 @@ public void testSqlForInvalidArguments() {
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessage("Invalid dialect: (empty string)");
   }
+
+  @Test
+  public void testReplaceViewWithUpdatedColumnDoc() {
+    TableIdentifier identifier = TableIdentifier.of("ns", "view");
+
+    if (requiresNamespaceCreate()) {
+      catalog().createNamespace(identifier.namespace());
+    }
+
+    View view =
+        catalog()
+            .buildView(identifier)
+            .withSchema(SCHEMA)
+            .withDefaultNamespace(identifier.namespace())
+            .withDefaultCatalog(catalog().name())
+            .withQuery("spark", "select * from ns.tbl")
+            .create();
+
+    view.updateColumnDoc("data", "some docs for data").commit();
+
+    assertThat(view.schemas().size()).isEqualTo(2);
+    assertThat(view.schema().findField("data").doc()).isEqualTo("some docs for data");
+  }
+
+  @Test
+  public void testReplaceViewColumnDocNonExistingFieldFails() {
+    TableIdentifier identifier = TableIdentifier.of("ns", "view");
+
+    if (requiresNamespaceCreate()) {
+      catalog().createNamespace(identifier.namespace());
+    }
+
+    View view =
+        catalog()
+            .buildView(identifier)
+            .withSchema(SCHEMA)
+            .withDefaultNamespace(identifier.namespace())
+            .withDefaultCatalog(catalog().name())
+            .withQuery("spark", "select * from ns.tbl")
+            .create();
+
+    assertThatThrownBy(() -> view.updateColumnDoc("non_existing", "non_existing").commit())
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Field non_existing does not exist");
+  }
 }