Initial MVP implementation of Catalog Federation to remote Iceberg REST Catalogs

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

@@ -378,7 +378,7 @@ public void testIcebergCreateTablesInExternalCatalog() throws IOException {
                       .withPartitionSpec(PartitionSpec.unpartitioned())
                       .create())
           .isInstanceOf(BadRequestException.class)
-          .hasMessage("Malformed request: Cannot create table on external catalogs.");
+          .hasMessage("Malformed request: Cannot create table on static-facade external catalogs.");
     }
   }
 
@@ -515,7 +515,7 @@ public void testIcebergUpdateTableInExternalCatalog() throws IOException {
                               10L))
                       .commit())
           .isInstanceOf(BadRequestException.class)
-          .hasMessage("Malformed request: Cannot update table on external catalogs.");
+          .hasMessage("Malformed request: Cannot update table on static-facade external catalogs.");
     }
   }
 

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.Optional;
 import org.apache.polaris.core.admin.model.StorageConfigInfo;
+import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.persistence.cache.EntityWeigher;
 
 /**
@@ -36,6 +37,22 @@ protected FeatureConfiguration(
     super(key, description, defaultValue, catalogConfig);
   }
 
+  /**
+   * Helper for the common scenario of gating a feature with a boolean FeatureConfiguration, where
+   * we want to throw an UnsupportedOperationException if it's not enabled.
+   */
+  public static void enforceFeatureEnabledOrThrow(
+      CallContext callContext, FeatureConfiguration<Boolean> featureConfig) {
+    boolean enabled =
+        callContext
+            .getPolarisCallContext()
+            .getConfigurationStore()
+            .getConfiguration(callContext.getPolarisCallContext(), featureConfig);
+    if (!enabled) {
+      throw new UnsupportedOperationException("Feature not enabled: " + featureConfig.key);
+    }
+  }
+
   public static final FeatureConfiguration<Boolean>
       ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING =
           PolarisConfiguration.<Boolean>builder()
@@ -201,4 +218,13 @@ protected FeatureConfiguration(
                   + " requires experimentation in the specific deployment environment")
           .defaultValue(100 * EntityWeigher.WEIGHT_PER_MB)
           .buildFeatureConfiguration();
+
+  public static final FeatureConfiguration<Boolean> ENABLE_CATALOG_FEDERATION =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ENABLE_CATALOG_FEDERATION")
+          .description(
+              "If true, allows creating and using ExternalCatalogs containing ConnectionConfigInfos"
+                  + " to perform federation to remote catalogs.")
+          .defaultValue(false)
+          .buildFeatureConfiguration();
 }

polaris-core/src/main/java/org/apache/polaris/core/connection/AuthenticationParametersDpo.java

