diff --git a/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDatasetBulkInsertHelper.java b/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDatasetBulkInsertHelper.java
index 0ccd33d7d..47e97aaeb 100644
--- a/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDatasetBulkInsertHelper.java
+++ b/hudi-spark-datasource/hudi-spark/src/main/java/org/apache/hudi/HoodieDatasetBulkInsertHelper.java
@@ -145,10 +145,14 @@ public class HoodieDatasetBulkInsertHelper {
             functions.lit("").cast(DataTypes.StringType));
 
     List<Column> originalFields =
-        Arrays.stream(rowsWithMetaCols.schema().fields()).filter(field -> !field.name().contains("_hoodie_")).map(f -> new Column(f.name())).collect(Collectors.toList());
+        Arrays.stream(rowsWithMetaCols.schema().fields())
+            .filter(field -> !HoodieRecord.HOODIE_META_COLUMNS_WITH_OPERATION.contains(field.name()))
+            .map(f -> new Column(f.name())).collect(Collectors.toList());
 
     List<Column> metaFields =
-        Arrays.stream(rowsWithMetaCols.schema().fields()).filter(field -> field.name().contains("_hoodie_")).map(f -> new Column(f.name())).collect(Collectors.toList());
+        Arrays.stream(rowsWithMetaCols.schema().fields())
+            .filter(field -> HoodieRecord.HOODIE_META_COLUMNS_WITH_OPERATION.contains(field.name()))
+            .map(f -> new Column(f.name())).collect(Collectors.toList());
 
     // reorder such that all meta columns are at the beginning followed by original columns
     List<Column> allCols = new ArrayList<>();