From 91d537153c713529a3c640eab65d8a9ccbb4443b Mon Sep 17 00:00:00 2001
From: tcodehuber <tcodehuber@gmail.com>
Date: Tue, 16 Jan 2024 16:47:34 +0800
Subject: [PATCH] [AMORO-2375] Add more detail exception log In
 SchemaUtil.selectInOrder (#2493)

* [AMORO-2375] Add more detail exception log In SchemaUtil.selectInOrder

* [AMORO-2375] Add more detail exception log In SchemaUtil.selectInOrder

* refactor code
---
 .../java/com/netease/arctic/utils/SchemaUtil.java    | 12 ++++++++++++
 1 file changed, 12 insertions(+)

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);
+      }
+    }
+  }
 }