/*
 * 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.flink.table.plan.metadata

import org.apache.flink.table.functions.sql.internal.{SqlRuntimeFilterBuilderFunction, SqlRuntimeFilterFunction}
import org.apache.flink.table.plan.metadata.SelectivityEstimator._
import org.apache.flink.table.plan.schema.{FlinkTable, TableSourceTable}
import org.apache.flink.table.plan.stats._
import org.apache.flink.table.plan.util.{FlinkRelOptUtil, FlinkRexUtil, PartitionPredicateExtractor, RexNodeExtractor}
import org.apache.flink.table.sources.PartitionableTableSource

import org.apache.calcite.avatica.util.DateTimeUtils
import org.apache.calcite.plan.{RelOptPredicateList, RelOptUtil}
import org.apache.calcite.rel.RelNode
import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFamily}
import org.apache.calcite.rel.core.TableScan
import org.apache.calcite.rel.metadata.RelMdUtil
import org.apache.calcite.rex._
import org.apache.calcite.sql.`type`.SqlTypeFamily
import org.apache.calcite.sql.fun.SqlStdOperatorTable._
import org.apache.calcite.sql.{SqlKind, SqlOperator}
import org.apache.calcite.util.{ImmutableBitSet, TimeString}

import java.lang.{Double => JDouble}

import scala.collection.JavaConversions._
import scala.collection.mutable

/**
  * Estimates selectivity of rows meeting a filter predicate on a RelNode.
  *
  * <p>For some unsupported cases, we use the default selectivity values
  * referred to [[org.apache.calcite.rel.metadata.RelMdUtil.guessSelectivity]].
  *
  * <p><strong>NOTES:</strong>
  * In current stage, we don't have advanced statistics such as sketches or histograms,
  * we have to assume uniform distribution.
  * We should update estimation methods in the future if advanced statistics is ready.
  *
  * <p>Part of the code is referred from Apache Spark.
  *
  * @param rel RelNode
  * @param mq  Metadata query
  */
