}
@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);
List<Operator> softPrepredecessors=null;
// get a clone of softPrepredecessors to avoid ConcurrentModificationException
if (currentPlan.getSoftLinkPredecessors(limit)!=null) {
softPrepredecessors=new ArrayList<Operator>(
currentPlan.getSoftLinkPredecessors(limit));
}
if (softPrepredecessors!=null) {
for (Operator op : softPrepredecessors) {
currentPlan.removeSoftLink(op, limit);
}
}
currentPlan.removeAndReconnect(limit);
currentPlan.insertBetween(prepredecessor, limit, pred);
if (softPrepredecessors!=null) {
for (Operator op : softPrepredecessors) {
currentPlan.createSoftLink(op, limit);
}
}
} else if (limit.getLimitPlan() == null) {
// TODO selectively enable optimizations for variable limit
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 LOLoad) {
// Push limit to load
LOLoad load = (LOLoad) pred;
if (load.getLimit() == -1)
load.setLimit(limit.getLimit());
else
load.setLimit(load.getLimit() < limit.getLimit() ? load
.getLimit() : limit.getLimit());
} 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) {