Add support for catalog federation in the CLI

client/python/cli/command/__init__.py

@@ -40,6 +40,7 @@ def options_get(key, f=lambda x: x):
         properties = Parser.parse_properties(options_get(Arguments.PROPERTY))
         set_properties = Parser.parse_properties(options_get(Arguments.SET_PROPERTY))
         remove_properties = options_get(Arguments.REMOVE_PROPERTY)
+        catalog_client_scopes = options_get(Arguments.CATALOG_CLIENT_SCOPE)
 
         command = None
         if options.command == Commands.CATALOGS:
@@ -61,7 +62,24 @@ def options_get(key, f=lambda x: x):
                 catalog_name=options_get(Arguments.CATALOG),
                 properties={} if properties is None else properties,
                 set_properties={} if set_properties is None else set_properties,
-                remove_properties=[] if remove_properties is None else remove_properties
+                hadoop_warehouse=options_get(Arguments.HADOOP_WAREHOUSE),
+                iceberg_remote_catalog_name=options_get(Arguments.ICEBERG_REMOTE_CATALOG_NAME),
+                remove_properties=[] if remove_properties is None else remove_properties,
+                catalog_connection_type=options_get(Arguments.CATALOG_CONNECTION_TYPE),
+                catalog_authentication_type=options_get(Arguments.CATALOG_AUTHENTICATION_TYPE),
+                catalog_service_identity_type=options_get(Arguments.CATALOG_SERVICE_IDENTITY_TYPE),
+                catalog_service_identity_iam_arn=options_get(Arguments.CATALOG_SERVICE_IDENTITY_IAM_ARN),
+                catalog_uri=options_get(Arguments.CATALOG_URI),
+                catalog_token_uri=options_get(Arguments.CATALOG_TOKEN_URI),
+                catalog_client_id=options_get(Arguments.CATALOG_CLIENT_ID),
+                catalog_client_secret=options_get(Arguments.CATALOG_CLIENT_SECRET),
+                catalog_client_scopes=[] if catalog_client_scopes is None else catalog_client_scopes,
+                catalog_bearer_token=options_get(Arguments.CATALOG_BEARER_TOKEN),
+                catalog_role_arn=options_get(Arguments.CATALOG_ROLE_ARN),
+                catalog_role_session_name=options_get(Arguments.CATALOG_ROLE_SESSION_NAME),
+                catalog_external_id=options_get(Arguments.CATALOG_EXTERNAL_ID),
+                catalog_signing_region=options_get(Arguments.CATALOG_SIGNING_REGION),
+                catalog_signing_name=options_get(Arguments.CATALOG_SIGNING_NAME)
             )
         elif options.command == Commands.PRINCIPALS:
             from cli.command.principals import PrincipalsCommand

client/python/cli/command/catalogs.py

@@ -19,14 +19,17 @@
 from dataclasses import dataclass, field
 from typing import Dict, Optional, List
 
-from pydantic import StrictStr
+from pydantic import StrictStr, SecretStr
 
 from cli.command import Command
-from cli.constants import StorageType, CatalogType, Subcommands, Arguments
+from cli.constants import StorageType, CatalogType, CatalogConnectionType, Subcommands, Arguments, AuthenticationType, \
+    ServiceIdentityType
 from cli.options.option_tree import Argument
 from polaris.management import PolarisDefaultApi, Catalog, CreateCatalogRequest, UpdateCatalogRequest, \
     StorageConfigInfo, ExternalCatalog, AwsStorageConfigInfo, AzureStorageConfigInfo, GcpStorageConfigInfo, \
-    PolarisCatalog, CatalogProperties
+    PolarisCatalog, CatalogProperties, AuthenticationParameters, BearerAuthenticationParameters, \
+    OAuthClientCredentialsParameters, SigV4AuthenticationParameters, HadoopConnectionConfigInfo, \
+    IcebergRestConnectionConfigInfo, AwsIamServiceIdentityInfo
 
 
 @dataclass
@@ -59,15 +62,56 @@ class CatalogsCommand(Command):
     properties: Dict[str, StrictStr]
     set_properties: Dict[str, StrictStr]
     remove_properties: List[str]
