IcebergCatalog- updateNamespaceProperties on single commit

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/CatalogHandlerUtils.java

@@ -239,13 +239,17 @@ public UpdateNamespacePropertiesResponse updateNamespaceProperties(
     Map<String, String> startProperties = catalog.loadNamespaceMetadata(namespace);
     Set<String> missing = Sets.difference(removals, startProperties.keySet());
 
-    if (!updates.isEmpty()) {
-      catalog.setProperties(namespace, updates);
-    }
+    if (catalog instanceof IcebergCatalog polarisCatalog) {
+      polarisCatalog.updateProperties(namespace, removals, updates);
+    } else {
+      if (!updates.isEmpty()) {
+        catalog.setProperties(namespace, updates);
+      }
 
-    if (!removals.isEmpty()) {
-      // remove the original set just in case there was an update just after loading properties
-      catalog.removeProperties(namespace, removals);
+      if (!removals.isEmpty()) {
+        // remove the original set just in case there was an update just after loading properties
+        catalog.removeProperties(namespace, removals);
+      }
     }
 
     return UpdateNamespacePropertiesResponse.builder()

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalog.java

@@ -735,6 +735,63 @@ public boolean setProperties(Namespace namespace, Map<String, String> properties
     return true;
   }
 
+  /**
+   * Update properties for a namespace.
+   *
+   * @param namespace the namespace to update properties for
+   * @param removals the set of properties to remove
+   * @param updates the map of properties to update
+   * @return true if the properties were updated, false otherwise
+   */
+  public boolean updateProperties(
+      Namespace namespace, Set<String> removals, Map<String, String> updates)
+      throws NoSuchNamespaceException {
+    PolarisResolvedPathWrapper resolvedEntities = resolvedEntityView.getResolvedPath(namespace);
+    if (resolvedEntities == null) {
+      throw noSuchNamespaceException(namespace);
+    }
+    PolarisEntity entity = resolvedEntities.getRawLeafEntity();
+    Map<String, String> newProperties = new HashMap<>(entity.getPropertiesAsMap());
+
+    // Merge new properties into existing map.
+    newProperties.putAll(updates);
+    // Remove properties.
+    removals.forEach(newProperties::remove);
+
+    PolarisEntity updatedEntity =
+        new PolarisEntity.Builder(entity).setProperties(newProperties).build();
+
+    if (!realmConfig.getConfig(FeatureConfiguration.ALLOW_NAMESPACE_LOCATION_OVERLAP)) {
+      LOGGER.debug("Validating no overlap with sibling tables or namespaces");
+      validateNoLocationOverlap(
+          NamespaceEntity.of(updatedEntity), resolvedEntities.getRawParentPath());
+    } else {
+      LOGGER.debug("Skipping location overlap validation for namespace '{}'", namespace);
+    }
+    if (!realmConfig.getConfig(
+        BehaviorChangeConfiguration.ALLOW_NAMESPACE_CUSTOM_LOCATION, catalogEntity)) {
+      if (updates.containsKey(PolarisEntityConstants.ENTITY_BASE_LOCATION)) {
+        validateNamespaceLocation(NamespaceEntity.of(entity), resolvedEntities);
+      }
+    }
+
+    List<PolarisEntity> parentPath = resolvedEntities.getRawFullPath();
+    PolarisEntity returnedEntity =
+        Optional.ofNullable(
+                getMetaStoreManager()
+                    .updateEntityPropertiesIfNotChanged(
+                        getCurrentPolarisContext(),
+                        PolarisEntity.toCoreList(parentPath),
+                        updatedEntity)
+                    .getEntity())
+            .map(PolarisEntity::new)
+            .orElse(null);
+    if (returnedEntity == null) {
+      throw new CommitConflictException("Concurrent modification of namespace: %s", namespace);
+    }
+    return true;
+  }
+
   @Override
   public boolean removeProperties(Namespace namespace, Set<String> properties)
       throws NoSuchNamespaceException {

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/AbstractIcebergCatalogTest.java

@@ -35,6 +35,7 @@
 import com.azure.core.exception.HttpResponseException;
 import com.google.cloud.storage.StorageException;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Streams;
 import io.quarkus.test.junit.QuarkusMock;
@@ -2614,4 +2615,30 @@ public void testPaginatedListNamespaces() {
       }
     }
   }
+
+  @Test
+  public void testUpdateNamespacePropertiesAtomic() {
+    catalog.createNamespace(
+        NS,
+        ImmutableMap.of(
+            "prop1", "val1",
+            "prop2", "val2",
+            "prop3", "val3"));
+
+    catalog.updateProperties(
+        NS,
+        ImmutableSet.of("prop1", "prop2", "missing_prop"),
+        ImmutableMap.of(
+            "prop3", "new_val3",
+            "prop4", "val4",
+            "prop5", "val5"));
+
+    Map<String, String> properties = catalog.loadNamespaceMetadata(NS);
+    Assertions.assertThat(properties).containsEntry("prop3", "new_val3");
+    Assertions.assertThat(properties).containsEntry("prop4", "val4");
+    Assertions.assertThat(properties).containsEntry("prop5", "val5");
+    Assertions.assertThat(properties).doesNotContainKey("prop1");
+    Assertions.assertThat(properties).doesNotContainKey("prop2");
+  }
 }
+