Implement federation to HadoopCatalog (#1466)

eric-maynard · web-flow · commit 24683d86b46c · 2025-05-09T15:18:05.000-07:00
* wip

* quarkus fixes

* autolint

* hadoop impl

* autolint

* Refactors

* refactored

* autolint

* add config

* autolint

* stable

* Remove breakpoint anchor

* add line to application.properties

* yank HADOOP

* autolint
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java b/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.connection.ConnectionType;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.persistence.cache.EntityWeigher;
 
@@ -241,4 +242,11 @@ public static void enforceFeatureEnabledOrThrow(
           .description("If true, the policy-store endpoints are enabled")
           .defaultValue(true)
           .buildFeatureConfiguration();
+
+  public static final FeatureConfiguration<List<String>> SUPPORTED_CATALOG_CONNECTION_TYPES =
+      PolarisConfiguration.<List<String>>builder()
+          .key("SUPPORTED_CATALOG_CONNECTION_TYPES")
+          .description("The list of supported catalog connection types for federation")
+          .defaultValue(List.of(ConnectionType.ICEBERG_REST.name()))
+          .buildFeatureConfiguration();
 }
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/connection/AuthenticationParametersDpo.java b/polaris-core/src/main/java/org/apache/polaris/core/connection/AuthenticationParametersDpo.java
@@ -25,6 +25,7 @@
 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.connection.iceberg.IcebergCatalogPropertiesProvider;
 import org.apache.polaris.core.secrets.UserSecretReference;
 
 /**
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/connection/ConnectionConfigInfoDpo.java b/polaris-core/src/main/java/org/apache/polaris/core/connection/ConnectionConfigInfoDpo.java
@@ -33,7 +33,11 @@
 import java.util.Map;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.admin.model.ConnectionConfigInfo;
+import org.apache.polaris.core.admin.model.HadoopConnectionConfigInfo;
 import org.apache.polaris.core.admin.model.IcebergRestConnectionConfigInfo;
+import org.apache.polaris.core.connection.hadoop.HadoopConnectionConfigInfoDpo;
+import org.apache.polaris.core.connection.iceberg.IcebergCatalogPropertiesProvider;
+import org.apache.polaris.core.connection.iceberg.IcebergRestConnectionConfigInfoDpo;
 import org.apache.polaris.core.secrets.UserSecretReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -48,6 +52,7 @@
     property = "connectionTypeCode")
 @JsonSubTypes({
   @JsonSubTypes.Type(value = IcebergRestConnectionConfigInfoDpo.class, name = "1"),
+  @JsonSubTypes.Type(value = HadoopConnectionConfigInfoDpo.class, name = "2"),
 })
 public abstract class ConnectionConfigInfoDpo implements IcebergCatalogPropertiesProvider {
   private static final Logger logger = LoggerFactory.getLogger(ConnectionConfigInfoDpo.class);
@@ -140,11 +145,12 @@ public static ConnectionConfigInfoDpo fromConnectionConfigInfoModelWithSecrets(
       ConnectionConfigInfo connectionConfigurationModel,
       Map<String, UserSecretReference> secretReferences) {
     ConnectionConfigInfoDpo config = null;
+    final AuthenticationParametersDpo authenticationParameters;
     switch (connectionConfigurationModel.getConnectionType()) {
       case ICEBERG_REST:
         IcebergRestConnectionConfigInfo icebergRestConfigModel =
             (IcebergRestConnectionConfigInfo) connectionConfigurationModel;
-        AuthenticationParametersDpo authenticationParameters =
+        authenticationParameters =
             AuthenticationParametersDpo.fromAuthenticationParametersModelWithSecrets(
                 icebergRestConfigModel.getAuthenticationParameters(), secretReferences);
         config =
@@ -153,6 +159,18 @@ public static ConnectionConfigInfoDpo fromConnectionConfigInfoModelWithSecrets(
                 authenticationParameters,
                 icebergRestConfigModel.getRemoteCatalogName());
         break;
+      case HADOOP:
+        HadoopConnectionConfigInfo hadoopConfigModel =
+            (HadoopConnectionConfigInfo) connectionConfigurationModel;
+        authenticationParameters =
+            AuthenticationParametersDpo.fromAuthenticationParametersModelWithSecrets(
+                hadoopConfigModel.getAuthenticationParameters(), secretReferences);
+        config =
+            new HadoopConnectionConfigInfoDpo(
+                hadoopConfigModel.getUri(),
+                authenticationParameters,
+                hadoopConfigModel.getWarehouse());
+        break;
       default:
         throw new IllegalStateException(
             "Unsupported connection type: " + connectionConfigurationModel.getConnectionType());
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/connection/ConnectionType.java b/polaris-core/src/main/java/org/apache/polaris/core/connection/ConnectionType.java
@@ -32,6 +32,7 @@
 public enum ConnectionType {
   NULL_TYPE(0),
   ICEBERG_REST(1),
+  HADOOP(2),
   ;
 
   private static final ConnectionType[] REVERSE_MAPPING_ARRAY;
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/connection/hadoop/HadoopConnectionConfigInfoDpo.java b/polaris-core/src/main/java/org/apache/polaris/core/connection/hadoop/HadoopConnectionConfigInfoDpo.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.hadoop;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.MoreObjects;
+import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.polaris.core.admin.model.ConnectionConfigInfo;
+import org.apache.polaris.core.admin.model.HadoopConnectionConfigInfo;
+import org.apache.polaris.core.connection.AuthenticationParametersDpo;
+import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
+import org.apache.polaris.core.connection.ConnectionType;
+import org.apache.polaris.core.secrets.UserSecretsManager;
+
+/**
+ * The internal persistence-object counterpart to {@link
+ * org.apache.polaris.core.admin.model.HadoopConnectionConfigInfo} defined in the API model.
+ */
+public class HadoopConnectionConfigInfoDpo extends ConnectionConfigInfoDpo {
+
+  private final String warehouse;
+
+  public HadoopConnectionConfigInfoDpo(
+      @JsonProperty(value = "uri", required = true) @Nonnull String uri,
+      @JsonProperty(value = "authenticationParameters", required = true) @Nonnull
+          AuthenticationParametersDpo authenticationParameters,
+      @JsonProperty(value = "warehouse", required = false) @Nullable String remoteCatalogName) {
+    super(ConnectionType.HADOOP.getCode(), uri, authenticationParameters);
+    this.warehouse = remoteCatalogName;
+  }
+
+  public String getWarehouse() {
+    return warehouse;
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("connectionTypeCode", getConnectionTypeCode())
+        .add("uri", getUri())
+        .add("warehouse", getWarehouse())
+        .add("authenticationParameters", getAuthenticationParameters().toString())
+        .toString();
+  }
+
+  @Override
+  public @Nonnull Map<String, String> asIcebergCatalogProperties(
+      UserSecretsManager secretsManager) {
+    HashMap<String, String> properties = new HashMap<>();
+    properties.put(CatalogProperties.URI, getUri());
+    if (getWarehouse() != null) {
+      properties.put(CatalogProperties.WAREHOUSE_LOCATION, getWarehouse());
+    }
+    properties.putAll(getAuthenticationParameters().asIcebergCatalogProperties(secretsManager));
+    return properties;
+  }
+
+  @Override
+  public ConnectionConfigInfo asConnectionConfigInfoModel() {
+    return HadoopConnectionConfigInfo.builder()
+        .setConnectionType(ConnectionConfigInfo.ConnectionTypeEnum.HADOOP)
+        .setUri(getUri())
+        .setWarehouse(getWarehouse())
+        .setAuthenticationParameters(
+            getAuthenticationParameters().asAuthenticationParametersModel())
+        .build();
+  }
+}
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/connection/iceberg/IcebergCatalogPropertiesProvider.java b/polaris-core/src/main/java/org/apache/polaris/core/connection/iceberg/IcebergCatalogPropertiesProvider.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.polaris.core.connection;
+package org.apache.polaris.core.connection.iceberg;
 
 import jakarta.annotation.Nonnull;
 import java.util.Map;
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/connection/iceberg/IcebergRestConnectionConfigInfoDpo.java b/polaris-core/src/main/java/org/apache/polaris/core/connection/iceberg/IcebergRestConnectionConfigInfoDpo.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.polaris.core.connection;
+package org.apache.polaris.core.connection.iceberg;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.MoreObjects;
@@ -27,6 +27,9 @@
 import org.apache.iceberg.CatalogProperties;
 import org.apache.polaris.core.admin.model.ConnectionConfigInfo;
 import org.apache.polaris.core.admin.model.IcebergRestConnectionConfigInfo;
+import org.apache.polaris.core.connection.AuthenticationParametersDpo;
+import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
+import org.apache.polaris.core.connection.ConnectionType;
 import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
diff --git a/quarkus/defaults/src/main/resources/application.properties b/quarkus/defaults/src/main/resources/application.properties
@@ -111,6 +111,7 @@ polaris.realm-context.require-header=false
 polaris.features.defaults."ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING"=false
 polaris.features.defaults."SUPPORTED_CATALOG_STORAGE_TYPES"=["S3","GCS","AZURE","FILE"]
 # polaris.features.defaults."ENABLE_CATALOG_FEDERATION"=true
+polaris.features.defaults."SUPPORTED_CATALOG_CONNECTION_TYPES"=["ICEBERG_REST"]
 
 # realm overrides
 # polaris.features.realm-overrides."my-realm"."INITIALIZE_DEFAULT_CATALOG_FILEIO_FOR_TEST"=true
diff --git a/quarkus/service/src/test/java/org/apache/polaris/service/quarkus/admin/PolarisAuthzTestBase.java b/quarkus/service/src/test/java/org/apache/polaris/service/quarkus/admin/PolarisAuthzTestBase.java
@@ -87,8 +87,8 @@
 import org.apache.polaris.service.config.DefaultConfigurationStore;
 import org.apache.polaris.service.config.RealmEntityManagerFactory;
 import org.apache.polaris.service.config.ReservedProperties;
-import org.apache.polaris.service.context.CallContextCatalogFactory;
-import org.apache.polaris.service.context.PolarisCallContextCatalogFactory;
+import org.apache.polaris.service.context.catalog.CallContextCatalogFactory;
+import org.apache.polaris.service.context.catalog.PolarisCallContextCatalogFactory;
 import org.apache.polaris.service.events.PolarisEventListener;
 import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl;
 import org.apache.polaris.service.task.TaskExecutor;
diff --git a/quarkus/service/src/test/java/org/apache/polaris/service/quarkus/catalog/IcebergCatalogHandlerAuthzTest.java b/quarkus/service/src/test/java/org/apache/polaris/service/quarkus/catalog/IcebergCatalogHandlerAuthzTest.java
@@ -68,8 +68,8 @@
 import org.apache.polaris.service.catalog.iceberg.IcebergCatalogHandler;
 import org.apache.polaris.service.catalog.io.DefaultFileIOFactory;
 import org.apache.polaris.service.config.RealmEntityManagerFactory;
-import org.apache.polaris.service.context.CallContextCatalogFactory;
-import org.apache.polaris.service.context.PolarisCallContextCatalogFactory;
+import org.apache.polaris.service.context.catalog.CallContextCatalogFactory;
+import org.apache.polaris.service.context.catalog.PolarisCallContextCatalogFactory;
 import org.apache.polaris.service.http.IfNoneMatch;
 import org.apache.polaris.service.quarkus.admin.PolarisAuthzTestBase;
 import org.apache.polaris.service.types.NotificationRequest;
diff --git a/service/common/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java b/service/common/src/main/java/org/apache/polaris/service/admin/PolarisServiceImpl.java
@@ -23,6 +23,7 @@
 import jakarta.ws.rs.core.Response;
 import jakarta.ws.rs.core.SecurityContext;
 import java.util.List;
+import java.util.Locale;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.NotAuthorizedException;
@@ -37,6 +38,7 @@
 import org.apache.polaris.core.admin.model.CreateCatalogRoleRequest;
 import org.apache.polaris.core.admin.model.CreatePrincipalRequest;
 import org.apache.polaris.core.admin.model.CreatePrincipalRoleRequest;
+import org.apache.polaris.core.admin.model.ExternalCatalog;
 import org.apache.polaris.core.admin.model.GrantCatalogRoleRequest;
 import org.apache.polaris.core.admin.model.GrantPrincipalRoleRequest;
 import org.apache.polaris.core.admin.model.GrantResource;
@@ -141,6 +143,7 @@ public Response createCatalog(
     PolarisAdminService adminService = newAdminService(realmContext, securityContext);
     Catalog catalog = request.getCatalog();
     validateStorageConfig(catalog.getStorageConfigInfo());
+    validateConnectionConfigInfo(catalog);
     Catalog newCatalog = new CatalogEntity(adminService.createCatalog(request)).asCatalog();
     LOGGER.info("Created new catalog {}", newCatalog);
     return Response.status(Response.Status.CREATED).build();
@@ -163,6 +166,30 @@ private void validateStorageConfig(StorageConfigInfo storageConfigInfo) {
     }
   }
 
+  private void validateConnectionConfigInfo(Catalog catalog) {
+    if (catalog.getType() == Catalog.TypeEnum.EXTERNAL) {
+      if (catalog instanceof ExternalCatalog externalCatalog) {
+        if (externalCatalog.getConnectionConfigInfo() != null) {
+          String connectionType =
+              externalCatalog.getConnectionConfigInfo().getConnectionType().name();
+          List<String> supportedConnectionTypes =
+              callContext
+                  .getPolarisCallContext()
+                  .getConfigurationStore()
+                  .getConfiguration(
+                      callContext.getPolarisCallContext(),
+                      FeatureConfiguration.SUPPORTED_CATALOG_CONNECTION_TYPES)
+                  .stream()
+                  .map(s -> s.toUpperCase(Locale.ROOT))
+                  .toList();
+          if (!supportedConnectionTypes.contains(connectionType)) {
+            throw new IllegalStateException("Unsupported connection type: " + connectionType);
+          }
+        }
+      }
+    }
+  }
+
   /** From PolarisCatalogsApiService */
   @Override
   public Response deleteCatalog(
diff --git a/service/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogAdapter.java b/service/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogAdapter.java
@@ -75,7 +75,7 @@
 import org.apache.polaris.service.catalog.api.IcebergRestConfigurationApiService;
 import org.apache.polaris.service.catalog.common.CatalogAdapter;
 import org.apache.polaris.service.config.ReservedProperties;
-import org.apache.polaris.service.context.CallContextCatalogFactory;
+import org.apache.polaris.service.context.catalog.CallContextCatalogFactory;
 import org.apache.polaris.service.http.IcebergHttpUtil;
 import org.apache.polaris.service.http.IfNoneMatch;
 import org.apache.polaris.service.types.CommitTableRequest;
diff --git a/service/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java b/service/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java
@@ -54,6 +54,7 @@
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hadoop.HadoopCatalog;
 import org.apache.iceberg.rest.CatalogHandlers;
 import org.apache.iceberg.rest.HTTPClient;
 import org.apache.iceberg.rest.RESTCatalog;
@@ -79,7 +80,8 @@
 import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
 import org.apache.polaris.core.connection.ConnectionType;
-import org.apache.polaris.core.connection.IcebergRestConnectionConfigInfoDpo;
+import org.apache.polaris.core.connection.hadoop.HadoopConnectionConfigInfoDpo;
+import org.apache.polaris.core.connection.iceberg.IcebergRestConnectionConfigInfoDpo;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.CatalogEntity;
 import org.apache.polaris.core.entity.PolarisEntitySubType;
@@ -97,7 +99,7 @@
 import org.apache.polaris.service.catalog.SupportsNotifications;
 import org.apache.polaris.service.catalog.common.CatalogHandler;
 import org.apache.polaris.service.config.ReservedProperties;
-import org.apache.polaris.service.context.CallContextCatalogFactory;
+import org.apache.polaris.service.context.catalog.CallContextCatalogFactory;
 import org.apache.polaris.service.http.IcebergHttpUtil;
 import org.apache.polaris.service.http.IfNoneMatch;
 import org.apache.polaris.service.types.NotificationRequest;
@@ -212,6 +214,7 @@ protected void initializeCatalog() {
       Catalog federatedCatalog;
       ConnectionType connectionType =
           ConnectionType.fromCode(connectionConfigInfoDpo.getConnectionTypeCode());
+
       switch (connectionType) {
         case ICEBERG_REST:
           SessionCatalog.SessionContext context = SessionCatalog.SessionContext.createEmpty();
@@ -226,6 +229,12 @@ protected void initializeCatalog() {
               ((IcebergRestConnectionConfigInfoDpo) connectionConfigInfoDpo).getRemoteCatalogName(),
               connectionConfigInfoDpo.asIcebergCatalogProperties(getUserSecretsManager()));
           break;
+        case HADOOP:
+          federatedCatalog = new HadoopCatalog();
+          federatedCatalog.initialize(
+              ((HadoopConnectionConfigInfoDpo) connectionConfigInfoDpo).getWarehouse(),
+              connectionConfigInfoDpo.asIcebergCatalogProperties(getUserSecretsManager()));
+          break;
         default:
           throw new UnsupportedOperationException(
               "Connection type not supported: " + connectionType);
diff --git a/service/common/src/main/java/org/apache/polaris/service/context/catalog/CallContextCatalogFactory.java b/service/common/src/main/java/org/apache/polaris/service/context/catalog/CallContextCatalogFactory.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.polaris.service.context;
+package org.apache.polaris.service.context.catalog;
 
 import jakarta.ws.rs.core.SecurityContext;
 import org.apache.iceberg.catalog.Catalog;
@@ -25,6 +25,9 @@
 import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
 
 public interface CallContextCatalogFactory {
+
+  static final String WAREHOUSE_LOCATION_BASEDIR = "/tmp/iceberg_rest_server_warehouse_data/";
+
   Catalog createCallContextCatalog(
       CallContext context,
       AuthenticatedPolarisPrincipal authenticatedPrincipal,
diff --git a/service/common/src/main/java/org/apache/polaris/service/context/catalog/PolarisCallContextCatalogFactory.java b/service/common/src/main/java/org/apache/polaris/service/context/catalog/PolarisCallContextCatalogFactory.java
@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.polaris.service.context;
+package org.apache.polaris.service.context.catalog;
 
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
@@ -48,9 +48,6 @@ public class PolarisCallContextCatalogFactory implements CallContextCatalogFacto
   private static final Logger LOGGER =
       LoggerFactory.getLogger(PolarisCallContextCatalogFactory.class);
 
-  private static final String WAREHOUSE_LOCATION_BASEDIR =
-      "/tmp/iceberg_rest_server_warehouse_data/";
-
   private final RealmEntityManagerFactory entityManagerFactory;
   private final TaskExecutor taskExecutor;
   private final FileIOFactory fileIOFactory;
diff --git a/service/common/src/testFixtures/java/org/apache/polaris/service/TestServices.java b/service/common/src/testFixtures/java/org/apache/polaris/service/TestServices.java
@@ -55,8 +55,8 @@
 import org.apache.polaris.service.config.DefaultConfigurationStore;
 import org.apache.polaris.service.config.RealmEntityManagerFactory;
 import org.apache.polaris.service.config.ReservedProperties;
-import org.apache.polaris.service.context.CallContextCatalogFactory;
-import org.apache.polaris.service.context.PolarisCallContextCatalogFactory;
+import org.apache.polaris.service.context.catalog.CallContextCatalogFactory;
+import org.apache.polaris.service.context.catalog.PolarisCallContextCatalogFactory;
 import org.apache.polaris.service.events.PolarisEventListener;
 import org.apache.polaris.service.events.TestPolarisEventListener;
 import org.apache.polaris.service.persistence.InMemoryPolarisMetaStoreManagerFactory;
diff --git a/spec/polaris-management-service.yml b/spec/polaris-management-service.yml

Original file line number	Diff line number	Diff line change
`@@ -32,6 +32,7 @@`
`32`	`32`	`public enum ConnectionType {`
`33`	`33`	`NULL_TYPE(0),`
`34`	`34`	`ICEBERG_REST(1),`
	`35`	`+ HADOOP(2),`
`35`	`36`	`;`
`36`	`37`
`37`	`38`	`private static final ConnectionType[] REVERSE_MAPPING_ARRAY;`