+    hadoop_warehouse: str
+    iceberg_remote_catalog_name: str
+    catalog_connection_type: str
+    catalog_authentication_type: str
+    catalog_service_identity_type: str
+    catalog_service_identity_iam_arn: str
+    catalog_uri: str
+    catalog_token_uri: str
+    catalog_client_id: str
+    catalog_client_secret: str
+    catalog_client_scopes: List[str]
+    catalog_bearer_token: str
+    catalog_role_arn: str
+    catalog_role_session_name: str
+    catalog_external_id: str
+    catalog_signing_region: str
+    catalog_signing_name: str
 
     def validate(self):
         if self.catalogs_subcommand == Subcommands.CREATE:
-            if not self.storage_type:
-                raise Exception(f'Missing required argument:'
-                                f' {Argument.to_flag_name(Arguments.STORAGE_TYPE)}')
-            if not self.default_base_location:
-                raise Exception(f'Missing required argument:'
-                                f' {Argument.to_flag_name(Arguments.DEFAULT_BASE_LOCATION)}')
+            if self.catalog_type != CatalogType.EXTERNAL.value:
+                if not self.storage_type:
+                    raise Exception(f'Missing required argument:'
+                                    f' {Argument.to_flag_name(Arguments.STORAGE_TYPE)}')
+                if not self.default_base_location:
+                    raise Exception(f'Missing required argument:'
+                                    f' {Argument.to_flag_name(Arguments.DEFAULT_BASE_LOCATION)}')
+            else:
+                if self.catalog_authentication_type == AuthenticationType.OAUTH.value:
+                    if not self.catalog_token_uri or not self.catalog_client_id \
+                            or not self.catalog_client_secret or len(self.catalog_client_scopes) == 0:
+                        raise Exception(f"Authentication type 'OAUTH' requires"
+                                f" {Argument.to_flag_name(Arguments.CATALOG_TOKEN_URI)},"
+                                f" {Argument.to_flag_name(Arguments.CATALOG_CLIENT_ID)},"
+                                f" {Argument.to_flag_name(Arguments.CATALOG_CLIENT_SECRET)},"
+                                f" and at least one {Argument.to_flag_name(Arguments.CATALOG_CLIENT_SCOPE)}.")
+                elif self.catalog_authentication_type == AuthenticationType.BEARER.value:
+                    if not self.catalog_bearer_token:
+                        raise Exception(f"Missing required argument for authentication type 'BEARER':"
+                                f" {Argument.to_flag_name(Arguments.CATALOG_BEARER_TOKEN)}")
+                elif self.catalog_authentication_type == AuthenticationType.SIGV4.value:
+                    if not self.catalog_role_arn or not self.catalog_signing_region:
+                        raise Exception(f"Authentication type 'SIGV4 requires"
+                                f" {Argument.to_flag_name(Arguments.CATALOG_ROLE_ARN)}"
+                                f" and {Argument.to_flag_name(Arguments.CATALOG_SIGNING_REGION)}")
+
+        if self.catalog_service_identity_type == ServiceIdentityType.AWS_IAM.value:
+            if not self.catalog_service_identity_iam_arn:
+                        raise Exception(f"Missing required argument for service identity type 'AWS_IAM':"
+                                f" {Argument.to_flag_name(Arguments.CATALOG_SERVICE_IDENTITY_IAM_ARN)}")
 
         if self.storage_type == StorageType.S3.value:
             if not self.role_arn:
@@ -137,31 +181,94 @@ def _build_storage_config_info(self):
             )
         return config
 
