/*
 * 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.plan.common.item.Item;
import com.actiontech.dble.plan.common.item.Item.ItemType;
import com.actiontech.dble.plan.common.item.ItemField;
import com.actiontech.dble.plan.common.item.function.ItemFunc;
import com.actiontech.dble.plan.common.item.subquery.ItemSubQuery;
import com.actiontech.dble.plan.node.*;
import com.actiontech.dble.plan.node.PlanNode.PlanNodeType;
import com.actiontech.dble.plan.util.FilterUtils;
import com.actiontech.dble.plan.util.PlanUtil;
import com.actiontech.dble.singleton.TraceManager;
import com.google.common.collect.ImmutableMap;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

/**
 * push down filter
 * <p>
 * <pre>
 * a. if conditions contains || ,then stop pushing down ,otherwise it is not correct
 * b. if conditions's column/value contains function,then stop pushing down
 * (TODO: CHECK FIELD IN FUNCTION TO DECIDE IF IT CAN BE PUSHED)
 * c.if conditions's column/value is from child's function,then stop pushing down
 *
 * cases: 1. where pushed to leaf, get the joinFilter
 * node type: JoinNode/QueryNode
 * Notice:if JoinNode is outter,stop pushing down
 *
 * eg: tabl1.join(table2).query(
 * "table1.id>5 && table2.id<10 && table1.name = table2.name")
 * after optimized:
 * table1.query("table1.id>5").join(table2.query("table2.id<10").on(
 * "table1.name = table2.name")
 *
 * eg: table1.join(table2).query("table1.id = table2.id")
 * after optimized:
 * :table1.join(table2).on("table1.id = table2.id")
 *
 * 2. join's constant condition,like column = 1,push to leaf
 * node type:JoinNode
 *
 * eg: tabl1.join(table2).on("table1.id>5&&table2.id<10")
 * after optimized:
 * table1.query("table1.id>5").join(table2.query("table2.id<10")) t
 *
 *
 * 3. both 1 and 2
 * node type:JoinNode
 *
 * eg: table.join(table2).on(
 * "table1.id = table2.id and table1.id>5 && table2.id<10")
 *  after optimized::table1.query(
 * "table1.id>5 && table1.id<10").join(table2.query(
 * "table2.id>5 && table2.id<10")).on("table1.id = table2.id")
 */
public final class FilterPusher {
    private FilterPusher() {
    }

    public static PlanNode optimize(PlanNode qtn) {
        TraceManager.TraceObject traceObject = TraceManager.threadTrace("optimize-for-filter-push-down");
        try {
            mergeJoinOnFilter(qtn);
            pushJoinOnFilter(qtn);
            qtn = pushFilter(qtn, new ArrayList<>());
            //todo all subQueries will do
            if (qtn instanceof NoNameNode && qtn.isContainsSubQuery()) {
                for (ItemSubQuery subQuery : qtn.getSubQueries()) {
                    PlanNode subQtn = optimize(subQuery.getPlanNode());
                    subQuery.setPlanNode(subQtn);
                }
            }
            return qtn;
        } finally {
            TraceManager.log(ImmutableMap.of("plan-node", qtn), traceObject);
            TraceManager.finishSpan(traceObject);
        }
    }

    /**
     * merge inner joi's otheron to where if we can
     *
     */
    private static void mergeJoinOnFilter(PlanNode qtn) {
        if (PlanUtil.isGlobalOrER(qtn))
            return;
        if (qtn.type().equals(PlanNodeType.JOIN) && ((JoinNode) qtn).isInnerJoin()) {
            JoinNode jn = (JoinNode) qtn;
            Item otherJoinOn = jn.getOtherJoinOnFilter();
            jn.setOtherJoinOnFilter(null);
            jn.query(FilterUtils.and(otherJoinOn, jn.getWhereFilter()));
        }
        for (PlanNode child : qtn.getChildren()) {
            mergeJoinOnFilter(child);
        }
    }

