diff --git a/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
index 033b35985046..70adf5ec6277 100644
--- a/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
+++ b/mr/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java
@@ -23,6 +23,7 @@
 import java.io.UncheckedIOException;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.InputFormat;
@@ -45,6 +46,7 @@
 import org.apache.iceberg.TableScan;
 import org.apache.iceberg.data.DeleteFilter;
 import org.apache.iceberg.data.GenericDeleteFilter;
+import org.apache.iceberg.data.IdentityPartitionConverters;
 import org.apache.iceberg.data.InternalRecordWrapper;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.encryption.EncryptedFiles;
@@ -63,6 +65,7 @@
 import org.apache.iceberg.mr.Catalogs;
 import org.apache.iceberg.mr.InputFormatConfig;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.PartitionUtil;
 import org.apache.iceberg.util.SerializationUtil;
 import org.apache.iceberg.util.ThreadPools;
 
@@ -313,6 +316,9 @@ private CloseableIterable<T> openTask(FileScanTask currentTask, Schema readSchem
           encryptionManager.decrypt(
               EncryptedFiles.encryptedInput(io.newInputFile(file.location()), file.keyMetadata()));
 
+      Map<Integer, ?> partition =
+          PartitionUtil.constantsMap(currentTask, IdentityPartitionConverters::convertConstant);
+
       ReadBuilder<Record, ?> readBuilder =
           FormatModelRegistry.readBuilder(file.format(), Record.class, inputFile);
 
@@ -328,6 +334,7 @@ private CloseableIterable<T> openTask(FileScanTask currentTask, Schema readSchem
           (CloseableIterable<T>)
               readBuilder
                   .project(readSchema)
+                  .idToConstant(partition)
                   .split(currentTask.start(), currentTask.length())
                   .caseSensitive(caseSensitive)
                   .filter(currentTask.residual())
diff --git a/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java b/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java
index ce588a7e83e2..7be03cd7bc12 100644
--- a/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java
+++ b/mr/src/test/java/org/apache/iceberg/mr/TestIcebergInputFormats.java
@@ -31,6 +31,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.function.Function;
 import org.apache.hadoop.conf.Configuration;
@@ -47,19 +48,26 @@
 import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Files;
 import org.apache.iceberg.Parameter;
 import org.apache.iceberg.ParameterizedTestExtension;
 import org.apache.iceberg.Parameters;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TestHelpers.Row;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.FileHelpers;
+import org.apache.iceberg.data.GenericFileWriterFactory;
+import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.hadoop.HadoopCatalog;
 import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.DataWriter;
+import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.mr.mapred.Container;
 import org.apache.iceberg.mr.mapred.MapredIcebergInputFormat;
 import org.apache.iceberg.mr.mapreduce.IcebergInputFormat;
@@ -298,6 +306,49 @@ private void validateIdentityPartitionProjections(
     }
   }
 
+  @TestTemplate
+  void identityPartitionValuesMissingFromDataFile() throws Exception {
+    Table table = helper.createTable(LOG_SCHEMA, IDENTITY_PARTITION_SPEC);
+
+    // Tables imported with add_files/migrate have data files that do not physically store the
+    // identity partition columns; their values only exist in the manifest entry. See
+    // TableMigrationUtil#listPartition, which reads the partition values from Hive metadata.
+    Schema fileSchema = withColumns("id", "message");
+    Record record = GenericRecord.create(fileSchema);
+    record.setField("id", 1);
+    record.setField("message", "hello");
+
+    DataFile dataFile =
+        writeFileWithSchema(table, fileSchema, Row.of("2020-03-20", "info"), record);
+    table.newAppend().appendFile(dataFile).commit();
+
+    builder.project(LOG_SCHEMA);
+    List<Record> records = testInputFormat.create(builder.conf()).getRecords();
+
+    assertThat(records).hasSize(1);
+    assertThat(records.get(0).getField("date")).isEqualTo("2020-03-20");
+    assertThat(records.get(0).getField("level")).isEqualTo("info");
+    assertThat(records.get(0).getField("id")).isEqualTo(1);
+    assertThat(records.get(0).getField("message")).isEqualTo("hello");
+  }
+
+  private DataFile writeFileWithSchema(
+      Table table, Schema fileSchema, StructLike partition, Record record) throws IOException {
+    OutputFile outputFile =
+        Files.localOutput(temp.resolve("partial-schema-" + UUID.randomUUID()).toFile());
+    DataWriter<Record> writer =
+        new GenericFileWriterFactory.Builder(table)
+            .dataFileFormat(fileFormat)
+            .dataSchema(fileSchema)
+            .build()
+            .newDataWriter(FileHelpers.encrypt(outputFile), table.spec(), partition);
+    try (writer) {
+      writer.write(record);
+    }
+
+    return writer.toDataFile();
+  }
+
   @TestTemplate
   public void testSnapshotReads() throws Exception {
     helper.createUnpartitionedTable();