Review changes 6.

Author: Laszlo Pinter

mr/src/main/java/org/apache/iceberg/mr/Catalogs.java

@@ -202,14 +202,13 @@ static Optional<Catalog> loadCatalog(Configuration conf, String catalogName) {
     if (catalogType == null) {
       throw new NoSuchNamespaceException("Catalog definition for %s is not found.", catalogName);
     }
-    String name = catalogName == null ? ICEBERG_DEFAULT_CATALOG_NAME : catalogName;
 
-    switch (catalogType.toLowerCase()) {
-      case NO_CATALOG_TYPE:
-        return Optional.empty();
-      default:
-        return Optional.of(CatalogUtil.buildIcebergCatalog(name,
-                getCatalogProperties(conf, name, catalogType), conf));
+    if (NO_CATALOG_TYPE.equalsIgnoreCase(catalogType)) {
+      return Optional.empty();
+    } else {
+      String name = catalogName == null ? ICEBERG_DEFAULT_CATALOG_NAME : catalogName;
+      return Optional.of(CatalogUtil.buildIcebergCatalog(name,
+              getCatalogProperties(conf, name, catalogType), conf));
     }
   }
 
@@ -267,18 +266,18 @@ private static Map<String, String> addCatalogPropertiesIfMissing(Configuration c
    * @return type of the catalog, can be null
    */
   private static String getCatalogType(Configuration conf, String catalogName) {
-    if (catalogName == null) {
-      String catalogType = conf.get(InputFormatConfig.CATALOG);
-      if (catalogType == null) {
-        return HIVE_CATALOG_TYPE;
-      } else if (catalogType.equals(LOCATION)) {
+    if (catalogName != null) {
+      String catalogType = conf.get(String.format(InputFormatConfig.CATALOG_TYPE_TEMPLATE, catalogName));
+      if (catalogName.equals(ICEBERG_HADOOP_TABLE_NAME) || catalogType == null) {
         return NO_CATALOG_TYPE;
       } else {
-        return conf.get(InputFormatConfig.CATALOG);
+        return catalogType;
       }
     } else {
-      String catalogType = conf.get(String.format(InputFormatConfig.CATALOG_TYPE_TEMPLATE, catalogName));
-      if (catalogName.equals(ICEBERG_HADOOP_TABLE_NAME) || catalogType == null) {
+      String catalogType = conf.get(InputFormatConfig.CATALOG);
+      if (catalogType == null) {
+        return HIVE_CATALOG_TYPE;
+      } else if (catalogType.equals(LOCATION)) {
         return NO_CATALOG_TYPE;
       } else {
         return catalogType;

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

@@ -121,8 +121,8 @@ public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
       String tables = jobConf.get(InputFormatConfig.OUTPUT_TABLES);
       String catalogName = tableDesc.getProperties().getProperty(InputFormatConfig.CATALOG_NAME);
       if (catalogName != null) {
-        tables = tables == null ? catalogName + CATALOG_NAME_SEPARATOR + tableDesc.getTableName() :
-                tables + TABLE_NAME_SEPARATOR + catalogName + CATALOG_NAME_SEPARATOR + tableDesc.getTableName();
+        String tableWithCatalogName = catalogName + CATALOG_NAME_SEPARATOR + tableDesc.getTableName();
+        tables = tables == null ? tableWithCatalogName : tables + TABLE_NAME_SEPARATOR + tableWithCatalogName;
       } else {
         tables = tables == null ? tableDesc.getTableName() : tables + TABLE_NAME_SEPARATOR + tableDesc.getTableName();
       }
@@ -174,7 +174,7 @@ public static Table table(Configuration config, String name) {
   /**
    * Returns the names of the output tables stored in the configuration.
    * @param config The configuration used to get the data from
-   * @return The collection of the table names as returned by TableDesc.getTableName()
+   * @return The collection of catalog name - table name pairs.
    */
   public static Collection<Pair<String, String>> outputTables(Configuration config) {
     Collection<String> tables = TABLE_NAME_SPLITTER.splitToList(config.get(InputFormatConfig.OUTPUT_TABLES));

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

@@ -68,7 +68,7 @@ static TestHiveShell shell(Map<String, String> configs) {
     TestHiveShell shell = new TestHiveShell();
     shell.setHiveConfValue("hive.notification.event.poll.interval", "-1");
     shell.setHiveConfValue("hive.tez.exec.print.summary", "true");
-    configs.entrySet().forEach(e -> shell.setHiveConfValue(e.getKey(), e.getValue()));
+    configs.forEach((k, v) -> shell.setHiveConfValue(k, v));
     // We would like to make sure that ORC reading overrides this config, so reading Iceberg tables could work in
     // systems (like Hive 3.2 and higher) where this value is set to true explicitly.
     shell.setHiveConfValue(OrcConf.FORCE_POSITIONAL_EVOLUTION.getHiveConfName(), "true");

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

@@ -361,7 +361,6 @@ public void testCreateTableAboveExistingTable() throws IOException {
         Collections.emptyList());
 
     if (testTableType == TestTables.TestTableType.HIVE_CATALOG) {
-
       // In HiveCatalog we just expect an exception since the table is already exists
       AssertHelpers.assertThrows("should throw exception", IllegalArgumentException.class,
           "customers already exists", () -> {