Save catalogName in jobConf

Author: Laszlo Pinter

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

@@ -47,6 +47,7 @@ private InputFormatConfig() {
   public static final String TABLE_SCHEMA = "iceberg.mr.table.schema";
   public static final String PARTITION_SPEC = "iceberg.mr.table.partition.spec";
   public static final String SERIALIZED_TABLE_PREFIX = "iceberg.mr.serialized.table.";
+  public static final String TABLE_CATALOG_PREFIX = "iceberg.mr.table.catalog";
   public static final String LOCALITY = "iceberg.mr.locality";
   public static final String CATALOG = "iceberg.mr.catalog";
   public static final String HADOOP_CATALOG_WAREHOUSE_LOCATION = "iceberg.mr.catalog.hadoop.warehouse.location";

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

@@ -51,7 +51,6 @@
 import org.apache.iceberg.mr.InputFormatConfig;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.util.concurrent.ThreadFactoryBuilder;
-import org.apache.iceberg.util.Pair;
 import org.apache.iceberg.util.Tasks;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -94,7 +93,7 @@ public void commitTask(TaskAttemptContext originalContext) throws IOException {
     TaskAttemptID attemptID = context.getTaskAttemptID();
     JobConf jobConf = context.getJobConf();
     Map<String, HiveIcebergRecordWriter> writers = HiveIcebergRecordWriter.getWriters(attemptID);
-    Collection<Pair<String, String>> outputs = HiveIcebergStorageHandler.outputTables(context.getJobConf());
+    Collection<String> outputs = HiveIcebergStorageHandler.outputTables(context.getJobConf());
 
     ExecutorService tableExecutor = tableExecutor(jobConf, outputs.size());
     try {
@@ -105,8 +104,8 @@ public void commitTask(TaskAttemptContext originalContext) throws IOException {
           .throwFailureWhenFinished()
           .executeWith(tableExecutor)
           .run(output -> {
-            Table table = HiveIcebergStorageHandler.table(context.getJobConf(), output.second());
-            HiveIcebergRecordWriter writer = writers.get(output.second());
+            Table table = HiveIcebergStorageHandler.table(context.getJobConf(), output);
+            HiveIcebergRecordWriter writer = writers.get(output);
             DataFile[] closedFiles = writer != null ? writer.dataFiles() : new DataFile[0];
             String fileForCommitLocation = generateFileForCommitLocation(table.location(), jobConf,
                 attemptID.getJobID(), attemptID.getTaskID().getId());
@@ -158,7 +157,7 @@ public void commitJob(JobContext originalContext) throws IOException {
     long startTime = System.currentTimeMillis();
     LOG.info("Committing job {} has started", jobContext.getJobID());
 
-    Collection<Pair<String, String>> outputs = HiveIcebergStorageHandler.outputTables(jobContext.getJobConf());
+    Collection<String> outputs = HiveIcebergStorageHandler.outputTables(jobContext.getJobConf());
     Collection<String> jobLocations = new ConcurrentLinkedQueue<>();
 
     ExecutorService fileExecutor = fileExecutor(jobConf);
@@ -170,9 +169,10 @@ public void commitJob(JobContext originalContext) throws IOException {
           .stopOnFailure()
           .executeWith(tableExecutor)
           .run(output -> {
-            Table table = HiveIcebergStorageHandler.table(jobConf, output.second());
+            Table table = HiveIcebergStorageHandler.table(jobConf, output);
+            String catalogName = HiveIcebergStorageHandler.catalogName(jobConf, output);
             jobLocations.add(generateJobLocation(table.location(), jobConf, jobContext.getJobID()));
-            commitTable(table.io(), fileExecutor, jobContext, output.second(), table.location(), output.first());
+            commitTable(table.io(), fileExecutor, jobContext, output, table.location(), catalogName);
           });
     } finally {
       fileExecutor.shutdown();
@@ -199,7 +199,7 @@ public void abortJob(JobContext originalContext, int status) throws IOException
     JobConf jobConf = jobContext.getJobConf();
 
     LOG.info("Job {} is aborted. Data file cleaning started", jobContext.getJobID());
-    Collection<Pair<String, String>> outputs = HiveIcebergStorageHandler.outputTables(jobContext.getJobConf());
+    Collection<String> outputs = HiveIcebergStorageHandler.outputTables(jobContext.getJobConf());
     Collection<String> jobLocations = new ConcurrentLinkedQueue<>();
 
     ExecutorService fileExecutor = fileExecutor(jobConf);
@@ -213,7 +213,7 @@ public void abortJob(JobContext originalContext, int status) throws IOException
           .run(output -> {
             LOG.info("Cleaning job for table {}", jobContext.getJobID(), output);
 
-            Table table = HiveIcebergStorageHandler.table(jobConf, output.second());
+            Table table = HiveIcebergStorageHandler.table(jobConf, output);
             jobLocations.add(generateJobLocation(table.location(), jobConf, jobContext.getJobID()));
             Collection<DataFile> dataFiles = dataFiles(fileExecutor, table.location(), jobContext, table.io(), false);
 

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

@@ -44,15 +44,12 @@
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.base.Splitter;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
-import org.apache.iceberg.util.Pair;
 import org.apache.iceberg.util.SerializationUtil;
 
 public class HiveIcebergStorageHandler implements HiveStoragePredicateHandler, HiveStorageHandler {
   private static final Splitter TABLE_NAME_SPLITTER = Splitter.on("..");
   private static final String TABLE_NAME_SEPARATOR = "..";
-  public static final String CATALOG_NAME_SEPARATOR = "::";
 
   static final String WRITE_KEY = "HiveIcebergStorageHandler_write";
 
@@ -116,19 +113,18 @@ public void configureInputJobCredentials(TableDesc tableDesc, Map<String, String
   public void configureJobConf(TableDesc tableDesc, JobConf jobConf) {
     if (tableDesc != null && tableDesc.getProperties() != null &&
         tableDesc.getProperties().get(WRITE_KEY) != null) {
-      Preconditions.checkArgument(!tableDesc.getTableName().contains(TABLE_NAME_SEPARATOR),
-          "Can not handle table " + tableDesc.getTableName() + ". Its name contains '" + TABLE_NAME_SEPARATOR + "'");
+      String tableName = tableDesc.getTableName();
+      Preconditions.checkArgument(!tableName.contains(TABLE_NAME_SEPARATOR),
+          "Can not handle table " + tableName + ". Its name contains '" + TABLE_NAME_SEPARATOR + "'");
       String tables = jobConf.get(InputFormatConfig.OUTPUT_TABLES);
+      tables = tables == null ? tableName : tables + TABLE_NAME_SEPARATOR + tableName;
+      jobConf.set("mapred.output.committer.class", HiveIcebergOutputCommitter.class.getName());
+      jobConf.set(InputFormatConfig.OUTPUT_TABLES, tables);
+
       String catalogName = tableDesc.getProperties().getProperty(InputFormatConfig.CATALOG_NAME);
       if (catalogName != null) {
-        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();
+        jobConf.set(InputFormatConfig.TABLE_CATALOG_PREFIX + tableName, catalogName);
       }
-
-      jobConf.set("mapred.output.committer.class", HiveIcebergOutputCommitter.class.getName());
-      jobConf.set(InputFormatConfig.OUTPUT_TABLES, tables);
     }
   }
 
@@ -174,19 +170,20 @@ 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 catalog name - table name pairs.
+   * @return The collection of the table names as returned by TableDesc.getTableName()
    */
-  public static Collection<Pair<String, String>> outputTables(Configuration config) {
-    Collection<String> tables = TABLE_NAME_SPLITTER.splitToList(config.get(InputFormatConfig.OUTPUT_TABLES));
-    Collection<Pair<String, String>> result = Lists.newArrayList();
-    tables.stream().map(t -> t.split(CATALOG_NAME_SEPARATOR, 2)).forEach(s -> {
-      if (s.length < 2) {
-        result.add(Pair.of(null, s[0]));
-      } else {
-        result.add(Pair.of(s[0], s[1]));
-      }
-    });
-    return result;
+  public static Collection<String> outputTables(Configuration config) {
+    return TABLE_NAME_SPLITTER.splitToList(config.get(InputFormatConfig.OUTPUT_TABLES));
+  }
+
+  /**
+   * Returns the catalog name serialized to the configuration.
+   * @param config The configuration used to get the data from
+   * @param name The name of the table we neeed as returned by TableDesc.getTableName()
+   * @return catalog name
+   */
+  public static String catalogName(Configuration config, String name) {
+    return config.get(InputFormatConfig.TABLE_CATALOG_PREFIX + name);
   }
 
   /**

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

@@ -229,8 +229,7 @@ private JobConf jobConf(Table table, int taskNum) {
     conf.setNumMapTasks(taskNum);
     conf.setNumReduceTasks(0);
     conf.set(HiveConf.ConfVars.HIVEQUERYID.varname, QUERY_ID);
-    conf.set(InputFormatConfig.OUTPUT_TABLES, table.properties().get(InputFormatConfig.CATALOG_NAME) +
-            HiveIcebergStorageHandler.CATALOG_NAME_SEPARATOR + table.name());
+    conf.set(InputFormatConfig.OUTPUT_TABLES, table.name());
     conf.set(InputFormatConfig.SERIALIZED_TABLE_PREFIX + table.name(), SerializationUtil.serializeToBase64(table));
 
     Map<String, String> propMap = Maps.newHashMap();