/*
 * 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.carbondata.core.scan.expression;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.scan.expression.conditional.GreaterThanEqualToExpression;
import org.apache.carbondata.core.scan.expression.conditional.GreaterThanExpression;
import org.apache.carbondata.core.scan.expression.conditional.LessThanEqualToExpression;
import org.apache.carbondata.core.scan.expression.conditional.LessThanExpression;
import org.apache.carbondata.core.scan.expression.logical.AndExpression;
import org.apache.carbondata.core.scan.expression.logical.OrExpression;
import org.apache.carbondata.core.scan.expression.logical.RangeExpression;
import org.apache.carbondata.core.scan.expression.logical.TrueExpression;
import org.apache.carbondata.core.scan.filter.intf.ExpressionType;

import static org.apache.carbondata.core.scan.filter.intf.ExpressionType.FALSE;
import static org.apache.carbondata.core.scan.filter.intf.ExpressionType.GREATERTHAN;
import static org.apache.carbondata.core.scan.filter.intf.ExpressionType.GREATERTHAN_EQUALTO;
import static org.apache.carbondata.core.scan.filter.intf.ExpressionType.LESSTHAN;
import static org.apache.carbondata.core.scan.filter.intf.ExpressionType.LESSTHAN_EQUALTO;

import org.apache.log4j.Logger;

public class RangeExpressionEvaluator {
  private static final Logger LOG =
      LogServiceFactory.getLogService(RangeExpressionEvaluator.class.getName());
  private Expression expr;
  private Expression srcNode;
  private Expression tarNode;
  private Expression tarParentNode;

  public RangeExpressionEvaluator(Expression expr) {
    this.expr = expr;
  }

  public Expression getExpr() {
    return expr;
  }

  public void setExpr(Expression expr) {
    this.expr = expr;
  }

  private Expression getSrcNode() {
    return srcNode;
  }

  private void setTarNode(Expression expr) {
    this.tarNode = expr;
  }

  private void setTarParentNode(Expression expr) {
    this.tarParentNode = expr;
  }

  /**
   * This method evaluates is any greater than or less than expression can be transformed
   * into a single RANGE filter.
   */
  public void rangeExpressionEvaluatorMapBased() {
    // The algorithm :
    // Get all the nodes of the Expression Tree and fill it into a MAP.
    // The Map structure will be currentNode, ColumnName, LessThanOrGreaterThan, Value, ParentNode
    // Group the rows in MAP according to the columns and then evaluate if it can be transformed
    // into a RANGE or not.
    //
    //            AND                                           AND
    //             |                                             |
    //            / \                                           / \
    //           /   \                                         /   \
    //        Less   Greater         =>                    TRUE   Range
    //         / \    / \                                         / \
    //        /   \  /   \                                       /   \
    //       a    10 a   5                                   Less   greater
    //                                                        /\      /\
    //                                                       /  \    /  \
    //                                                      a   10  a    5
    //

    Map<String, List<FilterModificationNode>> filterExpressionMap;
    filterExpressionMap = convertFilterTreeToMap();
    replaceWithRangeExpression(filterExpressionMap);
    filterExpressionMap.clear();
  }

  public void replaceWithRangeExpression(
      Map<String, List<FilterModificationNode>> filterExpressionMap) {

    List<FilterModificationNode> deleteExp = new ArrayList<>();
    Iterator<Map.Entry<String, List<FilterModificationNode>>> iterator =
        filterExpressionMap.entrySet().iterator();
    Map.Entry<String, List<FilterModificationNode>> nextEntry = null;
    while (iterator.hasNext()) {
      nextEntry = iterator.next();
      List<FilterModificationNode> filterExp = nextEntry.getValue();
      if (filterExp.size() > 1) {
        // There are multiple Expression for the same column traverse and check if they can
        // form a range.
        FilterModificationNode startMin = null;
        FilterModificationNode endMax = null;

        for (FilterModificationNode exp : filterExp) {
          if ((exp.getExpType() == GREATERTHAN) || (exp.getExpType() == GREATERTHAN_EQUALTO)) {
            if ((null == endMax) || ((checkLiteralValue(exp.getCurrentExp(),
                endMax.getCurrentExp())))) {
              if (null == startMin) {
                startMin = exp;
              } else {
                // There is already some value in startMin so check which one is greater.
                LiteralExpression srcLiteral = getChildLiteralExpression(startMin.getCurrentExp());
                LiteralExpression tarLiteral = getChildLiteralExpression(exp.getCurrentExp());

                ExpressionResult srcExpResult = srcLiteral.evaluate(null);
                ExpressionResult tarExpResult = tarLiteral.evaluate(null);
                if (srcExpResult.compareTo(tarExpResult) < 0) {
                  // Before replacing the startMin add the current StartMin into deleteExp List
                  // as they will be replaced with TRUE expression after RANGE formation.
                  deleteExp.add(startMin);
                  startMin = exp;
                }
              }
            }
          }
          if ((exp.getExpType() == LESSTHAN) || (exp.getExpType() == LESSTHAN_EQUALTO)) {
            if ((null == startMin) || ((checkLiteralValue(exp.getCurrentExp(),
                startMin.getCurrentExp())))) {
              if (null == endMax) {
                endMax = exp;
              } else {
                // There is already some value in endMax so check which one is less.
                LiteralExpression srcLiteral = getChildLiteralExpression(endMax.getCurrentExp());
                LiteralExpression tarLiteral = getChildLiteralExpression(exp.getCurrentExp());

                ExpressionResult srcExpResult = srcLiteral.evaluate(null);
                ExpressionResult tarExpResult = tarLiteral.evaluate(null);
                if (srcExpResult.compareTo(tarExpResult) > 0) {
                  // Before replacing the endMax add the current endMax into deleteExp List
                  // as they will be replaced with TRUE expression after RANGE formation.
                  deleteExp.add(endMax);
                  endMax = exp;
                }
              }
            }
          }
        }

        if ((null != startMin) && (null != endMax)) {
          LOG.info(
              "GreaterThan and LessThan Filter Expression changed to Range Expression for column "
                  + nextEntry.getKey());
          // the node can be converted to RANGE.
          Expression n1 = startMin.getCurrentExp();
          Expression n2 = endMax.getCurrentExp();

          RangeExpression rangeTree = new RangeExpression(n1, n2);
          Expression srcParentNode = startMin.getParentExp();
          Expression tarParentNode = endMax.getParentExp();

          srcParentNode.findAndSetChild(startMin.getCurrentExp(), rangeTree);
          tarParentNode.findAndSetChild(endMax.getCurrentExp(), new TrueExpression(null));

          if (deleteExp.size() > 0) {
            // There are some expression to Delete as they are Redundant after Range Formation.
            for (FilterModificationNode trueExp : deleteExp) {
              trueExp.getParentExp()
                  .findAndSetChild(trueExp.getCurrentExp(), new TrueExpression(null));
            }
          }
        }
      }
    }
  }

  private Map<String, List<FilterModificationNode>> convertFilterTreeToMap() {
    // Traverse the Filter Tree and add the nodes in filterExpressionMap.
    // Only those nodes will be added which has got LessThan, LessThanEqualTo
    // GreaterThan, GreaterThanEqualTo Expression Only.
    Map<String, List<FilterModificationNode>> filterExpressionMap =
        new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);

    // Traverse the Tree.
    fillExpressionMap(filterExpressionMap, null, null);

    return filterExpressionMap;
  }

  private void evaluateOrExpression(Expression currentNode, Expression orExpChild) {
    Map<String, List<FilterModificationNode>> filterExpressionMapNew =
        new HashMap<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
    fillExpressionMap(filterExpressionMapNew, orExpChild, currentNode);
    replaceWithRangeExpression(filterExpressionMapNew);
    filterExpressionMapNew.clear();
  }

  private void fillExpressionMap(Map<String, List<FilterModificationNode>> filterExpressionMap,
      Expression currentNode, Expression parentNode) {
    if (null == currentNode) {
      currentNode = this.getExpr();
      parentNode  = currentNode;
    }

    // if the parentNode is a ANDExpression and the current node is LessThan, GreaterThan
    // then add the node into filterExpressionMap.
    if ((parentNode instanceof AndExpression) && (isLessThanGreaterThanExp(currentNode)
        && eligibleForRangeExpConversion(currentNode))) {
      addFilterExpressionMap(filterExpressionMap, currentNode, parentNode);
    }
    // In case of Or Exp we have to evaluate both the subtrees of expression separately
    // else it will combine the results of both the subtrees into one expression
    // which wont give us correct result
    if (currentNode instanceof OrExpression) {
      Expression leftChild = ((OrExpression) currentNode).left;
      Expression rightChild = ((OrExpression) currentNode).right;
      if (null != leftChild) {
        evaluateOrExpression(currentNode, leftChild);
      }
      if (null != rightChild) {
        evaluateOrExpression(currentNode, rightChild);
      }
    } else {
      for (Expression exp : currentNode.getChildren()) {
        if (null != exp) {
          fillExpressionMap(filterExpressionMap, exp, currentNode);
        }
      }
    }
  }

  private void addFilterExpressionMap(Map<String, List<FilterModificationNode>> filterExpressionMap,
      Expression currentNode, Expression parentNode) {
    String colName = getColumnName(currentNode);
    ExpressionType expType = getExpressionType(currentNode);

    FilterModificationNode filterExpression =
        new FilterModificationNode(currentNode, parentNode, expType);

    if (null == filterExpressionMap.get(colName)) {
      filterExpressionMap.put(colName, new ArrayList<FilterModificationNode>());
    }
    filterExpressionMap.get(colName).add(filterExpression);
  }

  /**
   * This method checks if the Expression is among LessThan, LessThanEqualTo,
   * GreaterThan or GreaterThanEqualTo
   *
   * @param expr
   * @return
   */
  private boolean isLessThanGreaterThanExp(Expression expr) {
    return (expr instanceof LessThanEqualToExpression) || (expr instanceof LessThanExpression)
        || (expr instanceof GreaterThanEqualToExpression)
        || (expr instanceof GreaterThanExpression);
  }

  /**
   * This method verifies if the Expression is qualified for Range Expression conversion.
   *
   * @param expChild
   * @return
   */
  private boolean eligibleForRangeExpConversion(Expression expChild) {
    for (Expression exp : expChild.getChildren()) {
      if (exp instanceof ColumnExpression) {
        return ((ColumnExpression) exp).isDimension() &&
            !(((ColumnExpression) exp).getDimension().getDataType().isComplexType());
      }
    }
    return false;
  }

  /**
   * This method returns the Column name from the ColumnExpression ExpressionType.
   *
   * @param andNode
   * @return
   */
  private String getColumnName(Expression andNode) {
    // returns the Column Name from Column Expression.
    for (Expression exp : andNode.getChildren()) {
      if (exp instanceof ColumnExpression) {
        return ((ColumnExpression) exp).getColumnName();
      }
    }
    return null;
  }

  /**
   * This method returns the Value from the Literal ExpressionType.
   *
   * @param exp
   * @return
   */
  private Object getLiteralValue(Expression exp) {
    for (Expression expr : exp.getChildren()) {
      if (expr instanceof LiteralExpression) {
        return (((LiteralExpression) expr).getLiteralExpValue());
      }
    }
    return null;
  }

  /**
   * This method returns the DataType of the Literal Expression.
   *
   * @param exp
   * @return
   */
  private DataType getLiteralDataType(Expression exp) {
    for (Expression expr : exp.getChildren()) {
      if (expr instanceof LiteralExpression) {
        return (((LiteralExpression) expr).getLiteralExpDataType());
      }
    }
    return null;
  }

  /**
   * This method returns the DataType of the Literal Expression.
   *
   * @param exp
   * @return
   */
  private LiteralExpression getChildLiteralExpression(Expression exp) {
    for (Expression expr : exp.getChildren()) {
      if (expr instanceof LiteralExpression) {
        return ((LiteralExpression) expr);
      }
    }
    return null;
  }

  /**
   * This method returns the ExpressionType based on the Expression.
   *
   * @param exp
   * @return
   */
  private ExpressionType getExpressionType(Expression exp) {
    // return the expressionType. Note among the children of the
    // andNode one should be columnExpression others should be
    // LessThan, LessThanEqualTo, GreaterThan, GreaterThanEqualTo.
    //
    if (exp instanceof LessThanExpression) {
      return LESSTHAN;
    } else if (exp instanceof LessThanEqualToExpression) {
      return LESSTHAN_EQUALTO;
    } else if (exp instanceof GreaterThanExpression) {
      return GREATERTHAN;
    } else if (exp instanceof GreaterThanEqualToExpression) {
      return GREATERTHAN_EQUALTO;
    } else {
      return FALSE;
    }
  }

  /**
   * This method checks if the Source Expression matches with Target Expression.
   *
   * @param src
   * @param tar
   * @return
   */
  private boolean matchExpType(ExpressionType src, ExpressionType tar) {
    return (((src == LESSTHAN) || (src == LESSTHAN_EQUALTO)) && ((tar == GREATERTHAN) || (tar
        == GREATERTHAN_EQUALTO))) || (((src == GREATERTHAN) || (src == GREATERTHAN_EQUALTO)) && (
        (tar == LESSTHAN) || (tar == LESSTHAN_EQUALTO)));
  }

  /**
   * This Method Traverses the Expression Tree to find the corresponding node of the Range
   * Expression. If one node of Range Expression is LessThan then a corresponding GreaterThan
   * will be chosen or vice versa.
   *
   * @param currentNode
   * @param parentNode
   * @return
   */
  private Expression traverseTree(Expression currentNode, Expression parentNode) {
    Expression result = null;

    if (null == parentNode) {
      currentNode = this.getExpr();
      parentNode = currentNode;
    }

    if (!this.getSrcNode().equals(currentNode) && isLessThanGreaterThanExp(currentNode)) {
      String srcColumnName = getColumnName(this.getSrcNode());
      String tarColumnName = getColumnName(currentNode);
      ExpressionType srcExpType = getExpressionType(this.getSrcNode());
      ExpressionType tarExpType = getExpressionType(currentNode);

      if ((null != srcColumnName) && (null != tarColumnName) && (srcColumnName
          .equals(tarColumnName)) && (srcExpType != ExpressionType.FALSE) && (tarExpType
          != ExpressionType.FALSE) && ((matchExpType(srcExpType, tarExpType)) && checkLiteralValue(
          this.getSrcNode(), currentNode))) {
        this.setTarNode(currentNode);
        this.setTarParentNode(parentNode);
        return parentNode;
      }
    }

    for (Expression exp : currentNode.getChildren()) {
      if (null != exp && !(exp instanceof RangeExpression)) {
        result = traverseTree(exp, currentNode);
        if (null != result) {
          return result;
        }
      }
    }
    return null;
  }

  /**
   * This method will check if the literal values attached to GreaterThan of GreaterThanEqualTo
   * is less or equal to LessThan and LessThanEqualTo literal.
   *
   * @param src
   * @param tar
   * @return
   */
  private boolean checkLiteralValue(Expression src, Expression tar) {
    ExpressionType srcExpressionType = getExpressionType(src);
    ExpressionType tarExpressionType = getExpressionType(tar);
    LiteralExpression srcLiteral = getChildLiteralExpression(src);
    LiteralExpression tarLiteral = getChildLiteralExpression(tar);

    ExpressionResult srcExpResult = srcLiteral.evaluate(null);
    ExpressionResult tarExpResult = tarLiteral.evaluate(null);

    switch (srcExpressionType) {
      case LESSTHAN:
      case LESSTHAN_EQUALTO:
        switch (tarExpressionType) {
          case GREATERTHAN:
            if (srcExpResult.compareTo(tarExpResult) > 0) {
              return true;
            }
            break;
          case GREATERTHAN_EQUALTO:
            if (srcExpResult.compareTo(tarExpResult) >= 0) {
              return true;
            }
            break;
        }
        break;
      case GREATERTHAN:
      case GREATERTHAN_EQUALTO:
        switch (tarExpressionType) {
          case LESSTHAN:
            if (srcExpResult.compareTo(tarExpResult) < 0) {
              return true;
            }
            break;
          case LESSTHAN_EQUALTO:
            if (srcExpResult.compareTo(tarExpResult) <= 0) {
              return true;
            }
            break;
        }
        break;
    }

    return false;
  }
}