    private static PlanNode pushFilter(PlanNode qtn, List<Item> dnfNodeToPush) {
        List<Item> subHavingList = new ArrayList<>();
        for (Item filter : dnfNodeToPush) {
            if (filter.isWithSumFunc()) {
                subHavingList.add(filter);
            }
        }
        if (!subHavingList.isEmpty()) {
            qtn.having(FilterUtils.and(qtn.getHavingFilter(), FilterUtils.and(subHavingList)));
            dnfNodeToPush.removeAll(subHavingList);
        }

        // root node will receive filter as where ,otherwise merge the current where and push down
        if (qtn.getChildren().isEmpty() || (PlanUtil.isGlobalOrER(qtn) && qtn.type() != PlanNodeType.MERGE)) {
            Item node = FilterUtils.and(dnfNodeToPush);
            if (node != null) {
                qtn.query(FilterUtils.and(qtn.getWhereFilter(), node));
                qtn.setContainsSubQuery(qtn.getWhereFilter().isWithSubQuery() || isWithSubQuery(dnfNodeToPush));
            }
            return qtn;
        }

        Item filterInWhere = qtn.getWhereFilter();
        //left/right join: where filter can't be push to child
        if (filterInWhere != null) {
            List<Item> splits = FilterUtils.splitFilter(filterInWhere);
            qtn.query(null);
            dnfNodeToPush.addAll(splits);
        }

        if (qtn.type() == PlanNodeType.QUERY) {
            return getQueryNode(qtn, dnfNodeToPush);
        } else if (qtn.type() == PlanNodeType.JOIN) {
            return getJoinNode(qtn, dnfNodeToPush);
        } else if (qtn.type() == PlanNodeType.MERGE) {
            return getMergeNode(qtn, dnfNodeToPush);
        }

        return qtn;
    }

    // qtn is a query node, only has one child
    private static PlanNode getQueryNode(PlanNode qtn, List<Item> dnfNodeToPush) {
        if (qtn.getSubQueries().size() > 0) {
            Item node = FilterUtils.and(dnfNodeToPush);
            if (node != null) {
                qtn.query(FilterUtils.and(qtn.getWhereFilter(), node));
                qtn.setContainsSubQuery(qtn.getWhereFilter().isWithSubQuery() || isWithSubQuery(dnfNodeToPush));
            }
            return qtn;
        }

        refreshPdFilters(qtn, dnfNodeToPush);
        PlanNode child = pushFilter(qtn.getChild(), dnfNodeToPush);
        ((QueryNode) qtn).setChild(child);
        return qtn;
    }

    private static PlanNode getMergeNode(PlanNode qtn, List<Item> dnfNodeToPush) {
        // union's where can be pushed down ,but it must be replace to child's condition

        Item node = FilterUtils.and(dnfNodeToPush);
        if (node != null) {
            qtn.query(FilterUtils.and(qtn.getWhereFilter(), node));
            qtn.setContainsSubQuery(qtn.getWhereFilter().isWithSubQuery() || isWithSubQuery(dnfNodeToPush));
        }
        Item mergeWhere = qtn.getWhereFilter();
        // push down merge's condition
        qtn.query(null);
        List<Item> pushFilters = PlanUtil.getPushItemsToUnionChild((MergeNode) qtn, mergeWhere,
                ((MergeNode) qtn).getColIndexs());
        List<PlanNode> childs = qtn.getChildren();
        for (int index = 0; index < childs.size(); index++) {
            PlanNode child = childs.get(index);
            if (pushFilters != null) {
                Item pushFilter = pushFilters.get(index);
                if (pushFilter.isWithSumFunc()) {
                    child.having(FilterUtils.and(child.getHavingFilter(), pushFilter));
                } else {
                    child.query(FilterUtils.and(child.getWhereFilter(), pushFilter));
                }
            }
            FilterPusher.optimize(child);
        }
        return qtn;
    }