class SelectivityEstimator(rel: RelNode, mq: FlinkRelMetadataQuery)
    extends RexVisitorImpl[Option[Double]](true) {

  private val rexBuilder = rel.getCluster.getRexBuilder
  private val maxCnfNodeCount = FlinkRelOptUtil.getMaxCnfNodeCount(rel)

  // these default values is referred to RelMdUtil#guessSelectivity
  private[flink] val defaultComparisonSelectivity = Some(0.5d)
  private[flink] val defaultEqualsSelectivity = Some(0.15d)
  private[flink] val defaultIsNullSelectivity = Some(0.1d)
  private[flink] val defaultIsNotNullSelectivity = Some(1.0 - defaultIsNullSelectivity.get)
  private[flink] val defaultLikeSelectivity = Some(0.05d)
  private[flink] val defaultSelectivity = Some(0.25d)

  /**
    * Returns a percentage of rows meeting a filter predicate on TableScan node.
    *
    * @param predicate predicate whose selectivity is to be estimated against scan's output.
    * @return estimated selectivity (between 0.0 and 1.0),
    *         or None if no reliable estimate can be determined.
    */
  def evaluate(predicate: RexNode): Option[Double] = {
    try {
      if (predicate == null) {
        Some(1.0)
      } else {
        val rexSimplify = new RexSimplify(
          rexBuilder, RelOptPredicateList.EMPTY, true, RexUtil.EXECUTOR)
        val simplifiedPredicate = rexSimplify.simplify(predicate)
        if (simplifiedPredicate.isAlwaysTrue) {
          Some(1.0)
        } else if (simplifiedPredicate.isAlwaysFalse) {
          Some(0.0)
        } else {
          simplifiedPredicate.accept(this)
        }
      }
    } catch {
      // if found unsupported operations, fallback
      case _: Throwable => None
    }
  }

  /**
    * Get partition field names from PartitionableTableSource.
    *
    * @return partition field names if the TableSource in scan is PartitionableTableSource,
    *         otherwise None.
    */
  private def getPartitionFieldNames(scan: TableScan): Option[Array[String]] = {
    val table = scan.getTable.unwrap(classOf[TableSourceTable])
    table match {
      case t: TableSourceTable =>
        t.tableSource match {
          case p: PartitionableTableSource =>
            Some(p.getPartitionFieldNames)
          case _ => None
        }
      case _ => None
    }
  }

  override def visitCall(call: RexCall): Option[Double] = {
    call.getOperator match {
      case AND =>
        val predicates = splitAndPredicate(call)
        // predicates.size == 1 indicates that
        // the RexInputRef of each sub-condition in this AND call are the same.
        if (predicates.size == 1) {
          estimateAndPredicate(call)
        } else {
          val selectivity = predicates.map(estimateOperand)
          Some(selectivity.product)
        }

      case OR =>
        val predicates = splitOrPredicate(call)
        if (predicates.size == 1) {
          predicates.head match {
            case c: RexCall if c.getOperator == IN => estimateSinglePredicate(c)
            case _ => estimateOrPredicate(call)
          }
        } else {
          val selectivity = predicates.map(estimateOperand)
          Some(math.min(1.0, selectivity.sum - selectivity.product))
        }

      case NOT =>
        val operands = call.getOperands
        val selectivity = estimateOperand(operands.head)
        Some(1.0 - selectivity)

      case _ =>
        estimateSinglePredicate(call)
    }
  }

  private def estimateOperand(operand: RexNode): Double = {
    val subSelectivity = operand.accept(this)
    if (subSelectivity != null) subSelectivity.getOrElse(1.0) else 1.0
  }

  /**
    * Returns a percentage of rows meeting a single condition in Filter node.
    *
    * We will do partition pruning with those supported partition-pruning predicates,
    * and the rowCount in [[org.apache.flink.table.plan.stats.TableStats]] should exclude the
    * rows of the pruned partitions. So we should ignore partition predicate here.
    *
    * @param singlePredicate predicate whose selectivity is to be estimated against scan's output.
    * @return an optional double value to show the percentage of rows meeting a given condition.
    *         It returns None if the condition is not supported.
    */
  private def estimateSinglePredicate(singlePredicate: RexCall): Option[Double] = {
    if (isSupportedPartitionPredicate(singlePredicate)) {
      return estimatePartitionPredicate(singlePredicate)
    }

    val operands = singlePredicate.getOperands
    singlePredicate.getOperator match {
      case EQUALS =>
        estimateComparison(EQUALS, operands.head, operands.last)

      case NOT_EQUALS =>
        val selectivity = estimateComparison(EQUALS, operands.head, operands.last)
        Some(1.0 - selectivity.getOrElse(1.0))

      case GREATER_THAN =>
        estimateComparison(GREATER_THAN, operands.head, operands.last)

      case GREATER_THAN_OR_EQUAL =>
        estimateComparison(GREATER_THAN_OR_EQUAL, operands.head, operands.last)

      case LESS_THAN =>
        estimateComparison(LESS_THAN, operands.head, operands.last)

      case LESS_THAN_OR_EQUAL =>
        estimateComparison(LESS_THAN_OR_EQUAL, operands.head, operands.last)

      case IS_NULL =>
        estimateIsNull(operands.head)

      case IS_NOT_NULL =>
        estimateIsNotNull(operands.head)

      case IN =>
        estimateIn(operands.head, operands.slice(1, operands.size))

      case RelMdUtil.ARTIFICIAL_SELECTIVITY_FUNC =>
        Option(RelMdUtil.getSelectivityValue(singlePredicate))

      case LIKE =>
        defaultLikeSelectivity

      // RuntimeFilter should not break Cost's estimate.
      // (Because it may be useless, just advance Join's filtering only).
      case _: SqlRuntimeFilterBuilderFunction | _: SqlRuntimeFilterFunction =>
        Some(1.0D)

      case _ =>
        defaultSelectivity
    }
  }

  /**
    * Decomposes a predicate into a list of expressions that are AND'ed together,
    * and the expressions with same RexInputRefs will be converted into an AND.
    * e.g.
    * input predicate: `a > 10 AND b < 40.0 AND a < 20 AND CAST(b AS INTEGER) > 30`
    * output expressions: `a > 10 AND a < 20` and `b < 40.0` and `CAST(b AS INTEGER) > 30`
    */
  private def splitAndPredicate(call: RexCall): Seq[RexNode] = {
    val cnf = FlinkRexUtil.toCnf(rexBuilder, maxCnfNodeCount, call)
    val conjunctions = RelOptUtil.conjunctions(cnf)

    val combinations = new mutable.HashMap[String, mutable.ListBuffer[RexNode]]()
    val nonCombinations = new mutable.ListBuffer[RexNode]()

    val visitor = new RexVisitorImpl[Unit](true) {
      override def visitCall(call: RexCall): Unit = {
        call.getOperator match {
          case AND => throw new RuntimeException("This should not happen.")
          // TODO supports more operators
          case GREATER_THAN | GREATER_THAN_OR_EQUAL | LESS_THAN | LESS_THAN_OR_EQUAL =>
            val (left, right) = (call.operands.get(0), call.operands.get(1))
            (left, right) match {
              case (i: RexInputRef, _: RexLiteral) =>
                combinations.getOrElseUpdate(i.toString, mutable.ListBuffer[RexNode]()) += call
              case (_: RexLiteral, i: RexInputRef) =>
                combinations.getOrElseUpdate(i.toString, mutable.ListBuffer[RexNode]()) += call
              case _ => nonCombinations += call
            }
          case _ => nonCombinations += call
        }
      }
    }

    conjunctions.foreach(_.accept(visitor))

    val combined = combinations.values.map { r =>
      RexUtil.composeConjunction(rexBuilder, r, false)
    }.toSeq

    combined ++ nonCombinations
  }

  /**
    * Decomposes a predicate into a list of expressions that are OR'ed together,
    * and the expressions with same RexInputRefs connected with EQUALS will be converted into an IN.
    * e.g.
    * input predicate: `a = 10 or a = 20 OR a > 40.0 OR a < 0 OR b > 30`
    * output expressions: `a IN (10, 20)` and `(a > 40 OR a < 0)` and `b > 30`
    */
  private def splitOrPredicate(call: RexCall): Seq[RexNode] = {
    val combinations = new mutable.HashMap[String, mutable.ListBuffer[RexNode]]()
    val nonCombinations = new mutable.ListBuffer[RexNode]()

    val visitor = new RexVisitorImpl[Unit](true) {
      override def visitCall(call: RexCall): Unit = {
        call.getOperator match {
          // TODO supports more operators
          case GREATER_THAN | GREATER_THAN_OR_EQUAL | LESS_THAN | LESS_THAN_OR_EQUAL | EQUALS =>
            val (left, right) = (call.operands.get(0), call.operands.get(1))
            (left, right) match {
              case (i: RexInputRef, _: RexLiteral) =>
                combinations.getOrElseUpdate(i.toString, mutable.ListBuffer[RexNode]()) += call
              case (l: RexLiteral, i: RexInputRef) =>
                val newCall = if (call.getOperator == EQUALS) {
                  call.clone(call.getType, List(i, l))
                } else {
                  call
                }
                combinations.getOrElseUpdate(i.toString, mutable.ListBuffer[RexNode]()) += newCall
              case _ => nonCombinations += call
            }
          case _ => nonCombinations += call
        }
      }
    }

    val disjunctions = RelOptUtil.disjunctions(call)
    disjunctions.foreach(_.accept(visitor))

    val combined = combinations.values.flatMap { r =>
      val (equalsNodes, nonEqualsNodes) = r.partition(_.asInstanceOf[RexCall].getOperator == EQUALS)
      if (equalsNodes.size > 1) {
        val inputRef = equalsNodes.head.asInstanceOf[RexCall].getOperands.head
        val valuesNode = equalsNodes.map(_.asInstanceOf[RexCall].getOperands.last)
        val inNode = rexBuilder.makeCall(IN, List(inputRef) ++ valuesNode)
        if (nonEqualsNodes.isEmpty) {
          Seq(inNode)
        } else {
          Seq(inNode, RexUtil.composeDisjunction(rexBuilder, nonEqualsNodes, false))
        }
      } else {
        Seq(RexUtil.composeDisjunction(rexBuilder, r, false))
      }
    }.toSeq

    combined ++ nonCombinations
  }

  /**
    * Returns a percentage of rows meeting an AND condition with same RexInputRef in Filter node.
    * Each sub-condition in AND is a binary comparison condition,
    * and only binary comparison operator <, <=, >, >= are supported now.
    *
    * The condition is like: `a > 10 AND a < 20`
    *
    * @param andPredicate predicate whose selectivity is to be estimated against scan's output.
    * @return an optional double value to show the percentage of rows meeting a given condition.
    *         It returns None if the condition is not supported.
    */
  private def estimateAndPredicate(andPredicate: RexCall): Option[Double] = {
    val cnf = FlinkRexUtil.toCnf(rexBuilder, maxCnfNodeCount, andPredicate)
    val conjunctions = RelOptUtil.conjunctions(cnf)

    val inputRefs = mutable.HashSet[Int]()
    val supported = conjunctions.forall {
      case c: RexCall =>
        c.getOperator match {
          case GREATER_THAN | GREATER_THAN_OR_EQUAL | LESS_THAN | LESS_THAN_OR_EQUAL =>
            val (left, right) = (c.operands.get(0), c.operands.get(1))
            (left, right) match {
              case (i: RexInputRef, _: RexLiteral) => inputRefs += i.getIndex
              case (_: RexLiteral, i: RexInputRef) => inputRefs += i.getIndex
              case _ => throw new RuntimeException("This should not happen.")
            }
            // TODO support partition predicate and non-numeric type
            !isSupportedPartitionPredicate(c) &&
              canConvertToNumericType(left.getType) &&
              canConvertToNumericType(right.getType)
          case _ => throw new RuntimeException("This should not happen.")
        }
      case _ => throw new RuntimeException("This should not happen.")
    }
    require(inputRefs.size == 1)

    if (!supported) {
      val selectivity = conjunctions.map(estimateOperand)
      return Some(selectivity.product)
    }

    val predicateIntervals = conjunctions.map {
      case c: RexCall =>
        val (left, right) = (c.operands.get(0), c.operands.get(1))
        c.getOperator match {
          case GREATER_THAN =>
            (left, right) match {
              case (i: RexInputRef, l: RexLiteral) =>
                ValueInterval(literalToDouble(l), null, includeLower = false, includeUpper = false)
              case (l: RexLiteral, _: RexInputRef) =>
                ValueInterval(null, literalToDouble(l), includeLower = false, includeUpper = false)
              case _ => throw new RuntimeException("This should not happen.")
            }
          case GREATER_THAN_OR_EQUAL =>
            (left, right) match {
              case (_: RexInputRef, l: RexLiteral) =>
                ValueInterval(literalToDouble(l), null, includeLower = true, includeUpper = false)
              case (l: RexLiteral, _: RexInputRef) =>
                ValueInterval(null, literalToDouble(l), includeLower = false, includeUpper = true)
              case _ => throw new RuntimeException("This should not happen.")
            }
          case LESS_THAN =>
            (left, right) match {
              case (_: RexInputRef, l: RexLiteral) =>
                ValueInterval(null, literalToDouble(l), includeLower = false, includeUpper = false)
              case (l: RexLiteral, _: RexInputRef) =>
                ValueInterval(literalToDouble(l), null, includeLower = false, includeUpper = false)
              case _ => throw new RuntimeException("This should not happen.")
            }
          case LESS_THAN_OR_EQUAL =>
            (left, right) match {
              case (_: RexInputRef, l: RexLiteral) =>
                ValueInterval(null, literalToDouble(l), includeLower = false, includeUpper = true)
              case (l: RexLiteral, _: RexInputRef) =>
                ValueInterval(literalToDouble(l), null, includeLower = true, includeUpper = false)
              case _ => throw new RuntimeException("This should not happen.")
            }
          case _ => throw new RuntimeException("This should not happen.")
        }
      case _ => throw new RuntimeException("This should not happen.")
    }

    estimateAndNumericComparison(inputRefs.head, predicateIntervals)
  }

  /**
    * Returns a percentage of rows meeting an AND condition and
    * each sub-condition is a binary numeric comparison expression.
    * This method evaluate expression for Numeric/Boolean/Date/Time/Timestamp columns.
    */
  private def estimateAndNumericComparison(
      inputRefIndex: Int,
      intervals: Seq[ValueInterval]): Option[Double] = {
    require(intervals.nonEmpty)
    val default = Some(math.pow(defaultComparisonSelectivity.get, intervals.size))

    val columnInterval = mq.getColumnInterval(rel, inputRefIndex)
    if (columnInterval == null) {
      return default
    }

    // convert values in columnInterval to double type
    val convertedInterval = convertValueInterval(columnInterval, SqlTypeFamily.NUMERIC)
    val andInterval = intervals.foldLeft(convertedInterval)(ValueInterval.intersect)

    columnInterval match {
      case ValueInterval.empty => Some(0.0)
      case ValueInterval.infinite => default
      case _: LeftSemiInfiniteValueInterval => default
      case _: RightSemiInfiniteValueInterval => default
      case v: FiniteValueInterval =>
        andInterval match {
          case ValueInterval.empty => Some(0.0)
          case lv: FiniteValueInterval =>
            // TODO not take include min/max into consideration now
            val (min, max) = (comparableToDouble(v.lower), comparableToDouble(v.upper))
            val (litMin, litMax) = (comparableToDouble(lv.lower), comparableToDouble(lv.upper))
            Some((litMax - litMin) / (max - min))
          case _ =>
            // predicate like `a > 10 AND a > 20` had been simplified in `evaluate` method
            throw new RuntimeException("This should not happen.")
        }
    }
  }

  /**
    * Returns a percentage of rows meeting an OR condition with same RexInputRef in Filter node.
    * Each sub-condition in OR is a binary comparison condition,
    * and only binary comparison operator <, <=, >, >= are supported now.
    *
    * The condition is like: `a > 20 OR a < 10`
    *
    * @param orPredicate predicate whose selectivity is to be estimated against scan's output.
    * @return an optional double value to show the percentage of rows meeting a given condition.
    *         It returns None if the condition is not supported.
    */
  private def estimateOrPredicate(orPredicate: RexCall): Option[Double] = {
    val disjunctions = RelOptUtil.disjunctions(orPredicate)
    val result = disjunctions.map(estimateOperand)
    Some(math.min(1.0, result.sum))
  }

  /**
    * Returns whether a predicate is supported partition-pruning predicate.
    *
    * @param singlePredicate predicate whose selectivity is to be estimated against scan's output.
    * @return true if the TableSource in scan is PartitionableTableSource and the columns in the
    *         predicate are all partition columns, otherwise false.
    */
  private def isSupportedPartitionPredicate(singlePredicate: RexCall): Boolean = {
    rel match {
      case ts: TableScan =>
        val partitionFieldNames = getPartitionFieldNames(ts)
        partitionFieldNames match {
          // partitionable table
          case Some(names) if names.nonEmpty =>
            val (convertedPredicates, unconvertedRexNodes) =
              RexNodeExtractor.extractConjunctiveConditions(
                singlePredicate,
                maxCnfNodeCount,
                ts.getRowType.getFieldNames,
                ts.getCluster.getRexBuilder,
                FlinkRelOptUtil.getFunctionCatalog(ts))
            val (_, nonPartitionPredicates) =
              PartitionPredicateExtractor.extractPartitionPredicates(convertedPredicates, names)
            unconvertedRexNodes.isEmpty && nonPartitionPredicates.isEmpty

          // non-partitionable table
          case _ => false
        }
      case _ => false
    }
  }

  /**
    * Returns a percentage of rows meeting a partition expression.
    *
    * @param partitionPredicate partition predicate whose selectivity is to be estimated against
    *                           scan's output.
    * @return an optional double value to show the percentage of rows meeting a given condition.
    *         It returns None if the condition is not supported.
    */
  private def estimatePartitionPredicate(partitionPredicate: RexCall): Option[Double] = {
    assert(this.rel.isInstanceOf[TableScan])
    val table = rel.asInstanceOf[TableScan].getTable.unwrap(classOf[TableSourceTable])
    val partitionableTableSource = table match {
      case t: TableSourceTable =>
        t.tableSource match {
          case p: PartitionableTableSource => p
          case _ => throw new RuntimeException("This should not happen.")
        }
      case _ => throw new RuntimeException("This should not happen.")
    }

    // TODO deal with filter partial push-down
    if (partitionableTableSource.isFilterPushedDown) {
      // Ignore any predicates on partition columns because we have already
      // accounted for these in the Table row count.
      Some(1.0)
    } else {
      // We do not use PartitionableTableSource#getAllPartitions method to get the number of
      // partitions, because this method may involve some heavy work(e.g. reads metadata using I/O).
      // We assume the number of partitions is 100 here.
      val numOfPartitions = 100
      partitionPredicate.getOperator match {
        case EQUALS => Some(1.0 / numOfPartitions)
        case NOT_EQUALS => Some(1.0 - 1.0 / numOfPartitions)
        case IS_NULL => Some(0.0)
        case IS_NOT_NULL => Some(1.0)
        case IN =>
          val inSet = partitionPredicate.getOperands.subList(1, partitionPredicate.getOperands.size)
          checkInSet(inSet)
          val sizeOfInSet = inSet.size()
          Some(Math.min(1.0, sizeOfInSet * 1.0 / numOfPartitions))
        case _ => Some(5 * (1.0 / numOfPartitions))
      }
    }
  }

  /**
    * Returns a percentage of rows meeting a binary comparison expression containing two columns.
    *
    * We will use default comparison selectivity(0.5), in the following cases:
    * 1. the type is not supported
    * 2. the operator is not supported
    * 3. the specified column's stats is empty
    *
    * @param op    a binary comparison operator, including =, <=>, <, <=, >, >=
    * @param left  the left RexInputRef
    * @param right the right RexInputRef
    * @return an optional double value to show the percentage of rows meeting a given condition.
    *         It returns None if no statistics collected for a given column.
    */
  private def estimateComparison(op: SqlOperator, left: RexNode, right: RexNode): Option[Double] = {
    if (!isSupportedComparisonType(left.getType) || !isSupportedComparisonType(right.getType)) {
      val default = op match {
        case EQUALS => defaultEqualsSelectivity
        case _ => defaultComparisonSelectivity
      }
      return default
    }

    op match {
      case EQUALS => (left, right) match {
        case (i: RexInputRef, l: RexLiteral) => estimateEquals(i, l)
        case (l: RexLiteral, i: RexInputRef) => estimateEquals(i, l)
        case (l: RexInputRef, r: RexInputRef) => estimateComparison(EQUALS, l, r)
        case _ => defaultEqualsSelectivity
      }
      case LESS_THAN => (left, right) match {
        case (i: RexInputRef, l: RexLiteral) => estimateComparison(LESS_THAN, i, l)
        case (l: RexLiteral, i: RexInputRef) => estimateComparison(GREATER_THAN, i, l)
        case (l: RexInputRef, r: RexInputRef) => estimateComparison(LESS_THAN, l, r)
        case _ => defaultComparisonSelectivity
      }
      case LESS_THAN_OR_EQUAL => (left, right) match {
        case (i: RexInputRef, l: RexLiteral) => estimateComparison(LESS_THAN_OR_EQUAL, i, l)
        case (l: RexLiteral, i: RexInputRef) => estimateComparison(GREATER_THAN_OR_EQUAL, i, l)
        case (l: RexInputRef, r: RexInputRef) => estimateComparison(LESS_THAN_OR_EQUAL, l, r)
        case _ => defaultComparisonSelectivity
      }
      case GREATER_THAN => (left, right) match {
        case (i: RexInputRef, l: RexLiteral) => estimateComparison(GREATER_THAN, i, l)
        case (l: RexLiteral, i: RexInputRef) => estimateComparison(LESS_THAN, i, l)
        case (l: RexInputRef, r: RexInputRef) => estimateComparison(GREATER_THAN, l, r)
        case _ => defaultComparisonSelectivity
      }
      case GREATER_THAN_OR_EQUAL => (left, right) match {
        case (i: RexInputRef, l: RexLiteral) => estimateComparison(GREATER_THAN_OR_EQUAL, i, l)
        case (l: RexLiteral, i: RexInputRef) => estimateComparison(LESS_THAN_OR_EQUAL, i, l)
        case (l: RexInputRef, r: RexInputRef) => estimateComparison(GREATER_THAN_OR_EQUAL, l, r)
        case _ => defaultComparisonSelectivity
      }
      case _ => defaultComparisonSelectivity
    }
  }

  /**
    * Returns a percentage of rows meeting an equality (=) expression.
    *
    * @param inputRef a RexInputRef
    * @param literal  a literal value (or constant)
    * @return an optional double value to show the percentage of rows meeting a given condition.
    *         It returns None if no statistics collected for a given column.
    */
  private def estimateEquals(inputRef: RexInputRef, literal: RexLiteral): Option[Double] = {
    if (literal.isNull) {
      return estimateIsNull(inputRef)
    }
    val columnInterval = mq.getColumnInterval(rel, inputRef.getIndex)
    if (columnInterval == null) {
      return defaultEqualsSelectivity
    }
    val convertedInterval = convertValueInterval(columnInterval, inputRef.getType)
    convertedInterval match {
      case ValueInterval.infinite => defaultEqualsSelectivity
      case ValueInterval.empty => Some(0.0)
      case _ =>
        if (ValueInterval.contains(convertedInterval, literalToComparable(literal))) {
          val bitSetOfInputRef = ImmutableBitSet.of(inputRef.getIndex)
          val ndv = mq.getDistinctRowCount(rel, bitSetOfInputRef, null)
          if (ndv == null) {
            defaultEqualsSelectivity
          } else {
            // there is no possible that ndv < 1.0 because of the protection in RelMetadataQuery.
            Some(1.0 / ndv)
          }
        } else {
          Some(0.0)
        }
    }
  }

  /**
    * Returns a percentage of rows meeting a binary comparison expression.
    *
    * @param op       a binary comparison operator, including <, <=, >, >=
    * @param inputRef a RexInputRef
    * @param literal  a literal value (or constant)
    * @return an optional double value to show the percentage of rows meeting a given condition.
    *         It returns None if no statistics collected for a given column.
    */
  private def estimateComparison(
      op: SqlOperator,
      inputRef: RexInputRef,
      literal: RexLiteral): Option[Double] = {
    if (literal.isNull) {
      throw new IllegalArgumentException("Numeric comparison does not support null literal here.")
    }
    if (canConvertToNumericType(inputRef.getType)) {
      estimateNumericComparison(op, inputRef, literal)
    } else {
      // TODO: It is difficult to support binary comparisons for non-numeric type
      // without advanced statistics like histogram.
      defaultComparisonSelectivity
    }
  }

  /**
    * Returns a percentage of rows meeting a binary numeric comparison expression.
    * This method evaluate expression for Numeric/Boolean/Date/Time/Timestamp columns.
    *
    * @param op       a binary comparison operator, including <, <=, >, >=
    * @param inputRef a RexInputRef
    * @param literal  a literal value (or constant)
    * @return an optional double value to show the percentage of rows meeting a given condition.
    *         It returns None if no statistics collected for a given column.
    */
  private def estimateNumericComparison(
      op: SqlOperator,
      inputRef: RexInputRef,
      literal: RexLiteral): Option[Double] = {
    val columnInterval = mq.getColumnInterval(rel, inputRef.getIndex)
    if (columnInterval == null) {
      return defaultComparisonSelectivity
    }
    columnInterval match {
      case ValueInterval.infinite => defaultComparisonSelectivity
      case ValueInterval.empty => Some(0.0)
      case _ =>
        val (min, includeMin) = columnInterval match {
          case hasLower: WithLower => (comparableToDouble(hasLower.lower), hasLower.includeLower)
          case _ => (null, true)
        }
        val (max, includeMax) = columnInterval match {
          case hasUpper: WithUpper => (comparableToDouble(hasUpper.upper), hasUpper.includeUpper)
          case _ => (null, true)
        }
        val lit = literalToDouble(literal)
        val (noOverlap, completeOverlap) = op match {
          case LESS_THAN => (
            greaterThanOrEqualTo(min, lit),
            if (includeMax) lessThan(max, lit) else lessThanOrEqualTo(max, lit))
          case LESS_THAN_OR_EQUAL => (
            if (includeMin) greaterThan(min, lit) else greaterThanOrEqualTo(min, lit),
            lessThanOrEqualTo(max, lit))
          case GREATER_THAN => (
            lessThanOrEqualTo(max, lit),
            if (includeMin) greaterThan(min, lit) else greaterThanOrEqualTo(min, lit))
          case GREATER_THAN_OR_EQUAL => (
            if (includeMax) lessThan(max, lit) else lessThanOrEqualTo(max, lit),
            greaterThanOrEqualTo(min, lit))
        }

        val selectivity = if (noOverlap) {
          0.0
        } else if (completeOverlap) {
          1.0
        } else if (min != null && max != null) {
          val bitSetOfInputRef = ImmutableBitSet.of(inputRef.getIndex)
          val ndv = mq.getDistinctRowCount(rel, bitSetOfInputRef, null)
          op match {
            // there is no possible that ndv < 1.0 because of the protection in RelMetadataQuery.
            case LESS_THAN =>
              if (doubleEquals(lit, max)) {
                if (includeMax) {
                  Option(ndv).map(v => 1.0 - (1.0 / v))
                    .getOrElse(defaultComparisonSelectivity.get)
                } else {
                  1.0
                }
              } else {
                // TODO not take includeMin into consideration now
                (lit - min) / (max - min)
              }
            case LESS_THAN_OR_EQUAL =>
              if (doubleEquals(lit, min)) {
                if (includeMin) {
                  // The boundary value is the only satisfying value.
                  Option(ndv).map(1.0 / _).getOrElse(defaultComparisonSelectivity.get)
                } else {
                  0.0
                }
              } else {
                // TODO not take includeMax into consideration now
                (lit - min) / (max - min)
              }
            case GREATER_THAN =>
              if (doubleEquals(lit, min)) {
                if (includeMin) {
                  Option(ndv).map(v => 1.0 - (1.0 / v))
                    .getOrElse(defaultComparisonSelectivity.get)
                } else {
                  1.0
                }
              } else {
                // TODO not take includeMin into consideration now
                (max - lit) / (max - min)
              }
            case GREATER_THAN_OR_EQUAL =>
              if (doubleEquals(lit, max)) {
                if (includeMax) {
                  Option(ndv).map(1.0 / _).getOrElse(defaultComparisonSelectivity.get)
                } else {
                  0.0
                }
              } else {
                // TODO not take includeMax into consideration now
                (max - lit) / (max - min)
              }
          }
        } else {
          defaultComparisonSelectivity.get
        }
        Some(selectivity)
    }
  }

  /**
    * Returns a percentage of rows meeting a binary comparison expression containing two columns.
    * In SQL queries, we also see predicate expressions involving two columns
    * such as "column-1 (op) column-2" where column-1 and column-2 belong to same table.
    * Note that, if column-1 and column-2 belong to different tables, then it is a join
    * operator's work, NOT a filter operator's work.
    *
    * @param op    a binary comparison operator, including =, <, <=, >, >=
    * @param left  the left RexInputRef
    * @param right the right RexInputRef
    * @return an optional double value to show the percentage of rows meeting a given condition.
    *         It returns None if no statistics collected for a given column.
    */
  private def estimateComparison(
      op: SqlOperator,
      left: RexInputRef,
      right: RexInputRef): Option[Double] = {
    if (canConvertToNumericType(left.getType) && canConvertToNumericType(right.getType)) {
      estimateNumericComparison(op, left, right)
    } else {
      // TODO: It is difficult to support binary comparisons for non-numeric type
      // without advanced statistics like histogram.
      op match {
        case EQUALS => defaultEqualsSelectivity
        case _ => defaultComparisonSelectivity
      }
    }
  }

  /**
    * Returns a percentage of rows meeting a binary numeric comparison expression
    * containing two columns.
    * This method evaluate expression for Numeric/Boolean/Date/Time/Timestamp columns.
    *
    * @param op    a binary comparison operator, including =, <, <=, >, >=
    * @param left  the left RexInputRef
    * @param right the right RexInputRef
    * @return an optional double value to show the percentage of rows meeting a given condition.
    *         It returns None if no statistics collected for a given column.
    */
  private def estimateNumericComparison(
      op: SqlOperator,
      left: RexInputRef,
      right: RexInputRef): Option[Double] = {
    val selectivityWithoutStats = op match {
      case EQUALS => defaultEqualsSelectivity
      case _ => defaultComparisonSelectivity
    }
    val leftInterval = mq.getColumnInterval(this.rel, left.getIndex)
    val rightInterval = mq.getColumnInterval(this.rel, right.getIndex)
    if (leftInterval == null ||
      leftInterval == ValueInterval.infinite ||
      rightInterval == null ||
      rightInterval == ValueInterval.infinite) {
      return selectivityWithoutStats
    } else if (leftInterval == ValueInterval.empty || rightInterval == ValueInterval.empty) {
      return Some(0.0)
    }

    val leftNullCount = mq.getColumnNullCount(this.rel, left.getIndex)
    val rightNullCount = mq.getColumnNullCount(this.rel, right.getIndex)
    if (leftNullCount == null || rightNullCount == null) {
      return selectivityWithoutStats
    }

    val bitSetOfLeftInputRef = ImmutableBitSet.of(left.getIndex)
    val leftNdv = mq.getDistinctRowCount(this.rel, bitSetOfLeftInputRef, null)
    val bitSetOfRightInputRef = ImmutableBitSet.of(right.getIndex)
    val rightNdv = mq.getDistinctRowCount(this.rel, bitSetOfRightInputRef, null)
    if (op.equals(EQUALS) &&
      (!(leftInterval.isInstanceOf[FiniteValueInterval]
        && rightInterval.isInstanceOf[FiniteValueInterval])) ||
      leftNdv == null || rightNdv == null) {
      return defaultEqualsSelectivity
    }
    // left interval
    val (leftMin, leftIncludeMin) = leftInterval match {
      case hasLower: WithLower => (comparableToDouble(hasLower.lower), hasLower.includeLower)
      case _ => (null, true)
    }
    val (leftMax, leftIncludeMax) = leftInterval match {
      case hasUpper: WithUpper => (comparableToDouble(hasUpper.upper), hasUpper.includeUpper)
      case _ => (null, true)
    }
    val (rightMin, rightIncludeMin) = rightInterval match {
      case hasLower: WithLower => (comparableToDouble(hasLower.lower), hasLower.includeLower)
      case _ => (null, true)
    }
    val (rightMax, rightIncludeMax) = rightInterval match {
      case hasUpper: WithUpper => (comparableToDouble(hasUpper.upper), hasUpper.includeUpper)
      case _ => (null, true)
    }
    // determine the overlapping degree between predicate interval and column's interval
    val (noOverlap, completeOverlap) = op match {
      // Left < Right or Left <= Right
      // - no overlap:
      //      rightMin           rightMax     leftMin       leftMax
      // --------+------------------+------------+-------------+------->
      // - complete overlap: (If null values exists, we set it to partial overlap.)
      //      leftMin            leftMax      rightMin      rightMax
      // --------+------------------+------------+-------------+------->
      case LESS_THAN =>
        (
          greaterThanOrEqualTo(leftMin, rightMax),
          if (leftIncludeMax && rightIncludeMin) {
            lessThan(leftMax, rightMin)
          } else {
            lessThanOrEqualTo(leftMax, rightMin)
          })
      case LESS_THAN_OR_EQUAL =>
        (
          if (leftIncludeMin && rightIncludeMax) {
            greaterThan(leftMin, rightMax)
          } else {
            greaterThanOrEqualTo(leftMin, rightMax)
          },
          lessThanOrEqualTo(leftMax, rightMin))
      // Left > Right or Left >= Right
      // - no overlap:
      //      leftMin            leftMax      rightMin      rightMax
      // --------+------------------+------------+-------------+------->
      // - complete overlap: (If null values exists, we set it to partial overlap.)
      //      rightMin           rightMax     leftMin       leftMax
      // --------+------------------+------------+-------------+------->
      case GREATER_THAN =>
        (
          lessThanOrEqualTo(leftMax, rightMin),
          if (leftIncludeMin && rightIncludeMax) {
            greaterThan(leftMin, rightMax)
          } else {
            greaterThanOrEqualTo(leftMin, rightMax)
          })
      case GREATER_THAN_OR_EQUAL =>
        (
          if (leftIncludeMax && rightIncludeMin) {
            lessThan(leftMax, rightMin)
          } else {
            lessThanOrEqualTo(leftMax, rightMin)
          },
          greaterThanOrEqualTo(leftMin, rightMax))
      // Left = Right
      // - no overlap:
      //      leftMin            leftMax      rightMin      rightMax
      // --------+------------------+------------+-------------+------->
      //      rightMin           rightMax     leftMin       leftMax
      // --------+------------------+------------+-------------+------->
      // - complete overlap:
      //      leftMin            leftMin
      //      rightMin           rightMax
      // --------+------------------+------->
      case EQUALS =>
        // now, min/max/ndv is not null and ndv >= 0
        val isLeftSmallerThanRight = if (leftIncludeMax && rightIncludeMin) {
          leftMax < rightMin
        } else {
          leftMax <= rightMin
        }
        val isLeftBiggerThanRight = if (rightIncludeMax && leftIncludeMin) {
          rightMax < leftMin
        } else {
          rightMax <= leftMin
        }
        (
          isLeftSmallerThanRight || isLeftBiggerThanRight,
          (leftMin == rightMin) && (leftMax == rightMax)
            && (leftIncludeMin == rightIncludeMin)
            && (leftIncludeMax == rightIncludeMax)
            && (leftNdv == rightNdv))
    }
    val allNotNull = (leftNullCount == 0) && (rightNullCount == 0)
    val selectivity = if (noOverlap) {
      0.0
    } else if (completeOverlap && allNotNull) {
      1.0
    } else {
      // For partial overlap, we use an empirical value 1/3 as suggested by the book
      // "Database Systems, the complete book".
      1.0 / 3.0
    }
    Some(selectivity)
  }

  /**
    * Returns a percentage of rows meeting "IS NULL" condition.
    *
    * We will use default is_null selectivity(0.1), if the column stats is empty.
    *
    * @param input a RexNode
    * @return an optional double value to show the percentage of rows meeting a given condition
    *         It returns None if no statistics collected for a given column.
    */
  private def estimateIsNull(input: RexNode): Option[Double] = {
    val inputRef = convertToRexInputRef(input)
    val nullCount = mq.getColumnNullCount(this.rel, inputRef.getIndex)
    if (nullCount == null) {
      return defaultIsNullSelectivity
    }

    val rowCount = mq.getRowCount(this.rel)
    if (rowCount == null) {
      defaultIsNullSelectivity
    } else {
      // there is no possible that rowCount is 0 because of the protection in RelMetadataQuery.
      Some(Math.min(nullCount / rowCount, 1.0))
    }
  }

  /**
    * Returns a percentage of rows meeting "IS NOT NULL" condition.
    *
    * We will use default is_not_null selectivity(0.9), if the column stats is empty.
    *
    * @param input a RexNode
    * @return an optional double value to show the percentage of rows meeting a given condition
    *         It returns None if no statistics collected for a given column.
    */
  private def estimateIsNotNull(input: RexNode): Option[Double] = {
    val inputRef = convertToRexInputRef(input)
    val nullCount = mq.getColumnNullCount(this.rel, inputRef.getIndex)
    if (nullCount == null) {
      return defaultIsNotNullSelectivity
    }

    val rowCount = mq.getRowCount(this.rel)
    val selectivity = if (rowCount == 0) {
      0.0
    } else {
      1.0 - (nullCount / rowCount)
    }
    Some(selectivity)
  }

  /**
    * Returns a percentage of rows meeting "IN" operator expression.
    *
    * @param input a RexNode
    * @param inSet a set of literal values
    * @return an optional double value to show the percentage of rows meeting a given condition
    *         It returns None if no statistics exists for a given column.
    */
  private def estimateIn(input: RexNode, inSet: Seq[RexNode]): Option[Double] = {
    val inputRef = convertToRexInputRef(input)
    checkInSet(inSet)

    val defaultInSelectivity = Some(Math.min(1.0, defaultEqualsSelectivity.get * inSet.size))

    if (!isSupportedComparisonType(input.getType)) {
      return defaultInSelectivity
    }
    val columnInterval = mq.getColumnInterval(this.rel, inputRef.getIndex)
    if (columnInterval == null) {
      val ndv = mq.getDistinctRowCount(rel, ImmutableBitSet.of(inputRef.getIndex), null)
      return if (ndv == null) {
        defaultInSelectivity
      } else {
        Some(inSet.size / ndv)
      }
    }

    val interval = convertValueInterval(columnInterval, inputRef.getType)
    interval match {
      case ValueInterval.infinite => defaultInSelectivity
      case ValueInterval.empty => Some(0.0)
      case _ =>
        val validInSet = inSet.filter(
          v => ValueInterval.contains(interval, literalToComparable(v.asInstanceOf[RexLiteral])))

        if (validInSet.isEmpty) {
          Some(0.0)
        } else {
          val bitSetOfInputRef = ImmutableBitSet.of(inputRef.getIndex)
          val ndv = mq.getDistinctRowCount(rel, bitSetOfInputRef, null)
          if (ndv == null) {
            defaultInSelectivity
          } else {
            // there is no possible that ndv < 1.0 because of the protection in RelMetadataQuery.
            // newNdv should not be greater than the old ndv. For example, column has only 2 values
            // 1 and 6. The predicate column IN (1, 2, 3, 4, 5). validSet.size is 5.
            val newNdv = Math.min(ndv, validInSet.size)
            // return the filter selectivity. Without advanced statistics such as histograms,
            // we have to assume uniform distribution.
            Some(Math.min(1.0, newNdv * 1.0 / ndv))
          }
        }
    }
  }

  /**
    * Checks whether all values are literal. If contains non-literal values, throw an exception.
    */
  private def checkInSet(inSet: Seq[RexNode]): Unit = {
    if (inSet.exists(!_.isA(SqlKind.LITERAL))) {
      throw new IllegalArgumentException(s"${
        inSet.mkString(",")
      } contain some non-literal values.")
    }
  }

  /**
    * If a rexNode is RexInputRef type, convert it to RexInputRef; else throw an exception.
    */
  private def convertToRexInputRef(rexNode: RexNode): RexInputRef = {
    rexNode match {
      case i: RexInputRef => i
      case _ => throw new IllegalArgumentException(s"$rexNode is not a RexInputRef.")
    }
  }

  private def doubleEquals(v1: Double, v2: Double): Boolean = Math.abs(v1 - v2) < 1e-6

}

