/**
 * 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.tajo.engine.planner.rewrite;

import com.google.common.collect.*;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.tajo.algebra.JoinType;
import org.apache.tajo.catalog.CatalogUtil;
import org.apache.tajo.catalog.Column;
import org.apache.tajo.catalog.Schema;
import org.apache.tajo.catalog.TableDesc;
import org.apache.tajo.engine.eval.*;
import org.apache.tajo.engine.exception.InvalidQueryException;
import org.apache.tajo.engine.planner.*;
import org.apache.tajo.engine.planner.logical.*;
import org.apache.tajo.engine.planner.rewrite.FilterPushDownRule.FilterPushDownContext;
import org.apache.tajo.util.TUtil;

import java.util.*;

/**
 * This rule tries to push down all filter conditions into logical nodes as lower as possible.
 * It is likely to significantly reduces the intermediate data.
 */
public class FilterPushDownRule extends BasicLogicalPlanVisitor<FilterPushDownContext, LogicalNode>
    implements RewriteRule {
  private final static Log LOG = LogFactory.getLog(FilterPushDownRule.class);
  private static final String NAME = "FilterPushDown";

  static class FilterPushDownContext {
    Set<EvalNode> pushingDownFilters = new HashSet<EvalNode>();

    public void clear() {
      pushingDownFilters.clear();
    }
    public void setFiltersTobePushed(Collection<EvalNode> workingEvals) {
      this.pushingDownFilters.clear();
      this.pushingDownFilters.addAll(workingEvals);
    }
    public void addFiltersTobePushed(Collection<EvalNode> workingEvals) {
      this.pushingDownFilters.addAll(workingEvals);
    }

    public void setToOrigin(Map<EvalNode, EvalNode> evalMap) {
      //evalMap: copy -> origin
      List<EvalNode> origins = new ArrayList<EvalNode>();
      for (EvalNode eval : pushingDownFilters) {
        EvalNode origin = evalMap.get(eval);
        if (origin != null) {
          origins.add(origin);
        }
      }
      setFiltersTobePushed(origins);
    }
  }

  @Override
  public String getName() {
    return NAME;
  }

  @Override
  public boolean isEligible(LogicalPlan plan) {
    for (LogicalPlan.QueryBlock block : plan.getQueryBlocks()) {
      if (block.hasNode(NodeType.SELECTION) || block.hasNode(NodeType.JOIN)) {
        return true;
      }
    }
    return false;
  }

  @Override
  public LogicalPlan rewrite(LogicalPlan plan) throws PlanningException {
    /*
    FilterPushDown rule: processing when visits each node
      - If a target which is corresponding on a filter EvalNode's column is not FieldEval, do not PushDown.
      - Replace filter EvalNode's column with child node's output column.
        If there is no child node's output column, do not PushDown.
      - When visit ScanNode, add filter eval to ScanNode's qual
      - When visit GroupByNode, Find aggregation column in a filter EvalNode and
        . If a parent is HavingNode, add filter eval to parent HavingNode.
        . It not, create new HavingNode and set parent's child.
     */
    FilterPushDownContext context = new FilterPushDownContext();
    for (LogicalPlan.QueryBlock block : plan.getQueryBlocks()) {
      context.clear();
      this.visit(context, plan, block, block.getRoot(), new Stack<LogicalNode>());
    }

    if (LOG.isDebugEnabled()) {
      LOG.debug("=============================================");
      LOG.debug("FilterPushDown Optimized Query: \n" + plan.toString());
      LOG.debug("=============================================");
    }
    return plan;
  }

  @Override
  public LogicalNode visitFilter(FilterPushDownContext context, LogicalPlan plan, LogicalPlan.QueryBlock block,
                                 SelectionNode selNode, Stack<LogicalNode> stack) throws PlanningException {
    context.pushingDownFilters.addAll(Sets.newHashSet(AlgebraicUtil.toConjunctiveNormalFormArray(selNode.getQual())));

    stack.push(selNode);
    visit(context, plan, block, selNode.getChild(), stack);
    stack.pop();

    if(context.pushingDownFilters.size() == 0) {
      // remove the selection operator if there is no search condition after selection push.
      LogicalNode node = stack.peek();
      if (node instanceof UnaryNode) {
        UnaryNode unary = (UnaryNode) node;
        unary.setChild(selNode.getChild());
      } else {
        throw new InvalidQueryException("Unexpected Logical Query Plan");
      }
    } else { // if there remain search conditions

      // check if it can be evaluated here
      Set<EvalNode> matched = TUtil.newHashSet();
      for (EvalNode eachEval : context.pushingDownFilters) {
        if (LogicalPlanner.checkIfBeEvaluatedAtThis(eachEval, selNode)) {
          matched.add(eachEval);
        }
      }

      // if there are search conditions which can be evaluated here,
      // push down them and remove them from context.pushingDownFilters.
      if (matched.size() > 0) {
        selNode.setQual(AlgebraicUtil.createSingletonExprFromCNF(matched.toArray(new EvalNode[matched.size()])));
        context.pushingDownFilters.removeAll(matched);
      }
    }

    return selNode;
  }

  @Override
  public LogicalNode visitJoin(FilterPushDownContext context, LogicalPlan plan, LogicalPlan.QueryBlock block, JoinNode joinNode,
                               Stack<LogicalNode> stack) throws PlanningException {
    // here we should stop selection pushdown on the null supplying side(s) of an outer join
    // get the two operands of the join operation as well as the join type
    JoinType joinType = joinNode.getJoinType();
    EvalNode joinQual = joinNode.getJoinQual();
    if (joinQual != null && LogicalPlanner.isOuterJoin(joinType)) {
      BinaryEval binaryEval = (BinaryEval) joinQual;
      // if both are fields
      if (binaryEval.getLeftExpr().getType() == EvalType.FIELD &&
          binaryEval.getRightExpr().getType() == EvalType.FIELD) {

        String leftTableName = ((FieldEval) binaryEval.getLeftExpr()).getQualifier();
        String rightTableName = ((FieldEval) binaryEval.getRightExpr()).getQualifier();
        List<String> nullSuppliers = Lists.newArrayList();
        Set<String> leftTableSet = Sets.newHashSet(PlannerUtil.getRelationLineageWithinQueryBlock(plan,
            joinNode.getLeftChild()));
        Set<String> rightTableSet = Sets.newHashSet(PlannerUtil.getRelationLineageWithinQueryBlock(plan,
            joinNode.getRightChild()));

        // some verification
        if (joinType == JoinType.FULL_OUTER) {
          nullSuppliers.add(leftTableName);
          nullSuppliers.add(rightTableName);

          // verify that these null suppliers are indeed in the left and right sets
          if (!rightTableSet.contains(nullSuppliers.get(0)) && !leftTableSet.contains(nullSuppliers.get(0))) {
            throw new InvalidQueryException("Incorrect Logical Query Plan with regard to outer join");
          }
          if (!rightTableSet.contains(nullSuppliers.get(1)) && !leftTableSet.contains(nullSuppliers.get(1))) {
            throw new InvalidQueryException("Incorrect Logical Query Plan with regard to outer join");
          }

        } else if (joinType == JoinType.LEFT_OUTER) {
          nullSuppliers.add(((RelationNode)joinNode.getRightChild()).getCanonicalName());
          //verify that this null supplier is indeed in the right sub-tree
          if (!rightTableSet.contains(nullSuppliers.get(0))) {
            throw new InvalidQueryException("Incorrect Logical Query Plan with regard to outer join");
          }
        } else if (joinType == JoinType.RIGHT_OUTER) {
          if (((RelationNode)joinNode.getRightChild()).getCanonicalName().equals(rightTableName)) {
            nullSuppliers.add(leftTableName);
          } else {
            nullSuppliers.add(rightTableName);
          }

          // verify that this null supplier is indeed in the left sub-tree
          if (!leftTableSet.contains(nullSuppliers.get(0))) {
            throw new InvalidQueryException("Incorrect Logical Query Plan with regard to outer join");
          }
        }
      }
    }

    // get evals from ON clause
    List<EvalNode> onConditions = new ArrayList<EvalNode>();
    if (joinNode.hasJoinQual()) {
      onConditions.addAll(Sets.newHashSet(AlgebraicUtil.toConjunctiveNormalFormArray(joinNode.getJoinQual())));
    }

    boolean isTopMostJoin = stack.peek().getType() != NodeType.JOIN;

    List<EvalNode> outerJoinPredicationEvals = new ArrayList<EvalNode>();
    List<EvalNode> outerJoinFilterEvalsExcludePredication = new ArrayList<EvalNode>();
    if (LogicalPlanner.isOuterJoin(joinNode.getJoinType())) {
      // TAJO-853
      // In the case of top most JOIN, all filters except JOIN condition aren't pushed down.
      // That filters are processed by SELECTION NODE.
      Set<String> nullSupplyingTableNameSet;
      if (joinNode.getJoinType() == JoinType.RIGHT_OUTER) {
        nullSupplyingTableNameSet = TUtil.newHashSet(PlannerUtil.getRelationLineage(joinNode.getLeftChild()));
      } else {
        nullSupplyingTableNameSet = TUtil.newHashSet(PlannerUtil.getRelationLineage(joinNode.getRightChild()));
      }

      Set<String> preservedTableNameSet;
      if (joinNode.getJoinType() == JoinType.RIGHT_OUTER) {
        preservedTableNameSet = TUtil.newHashSet(PlannerUtil.getRelationLineage(joinNode.getRightChild()));
      } else {
        preservedTableNameSet = TUtil.newHashSet(PlannerUtil.getRelationLineage(joinNode.getLeftChild()));
      }

      List<EvalNode> removedFromFilter = new ArrayList<EvalNode>();
      for (EvalNode eachEval: context.pushingDownFilters) {
        if (EvalTreeUtil.isJoinQual(eachEval, true)) {
          outerJoinPredicationEvals.add(eachEval);
          removedFromFilter.add(eachEval);
        } else {
          Set<Column> columns = EvalTreeUtil.findUniqueColumns(eachEval);
          boolean canPushDown = true;
          for (Column eachColumn: columns) {
            if (nullSupplyingTableNameSet.contains(eachColumn.getQualifier())) {
              canPushDown = false;
              break;
            }
          }
          if (!canPushDown) {
            outerJoinFilterEvalsExcludePredication.add(eachEval);
            removedFromFilter.add(eachEval);
          }
        }
      }

      context.pushingDownFilters.removeAll(removedFromFilter);

      for (EvalNode eachOnEval: onConditions) {
        if (EvalTreeUtil.isJoinQual(eachOnEval, true)) {
          // If join condition, processing in the JoinNode.
          outerJoinPredicationEvals.add(eachOnEval);
        } else {
          // If Eval has a column which belong to Preserved Row table, not using to push down but using JoinCondition
          Set<Column> columns = EvalTreeUtil.findUniqueColumns(eachOnEval);
          boolean canPushDown = true;
          for (Column eachColumn: columns) {
            if (preservedTableNameSet.contains(eachColumn.getQualifier())) {
              canPushDown = false;
              break;
            }
          }
          if (canPushDown) {
            context.pushingDownFilters.add(eachOnEval);
          } else {
            outerJoinPredicationEvals.add(eachOnEval);
          }
        }
      }
    } else {
      context.pushingDownFilters.addAll(onConditions);
    }

    LogicalNode left = joinNode.getLeftChild();
    LogicalNode right = joinNode.getRightChild();

    List<EvalNode> notMatched = new ArrayList<EvalNode>();
    // Join's input schema = right child output columns + left child output columns
    Map<EvalNode, EvalNode> transformedMap = findCanPushdownAndTransform(context, joinNode, left, notMatched, null, true,
        0);
    context.setFiltersTobePushed(transformedMap.keySet());
    visit(context, plan, block, left, stack);

    context.setToOrigin(transformedMap);
    context.addFiltersTobePushed(notMatched);

    notMatched.clear();
    transformedMap = findCanPushdownAndTransform(context, joinNode, right, notMatched, null, true, left.getOutSchema().size());
    context.setFiltersTobePushed(new HashSet<EvalNode>(transformedMap.keySet()));

    visit(context, plan, block, right, stack);

    context.setToOrigin(transformedMap);
    context.addFiltersTobePushed(notMatched);

    notMatched.clear();
    List<EvalNode> matched = Lists.newArrayList();
    if(LogicalPlanner.isOuterJoin(joinNode.getJoinType())) {
      matched.addAll(outerJoinPredicationEvals);
    } else {
      for (EvalNode eval : context.pushingDownFilters) {
        if (LogicalPlanner.checkIfBeEvaluatedAtJoin(block, eval, joinNode, isTopMostJoin)) {
          matched.add(eval);
        }
      }
    }

    EvalNode qual = null;
    if (matched.size() > 1) {
      // merged into one eval tree
      qual = AlgebraicUtil.createSingletonExprFromCNF(
          matched.toArray(new EvalNode[matched.size()]));
    } else if (matched.size() == 1) {
      // if the number of matched expr is one
      qual = matched.get(0);
    }

    if (qual != null) {
      joinNode.setJoinQual(qual);

      if (joinNode.getJoinType() == JoinType.CROSS) {
        joinNode.setJoinType(JoinType.INNER);
      }
      context.pushingDownFilters.removeAll(matched);
    }

    context.pushingDownFilters.addAll(outerJoinFilterEvalsExcludePredication);
    return joinNode;
  }

  private Map<EvalNode, EvalNode> transformEvalsWidthByPassNode(
      Collection<EvalNode> originEvals, LogicalPlan plan,
      LogicalPlan.QueryBlock block,
      LogicalNode node, LogicalNode childNode) throws PlanningException {
    // transformed -> pushingDownFilters
    Map<EvalNode, EvalNode> transformedMap = new HashMap<EvalNode, EvalNode>();

    if (originEvals.isEmpty()) {
      return transformedMap;
    }

    if (node.getType() == NodeType.UNION) {
      // If node is union, All eval's columns are simple name and matched with child's output schema.
      Schema childOutSchema = childNode.getOutSchema();
      for (EvalNode eval : originEvals) {
        EvalNode copy;
        try {
          copy = (EvalNode) eval.clone();
        } catch (CloneNotSupportedException e) {
          throw new PlanningException(e);
        }

        Set<Column> columns = EvalTreeUtil.findUniqueColumns(copy);
        for (Column c : columns) {
          Column column = childOutSchema.getColumn(c.getSimpleName());
          if (column == null) {
            throw new PlanningException(
                "Invalid Filter PushDown on SubQuery: No such a corresponding column '"
                    + c.getQualifiedName() + " for FilterPushDown(" + eval + "), " +
                    "(PID=" + node.getPID() + ", Child=" + childNode.getPID() + ")");
          }
          EvalTreeUtil.changeColumnRef(copy, c.getSimpleName(), column.getQualifiedName());
        }

        transformedMap.put(copy, eval);
      }
      return transformedMap;
    }

    if (childNode.getType() == NodeType.UNION) {
      // If child is union, remove qualifier from eval's column
      for (EvalNode eval : originEvals) {
        EvalNode copy;
        try {
          copy = (EvalNode) eval.clone();
        } catch (CloneNotSupportedException e) {
          throw new PlanningException(e);
        }

        Set<Column> columns = EvalTreeUtil.findUniqueColumns(copy);
        for (Column c : columns) {
          if (c.hasQualifier()) {
            EvalTreeUtil.changeColumnRef(copy, c.getQualifiedName(), c.getSimpleName());
          }
        }

        transformedMap.put(copy, eval);
      }

      return transformedMap;
    }

    // node in column -> child out column
    Map<String, String> columnMap = new HashMap<String, String>();

    for (int i = 0; i < node.getInSchema().size(); i++) {
      String inColumnName = node.getInSchema().getColumn(i).getQualifiedName();
      Column childOutColumn = childNode.getOutSchema().getColumn(i);
      columnMap.put(inColumnName, childOutColumn.getQualifiedName());
    }

    // Rename from upper block's one to lower block's one
    for (EvalNode matchedEval : originEvals) {
      EvalNode copy;
      try {
        copy = (EvalNode) matchedEval.clone();
      } catch (CloneNotSupportedException e) {
        throw new PlanningException(e);
      }

      Set<Column> columns = EvalTreeUtil.findUniqueColumns(copy);
      boolean allMatched = true;
      for (Column c : columns) {
        if (columnMap.containsKey(c.getQualifiedName())) {
          EvalTreeUtil.changeColumnRef(copy, c.getQualifiedName(), columnMap.get(c.getQualifiedName()));
        } else {
          if (childNode.getType() == NodeType.GROUP_BY) {
            if (((GroupbyNode) childNode).isAggregationColumn(c.getSimpleName())) {
              allMatched = false;
              break;
            }
          } else {
            throw new PlanningException(
                "Invalid Filter PushDown on SubQuery: No such a corresponding column '"
                    + c.getQualifiedName() + " for FilterPushDown(" + matchedEval + "), " +
                    "(PID=" + node.getPID() + ", Child=" + childNode.getPID() + ")"
            );
          }
        }
      }
      if (allMatched) {
        transformedMap.put(copy, matchedEval);
      }
    }

    return transformedMap;
  }

  @Override
  public LogicalNode visitTableSubQuery(FilterPushDownContext context, LogicalPlan plan, LogicalPlan.QueryBlock block,
                                        TableSubQueryNode node, Stack<LogicalNode> stack) throws PlanningException {
    List<EvalNode> matched = Lists.newArrayList();
    for (EvalNode eval : context.pushingDownFilters) {
      if (LogicalPlanner.checkIfBeEvaluatedAtRelation(block, eval, node)) {
        matched.add(eval);
      }
    }

    // transformed -> pushingDownFilters
    Map<EvalNode, EvalNode> transformedMap =
        transformEvalsWidthByPassNode(matched, plan, block, node, node.getSubQuery());

    context.setFiltersTobePushed(new HashSet<EvalNode>(transformedMap.keySet()));
    visit(context, plan, plan.getBlock(node.getSubQuery()));
    context.setToOrigin(transformedMap);

    return node;
  }

  @Override
  public LogicalNode visitUnion(FilterPushDownContext context, LogicalPlan plan,
                                LogicalPlan.QueryBlock block, UnionNode unionNode,
                                Stack<LogicalNode> stack) throws PlanningException {
    LogicalNode leftNode = unionNode.getLeftChild();

    List<EvalNode> origins = new ArrayList<EvalNode>(context.pushingDownFilters);

    // transformed -> pushingDownFilters
    Map<EvalNode, EvalNode> transformedMap = transformEvalsWidthByPassNode(origins, plan, block, unionNode, leftNode);
    context.setFiltersTobePushed(new HashSet<EvalNode>(transformedMap.keySet()));
    visit(context, plan, plan.getBlock(leftNode));

    if (!context.pushingDownFilters.isEmpty()) {
      errorFilterPushDown(plan, leftNode, context);
    }

    LogicalNode rightNode = unionNode.getRightChild();
    transformedMap = transformEvalsWidthByPassNode(origins, plan, block, unionNode, rightNode);
    context.setFiltersTobePushed(new HashSet<EvalNode>(transformedMap.keySet()));
    visit(context, plan, plan.getBlock(rightNode), rightNode, stack);

    if (!context.pushingDownFilters.isEmpty()) {
      errorFilterPushDown(plan, rightNode, context);
    }

    // notify all filter matched to upper
    context.pushingDownFilters.clear();
    return unionNode;
  }

  @Override
  public LogicalNode visitProjection(FilterPushDownContext context,
                                     LogicalPlan plan,
                                     LogicalPlan.QueryBlock block,
                                     ProjectionNode projectionNode,
                                     Stack<LogicalNode> stack) throws PlanningException {
    LogicalNode childNode = projectionNode.getChild();

    List<EvalNode> notMatched = new ArrayList<EvalNode>();

    //copy -> origin
    BiMap<EvalNode, EvalNode> transformedMap = findCanPushdownAndTransform(
        context, projectionNode, childNode, notMatched, null, false, 0);

    context.setFiltersTobePushed(transformedMap.keySet());

    stack.push(projectionNode);
    childNode = visit(context, plan, plan.getBlock(childNode), childNode, stack);
    stack.pop();

    // find not matched after visiting child
    for (EvalNode eval: context.pushingDownFilters) {
      notMatched.add(transformedMap.get(eval));
    }

    EvalNode qual = null;
    if (notMatched.size() > 1) {
      // merged into one eval tree
      qual = AlgebraicUtil.createSingletonExprFromCNF(notMatched.toArray(new EvalNode[notMatched.size()]));
    } else if (notMatched.size() == 1) {
      // if the number of matched expr is one
      qual = notMatched.get(0);
    }

    // If there is not matched node add SelectionNode and clear context.pushingDownFilters
    if (qual != null && LogicalPlanner.checkIfBeEvaluatedAtThis(qual, projectionNode)) {
      SelectionNode selectionNode = plan.createNode(SelectionNode.class);
      selectionNode.setInSchema(childNode.getOutSchema());
      selectionNode.setOutSchema(childNode.getOutSchema());
      selectionNode.setQual(qual);
      block.registerNode(selectionNode);

      projectionNode.setChild(selectionNode);
      selectionNode.setChild(childNode);

      // clean all remain filters because all conditions are merged into a qual
      context.pushingDownFilters.clear();
    }

    // if there are remain filters, recover the original names and give back to the upper query block.
    if (context.pushingDownFilters.size() > 0) {
      ImmutableSet<EvalNode> copy = ImmutableSet.copyOf(context.pushingDownFilters);
      context.pushingDownFilters.clear();
      context.pushingDownFilters.addAll(reverseTransform(transformedMap, copy));
    }

    return projectionNode;
  }

  private Collection<EvalNode> reverseTransform(BiMap<EvalNode, EvalNode> map, Set<EvalNode> remainFilters) {
    Set<EvalNode> reversed = Sets.newHashSet();
    for (EvalNode evalNode : remainFilters) {
      reversed.add(map.get(evalNode));
    }
    return reversed;
  }

  private BiMap<EvalNode, EvalNode> findCanPushdownAndTransform(
      FilterPushDownContext context, Projectable node,
      LogicalNode childNode, List<EvalNode> notMatched,
      Set<String> partitionColumns,
      boolean ignoreJoin, int columnOffset) throws PlanningException {
    // canonical name -> target
    Map<String, Target> nodeTargetMap = new HashMap<String, Target>();
    for (Target target : node.getTargets()) {
      nodeTargetMap.put(target.getCanonicalName(), target);
    }

    // copy -> origin
    BiMap<EvalNode, EvalNode> matched = HashBiMap.create();

    for (EvalNode eval : context.pushingDownFilters) {
      if (ignoreJoin && EvalTreeUtil.isJoinQual(eval, true)) {
        notMatched.add(eval);
        continue;
      }
      // If all column is field eval, can push down.
      Set<Column> evalColumns = EvalTreeUtil.findUniqueColumns(eval);
      boolean columnMatched = true;
      for (Column c : evalColumns) {
        Target target = nodeTargetMap.get(c.getQualifiedName());
        if (target == null) {
          columnMatched = false;
          break;
        }
        if (target.getEvalTree().getType() != EvalType.FIELD) {
          columnMatched = false;
          break;
        }
      }

      if (columnMatched) {
        // transform eval column to child's output column
        EvalNode copyEvalNode = transformEval(node, childNode, eval, nodeTargetMap, partitionColumns, columnOffset);
        if (copyEvalNode != null) {
          matched.put(copyEvalNode, eval);
        } else {
          notMatched.add(eval);
        }
      } else {
        notMatched.add(eval);
      }
    }

    return matched;
  }

  private EvalNode transformEval(Projectable node, LogicalNode childNode, EvalNode origin,
                                 Map<String, Target> targetMap, Set<String> partitionColumns,
                                 int columnOffset) throws PlanningException {
    Schema outputSchema = childNode != null ? childNode.getOutSchema() : node.getInSchema();
    EvalNode copy;
    try {
      copy = (EvalNode) origin.clone();
    } catch (CloneNotSupportedException e) {
      throw new PlanningException(e);
    }
    Set<Column> columns = EvalTreeUtil.findUniqueColumns(copy);
    for (Column c: columns) {
      Target target = targetMap.get(c.getQualifiedName());
      if (target == null) {
        throw new PlanningException(
            "Invalid Filter PushDown: No such a corresponding target '"
                + c.getQualifiedName() + " for FilterPushDown(" + origin + "), " +
                "(PID=" + node.getPID() + ")"
        );
      }
      EvalNode targetEvalNode = target.getEvalTree();
      if (targetEvalNode.getType() != EvalType.FIELD) {
        throw new PlanningException(
            "Invalid Filter PushDown: '" + c.getQualifiedName() + "' target is not FieldEval " +
                "(PID=" + node.getPID() + ")"
        );
      }

      FieldEval fieldEval = (FieldEval)targetEvalNode;
      Column targetInputColumn = fieldEval.getColumnRef();

      int index;
      if (targetInputColumn.hasQualifier()) {
        index = node.getInSchema().getColumnId(targetInputColumn.getQualifiedName());
      } else {
        index = node.getInSchema().getColumnIdByName(targetInputColumn.getQualifiedName());
      }
      if (columnOffset > 0) {
        index = index - columnOffset;
      }
      if (index < 0 || index >= outputSchema.size()) {
        if (partitionColumns != null && !partitionColumns.isEmpty() && node instanceof ScanNode) {
          ScanNode scanNode = (ScanNode)node;
          boolean isPartitionColumn = false;
          if (CatalogUtil.isFQColumnName(partitionColumns.iterator().next())) {
            isPartitionColumn = partitionColumns.contains(
                CatalogUtil.buildFQName(scanNode.getTableName(), c.getSimpleName()));
          } else {
            isPartitionColumn = partitionColumns.contains(c.getSimpleName());
          }
          if (isPartitionColumn) {
            EvalTreeUtil.changeColumnRef(copy, c.getQualifiedName(),
                scanNode.getCanonicalName() + "." + c.getSimpleName());
          } else {
            return null;
          }
        } else {
          return null;
        }
      } else {
        Column outputColumn = outputSchema.getColumn(index);
        EvalTreeUtil.changeColumnRef(copy, c.getQualifiedName(), outputColumn.getQualifiedName());
      }
    }

    return copy;
  }

  /**
   * Find aggregation columns in filter eval and add having clause or add HavingNode.
   * @param context
   * @param plan
   * @param block
   * @param parentNode  If null, having is parent
   * @param havingNode      If null, projection is parent
   * @param groupByNode
   * @return matched origin eval
   * @throws PlanningException
   */
  private List<EvalNode> addHavingNode(FilterPushDownContext context, LogicalPlan plan,
                                       LogicalPlan.QueryBlock block,
                                       UnaryNode parentNode,
                                       HavingNode havingNode,
                                       GroupbyNode groupByNode) throws PlanningException {
    // find aggregation column
    Set<Column> groupingColumns = new HashSet<Column>(Arrays.asList(groupByNode.getGroupingColumns()));
    Set<String> aggrFunctionOutColumns = new HashSet<String>();
    for (Column column : groupByNode.getOutSchema().getColumns()) {
      if (!groupingColumns.contains(column)) {
        aggrFunctionOutColumns.add(column.getQualifiedName());
      }
    }

    List<EvalNode> aggrEvalOrigins = new ArrayList<EvalNode>();
    List<EvalNode> aggrEvals = new ArrayList<EvalNode>();

    for (EvalNode eval : context.pushingDownFilters) {
      EvalNode copy = null;
      try {
        copy = (EvalNode)eval.clone();
      } catch (CloneNotSupportedException e) {
      }
      boolean isEvalAggrFunction = false;
      for (Column evalColumn : EvalTreeUtil.findUniqueColumns(copy)) {
        if (aggrFunctionOutColumns.contains(evalColumn.getSimpleName())) {
          EvalTreeUtil.changeColumnRef(copy, evalColumn.getQualifiedName(), evalColumn.getSimpleName());
          isEvalAggrFunction = true;
          break;
        }
      }
      if (isEvalAggrFunction) {
        aggrEvals.add(copy);
        aggrEvalOrigins.add(eval);
      }
    }

    if (aggrEvals.isEmpty()) {
      return aggrEvalOrigins;
    }

    // transform

    HavingNode workingHavingNode;
    if (havingNode != null) {
      workingHavingNode = havingNode;
      aggrEvals.add(havingNode.getQual());
    } else {
      workingHavingNode = plan.createNode(HavingNode.class);
      block.registerNode(workingHavingNode);
      parentNode.setChild(workingHavingNode);
      workingHavingNode.setChild(groupByNode);
    }

    EvalNode qual = null;
    if (aggrEvals.size() > 1) {
      // merged into one eval tree
      qual = AlgebraicUtil.createSingletonExprFromCNF(aggrEvals.toArray(new EvalNode[aggrEvals.size()]));
    } else if (aggrEvals.size() == 1) {
      // if the number of matched expr is one
      qual = aggrEvals.get(0);
    }

    // If there is not matched node add SelectionNode and clear context.pushingDownFilters
    if (qual != null) {
      workingHavingNode.setQual(qual);
    }

    return aggrEvalOrigins;
  }

  @Override
  public LogicalNode visitWindowAgg(FilterPushDownContext context, LogicalPlan plan,
                                  LogicalPlan.QueryBlock block, WindowAggNode winAggNode,
                                  Stack<LogicalNode> stack) throws PlanningException {
    stack.push(winAggNode);
    super.visitWindowAgg(context, plan, block, winAggNode, stack);
    stack.pop();
    return winAggNode;
  }


  @Override
  public LogicalNode visitGroupBy(FilterPushDownContext context, LogicalPlan plan,
                                  LogicalPlan.QueryBlock block, GroupbyNode groupbyNode,
                                  Stack<LogicalNode> stack) throws PlanningException {
    LogicalNode parentNode = stack.peek();
    List<EvalNode> aggrEvals;
    if (parentNode.getType() == NodeType.HAVING) {
      aggrEvals = addHavingNode(context, plan, block, null, (HavingNode)parentNode, groupbyNode);
    } else {
      aggrEvals = addHavingNode(context, plan, block, (UnaryNode)parentNode, null, groupbyNode);
    }

    if (aggrEvals != null) {
      // remove aggregation eval from conext
      context.pushingDownFilters.removeAll(aggrEvals);
    }

    List<EvalNode> notMatched = new ArrayList<EvalNode>();
    // transform
    Map<EvalNode, EvalNode> tranformed =
        findCanPushdownAndTransform(context, groupbyNode,groupbyNode.getChild(), notMatched, null, false, 0);

    context.setFiltersTobePushed(tranformed.keySet());
    LogicalNode current = super.visitGroupBy(context, plan, block, groupbyNode, stack);

    context.setToOrigin(tranformed);
    context.addFiltersTobePushed(notMatched);

    return current;
  }

  @Override
  public LogicalNode visitScan(FilterPushDownContext context, LogicalPlan plan,
                               LogicalPlan.QueryBlock block, ScanNode scanNode,
                               Stack<LogicalNode> stack) throws PlanningException {
    List<EvalNode> matched = Lists.newArrayList();

    // find partition column and check matching
    Set<String> partitionColumns = new HashSet<String>();
    TableDesc table = scanNode.getTableDesc();
    boolean hasQualifiedName = false;
    if (table.hasPartition()) {
      for (Column c: table.getPartitionMethod().getExpressionSchema().getColumns()) {
        partitionColumns.add(c.getQualifiedName());
        hasQualifiedName = c.hasQualifier();
      }
    }
    Set<EvalNode> partitionEvals = new HashSet<EvalNode>();
    for (EvalNode eval : context.pushingDownFilters) {
      if (table.hasPartition()) {
        Set<Column> columns = EvalTreeUtil.findUniqueColumns(eval);
        if (columns.size() != 1) {
          continue;
        }
        Column column = columns.iterator().next();

        // If catalog runs with HCatalog, partition column is a qualified name
        // Else partition column is a simple name
        boolean isPartitionColumn = false;
        if (hasQualifiedName) {
          isPartitionColumn = partitionColumns.contains(CatalogUtil.buildFQName(table.getName(), column.getSimpleName()));
        } else {
          isPartitionColumn = partitionColumns.contains(column.getSimpleName());
        }
        if (isPartitionColumn) {
          EvalNode copy;
          try {
            copy = (EvalNode) eval.clone();
          } catch (CloneNotSupportedException e) {
            throw new PlanningException(e);
          }
          EvalTreeUtil.changeColumnRef(copy, column.getQualifiedName(),
              scanNode.getCanonicalName() + "." + column.getSimpleName());
          matched.add(copy);
          partitionEvals.add(eval);
        }
      }
    }

    context.pushingDownFilters.removeAll(partitionEvals);

    List<EvalNode> notMatched = new ArrayList<EvalNode>();

    // transform
    Map<EvalNode, EvalNode> transformed =
        findCanPushdownAndTransform(context, scanNode, null, notMatched, partitionColumns, true, 0);

    for (EvalNode eval : transformed.keySet()) {
      if (LogicalPlanner.checkIfBeEvaluatedAtRelation(block, eval, scanNode)) {
        matched.add(eval);
      }
    }

    EvalNode qual = null;
    if (matched.size() > 1) {
      // merged into one eval tree
      qual = AlgebraicUtil.createSingletonExprFromCNF(
          matched.toArray(new EvalNode[matched.size()]));
    } else if (matched.size() == 1) {
      // if the number of matched expr is one
      qual = matched.iterator().next();
    }

    if (qual != null) { // if a matched qual exists
      scanNode.setQual(qual);
    }

    for (EvalNode matchedEval: matched) {
      transformed.remove(matchedEval);
    }

    context.setToOrigin(transformed);
    context.addFiltersTobePushed(notMatched);

    return scanNode;
  }

  private void errorFilterPushDown(LogicalPlan plan, LogicalNode node,
                                   FilterPushDownContext context) throws PlanningException {
    String notMatchedNodeStr = "";
    String prefix = "";
    for (EvalNode notMatchedNode: context.pushingDownFilters) {
      notMatchedNodeStr += prefix + notMatchedNode;
      prefix = ", ";
    }
    throw new PlanningException("FilterPushDown failed cause some filters not matched: " + notMatchedNodeStr + "\n" +
        "Error node: " + node.getPlanString() + "\n" +
        plan.toString());
  }
}