private static Path setBaseWork(Configuration conf, BaseWork w, Path hiveScratchDir, String name, boolean useCache) {
    Kryo kryo = SerializationUtilities.borrowKryo(conf);
    try {
      setPlanPath(conf, hiveScratchDir);

      Path planPath = getPlanPath(conf, name);
      setHasWork(conf, name);

      OutputStream out = null;

      final long serializedSize;
      final String planMode;
      if (HiveConf.getBoolVar(conf, ConfVars.HIVE_RPC_QUERY_PLAN)) {
        // add it to the conf
        ByteArrayOutputStream byteOut = new ByteArrayOutputStream();
        try {
          out = new DeflaterOutputStream(byteOut, new Deflater(Deflater.BEST_SPEED));
          SerializationUtilities.serializePlan(kryo, w, out);
          out.close();
          out = null;
        } finally {
          IOUtils.closeStream(out);
        }
        final String serializedPlan = Base64.getEncoder().encodeToString(byteOut.toByteArray());
        serializedSize = serializedPlan.length();
        planMode = "RPC";
        conf.set(planPath.toUri().getPath(), serializedPlan);
      } else {
        // use the default file system of the conf
        FileSystem fs = planPath.getFileSystem(conf);
        try {
          out = fs.create(planPath);
          SerializationUtilities.serializePlan(kryo, w, out);
          out.close();
          out = null;
          long fileLen = fs.getFileStatus(planPath).getLen();
          serializedSize = fileLen;
          planMode = "FILE";
        } finally {
          IOUtils.closeStream(out);
        }

        // Serialize the plan to the default hdfs instance
        // Except for hadoop local mode execution where we should be
        // able to get the plan directly from the cache
        if (useCache && !ShimLoader.getHadoopShims().isLocalMode(conf)) {
          // Set up distributed cache
          if (!DistributedCache.getSymlink(conf)) {
            DistributedCache.createSymlink(conf);
          }
          String uriWithLink = planPath.toUri().toString() + "#" + name;
          DistributedCache.addCacheFile(new URI(uriWithLink), conf);

          // set replication of the plan file to a high number. we use the same
          // replication factor as used by the hadoop jobclient for job.xml etc.
          short replication = (short) conf.getInt("mapred.submit.replication", 10);
          fs.setReplication(planPath, replication);
        }
      }

      LOG.info("Serialized plan (via {}) - name: {} size: {}", planMode, w.getName(),
          humanReadableByteCount(serializedSize));
      // Cache the plan in this process
      gWorkMap.get(conf).put(planPath, w);
      return planPath;
    } catch (Exception e) {
      String msg = "Error caching " + name;
      LOG.error(msg, e);
      throw new RuntimeException(msg, e);
    } finally {
      SerializationUtilities.releaseKryo(kryo);
    }
  }
Returns the Map or Reduce plan
Side effect: the BaseWork returned is also placed in the gWorkMap
@param conf
@param name
@return BaseWork based on the name supplied will return null if name is null
@throws RuntimeException if the configuration files are not proper or if plan can not be loaded
ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java