object SelectivityEstimator {

  /**
    * Convert ValueInterval,
    * for Numeric/Boolean/Date/Time/Timestamp, min/max will be converted to Double,
    * for CHARACTER, min/max will be converted to String.
    */
  def convertValueInterval(interval: ValueInterval, dataType: RelDataType): ValueInterval = {
    require(interval != null && dataType != null)
    convertValueInterval(interval, dataType.getFamily)
  }

  /**
    * Convert ValueInterval,
    * for Numeric/Boolean/Date/Time/Timestamp, min/max will be converted to Double,
    * for CHARACTER, min/max will be converted to String.
    */
  def convertValueInterval(
      interval: ValueInterval,
      typeFamily: RelDataTypeFamily): ValueInterval = {
    require(interval != null && typeFamily != null)
    interval match {
      case ValueInterval.empty | ValueInterval.infinite => interval
      case _ =>
        val (lower, includeLower) = interval match {
          case li: WithLower => (li.lower, li.includeLower)
          case _ => (null, false)
        }
        val (upper, includeUpper) = interval match {
          case ui: WithUpper => (ui.upper, ui.includeUpper)
          case _ => (null, false)
        }
        typeFamily match {
          case SqlTypeFamily.NUMERIC | SqlTypeFamily.BOOLEAN | SqlTypeFamily.DATE |
               SqlTypeFamily.TIME | SqlTypeFamily.TIMESTAMP =>
            ValueInterval(
              comparableToDouble(lower),
              comparableToDouble(upper),
              includeLower,
              includeUpper)
          case SqlTypeFamily.CHARACTER =>
            ValueInterval(
              lower.toString,
              upper.toString,
              includeLower,
              includeUpper)
          case _ => throw new UnsupportedOperationException(s"Unsupported typeFamily: $typeFamily")
        }
    }
  }

