Nessie: Support Namespace properties (apache#4610)

nessie/src/main/java/org/apache/iceberg/nessie/NessieCatalog.java

@@ -223,16 +223,16 @@ public boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyExcept
 
   @Override
   public boolean setProperties(Namespace namespace, Map<String, String> properties) {
-    throw new UnsupportedOperationException(
-        "Cannot set properties for namespace '" + namespace +
-            "': setProperties is not supported by the NessieCatalog");
+    client.setProperties(namespace, properties);
+    // always successful, otherwise an exception is thrown
+    return true;
   }
 
   @Override
   public boolean removeProperties(Namespace namespace, Set<String> properties) {
-    throw new UnsupportedOperationException(
-        "Cannot remove properties for namespace '" + namespace +
-            "': removeProperties is not supported by the NessieCatalog");
+    client.removeProperties(namespace, properties);
+    // always successful, otherwise an exception is thrown
+    return true;
   }
 
   @Override

nessie/src/main/java/org/apache/iceberg/nessie/NessieIcebergClient.java

@@ -22,6 +22,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -34,7 +35,6 @@
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.relocated.com.google.common.base.Suppliers;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.util.Tasks;
 import org.projectnessie.client.NessieConfigConstants;
 import org.projectnessie.client.api.CommitMultipleOperationsBuilder;
@@ -170,6 +170,7 @@ public void createNamespace(Namespace namespace, Map<String, String> metadata) {
       getApi().createNamespace()
           .reference(getRef().getReference())
           .namespace(org.projectnessie.model.Namespace.of(namespace.levels()))
+          .properties(metadata)
           .create();
       refresh();
     } catch (NessieNamespaceAlreadyExistsException e) {
@@ -217,17 +218,53 @@ public boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyExcept
 
   public Map<String, String> loadNamespaceMetadata(Namespace namespace) throws NoSuchNamespaceException {
     try {
-      getApi().getNamespace()
+      return getApi().getNamespace()
           .reference(getRef().getReference())
           .namespace(org.projectnessie.model.Namespace.of(namespace.levels()))
-          .get();
+          .get()
+          .getProperties();
     } catch (NessieNamespaceNotFoundException e) {
       throw new NoSuchNamespaceException(e, "Namespace does not exist: %s", namespace);
     } catch (NessieReferenceNotFoundException e) {
       throw new RuntimeException(String.format("Cannot load Namespace '%s': " +
           "ref '%s' is no longer valid.", namespace, getRef().getName()), e);
     }
-    return ImmutableMap.of();
+  }
+
+  public void setProperties(Namespace namespace, Map<String, String> properties) {
+    try {
+      getApi()
+          .updateProperties()
+          .reference(getRef().getReference())
+          .namespace(org.projectnessie.model.Namespace.of(namespace.levels()))
+          .updateProperties(properties)
+          .update();
+      refresh();
+    } catch (NessieNamespaceNotFoundException e) {
+      throw new NoSuchNamespaceException(e, "Namespace does not exist: %s", namespace);
+    } catch (NessieNotFoundException e) {
+      throw new RuntimeException(
+          String.format("Cannot update properties on Namespace '%s': ref '%s' is no longer valid.",
+              namespace, getRef().getName()), e);
+    }
+  }
+
+  public void removeProperties(Namespace namespace, Set<String> properties) {
+    try {
+      getApi()
+          .updateProperties()
+          .reference(getRef().getReference())
+          .namespace(org.projectnessie.model.Namespace.of(namespace.levels()))
+          .removeProperties(properties)
+          .update();
+      refresh();
+    } catch (NessieNamespaceNotFoundException e) {
+      throw new NoSuchNamespaceException(e, "Namespace does not exist: %s", namespace);
+    } catch (NessieNotFoundException e) {
+      throw new RuntimeException(
+          String.format("Cannot remove properties from Namespace '%s': ref '%s' is no longer valid.",
+              namespace, getRef().getName()), e);
+    }
   }
 
   public void renameTable(TableIdentifier from, TableIdentifier to) {

nessie/src/test/java/org/apache/iceberg/nessie/TestNamespace.java

@@ -19,13 +19,17 @@
 
 package org.apache.iceberg.nessie;
 
-import java.util.Collections;
+import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Types;
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -104,16 +108,36 @@ public void testCreatingAndDroppingNamespaceWithContent() throws NessieNotFoundE
 
   @Test
   public void testSettingProperties() {
-    Assertions.assertThatThrownBy(() -> catalog.setProperties(Namespace.of("test"), Collections.emptyMap()))
-        .isInstanceOf(UnsupportedOperationException.class)
-        .hasMessage("Cannot set properties for namespace 'test': setProperties is not supported by the NessieCatalog");
+    Map<String, String> properties = ImmutableMap.of("prop", "val");
+    Namespace namespace = Namespace.of("withProperties");
+    catalog.createNamespace(namespace, properties);
+    Assertions.assertThat(catalog.namespaceExists(namespace)).isTrue();
+    Assertions.assertThat(catalog.loadNamespaceMetadata(namespace)).isEqualTo(properties);
+
+    ImmutableMap<String, String> updatedProperties = ImmutableMap.of("prop2", "val2", "prop", "new_val");
+    catalog.setProperties(namespace, updatedProperties);
+    Assertions.assertThat(catalog.loadNamespaceMetadata(namespace)).isEqualTo(updatedProperties);
+
+    Assertions.assertThatThrownBy(() -> catalog.setProperties(Namespace.of("unknown"), updatedProperties))
+        .isInstanceOf(NoSuchNamespaceException.class)
+        .hasMessage("Namespace does not exist: unknown");
   }
 
   @Test
   public void testRemovingProperties() {
-    Assertions.assertThatThrownBy(() -> catalog.removeProperties(Namespace.of("test"), Collections.emptySet()))
-        .isInstanceOf(UnsupportedOperationException.class)
-        .hasMessage(
-            "Cannot remove properties for namespace 'test': removeProperties is not supported by the NessieCatalog");
+    Map<String, String> properties = ImmutableMap.of("prop2", "val2", "prop", "val");
+    Namespace namespace = Namespace.of("withPropertyDeletes");
+    catalog.createNamespace(namespace, properties);
+    Assertions.assertThat(catalog.namespaceExists(namespace)).isTrue();
+    Assertions.assertThat(catalog.loadNamespaceMetadata(namespace)).isEqualTo(properties);
+
+    Set<String> toRemove = Sets.newHashSet(Arrays.asList("prop1", "prop2", "prop3"));
+    catalog.removeProperties(namespace, toRemove);
+    Assertions.assertThat(catalog.loadNamespaceMetadata(namespace))
+        .isEqualTo(ImmutableMap.of("prop", "val"));
+
+    Assertions.assertThatThrownBy(() -> catalog.removeProperties(Namespace.of("unknown"), toRemove))
+        .isInstanceOf(NoSuchNamespaceException.class)
+        .hasMessage("Namespace does not exist: unknown");
   }
 }

nessie/src/test/java/org/apache/iceberg/nessie/TestNessieCatalog.java

@@ -136,7 +136,7 @@ protected NessieCatalog catalog() {
 
   @Override
   protected boolean supportsNamespaceProperties() {
-    return false;
+    return true;
   }
 
   @Override