/*
 * 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.calcite.FlinkTypeFactory
import org.apache.flink.table.functions.sql.SqlIncrSumAggFunction
import org.apache.flink.table.functions.utils.ScalarSqlFunction
import org.apache.flink.table.plan.FlinkJoinRelType
import org.apache.flink.table.plan.`trait`.RelModifiedMonotonicity
import org.apache.flink.table.plan.metadata.FlinkMetadata.ModifiedMonotonicityMeta
import org.apache.flink.table.plan.nodes.logical._
import org.apache.flink.table.plan.nodes.physical.batch.BatchExecGroupAggregateBase
import org.apache.flink.table.plan.nodes.physical.stream._
import org.apache.flink.table.plan.schema.{DataStreamTable, IntermediateRelNodeTable}
import org.apache.flink.table.plan.stats.{WithLower, WithUpper}

import org.apache.calcite.plan.hep.HepRelVertex
import org.apache.calcite.plan.volcano.RelSubset
import org.apache.calcite.rel.core._
import org.apache.calcite.rel.metadata._
import org.apache.calcite.rel.{AbstractRelNode, RelCollation, RelFieldCollation, RelNode}
import org.apache.calcite.rex.{RexCall, RexCallBinding, RexInputRef, RexNode}
import org.apache.calcite.sql.{SqlKind, SqlOperatorBinding}
import org.apache.calcite.sql.fun.{SqlCountAggFunction, SqlMinMaxAggFunction, SqlSumAggFunction, SqlSumEmptyIsZeroAggFunction}
import org.apache.calcite.sql.validate.SqlMonotonicity
import org.apache.calcite.sql.validate.SqlMonotonicity._
import org.apache.calcite.util.{ImmutableIntList, Util}

import java.lang.{Byte => JByte, Double => JDouble, Float => JFloat, Integer => JInt, Long => JLong, Short => JShort, String => JString}
import java.math.{BigDecimal => JBigDecimal}
import java.sql.{Date, Time, Timestamp}
import java.util
import java.util.{List => JList}

import scala.collection.JavaConversions._

/**
  * FlinkRelMdModifiedMonotonicity supplies a default implementation of
  * [[FlinkRelMetadataQuery.getRelModifiedMonotonicity]] for logical algebra.
  */
class FlinkRelMdModifiedMonotonicity private extends MetadataHandler[ModifiedMonotonicityMeta] {

  override def getDef: MetadataDef[ModifiedMonotonicityMeta] =
    FlinkMetadata.ModifiedMonotonicityMeta.DEF

  /**
    * Utility to create a RelModifiedMonotonicity which all fields is modified constant which
    * means all the field's value will not be modified.
    */
  def constants(rel: RelNode): RelModifiedMonotonicity = {
    new RelModifiedMonotonicity(Array.fill(rel.getRowType.getFieldCount)(CONSTANT))
  }

  def notMonotonic(rel: RelNode): RelModifiedMonotonicity = {
    new RelModifiedMonotonicity(Array.fill(rel.getRowType.getFieldCount)(NOT_MONOTONIC))
  }

  // ---------------------------  Abstract RelNode ------------------------------

  def getRelModifiedMonotonicity(rel: RelNode, mq: RelMetadataQuery): RelModifiedMonotonicity = {
    val rowSize = rel.getRowType.getFieldCount
    rel match {
      case _: StreamExecCorrelate | _: Correlate =>
        getBasicMono(rel.getInput(0), mq, rowSize)
      case wa: StreamExecWatermarkAssigner =>
        getBasicMono(wa.getInput, mq, rowSize)
      case mb: StreamExecMiniBatchAssigner =>
        getBasicMono(mb.getInput, mq, rowSize)
      case _: StreamExecTemporalTableFunctionJoin =>
        getBasicMono(rel.getInput(0), mq, rowSize)
      case _: StreamExecExpand =>
        getBasicMono(rel.getInput(0), mq, rowSize)
      case _ => null
    }
  }

  def getRelModifiedMonotonicity(subset: RelSubset, mq: RelMetadataQuery)
  : RelModifiedMonotonicity = {
    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    fmq.getRelModifiedMonotonicity(Util.first(subset.getBest, subset.getOriginal))
  }

