[INLONG-9231][Sort] Find no audit time field when the field is in upp…

…er case (apache#9232)
vernedeng · Nov 8, 2023 · d116f1a · d116f1a
1 parent 33df4f5
commit d116f1a
Showing 1 changed file with 7 additions and 4 deletions.
diff --git a/...nectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/utils/SinkMetadataUtils.java b/...nectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/utils/SinkMetadataUtils.java
@@ -32,6 +32,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Set;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 /**
@@ -49,13 +50,14 @@ public class SinkMetadataUtils implements Serializable {
     public SinkMetadataUtils(List<String> metadataKeys, DataType consumedDataType) {
         Set<String> metadataKeySet = ImmutableSet.copyOf(metadataKeys);
         List<RowType.RowField> metaFields = ((RowType) consumedDataType.getLogicalType()).getFields();
-
+        List<String> names = metaFields.stream().map(RowType.RowField::getName).collect(Collectors.toList());
+        log.info("start to config SinkMetadataUtils, metaKeys={}, consume fields={}", metadataKeys, names);
         // get related converters by real keys
         // the pos of physical column will be replaced by IcebergWritableMetadata.NULL
         this.converters = metaFields.stream()
                 .map(RowType.RowField::getName)
                 .map(key -> Stream.of(IcebergWritableMetadata.values())
-                        .filter(m -> m.getKey().equals(key))
+                        .filter(m -> m.getKey().equalsIgnoreCase(key))
                         .findFirst()
                         .orElse(IcebergWritableMetadata.NULL))
                 .map(IcebergWritableMetadata::getConverter)
@@ -65,15 +67,16 @@ public SinkMetadataUtils(List<String> metadataKeys, DataType consumedDataType) {
         ImmutableBiMap.Builder<String, Integer> builder = ImmutableBiMap.builder();
         for (int i = 0; i < metaFields.size(); i++) {
             String name = metaFields.get(i).getName();
-            if (metadataKeySet.contains(name)) {
-                builder.put(name, i);
+            if (metadataKeySet.contains(name.toLowerCase())) {
+                builder.put(name.toLowerCase(), i);
             }
         }
         this.field2posMap = builder.build();
         this.pos2Field = field2posMap.inverse();
 
         // for audit time
         DATA_TIME_INDEX = field2posMap.getOrDefault(Constants.META_AUDIT_DATA_TIME, -1);
+        log.info("find data time index={}, filed2posMap={}", DATA_TIME_INDEX, field2posMap);
     }
 
     public Integer getMetadataPosByName(String name) {