Hive: Use Hive table location in HiveIcebergSplit

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java

@@ -56,6 +56,7 @@ public class HiveIcebergInputFormat extends MapredIcebergInputFormat<Record>
           Configuration.class, ExprNodeGenericFuncDesc.class)
       .orNoop()
       .buildStatic();
+  static final String SPLIT_LOCATION = "iceberg.hive.split.location";
 
   @Override
   public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
@@ -72,7 +73,7 @@ public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
       }
     }
 
-    String location = job.get(InputFormatConfig.TABLE_LOCATION);
+    String location = job.get(SPLIT_LOCATION);
     return Arrays.stream(super.getSplits(job, numSplits))
                  .map(split -> new HiveIcebergSplit((IcebergSplit) split, location))
                  .toArray(InputSplit[]::new);

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergSerDe.java

@@ -27,8 +27,11 @@
 import org.apache.hadoop.hive.serde2.SerDeStats;
 import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
 import org.apache.hadoop.io.Writable;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.mr.Catalogs;
+import org.apache.iceberg.mr.InputFormatConfig;
 import org.apache.iceberg.mr.hive.serde.objectinspector.IcebergObjectInspector;
 import org.apache.iceberg.mr.mapred.Container;
 
@@ -38,10 +41,15 @@ public class HiveIcebergSerDe extends AbstractSerDe {
 
   @Override
   public void initialize(@Nullable Configuration configuration, Properties serDeProperties) throws SerDeException {
-    Table table = Catalogs.loadTable(configuration, serDeProperties);
-
+    Schema tableSchema;
+    if (configuration.get(InputFormatConfig.TABLE_SCHEMA) != null) {
+      tableSchema = SchemaParser.fromJson(configuration.get(InputFormatConfig.TABLE_SCHEMA));
+    } else {
+      Table table = Catalogs.loadTable(configuration, serDeProperties);
+      tableSchema = table.schema();
+    }
     try {
-      this.inspector = IcebergObjectInspector.create(table.schema());
+      this.inspector = IcebergObjectInspector.create(tableSchema);
     } catch (Exception e) {
       throw new SerDeException(e);
     }

mr/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java

@@ -43,6 +43,7 @@
 public class HiveIcebergStorageHandler implements HiveStoragePredicateHandler, HiveStorageHandler {
 
   private static final String NAME = "name";
+  private static final String LOCATION = "location";
 
   private Configuration conf;
 
@@ -79,6 +80,7 @@ public void configureInputJobProperties(TableDesc tableDesc, Map<String, String>
     map.put(InputFormatConfig.TABLE_IDENTIFIER, props.getProperty(NAME));
     map.put(InputFormatConfig.TABLE_LOCATION, table.location());
     map.put(InputFormatConfig.TABLE_SCHEMA, SchemaParser.toJson(table.schema()));
+    map.put(HiveIcebergInputFormat.SPLIT_LOCATION, props.getProperty(LOCATION));
   }
 
   @Override

mr/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStorageHandlerWithHiveCatalogAndLinkedinMetadata.java

@@ -0,0 +1,88 @@
+/*
+ * 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.mr.hive;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.List;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.common.DynMethods;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.hive.HiveCatalogs;
+import org.apache.iceberg.mr.TestHelper;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.rules.TemporaryFolder;
+
+public class TestHiveIcebergStorageHandlerWithHiveCatalogAndLinkedinMetadata
+    extends TestHiveIcebergStorageHandlerWithHiveCatalog {
+
+  private static final FileFormat FILE_FORMAT = FileFormat.PARQUET;
+
+  private HiveCatalog hiveCatalog;
+  private TemporaryFolder temporaryFolder;
+
+  @Override
+  public TestTables testTables(Configuration conf, TemporaryFolder temp) {
+    hiveCatalog = HiveCatalogs.loadCatalog(conf);
+    temporaryFolder = temp;
+    return super.testTables(conf, temp);
+  }
+
+  @Override
+  protected Table createIcebergTable(String tableName, Schema schema, List<Record> records) throws IOException {
+    // This code is derived from TestTables. There was no easy way to alter table location without changing
+    // bunch of interfaces. With this code the same outcome is achieved.
+    TableIdentifier tableIdentifier = TableIdentifier.parse("default." + tableName);
+    Table table = hiveCatalog.createTable(
+        tableIdentifier,
+        schema,
+        PartitionSpec.unpartitioned(),
+        getLocationWithoutURI(tableIdentifier),
+        ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, FILE_FORMAT.name()));
+
+    if (!records.isEmpty()) {
+      table
+          .newAppend()
+          .appendFile(TestHelper.writeFile(table, null, records, FILE_FORMAT, temporaryFolder.newFile()))
+          .commit();
+    }
+    return table;
+  }
+
+  private String getLocationWithoutURI(TableIdentifier tableIdentifier) {
+    try {
+      String location =  DynMethods.builder("defaultWarehouseLocation")
+          .hiddenImpl(HiveCatalog.class, TableIdentifier.class)
+          .build()
+          .invoke(hiveCatalog, tableIdentifier);
+      return new URI(location).getPath();
+    } catch (URISyntaxException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}