  def getRelModifiedMonotonicity(
    hepRelVertex: HepRelVertex, mq: RelMetadataQuery): RelModifiedMonotonicity = {

    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    fmq.getRelModifiedMonotonicity(hepRelVertex.getCurrentRel)
  }

  def getRelModifiedMonotonicity(rel: TableScan, mq: RelMetadataQuery): RelModifiedMonotonicity = {
    val monotonicity: RelModifiedMonotonicity = rel match {
      case _: FlinkLogicalNativeTableScan | _: StreamExecDataStreamScan =>
        val table = rel.getTable.unwrap(classOf[DataStreamTable[Any]])
        table.statistic.getRelModifiedMonotonicity
      case _: FlinkLogicalIntermediateTableScan | _: StreamExecIntermediateTableScan =>
        val table = rel.getTable.unwrap(classOf[IntermediateRelNodeTable])
        table.statistic.getRelModifiedMonotonicity
      case _ => null
    }

    if (monotonicity != null) {
      monotonicity
    } else {
      new RelModifiedMonotonicity(Array.fill(rel.getRowType.getFieldCount)(CONSTANT))
    }
  }

  def getRelModifiedMonotonicity(rel: Aggregate, mq: RelMetadataQuery): RelModifiedMonotonicity = {
    getAggModifiedMono(rel.getInput, mq, rel.getAggCallList.toList, rel.getGroupSet.toArray)
  }

  def getRelModifiedMonotonicity(calc: Calc, mq: RelMetadataQuery): RelModifiedMonotonicity = {
    val projects = calc.getProgram.getProjectList.map(calc.getProgram.expandLocalRef)
    getProjectMonotonicity(projects, calc.getInput, mq)
  }

  def getRelModifiedMonotonicity(rel: Project, mq: RelMetadataQuery): RelModifiedMonotonicity = {
    getProjectMonotonicity(rel.getProjects, rel.getInput, mq)
  }

  def getRelModifiedMonotonicity(rel: Union, mq: RelMetadataQuery): RelModifiedMonotonicity = {
    getUnionMonotonicity(rel, mq)
  }

  // --------------------------- FlinkLogical RelNode ------------------------------

  def getRelModifiedMonotonicity(
      rel: FlinkLogicalWindowAggregate,
      mq: RelMetadataQuery): RelModifiedMonotonicity = {
    null
  }

  def getRelModifiedMonotonicity(
    rel: FlinkLogicalLastRow,
    mq: RelMetadataQuery): RelModifiedMonotonicity = {

    if (allAppend(mq, rel.getInput)) {
      val mono = new RelModifiedMonotonicity(Array.fill(rel.getRowType.getFieldCount)(MONOTONIC))
      rel.getUniqueKeys.foreach(e => mono.fieldMonotonicities(e) = CONSTANT)
      mono
    } else {
      null
    }
  }

  def getRelModifiedMonotonicity(
    rel: FlinkLogicalOverWindow,
    mq: RelMetadataQuery): RelModifiedMonotonicity = {

    constants(rel)
  }

  def getRelModifiedMonotonicity(join: FlinkLogicalJoin, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    val joinInfo = join.analyzeCondition
    val joinType = FlinkJoinRelType.toFlinkJoinRelType(join.getJoinType)
    getJoinMonotonicity(
      joinInfo,
      joinType,
      join.getLeft,
      join.getRight,
      joinInfo.leftKeys,
      joinInfo.rightKeys,
      mq)
  }

  def getRelModifiedMonotonicity(rel: FlinkLogicalUnion, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    getUnionMonotonicity(rel, mq)
  }

  def getRelModifiedMonotonicity(
    rel: FlinkLogicalSemiJoin,
    mq: RelMetadataQuery): RelModifiedMonotonicity = {

    if (!rel.isAnti) {
      val joinInfo = JoinInfo.of(rel.getLeft, rel.getRight, rel.getCondition)
      getJoinMonotonicity(
        joinInfo,
        FlinkJoinRelType.SEMI,
        rel.getLeft,
        rel.getRight,
        joinInfo.leftKeys,
        joinInfo.rightKeys,
        mq)
    } else {
      null
    }
  }

