/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.pig.newplan.logical.rules;

import java.util.Iterator;
import java.util.List;

import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.util.Pair;
import org.apache.pig.newplan.Operator;
import org.apache.pig.newplan.OperatorPlan;
import org.apache.pig.newplan.OperatorSubPlan;
import org.apache.pig.newplan.logical.expression.AndExpression;
import org.apache.pig.newplan.logical.expression.LogicalExpression;
import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan;
import org.apache.pig.newplan.logical.expression.ProjectExpression;
import org.apache.pig.newplan.logical.relational.LOFilter;
import org.apache.pig.newplan.logical.relational.LogicalPlan;
import org.apache.pig.newplan.logical.relational.LogicalRelationalOperator;
import org.apache.pig.newplan.optimizer.Rule;
import org.apache.pig.newplan.optimizer.Transformer;

public class SplitFilter extends Rule {    

    public SplitFilter(String n) {
        super(n, false);       
    }

    @Override
    public Transformer getNewTransformer() {        
        return new SplitFilterTransformer();
    }

    public class SplitFilterTransformer extends Transformer {
        private OperatorSubPlan subPlan;

        @Override
        public boolean check(OperatorPlan matched) throws FrontendException {
            LOFilter filter = (LOFilter)matched.getSources().get(0);
            LogicalExpressionPlan cond = filter.getFilterPlan();
            LogicalExpression root = (LogicalExpression) cond.getSources().get(0);
            if (root instanceof AndExpression && currentPlan.getSoftLinkPredecessors(filter)==null) {
                return true;
            }
            
            return false;
        }

        @Override
        public void transform(OperatorPlan matched) throws FrontendException {
            subPlan = new OperatorSubPlan(currentPlan);
            
            // split one LOFilter into 2 by "AND"
            LOFilter filter = (LOFilter)matched.getSources().get(0);
            LogicalExpressionPlan cond = filter.getFilterPlan();
            LogicalExpression root = (LogicalExpression) cond.getSources().get(0);
            if (!(root instanceof AndExpression)) {
                return;
            }
            LogicalExpressionPlan op1 = new LogicalExpressionPlan();
            op1.add((LogicalExpression)cond.getSuccessors(root).get(0));
            fillSubPlan(cond, op1, (LogicalExpression)cond.getSuccessors(root).get(0));
            
            LogicalExpressionPlan op2 = new LogicalExpressionPlan();
            op2.add((LogicalExpression)cond.getSuccessors(root).get(1));
            fillSubPlan(cond, op2, (LogicalExpression)cond.getSuccessors(root).get(1));
            
            filter.setFilterPlan(op1);
            LOFilter filter2 = new LOFilter((LogicalPlan)currentPlan, op2);
            currentPlan.add(filter2);
            
            Operator succed = null;
            List<Operator> succeds = currentPlan.getSuccessors(filter);
            if (succeds != null) {
                succed = succeds.get(0);
                subPlan.add(succed);
                currentPlan.insertBetween(filter, filter2, succed);
            } else {
                currentPlan.connect(filter, 0, filter2, 0); 
            }
            
            subPlan.add(filter);
            subPlan.add(filter2);
            Iterator<Operator> iter = filter2.getFilterPlan().getOperators();
            while (iter.hasNext()) {
                Operator oper = iter.next();
                if (oper instanceof ProjectExpression) {
                    ((ProjectExpression)oper).setAttachedRelationalOp(filter2);
                }
            }
        }
        
        @Override
        public OperatorPlan reportChanges() {
            return subPlan;
        }
        
        private void fillSubPlan(OperatorPlan origPlan, 
                OperatorPlan subPlan, Operator startOp) throws FrontendException {
                       
            List<Operator> l = origPlan.getSuccessors(startOp);
            if (l != null) {
                for(Operator le: l) {
                    subPlan.add(le);
                    subPlan.connect(startOp, le);
                    fillSubPlan(origPlan, subPlan, le);
                }            
            }
        }

    }

    @Override
    protected OperatorPlan buildPattern() {        
        // the pattern that this rule looks for
        // is filter
        LogicalPlan plan = new LogicalPlan();      
        LogicalRelationalOperator op2 = new LOFilter(plan);
        plan.add(op2);
        
        return plan;
    }
}