Move 'Hive Metadata Scan: Support case insensitive name mapping' (PR 52) to hivelink-core (linkedin#102)

Author: ljfgem

build.gradle

@@ -528,6 +528,9 @@ project(':iceberg-hivelink') {
       exclude group: 'com.zaxxer', module: 'HikariCP'
     }
 
+    testCompile project(':iceberg-data')
+    testCompile project(':iceberg-orc')
+
     // By default, hive-exec is a fat/uber jar and it exports a guava library
     // that's really old. We use the core classifier to be able to override our guava
     // version. Luckily, hive-exec seems to work okay so far with this version of guava

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

@@ -130,18 +130,6 @@ static TableMetadata newTableMetadata(Schema schema,
         .build();
   }
 
-  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),
-        SortOrder.unsorted().orderId(), ImmutableList.of(SortOrder.unsorted()),
-        ImmutableMap.copyOf(properties), -1, ImmutableList.of(),
-        ImmutableList.of(), ImmutableList.of());
-  }
-
   public static class SnapshotLogEntry implements HistoryEntry {
     private final long timestampMillis;
     private final long snapshotId;
@@ -257,7 +245,7 @@ public String toString() {
   private final List<MetadataUpdate> changes;
 
   @SuppressWarnings("checkstyle:CyclomaticComplexity")
-  TableMetadata(String metadataFileLocation,
+  TableMetadata(InputFile file,
                 int formatVersion,
                 String uuid,
                 String location,

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

@@ -54,19 +54,6 @@ 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.
    *
@@ -259,8 +246,8 @@ private static <S, T> S visit(MappedFields mapping, Visitor<S, T> visitor) {
     return visitor.fields(mapping, fieldResults);
   }
 
-  private static class CreateMapping extends TypeUtil.SchemaVisitor<MappedFields> {
-    private static final CreateMapping INSTANCE = new CreateMapping();
+  public static class CreateMapping extends TypeUtil.SchemaVisitor<MappedFields> {
+    public static final CreateMapping INSTANCE = new CreateMapping();
 
     private CreateMapping() {
     }

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

@@ -54,7 +54,7 @@ public static NameMapping of(MappedFields fields) {
     this(mapping, true);
   }
 
-  NameMapping(MappedFields mapping, boolean caseSensitive) {
+  public NameMapping(MappedFields mapping, boolean caseSensitive) {
     this.mapping = mapping;
     this.caseSensitive = caseSensitive;
     lazyFieldsById();

hivelink-core/src/main/java/org/apache/iceberg/hivelink/core/LegacyHiveTableOperations.java

@@ -28,6 +28,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.UUID;
 import java.util.stream.Collectors;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
@@ -40,6 +41,7 @@
 import org.apache.iceberg.Metrics;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortOrder;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableProperties;
@@ -51,12 +53,13 @@
 import org.apache.iceberg.hadoop.HadoopFileIO;
 import org.apache.iceberg.hive.HiveClientPool;
 import org.apache.iceberg.hivelink.core.utils.FileSystemUtils;
+import org.apache.iceberg.hivelink.core.utils.MappingUtil;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.LocationProvider;
-import org.apache.iceberg.mapping.MappingUtil;
 import org.apache.iceberg.mapping.NameMappingParser;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.types.Types;
@@ -68,6 +71,9 @@
 public class LegacyHiveTableOperations extends BaseMetastoreTableOperations {
 
   private static final Logger LOG = LoggerFactory.getLogger(LegacyHiveTableOperations.class);
+  private static final long INITIAL_SEQUENCE_NUMBER = 0;
+  private static final int DEFAULT_TABLE_FORMAT_VERSION = 1;
+  private static final int INITIAL_SPEC_ID = 0;
 
   private final HiveClientPool metaClients;
   private final String databaseName;
@@ -105,7 +111,7 @@ protected void doRefresh() {
       // Provide a case insensitive name mapping for Hive tables
       tableProperties.put(TableProperties.DEFAULT_NAME_MAPPING,
           NameMappingParser.toJson(MappingUtil.create(schema, false)));
-      TableMetadata metadata = TableMetadata.newTableMetadataWithoutFreshIds(schema, spec,
+      TableMetadata metadata = newTableMetadataWithoutFreshIds(schema, spec,
           hiveTable.getSd().getLocation(), tableProperties);
       setCurrentMetadata(metadata);
     } catch (TException e) {
@@ -279,4 +285,16 @@ public String metadataFileLocation(String filename) {
   public LocationProvider locationProvider() {
     throw new UnsupportedOperationException("Writes not supported for Hive tables without Iceberg metadata");
   }
+
+  private 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),
+            SortOrder.unsorted().orderId(), ImmutableList.of(SortOrder.unsorted()),
+            ImmutableMap.copyOf(properties), -1, ImmutableList.of(),
+            ImmutableList.of(), ImmutableList.of());
+  }
 }

hivelink-core/src/main/java/org/apache/iceberg/hivelink/core/schema/MergeHiveSchemaWithAvro.java

@@ -44,7 +44,7 @@
  * 2. Copies field names, nullability, default value, field props from the Avro schema
  * 3. Copies field type from the Hive schema.
  *    TODO: We should also handle some cases of type promotion where the types in Avro are potentially more correct
- *    e.g.BINARY in Hive -> FIXED in Avro, STRING in Hive -> ENUM in Avro, etc
+ *    e.g.BINARY in Hive to FIXED in Avro, STRING in Hive to ENUM in Avro, etc
  * 4. Retains fields found only in the Hive schema; Ignores fields found only in the Avro schema
  * 5. Fields found only in Hive schema are represented as optional fields in the resultant Avro schema
  * 6. For fields found only in Hive schema, field names are sanitized to make them compatible with Avro identifier spec

hivelink-core/src/main/java/org/apache/iceberg/hivelink/core/utils/MappingUtil.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.hivelink.core.utils;
+
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.mapping.MappedFields;
+import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.types.TypeUtil;
+
+public class MappingUtil {
+
+  private MappingUtil() {
+  }
+
+  /**
+   * 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) {
+    final MappedFields mappedFields = TypeUtil.visit(schema,
+        org.apache.iceberg.mapping.MappingUtil.CreateMapping.INSTANCE);
+    return new NameMapping(mappedFields, caseSensitive);
+  }
+}

data/src/test/java/org/apache/iceberg/data/TestReadFileWithCaseMismatch.java renamed to hivelink-core/src/test/java/org/apache/iceberg/hivelink/core/TestReadFileWithCaseMismatch.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.iceberg.data;
+package org.apache.iceberg.hivelink.core;
 
 import java.io.Closeable;
 import java.io.File;
@@ -35,10 +35,11 @@
 import org.apache.iceberg.Files;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.Record;
 import org.apache.iceberg.data.avro.DataReader;
 import org.apache.iceberg.data.orc.GenericOrcReader;
+import org.apache.iceberg.hivelink.core.utils.MappingUtil;
 import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.mapping.MappingUtil;
 import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.orc.ORC;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;