+    def _build_connection_config_info(self):
+        if self.catalog_type != CatalogType.EXTERNAL.value:
+            return None
+
+        auth_params = None
+        if self.catalog_authentication_type == AuthenticationType.OAUTH.value:
+            auth_params = OAuthClientCredentialsParameters(
+                authentication_type=self.catalog_authentication_type.upper(),
+                token_uri=self.catalog_token_uri,
+                client_id=self.catalog_client_id,
+                client_secret=SecretStr(self.catalog_client_secret),
+                scopes=self.catalog_client_scopes
+            )
+        elif self.catalog_authentication_type == AuthenticationType.BEARER.value:
+            auth_params = BearerAuthenticationParameters(
+                authentication_type=self.catalog_authentication_type.upper(),
+                bearer_token=SecretStr(self.catalog_bearer_token)
+            )
+        elif self.catalog_authentication_type == AuthenticationType.SIGV4.value:
+            auth_params = SigV4AuthenticationParameters(
+                authentication_type=self.catalog_authentication_type.upper(),
+                role_arn=self.catalog_role_arn,
+                role_session_name=self.catalog_role_session_name,
+                external_id=self.catalog_external_id,
+                signing_region=self.catalog_signing_region,
+                signing_name=self.catalog_signing_name,
+            )
+        elif self.catalog_authentication_type is not None:
+            raise Exception("Unknown authentication type:", self.catalog_authentication_type)
+
+        service_identity = None
+        if self.catalog_service_identity_type == ServiceIdentityType.AWS_IAM:
+            service_identity = AwsIamServiceIdentityInfo(
+                identity_type=self.catalog_service_identity_type.upper(),
+                iam_arn=self.catalog_service_identity_iam_arn
+            )
+        elif self.catalog_service_identity_type is not None:
+            raise Exception("Unknown service identity type:", self.catalog_service_identity_type)
+
+        config = None
+        if self.catalog_connection_type == CatalogConnectionType.HADOOP.value:
+            config = HadoopConnectionConfigInfo(
+                connection_type=self.catalog_connection_type.upper(),
+                uri=self.catalog_uri,
+                authentication_parameters=auth_params,
+                service_identity=service_identity,
+                warehouse=self.hadoop_warehouse
+            )
+        elif self.catalog_connection_type == CatalogConnectionType.ICEBERG.value:
+            config = IcebergRestConnectionConfigInfo(
+                connection_type=self.catalog_connection_type.upper().replace('-', '_'),
+                uri=self.catalog_uri,
+                authentication_parameters=auth_params,
+                service_identity=service_identity,
+                remote_catalog_name=self.iceberg_remote_catalog_name
+            )
+        elif self.catalog_connection_type is not None:
+            raise Exception("Unknown catalog connection type:", self.catalog_connection_type)
+        return config
+
     def execute(self, api: PolarisDefaultApi) -> None:
         if self.catalogs_subcommand == Subcommands.CREATE:
-            config = self._build_storage_config_info()
+            storage_config = self._build_storage_config_info()
+            connection_config = self._build_connection_config_info()
             if self.catalog_type == CatalogType.EXTERNAL.value:
                 request = CreateCatalogRequest(
                     catalog=ExternalCatalog(
                         type=self.catalog_type.upper(),
                         name=self.catalog_name,
-                        storage_config_info=config,
+                        storage_config_info=storage_config,
                         properties=CatalogProperties(
                             default_base_location=self.default_base_location,
                             additional_properties=self.properties
-                        )
+                        ),
+                        connection_config_info=connection_config
                     )
                 )
             else:
                 request = CreateCatalogRequest(
                     catalog=PolarisCatalog(
                         type=self.catalog_type.upper(),
                         name=self.catalog_name,
-                        storage_config_info=config,
+                        storage_config_info=storage_config,
                         properties=CatalogProperties(
                             default_base_location=self.default_base_location,
                             additional_properties=self.properties
-                        )
+                        ),
+                        connection_config_info=connection_config
                     )
                 )
             api.create_catalog(request)

client/python/cli/constants.py

@@ -48,6 +48,33 @@ class PrincipalType(Enum):
     SERVICE = 'service'
 
 
