Add missing column

Author: Fokko

pyiceberg/table/inspect.py

@@ -404,6 +404,7 @@ def _get_all_manifests_schema(self) -> pa.Schema:
 
         all_manifests_schema = self._get_manifests_schema()
         all_manifests_schema = all_manifests_schema.append(pa.field("reference_snapshot_id", pa.int64(), nullable=False))
+        all_manifests_schema = all_manifests_schema.append(pa.field("key_metadata", pa.binary(), nullable=True))
         return all_manifests_schema
 
     def _generate_manifests_table(self, snapshot: Snapshot | None, is_all_manifests_table: bool = False) -> pa.Table:
@@ -468,6 +469,7 @@ def _partition_summaries_to_rows(
                 }
                 if is_all_manifests_table:
                     manifest_row["reference_snapshot_id"] = snapshot.snapshot_id
+                    manifest_row["key_metadata"] = manifest.key_metadata
                 manifests.append(manifest_row)
 
         return pa.Table.from_pylist(

tests/integration/test_inspect_table.py

@@ -1012,6 +1012,7 @@ def test_inspect_all_manifests(spark: SparkSession, session_catalog: Catalog, fo
         "deleted_delete_files_count",
         "partition_summaries",
         "reference_snapshot_id",
+        "key_metadata",
     ]
 
     int_cols = [