diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java
index a8e82d101fbf..f53859ef97d6 100644
--- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java
+++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java
@@ -161,7 +161,8 @@ protected Dataset<FileInfo> contentFileDS(Table table, Set<Long> snapshotIds) {
                 "length",
                 "0 as sequenceNumber",
                 "partition_spec_id as partitionSpecId",
-                "added_snapshot_id as addedSnapshotId")
+                "added_snapshot_id as addedSnapshotId",
+                "key_metadata as keyMetadata")
             .dropDuplicates("path")
             .repartition(numShufflePartitions) // avoid adaptive execution combining tasks
             .as(ManifestFileBean.ENCODER);
diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/ManifestFileBean.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/ManifestFileBean.java
index fd4639897743..599e27b71c45 100644
--- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/ManifestFileBean.java
+++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/ManifestFileBean.java
@@ -37,6 +37,7 @@ public class ManifestFileBean implements ManifestFile, Serializable {
   private Integer content = null;
   private Long sequenceNumber = null;
   private Long firstRowId = null;
+  private byte[] keyMetadata = null;
 
   public static ManifestFileBean fromManifest(ManifestFile manifest) {
     ManifestFileBean bean = new ManifestFileBean();
@@ -48,6 +49,7 @@ public static ManifestFileBean fromManifest(ManifestFile manifest) {
     bean.setContent(manifest.content().id());
     bean.setSequenceNumber(manifest.sequenceNumber());
     bean.setFirstRowId(manifest.firstRowId());
+    bean.setKeyMetadata(manifest.keyMetadata() == null ? null : manifest.keyMetadata().array());
 
     return bean;
   }
@@ -104,6 +106,14 @@ public void setFirstRowId(Long firstRowId) {
     this.firstRowId = firstRowId;
   }
 
+  public byte[] getKeyMetadata() {
+    return keyMetadata;
+  }
+
+  public void setKeyMetadata(byte[] keyMetadata) {
+    this.keyMetadata = keyMetadata;
+  }
+
   @Override
   public String path() {
     return path;
@@ -176,7 +186,7 @@ public List<PartitionFieldSummary> partitions() {
 
   @Override
   public ByteBuffer keyMetadata() {
-    return null;
+    return keyMetadata == null ? null : ByteBuffer.wrap(keyMetadata);
   }
 
   @Override