/*
 * JBoss, Home of Professional Open Source.
 * See the COPYRIGHT.txt file distributed with this work for information
 * regarding copyright ownership.  Some portions may be licensed
 * to Red Hat, Inc. under one or more contributor license agreements.
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301 USA.
 * Modified by
 * @author  Data Virtuality GmbH
 * @date    $Date$
 *  
 */
// https://issues.jboss.org/browse/TEIID-3208

package org.teiid.query.optimizer.relational.rules;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

import org.teiid.api.exception.query.QueryMetadataException;
import org.teiid.api.exception.query.QueryPlannerException;
import org.teiid.core.TeiidComponentException;
import org.teiid.query.analysis.AnalysisRecord;
import org.teiid.query.metadata.QueryMetadataInterface;
import org.teiid.query.optimizer.capabilities.CapabilitiesFinder;
import org.teiid.query.optimizer.relational.OptimizerRule;
import org.teiid.query.optimizer.relational.RelationalPlanner;
import org.teiid.query.optimizer.relational.RuleStack;
import org.teiid.query.optimizer.relational.plantree.NodeConstants;
import org.teiid.query.optimizer.relational.plantree.NodeConstants.Info;
import org.teiid.query.optimizer.relational.plantree.NodeEditor;
import org.teiid.query.optimizer.relational.plantree.PlanNode;
import org.teiid.query.rewriter.QueryRewriter;
import org.teiid.query.sql.lang.CompareCriteria;
import org.teiid.query.sql.lang.Criteria;
import org.teiid.query.sql.lang.IsNullCriteria;
import org.teiid.query.sql.lang.JoinType;
import org.teiid.query.sql.symbol.ElementSymbol;
import org.teiid.query.sql.symbol.Expression;
import org.teiid.query.sql.symbol.Function;
import org.teiid.query.sql.symbol.GroupSymbol;
import org.teiid.query.sql.visitor.GroupsUsedByElementsVisitor;
import org.teiid.query.util.CommandContext;


/**
 * Pushes on criteria out of the on clause if possible.
 * 
 * If the join no longer contains criteria, it will be changed into a cross join.
 * 
 * Upon a successful push, RulePushSelectCriteria will be run again.
 */
public final class RulePushNonJoinCriteria implements OptimizerRule {
	
	private boolean firstRun = true;
	
	public RulePushNonJoinCriteria(boolean firstRun) {
		this.firstRun = firstRun;
	}

	/**
	 * Execute the rule as described in the class comments.
	 * @param plan Incoming query plan, may be modified during method and may be returned from method
	 * @param metadata Metadata source
	 * @param rules Rules from optimizer rule stack, may be manipulated during method
	 * @return Updated query plan if rule fired, else original query plan
	 */
	public PlanNode execute(PlanNode plan, QueryMetadataInterface metadata, CapabilitiesFinder capFinder, RuleStack rules, AnalysisRecord analysisRecord, CommandContext context)
		throws QueryPlannerException, QueryMetadataException, TeiidComponentException {

        boolean treeChanged = false;
        boolean removeCopiedFlag = false;
        boolean pushRuleRaiseNull = false;
        
        for (PlanNode node : NodeEditor.findAllNodes(plan, NodeConstants.Types.JOIN)) {
            List criteria = (List)node.getProperty(NodeConstants.Info.JOIN_CRITERIA);
                        
            JoinType joinType = (JoinType)node.getProperty(NodeConstants.Info.JOIN_TYPE);
            
            //criteria cannot be pushed out of a full outer join clause
            if (joinType == JoinType.JOIN_FULL_OUTER || joinType == JoinType.JOIN_CROSS) {
                continue;
            }
            
            Iterator crits = criteria.iterator();
            while (crits.hasNext()) {
                Criteria crit = (Criteria)crits.next();
                                
                //special case handling for true/false criteria
                if (crit.equals(QueryRewriter.FALSE_CRITERIA) || crit.equals(QueryRewriter.UNKNOWN_CRITERIA)) {
                    if (joinType == JoinType.JOIN_INNER) {
                        FrameUtil.replaceWithNullNode(node);
                    } else {
                        //must be a left or right outer join, replace the inner side with null  
                        FrameUtil.replaceWithNullNode(JoinUtil.getInnerSideJoinNodes(node)[0]);
                        removeCopiedFlag = true;
                    }
                    //since a null node has been created, raise it to its highest point
                    pushRuleRaiseNull = true;
                    treeChanged = true;
                    break;
                } else if (crit.equals(QueryRewriter.TRUE_CRITERIA)) {
                    crits.remove();
                    break;
                }
                
                if (pushCriteria(node, crit, crits, metadata)) {
                    treeChanged = true;
                }
            }
            
            //degrade the join if there is no criteria left
            if (criteria.isEmpty() && joinType == JoinType.JOIN_INNER) {
                node.setProperty(NodeConstants.Info.JOIN_TYPE, JoinType.JOIN_CROSS);
                if (node.getProperty(NodeConstants.Info.ORIGINAL_JOIN_TYPE) == null)
                    node.setProperty(NodeConstants.Info.ORIGINAL_JOIN_TYPE, joinType);
	            node.setProperty(NodeConstants.Info.IS_IMPLICIT_CROSS_JOIN, true);
                treeChanged = true;
            }
             
            if (removeCopiedFlag) {
                //allow the criteria above the join to be eligible for pushing and copying
                PlanNode parent = node.getParent();
                while (parent != null && parent.getType() == NodeConstants.Types.SELECT) {
                    parent.setProperty(NodeConstants.Info.IS_COPIED, Boolean.FALSE);
                    parent = parent.getParent();
                }
            }            
        }
        
        if (treeChanged) {
            rules.push(RuleConstants.PUSH_SELECT_CRITERIA);
        }
        
        if (pushRuleRaiseNull) {
            rules.push(RuleConstants.RAISE_NULL);
        }

		return plan;
	}
    