  /**
    * Get literal value as Comparable.
    *
    * The class may be different between literal value and min/max value,
    * and different classes can't be compared. So
    * for Numeric/Boolean/Date/Time/Timestamp, literal will be converted to Double uniformly,
    * for CHARACTER, literal will be converted to String.
    *
    * e.g. min/max values are Double: min=10.0 and max=100.0,
    * however the literal values are Integer: select * from tb where a in (20, 30, 40)
    */
  def literalToComparable(literal: RexLiteral): Comparable[_] = {
    if (!literal.isNull) {
      literal.getType.getFamily match {
        case SqlTypeFamily.NUMERIC | SqlTypeFamily.BOOLEAN | SqlTypeFamily.DATE |
             SqlTypeFamily.TIME | SqlTypeFamily.TIMESTAMP => literalToDouble(literal)
        case SqlTypeFamily.CHARACTER => literal.getValueAs(classOf[String])
        case _ => throw new UnsupportedOperationException(
          s"Can't get value as comparable from literal type: ${literal.getType}")
      }
    } else {
      null
    }
  }

  def comparableToDouble(value: Any): JDouble = {
    value match {
      case null => null
      case v: Number => v.doubleValue()
      case v: Boolean => if (v) 1.0 else 0.0
      // Returns days since 1970-01-01 for Date type value
      case v: java.sql.Date => JDouble.valueOf(v.getTime / DateTimeUtils.MILLIS_PER_DAY)
      // Returns milliseconds since 1970-01-01 00:00:00 for Timestamp type value
      case v: java.sql.Timestamp => JDouble.valueOf(v.getTime)
      // Returns milliseconds since midnight for Time type value
      case v: java.sql.Time => JDouble.valueOf(new TimeString(v.toString).getMillisOfDay)
      case _ =>
        throw new UnsupportedOperationException(
          s"Can't convert comparable type: ${value.getClass} to Double")
    }
  }

