/*
 * 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.expressions

import java.util
import java.math.BigDecimal

import scala.collection.mutable
import com.google.common.collect.ImmutableList
import org.apache.calcite.rel.RelFieldCollation
import org.apache.calcite.rex.RexWindowBound._
import org.apache.calcite.rex.{RexCall, RexFieldCollation, RexNode, RexWindowBound}
import org.apache.calcite.sql._
import org.apache.calcite.sql.`type`.OrdinalReturnTypeInference
import org.apache.calcite.sql.parser.SqlParserPos
import org.apache.calcite.tools.RelBuilder
import org.apache.flink.table.api._
import org.apache.flink.table.api.functions.{ScalarFunction, TableFunction}
import org.apache.flink.table.api.scala.{CURRENT_ROW, UNBOUNDED_ROW}
import org.apache.flink.table.calcite.{FlinkPlannerImpl, FlinkTypeFactory}
import org.apache.flink.table.functions.sql.internal.SqlThrowExceptionFunction
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils._
import org.apache.flink.table.plan.logical.{LogicalExprVisitor, LogicalNode, LogicalTableFunctionCall}
import org.apache.flink.table.api.types._
import org.apache.flink.table.typeutils.TypeCheckUtils.isTimeInterval
import org.apache.flink.table.typeutils.TypeUtils
import org.apache.flink.table.validate.{ValidationFailure, ValidationResult, ValidationSuccess}

import _root_.scala.collection.JavaConverters._

/**
  * General expression for unresolved function calls. The function can be a built-in
  * scalar function or a user-defined scalar function.
  */
case class Call(functionName: String, args: Seq[Expression]) extends Expression {

  override private[flink] def children: Seq[Expression] = args

  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
    throw UnresolvedException(s"trying to convert UnresolvedFunction $functionName to RexNode")
  }

  override def toString = s"\\$functionName(${args.mkString(", ")})"

  override private[flink] def resultType =
    throw UnresolvedException(s"calling resultType on UnresolvedFunction $functionName")

  override private[flink] def validateInput(): ValidationResult =
    ValidationFailure(s"Unresolved function call: $functionName")

  override def accept[T](logicalExprVisitor: LogicalExprVisitor[T]): T =
    logicalExprVisitor.visit(this)
}

/**
  * Over call with unresolved alias for over window.
  *
  * @param agg The aggregation of the over call.
  * @param alias The alias of the referenced over window.
  */
case class UnresolvedOverCall(agg: Expression, alias: Expression) extends Expression {

  override private[flink] def validateInput() =
    ValidationFailure(s"Over window with alias $alias could not be resolved.")

  override private[flink] def resultType = agg.resultType

  override private[flink] def children = Seq()

  override def accept[T](logicalExprVisitor: LogicalExprVisitor[T]): T =
    logicalExprVisitor.visit(this)
}

/**
  * Over expression for Calcite over transform.
  *
  * @param agg            over-agg expression
  * @param partitionBy    The fields by which the over window is partitioned
  * @param orderBy        The field by which the over window is sorted
  * @param preceding      The lower bound of the window
  * @param following      The upper bound of the window
  */
