// Removed POSort, if the successor require a databag, we need to
// add a PORelationToExprProject
// to convert tuples into databag
for (POToChange sortToRemove : sortsToRemove) {
numSortRemoved++;
POSort oldSort = (POSort) sortToRemove.oper;
String scope = oldSort.getOperatorKey().scope;
List<PhysicalOperator> preds = sortToRemove.plan
.getPredecessors(sortToRemove.oper);
List<PhysicalOperator> succs = sortToRemove.plan
.getSuccessors(sortToRemove.oper);
POProject project = null;
if ((preds == null
|| preds.get(0).getResultType() != DataType.BAG
&& oldSort.getResultType() == DataType.BAG) // sort to remove do change the result type
&& (succs == null || !(succs.get(0) instanceof PORelationToExprProject))) // successor is not PORelationToExprProject
{
project = new PORelationToExprProject(new OperatorKey(
scope, NodeIdGenerator.getGenerator()
.getNextNodeId(scope)), oldSort
.getRequestedParallelism());
project.setInputs(oldSort.getInputs());
project.setResultType(DataType.BAG);
project.setStar(true);
}
if (project == null)
sortToRemove.plan.removeAndReconnect(sortToRemove.oper);