private ReduceSinkOperator getReduceSinkOp(List<Integer> keyPositions, List<Integer> sortOrder,
        List<Integer> sortNullOrder, ArrayList<ExprNodeDesc> allCols, Operator<? extends OperatorDesc> parent,
        AcidUtils.Operation writeType) {
      // we will clone here as RS will update bucket column key with its
      // corresponding with bucket number and hence their OIs
      final ArrayList<ExprNodeDesc> keyCols = keyPositions.stream()
          .map(id -> allCols.get(id).clone())
          .collect(Collectors.toCollection(ArrayList::new));
      ArrayList<ExprNodeDesc> valCols = Lists.newArrayList();
      for (int i = 0; i < allCols.size(); i++) {
        if (i != granularityKeyPos && i != partitionKeyPos) {
          valCols.add(allCols.get(i).clone());
        }
      }

      final ArrayList<ExprNodeDesc> partCols =
          keyPositions.stream().map(id -> allCols.get(id).clone())
              .collect(Collectors.toCollection(ArrayList::new));

      // map _col0 to KEY._col0, etc
      Map<String, ExprNodeDesc> colExprMap = Maps.newHashMap();
      Map<String, String> nameMapping = new HashMap<>();
      final ArrayList<String> keyColNames = Lists.newArrayList();
      final ArrayList<String> valColNames = Lists.newArrayList();
      for (ExprNodeDesc exprNodeDesc : keyCols) {
        keyColNames.add(exprNodeDesc.getExprString());
        String key = Utilities.ReduceField.KEY + "." + exprNodeDesc.getExprString();
        colExprMap.put(key, exprNodeDesc);
        nameMapping.put(exprNodeDesc.getExprString(), key);
      }
      for (ExprNodeDesc exprNodeDesc : valCols) {
        valColNames.add(exprNodeDesc.getExprString());
        String key = Utilities.ReduceField.VALUE + "." + exprNodeDesc.getExprString();
        colExprMap.put(key, exprNodeDesc);
        nameMapping.put(exprNodeDesc.getExprString(), key);
      }


      // order and null order
      final String orderStr = StringUtils.repeat("+", sortOrder.size());
      final String nullOrderStr = StringUtils.repeat("a", sortNullOrder.size());

      // Create Key/Value TableDesc. When the operator plan is split into MR tasks,
      // the reduce operator will initialize Extract operator with information
      // from Key and Value TableDesc
      final List<FieldSchema> fields = PlanUtils.getFieldSchemasFromColumnList(keyCols,
          keyColNames, 0, "");
      final TableDesc keyTable = PlanUtils.getReduceKeyTableDesc(fields, orderStr, nullOrderStr);
      List<FieldSchema> valFields = PlanUtils.getFieldSchemasFromColumnList(valCols,
          valColNames, 0, "");
      final TableDesc valueTable = PlanUtils.getReduceValueTableDesc(valFields);
      List<List<Integer>> distinctColumnIndices = Lists.newArrayList();

      // Number of reducers is set to default (-1)
      final ReduceSinkDesc rsConf = new ReduceSinkDesc(keyCols, keyCols.size(), valCols,
          keyColNames, distinctColumnIndices, valColNames, -1, partCols, -1, keyTable,
          valueTable, writeType);

      final ArrayList<ColumnInfo> signature =
          parent.getSchema().getSignature()
              .stream()
              .map(e -> new ColumnInfo(e))
              .map(columnInfo ->
                {
                  columnInfo.setInternalName(nameMapping.get(columnInfo.getInternalName()));
                  return columnInfo;
               })
              .collect(Collectors.toCollection(ArrayList::new));
      final ReduceSinkOperator op = (ReduceSinkOperator) OperatorFactory.getAndMakeChild(
          rsConf, new RowSchema(signature), parent);
      op.setColumnExprMap(colExprMap);
      return op;
    }
Introduces a RS before FS to partition data by configuration specified
time granularity.
ql/src/java/org/apache/hadoop/hive/ql/optimizer/SortedDynPartitionTimeGranularityOptimizer.java