}
@Override
public void transform(OperatorPlan matched) throws FrontendException {
LOLimit limit = (LOLimit) matched.getSources().get(0);
// Find the next foreach operator.
List<Operator> preds = currentPlan.getPredecessors(limit);
Operator pred = preds.get(0);
if (pred instanceof LOForEach) {
// We can safely move LOLimit up
// Get operator before LOForEach
Operator prepredecessor = currentPlan.getPredecessors(pred)
.get(0);
currentPlan.removeAndReconnect(limit);
currentPlan.insertBetween(prepredecessor, limit, pred);
} else if (pred instanceof LOCross || pred instanceof LOUnion) {
// Limit can be duplicated, and the new instance pushed in front
// of an operator for the following operators
// (that is, if you have X->limit, you can transform that to
// limit->X->limit):
LOLimit newLimit = null;
List<Operator> nodesToProcess = new ArrayList<Operator>();
for (Operator prepredecessor : currentPlan
.getPredecessors(pred))
nodesToProcess.add(prepredecessor);
for (Operator prepredecessor : nodesToProcess) {
if (prepredecessor instanceof LOLimit) {
LOLimit l = (LOLimit) prepredecessor;
l.setLimit(l.getLimit() < limit.getLimit() ? l
.getLimit() : limit.getLimit());
} else {
newLimit = new LOLimit((LogicalPlan) currentPlan, limit
.getLimit());
currentPlan.insertBetween(prepredecessor, newLimit, pred);
}
}
} else if (pred instanceof LOSort) {
LOSort sort = (LOSort) pred;
if (sort.getLimit() == -1)
sort.setLimit(limit.getLimit());
else
sort.setLimit(sort.getLimit() < limit.getLimit() ? sort
.getLimit() : limit.getLimit());
// remove the limit
currentPlan.removeAndReconnect(limit);
} else if (pred instanceof LOLimit) {
// Limit is merged into another LOLimit
LOLimit beforeLimit = (LOLimit) pred;
beforeLimit
.setLimit(beforeLimit.getLimit() < limit.getLimit() ? beforeLimit
.getLimit()
: limit.getLimit());
// remove the limit
currentPlan.removeAndReconnect(limit);
} else if (pred instanceof LOSplitOutput) {