src/org/apache/pig/backend/hadoop/executionengine/util/SecondaryKeyOptimizerUtil.java [130:185]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        SortKeyInfo result = new SortKeyInfo();
        List<PhysicalPlan> plans = rearrange.getPlans();
        nextPlan: for (int i = 0; i < plans.size(); i++) {
            PhysicalPlan plan = plans.get(i);
            ColumnChainInfo columnChainInfo = new ColumnChainInfo();
            if (plan.getRoots() == null) {
                log.debug("POLocalRearrange plan is null");
                return null;
            } else if (plan.getRoots().size() != 1) {
                // POLocalRearrange plan contains more than 1 root.
                // Probably there is an Expression operator in the local
                // rearrangement plan, skip this plan
                continue nextPlan;
            } else {
                List<Integer> columns = new ArrayList<Integer>();
                columns
                        .add(rearrange.getIndex()
                                & PigNullableWritable.idxSpace);

                // The first item inside columnChainInfo is set to type Tuple.
                // This value is not actually in use, but it intends to match
                // the type of POProject in reduce side
                columnChainInfo.insert(columns, DataType.TUPLE);

                PhysicalOperator node = plan.getRoots().get(0);
                while (node != null) {
                    if (node instanceof POProject) {
                        POProject project = (POProject) node;
                        if(project.isProjectToEnd()){
                            columnChainInfo.insert(project.getStartCol(),
                                    project.getResultType());
                        }else {
                            columnChainInfo.insert(
                                    project.getColumns(), project.getResultType());
                        }

                        if (plan.getSuccessors(node) == null)
                            node = null;
                        else if (plan.getSuccessors(node).size() != 1) {
                            log.debug(node + " have more than 1 successor");
                            node = null;
                        } else
                            node = plan.getSuccessors(node).get(0);
                    } else
                        // constant, UDF, we will pass
                        continue nextPlan;
                }
            }
            // Let's assume all main key is sorted ascendant, we can further
            // optimize it to match one of the nested sort/distinct key, because we do not
            // really care about how cogroup key are sorted; But it may not be the case
            // if sometime we switch all the comparator to byte comparator, so just
            // leave it as it is for now
            result.insertColumnChainInfo(i, columnChainInfo, true);
        }
        return result;
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/SecondaryKeyOptimizer.java [85:140]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
        SortKeyInfo result = new SortKeyInfo();
        List<PhysicalPlan> plans = rearrange.getPlans();
        nextPlan: for (int i = 0; i < plans.size(); i++) {
            PhysicalPlan plan = plans.get(i);
            ColumnChainInfo columnChainInfo = new ColumnChainInfo();
            if (plan.getRoots() == null) {
                log.debug("POLocalRearrange plan is null");
                return null;
            } else if (plan.getRoots().size() != 1) {
                // POLocalRearrange plan contains more than 1 root.
                // Probably there is an Expression operator in the local
                // rearrangement plan, skip this plan
                continue nextPlan;
            } else {
                List<Integer> columns = new ArrayList<Integer>();
                columns
                        .add(rearrange.getIndex()
                                & PigNullableWritable.idxSpace);
                
                // The first item inside columnChainInfo is set to type Tuple.
                // This value is not actually in use, but it intends to match
                // the type of POProject in reduce side
                columnChainInfo.insert(columns, DataType.TUPLE);

                PhysicalOperator node = plan.getRoots().get(0);
                while (node != null) {
                    if (node instanceof POProject) {
                        POProject project = (POProject) node;
                        if(project.isProjectToEnd()){
                            columnChainInfo.insert(project.getStartCol(), 
                                    project.getResultType());
                        }else {
                            columnChainInfo.insert(
                                    project.getColumns(), project.getResultType());
                        }
                        
                        if (plan.getSuccessors(node) == null)
                            node = null;
                        else if (plan.getSuccessors(node).size() != 1) {
                            log.debug(node + " have more than 1 successor");
                            node = null;
                        } else
                            node = plan.getSuccessors(node).get(0);
                    } else
                        // constant, UDF, we will pass
                        continue nextPlan;
                }
            }
            // Let's assume all main key is sorted ascendant, we can further
            // optimize it to match one of the nested sort/distinct key, because we do not
            // really care about how cogroup key are sorted; But it may not be the case
            // if sometime we switch all the comparator to byte comparator, so just
            // leave it as it is for now
            result.insertColumnChainInfo(i, columnChainInfo, true);
        }
        return result;
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -