Glue: GetTable/GetTables on a resource link read through to the target table

moto/glue/models.py

@@ -312,6 +312,7 @@ def create_table(
             table_input,
             catalog_id=self.account_id,
             open_table_format_input=open_table_format_input,
+            backend=self,
         )
         database.tables[table_name] = table
         return table
@@ -1706,17 +1707,29 @@ def as_dict(self) -> dict[str, Any]:
 
 
 class FakeTable(BaseModel):
+    # Fields that AWS Glue reads through from the target table when a table
+    # is a resource link (i.e. its definition contains a TargetTable).
+    # See https://docs.aws.amazon.com/lake-formation/latest/dg/resource-links-glue-apis.html
+    READ_THROUGH_FIELDS = (
+        "StorageDescriptor",
+        "PartitionKeys",
+        "TableType",
+        "Parameters",
+    )
+
     def __init__(
         self,
         database_name: str,
         table_name: str,
         table_input: dict[str, Any],
         catalog_id: str,
         open_table_format_input: dict[str, Any] | None = None,
+        backend: "GlueBackend | None" = None,
     ):
         self.database_name = database_name
         self.name = table_name
         self.catalog_id = catalog_id
+        self.backend = backend
         self.partitions: dict[str, FakePartition] = OrderedDict()
         self.created_time = utcnow()
         self.updated_time: datetime | None = None
@@ -1749,19 +1762,49 @@ def delete_version(self, version_id: str) -> None:
 
     def as_dict(self, version: str | None = None) -> dict[str, Any]:
         version = version or self._current_version  # type: ignore
+        version_data = self.get_version(str(version))
         obj = {
             "DatabaseName": self.database_name,
             "Name": self.name,
             "CreateTime": self.created_time,
-            **self.get_version(str(version)),
+            **version_data,
             # Add VersionId after we get the version-details, just to make sure that it's a valid version (int)
             "VersionId": str(version),
             "CatalogId": self.catalog_id,
         }
         if self.updated_time is not None:
             obj["UpdateTime"] = self.updated_time
+
+        target_table = version_data.get("TargetTable")
+        if target_table and self.backend is not None:
+            self._merge_target_table(obj, target_table)
         return obj
 
+    def _merge_target_table(
+        self, obj: dict[str, Any], target_table: dict[str, Any]
+    ) -> None:
+        """
+        If this table is a resource link (its definition contains a
+        TargetTable), read through to the target table and merge its
+        schema/storage fields into the response - mirroring AWS Glue, which
+        issues a second GetTable-call against the target table and merges
+        the result. The link's own Name/DatabaseName/TargetTable are kept.
+
+        If the target table does not exist, the link is returned as-is.
+        """
+        target_database_name = target_table.get("DatabaseName")
+        target_name = target_table.get("Name")
+        if not target_database_name or not target_name:
+            return
+        try:
+            target = self.backend.get_table(target_database_name, target_name)  # type: ignore[union-attr]
+        except (DatabaseNotFoundException, TableNotFoundException):
+            return
+        target_version_data = target.get_version(str(target._current_version))
+        for field in self.READ_THROUGH_FIELDS:
+            if field in target_version_data:
+                obj[field] = target_version_data[field]
+
     def create_partition(self, partiton_input: dict[str, Any]) -> None:
         partition = FakePartition(self.database_name, self.name, partiton_input)
         key = str(partition.values)

tests/test_glue/test_datacatalog.py

@@ -320,6 +320,104 @@ def test_get_tables_expression():
     assert len(tables_star_expression6) == 2
 
 
+@mock_aws
+def test_get_table_resource_link_reads_through_target_table():
+    client = boto3.client("glue", region_name="us-east-1")
+    database_name = "myspecialdatabase"
+    helpers.create_database(client, database_name)
+
+    target_table_name = "targettable"
+    columns = [{"Name": "country", "Type": "string"}]
+    target_table_input = helpers.create_table_input(
+        database_name, target_table_name, columns=columns
+    )
+    helpers.create_table(client, database_name, target_table_name, target_table_input)
+
+    link_table_name = "resourcelinktable"
+    link_table_input = {
+        "Name": link_table_name,
+        "TargetTable": {
+            "CatalogId": ACCOUNT_ID,
+            "DatabaseName": database_name,
+            "Name": target_table_name,
+        },
+    }
+    helpers.create_table(client, database_name, link_table_name, link_table_input)
+
+    response = helpers.get_table(client, database_name, link_table_name)
+    table = response["Table"]
+
+    # The link keeps its own identity ...
+    assert table["Name"] == link_table_name
+    assert table["DatabaseName"] == database_name
+    assert table["TargetTable"] == link_table_input["TargetTable"]
+    # ... but reads through the schema/storage of the target table
+    assert table["StorageDescriptor"] == target_table_input["StorageDescriptor"]
+    assert table["PartitionKeys"] == target_table_input["PartitionKeys"]
+    assert table["TableType"] == target_table_input["TableType"]
+
+
+@mock_aws
+def test_get_tables_resource_link_reads_through_target_table():
+    client = boto3.client("glue", region_name="us-east-1")
+    database_name = "myspecialdatabase"
+    helpers.create_database(client, database_name)
+
+    target_table_name = "targettable"
+    columns = [{"Name": "country", "Type": "string"}]
+    target_table_input = helpers.create_table_input(
+        database_name, target_table_name, columns=columns
+    )
+    helpers.create_table(client, database_name, target_table_name, target_table_input)
+
+    link_table_name = "resourcelinktable"
+    link_table_input = {
+        "Name": link_table_name,
+        "TargetTable": {
+            "CatalogId": ACCOUNT_ID,
+            "DatabaseName": database_name,
+            "Name": target_table_name,
+        },
+    }
+    helpers.create_table(client, database_name, link_table_name, link_table_input)
+
+    response = helpers.get_tables(client, database_name)
+    tables = {table["Name"]: table for table in response["TableList"]}
+
+    assert len(tables) == 2
+    link_table = tables[link_table_name]
+    assert link_table["TargetTable"] == link_table_input["TargetTable"]
+    assert link_table["StorageDescriptor"] == target_table_input["StorageDescriptor"]
+    assert link_table["PartitionKeys"] == target_table_input["PartitionKeys"]
+
+
+@mock_aws
+def test_get_table_resource_link_missing_target_table():
+    client = boto3.client("glue", region_name="us-east-1")
+    database_name = "myspecialdatabase"
+    helpers.create_database(client, database_name)
+
+    link_table_name = "resourcelinktable"
+    link_table_input = {
+        "Name": link_table_name,
+        "TargetTable": {
+            "CatalogId": ACCOUNT_ID,
+            "DatabaseName": database_name,
+            "Name": "doesnotexist",
+        },
+    }
+    helpers.create_table(client, database_name, link_table_name, link_table_input)
+
+    response = helpers.get_table(client, database_name, link_table_name)
+    table = response["Table"]
+
+    # No target to read through from - the link is returned as-is
+    assert table["Name"] == link_table_name
+    assert table["TargetTable"] == link_table_input["TargetTable"]
+    assert "StorageDescriptor" not in table
+    assert "PartitionKeys" not in table
+
+
 @mock_aws
 def test_get_table_versions():
     client = boto3.client("glue", region_name="us-east-1")