    private static PlanNode getJoinNode(PlanNode qtn, List<Item> dnfNodeToPush) {
        JoinNode jn = (JoinNode) qtn;
        List<Item> dnfNodetoPushToLeft = new LinkedList<>();
        List<Item> dnfNodetoPushToRight = new LinkedList<>();
        List<Item> leftCopyedPushFilters = new LinkedList<>();
        List<Item> rightCopyedPushFilters = new LinkedList<>();
        List<Item> dnfNodeToCurrent = new LinkedList<>();

        PlanUtil.findJoinColumnsAndRemoveIt(dnfNodeToPush, jn);
        for (Item filter : dnfNodeToPush) {
            // ex. 1 = -1
            if (filter.getReferTables().size() == 0) {
                dnfNodetoPushToLeft.add(filter);
                dnfNodetoPushToRight.add(filter);
                continue;
            }
            if (PlanUtil.canPush(filter, jn.getLeftNode(), jn)) {
                dnfNodetoPushToLeft.add(filter);
            } else if (PlanUtil.canPush(filter, jn.getRightNode(), jn)) {
                dnfNodetoPushToRight.add(filter);
            } else {
                dnfNodeToCurrent.add(filter);
            }
        }
        if (jn.isInnerJoin() || jn.isLeftOuterJoin() || jn.isRightOuterJoin()) {
            // push the left expr to join filter's right condition
            rightCopyedPushFilters.addAll(
                    copyFilterToJoinOnColumns(dnfNodetoPushToLeft, jn.getLeftKeys(), jn.getRightKeys()));

            // push the right expr to join filter's left condition
            leftCopyedPushFilters.addAll(
                    copyFilterToJoinOnColumns(dnfNodetoPushToRight, jn.getRightKeys(), jn.getLeftKeys()));
        }

        // if it can not push down, merge to current where
        Item node = FilterUtils.and(dnfNodeToCurrent);
        if (node != null) {
            qtn.query(FilterUtils.and(qtn.getWhereFilter(), node));
        }

        if (jn.isInnerJoin() || jn.isLeftOuterJoin() || jn.isRightOuterJoin()) {
            if (jn.isLeftOuterJoin()) {
                // left join, push down the right join ,but still contains right joins condition
                jn.query(FilterUtils.and(qtn.getWhereFilter(), FilterUtils.and(dnfNodetoPushToRight)));
            }
            if (jn.isRightOuterJoin()) {
                // right join, push down the left join ,but still contains left joins condition
                jn.query(FilterUtils.and(qtn.getWhereFilter(), FilterUtils.and(dnfNodetoPushToLeft)));
            }
            // merge
            dnfNodetoPushToRight.addAll(rightCopyedPushFilters);
            dnfNodetoPushToLeft.addAll(leftCopyedPushFilters);

            refreshPdFilters(jn, dnfNodetoPushToLeft);
            refreshPdFilters(jn, dnfNodetoPushToRight);
            jn.setLeftNode(pushFilter(jn.getLeftNode(), dnfNodetoPushToLeft));
            jn.setRightNode(pushFilter(((JoinNode) qtn).getRightNode(), dnfNodetoPushToRight));
        } else {
            if (!dnfNodeToPush.isEmpty()) {
                jn.query(FilterUtils.and(qtn.getWhereFilter(), FilterUtils.and(dnfNodeToPush)));
            }
        }
        return jn;
    }

