Add IcebergCatalogPrefixParser (#923)

This is an extension point to allow customizing the relationship
between realm / catalog and Iceberg REST API prefix.

service/common/src/main/java/org/apache/polaris/service/catalog/DefaultIcebergCatalogPrefixParser.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.catalog;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import org.apache.polaris.core.context.RealmContext;
+
+@ApplicationScoped
+public class DefaultIcebergCatalogPrefixParser implements IcebergCatalogPrefixParser {
+  @Override
+  public String prefixToCatalogName(RealmContext realm, String prefix) {
+    return prefix;
+  }
+
+  @Override
+  public String catalogNameToPrefix(RealmContext realm, String catalogName) {
+    return catalogName;
+  }
+}

service/common/src/main/java/org/apache/polaris/service/catalog/IcebergCatalogAdapter.java

@@ -126,6 +126,7 @@ public class IcebergCatalogAdapter
   private final PolarisConfigurationStore configurationStore;
   private final PolarisDiagnostics diagnostics;
   private final PolarisAuthorizer polarisAuthorizer;
+  private final IcebergCatalogPrefixParser prefixParser;
 
   @Inject
   public IcebergCatalogAdapter(
@@ -136,7 +137,8 @@ public IcebergCatalogAdapter(
       PolarisMetaStoreSession session,
       PolarisConfigurationStore configurationStore,
       PolarisDiagnostics diagnostics,
-      PolarisAuthorizer polarisAuthorizer) {
+      PolarisAuthorizer polarisAuthorizer,
+      IcebergCatalogPrefixParser prefixParser) {
     this.realmContext = realmContext;
     this.catalogFactory = catalogFactory;
     this.entityManager = entityManager;
@@ -145,6 +147,7 @@ public IcebergCatalogAdapter(
     this.configurationStore = configurationStore;
     this.diagnostics = diagnostics;
     this.polarisAuthorizer = polarisAuthorizer;
+    this.prefixParser = prefixParser;
   }
 
   /**
@@ -153,8 +156,9 @@ public IcebergCatalogAdapter(
    */
   private Response withCatalog(
       SecurityContext securityContext,
-      String catalogName,
+      String prefix,
       Function<PolarisCatalogHandlerWrapper, Response> action) {
+    String catalogName = prefixParser.prefixToCatalogName(realmContext, prefix);
     try (PolarisCatalogHandlerWrapper wrapper = newHandlerWrapper(securityContext, catalogName)) {
       return action.apply(wrapper);
     } catch (RuntimeException e) {
@@ -629,10 +633,11 @@ public Response getConfig(
     Map<String, String> properties =
         PolarisEntity.of(resolvedReferenceCatalog.getEntity()).getPropertiesAsMap();
 
+    String prefix = prefixParser.catalogNameToPrefix(realmContext, warehouse);
     return Response.ok(
             ConfigResponse.builder()
                 .withDefaults(properties) // catalog properties are defaults
-                .withOverrides(ImmutableMap.of("prefix", warehouse))
+                .withOverrides(ImmutableMap.of("prefix", prefix))
                 .withEndpoints(
                     ImmutableList.<Endpoint>builder()
                         .addAll(DEFAULT_ENDPOINTS)

service/common/src/main/java/org/apache/polaris/service/catalog/IcebergCatalogPrefixParser.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.catalog;
+
+import org.apache.polaris.core.context.RealmContext;
+
+/** An extension point for converting Iceberg REST API "prefix" values to Polaris Catalog names. */
+public interface IcebergCatalogPrefixParser {
+
+  /**
+   * Produces the name of a Polaris catalog from the given Iceberg Catalog REST API "prefix" for the
+   * specified realm.
+   *
+   * @param realm identifies the realm where the API request is to be served.
+   * @param prefix the "prefix" according to the Iceberg REST Catalog API specification.
+   * @return Polaris Catalog name
+   */
+  String prefixToCatalogName(RealmContext realm, String prefix);
+
+  /**
+   * Produces the "prefix" according to the Iceberg REST Catalog API specification for the given
+   * Polaris catalog name in the specified realm.
+   *
+   * @param realm identifies the realm owning the catalog
+   * @param catalogName name of a Polaris catalog.
+   * @return the "prefix" for the Iceberg REST client to be used for requests to the given catalog.
+   */
+  String catalogNameToPrefix(RealmContext realm, String catalogName);
+}

service/common/src/testFixtures/java/org/apache/polaris/service/TestServices.java

@@ -40,6 +40,7 @@
 import org.apache.polaris.core.persistence.PolarisMetaStoreSession;
 import org.apache.polaris.service.admin.PolarisServiceImpl;
 import org.apache.polaris.service.admin.api.PolarisCatalogsApi;
+import org.apache.polaris.service.catalog.DefaultIcebergCatalogPrefixParser;
 import org.apache.polaris.service.catalog.IcebergCatalogAdapter;
 import org.apache.polaris.service.catalog.api.IcebergRestCatalogApi;
 import org.apache.polaris.service.catalog.api.IcebergRestCatalogApiService;
@@ -163,7 +164,8 @@ public TestServices build() {
               metaStoreSession,
               configurationStore,
               polarisDiagnostics,
-              authorizer);
+              authorizer,
+              new DefaultIcebergCatalogPrefixParser());
 
       IcebergRestCatalogApi restApi = new IcebergRestCatalogApi(service);