class JoinFilterPushDown extends LogicalOperatorTransform {

    @Override
    public OperatorNode<SequenceOperator> visitSequenceOperator(OperatorNode<SequenceOperator> node) {
        // if this is a FILTER AND it contains a JOIN (perhaps with some other transforms in the way)
        //   AND the filter contains only references to the left side of the join
        if (node.getOperator() != SequenceOperator.FILTER) {
            return super.visitSequenceOperator(node);
        }
        // we have a FILTER, see if there's a JOIN underneath
        OperatorNode<SequenceOperator> target = node.getArgument(0);
        OperatorNode<ExpressionOperator> filter = node.getArgument(1);
        // It has to be *directly* underneath due to the way logical operands are constructed by the current parser
        // there may be a stack of JOINs, but we can attack each one in sequence

        if (target.getOperator() == SequenceOperator.JOIN || target.getOperator() == SequenceOperator.LEFT_JOIN) {
            List<OperatorNode<ExpressionOperator>> top = Lists.newArrayList();
            List<OperatorNode<ExpressionOperator>> leftFilter = Lists.newArrayList();
            List<OperatorNode<ExpressionOperator>> rightFilter = Lists.newArrayList();
            OperatorNode<SequenceOperator> leftSide = target.getArgument(0);
            OperatorNode<SequenceOperator> rightSide = target.getArgument(1);
            OperatorNode<ExpressionOperator> joinExpr = target.getArgument(2);
            if (filter.getOperator() == ExpressionOperator.AND) {
                flatten(top, (List<OperatorNode<ExpressionOperator>>) filter.getArgument(0));
            } else {
                top.add(filter);
            }
            Iterator<OperatorNode<ExpressionOperator>> topIterator = top.iterator();
            while (topIterator.hasNext()) {
                OperatorNode<ExpressionOperator> clause = topIterator.next();
                Set<String> left = findSources(leftSide);
                Set<String> right = findSources(rightSide);
                Set<String> referencedFilter = findReferencedSources(clause);
                boolean useLeft = !Sets.intersection(referencedFilter, left).isEmpty();
                boolean useRight = !Sets.intersection(referencedFilter, right).isEmpty();
                if (useLeft && useRight) {
                    // can't do anything
                } else if (useLeft) {
                    leftFilter.add(clause);
                    topIterator.remove();
                } else if (useRight) {
                    rightFilter.add(clause);
                    topIterator.remove();
                }
            }
            OperatorNode<SequenceOperator> result = node;
            if (rightFilter.size() > 0) {
                rightSide = visitSequenceOperator(new MergeFilters().visitSequenceOperator(OperatorNode.create(node.getLocation(), SequenceOperator.FILTER, rightSide, createFilter(rightFilter))));
            }
            if (leftFilter.size() > 0) {
                leftSide = visitSequenceOperator(new MergeFilters().visitSequenceOperator(OperatorNode.create(node.getLocation(), SequenceOperator.FILTER, leftSide, createFilter(leftFilter))));
            }
            if (rightFilter.size() > 0 || leftFilter.size() > 0) {
                result = OperatorNode.create(target.getLocation(), target.getAnnotations(), target.getOperator(),
                        leftSide, rightSide, joinExpr);
                if (top.size() > 0) {
                    result = OperatorNode.create(node.getLocation(), node.getAnnotations(), node.getOperator(), result, createFilter(top));
                } else {
                    for (Map.Entry<String, Object> e : node.getAnnotations().entrySet()) {
                        result.putAnnotation(e.getKey(), e.getValue());
                    }
                }
            }
            return super.visitSequenceOperator(result);
        }
        return node;
    }

}