    /**
     * inner join's other join on would add to where when FilterPre and push down,
     * when Left join, eg" select * from t1 left join t2 on t1.id=t2.id and t1.id = 10 and
     * t2.name = 'aaa' push down t2.id=10 and t2.name='aaa'
     */
    private static void pushJoinOnFilter(PlanNode qtn) {
        if (PlanUtil.isGlobalOrER(qtn))
            return;
        if (qtn.type().equals(PlanNodeType.JOIN)) {
            JoinNode jn = (JoinNode) qtn;
            Item otherJoinOn = jn.getOtherJoinOnFilter();
            if (jn.isLeftOuterJoin() && otherJoinOn != null) {
                List<Item> pushToRightNode = new ArrayList<>();
                List<Item> splitedFilters = FilterUtils.splitFilter(otherJoinOn);
                for (Item filter : splitedFilters) {
                    if (filter.getReferTables().isEmpty())
                        pushToRightNode.add(filter);
                    else if (PlanUtil.canPush(filter, jn.getRightNode(), jn))
                        pushToRightNode.add(filter);
                    else if (PlanUtil.canPush(filter, jn.getLeftNode(), jn)) {
                        Item copyedFilter = copyFilterToJoinOnColumns(filter, jn.getRightKeys(), jn.getLeftKeys());
                        if (copyedFilter != null)
                            pushToRightNode.add(copyedFilter);
                    }
                }
                if (!pushToRightNode.isEmpty()) {
                    splitedFilters.removeAll(pushToRightNode);
                    Item newOtherJoinOn = FilterUtils.and(splitedFilters);
                    jn.setOtherJoinOnFilter(newOtherJoinOn);
                    refreshPdFilters(jn, pushToRightNode);
                    List<Item> subHavingList = new ArrayList<>();
                    List<Item> subWhereList = new ArrayList<>();
                    for (Item filter : pushToRightNode) {
                        if (filter.isWithSumFunc()) {
                            subHavingList.add(filter);
                        } else {
                            subWhereList.add(filter);
                        }
                    }
                    Item subHaving = FilterUtils.and(subHavingList);
                    Item subWhere = FilterUtils.and(subWhereList);
                    jn.getRightNode().having(FilterUtils.and(jn.getRightNode().getHavingFilter(), subHaving));
                    jn.getRightNode().setWhereFilter(FilterUtils.and(jn.getRightNode().getWhereFilter(), subWhere));
                }
            }
        }
        for (PlanNode child : qtn.getChildren())
            pushJoinOnFilter(child);
    }

    /**
     * copyFilterToJoinOnColumns
     *
     * @param dnf          DNF filter to be copied
     * @param qnColumns    origin node's join column
     * @param otherColumns otherColumns
     */
    private static List<Item> copyFilterToJoinOnColumns(List<Item> dnf, List<Item> qnColumns, List<Item> otherColumns) {
        List<Item> newIFilterToPush = new LinkedList<>();
        for (Item filter : dnf) {
            Item newFilter = copyFilterToJoinOnColumns(filter, qnColumns, otherColumns);
            if (newFilter != null)
                newIFilterToPush.add(newFilter);
        }
        return newIFilterToPush;
    }

    private static Item copyFilterToJoinOnColumns(Item filter, List<Item> qnColumns, List<Item> otherColumns) {
        if (filter.type().equals(ItemType.FUNC_ITEM) || filter.type().equals(ItemType.COND_ITEM)) {
            return replaceFunctionArg((ItemFunc) filter, qnColumns, otherColumns);
        }
        return null;
    }

    private static void refreshPdFilters(PlanNode qtn, List<Item> filters) {
        for (int index = 0; index < filters.size(); index++) {
            Item toPsFilter = filters.get(index);
            Item pdFilter = PlanUtil.pushDownItem(qtn, toPsFilter);
            filters.set(index, pdFilter);
        }
    }

    /**
     * replaceFunctionArg
     *
     * @return if f also contains selectable which is not sels1 ,return null
     */
    private static ItemFunc replaceFunctionArg(ItemFunc f, List<Item> sels1, List<Item> sels2) {
        ItemFunc ret = (ItemFunc) f.cloneStruct();
        boolean hasFieldInFunc = false;
        for (int index = 0; index < ret.getArgCount(); index++) {
            Item arg = ret.arguments().get(index);
            if (arg instanceof ItemFunc) {
                ItemFunc newfArg = replaceFunctionArg((ItemFunc) arg, sels1, sels2);
                if (newfArg == null)
                    return null;
                else
                    ret.arguments().set(index, newfArg);
            } else if (arg instanceof ItemField) {
                hasFieldInFunc = true;
                int tmpIndex = sels1.indexOf(arg);
                if (tmpIndex < 0) {
                    return null;
                } else {
                    Item newArg = sels2.get(tmpIndex);
                    ret.arguments().set(index, newArg.cloneStruct());
                }
            }  // else do nothing;
        }
        if (!hasFieldInFunc) {
            return null;
        }
        ret.setItemName(null);
        ret.setPushDownName(null);
        PlanUtil.refreshReferTables(ret);
        return ret;
    }

    private static boolean isWithSubQuery(List<Item> filters) {
        for (Item filter : filters) {
            if (filter.isWithSubQuery()) {
                return true;
            }
        }
        return false;
    }
}