case class OverCall(
    agg: Expression,
    partitionBy: Seq[Expression],
    orderBy: Seq[Expression],
    var preceding: Expression,
    var following: Expression,
    tableEnv: TableEnvironment) extends Expression {

  override def toString: String = s"$agg OVER (" +
    s"PARTITION BY (${partitionBy.mkString(", ")}) " +
    s"ORDER BY (${orderBy.mkString(", ")})" +
    s"PRECEDING $preceding " +
    s"FOLLOWING $following)"

  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {

    val rexBuilder = relBuilder.getRexBuilder

    // assemble aggregation
    val operator: SqlAggFunction = agg.asInstanceOf[Aggregation].getSqlAggFunction()
    val aggResultType = relBuilder
      .getTypeFactory.asInstanceOf[FlinkTypeFactory]
      .createTypeFromInternalType(agg.resultType, isNullable = true)

    // assemble exprs by agg children
    val aggExprs = agg.asInstanceOf[Aggregation].children.map(_.toRexNode(relBuilder)).asJava

    // assemble order by key
    def collation(
      node: RexNode, direction: RelFieldCollation.Direction,
      nullDirection: RelFieldCollation.NullDirection, kinds: mutable.Set[SqlKind]): RexNode = {
      node.getKind match {
        case SqlKind.DESCENDING =>
          kinds.add(node.getKind)
          collation(
            node.asInstanceOf[RexCall].getOperands.get(0),
            RelFieldCollation.Direction.DESCENDING, nullDirection, kinds)
        case SqlKind.NULLS_FIRST =>
          kinds.add(node.getKind)
          collation(
            node.asInstanceOf[RexCall].getOperands.get(0), direction,
            RelFieldCollation.NullDirection.FIRST, kinds)
        case SqlKind.NULLS_LAST =>
          kinds.add(node.getKind)
          collation(
            node.asInstanceOf[RexCall].getOperands.get(0), direction,
            RelFieldCollation.NullDirection.LAST, kinds)
        case _ =>
          if (nullDirection == null) {
            // Set the null direction if not specified.
            // Consistent with HIVE/SPARK/MYSQL/BLINK-RUNTIME.
            if (FlinkPlannerImpl.defaultNullCollation.last(
              direction.equals(RelFieldCollation.Direction.DESCENDING))) {
              kinds.add(SqlKind.NULLS_LAST)
            } else {
              kinds.add(SqlKind.NULLS_FIRST)
            }
          }
          node
      }
    }

    val orderKeysBuilder = new ImmutableList.Builder[RexFieldCollation]()
    for (orderExp <- orderBy) {
      val kinds: mutable.Set[SqlKind] = mutable.Set()
      val rexNode = collation(
        orderExp.toRexNode, RelFieldCollation.Direction.ASCENDING, null, kinds)
      val orderKey = new RexFieldCollation(rexNode, kinds.asJava)
      orderKeysBuilder.add(orderKey)
    }
    val orderKeys = orderKeysBuilder.build()

    // assemble partition by keys
    val partitionKeys = partitionBy.map(_.toRexNode).asJava

    // assemble bounds
    var isPhysical: Boolean = false
    var lowerBound, upperBound: RexWindowBound = null
    agg match {
      case _: RowNumber | _: Rank | _: DenseRank =>
        isPhysical = true
        lowerBound = createBound(relBuilder, UNBOUNDED_ROW, SqlKind.PRECEDING)
        upperBound = createBound(relBuilder, CURRENT_ROW, SqlKind.FOLLOWING)
      case _ =>
        isPhysical = preceding.resultType == DataTypes.INTERVAL_ROWS
        lowerBound = createBound(relBuilder, preceding, SqlKind.PRECEDING)
        upperBound = createBound(relBuilder, following, SqlKind.FOLLOWING)
    }

    // build RexOver
    rexBuilder.makeOver(
      aggResultType,
      operator,
      aggExprs,
      partitionKeys,
      orderKeys,
      lowerBound,
      upperBound,
      isPhysical,
      true,
      false,
      false)
  }

  private def createBound(
    relBuilder: RelBuilder,
    bound: Expression,
    sqlKind: SqlKind): RexWindowBound = {

    bound match {
      case _: UnboundedRow | _: UnboundedRange =>
        val unbounded = if (sqlKind == SqlKind.PRECEDING) {
          SqlWindow.createUnboundedPreceding(SqlParserPos.ZERO)
        } else {
          SqlWindow.createUnboundedFollowing(SqlParserPos.ZERO)
        }
        create(unbounded, null)
      case _: CurrentRow | _: CurrentRange =>
        val currentRow = SqlWindow.createCurrentRow(SqlParserPos.ZERO)
        create(currentRow, null)
      case b: Literal =>
        val DECIMAL_PRECISION_NEEDED_FOR_LONG = 19
        val returnType = relBuilder
          .getTypeFactory.asInstanceOf[FlinkTypeFactory]
          .createTypeFromInternalType(
            DecimalType.of(DECIMAL_PRECISION_NEEDED_FOR_LONG, 0),
            isNullable = true)

        val sqlOperator = new SqlPostfixOperator(
          sqlKind.name,
          sqlKind,
          2,
          new OrdinalReturnTypeInference(0),
          null,
          null)

        val operands: Array[SqlNode] = new Array[SqlNode](1)
        operands(0) = SqlLiteral.createExactNumeric("1", SqlParserPos.ZERO)

        val node = new SqlBasicCall(sqlOperator, operands, SqlParserPos.ZERO)

        val expressions: util.ArrayList[RexNode] = new util.ArrayList[RexNode]()
        b.value match {
          case v: Double => expressions.add(relBuilder.literal(
            BigDecimal.valueOf(v.asInstanceOf[Number].doubleValue())))
          case _ => expressions.add(relBuilder.literal(b.value))
        }

        val rexNode = relBuilder.getRexBuilder.makeCall(returnType, sqlOperator, expressions)

        create(node, rexNode)
    }
  }

  override private[flink] def children: Seq[Expression] =
    Seq(agg) ++ orderBy ++ partitionBy ++ Seq(preceding) ++ Seq(following)

  override private[flink] def resultType = agg.resultType

  override private[flink] def validateInput(): ValidationResult = {

    // check that agg expression is aggregation
    agg match {
      case _: Aggregation =>
        ValidationSuccess
      case _ =>
        return ValidationFailure(s"OVER can only be applied on an aggregation.")
    }

    // check partitionBy expression keys are resolved field reference
    partitionBy.foreach {
      case r: ResolvedFieldReference
        if TypeConverters.createExternalTypeInfoFromDataType(r.resultType).isKeyType  =>
        ValidationSuccess
      case r: ResolvedFieldReference =>
        return ValidationFailure(s"Invalid PartitionBy expression: $r. " +
          s"Expression must return key type.")
      case r =>
        return ValidationFailure(s"Invalid PartitionBy expression: $r. " +
          s"Expression must be a resolved field reference.")
    }

    // check preceding is valid
    preceding match {
      case _: CurrentRow | _: CurrentRange | _: UnboundedRow | _: UnboundedRange =>
        ValidationSuccess
      case Literal(_: Long, DataTypes.INTERVAL_ROWS) =>
        ValidationSuccess
      case Literal(_: Long, DataTypes.INTERVAL_RANGE) |
           Literal(_: Double, DataTypes.INTERVAL_RANGE) =>
        ValidationSuccess
      case Literal(v: Long, b) if isTimeInterval(b) && v >= 0 =>
        ValidationSuccess
      case Literal(_, b) if isTimeInterval(b) =>
        return ValidationFailure("Preceding time interval must be equal or larger than 0.")
      case Literal(_, _) =>
        return ValidationFailure("Preceding must be a row interval or time interval literal.")
    }

    // check following is valid
    following match {
      case _: CurrentRow | _: CurrentRange | _: UnboundedRow | _: UnboundedRange =>
        ValidationSuccess
      case Literal(_: Long, DataTypes.INTERVAL_ROWS) =>
        ValidationSuccess
      case Literal(_: Long, DataTypes.INTERVAL_RANGE) |
           Literal(_: Double, DataTypes.INTERVAL_RANGE) =>
        ValidationSuccess
      case Literal(v: Long, b) if isTimeInterval(b) && v >= 0 =>
        ValidationSuccess
      case Literal(_, b) if isTimeInterval(b) =>
        return ValidationFailure("Following time interval must be equal or larger than 0.")
      case Literal(_, _) =>
        return ValidationFailure("Following must be a row interval or time interval literal.")
    }

    // check that preceding and following are of same type
    (preceding, following) match {
      case (p: Expression, f: Expression) if p.resultType == f.resultType =>
        ValidationSuccess
      case (_: UnboundedRange, f: Expression) if f.resultType == DataTypes.INTERVAL_MILLIS =>
        ValidationSuccess
      case (_: CurrentRange, f: Expression) if f.resultType == DataTypes.INTERVAL_MILLIS =>
        ValidationSuccess
      case (p: Expression, _: UnboundedRange) if p.resultType == DataTypes.INTERVAL_MILLIS =>
        ValidationSuccess
      case (p: Expression, _: CurrentRange) if p.resultType == DataTypes.INTERVAL_MILLIS =>
        ValidationSuccess
      case _ =>
        return ValidationFailure("Preceding and following must be of same interval type.")
    }

    if (tableEnv.isInstanceOf[StreamTableEnvironment]) {
      validateInputForStream()
    } else {
      ValidationSuccess
    }
  }

  private[flink] def validateInputForStream(): ValidationResult = {
    // check preceding/following range
    preceding match {
      case Literal(_, DataTypes.INTERVAL_RANGE) =>
        return ValidationFailure("Stream table API does not support value range.")
      case Literal(v: Long, DataTypes.INTERVAL_ROWS) if v < 0 =>
        return ValidationFailure("Stream table API does not support negative preceding.")
      case _ =>
        ValidationSuccess
    }
    following match {
      case _: UnboundedRow | _: UnboundedRange =>
        return ValidationFailure("Stream table API does not support unbounded following.")
      case Literal(_, DataTypes.INTERVAL_RANGE) =>
        return ValidationFailure("Stream table API does not support value range.")
      case Literal(v: Long, DataTypes.INTERVAL_ROWS) if v > 0 =>
        return ValidationFailure("Stream table API does not support positive following.")
      case _ =>
        ValidationSuccess
    }

    // check lead/lag offset
    agg match {
      case a: Lead =>
        if (a.getOffsetValue > 0) {
          return ValidationFailure("Stream table API does not support positive offset for lead.")
        }
      case a: Lag =>
        if (a.getOffsetValue < 0) {
          return ValidationFailure("Stream table API does not support negative offset for lag.")
        }
      case _ =>
        ValidationSuccess
    }

    ValidationSuccess
  }

  override def accept[T](logicalExprVisitor: LogicalExprVisitor[T]): T =
    logicalExprVisitor.visit(this)
}

