public static void clearWork(Configuration conf) {
    Path mapPath = getPlanPath(conf, MAP_PLAN_NAME);
    Path reducePath = getPlanPath(conf, REDUCE_PLAN_NAME);

    // if the plan path hasn't been initialized just return, nothing to clean.
    if (mapPath == null && reducePath == null) {
      return;
    }


    try {
      if (!HiveConf.getBoolVar(conf, ConfVars.HIVE_RPC_QUERY_PLAN)) {
        FileSystem fs = mapPath.getFileSystem(conf);
        try {
          fs.delete(mapPath, true);
        } catch (FileNotFoundException e) {
          // delete if exists, don't panic if it doesn't
        }
        try {
          fs.delete(reducePath, true);
        } catch (FileNotFoundException e) {
          // delete if exists, don't panic if it doesn't
        }
      }
    } catch (Exception e) {
      LOG.warn("Failed to clean-up tmp directories.", e);
    } finally {
      // where a single process works with multiple plans - we must clear
      // the cache before working with the next plan.
      clearWorkMapForConf(conf);
    }
  }
ReduceField:
KEY: record key
VALUE: record value
ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java