private List<List<Integer>> getDistinctColIndicesForReduceSink(QBParseInfo parseInfo,
                                                                 String dest,
                                                                 List<ExprNodeDesc> reduceKeys, RowResolver reduceSinkInputRowResolver,
                                                                 RowResolver reduceSinkOutputRowResolver, List<String> outputKeyColumnNames,
                                                                 Map<String, ExprNodeDesc> colExprMap)
      throws SemanticException {

    List<List<Integer>> distinctColIndices = new ArrayList<List<Integer>>();

    // If there is a distinctFuncExp, add all parameters to the reduceKeys.
    if (!parseInfo.getDistinctFuncExprsForClause(dest).isEmpty()) {
      List<ASTNode> distFuncs = parseInfo.getDistinctFuncExprsForClause(dest);
      String colName = getColumnInternalName(reduceKeys.size());
      outputKeyColumnNames.add(colName);
      for (int i = 0; i < distFuncs.size(); i++) {
        ASTNode value = distFuncs.get(i);
        int numExprs = 0;
        List<Integer> distinctIndices = new ArrayList<Integer>();
        // 0 is function name
        for (int j = 1; j < value.getChildCount(); j++) {
          ASTNode parameter = (ASTNode) value.getChild(j);
          ExprNodeDesc expr = genExprNodeDesc(parameter, reduceSinkInputRowResolver);
          // see if expr is already present in reduceKeys.
          // get index of expr in reduceKeys
          int ri;
          for (ri = 0; ri < reduceKeys.size(); ri++) {
            if (reduceKeys.get(ri).getExprString().equals(expr.getExprString())) {
              break;
            }
          }
          // add the expr to reduceKeys if it is not present
          if (ri == reduceKeys.size()) {
            String name = getColumnInternalName(numExprs);
            String field = Utilities.ReduceField.KEY.toString() + "." + colName
                + ":" + i
                + "." + name;
            ColumnInfo colInfo = new ColumnInfo(field, expr.getTypeInfo(), null, false);
            reduceSinkOutputRowResolver.putExpression(parameter, colInfo);
            colExprMap.put(field, expr);
            reduceKeys.add(expr);
          }
          // add the index of expr in reduceKeys to distinctIndices
          distinctIndices.add(ri);
          numExprs++;
        }
        distinctColIndices.add(distinctIndices);
      }
    }

    return distinctColIndices;
  }
Generate the ReduceSinkOperator for the Group By Query Block
(qb.getPartInfo().getXXX(dest)). The new ReduceSinkOperator will be a child
of inputOperatorInfo.

It will put all Group By keys and the distinct field (if any) in the
map-reduce sort key, and all other fields in the map-reduce value.

@param numPartitionFields
         the number of fields for map-reduce partitioning. This is usually
         the number of fields in the Group By keys.
@return the new ReduceSinkOperator.
@throws SemanticException
ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java