  def getRelModifiedMonotonicity(rel: FlinkLogicalRank, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    getRankMonotonicity(rel.getInput,
      rel,
      rel.partitionKey.toArray,
      rel.outputRankFunColumn,
      rel.sortCollation,
      mq)
  }

  // --------------------------- StreamExec RelNode ------------------------------

  def getRelModifiedMonotonicity(rel: StreamExecUnion, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    getUnionMonotonicity(rel, mq)
  }

  def getRelModifiedMonotonicity(rel: StreamExecGroupAggregate, mq: RelMetadataQuery)
  : RelModifiedMonotonicity = {
    getAggModifiedMono(rel.getInput, mq, rel.aggCalls.toList, rel.getGroupings)
  }

  def getRelModifiedMonotonicity(rel: StreamExecLocalGroupAggregate, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    getAggModifiedMono(rel.getInput, mq, rel.aggCalls.toList, rel.getGroupings)
  }

  def getRelModifiedMonotonicity(rel: StreamExecGlobalGroupAggregate, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    // global and local agg should have same update mono
    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    fmq.getRelModifiedMonotonicity(rel.getInput)
  }

  def getRelModifiedMonotonicity(rel: StreamExecIncrementalGroupAggregate, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    getAggModifiedMono(rel.getInput, mq, rel.finalAggCalls.toList, rel.groupKey)
  }

  def getRelModifiedMonotonicity(rel: StreamExecGroupWindowAggregate, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    if (allAppend(mq, rel.getInput) && !rel.producesUpdates) {
      constants(rel)
    } else {
      null
    }
  }

  def getRelModifiedMonotonicity(
    rel: StreamExecOverAggregate,
    mq: RelMetadataQuery): RelModifiedMonotonicity = {

    constants(rel)
  }

  def getRelModifiedMonotonicity(
    rel: StreamExecExchange,
    mq: RelMetadataQuery): RelModifiedMonotonicity = {

    // for exchange, get correspond from input mono
    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    fmq.getRelModifiedMonotonicity(rel.getInput)
  }

  def getRelModifiedMonotonicity(rel: StreamExecWindowJoin, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    // window join won't have update
    constants(rel)
  }

  def getRelModifiedMonotonicity(rel: BatchExecGroupAggregateBase, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    null
  }

  def getRelModifiedMonotonicity(join: Join, mq: RelMetadataQuery): RelModifiedMonotonicity = {
    val joinInfo = join.analyzeCondition
    val joinType = FlinkJoinRelType.toFlinkJoinRelType(join.getJoinType)
    joinInfo.leftKeys.toIntArray
    getJoinMonotonicity(
      joinInfo,
      joinType,
      join.getLeft,
      join.getRight,
      joinInfo.leftKeys,
      joinInfo.rightKeys,
      mq)
  }

  def getRelModifiedMonotonicity(join: StreamExecJoin, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    getJoinMonotonicity(
      join.joinInfo,
      join.joinType,
      join.getLeft,
      join.getRight,
      join.joinInfo.leftKeys,
      join.joinInfo.rightKeys,
      mq)
  }

  def getRelModifiedMonotonicity(rel: StreamExecRank, mq: RelMetadataQuery):
  RelModifiedMonotonicity = {
    getRankMonotonicity(
      rel.getInput,
      rel,
      rel.partitionKey.toArray,
      rel.outputRankFunColumn,
      rel.sortCollation,
      mq)
  }

  def getRelModifiedMonotonicity(
    rel: StreamExecLastRow,
    mq: RelMetadataQuery): RelModifiedMonotonicity = {

    if (allAppend(mq, rel.getInput)) {
      val mono = new RelModifiedMonotonicity(Array.fill(rel.getRowType.getFieldCount)(MONOTONIC))
      rel.getUniqueKeys.foreach(e => mono.fieldMonotonicities(e) = CONSTANT)
      mono
    } else {
      null
    }
  }

  /*********  Common Function **********/

