diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java
index 22fa31539d887..a9ff1f85478cb 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/io/HoodieCreateHandle.java
@@ -140,6 +140,8 @@ public void write(HoodieRecord record, Option<IndexedRecord> avroRecord) {
         // Convert GenericRecord to GenericRecord with hoodie commit metadata in schema
         IndexedRecord recordWithMetadataInSchema = rewriteRecord((GenericRecord) avroRecord.get());
         if (preserveHoodieMetadata) {
+          // do not preserve FILENAME_METADATA_FIELD
+          recordWithMetadataInSchema.put(HoodieRecord.HOODIE_META_COLUMNS_NAME_TO_POS.get(HoodieRecord.FILENAME_METADATA_FIELD), path.getName());
           fileWriter.writeAvro(record.getRecordKey(), recordWithMetadataInSchema);
         } else {
           fileWriter.writeAvroWithMetadata(recordWithMetadataInSchema, record);
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
index 566db224e61ed..aa3ead4e2b296 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieClientOnCopyOnWriteStorage.java
@@ -1836,9 +1836,12 @@ private void verifyRecordsWrittenWithPreservedMetadata(Set<String> commitTimes,
     Map<String, List<GenericRecord>> recordsByCommitTime = records.stream()
         .collect(Collectors.groupingBy(r -> r.get(HoodieRecord.COMMIT_TIME_METADATA_FIELD).toString()));
     assertTrue(commitTimes.containsAll(recordsByCommitTime.keySet()));
+    Set<String> expectedFileIds = allStatus.stream().map(WriteStatus::getFileId).collect(Collectors.toSet());
     for (GenericRecord record : records) {
       String recordKey = record.get(HoodieRecord.RECORD_KEY_METADATA_FIELD).toString();
       assertTrue(expectedKeys.contains(recordKey));
+      String fileName = record.get(HoodieRecord.FILENAME_METADATA_FIELD).toString();
+      assertTrue(expectedFileIds.contains(FSUtils.getFileId(fileName)));
     }
   }