@Override
  public List<InputSplit> getSplits(JobContext jobContext)
    throws IOException, InterruptedException {
    Configuration conf = jobContext.getConfiguration();
    // Set up recursive reads for sub-directories.
    // (Otherwise, sub-directories produced by Hive UNION operations won't be readable.)
    conf.setBoolean("mapred.input.dir.recursive", true);

    //Get the job info from the configuration,
    //throws exception if not initialized
    InputJobInfo inputJobInfo;
    try {
      inputJobInfo = getJobInfo(conf);
    } catch (Exception e) {
      throw new IOException(e);
    }

    List<InputSplit> splits = new ArrayList<InputSplit>();
    List<PartInfo> partitionInfoList = inputJobInfo.getPartitions();
    if (partitionInfoList == null) {
      //No partitions match the specified partition filter
      return splits;
    }

    HiveStorageHandler storageHandler;
    Map<String,String> hiveProps = null;
    //For each matching partition, call getSplits on the underlying InputFormat
    for (PartInfo partitionInfo : partitionInfoList) {
      JobConf jobConf = HCatUtil.getJobConfFromContext(jobContext);
      if (hiveProps == null) {
        hiveProps = HCatUtil.getHCatKeyHiveConf(jobConf);
      }
      List<String> setInputPath = setInputPath(jobConf, partitionInfo.getLocation());
      if (setInputPath.isEmpty()) {
        continue;
      }
      Map<String, String> jobProperties = partitionInfo.getJobProperties();

      HCatUtil.copyJobPropertiesToJobConf(hiveProps, jobConf);
      HCatUtil.copyJobPropertiesToJobConf(jobProperties, jobConf);

      storageHandler = HCatUtil.getStorageHandler(
        jobConf, partitionInfo);

      //Get the input format
      Class inputFormatClass = storageHandler.getInputFormatClass();
      org.apache.hadoop.mapred.InputFormat inputFormat =
        getMapRedInputFormat(jobConf, inputFormatClass);

      //Call getSplit on the InputFormat, create an HCatSplit for each
      //underlying split. When the desired number of input splits is missing,
      //use a default number (denoted by zero).
      //TODO(malewicz): Currently each partition is split independently into
      //a desired number. However, we want the union of all partitions to be
      //split into a desired number while maintaining balanced sizes of input
      //splits.
      int desiredNumSplits =
        conf.getInt(HCatConstants.HCAT_DESIRED_PARTITION_NUM_SPLITS, 0);
      org.apache.hadoop.mapred.InputSplit[] baseSplits =
        inputFormat.getSplits(jobConf, desiredNumSplits);

      for (org.apache.hadoop.mapred.InputSplit split : baseSplits) {
        splits.add(new HCatSplit(partitionInfo, split));
      }
    }

    return splits;
  }
Logically split the set of input files for the job. Returns the
underlying InputFormat's splits
@param jobContext the job context object
@return the splits, an HCatInputSplit wrapper over the storage
        handler InputSplits
@throws IOException or InterruptedException
hcatalog/core/src/main/java/org/apache/hive/hcatalog/mapreduce/HCatBaseInputFormat.java