/*
 * Copyright (C) 2016-2022 ActionTech.
 * License: http://www.gnu.org/licenses/gpl.html GPL version 2 or higher.
 */

package com.actiontech.dble.plan.optimizer;

import com.actiontech.dble.DbleServer;
import com.actiontech.dble.config.ErrorCode;
import com.actiontech.dble.config.model.sharding.table.ERTable;
import com.actiontech.dble.plan.NamedField;
import com.actiontech.dble.plan.common.exception.MySQLOutPutException;
import com.actiontech.dble.plan.common.item.Item;
import com.actiontech.dble.plan.common.item.ItemField;
import com.actiontech.dble.plan.common.item.function.ItemFunc;
import com.actiontech.dble.plan.common.item.function.operator.cmpfunc.ItemFuncEqual;
import com.actiontech.dble.plan.node.JoinNode;
import com.actiontech.dble.plan.node.PlanNode;
import com.actiontech.dble.plan.node.TableNode;
import com.actiontech.dble.plan.util.FilterUtils;
import com.actiontech.dble.plan.util.PlanUtil;
import com.actiontech.dble.route.parser.util.Pair;
import com.actiontech.dble.util.StringUtil;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.*;
import java.util.Map.Entry;
import java.util.function.Predicate;

public class JoinChooser {
    private static final Logger LOGGER = LogManager.getLogger(JoinChooser.class);
    private final List<JoinRelations> joinRelations = new LinkedList<>();
    private final List<PlanNode> joinUnits = new ArrayList<>();
    private final Map<PlanNode, JoinRelationDag> dagNodes = new HashMap<>();
    private final Map<ERTable, Set<ERTable>> erRelations;
    private final int charsetIndex;
    private final JoinNode orgNode;
    private final List<Item> otherJoinOns = new ArrayList<>();
    @Nonnull
    private final HintPlanInfo hintPlanInfo;
    private final Comparator<JoinRelationDag> defaultCmp = (o1, o2) -> {
        if (o1.relations.erRelationLst.size() > 0 && o2.relations.erRelationLst.size() > 0) {
            if (o1.relations.isInner == o2.relations.isInner) { // both er，both inner or not inner
                return 0;
            } else if (o1.relations.isInner) { // both er，o1 inner,o2  notinner
                return -1;
            } else { //if (o2.relations.isInner) { both er，o1 not inner,o2 inner
                return 1;
            }
        } else if (o1.relations.erRelationLst.size() > 0) { // if o2 is not ER join, o1 is ER join, o1<o2
            return -1;
        } else if (o2.relations.erRelationLst.size() > 0) { // if o1 is not ER join, o2 is ER join, o1>o2
            return 1;
        } else {
            // both o1,o2 are not ER join, global table should be first
            boolean o1Global = o1.node.getUnGlobalTableCount() == 0;
            boolean o2Global = o2.node.getUnGlobalTableCount() == 0;
            if (o1Global == o2Global) {
                if (o1.relations.isInner == o2.relations.isInner) { // both er，both inner or not inner
                    return 0;
                } else if (o1.relations.isInner) { // both er，o1 inner,o2  notinner
                    return -1;
                } else { //if (o2.relations.isInner) { both er，o1 not inner,o2 inner
                    return 1;
                }
            } else if (o1Global) {
                return -1;
            } else { // if (o2Global) {
                return 1;
            }
        }
    };

    public JoinChooser(JoinNode qtn, Map<ERTable, Set<ERTable>> erRelations, @Nonnull HintPlanInfo hintPlanInfo) {
        this.orgNode = qtn;
        this.erRelations = erRelations;
        this.charsetIndex = orgNode.getCharsetIndex();
        this.hintPlanInfo = hintPlanInfo;
    }

    public JoinChooser(JoinNode qtn, @Nonnull HintPlanInfo hintPlanInfo) {
        this(qtn, DbleServer.getInstance().getConfig().getErRelations(), hintPlanInfo);
    }

    public JoinNode optimize() {
        if (erRelations == null && hintPlanInfo.isZeroNode()) {
            return orgNode;
        }
        return innerJoinOptimizer();
    }


