@Override
  void close() {
    if (cache != null) {
      for (String key : cacheKeys) {
        cache.release(key);
      }
    }

    if (dynamicValueCache != null) {
      for (String k : dynamicValueCacheKeys) {
        dynamicValueCache.release(k);
      }
    }

    try {
      if (isAborted()) {
        for (ReduceRecordSource rs : sources) {
          if (!rs.close()) {
            setAborted(false); // Preserving the old logic. Hmm...
            break;
          }
        }
      }

      boolean abort = isAborted();
      reducer.close(abort);
      if (mergeWorkList != null) {
        for (BaseWork redWork : mergeWorkList) {
          ((ReduceWork) redWork).getReducer().close(abort);
        }
      }

      // Need to close the dummyOps as well. The operator pipeline
      // is not considered "closed/done" unless all operators are
      // done. For broadcast joins that includes the dummy parents.
      List<HashTableDummyOperator> dummyOps = reduceWork.getDummyOps();
      if (dummyOps != null) {
        for (Operator<?> dummyOp : dummyOps) {
          dummyOp.close(abort);
        }
      }
      ReportStats rps = new ReportStats(reporter, jconf);
      reducer.preorderMap(rps);

    } catch (Exception e) {
      if (!isAborted()) {
        // signal new failure to map-reduce
        LOG.error("Hit error while closing operators - failing tree");
        throw new RuntimeException("Hive Runtime Error while closing operators: " + e.getMessage(), e);
      }
    } finally {
      Utilities.clearWorkMap(jconf);
      MapredContext.close();
    }
  }
Get the inputs that should be streamed through reduce plan.

@param inputs
@return
@throws Exception
ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java