  def getUnionMonotonicity(rel: AbstractRelNode, mq: RelMetadataQuery): RelModifiedMonotonicity = {
    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    if (rel.getInputs.exists(p => containsDelete(fmq, p))) {
      null
    } else {
      val monos = rel.getInputs.map(fmq.getRelModifiedMonotonicity)
      val head = monos.head
      if (monos.forall(head.equals(_))) {
        head
      } else {
        notMonotonic(rel)
      }
    }
  }

  def getJoinMonotonicity(
      joinInfo: JoinInfo,
      joinRelType: FlinkJoinRelType,
      left: RelNode,
      right: RelNode,
      leftKeys: ImmutableIntList,
      rightKeys: ImmutableIntList,
      mq: RelMetadataQuery): RelModifiedMonotonicity = {

    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)

    // if group set contains update return null
    val containDelete = containsDelete(fmq, left) || containsDelete(fmq, right)
    val containUpdate = containsUpdate(fmq, left) || containsUpdate(fmq, right)

    val keyPaireAllAppend =
      leftKeys.toIntArray.forall({
        val mono = fmq.getRelModifiedMonotonicity(left)
        mono != null && mono.fieldMonotonicities(_) == CONSTANT
      }) &&
        rightKeys.toIntArray.forall({
          val mono = fmq.getRelModifiedMonotonicity(right)
          mono != null && mono.fieldMonotonicities(_) == CONSTANT
        })

