@Override
  public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
    PerfLogger perfLogger = SessionState.getPerfLogger();
    perfLogger.perfLogBegin(CLASS_NAME, PerfLogger.GET_SPLITS);
    init(job);

    ArrayList<InputSplit> result = new ArrayList<InputSplit>();

    Path[] paths = getInputPaths(job);

    List<Path> nonCombinablePaths = new ArrayList<Path>(paths.length / 2);
    List<Path> combinablePaths = new ArrayList<Path>(paths.length / 2);

    int numThreads = Math.min(MAX_CHECK_NONCOMBINABLE_THREAD_NUM,
        (int) Math.ceil((double) paths.length / DEFAULT_NUM_PATH_PER_THREAD));

    try {
      Set<Integer> nonCombinablePathIndices = getNonCombinablePathIndices(job, paths, numThreads);
      for (int i = 0; i < paths.length; i++) {
        if (nonCombinablePathIndices.contains(i)) {
          nonCombinablePaths.add(paths[i]);
        } else {
          combinablePaths.add(paths[i]);
        }
      }
    } catch (Exception e) {
      LOG.error("Error checking non-combinable path", e);
      perfLogger.perfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
      throw new IOException(e);
    }

    // Store the previous value for the path specification
    String oldPaths = job.get(org.apache.hadoop.mapreduce.lib.input.FileInputFormat.INPUT_DIR);
    LOG.debug("The received input paths are: [{}] against the property {}", oldPaths,
        org.apache.hadoop.mapreduce.lib.input.FileInputFormat.INPUT_DIR);

    // Process the normal splits
    if (nonCombinablePaths.size() > 0) {
      FileInputFormat.setInputPaths(job, nonCombinablePaths.toArray
          (new Path[nonCombinablePaths.size()]));
      InputSplit[] splits = super.getSplits(job, numSplits);
      for (InputSplit split : splits) {
        result.add(split);
      }
    }

    // Process the combine splits
    if (combinablePaths.size() > 0) {
      FileInputFormat.setInputPaths(job, combinablePaths.toArray
          (new Path[combinablePaths.size()]));
      Map<Path, PartitionDesc> pathToPartitionInfo = this.pathToPartitionInfo != null ?
          this.pathToPartitionInfo : Utilities.getMapWork(job).getPathToPartitionInfo();
      InputSplit[] splits = getCombineSplits(job, numSplits, pathToPartitionInfo);
      for (InputSplit split : splits) {
        result.add(split);
      }
    }

    // Restore the old path information back
    // This is just to prevent incompatibilities with previous versions Hive
    // if some application depends on the original value being set.
    if (oldPaths != null) {
      job.set(org.apache.hadoop.mapreduce.lib.input.FileInputFormat.INPUT_DIR, oldPaths);
    }

    // clear work from ThreadLocal after splits generated in case of thread is reused in pool.
    Utilities.clearWorkMapForConf(job);

    if (result.isEmpty() && paths.length > 0 && job.getBoolean(Utilities.ENSURE_OPERATORS_EXECUTED, false)) {
      // If there are no inputs; the Execution engine skips the operator tree.
      // To prevent it from happening; an opaque  ZeroRows input is added here - when needed.
      result.add(
          new HiveInputSplit(new NullRowsInputFormat.DummyInputSplit(paths[0]), ZeroRowsInputFormat.class.getName()));
    }

    LOG.info("Number of all splits " + result.size());
    perfLogger.perfLogEnd(CLASS_NAME, PerfLogger.GET_SPLITS);
    return result.toArray(new InputSplit[result.size()]);
  }
Create Hive splits based on CombineFileSplit.
ql/src/java/org/apache/hadoop/hive/ql/io/CombineHiveInputFormat.java