Fixes 16595: Parse Iceberg REST table FQN from identifier (#16596)

ingestion/src/metadata/ingestion/source/database/iceberg/catalog/rest.py

@@ -37,7 +37,7 @@ class IcebergRestCatalog(IcebergCatalogBase):
                 "'connection' is not an instance of 'RestCatalogConnection'"
             )
 
-        if catalog.connection.credential:
+        if catalog.connection.credential and catalog.connection.credential.clientSecret:
             client_id = catalog.connection.credential.clientId.get_secret_value()
             client_secret = (
                 catalog.connection.credential.clientSecret.get_secret_value()

ingestion/src/metadata/ingestion/source/database/iceberg/helper.py

@@ -37,14 +37,14 @@ def namespace_to_str(namespace: tuple[str]) -> str:
     return ".".join(namespace)
 
 
-def get_table_name_as_str(table: pyiceberg.table.Table) -> str:
-    """Returns the Table Name as Tring from a PyIceberg Table.
+def get_table_name_as_str(table: Tuple[str]) -> str:
+    """Returns the Table Name as String from a PyIceberg table identifier tuple.
 
     The PyIceberg table name is returned as tuple and we turn them into a String
     concatenating the items with a '.' in between.
     """
     # We are skipping the first item because it is the schema name.
-    return ".".join(table.name()[1:])
+    return ".".join(table[1:])
 
 
 def get_column_from_partition(
@@ -94,7 +94,7 @@ def get_column_partition_type(
 def get_owner_from_table(
     table: pyiceberg.table.Table, property_key: str
 ) -> Optional[str]:
-    """Retrives the owner information from given Table Property."""
+    """Retrieves the owner information from given Table Property."""
     return table.properties.get(property_key)
 
 

ingestion/src/metadata/ingestion/source/database/iceberg/metadata.py

@@ -181,7 +181,8 @@ class IcebergSource(DatabaseServiceSource):
         for table_identifier in self.iceberg.list_tables(namespace):
             try:
                 table = self.iceberg.load_table(table_identifier)
-                table_name = get_table_name_as_str(table)
+                # extract table name from table identifier, which does not include catalog name
+                table_name = get_table_name_as_str(table_identifier)
                 table_fqn = fqn.build(
                     self.metadata,
                     entity_type=Table,

openmetadata-spec/src/main/resources/json/schema/entity/services/connections/database/iceberg/restCatalogConnection.json

@@ -30,11 +30,7 @@
           "format": "password"
         }
       },
-      "additionalProperties": false,
-      "required": [
-        "clientId",
-        "clientSecret"
-      ]
+      "additionalProperties": false
     },
     "token": {
       "title": "Token",