[HIVE-8394] HIVE-7803 doesn't handle Pig MultiQuery, can cause data-loss.

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 0.12.0, 0.13.1, 0.14.0
• Fix Version/s: 0.14.0
• Component/s: HCatalog
• Assignee: Mithun Radhakrishnan
• Reporter: Mithun Radhakrishnan
• Created: 08/Oct/14 01:17
• Updated: 13/Nov/14 19:41
• Resolved: 02/Nov/14 23:50

Description

We've found situations in production where Pig queries using HCatStorer, dynamic partitioning and opt.multiquery=true that produce partitions in the output table, but the corresponding directories have no data files (in spite of Pig reporting non-zero records written to HDFS). I don't yet have a distilled test-case for this.

Here's the code from FileOutputCommitterContainer after HIVE-7803:

FileOutputCommitterContainer.java

  @Override
  public void commitTask(TaskAttemptContext context) throws IOException {
    String jobInfoStr = context.getConfiguration().get(FileRecordWriterContainer.DYN_JOBINFO);
    if (!dynamicPartitioningUsed) {
         //See HCATALOG-499
      FileOutputFormatContainer.setWorkOutputPath(context);
      getBaseOutputCommitter().commitTask(HCatMapRedUtil.createTaskAttemptContext(context));
    } else if (jobInfoStr != null) {
      ArrayList<String> jobInfoList = (ArrayList<String>)HCatUtil.deserialize(jobInfoStr);
      org.apache.hadoop.mapred.TaskAttemptContext currTaskContext = HCatMapRedUtil.createTaskAttemptContext(context);
      for (String jobStr : jobInfoList) {
    	OutputJobInfo localJobInfo = (OutputJobInfo)HCatUtil.deserialize(jobStr);
    	FileOutputCommitter committer = new FileOutputCommitter(new Path(localJobInfo.getLocation()), currTaskContext);
    	committer.commitTask(currTaskContext);
      }
    }
  }

The serialized jobInfoList can't be retrieved, and hence the commit never completes. This is because Pig's MapReducePOStoreImpl deliberately clones both the TaskAttemptContext and the contained Configuration instance, thus separating the Configuration instances passed to FileOutputCommitterContainer::commitTask() and FileRecordWriterContainer::close(). Anything set by the RecordWriter is unavailable to the Committer.

One approach would have been to store state in the FileOutputFormatContainer. But that won't work since this is constructed via reflection in HCatOutputFormat (itself constructed via reflection by PigOutputFormat via HCatStorer). There's no guarantee that the instance is preserved.

My only recourse seems to be to use a Singleton to store shared state. I'm loath to indulge in this brand of shenanigans. (Statics and container-reuse in Tez might not play well together, for instance.) It might work if we're careful about tearing down the singleton.

Any other ideas?

Issue Links

• duplicates: HIVE-7803 Enable Hadoop speculative execution may cause corrupt output directory (dynamic partition)