Review changes 2.

Author: Laszlo Pinter

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

@@ -194,36 +194,14 @@ public static boolean dropTable(Configuration conf, Properties props) {
    */
   public static boolean hiveCatalog(Configuration conf, Properties props) {
     String catalogName = props.getProperty(InputFormatConfig.TABLE_CATALOG);
-    if (catalogName != null) {
-      return HIVE.equalsIgnoreCase(conf.get(String.format(InputFormatConfig.CATALOG_TYPE_TEMPLATE, catalogName)));
-    } else {
-      if (HIVE.equalsIgnoreCase(conf.get(String.format(InputFormatConfig.CATALOG_TYPE_TEMPLATE, DEFAULT_CATALOG)))) {
-        return true;
-      } else {
-        return HIVE.equalsIgnoreCase(conf.get(InputFormatConfig.CATALOG));
-      }
-    }
+    return HIVE.equalsIgnoreCase(getCatalogType(conf, catalogName));
   }
 
   @VisibleForTesting
   static Optional<Catalog> loadCatalog(Configuration conf, String catalogName) {
-    String catalogType;
     String name = catalogName == null ? DEFAULT_CATALOG : catalogName;
-    catalogType = conf.get(String.format(InputFormatConfig.CATALOG_TYPE_TEMPLATE, name));
-
-    // keep both catalog configuration methods for seamless transition
-    if (catalogType != null) {
-      // new logic
-      LOG.debug("Using catalog configuration from table properties.");
-      return loadCatalog(conf, name, catalogType);
-    } else {
-      // old logic
-      // use catalog {@link InputFormatConfig.CATALOG} stored in global hive config if table specific catalog
-      // configuration or default catalog definition is missing
-      LOG.debug("Using catalog configuration from global configuration.");
-      catalogType = conf.get(InputFormatConfig.CATALOG);
-      return loadCatalog(conf, name, catalogType);
-    }
+    String catalogType = getCatalogType(conf, name);
+    return loadCatalog(conf, name, catalogType);
   }
 
   private static Optional<Catalog> loadCatalog(Configuration conf, String catalogName, String catalogType) {
@@ -237,8 +215,7 @@ private static Optional<Catalog> loadCatalog(Configuration conf, String catalogN
     switch (catalogType.toLowerCase()) {
       case HADOOP:
         if (properties.containsKey(CatalogProperties.WAREHOUSE_LOCATION)) {
-          catalog = CatalogUtil.loadCatalog(HadoopCatalog.class.getName(), catalogName,
-                  getCatalogProperties(conf, catalogName), conf);
+          catalog = CatalogUtil.loadCatalog(HadoopCatalog.class.getName(), catalogName, properties, conf);
         } else {
           String warehouseLocation = conf.get(InputFormatConfig.HADOOP_CATALOG_WAREHOUSE_LOCATION);
           catalog = (warehouseLocation != null) ? new HadoopCatalog(conf, warehouseLocation) : new HadoopCatalog(conf);
@@ -282,4 +259,17 @@ private static Map<String, String> getCatalogProperties(Configuration conf, Stri
     });
     return properties;
   }
+
+  private static String getCatalogType(Configuration conf, String catalogName) {
+    String name = catalogName == null ? DEFAULT_CATALOG : catalogName;
+    String catalogType = conf.get(String.format(InputFormatConfig.CATALOG_TYPE_TEMPLATE, name));
+    // keep both catalog configuration methods for seamless transition
+    if (catalogType != null) {
+      LOG.debug("Using catalog configuration from table properties.");
+      return catalogType;
+    } else {
+      LOG.debug("Using catalog configuration from global configuration.");
+      return conf.get(InputFormatConfig.CATALOG);
+    }
+  }
 }

mr/src/test/java/org/apache/iceberg/mr/TestCatalogs.java

@@ -79,10 +79,7 @@ public void testLoadTableFromLocation() throws IOException {
   public void testLoadTableFromCatalog() throws IOException {
     String defaultCatalogName = "default";
     String warehouseLocation = temp.newFolder("hadoop", "warehouse").toString();
-    conf.set(String.format(InputFormatConfig.CATALOG_WAREHOUSE_TEMPLATE, defaultCatalogName), warehouseLocation);
-    conf.set(String.format(InputFormatConfig.CATALOG_TYPE_TEMPLATE, defaultCatalogName), Catalogs.CUSTOM);
-    conf.set(String.format(InputFormatConfig.CATALOG_CLASS_TEMPLATE, defaultCatalogName),
-            CustomHadoopCatalog.class.getName());
+    setCustomCatalogProperties(defaultCatalogName, warehouseLocation);
 
     AssertHelpers.assertThrows(
             "Should complain about table identifier not set", IllegalArgumentException.class,
@@ -145,10 +142,7 @@ public void testCreateDropTableToCatalog() throws IOException {
     String defaultCatalogName = "default";
     String warehouseLocation = temp.newFolder("hadoop", "warehouse").toString();
 
-    conf.set(String.format(InputFormatConfig.CATALOG_WAREHOUSE_TEMPLATE, defaultCatalogName), warehouseLocation);
-    conf.set(String.format(InputFormatConfig.CATALOG_CLASS_TEMPLATE, defaultCatalogName),
-            CustomHadoopCatalog.class.getName());
-    conf.set(String.format(InputFormatConfig.CATALOG_TYPE_TEMPLATE, defaultCatalogName), Catalogs.CUSTOM);
+    setCustomCatalogProperties(defaultCatalogName, warehouseLocation);
 
     Properties missingSchema = new Properties();
     missingSchema.put("name", identifier.toString());
@@ -298,4 +292,10 @@ public Catalog load(Configuration conf) {
       return new CustomHadoopCatalog(conf);
     }
   }
+
+  private void setCustomCatalogProperties(String catalogName, String warehouseLocation) {
+    conf.set(String.format(InputFormatConfig.CATALOG_WAREHOUSE_TEMPLATE, catalogName), warehouseLocation);
+    conf.set(String.format(InputFormatConfig.CATALOG_CLASS_TEMPLATE, catalogName), CustomHadoopCatalog.class.getName());
+    conf.set(String.format(InputFormatConfig.CATALOG_TYPE_TEMPLATE, catalogName), Catalogs.CUSTOM);
+  }
 }