object ScalarFunctionCall {
  def apply(
      scalarFunction: ScalarFunction,
      parameters: Array[Expression]): ScalarFunctionCall =
    new ScalarFunctionCall(scalarFunction, parameters)
}

/**
  * Expression for calling a user-defined scalar functions.
  *
  * @param scalarFunction scalar function to be called (might be overloaded)
  * @param parameters actual parameters that determine target evaluation method
  */
case class ScalarFunctionCall(
    scalarFunction: ScalarFunction,
    parameters: Seq[Expression])
  extends Expression {

  override private[flink] def children: Seq[Expression] = parameters

  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
    val typeFactory = relBuilder.getTypeFactory.asInstanceOf[FlinkTypeFactory]
    relBuilder.call(
      createScalarSqlFunction(
        scalarFunction.functionIdentifier,
        scalarFunction.toString,
        scalarFunction,
        typeFactory),
      parameters.map(_.toRexNode): _*)
  }

  override def toString =
    s"${scalarFunction.getClass.getCanonicalName}(${parameters.mkString(", ")})"

  override private[flink] def resultType =
    getResultTypeOfCTDFunction(
      scalarFunction,
      parameters.toArray,
      () => {extractTypeFromScalarFunc(
        scalarFunction, parameters.map(_.resultType).toArray)}).toInternalType

  override private[flink] def validateInput(): ValidationResult = {
    val signature = children.map(_.resultType)
    // look for a signature that matches the input types
    if (getEvalUserDefinedMethod(scalarFunction, signature).isEmpty) {
      ValidationFailure(s"Given parameters do not match any signature. \n" +
        s"Actual: ${signatureToString(signature)} \n" +
        s"Expected: ${signaturesToString(scalarFunction, "eval")}")
    } else {
      ValidationSuccess
    }
  }

  override def accept[T](logicalExprVisitor: LogicalExprVisitor[T]): T =
    logicalExprVisitor.visit(this)
}