+class CatalogConnectionType(Enum):
+    """
+    Represents a ConnectionType for an EXTERNAL catalog -- see ConnectionConfigInfo in the spec
+    """
+
+    HADOOP = 'hadoop'
+    ICEBERG = 'iceberg-rest'
+
+
+class AuthenticationType(Enum):
+    """
+    Represents a AuthenticationType for an EXTERNAL catalog -- see AuthenticationParameters in the spec
+    """
+
+    OAUTH = 'oauth'
+    BEARER = 'bearer'
+    SIGV4 = 'sigv4'
+
+
+class ServiceIdentityType(Enum):
+    """
+    Represents a Service Identity Type for an EXTERNAL catalog -- see ServiceIdentityInfo in the spec
+    """
+
+    AWS_IAM = 'aws_iam'
+
+
 class Commands:
     """
     Represents the various commands available in the CLI
@@ -136,6 +163,23 @@ class Arguments:
     REGION = 'region'
     PROFILE = 'profile'
     PROXY = 'proxy'
+    HADOOP_WAREHOUSE = 'hadoop_warehouse'
+    ICEBERG_REMOTE_CATALOG_NAME = 'iceberg_remote_catalog_name'
+    CATALOG_CONNECTION_TYPE = 'catalog_connection_type'
+    CATALOG_AUTHENTICATION_TYPE = 'catalog_authentication_type'
+    CATALOG_SERVICE_IDENTITY_TYPE = 'catalog_service_identity_type'
+    CATALOG_SERVICE_IDENTITY_IAM_ARN = 'catalog_service_identity_iam_arn'
+    CATALOG_URI = 'catalog_uri'
+    CATALOG_TOKEN_URI = 'catalog_token_uri'
+    CATALOG_CLIENT_ID = 'catalog_client_id'
+    CATALOG_CLIENT_SECRET = 'catalog_client_secret'
+    CATALOG_CLIENT_SCOPE = 'catalog_client_scope'
+    CATALOG_BEARER_TOKEN = 'catalog_bearer_token'
+    CATALOG_ROLE_ARN = 'catalog_role_arn'
+    CATALOG_ROLE_SESSION_NAME = 'catalog_role_session_name'
+    CATALOG_EXTERNAL_ID = 'catalog_external_id'
+    CATALOG_SIGNING_REGION = 'catalog_signing_region'
+    CATALOG_SIGNING_NAME = 'catalog_signing_name'
 
 
 class Hints:
@@ -180,6 +224,41 @@ class Create:
         class Update:
             DEFAULT_BASE_LOCATION = 'A new default base location for the catalog'
 
+        class External:
+            CATALOG_CONNECTION_TYPE = 'The type of external catalog in [ICEBERG, HADOOP].'
+            CATALOG_AUTHENTICATION_TYPE = 'The type of authentication in [OAUTH, BEARER, SIGV4]'
+            CATALOG_SERVICE_IDENTITY_TYPE = 'The type of service identity in [AWS_IAM]'
+
+            CATALOG_SERVICE_IDENTITY_IAM_ARN = ('When using the AWS_IAM service identity type, this is the ARN '
+                                                'of the IAM user or IAM role Polaris uses to assume roles and '
+                                                'then access external resources.')
+
+            CATALOG_URI = 'The URI of the external catalog'
+            HADOOP_WAREHOUSE = 'The warehouse to use when federating to a HADOOP catalog'
+            ICEBERG_REMOTE_CATALOG_NAME = 'The remote catalog name when federating to an Iceberg REST catalog'
+
+
+            CATALOG_TOKEN_URI = '(For authentication type OAUTH) Token server URI'
+            CATALOG_CLIENT_ID = '(For authentication type OAUTH) oauth client id'
+            CATALOG_CLIENT_SECRET = '(For authentication type OAUTH) oauth client secret (input-only)'
+            CATALOG_CLIENT_SCOPE = ('(For authentication type OAUTH) oauth scopes to specify when exchanging '
+                                    'for a short-lived access token. Multiple can be provided by specifying'
+                                    ' this option more than once')
+
+            CATALOG_BEARER_TOKEN = '(For authentication type BEARER) Bearer token (input-only)'
+
+            CATALOG_ROLE_ARN = ('(For authentication type SIGV4) The aws IAM role arn assumed by polaris '
+                                'userArn when signing requests')
+            CATALOG_ROLE_SESSION_NAME = ('(For authentication type SIGV4) The role session name to be used '
+                                         'by the SigV4 protocol for signing requests')
+            CATALOG_EXTERNAL_ID = ('(For authentication type SIGV4) An optional external id used to establish '
+                                   'a trust relationship with AWS in the trust policy')
+            CATALOG_SIGNING_REGION = ('(For authentication type SIGV4) Region to be used by the SigV4 protocol '
+                                      'for signing requests')
+            CATALOG_SIGNING_NAME = ('(For authentication type SIGV4) The service name to be used by the SigV4 '
+                                    'protocol for signing requests, the default signing name is "execute-api" '
+                                    'is if not provided')
+
     class Principals:
         class Create:
             TYPE = 'The type of principal to create in [SERVICE]'

client/python/cli/options/option_tree.py

@@ -19,7 +19,8 @@
 from dataclasses import dataclass, field
 from typing import List
 
-from cli.constants import StorageType, CatalogType, PrincipalType, Hints, Commands, Arguments, Subcommands, Actions
+from cli.constants import StorageType, CatalogType, PrincipalType, Hints, Commands, Arguments, Subcommands, Actions, \
+    CatalogConnectionType, AuthenticationType, ServiceIdentityType
 
 
 @dataclass
@@ -74,6 +75,36 @@ class OptionTree:
         Argument(Arguments.CATALOG_ROLE, str, Hints.CatalogRoles.CATALOG_ROLE)
     ]
 
+    _FEDERATION_ARGS = [
+        Argument(Arguments.CATALOG_CONNECTION_TYPE, str,
+                 Hints.Catalogs.External.CATALOG_CONNECTION_TYPE, lower=True,
+                 choices=[ct.value for ct in CatalogConnectionType]),
+        Argument(Arguments.ICEBERG_REMOTE_CATALOG_NAME, str,
+                 Hints.Catalogs.External.ICEBERG_REMOTE_CATALOG_NAME),
+        Argument(Arguments.HADOOP_WAREHOUSE, str,
+                 Hints.Catalogs.External.HADOOP_WAREHOUSE),
+        Argument(Arguments.CATALOG_AUTHENTICATION_TYPE, str,
+                 Hints.Catalogs.External.CATALOG_AUTHENTICATION_TYPE, lower=True,
+                 choices=[at.value for at in AuthenticationType]),
+        Argument(Arguments.CATALOG_SERVICE_IDENTITY_TYPE, str,
+                 Hints.Catalogs.External.CATALOG_SERVICE_IDENTITY_TYPE, lower=True,
+                 choices=[st.value for st in ServiceIdentityType]),
+        Argument(Arguments.CATALOG_SERVICE_IDENTITY_IAM_ARN, str,
+                 Hints.Catalogs.External.CATALOG_SERVICE_IDENTITY_IAM_ARN),
+        Argument(Arguments.CATALOG_URI, str, Hints.Catalogs.External.CATALOG_URI),
+        Argument(Arguments.CATALOG_TOKEN_URI, str, Hints.Catalogs.External.CATALOG_TOKEN_URI),
+        Argument(Arguments.CATALOG_CLIENT_ID, str, Hints.Catalogs.External.CATALOG_CLIENT_ID),
+        Argument(Arguments.CATALOG_CLIENT_SECRET, str, Hints.Catalogs.External.CATALOG_CLIENT_SECRET),
+        Argument(Arguments.CATALOG_CLIENT_SCOPE, str,
+                 Hints.Catalogs.External.CATALOG_CLIENT_SCOPE, allow_repeats=True),
+        Argument(Arguments.CATALOG_BEARER_TOKEN, str, Hints.Catalogs.External.CATALOG_BEARER_TOKEN),
+        Argument(Arguments.CATALOG_ROLE_ARN, str, Hints.Catalogs.External.CATALOG_ROLE_ARN),
+        Argument(Arguments.CATALOG_ROLE_SESSION_NAME, str, Hints.Catalogs.External.CATALOG_ROLE_SESSION_NAME),
+        Argument(Arguments.CATALOG_EXTERNAL_ID, str, Hints.Catalogs.External.CATALOG_EXTERNAL_ID),
+        Argument(Arguments.CATALOG_SIGNING_REGION, str, Hints.Catalogs.External.CATALOG_SIGNING_REGION),
+        Argument(Arguments.CATALOG_SIGNING_NAME, str, Hints.Catalogs.External.CATALOG_SIGNING_NAME, lower=True)
+    ]
+
     @staticmethod
     def get_tree() -> List[Option]:
         return [
@@ -94,7 +125,7 @@ def get_tree() -> List[Option]:
                     Argument(Arguments.CONSENT_URL, str, Hints.Catalogs.Create.CONSENT_URL),
                     Argument(Arguments.SERVICE_ACCOUNT, str, Hints.Catalogs.Create.SERVICE_ACCOUNT),
                     Argument(Arguments.PROPERTY, str, Hints.PROPERTY, allow_repeats=True),
-                ], input_name=Arguments.CATALOG),
+                ] + OptionTree._FEDERATION_ARGS, input_name=Arguments.CATALOG),
                 Option(Subcommands.DELETE, input_name=Arguments.CATALOG),
                 Option(Subcommands.GET, input_name=Arguments.CATALOG),
                 Option(Subcommands.LIST, args=[