@@ -0,0 +1,88 @@
+/*
+ * 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.core.connection;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import java.util.Map;
+import org.apache.polaris.core.admin.model.AuthenticationParameters;
+import org.apache.polaris.core.admin.model.BearerAuthenticationParameters;
+import org.apache.polaris.core.admin.model.OAuthClientCredentialsParameters;
+import org.apache.polaris.core.secrets.UserSecretReference;
+
+/**
+ * The internal persistence-object counterpart to AuthenticationParameters defined in the API model.
+ * Important: JsonSubTypes must be kept in sync with {@link AuthenticationType}.
+ */
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "authenticationTypeCode", visible = true)
+@JsonSubTypes({
+  @JsonSubTypes.Type(value = OAuthClientCredentialsParametersDpo.class, name = "1"),
+  @JsonSubTypes.Type(value = BearerAuthenticationParametersDpo.class, name = "2"),
+})
+public abstract class AuthenticationParametersDpo implements IcebergCatalogPropertiesProvider {
+
+  public static final String INLINE_CLIENT_SECRET_REFERENCE_KEY = "inlineClientSecretReference";
+  public static final String INLINE_BEARER_TOKEN_REFERENCE_KEY = "inlineBearerTokenReference";
+
+  @JsonProperty(value = "authenticationTypeCode")
+  private final int authenticationTypeCode;
+
+  public AuthenticationParametersDpo(
+      @JsonProperty(value = "authenticationTypeCode", required = true) int authenticationTypeCode) {
+    this.authenticationTypeCode = authenticationTypeCode;
+  }
+
+  public int getAuthenticationTypeCode() {
+    return authenticationTypeCode;
+  }
+
+  public abstract AuthenticationParameters asAuthenticationParametersModel();
+
+  public static AuthenticationParametersDpo fromAuthenticationParametersModelWithSecrets(
+      AuthenticationParameters authenticationParameters,
+      Map<String, UserSecretReference> secretReferences) {
+    final AuthenticationParametersDpo config;
+    switch (authenticationParameters.getAuthenticationType()) {
+      case OAUTH:
+        OAuthClientCredentialsParameters oauthClientCredentialsModel =
+            (OAuthClientCredentialsParameters) authenticationParameters;
+        config =
+            new OAuthClientCredentialsParametersDpo(
+                AuthenticationType.OAUTH.getCode(),
+                oauthClientCredentialsModel.getTokenUri(),
+                oauthClientCredentialsModel.getClientId(),
+                secretReferences.get(INLINE_CLIENT_SECRET_REFERENCE_KEY),
+                oauthClientCredentialsModel.getScopes());
+        break;
+      case BEARER:
+        BearerAuthenticationParameters bearerAuthenticationParametersModel =
+            (BearerAuthenticationParameters) authenticationParameters;
+        config =
+            new BearerAuthenticationParametersDpo(
+                AuthenticationType.BEARER.getCode(),
+                secretReferences.get(INLINE_BEARER_TOKEN_REFERENCE_KEY));
+        break;
+      default:
+        throw new IllegalStateException(
+            "Unsupported authentication type: " + authenticationParameters.getAuthenticationType());
+    }
+    return config;
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/connection/AuthenticationType.java

@@ -0,0 +1,42 @@
+/*
+ * 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.core.connection;
+
+/**
+ * The internal persistence-object counterpart to AuthenticationParameters.AuthenticationTypeEnum
+ * defined in the API model. We define integer type codes in this enum for better compatibility
+ * within persisted data in case the names of enum types are ever changed in place.
+ *
+ * <p>Important: Codes must be kept in-sync with JsonSubTypes annotated within {@link
+ * AuthenticationParametersDpo}.
+ */
+public enum AuthenticationType {
+  OAUTH(1),
+  BEARER(2);
+
+  private final int code;
+
+  AuthenticationType(int code) {
+    this.code = code;
+  }
+
+  public int getCode() {
+    return this.code;
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/connection/BearerAuthenticationParametersDpo.java

@@ -0,0 +1,73 @@
+/*
+ * 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.core.connection;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.MoreObjects;
+import jakarta.annotation.Nonnull;
+import java.util.Map;
+import org.apache.iceberg.rest.auth.OAuth2Properties;
+import org.apache.polaris.core.admin.model.AuthenticationParameters;
+import org.apache.polaris.core.admin.model.BearerAuthenticationParameters;
+import org.apache.polaris.core.secrets.UserSecretReference;
+import org.apache.polaris.core.secrets.UserSecretsManager;
+
+/**
+ * The internal persistence-object counterpart to BearerAuthenticationParameters defined in the API
+ * model.
+ */
+public class BearerAuthenticationParametersDpo extends AuthenticationParametersDpo {
+
+  @JsonProperty(value = "bearerTokenReference")
+  private final UserSecretReference bearerTokenReference;
+
+  public BearerAuthenticationParametersDpo(
+      @JsonProperty(value = "authenticationTypeCode", required = true) int authenticationTypeCode,
+      @JsonProperty(value = "bearerTokenReference", required = true) @Nonnull
+          UserSecretReference bearerTokenReference) {
+    super(authenticationTypeCode);
+    this.bearerTokenReference = bearerTokenReference;
+  }
+
+  public @Nonnull UserSecretReference getBearerTokenReference() {
+    return bearerTokenReference;
+  }
+
+  @Override
+  public @Nonnull Map<String, String> asIcebergCatalogProperties(
+      UserSecretsManager secretsManager) {
+    String bearerToken = secretsManager.readSecret(getBearerTokenReference());
+    return Map.of(OAuth2Properties.TOKEN, bearerToken);
+  }
+
+  @Override
+  public AuthenticationParameters asAuthenticationParametersModel() {
+    return BearerAuthenticationParameters.builder()
+        .setAuthenticationType(AuthenticationParameters.AuthenticationTypeEnum.BEARER)
+        .build();
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("authenticationTypeCode", getAuthenticationTypeCode())
+        .add("bearerTokenReference", getBearerTokenReference())
+        .toString();
+  }
+}