Hive Metadata Scan: Support case insensitive name mapping

core/src/main/java/org/apache/iceberg/TableMetadata.java

@@ -101,6 +101,17 @@ static TableMetadata newTableMetadata(Schema schema,
         ImmutableList.of(), ImmutableList.of());
   }
 
+  public static TableMetadata newTableMetadataWithoutFreshIds(Schema schema,
+      PartitionSpec spec,
+      String location,
+      Map<String, String> properties) {
+    return new TableMetadata(null, DEFAULT_TABLE_FORMAT_VERSION, UUID.randomUUID().toString(), location,
+        INITIAL_SEQUENCE_NUMBER, System.currentTimeMillis(),
+        -1, schema, INITIAL_SPEC_ID, ImmutableList.of(spec),
+        ImmutableMap.copyOf(properties), -1, ImmutableList.of(),
+        ImmutableList.of(), ImmutableList.of());
+  }
+
   public static class SnapshotLogEntry implements HistoryEntry {
     private final long timestampMillis;
     private final long snapshotId;

core/src/main/java/org/apache/iceberg/mapping/MappingUtil.java

@@ -54,6 +54,19 @@ public static NameMapping create(Schema schema) {
     return new NameMapping(TypeUtil.visit(schema, CreateMapping.INSTANCE));
   }
 
+  /**
+   * Create a name-based mapping for a schema.
+   * <p>
+   * The mapping returned by this method will use the schema's name for each field.
+   *
+   * @param schema a {@link Schema}
+   * @param caseSensitive whether names should be matched case sensitively
+   * @return a {@link NameMapping} initialized with the schema's fields and names
+   */
+  public static NameMapping create(Schema schema, boolean caseSensitive) {
+    return new NameMapping(TypeUtil.visit(schema, CreateMapping.INSTANCE), caseSensitive);
+  }
+
   /**
    * Update a name-based mapping using changes to a schema.
    *

core/src/main/java/org/apache/iceberg/mapping/NameMapping.java

@@ -22,6 +22,7 @@
 import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 
@@ -44,25 +45,47 @@ public static NameMapping of(MappedFields fields) {
   }
 
   private final MappedFields mapping;
+  private final boolean caseSensitive;
   private transient Map<Integer, MappedField> fieldsById;
   private transient Map<String, MappedField> fieldsByName;
+  private transient Map<String, MappedField> fieldsByNameLowercase;
 
   NameMapping(MappedFields mapping) {
+    this(mapping, true);
+  }
+
+  NameMapping(MappedFields mapping, boolean caseSensitive) {
     this.mapping = mapping;
+    this.caseSensitive = caseSensitive;
     lazyFieldsById();
     lazyFieldsByName();
+    if (!caseSensitive) {
+      lazyFieldsByNameLowercase();
+    }
   }
 
   public MappedField find(int id) {
     return lazyFieldsById().get(id);
   }
 
   public MappedField find(String... names) {
-    return lazyFieldsByName().get(DOT.join(names));
+    return find(DOT.join(names));
   }
 
   public MappedField find(List<String> names) {
-    return lazyFieldsByName().get(DOT.join(names));
+    return find(DOT.join(names));
+  }
+
+  private MappedField find(String qualifiedName) {
+    MappedField field = lazyFieldsByName().get(qualifiedName);
+    if (field == null && !caseSensitive) {
+      field = lazyFieldsByNameLowercase().get(qualifiedName.toLowerCase());
+    }
+    return field;
+  }
+
+  public boolean isCaseSensitive() {
+    return caseSensitive;
   }
 
   public MappedFields asMappedFields() {
@@ -83,6 +106,14 @@ private Map<String, MappedField> lazyFieldsByName() {
     return fieldsByName;
   }
 
+  private Map<String, MappedField> lazyFieldsByNameLowercase() {
+    if (fieldsByNameLowercase == null) {
+      this.fieldsByNameLowercase = lazyFieldsByName().entrySet().stream()
+          .collect(Collectors.toMap(x -> x.getKey().toLowerCase(), Map.Entry::getValue, (u, v) -> u));
+    }
+    return fieldsByNameLowercase;
+  }
+
   @Override
   public String toString() {
     if (mapping.fields().isEmpty()) {

core/src/main/java/org/apache/iceberg/mapping/NameMappingParser.java

@@ -41,6 +41,17 @@
  *       { "field-id": 5, "names": ["longitude", "long"] }
  *     ] } ]
  * </pre>
+ * or
+ * <pre>
+ * { "case-sensitive": false,
+ *   "mapping": [ { "field-id": 1, "names": ["id", "record_id"] },
+ *       { "field-id": 2, "names": ["data"] },
+ *       { "field-id": 3, "names": ["location"], "fields": [
+ *           { "field-id": 4, "names": ["latitude", "lat"] },
+ *           { "field-id": 5, "names": ["longitude", "long"] }
+ *       ] } ]
+ * }
+ * </pre>
  */
 public class NameMappingParser {
 
@@ -50,6 +61,8 @@ private NameMappingParser() {
   private static final String FIELD_ID = "field-id";
   private static final String NAMES = "names";
   private static final String FIELDS = "fields";
+  private static final String CASE_SENSITIVE = "case-sensitive";
+  private static final String MAPPING = "mapping";
 
   public static String toJson(NameMapping mapping) {
     try {
@@ -65,7 +78,11 @@ public static String toJson(NameMapping mapping) {
   }
 
   static void toJson(NameMapping nameMapping, JsonGenerator generator) throws IOException {
+    generator.writeStartObject();
+    generator.writeBooleanField(CASE_SENSITIVE, nameMapping.isCaseSensitive());
+    generator.writeFieldName(MAPPING);
     toJson(nameMapping.asMappedFields(), generator);
+    generator.writeEndObject();
   }
 
   private static void toJson(MappedFields mapping, JsonGenerator generator) throws IOException {
@@ -107,7 +124,14 @@ public static NameMapping fromJson(String json) {
   }
 
   static NameMapping fromJson(JsonNode node) {
-    return new NameMapping(fieldsFromJson(node));
+    Preconditions.checkArgument(node.isObject() || node.isArray(),
+        "Cannot parse non-object or non-array name mapping: %s", node);
+    if (node.isObject()) {
+      boolean caseSensitive = JsonUtil.getBool(CASE_SENSITIVE, node);
+      return new NameMapping(fieldsFromJson(node.get(MAPPING)), caseSensitive);
+    } else {
+      return new NameMapping(fieldsFromJson(node));
+    }
   }
 
   private static MappedFields fieldsFromJson(JsonNode node) {