    /** 
     * True if the criteria is pushed.
     * 
     * It's possible to push to the inner side of the join if the new criteria node
     * originates there
     * 
     * @param joinNode
     * @param tgtCrit
     * @param metadata 
     * @return
     */
    private boolean pushCriteria(PlanNode joinNode,
                                  Criteria tgtCrit, Iterator iter, QueryMetadataInterface metadata) {
        PlanNode newCritNode = RelationalPlanner.createSelectNode(tgtCrit, false);
        
        Set<GroupSymbol> groups = newCritNode.getGroups();
        
        PlanNode[] innerJoinNodes = JoinUtil.getInnerSideJoinNodes(joinNode);

        boolean pushed = false;

        for (int i = 0; i < innerJoinNodes.length; i++) {
            if (FrameUtil.findOriginatingNode(innerJoinNodes[i], groups) != null) {
                if (pushed) {
                    //create a new copy since the old one has been used
                    newCritNode = RelationalPlanner.createSelectNode(tgtCrit, false);
                }
                innerJoinNodes[i].addAsParent(newCritNode);
                pushed = true;
            }
        }
        
        if (pushed) {
        	iter.remove();
        } else if (firstRun && tgtCrit instanceof CompareCriteria) {
            CompareCriteria crit = (CompareCriteria) tgtCrit;
            
            Expression leftExpr = crit.getLeftExpression();
            Expression rightExpr = crit.getRightExpression();
                        
        	for (int i = 0; i < innerJoinNodes.length; i++) {
        		PlanNode node = FrameUtil.findJoinSourceNode(innerJoinNodes[i]);
        		
        		boolean outer = false;
        		for (PlanNode child : NodeEditor.findAllNodes(node, NodeConstants.Types.JOIN)) {
        			if (((JoinType)child.getProperty(Info.JOIN_TYPE)).isOuter()) {
        				outer = true;
        				break;
        			}
        		}
        		
        		if (!outer) {
        			continue;
        		}
        		
                Set<GroupSymbol> leftExprGroups = GroupsUsedByElementsVisitor.getGroups(leftExpr);            
                Set<GroupSymbol> rightExprGroups = GroupsUsedByElementsVisitor.getGroups(rightExpr);

                ArrayList<ElementSymbol> notNull = new ArrayList<ElementSymbol>(2);
                
        		if (node.getGroups().containsAll(leftExprGroups)) {
        			collectNotNull(leftExpr, notNull);
        		} else if (node.getGroups().containsAll(rightExprGroups)) {
        			collectNotNull(rightExpr, notNull);
        		}
        		
        		if (!notNull.isEmpty()) {
        			pushed = true;
        			for (ElementSymbol es : notNull) {
        				IsNullCriteria inc = new IsNullCriteria(es);
        				inc.setNegated(true);
        				PlanNode notNullCrit = RelationalPlanner.createSelectNode(inc, false);
        				notNullCrit.setProperty(NodeConstants.Info.IS_TEMPORARY, true);
        				innerJoinNodes[i].addAsParent(notNullCrit);
        			}
        		}
        	}
        }
        
        return pushed;
    }

	private void collectNotNull(Expression leftExpr,
			ArrayList<ElementSymbol> notNull) {
		if (leftExpr instanceof ElementSymbol) {
			notNull.add((ElementSymbol)leftExpr);
		} else if (leftExpr instanceof Function) {
			Function f = (Function)leftExpr;
			if (!f.getFunctionDescriptor().isNullDependent()) {
				for (Expression arg : f.getArgs()) {
					collectNotNull(arg, notNull);
				}
			}
		}
	}

    public String toString() {
		return "PushNonJoinCriteria"; //$NON-NLS-1$
	}

}
