@SuppressWarnings("nls")
  private Operator genJoinReduceSinkChild(ExprNodeDesc[] joinKeys,
                                          Operator<?> parent, String[] srcs, int tag) throws SemanticException {

    Operator dummy = Operator.createDummy();  // dummy for backtracking
    dummy.setParentOperators(Arrays.asList(parent));

    RowResolver inputRR = opParseCtx.get(parent).getRowResolver();
    RowResolver outputRR = new RowResolver();
    List<String> outputColumns = new ArrayList<String>();
    List<ExprNodeDesc> reduceKeys = new ArrayList<ExprNodeDesc>();
    List<ExprNodeDesc> reduceKeysBack = new ArrayList<ExprNodeDesc>();

    // Compute join keys and store in reduceKeys
    for (ExprNodeDesc joinKey : joinKeys) {
      reduceKeys.add(joinKey);
      reduceKeysBack.add(ExprNodeDescUtils.backtrack(joinKey, dummy, parent));
    }

    // Walk over the input row resolver and copy in the output
    ArrayList<ExprNodeDesc> reduceValues = new ArrayList<ExprNodeDesc>();

    List<ColumnInfo> columns = inputRR.getColumnInfos();
    int[] index = new int[columns.size()];
    for (int i = 0; i < columns.size(); i++) {
      ColumnInfo colInfo = columns.get(i);
      String[] nm = inputRR.reverseLookup(colInfo.getInternalName());
      String[] nm2 = inputRR.getAlternateMappings(colInfo.getInternalName());
      ExprNodeDesc expr = new ExprNodeColumnDesc(colInfo);

      // backtrack can be null when input is script operator
      ExprNodeDesc exprBack = ExprNodeDescUtils.backtrack(expr, dummy, parent);
      if (exprBack != null) {
        if (ExprNodeDescUtils.isConstant(exprBack)) {
          int kindex = reduceKeysBack.indexOf(exprBack);
          if (kindex >= 0) {
            addJoinKeyToRowSchema(outputRR, index, i, colInfo, nm, nm2, kindex);
            continue;
          }
        } else {
          int startIdx = 0;
          int kindex;
          // joinKey may present multiple times, add the duplicates to the schema with different internal name.
          // example: KEY.reducesinkkey0, KEY.reducesinkkey1
          //      join        LU_CUSTOMER        a16
          //      on         (a15.CUSTOMER_ID = a16.CUSTOMER_ID and pa11.CUSTOMER_ID = a16.CUSTOMER_ID)
          while ((kindex = ExprNodeDescUtils.indexOf(exprBack, reduceKeysBack, startIdx)) >= 0) {
            addJoinKeyToRowSchema(outputRR, index, i, colInfo, nm, nm2, kindex);
            startIdx = kindex + 1;
          }
          if (startIdx > 0) {
            // at least one instance found
            continue;
          }
        }
      }
      index[i] = -reduceValues.size() - 1;
      String outputColName = getColumnInternalName(reduceValues.size());

      reduceValues.add(expr);

      ColumnInfo newColInfo = new ColumnInfo(colInfo);
      String internalColName = Utilities.ReduceField.VALUE + "." + outputColName;
      newColInfo.setInternalName(internalColName);
      newColInfo.setTabAlias(nm[0]);

      outputRR.put(nm[0], nm[1], newColInfo);
      if (nm2 != null) {
        outputRR.addMappingOnly(nm2[0], nm2[1], newColInfo);
      }
      outputColumns.add(outputColName);
    }
    dummy.setParentOperators(null);

    int numReds = -1;

    // Use only 1 reducer in case of cartesian product
    if (reduceKeys.size() == 0) {
      numReds = 1;
      String error = StrictChecks.checkCartesian(conf);
      if (error != null) {
        throw new SemanticException(error);
      }
    }

    ReduceSinkDesc rsDesc = PlanUtils.getReduceSinkDesc(reduceKeys,
        reduceValues, outputColumns, false, tag,
        reduceKeys.size(), numReds, AcidUtils.Operation.NOT_ACID, defaultNullOrder);

    Map<String, String> translatorMap = new HashMap<String, String>();

    Map<String, ExprNodeDesc> colExprMap = new HashMap<String, ExprNodeDesc>();

    List<String> keyColNames = rsDesc.getOutputKeyColumnNames();
    for (int i = 0 ; i < keyColNames.size(); i++) {
      String oldName = keyColNames.get(i);
      String newName = Utilities.ReduceField.KEY + "." + oldName;
      colExprMap.put(newName, reduceKeys.get(i));
      translatorMap.put(oldName, newName);
    }
    List<String> valColNames = rsDesc.getOutputValueColumnNames();
    for (int i = 0 ; i < valColNames.size(); i++) {
      String oldName = valColNames.get(i);
      String newName = Utilities.ReduceField.VALUE + "." + oldName;
      colExprMap.put(newName, reduceValues.get(i));
      translatorMap.put(oldName, newName);
    }

    RowSchema defaultRs = new RowSchema(outputRR.getColumnInfos());

    List<ColumnInfo> newColumnInfos = new ArrayList<ColumnInfo>();
    for (ColumnInfo ci : outputRR.getColumnInfos()) {
      if (translatorMap.containsKey(ci.getInternalName())) {
        ci = new ColumnInfo(ci);
        ci.setInternalName(translatorMap.get(ci.getInternalName()));
      }
      newColumnInfos.add(ci);
    }

    ReduceSinkOperator rsOp = (ReduceSinkOperator) putOpInsertMap(
        OperatorFactory.getAndMakeChild(rsDesc, new RowSchema(newColumnInfos), parent), outputRR);

    rsOp.setValueIndex(index);
    rsOp.setColumnExprMap(colExprMap);
    rsOp.setInputAliases(srcs);
    return rsOp;
  }
Creates an expression for converting from a table column to a row column. For example:
The table column is int but the query provides a string in the row, then we need to cast automatically.
@param tableField The target table column
@param rowField The source row column
@param conversion The value of this boolean is set to true if we detect that a conversion is needed. This is a
                  hidden return value hidden here, to notify the caller that a cast was needed.
@param dest The destination table for the error message
@param columnNum The destination column id for the error message
@return The Expression describing the selected column. Note that `conversion` can be considered as a return value
        as well
@throws SemanticException If conversion were needed, but automatic conversion is not available
ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java