Unblock test createViewWithCustomMetadataLocation

integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogViewIntegrationBase.java

@@ -20,9 +20,16 @@
 
 import static org.apache.polaris.service.it.env.PolarisClient.polarisClient;
 
+import com.google.common.collect.ImmutableMap;
 import java.lang.reflect.Method;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.Map;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.rest.RESTCatalog;
+import org.apache.iceberg.view.BaseView;
+import org.apache.iceberg.view.View;
 import org.apache.iceberg.view.ViewCatalogTests;
 import org.apache.polaris.core.admin.model.Catalog;
 import org.apache.polaris.core.admin.model.CatalogProperties;
@@ -31,22 +38,24 @@
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.entity.CatalogEntity;
+import org.apache.polaris.core.entity.table.IcebergTableLikeEntity;
 import org.apache.polaris.service.it.env.ClientCredentials;
 import org.apache.polaris.service.it.env.IcebergHelper;
 import org.apache.polaris.service.it.env.ManagementApi;
 import org.apache.polaris.service.it.env.PolarisApiEndpoints;
 import org.apache.polaris.service.it.env.PolarisClient;
 import org.apache.polaris.service.it.ext.PolarisIntegrationTestExtension;
+import org.assertj.core.api.Assertions;
 import org.assertj.core.api.Assumptions;
 import org.assertj.core.configuration.PreferredAssumptionException;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
 
 /**
  * Import the full core Iceberg catalog tests by hitting the REST service via the RESTCatalog
@@ -175,15 +184,81 @@ protected boolean overridesRequestedLocation() {
     return true;
   }
 
-  /** TODO: Unblock this test, see: https://github.com/apache/polaris/issues/1273 */
   @Override
   @Test
-  @Disabled(
-      """
-      Disabled because the behavior is not applicable to Polaris.
-      To unblock, update this to expect an exception and add a Polaris-specific test.
-      """)
   public void createViewWithCustomMetadataLocation() {
-    super.createViewWithCustomMetadataLocation();
+    Assertions.assertThatThrownBy(super::createViewWithCustomMetadataLocation)
+        .isInstanceOf(ForbiddenException.class)
+        .hasMessageContaining("Forbidden: Invalid locations");
+  }
+
+  @Test
+  public void createViewWithCustomMetadataLocationUsingPolaris(@TempDir Path tempDir) {
+    TableIdentifier identifier = TableIdentifier.of("ns", "view");
+
+    String location = Paths.get(tempDir.toUri().toString()).toString();
+    String customLocation = Paths.get(tempDir.toUri().toString(), "custom-location").toString();
+    String customLocation2 = Paths.get(tempDir.toUri().toString(), "custom-location2").toString();
+    String customLocationChild =
+        Paths.get(tempDir.toUri().toString(), "custom-location/child").toString();
+
+    catalog()
+        .createNamespace(
+            identifier.namespace(),
+            ImmutableMap.of(
+                IcebergTableLikeEntity.USER_SPECIFIED_WRITE_METADATA_LOCATION_KEY, location));
+
+    Assertions.assertThat(catalog().viewExists(identifier)).as("View should not exist").isFalse();
+
+    // CAN create a view with a custom metadata location `baseLocation/customLocation`,
+    // as long as the location is within the parent namespace's `write.metadata.path=baseLocation`
+    View view =
+        catalog()
+            .buildView(identifier)
+            .withSchema(SCHEMA)
+            .withDefaultNamespace(identifier.namespace())
+            .withDefaultCatalog(catalog().name())
+            .withQuery("spark", "select * from ns.tbl")
+            .withProperty(
+                IcebergTableLikeEntity.USER_SPECIFIED_WRITE_METADATA_LOCATION_KEY, customLocation)
+            .withLocation(location)
+            .create();
+
+    Assertions.assertThat(view).isNotNull();
+    Assertions.assertThat(catalog().viewExists(identifier)).as("View should exist").isTrue();
+    Assertions.assertThat(view.properties()).containsEntry("write.metadata.path", customLocation);
+    Assertions.assertThat(((BaseView) view).operations().current().metadataFileLocation())
+        .isNotNull()
+        .startsWith(customLocation);
+
+    // CANNOT update the view with a new metadata location `baseLocation/customLocation2`,
+    // even though the new location is still under the parent namespace's
+    // `write.metadata.path=baseLocation`.
+    Assertions.assertThatThrownBy(
+            () ->
+                catalog()
+                    .loadView(identifier)
+                    .updateProperties()
+                    .set(
+                        IcebergTableLikeEntity.USER_SPECIFIED_WRITE_METADATA_LOCATION_KEY,
+                        customLocation2)
+                    .commit())
+        .isInstanceOf(ForbiddenException.class)
+        .hasMessageContaining("Forbidden: Invalid locations");
+
+    // CANNOT update the view with a child metadata location `baseLocation/customLocation/child`,
+    // even though it is a subpath of the original view's
+    // `write.metadata.path=baseLocation/customLocation`.
+    Assertions.assertThatThrownBy(
+            () ->
+                catalog()
+                    .loadView(identifier)
+                    .updateProperties()
+                    .set(
+                        IcebergTableLikeEntity.USER_SPECIFIED_WRITE_METADATA_LOCATION_KEY,
+                        customLocationChild)
+                    .commit())
+        .isInstanceOf(ForbiddenException.class)
+        .hasMessageContaining("Forbidden: Invalid locations");
   }
 }

polaris-core/src/main/java/org/apache/polaris/core/storage/PolarisStorageConfigurationInfo.java

@@ -181,6 +181,7 @@ public static Optional<PolarisStorageConfigurationInfo> forEntityPath(
                     "Allowing unstructured table location for entity: {}",
                     entityPathReversed.get(0).getName());
 
+                // TODO: figure out the purpose of adding `userSpecifiedWriteLocations`
                 List<String> locs =
                     userSpecifiedWriteLocations(entityPathReversed.get(0).getPropertiesAsMap());
                 return new StorageConfigurationOverride(