/**
  *
  * Expression for calling a user-defined table function with actual parameters.
  *
  * @param functionName function name
  * @param tableFunction user-defined table function
  * @param parameters actual parameters of function
  * @param externalResultType type information of returned table
  */
case class TableFunctionCall(
    functionName: String,
    tableFunction: TableFunction[_],
    parameters: Seq[Expression],
    externalResultType: DataType)
  extends Expression {

  private var aliases: Option[Seq[String]] = None

  override private[flink] def children: Seq[Expression] = parameters

  override private[flink] def resultType: InternalType = externalResultType.toInternalType

  /**
    * Assigns an alias for this table function's returned fields that the following operator
    * can refer to.
    *
    * @param aliasList alias for this table function's returned fields
    * @return this table function call
    */
  private[flink] def as(aliasList: Option[Seq[String]]): TableFunctionCall = {
    this.aliases = aliasList
    this
  }

  /**
    * Converts an API class to a logical node for planning.
    */
  private[flink] def toLogicalTableFunctionCall(child: LogicalNode): LogicalTableFunctionCall = {
    val originNames = getFieldInfo(resultType)._1

    // determine the final field names
    val fieldNames = if (aliases.isDefined) {
      val aliasList = aliases.get
      if (aliasList.length != originNames.length) {
        throw new ValidationException(
          s"List of column aliases must have same degree as table; " +
            s"the returned table of function '$functionName' has ${originNames.length} " +
            s"columns (${originNames.mkString(",")}), " +
            s"whereas alias list has ${aliasList.length} columns")
      } else {
        aliasList.toArray
      }
    } else {
      originNames
    }

    LogicalTableFunctionCall(
      functionName,
      tableFunction,
      parameters,
      externalResultType,
      fieldNames,
      child)
  }

  override def toString =
    s"${tableFunction.getClass.getCanonicalName}(${parameters.mkString(", ")})"

  override def accept[T](logicalExprVisitor: LogicalExprVisitor[T]): T =
    logicalExprVisitor.visit(this)
}

case class ThrowException(msg: Expression, tp: InternalType) extends UnaryExpression {

  override private[flink] def validateInput(): ValidationResult = {
    if (child.resultType == DataTypes.STRING) {
      ValidationSuccess
    } else {
      ValidationFailure(s"ThrowException operator requires String input, " +
          s"but $child is of type ${child.resultType}")
    }
  }

  override def toString: String = s"ThrowException($msg)"

  override private[flink] def toRexNode(implicit relBuilder: RelBuilder): RexNode = {
    relBuilder.call(
      new SqlThrowExceptionFunction(
        tp,
        relBuilder.getTypeFactory.asInstanceOf[FlinkTypeFactory]),
      msg.toRexNode)
  }

  override private[flink] def child = msg

  override private[flink] def resultType = tp

  override def accept[T](logicalExprVisitor: LogicalExprVisitor[T]): T =
    logicalExprVisitor.visit(this)
}