  def literalToDouble(literal: RexLiteral): JDouble = {
    if (!literal.isNull) {
      literal.getType.getFamily match {
        case SqlTypeFamily.NUMERIC =>
          literal.getValue3.toString.toDouble
        case SqlTypeFamily.BOOLEAN =>
          if (RexLiteral.booleanValue(literal)) 1.0 else 0.0
        case SqlTypeFamily.DATE | SqlTypeFamily.TIME | SqlTypeFamily.TIMESTAMP =>
          literal.getValue2.toString.toDouble
        case _ =>
          throw new UnsupportedOperationException(
            s"Can't get value as Double from literal type: ${literal.getType}")
      }
    } else {
      null
    }
  }

  def lessThan(left: Comparable[_], right: Comparable[_]): Boolean = {
    if (left == null || right == null) {
      false
    } else {
      left.asInstanceOf[Comparable[Any]].compareTo(right.asInstanceOf[Comparable[Any]]) < 0
    }
  }

  def lessThanOrEqualTo(left: Comparable[_], right: Comparable[_]): Boolean = {
    if (left == null || right == null) {
      false
    } else {
      left.asInstanceOf[Comparable[Any]].compareTo(right.asInstanceOf[Comparable[Any]]) <= 0
    }
  }

  def greaterThan(left: Comparable[_], right: Comparable[_]): Boolean = {
    if (left == null || right == null) {
      false
    } else {
      left.asInstanceOf[Comparable[Any]].compareTo(right.asInstanceOf[Comparable[Any]]) > 0
    }
  }

  def greaterThanOrEqualTo(left: Comparable[_], right: Comparable[_]): Boolean = {
    if (left == null || right == null) {
      false
    } else {
      left.asInstanceOf[Comparable[Any]].compareTo(right.asInstanceOf[Comparable[Any]]) >= 0
    }
  }

  /**
    * Currently, Numeric/Boolean/String/Date/Time/Timestamp are supported.
    */
  def isSupportedComparisonType(relType: RelDataType): Boolean = {
    relType.getFamily match {
      case SqlTypeFamily.NUMERIC | SqlTypeFamily.BOOLEAN | SqlTypeFamily.CHARACTER |
           SqlTypeFamily.DATE | SqlTypeFamily.TIME | SqlTypeFamily.TIMESTAMP => true
      case _ => false
    }
  }

  /**
    * Currently, value of Numeric/Boolean/Date/Time/Timestamp column can be converted to a number.
    */
  def canConvertToNumericType(relType: RelDataType): Boolean = {
    relType.getFamily match {
      case SqlTypeFamily.NUMERIC | SqlTypeFamily.BOOLEAN |
           SqlTypeFamily.DATE | SqlTypeFamily.TIME | SqlTypeFamily.TIMESTAMP => true
      case _ => false
    }
  }
}