    /**
     * inner join's ER, rebuild inner join's unit
     */
    private JoinNode innerJoinOptimizer() {
        initJoinUnits(orgNode);
        if (joinUnits.size() == 1) {
            return orgNode;
        }
        JoinNode relationJoin = null;
        if (initNodeRelations(orgNode) && joinRelations.size() > 0) {
            //make DAG
            JoinRelationDag root;
            try {
                root = initJoinRelationDag();
            } catch (OptimizeException e) {
                LOGGER.debug("Join order of  sql  doesn't support to be  optimized. Because {}. The sql is [{}]", e.getMessage(), orgNode);
                if (!hintPlanInfo.isZeroNode()) {
                    throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "we don't support optimize this sql use hints, because " + e.getMessage());
                } else {
                    return orgNode;
                }
            }
            leftCartesianNodes();

            //if custom ,check plan can Follow the rules：Topological Sorting of dag, CartesianNodes
            if (!hintPlanInfo.isZeroNode()) {
                relationJoin = joinWithHint(root);
            } else {
                // use auto plan
                relationJoin = makeBNFJoin(root, defaultCmp);
            }
        }
        // no relation join
        if (relationJoin == null) {
            if (!hintPlanInfo.isZeroNode()) {
                throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "we don't support optimize this sql use hints yet. Maybe this sql contains 'multi right join' or 'cartesian with relation' or 'subquery'.");
            }
            return orgNode;
        }

        // others' node is the join units which can not optimize, just merge them
        JoinNode ret = makeJoinWithCartesianNode(relationJoin);
        ret.setDistinct(orgNode.isDistinct());
        ret.setOrderBys(orgNode.getOrderBys());
        ret.setGroupBys(orgNode.getGroupBys());
        ret.select(orgNode.getColumnsSelected());
        ret.setLimitFrom(orgNode.getLimitFrom());
        ret.setLimitTo(orgNode.getLimitTo());
        ret.having(orgNode.getHavingFilter());
        //ret.setWhereFilter(orgNode.getWhereFilter());
        ret.setWhereFilter(FilterUtils.and(orgNode.getWhereFilter(), FilterUtils.and(otherJoinOns)));
        ret.setAlias(orgNode.getAlias());
        ret.setWithSubQuery(orgNode.isWithSubQuery());
        ret.setContainsSubQuery(orgNode.isContainsSubQuery());
        ret.getSubQueries().addAll(orgNode.getSubQueries());
        ret.setSql(orgNode.getSql());
        ret.setUpFields();
        return ret;
    }


    private JoinNode makeJoinWithCartesianNode(JoinNode node) {
        JoinNode left = node;
        for (PlanNode right : joinUnits) {
            left = new JoinNode(left, right, charsetIndex);
        }
        return left;
    }

    private void leftCartesianNodes() {
        if (joinUnits.size() > dagNodes.size()) {
            //Cartesian Product node
            joinUnits.removeIf(dagNodes::containsKey);
        } else {
            joinUnits.clear();
        }
    }

    @Nonnull
    private JoinRelationDag initJoinRelationDag() {
        JoinRelationDag root = createFirstNode();
        for (int i = 1; i < joinRelations.size(); i++) {
            root = addNodeToDag(root, joinRelations.get(i));
        }
        return root;
    }

    @Nonnull
    private JoinRelationDag createFirstNode() {
        JoinRelations firstRelation = joinRelations.get(0);
        // firstRelation should only have one left nodes
        if (firstRelation.prefixNodes.size() != 1) {
            throw new OptimizeException("firstRelation have " + firstRelation.prefixNodes.size() + " prefix node");
        }

        JoinRelationDag root = createDag(firstRelation.leftNodes.iterator().next());
        JoinRelationDag right = createDag(firstRelation, firstRelation.isInner);
        root.rightNodes.add(right);
        right.degree++;
        right.leftNodes.add(root);
        return root;
    }


    private boolean isSameNode(String hintNodeName, JoinRelationDag dagNode) {
        final PlanNode node = dagNode.node;
        if (hintNodeName == null) {
            return false;
        }
        final String unitName = getUnitName(node);
        if (unitName == null) {
            throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "can't optimize this sql, try to set alias for every tables in sql. Related nodes: " + node);
        }
        return Objects.equals(hintNodeName, unitName);


    }

    @Nullable
    private static String getUnitName(PlanNode node) {
        if (node.getAlias() != null) {
            return node.getAlias();
        } else if (node instanceof TableNode) {
            return ((TableNode) node).getTableName();
        }
        return null;

    }

    private JoinNode joinWithHint(JoinRelationDag root) {
        validateHint();
        if (dagNodes.size() != hintPlanInfo.nodeSize()) {
            throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "hint size " + hintPlanInfo.nodeSize() + " not equals to plan node size " + dagNodes.size() + ".");
        }
        if (root.degree != 0) {
            throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "exists any relations route to the root node: " + root);
        }
        final Iterator<String> hintIt = hintPlanInfo.getHintPlanNodeMap().keySet().stream().iterator();
        String nextHintNode = hintIt.next();

        {
            root = findNode(root, nextHintNode);
            if (root == null) {
                throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "no node match the root: " + nextHintNode);
            }
            if (!root.isFamilyInner) {
                throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "can't use '" + root + "' node for root. Because exists some left join relations point to this node. ");
            }
            flipRightNodeToLeft(root, null);
        }
        final JoinNodeBuilder joinNodeBuilder = new JoinNodeBuilder(root.node);
        Map<JoinRelationDag, DagLine> nextAccessDagNodes = new HashMap<>();
        nextAccessDagNodes.put(root, new DagLine(null, root));

        traversal:
        while (!nextAccessDagNodes.isEmpty()) {
            final Iterator<DagLine> it = nextAccessDagNodes.values().iterator();
            while (it.hasNext()) {
                final DagLine couldAccessDagNode = it.next();
                final JoinRelationDag currentNode = couldAccessDagNode.getTargetNode();
                final JoinRelationDag prevNode = couldAccessDagNode.getPrevNode();
                if (currentNode.visited) {
                    it.remove();
                    continue;
                }

                if (!isSameNode(nextHintNode, currentNode)) {
                    continue;
                }
                /*
                two reason
                1.flip may cause node degree increased.
                2.put rightNode ignore the degree.
                 */
                if (currentNode.degree != 0) {
                    //flip only if needed. Unnecessary flip may cause degree num of other nodes increased and become inaccessible.
                    final boolean flipSuccess = flipRightNodeToLeft(currentNode, prevNode);
                    if (!flipSuccess) {
                        throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "can't use this hints,because exists some left join relations point to node: " + currentNode);
                    }
                    if (currentNode.degree != 0) {
                        //In theory it won't happen
                        throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "can't parse use this hints. " + hintPlanInfo);
                    }
                }
                currentNode.markVisited();
                //matched
                if (prevNode != null) {
                    joinNodeBuilder.appendNodeToRight(currentNode);
                }

                //prepare for next traversal.
                it.remove();
                if (!hintIt.hasNext()) {
                    for (JoinRelationDag targetNode : nextAccessDagNodes.keySet()) {
                        if (!targetNode.visited) {
                            ////In theory it won't happen
                            throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "can't traversal all node use this hint." + hintPlanInfo);
                        }
                    }
                    break traversal;
                }
                nextHintNode = hintIt.next();
                for (JoinRelationDag rightNode : currentNode.rightNodes) {
                    if (rightNode.visited) {
                        continue;
                    }
                    --rightNode.degree;

                    /*
                    ignore the degree and delay the degree comparison .because the rightNode may need flip.
                    e.g.
                    select * from a inner join b on a.id=b.id inner join c on b.id=c.id inner join d on c.id=d.id where b.id=1
                    the join order is a->b->c->d and the hint is c->d->b->a.
                    we choose 'c' as root, then dag like this (c->d) & (c->b<-a)
                    when visited c. b is one of rightNodes of c， and the degree of b is 2.
                     */
                    //                    if (rightNode.degree == 0) {
                    nextAccessDagNodes.put(rightNode, new DagLine(currentNode, rightNode));
                    //                    }

                }
                continue traversal;
            }
            //when no nextAccessDagNodes match the nextHintNode
            String finalNextHintNode = nextHintNode;
            final boolean nodeExist = isHintNodeExist(finalNextHintNode);
            if (!nodeExist) {
                throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "You are using wrong hint.The node '" + nextHintNode + "' doesn't exist.");
            } else {
                throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "You are using wrong hint. please check the node '" + nextHintNode + "',there are no previous nodes connect to it.");
            }


        }
        if (hintIt.hasNext()) {
            throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "can't traversal all node use this hint. please check near the node '" + nextHintNode + "',may be contain orphaned node.");
        }
        return joinNodeBuilder.build();
    }


    private boolean isHintNodeExist(String finalNextHintNodeName) {
        return dagNodes.values().stream().anyMatch(node ->
                isSameNode(finalNextHintNodeName, node)
        );
    }

    private void validateHint() {
        if (hintPlanInfo.erRelyOn()) {
            throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "The ER relation in the hint currently only supports when it exists in the headmost of hint.");
        }
    }

    private JoinRelationDag findNode(JoinRelationDag root, String hintNodeName) {
        if (isSameNode(hintNodeName, root)) {
            return root;
        } else {
            for (JoinRelationDag rightNode : root.rightNodes) {
                final JoinRelationDag node = findNode(rightNode, hintNodeName);
                if (node != null) {
                    return node;
                }
            }
            return null;
        }
    }

    private boolean flipRightNodeToLeft(JoinRelationDag currentNode, JoinRelationDag prevNode) {
        final JoinRelations relations = currentNode.relations;
        if (relations == null) {
            return true;
        }
        if (!relations.isInner) {
            return false;
        }
        final List<OneToOneJoinRelation> splitRelations = splitAndExchangeRelations(relations);

        final Iterator<JoinRelationDag> it = currentNode.leftNodes.iterator();
        while (it.hasNext()) {
            final JoinRelationDag oldLeft = it.next();
            if (oldLeft.visited || oldLeft == prevNode) {
                continue;
            }
            if (oldLeft.relations != null && !oldLeft.relations.isInner) {
                throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "some errors near the node '" + getUnitName(oldLeft.node) + "'. Because left join and inner join can't point to same node.");
            }
            it.remove();
            //currentNode move to new left.
            currentNode.rightNodes.add(oldLeft);
            currentNode.degree--;
            //old left move to right.
            oldLeft.degree++;
            oldLeft.leftNodes.add(currentNode);
            oldLeft.rightNodes.remove(currentNode);

            final Iterator<OneToOneJoinRelation> splitIt = splitRelations.iterator();
            while (splitIt.hasNext()) {
                final OneToOneJoinRelation splitRelation = splitIt.next();
                if (splitRelation.rightNode == oldLeft.node) {
                    oldLeft.relations = addRelations(oldLeft.relations, splitRelation);

                    if (currentNode.relations != null) {
                        subRelation(currentNode.relations, splitRelation);
                    }
                    splitIt.remove();
                }
            }

        }
        return true;
    }

    private JoinNode makeBNFJoin(JoinRelationDag root, Comparator<JoinRelationDag> joinCmp) {
        List<JoinRelationDag> zeroDegreeList = new ArrayList<>();
        for (JoinRelationDag tree : root.rightNodes) {
            if (--tree.degree == 0) {
                zeroDegreeList.add(tree);
            }
        }
        final JoinNodeBuilder joinNodeBuilder = new JoinNodeBuilder(root.node);
        while (!zeroDegreeList.isEmpty()) {
            zeroDegreeList.sort(joinCmp);
            // zeroDegreeList contains no er relations
            boolean cleanList = zeroDegreeList.get(0).relations.erRelationLst.size() == 0;
            Iterator<JoinRelationDag> zeroDegreeIterator = zeroDegreeList.iterator();
            List<JoinRelationDag> newZeroDegreeList = new ArrayList<>();
            while (zeroDegreeIterator.hasNext()) {
                JoinRelationDag rightNode = zeroDegreeIterator.next();
                if (cleanList || rightNode.relations.erRelationLst.size() > 0) {
                    zeroDegreeIterator.remove();
                    joinNodeBuilder.appendNodeToRight(rightNode);
                    for (JoinRelationDag tree : rightNode.rightNodes) {
                        if (--tree.degree == 0) {
                            newZeroDegreeList.add(tree);
                        }
                    }
                } else { // not have any er-node, other nodes will sort with new 0 degree nodes
                    break;
                }
            }
            zeroDegreeList.addAll(newZeroDegreeList);
        }
        return joinNodeBuilder.build();
    }


    private JoinRelationDag addNodeToDag(JoinRelationDag root, JoinRelations relations) {
        Set<JoinRelationDag> prefixDagNodes = new HashSet<>(relations.prefixNodes.size());
        boolean familyInner = relations.isInner;
        for (PlanNode prefixNode : relations.prefixNodes) {
            JoinRelationDag tmp = dagNodes.get(prefixNode);
            if (tmp == null) {
                // eg: select b.* from  a inner join  b on a.id=b.id , sharding2 inner join   sharding2_child  on sharding2_child.id=sharding2.id ;
                // maybe multi DAGs, or need merge DAGs, optimizer cost too much
                throw new OptimizeException("this sql contains cartesian with relation");
            } else {
                prefixDagNodes.add(tmp);
                if (familyInner && !tmp.isFamilyInner) {
                    familyInner = false;
                }
            }
        }


        if (!familyInner || prefixDagNodes.size() == 1 || relations.erRelationLst.size() == 0) {
            // 1.left join can not be optimizer
            // 2. familyInner only one prefixDagNode, no need optimizer
            // 3. all join filter are not er, no need change direction
            JoinRelationDag right = createDag(relations, familyInner);
            for (JoinRelationDag prefixDagNode : prefixDagNodes) {
                prefixDagNode.rightNodes.add(right);
                right.degree++;
            }
            right.leftNodes.addAll(prefixDagNodes);
            return root;
        } else {
            Set<PlanNode> toChangeParent = new HashSet<>();
            for (JoinRelation joinRelation : relations.erRelationLst) {
                toChangeParent.add(joinRelation.left.planNode);
            }
            return optimizerInnerJoinOtherFilter(createDag(relations.rightNode), relations, prefixDagNodes, toChangeParent);
        }
    }

    private JoinRelationDag optimizerInnerJoinOtherFilter(JoinRelationDag newLeft, JoinRelations relations, Set<JoinRelationDag> orgLefts, Set<PlanNode> toChangeParent) {
        //if (relations.leftNodes.size() < relations.prefixNodes.size()) {
        //    todo:a inner join b on (ab) inner join c on (bc,ab)
        //}

        //eg: a inner join b on (ab) inner join c on (bc,ac)
        //change direction to:c inner join b on (bc) inner join a on (ab,ac)
        if (orgLefts.size() == 0 && relations == null) { // root and not with new
            return newLeft;
        }
        for (JoinRelationDag orgLeft : orgLefts) {
            if (toChangeParent.contains(orgLeft.node)) {
                // inner join prefixNodes==left nodes
                Set<PlanNode> toChange = orgLeft.relations == null ? null : orgLeft.relations.prefixNodes;
                optimizerInnerJoinOtherFilter(orgLeft, orgLeft.relations, orgLeft.leftNodes, toChange);
            }
        }
        List<OneToOneJoinRelation> splitRelationLst = splitAndExchangeRelations(relations);
        for (OneToOneJoinRelation splitRelation : splitRelationLst) {
            JoinRelationDag oldLeft = null;
            for (JoinRelationDag orgLeft : orgLefts) {
                if (splitRelation.rightNode == orgLeft.node) {
                    oldLeft = orgLeft;
                    break;
                }
            }
            //change direction
            assert oldLeft != null;
            oldLeft.leftNodes.add(newLeft);
            oldLeft.rightNodes.remove(newLeft);
            oldLeft.degree++;
            oldLeft.relations = addRelations(oldLeft.relations, splitRelation);

            if (newLeft.relations != null) {
                subRelation(newLeft.relations, splitRelation);
            }
            newLeft.leftNodes.remove(oldLeft); // root remove nothing
            newLeft.rightNodes.add(oldLeft);
            if (newLeft.degree > 0) { //root is 0
                newLeft.degree--;
            }
        }
        return newLeft;
    }

    private void subRelation(JoinRelations a, OneToOneJoinRelation b) {
        if (a == null) {
            return;
        }
        final Predicate<JoinRelation> predicate = joinRelation -> (joinRelation.right.planNode == b.rightNode && joinRelation.left.planNode == b.leftNode) || (joinRelation.right.planNode == b.leftNode && joinRelation.left.planNode == b.rightNode);
        a.erRelationLst.removeIf(predicate);
        a.normalRelationLst.removeIf(predicate);
        a.init();
    }


    private JoinRelations addRelations(JoinRelations a, OneToOneJoinRelation b) {
        if (a == null) {
            return b.convertToJoinRelations();
        }
        a.erRelationLst.addAll(b.erRelationLst);
        a.normalRelationLst.addAll(b.normalRelationLst);
        a.init();
        return a;
    }

    private List<OneToOneJoinRelation> splitAndExchangeRelations(JoinRelations relations) {
        PlanNode newLeftNode = relations.rightNode;

        List<OneToOneJoinRelation> relationLst = new ArrayList<>();
        Map<PlanNode, List<JoinRelation>> nodeToNormalMap = new HashMap<>();
        for (JoinRelation joinRelation : relations.normalRelationLst) {
            joinRelation = joinRelation.exchange();
            List<JoinRelation> tmpNormalList = nodeToNormalMap.get(joinRelation.right.planNode);
            if (tmpNormalList == null) {
                tmpNormalList = new ArrayList<>();
            }
            tmpNormalList.add(joinRelation);
            nodeToNormalMap.put(joinRelation.right.planNode, tmpNormalList);
        }
        for (JoinRelation joinRelation : relations.erRelationLst) {
            joinRelation = joinRelation.exchange();
            List<JoinRelation> tmpErRelationLst = new ArrayList<>(1);
            tmpErRelationLst.add(joinRelation);
            List<JoinRelation> tmpNormalRelationLst = nodeToNormalMap.remove(joinRelation.right.planNode);
            if (tmpNormalRelationLst == null) {
                tmpNormalRelationLst = new ArrayList<>(0);
            }
            OneToOneJoinRelation nodeRelations = new OneToOneJoinRelation(tmpErRelationLst, tmpNormalRelationLst, joinRelation.right.planNode, newLeftNode);
            relationLst.add(nodeRelations);
        }
        for (Entry<PlanNode, List<JoinRelation>> entry : nodeToNormalMap.entrySet()) {
            OneToOneJoinRelation nodeRelations = new OneToOneJoinRelation(new ArrayList<>(0), entry.getValue(), entry.getKey(), newLeftNode);
            relationLst.add(nodeRelations);
        }
        return relationLst;
    }

    // find the smallest join units in node
    private void initJoinUnits(JoinNode node) {
        for (int index = 0; index < node.getChildren().size(); index++) {
            PlanNode child = node.getChildren().get(index);
            if (isUnit(child)) {
                child = JoinProcessor.optimize(child, new HintPlanInfo());
                node.getChildren().set(index, child);
                this.joinUnits.add(child);
            } else {
                initJoinUnits((JoinNode) child);
            }
        }
    }


    private boolean isUnit(PlanNode node) {
        return node.type() != PlanNode.PlanNodeType.JOIN || node.isWithSubQuery();
    }

    private boolean initNodeRelations(JoinNode joinNode) {
        for (PlanNode unit : joinUnits) {
            // is unit
            if (unit == joinNode) {
                return true;
            }
        }
        PlanNode right = joinNode.getChildren().get(1);
        if ((!isUnit(right)) && (right.type().equals(PlanNode.PlanNodeType.JOIN))) {
            right = JoinProcessor.optimize(right, new HintPlanInfo());
            joinNode.setRightNode(right);
            return false; //will be not optimize current join node
        }

        PlanNode left = joinNode.getChildren().get(0);
        if ((!isUnit(left)) && (left.type().equals(PlanNode.PlanNodeType.JOIN))) {
            if (!initNodeRelations((JoinNode) left)) {
                return false;
            }
        }

        Item otherFilter = joinNode.getOtherJoinOnFilter();
        PlanNode rightNode = joinNode.getRightNode();
        if (joinNode.getJoinFilter().size() > 0) {
            List<JoinRelation> erRelationLst = new ArrayList<>(1);
            List<JoinRelation> normalRelationLst = new ArrayList<>(1);
            Set<PlanNode> leftNodes = new HashSet<>(2);
            for (ItemFuncEqual filter : joinNode.getJoinFilter()) {
                JoinColumnInfo columnInfoLeft = initJoinColumnInfo(filter.arguments().get(0));
                JoinColumnInfo columnInfoRight = initJoinColumnInfo(filter.arguments().get(1));
                boolean isERJoin = isErRelation(columnInfoLeft.erTable, columnInfoRight.erTable);
                if (columnInfoLeft.planNode != rightNode && columnInfoRight.planNode != rightNode) {
                    //  now may not happen:a join b on a,b join c on c,b and a,b; the last a,b can be other filter
                    //todo: if (isERJoin) and inner join try optimizer later
                    otherFilter = FilterUtils.and(otherFilter, filter);
                    continue;
                }
                JoinRelation nodeRelation = new JoinRelation(columnInfoLeft, columnInfoRight, filter);
                if (isERJoin) {
                    erRelationLst.add(nodeRelation);
                } else {
                    normalRelationLst.add(nodeRelation);
                }
                leftNodes.add(columnInfoLeft.planNode);
            }

            JoinRelations nodeRelations;
            if (joinNode.isInnerJoin()) {
                otherJoinOns.add(otherFilter);
                nodeRelations = new JoinRelations(erRelationLst, normalRelationLst, rightNode, leftNodes);
            } else {
                nodeRelations = new JoinRelations(erRelationLst, normalRelationLst, otherFilter, rightNode, leftNodes);
            }
            nodeRelations.init();
            joinRelations.add(nodeRelations);
        } else {
            if (joinNode.isInnerJoin()) {
                if (!hintPlanInfo.isZeroNode()) {
                    throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "we don't support optimize this sql use hints yet. Because this sql contains 'cartesian with relation'.");
                }
                otherJoinOns.add(otherFilter);
            } else {
                Set<PlanNode> leftNodes = new HashSet<>();
                getLeftNodes(joinNode.getLeftNode(), leftNodes);
                JoinRelations nodeRelations = new JoinRelations(new ArrayList<>(0), new ArrayList<>(0), otherFilter, rightNode, leftNodes);
                nodeRelations.init();
                joinRelations.add(nodeRelations);
            }
        }
        return true;
    }

    private void getLeftNodes(PlanNode child, Set<PlanNode> leftNodes) {
        if ((!isUnit(child)) && (child.type().equals(PlanNode.PlanNodeType.JOIN))) {
            getLeftNodes(((JoinNode) child).getLeftNode(), leftNodes);
            getLeftNodes(((JoinNode) child).getRightNode(), leftNodes);
        } else {
            leftNodes.add(child);
        }
    }

    private JoinColumnInfo initJoinColumnInfo(Item key) {
        JoinColumnInfo columnInfo = new JoinColumnInfo(key);
        for (PlanNode planNode : joinUnits) {
            Item tmpSel = nodeHasSelectTable(planNode, columnInfo.key);
            if (tmpSel != null) {
                columnInfo.planNode = planNode;
                columnInfo.erTable = getERKey(planNode, tmpSel);
                return columnInfo;
            }
        }
        throw new MySQLOutPutException(ErrorCode.ER_OPTIMIZER, "", "can not find table of:" + key);
    }


    private boolean onlyContainsOneTable(PlanNode queryNode) {
        PlanNode child = queryNode.getChildren().get(0);
        if (child.type() == PlanNode.PlanNodeType.TABLE) {
            return true;
        } else if (child.type() == PlanNode.PlanNodeType.QUERY) {
            return onlyContainsOneTable(child);
        } else {
            return false;
        }
    }

    private ERTable getERKey(PlanNode tn, Item column) {
        if (!(column instanceof ItemField))
            return null;
        Pair<TableNode, ItemField> pair = null;
        if (tn.type() == PlanNode.PlanNodeType.QUERY && onlyContainsOneTable(tn)) {
            pair = PlanUtil.findColumnInTableLeaf((ItemField) column, tn);
        } else if (tn.type() != PlanNode.PlanNodeType.TABLE && !PlanUtil.isERNode(tn)) {
            return null;
        } else {
            pair = PlanUtil.findColumnInTableLeaf((ItemField) column, tn);
        }
        if (pair == null)
            return null;
        TableNode tableNode = pair.getKey();
        ItemField col = pair.getValue();
        ERTable erTable = new ERTable(tableNode.getSchema(), tableNode.getPureName(), col.getItemName());
        if (tn.type() == PlanNode.PlanNodeType.TABLE || tn.type() == PlanNode.PlanNodeType.QUERY) {
            return erTable;
        } else {
            List<ERTable> erList = ((JoinNode) tn).getERkeys();
            for (ERTable cerKey : erList) {
                if (isErRelation(erTable, cerKey))
                    return erTable;
            }
            return null;
        }
    }


    private Item nodeHasSelectTable(PlanNode child, Item sel) {
        if (sel instanceof ItemField) {
            return nodeHasColumn(child, (ItemField) sel);
        } else if (sel.canValued()) {
            return sel;
        } else if (sel.type().equals(Item.ItemType.SUM_FUNC_ITEM)) {
            return null;
        } else {
            ItemFunc fcopy = (ItemFunc) sel.cloneStruct();
            for (int index = 0; index < fcopy.getArgCount(); index++) {
                Item arg = fcopy.arguments().get(index);
                Item argSel = nodeHasSelectTable(child, arg);
                if (argSel == null)
                    return null;
                else
                    fcopy.arguments().set(index, argSel);
            }
            PlanUtil.refreshReferTables(fcopy);
            fcopy.setPushDownName(null);
            return fcopy;
        }
    }

    private Item nodeHasColumn(PlanNode child, ItemField col) {
        String colName = col.getItemName();
        if (StringUtil.isEmpty(col.getTableName())) {
            for (Entry<NamedField, Item> entry : child.getOuterFields().entrySet()) {
                if (StringUtil.equalsIgnoreCase(colName, entry.getKey().getName())) {
                    return entry.getValue();
                }
            }
        } else {
            String table = col.getTableName();
            if (child.getAlias() == null) {
                for (Entry<NamedField, Item> entry : child.getOuterFields().entrySet()) {
                    if (StringUtil.equals(table, entry.getKey().getTable()) &&
                            StringUtil.equalsIgnoreCase(colName, entry.getKey().getName())) {
                        return entry.getValue();
                    }
                }
            } else {
                if (DbleServer.getInstance().getSystemVariables().isLowerCaseTableNames()) {
                    if (!StringUtil.equalsIgnoreCase(table, child.getAlias()))
                        return null;
                } else {
                    if (!StringUtil.equals(table, child.getAlias()))
                        return null;
                }
                for (Entry<NamedField, Item> entry : child.getOuterFields().entrySet()) {
                    if (StringUtil.equalsIgnoreCase(colName, entry.getKey().getName())) {
                        return entry.getValue();
                    }
                }
            }
        }
        return null;
    }

    private boolean isErRelation(ERTable er0, ERTable er1) {
        if (er0 == null || er1 == null) {
            return false;
        }
        Set<ERTable> erList = erRelations.get(er0);
        if (erList == null) {
            return false;
        }
        return erList.contains(er1);
    }

    public JoinRelationDag createDag(PlanNode node) {
        JoinRelationDag dagNode = dagNodes.get(node);
        if (dagNode == null) {
            dagNode = new JoinRelationDag(node);
            dagNodes.put(node, dagNode);
        }
        return dagNode;
    }


    public JoinRelationDag createDag(JoinRelations relations, boolean isFamilyInner) {
        JoinRelationDag dagNode = dagNodes.get(relations.rightNode);
        if (dagNode == null) {
            dagNode = new JoinRelationDag(relations, isFamilyInner);
            dagNodes.put(relations.rightNode, dagNode);
        }
        return dagNode;
    }

    private final class JoinNodeBuilder {
        private final PlanNode rootNode;
        private JoinNode result;

        private JoinNodeBuilder(PlanNode rootNode) {
            this.rootNode = rootNode;
        }

        private void appendNodeToRight(JoinRelationDag rightNodeOfJoin) {
            boolean leftIsNull = result == null;
            JoinNode joinNode = new JoinNode(leftIsNull ? rootNode : result, rightNodeOfJoin.node, charsetIndex);
            if (!rightNodeOfJoin.relations.isInner) {
                joinNode.setLeftOuterJoin();
            }
            List<ItemFuncEqual> filters = new ArrayList<>();
            for (JoinRelation joinRelation : rightNodeOfJoin.relations.erRelationLst) {
                filters.add(joinRelation.filter);
                if (leftIsNull) {
                    joinNode.getERkeys().add(joinRelation.left.erTable);
                } else {
                    joinNode.getERkeys().addAll((result).getERkeys());
                }
            }
            for (JoinRelation joinRelation : rightNodeOfJoin.relations.normalRelationLst) {
                filters.add(joinRelation.filter);
            }
            joinNode.setJoinFilter(filters);
            joinNode.setOtherJoinOnFilter(rightNodeOfJoin.relations.otherFilter);
            this.result = joinNode;
        }

        public JoinNode build() {
            return result;
        }
    }

    private static final class JoinRelationDag {
        private final PlanNode node;
        private int degree = 0;
        private JoinRelations relations;
        private final Set<JoinRelationDag> rightNodes = new LinkedHashSet<>();
        private final Set<JoinRelationDag> leftNodes = new LinkedHashSet<>();
        private boolean isFamilyInner = true;
        private boolean visited = false;

        private JoinRelationDag(PlanNode node) {
            this.node = node;
            this.relations = null;
        }

        private JoinRelationDag(JoinRelations relations, boolean isFamilyInner) {
            this.node = relations.rightNode;
            this.relations = relations;
            this.isFamilyInner = isFamilyInner;
        }


        public void markVisited() {
            this.visited = true;
        }

        @Override
        public int hashCode() {
            return super.hashCode();
        }

        @Override
        public boolean equals(Object obj) {
            return super.equals(obj);
        }

        @Override
        public String toString() {
            return "{" +
                    "node=" + getUnitName(node) +
                    '}';
        }
    }


    private class JoinRelations {
        private final List<JoinRelation> erRelationLst;
        private final List<JoinRelation> normalRelationLst;
        private final boolean isInner;
        private final Item otherFilter;
        private final Set<PlanNode> leftNodes;
        private final PlanNode rightNode;
        private final Set<PlanNode> prefixNodes = new HashSet<>();

        JoinRelations(List<JoinRelation> erRelationLst, List<JoinRelation> normalRelationLst, Item otherFilter, PlanNode rightNode, Set<PlanNode> leftNodes) {
            this.erRelationLst = erRelationLst;
            this.normalRelationLst = normalRelationLst;
            this.rightNode = rightNode;
            this.leftNodes = leftNodes;
            this.isInner = false;
            this.otherFilter = otherFilter;
        }

        JoinRelations(List<JoinRelation> erRelationLst, List<JoinRelation> normalRelationLst, PlanNode rightNode, Set<PlanNode> leftNodes) {
            this.erRelationLst = erRelationLst;
            this.normalRelationLst = normalRelationLst;
            this.rightNode = rightNode;
            this.leftNodes = leftNodes;
            this.isInner = true;
            this.otherFilter = null;
        }

        void init() {
            prefixNodes.clear();
            prefixNodes.addAll(leftNodes);
            if (otherFilter != null && otherFilter.getReferTables() != null) {
                for (PlanNode planNode : joinUnits) {
                    if (planNode != rightNode) {
                        Item tmpSel = nodeHasSelectTable(planNode, otherFilter);
                        if (tmpSel != null) {
                            prefixNodes.add(planNode);
                        }
                    }
                }
            }
        }

        @Override
        public String toString() {
            return String.valueOf(Iterables.concat(normalRelationLst, erRelationLst));
        }

    }


    private class OneToOneJoinRelation {
        private final List<JoinRelation> erRelationLst;
        private final List<JoinRelation> normalRelationLst;
        private final PlanNode leftNode;
        private final PlanNode rightNode;


        OneToOneJoinRelation(List<JoinRelation> erRelationLst, List<JoinRelation> normalRelationLst, PlanNode rightNode, PlanNode leftNode) {
            this.erRelationLst = erRelationLst;
            this.normalRelationLst = normalRelationLst;
            this.rightNode = rightNode;
            this.leftNode = leftNode;
        }


        public JoinRelations convertToJoinRelations() {
            final JoinRelations ret = new JoinRelations(erRelationLst, normalRelationLst, rightNode, Sets.newHashSet(leftNode));
            ret.init();
            return ret;
        }

        @Override
        public String toString() {
            return getUnitName(leftNode) + " --> " + getUnitName(rightNode);
        }
    }

    private class JoinRelation {
        private final JoinColumnInfo left;
        private final JoinColumnInfo right;
        private final ItemFuncEqual filter;

        JoinRelation(JoinColumnInfo left, JoinColumnInfo right, ItemFuncEqual filter) {
            this.left = left;
            this.right = right;
            this.filter = filter;
        }

        private JoinRelation exchange() {
            return new JoinRelation(this.right, this.left, FilterUtils.equal(right.key, left.key, charsetIndex));
        }

        @Override
        public String toString() {
            return left + " --> " + right;
        }
    }

    /**
     * JoinColumnInfo
     *
     * @author ActionTech
     */
    private static class JoinColumnInfo {
        private Item key; // join on's on key
        private PlanNode planNode; // treenode of the joinColumn belong to
        private ERTable erTable; //  joinColumn is er ,if so,save th parentkey

        JoinColumnInfo(Item key) {
            this.key = key;
            planNode = null;
            erTable = null;
        }

        @Override
        public int hashCode() {
            int hash = this.key.getTableName().hashCode();
            hash = hash * 31 + this.key.getItemName().toLowerCase().hashCode();
            return hash;
        }

        @Override
        public boolean equals(Object o) {
            if (o == null)
                return false;
            if (o == this)
                return true;
            if (!(o instanceof JoinColumnInfo)) {
                return false;
            }
            JoinColumnInfo other = (JoinColumnInfo) o;
            if (this.key == null)
                return false;
            return StringUtil.equals(this.key.getTableName(), other.key.getTableName()) &&
                    StringUtil.equalsIgnoreCase(this.key.getItemName(), other.key.getItemName());
        }

        @Override
        public String toString() {
            return "key:" + key;
        }

        public Item getKey() {
            return key;
        }

        public void setKey(Item key) {
            this.key = key;
        }

        public PlanNode getPlanNode() {
            return planNode;
        }

        public void setPlanNode(PlanNode planNode) {
            this.planNode = planNode;
        }

        public ERTable getErTable() {
            return erTable;
        }

        public void setErTable(ERTable erTable) {
            this.erTable = erTable;
        }
    }

    public static final class DagLine {

        private final JoinRelationDag prevNode;
        private final JoinRelationDag targetNode;

        public DagLine(JoinRelationDag prevNode, JoinRelationDag targetNode) {
            this.prevNode = prevNode;
            this.targetNode = targetNode;
        }

        public JoinRelationDag getPrevNode() {
            return prevNode;
        }

        public JoinRelationDag getTargetNode() {
            return targetNode;
        }

        @Override
        public String toString() {
            return "(" + prevNode + ", " + targetNode + ")";
        }

    }

    private static class OptimizeException extends RuntimeException {
        OptimizeException(String message) {
            super(message);
        }
    }
}