    if (!containDelete &&
      !joinRelType.equals(FlinkJoinRelType.ANTI) &&
      keyPaireAllAppend &&
      (containUpdate && joinInfo.isEqui || !containUpdate)) {

      // output rowtype of semi equals to the rowtype of left child
      if (joinRelType.equals(FlinkJoinRelType.SEMI)) {
        fmq.getRelModifiedMonotonicity(left)
      } else {
        val lmono = fmq.getRelModifiedMonotonicity(left).fieldMonotonicities
        val rmono = fmq.getRelModifiedMonotonicity(right).fieldMonotonicities
        new RelModifiedMonotonicity(lmono ++ rmono)
      }

    } else {
      null
    }
  }

  def getAggModifiedMono(
      input: RelNode,
      mq: RelMetadataQuery,
      aggCallList: List[AggregateCall],
      groupSet: Array[Int]): RelModifiedMonotonicity = {

    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    val childUpdateMono = fmq.getRelModifiedMonotonicity(input)
    val groupCnt = groupSet.length
    // if group by a update field or group by a field mono is null, just return null
    if (groupSet.exists(
      e => {
        childUpdateMono == null || childUpdateMono.fieldMonotonicities(e) != CONSTANT
      })) {
      return null
    }

    val currentUpdateMono = new RelModifiedMonotonicity(
      Array.fill(groupCnt)(CONSTANT) ++ Array.fill(aggCallList.size)(NOT_MONOTONIC))

    // get orig monotonicity ignore child
    aggCallList.zipWithIndex.foreach(
      e => {
        currentUpdateMono.fieldMonotonicities(e._2 + groupCnt) =
          getAggMonotonicity(e._1, fmq, input)
      })

    // need to reCalc monotonicity if child contains update
    val mono = currentUpdateMono.fieldMonotonicities
    if (containsUpdate(fmq, input)) {
      aggCallList.zipWithIndex.foreach(
        e => {
          if (e._1.getArgList.size() > 1) {
            mono(e._2 + groupCnt) = NOT_MONOTONIC
          } else if (e._1.getArgList.size() == 1) {
            val childMono = childUpdateMono.fieldMonotonicities(e._1.getArgList.head)
            val currentMono = mono(e._2 + groupCnt)
            // count will Increasing even child is NOT_MONOTONIC
            if (childMono != currentMono &&
              !e._1.getAggregation.isInstanceOf[SqlCountAggFunction]) {
              mono(e._2 + groupCnt) = NOT_MONOTONIC
            }
          }
        })
    }
    currentUpdateMono
  }

  def getProjectMonotonicity(
      projects: JList[RexNode],
      input: RelNode,
      mq: RelMetadataQuery): RelModifiedMonotonicity = {

    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    // contains delete
    if (containsDelete(mq, input)) {
      return null
    }

    // all append
    if (allAppend(fmq, input)) {
      return new RelModifiedMonotonicity(Array.fill(projects.size)(CONSTANT))
    }

    // contains update
    // init mono
    val groups = new util.LinkedList[Int]()
    val monos = Array.fill(projects.size())(NOT_MONOTONIC)
    val childMono = fmq.getRelModifiedMonotonicity(input).fieldMonotonicities
    // copy child mono
    projects.zipWithIndex.foreach(
      e => {
        def getInputFieldIndex(node: RexNode): Int = {
          node match {
            case ref: RexInputRef =>
              if (ref.getIndex >= childMono.length) {
                println("")
              }
              monos(e._2) = childMono(ref.getIndex)
              ref.getIndex
            case a: RexCall if a.getKind == SqlKind.AS || a.getKind == SqlKind.CAST =>
              getInputFieldIndex(a.getOperands.get(0))
            case c: RexCall if c.getOperands.size() == 1 =>
              c.getOperator match {
                case ssf: ScalarSqlFunction =>
                  val inputIndex = getInputFieldIndex(c.getOperands.get(0))
                  val inputCollations = mq.collations(input)
                  val binding = RexCallBinding.create(
                    input.getCluster.getTypeFactory, c, inputCollations)
                  val udfMono = getUdfMonotonicity(ssf, binding)

                  val inputMono = if (inputIndex > -1) {
                    childMono(inputIndex)
                  } else {
                    NOT_MONOTONIC
                  }
                  if (inputMono == udfMono) {
                    monos(e._2) = inputMono
                  } else {
                    monos(e._2) = NOT_MONOTONIC
                  }
                  inputIndex
                case _ => -1
              }
            case _ => -1
          }
        }
        getInputFieldIndex(e._1)
      })
    new RelModifiedMonotonicity(monos)
  }

  def getRankMonotonicity(
      input: RelNode,
      rank: RelNode,
      partitionKey: Array[Int],
      hasRowNumber: Boolean,
      sortCollation: RelCollation,
      mq: RelMetadataQuery): RelModifiedMonotonicity = {

    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    val childUpdateMono = fmq.getRelModifiedMonotonicity(input)

    // If child monotonicity is null, we should return early.
    if (childUpdateMono == null) {
      return null
    }

    // if partitionBy a update field or partitionBy a field whose mono is null, just return null
    if (partitionKey.exists(
      e => {
        childUpdateMono == null || childUpdateMono.fieldMonotonicities(e) != CONSTANT
      })) {
      return null
    }

    // init current mono
    val currentUpdateMono = notMonotonic(rank)
    // 1. partitionBy field is CONSTANT
    partitionKey.foreach(e => currentUpdateMono.fieldMonotonicities(e) = CONSTANT)
    // 2. row number filed is CONSTANT
    if(hasRowNumber) {
      currentUpdateMono.fieldMonotonicities(rank.getRowType.getFieldCount - 1) = CONSTANT
    }
    // 3. time attribute field is increasing
    (0 until rank.getRowType.getFieldCount).foreach(e => {
      if (FlinkTypeFactory.isTimeIndicatorType(rank.getRowType.getFieldList.get(e).getType)) {
        childUpdateMono.fieldMonotonicities(e) = INCREASING
      }
    })
    val fieldCollations = sortCollation.getFieldCollations
    if (fieldCollations.nonEmpty) {
      // 4. process the first collation field, we can only deduce the first collation field
      val firstCollation = fieldCollations.get(0)
      // Collation field index in child node will be same with Rank node,
      // see ProjectToLogicalProjectAndWindowRule for details.
      val childFieldMono = childUpdateMono.fieldMonotonicities(firstCollation.getFieldIndex)
      currentUpdateMono.fieldMonotonicities(firstCollation.getFieldIndex) =
        childFieldMono match {
          case SqlMonotonicity.INCREASING | SqlMonotonicity.CONSTANT
            if firstCollation.direction == RelFieldCollation.Direction.DESCENDING => INCREASING
          case SqlMonotonicity.DECREASING | SqlMonotonicity.CONSTANT
            if firstCollation.direction == RelFieldCollation.Direction.ASCENDING => DECREASING
          case _ => NOT_MONOTONIC
        }
    }

    currentUpdateMono
  }

  /**
    * These operator won't generate update itself
    */
  def getBasicMono(
      input: RelNode,
      mq: RelMetadataQuery,
      rowSize: Int): RelModifiedMonotonicity = {
    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    if (containsDelete(fmq, input)) {
      null
    } else if (allAppend(fmq, input)) {
      new RelModifiedMonotonicity(Array.fill(rowSize)(CONSTANT))
    } else {
      new RelModifiedMonotonicity(Array.fill(rowSize)(NOT_MONOTONIC))
    }
  }

  def containsDelete(mq: RelMetadataQuery, node: RelNode): Boolean = {
    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    fmq.getRelModifiedMonotonicity(node) == null
  }

  def containsUpdate(mq: RelMetadataQuery, node: RelNode): Boolean = {
    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    !containsDelete(fmq, node) &&
      fmq.getRelModifiedMonotonicity(node).fieldMonotonicities.exists(_ != CONSTANT)
  }

  def allAppend(mq: RelMetadataQuery, node: RelNode): Boolean = {
    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
    !containsDelete(fmq, node) &&
      fmq.getRelModifiedMonotonicity(node).fieldMonotonicities.forall(_ == CONSTANT)
  }

  def getUdfMonotonicity(udf: ScalarSqlFunction, binding: SqlOperatorBinding): SqlMonotonicity = {
    // get monotonicity info from ScalarSqlFunction directly.
    udf.getMonotonicity(binding)
  }

  def getAggMonotonicity(
    aggCall: AggregateCall,
    mq: RelMetadataQuery,
    input: RelNode): SqlMonotonicity = {

    val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)

    aggCall.getAggregation match {
      case _: SqlCountAggFunction => INCREASING
      case mm: SqlMinMaxAggFunction =>
        mm.kind match {
          case SqlKind.MAX => INCREASING
          case SqlKind.MIN => DECREASING
          case _ => NOT_MONOTONIC
        }

      case _: SqlIncrSumAggFunction => INCREASING

      case _: SqlSumAggFunction | _: SqlSumEmptyIsZeroAggFunction =>
        val valueInterval = fmq.getFilteredColumnInterval(
          input, aggCall.getArgList.head, aggCall.filterArg)
        if (valueInterval == null) {
          NOT_MONOTONIC
        } else {
          valueInterval match {
            case n1: WithLower =>
              val compare = isValueGreaterThanZero(n1.lower)
              if (compare >= 0) {
                INCREASING
              } else {
                NOT_MONOTONIC
              }
            case n2: WithUpper =>
              val compare = isValueGreaterThanZero(n2.upper)
              if (compare <= 0) {
                DECREASING
              } else {
                NOT_MONOTONIC
              }
            case _ =>
              // value range has no lower end
              NOT_MONOTONIC
          }
        }
      case _ => NOT_MONOTONIC
    }
  }

  private def isValueGreaterThanZero[T](value: Comparable[T]): Int = {
    value match {
      case i: JInt => i.compareTo(0)
      case l: JLong => l.compareTo(0L)
      case db: JDouble => db.compareTo(0d)
      case f: JFloat => f.compareTo(0f)
      case s: JShort => s.compareTo(0.toShort)
      case b: JByte => b.compareTo(0.toByte)
      case dec: JBigDecimal => dec.compareTo(JBigDecimal.ZERO)
      case _: Date | _: Time | _: Timestamp | _: JString =>
        //not interested here, just return negative
        -1
      case _ =>
        // other numeric types
        value.asInstanceOf[Comparable[Any]].compareTo(0.asInstanceOf[Comparable[Any]])
    }
  }

}

object FlinkRelMdModifiedMonotonicity {

  private val INSTANCE = new FlinkRelMdModifiedMonotonicity

  val SOURCE: RelMetadataProvider = ReflectiveRelMetadataProvider.reflectiveSource(
    FlinkMetadata.ModifiedMonotonicityMeta.METHOD, INSTANCE)

}