diff --git a/core/src/main/java/com/netease/arctic/utils/SchemaUtil.java b/core/src/main/java/com/netease/arctic/utils/SchemaUtil.java
index c14f90344b..73f5dc6e3f 100644
--- a/core/src/main/java/com/netease/arctic/utils/SchemaUtil.java
+++ b/core/src/main/java/com/netease/arctic/utils/SchemaUtil.java
@@ -85,6 +85,7 @@ public static Schema fillUpIdentifierFields(
   public static Schema selectInOrder(Schema baseSchema, List<String> fieldNames) {
     Preconditions.checkNotNull(fieldNames);
     Preconditions.checkNotNull(baseSchema);
+    validateSchemaFields(baseSchema, fieldNames);
 
     int schemaId = baseSchema.schemaId();
     List<Types.NestedField> fields =
@@ -92,4 +93,15 @@ public static Schema selectInOrder(Schema baseSchema, List<String> fieldNames) {
 
     return new Schema(schemaId, fields);
   }
+
+  private static void validateSchemaFields(Schema schema, List<String> requiredFields) {
+    Set<String> existingFields =
+        schema.columns().stream().map(Types.NestedField::name).collect(Collectors.toSet());
+    for (String requiredField : requiredFields) {
+      if (!existingFields.contains(requiredField)) {
+        throw new IllegalArgumentException(
+            "The required field in schema is missing: " + requiredField);
+      }
+    }
+  }
 }