/*
 * 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.functions.utils

import java.lang.reflect.{Method, Modifier}
import java.lang.{Integer => JInt, Long => JLong}
import java.sql.{Date, Time, Timestamp}
import com.google.common.primitives.Primitives
import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeFactory}
import org.apache.calcite.rex.{RexLiteral, RexNode}
import org.apache.calcite.sql.`type`.SqlTypeName
import org.apache.calcite.sql.{SqlFunction, SqlOperatorBinding}
import org.apache.commons.codec.binary.Base64
import org.apache.flink.api.common.functions.InvalidTypesException
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils._
import org.apache.flink.table.api.functions._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{TableEnvironment, TableException, ValidationException}
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.expressions._
import org.apache.flink.table.plan.logical._
import org.apache.flink.table.plan.schema.DeferredTypeFlinkTableFunction
import org.apache.flink.table.dataformat.{BaseRow, BinaryString, Decimal}
import org.apache.flink.table.errorcode.TableErrors
import org.apache.flink.table.hive.functions._
import org.apache.flink.table.api.types._
import org.apache.flink.table.typeutils.TypeUtils
import org.apache.flink.types.Row
import org.apache.flink.util.InstantiationUtil

import org.apache.hadoop.hive.ql.exec.{UDAF, UDF}
import org.apache.hadoop.hive.ql.udf.generic.{GenericUDAFResolver2, GenericUDF, GenericUDTF}

import scala.collection.JavaConversions._
import scala.collection.mutable
import scala.language.postfixOps

object UserDefinedFunctionUtils {

  /**
    * Checks if a user-defined function can be easily instantiated.
    */
  def checkForInstantiation(clazz: Class[_]): Unit = {
    if (!InstantiationUtil.isPublic(clazz)) {
      throw new ValidationException(s"Function class ${clazz.getCanonicalName} is not public.")
    }
    else if (!InstantiationUtil.isProperClass(clazz)) {
      throw new ValidationException(
        s"Function class ${clazz.getCanonicalName} is no proper class," +
        " it is either abstract, an interface, or a primitive type.")
    }
    else if (InstantiationUtil.isNonStaticInnerClass(clazz)) {
      throw new ValidationException(
        s"The class ${clazz.getCanonicalName} is an inner class, but" +
        " not statically accessible.")
    }
  }

  /**
    * Check whether this is a Scala object. It is forbidden to use [[TableFunction]] implemented
    * by a Scala object, since concurrent risks.
    */
  def checkNotSingleton(clazz: Class[_]): Unit = {
    // TODO it is not a good way to check singleton. Maybe improve it further.
    if (clazz.getFields.map(_.getName) contains "MODULE$") {
      throw new ValidationException(
        s"TableFunction implemented by class ${clazz.getCanonicalName} " +
          s"is a Scala object, it is forbidden since concurrent risks.")
    }
  }

  // ----------------------------------------------------------------------------------------------
  // Utilities for user-defined methods
  // ----------------------------------------------------------------------------------------------

  def throwValidationException(
      name: String,
      func: UserDefinedFunction,
      parameters: Array[InternalType]): Method = {
    throw new ValidationException(
      s"Given parameters of function '$name' do not match any signature. \n" +
          s"Actual: ${signatureToString(parameters)} \n" +
          s"Expected: ${signaturesToString(func, "eval")}")
  }

  private def getParamClassesConsiderVarArgs(
      isVarArgs: Boolean,
      matchingSignature: Array[Class[_]],
      expectedLength: Int): Array[Class[_]] = {
    var paramClasses = new mutable.ArrayBuffer[Class[_]]
    for (i <- 0 until expectedLength) {
      if (i < matchingSignature.length - 1) {
        paramClasses += matchingSignature(i)
      } else if (isVarArgs) {
        paramClasses += matchingSignature.last.getComponentType
      } else {
        // last argument is not an array type
        paramClasses += matchingSignature.last
      }
    }
    paramClasses.toArray
  }

  def getEvalMethodSignature(
      func: CustomTypeDefinedFunction,
      expectedTypes: Array[InternalType]): Array[Class[_]] = {
    val method = getEvalUserDefinedMethod(func, expectedTypes).getOrElse(
      throwValidationException(func.getClass.getCanonicalName, func, expectedTypes)
    )
    getParamClassesConsiderVarArgs(method.isVarArgs, method.getParameterTypes, expectedTypes.length)
  }

  def getAggUserDefinedInputTypes(
      func: AggregateFunction[_, _],
      externalAccType: DataType,
      expectedTypes: Array[InternalType]): Array[DataType] = {
    val accMethod = getAggFunctionUDIMethod(
      func, "accumulate", externalAccType, expectedTypes).getOrElse(
      throwValidationException(func.getClass.getCanonicalName, func, expectedTypes)
    )
    func.getUserDefinedInputTypes(
      getParamClassesConsiderVarArgs(
        accMethod.isVarArgs,
        // drop first, first must be Acc.
        accMethod.getParameterTypes.drop(1),
        expectedTypes.length)).zipWithIndex.map {
      case (t, i) =>
        // we don't trust GenericType.
        if (TypeUtils.isGeneric(t)) expectedTypes(i) else t
    }
  }

  /**
    * Returns signatures of accumulate methods matching the given signature of [[InternalType]].
    * Elements of the signature can be null (act as a wildcard).
    */
  def getAccumulateMethodSignature(
      function: AggregateFunction[_, _],
      expectedTypes: Seq[InternalType])
      : Option[Array[Class[_]]] = {
    getAggFunctionUDIMethod(
      function,
      "accumulate",
      getAccumulatorTypeOfAggregateFunction(function),
      expectedTypes
    ).map(_.getParameterTypes)
  }

  def getParameterTypes(
      function: UserDefinedFunction,
      signature: Array[Class[_]]): Array[InternalType] = {
    signature.map { c =>
      try {
        TypeExtractor.getForClass(c).toInternalType
      } catch {
        case ite: InvalidTypesException =>
          throw new ValidationException(
            s"Parameter types of function '${function.getClass.getCanonicalName}' cannot be " +
              s"automatically determined. Please provide type information manually.")
      }
    }
  }

  def getEvalUserDefinedMethod(
      function: CustomTypeDefinedFunction,
      expectedTypes: Seq[InternalType])
      : Option[Method] = {
    getUserDefinedMethod(
      function,
      "eval",
      typesToClasses(expectedTypes),
      expectedTypes.toArray,
      (paraClasses) => function.getParameterTypes(paraClasses))
  }

  def getAggFunctionUDIMethod(
      function: AggregateFunction[_, _],
      methodName: String,
      accType: DataType,
      expectedTypes: Seq[InternalType])
      : Option[Method] = {
    val input = (Array(accType) ++ expectedTypes).toSeq
    getUserDefinedMethod(
      function,
      methodName,
      typesToClasses(input),
      input.map(_.toInternalType).toArray,
      (cls) => Array(accType) ++
          function.getUserDefinedInputTypes(cls.drop(1)))
  }

  /**
    * Get method without match DateType.
    */
  def getUserDefinedMethod(
      function: UserDefinedFunction,
      methodName: String,
      signature: Seq[DataType])
  : Option[Method] = {
    getUserDefinedMethod(
      function,
      methodName,
      typesToClasses(signature),
      signature.map(_.toInternalType).toArray,
      (cls) => cls.indices.map((_) => null).toArray)
  }

  /**
    * Returns user defined method matching the given name and signature.
    *
    * @param function        function instance
    * @param methodName      method name
    * @param methodSignature an array of raw Java classes. We compare the raw Java classes not the
    *                        DateType. DateType does not matter during runtime (e.g.
    *                        within a MapFunction)
    */
  def getUserDefinedMethod(
      function: UserDefinedFunction,
      methodName: String,
      methodSignature: Array[Class[_]],
      internalTypes: Array[InternalType],
      parameterTypes: (Array[Class[_]] => Array[DataType]))
    : Option[Method] = {

    val methods = checkAndExtractMethods(function, methodName)

    var applyCnt = 0
    val filtered = methods
      // go over all the methods and filter out matching methods
      .filter {
        case cur if !cur.isVarArgs =>
          val signatures = cur.getParameterTypes
          val dataTypes = parameterTypes(signatures)
          // match parameters of signature to actual parameters
          methodSignature.length == signatures.length &&
            signatures.zipWithIndex.forall { case (clazz, i) => {
              if (methodSignature(i) == classOf[Object]) {
                // The element of the method signature comes from the Table API's apply().
                // We can not decide the type here. It is an Unresolved Expression.
                // Actually, we do not have to decide the type here, any method of the overrides
                // which matches the arguments count will do the job.
                // So here we choose any method is correct.
                applyCnt += 1
              }
              parameterTypeEquals(methodSignature(i), clazz) ||
                  parameterDataTypeEquals(internalTypes(i), dataTypes(i))
            }
          }
        case cur if cur.isVarArgs =>
          val signatures = cur.getParameterTypes
          val dataTypes = parameterTypes(signatures)
          methodSignature.zipWithIndex.forall {
            // non-varargs
            case (clazz, i) if i < signatures.length - 1  =>
              parameterTypeEquals(clazz, signatures(i)) ||
                  parameterDataTypeEquals(internalTypes(i), dataTypes(i))
            // varargs
            case (clazz, i) if i >= signatures.length - 1 =>
              parameterTypeEquals(clazz, signatures.last.getComponentType) ||
                  parameterDataTypeEquals(internalTypes(i), dataTypes(i))
          } || (methodSignature.isEmpty && signatures.length == 1) // empty varargs
    }

    // if there is a fixed method, compiler will call this method preferentially
    val fixedMethodsCount = filtered.count(!_.isVarArgs)
    val found = filtered.filter { cur =>
      fixedMethodsCount > 0 && !cur.isVarArgs ||
      fixedMethodsCount == 0 && cur.isVarArgs
    }.filter { cur =>
      // filter abstract methods
      !Modifier.isVolatile(cur.getModifiers)
    }

    // check if there is a Scala varargs annotation
    if (found.isEmpty &&
      methods.exists { method =>
        val signatures = method.getParameterTypes
        val dataTypes = parameterTypes(signatures)
        if (!method.isVarArgs && signatures.length != methodSignature.length) {
          false
        } else if (method.isVarArgs && signatures.length > methodSignature.length + 1) {
          false
        } else {
          signatures.zipWithIndex.forall {
            case (clazz, i) if i < signatures.length - 1 =>
              parameterTypeEquals(methodSignature(i), clazz) ||
                parameterDataTypeEquals(internalTypes(i), dataTypes(i))
            case (clazz, i) if i == signatures.length - 1 =>
              clazz.getName.equals("scala.collection.Seq")
          }
        }
      }) {
      throw new ValidationException(
        s"Scala-style variable arguments in '$methodName' methods are not supported. Please " +
          s"add a @scala.annotation.varargs annotation.")
    } else if (found.length > 1) {
      if (applyCnt > 0) {
        // As we can not decide type while apply() exists, so choose any one is correct
        return found.headOption
      }
      throw new ValidationException(
        s"Found multiple '$methodName' methods which match the signature.")
    }
    found.headOption
  }

  /**
    * Check if a given method exists in the given function
    */
  def ifMethodExistInFunction(method: String, function: UserDefinedFunction): Boolean = {
    val methods = function
      .getClass
      .getMethods
      .filter {
        m => m.getName == method
      }
    !methods.isEmpty
  }

  /**
    * Extracts methods and throws a [[ValidationException]] if no implementation
    * can be found, or implementation does not match the requirements.
    */
  def checkAndExtractMethods(
      function: UserDefinedFunction,
      methodName: String): Array[Method] = {
    val methods = function
      .getClass
      .getMethods
      .filter { m =>
        val modifiers = m.getModifiers
        m.getName == methodName &&
          Modifier.isPublic(modifiers) &&
          !Modifier.isAbstract(modifiers) &&
          !(function.isInstanceOf[TableFunction[_]] && Modifier.isStatic(modifiers))
      }

    if (methods.isEmpty) {
      throw new ValidationException(
        s"Function class '${function.getClass.getCanonicalName}' does not implement at least " +
          s"one method named '$methodName' which is public, not abstract and " +
          s"(in case of table functions) not static.")
    }

    methods
  }

  def getMethodSignatures(
      function: UserDefinedFunction,
      methodName: String): Array[Array[Class[_]]] = {
    checkAndExtractMethods(function, methodName).map(_.getParameterTypes)
  }

  // ----------------------------------------------------------------------------------------------
  // Utilities for SQL functions
  // ----------------------------------------------------------------------------------------------

  /**
    * Create [[SqlFunction]] for a [[ScalarFunction]]
    *
    * @param name function name
    * @param function scalar function
    * @param typeFactory type factory
    * @return the ScalarSqlFunction
    */
  def createScalarSqlFunction(
      name: String,
      displayName: String,
      function: ScalarFunction,
      typeFactory: FlinkTypeFactory)
    : SqlFunction = {
    new ScalarSqlFunction(name, displayName, function, typeFactory)
  }

  /**
    * Create [[SqlFunction]] for a [[TableFunction]]
    *
    * @param name function name
    * @param tableFunction table function
    * @param implicitResultType the implicit type information of returned table
    * @param typeFactory type factory
    * @return the TableSqlFunction
    */
  def createTableSqlFunction(
      name: String,
      displayName: String,
      tableFunction: TableFunction[_],
      implicitResultType: DataType,
      typeFactory: FlinkTypeFactory)
    : TableSqlFunction = {
    // we don't know the exact result type yet.
    val function = new DeferredTypeFlinkTableFunction(tableFunction, implicitResultType)
    new TableSqlFunction(name, displayName, tableFunction, implicitResultType,
      typeFactory, function)
  }

  /**
    * Create [[SqlFunction]] for an [[AggregateFunction]]
    *
    * @param name function name
    * @param aggFunction aggregate function
    * @param typeFactory type factory
    * @return the TableSqlFunction
    */
  def createAggregateSqlFunction(
      name: String,
      displayName: String,
      aggFunction: AggregateFunction[_, _],
      externalResultType: DataType,
      externalAccType: DataType,
      typeFactory: FlinkTypeFactory)
  : SqlFunction = {
    //check if a qualified accumulate method exists before create Sql function
    checkAndExtractMethods(aggFunction, "accumulate")

    AggSqlFunction(
      name,
      displayName,
      aggFunction,
      externalResultType,
      externalAccType,
      typeFactory,
      aggFunction.requiresOver)
  }

  // ----------------------------------------------------------------------------------------------
  // Utilities for user-defined functions
  // ----------------------------------------------------------------------------------------------

  /**
    * Tries to infer the DataType of an AggregateFunction's return type.
    *
    * @param aggregateFunction The AggregateFunction for which the return type is inferred.
    * @param extractedType The implicitly inferred type of the result type.
    *
    * @return The inferred result type of the AggregateFunction.
    */
  def getResultTypeOfAggregateFunction(
      aggregateFunction: AggregateFunction[_, _],
      extractedType: DataType = null)
    : DataType = {

    val resultType = aggregateFunction.getResultType
    if (resultType != null) {
      resultType
    } else if (extractedType != null) {
      extractedType
    } else {
      try {
        extractTypeFromAggregateFunction(aggregateFunction, 0)
      } catch {
        case ite: InvalidTypesException =>
          throw new TableException(
            "Cannot infer generic type of ${aggregateFunction.getClass}. " +
              "You can override AggregateFunction.getResultType() to specify the type.",
            ite
          )
      }
    }
  }

  /**
    * Tries to infer the InternalType of an AggregateFunction's accumulator type.
    *
    * @param aggregateFunction The AggregateFunction for which the accumulator type is inferred.
    * @param extractedType The implicitly inferred type of the accumulator type.
    *
    * @return The inferred accumulator type of the AggregateFunction.
    */
  def getAccumulatorTypeOfAggregateFunction(
    aggregateFunction: AggregateFunction[_, _],
    extractedType: DataType = null): DataType = {

    val accType = aggregateFunction.getAccumulatorType
    if (accType != null) {
      accType
    } else if (extractedType != null) {
      extractedType
    } else {
      try {
        extractTypeFromAggregateFunction(aggregateFunction, 1)
      } catch {
        case ite: InvalidTypesException =>
          throw new TableException(
            "Cannot infer generic type of ${aggregateFunction.getClass}. " +
              "You can override AggregateFunction.getAccumulatorType() to specify the type.",
            ite
          )
      }
    }
  }

  /**
    * Internal method to extract a type from an AggregateFunction's type parameters.
    *
    * @param aggregateFunction The AggregateFunction for which the type is extracted.
    * @param parameterTypePos The position of the type parameter for which the type is extracted.
    *
    * @return The extracted type.
    */
  @throws(classOf[InvalidTypesException])
  private def extractTypeFromAggregateFunction(
      aggregateFunction: AggregateFunction[_, _],
      parameterTypePos: Int): DataType = {

    new TypeInfoWrappedDataType(TypeExtractor.createTypeInfo(
      aggregateFunction,
      classOf[AggregateFunction[_, _]],
      aggregateFunction.getClass,
      parameterTypePos))
  }

  def getResultTypeOfScalarFunction(
      function: ScalarFunction,
      arguments: Array[AnyRef],
      argTypes: Array[InternalType]): DataType = {
    val userDefinedTypeInfo = function.getResultType(
      arguments, getEvalMethodSignature(function, argTypes))
    if (userDefinedTypeInfo != null) {
      userDefinedTypeInfo
    } else {
      extractTypeFromScalarFunc(function, argTypes)
    }
  }

  private[flink] def extractTypeFromScalarFunc(
      function: ScalarFunction,
      argTypes: Array[InternalType]): DataType = {
    try {TypeExtractor.getForClass(
        getResultTypeClassOfScalarFunction(function, argTypes))
    } catch {
      case _: InvalidTypesException =>
        throw new ValidationException(
          s"Return type of scalar function '${function.getClass.getCanonicalName}' cannot be " +
              s"automatically determined. Please provide type information manually.")
    }
  }

  /**
    * Returns the return type of the evaluation method matching the given signature.
    */
  def getResultTypeClassOfScalarFunction(
      function: ScalarFunction,
      argTypes: Array[InternalType]): Class[_] = {
    // find method for signature
    getEvalUserDefinedMethod(function, argTypes).getOrElse(
      throw new IllegalArgumentException("Given signature is invalid.")).getReturnType
  }

  /**
    * Returns the return type of the evaluation method matching the given signature.
    */
  def getResultTypeClassOfPythonScalarFunction(returnType: InternalType): Class[_] = {
    returnType match {
      case _: StringType => classOf[org.apache.flink.table.dataformat.BinaryString]
      case _: BooleanType => classOf[java.lang.Boolean]
      case _: ByteType => classOf[java.lang.Byte]
      case _: ShortType => classOf[java.lang.Short]
      case _: IntType => classOf[java.lang.Integer]
      case _: LongType => classOf[java.lang.Long]
      case _: FloatType => classOf[java.lang.Float]
      case _: DoubleType => classOf[java.lang.Double]
      case _: DecimalType => classOf[java.math.BigDecimal]
      case _: DateType => classOf[java.lang.Integer]
      case _: TimeType => classOf[java.lang.Integer]
      case _: TimestampType => classOf[java.lang.Long]
      case _: ByteArrayType => classOf[Array[Byte]]
    }
  }

  // ----------------------------------------------------------------------------------------------
  // Miscellaneous
  // ----------------------------------------------------------------------------------------------

  /**
    * Returns field names and field positions for a given [[DataType]].
    *
    * Field names are automatically extracted for [[RowType]].
    *
    * @param inputType The DataType to extract the field names and positions from.
    * @return A tuple of two arrays holding the field names and corresponding field positions.
    */
  def getFieldInfo(inputType: DataType)
    : (Array[String], Array[Int], Array[InternalType]) = {
    (
        TableEnvironment.getFieldNames(inputType),
        TableEnvironment.getFieldIndices(inputType),
        TableEnvironment.getFieldTypes(inputType))
  }

  /**
    * Prints one signature consisting of classes.
    */
  def signatureToString(signature: Array[Class[_]]): String =
  signature.map { clazz =>
    if (clazz == null) {
      "null"
    } else {
      clazz.getCanonicalName
    }
  }.mkString("(", ", ", ")")

  /**
    * Prints one signature consisting of DataType.
    */
  def signatureToString(signature: Seq[DataType]): String = {
    signatureToString(typesToClasses(signature))
  }

  /**
    * Prints all signatures of methods with given name in a class.
    */
  def signaturesToString(function: UserDefinedFunction, name: String): String = {
    getMethodSignatures(function, name).map(signatureToString).mkString(", ")
  }

  /**
    * Extracts type classes of [[DataType]] in a null-aware way.
    */
  def typesToClasses(types: Seq[DataType]): Array[Class[_]] =
    types.map { t =>
    if (t == null) {
      null
    } else {
      TypeUtils.getExternalClassForType(t)
    }
  }.toArray

  /**
    * Compares parameter candidate classes with expected classes. If true, the parameters match.
    * Candidate can be null (acts as a wildcard).
    */
  private def parameterTypeEquals(candidate: Class[_], expected: Class[_]): Boolean =
  candidate == null ||
    candidate == expected ||
    expected == classOf[Object] ||
    candidate == classOf[Object]  ||  // Special case when we don't know the type
    expected.isPrimitive && Primitives.wrap(expected) == candidate ||
    candidate == classOf[Date] && (expected == classOf[Int] || expected == classOf[JInt])  ||
    candidate == classOf[Time] && (expected == classOf[Int] || expected == classOf[JInt]) ||
    candidate == classOf[Timestamp] && (expected == classOf[Long] || expected == classOf[JLong]) ||
    candidate == classOf[BinaryString] && expected == classOf[String] ||
    candidate == classOf[String] && expected == classOf[BinaryString] ||
    classOf[BaseRow].isAssignableFrom(candidate) && expected == classOf[Row] ||
    candidate == classOf[Row] && classOf[BaseRow].isAssignableFrom(expected) ||
    candidate == classOf[Decimal] && expected == classOf[BigDecimal] ||
    candidate == classOf[BigDecimal] && expected == classOf[Decimal] ||
    (candidate.isArray &&
      expected.isArray &&
      candidate.getComponentType.isInstanceOf[Object] &&
      expected.getComponentType == classOf[Object])

  private def parameterDataTypeEquals(internal: InternalType, parameterType: DataType): Boolean =
    // There is a special equal to GenericType. We need rewrite type extract to BaseRow etc...
    parameterType.toInternalType == internal || internal.getTypeClass == parameterType.getTypeClass

  @throws[Exception]
  def serialize(function: UserDefinedFunction): String = {
    val byteArray = InstantiationUtil.serializeObject(function)
    Base64.encodeBase64URLSafeString(byteArray)
  }

  @throws[Exception]
  def deserialize(data: String): UserDefinedFunction = {
    val byteData = Base64.decodeBase64(data)
    InstantiationUtil
      .deserializeObject[UserDefinedFunction](byteData, Thread.currentThread.getContextClassLoader)
  }

  /**
    * Creates a [[LogicalTableFunctionCall]] by parsing a String expression.
    *
    * @param tableEnv The table environmenent to lookup the function.
    * @param udtf a String expression of a TableFunctionCall, such as "split(c)"
    * @return A LogicalTableFunctionCall.
    */
  def createLogicalFunctionCall(
      tableEnv: TableEnvironment,
      udtf: String): LogicalTableFunctionCall = {

    var alias: Option[Seq[String]] = None

    // unwrap an Expression until we get a TableFunctionCall
    def unwrap(expr: Expression): TableFunctionCall = expr match {
      case Alias(child, name, extraNames) =>
        alias = Some(Seq(name) ++ extraNames)
        unwrap(child)
      case Call(name, args) =>
        val function = tableEnv.chainedFunctionCatalog.lookupFunction(name, args)
        unwrap(function)
      case c: TableFunctionCall => c
      case _ =>
        throw new TableException(
          "Table(TableEnv, String) constructor only accept String that " +
            "define table function followed by some Alias.")
    }

    val functionCall: LogicalTableFunctionCall = unwrap(ExpressionParser.parseExpression(udtf))
      .as(alias).toLogicalTableFunctionCall(child = null)
    functionCall
  }

  def getOperandType(callBinding: SqlOperatorBinding): Seq[InternalType] = {
    val operandTypes = for (i <- 0 until callBinding.getOperandCount)
      yield callBinding.getOperandType(i)
    operandTypes.map { operandType =>
      if (operandType.getSqlTypeName == SqlTypeName.NULL) {
        null
      } else {
        FlinkTypeFactory.toInternalType(operandType)
      }
    }
  }

  private[table] def getResultTypeOfCTDFunction(
      func: CustomTypeDefinedFunction,
      params: Array[Expression],
      getImplicitResultType: () => DataType): DataType = {
    val arguments = params.map {
      case exp: Literal =>
        exp.value.asInstanceOf[AnyRef]
      case _ =>
        null
    }
    val signature = params.map { param =>
      if (param.valid) param.resultType
      else new GenericType(new Object().getClass)  // apply() can not decide type
    }
    val argTypes = getEvalMethodSignature(func, signature)
    val udt = func.getResultType(arguments, argTypes)
    if (udt != null) udt else getImplicitResultType()
  }

  /**
    * Transform the rex nodes to Objects
    * Only literal rex nodes will be transformed, non-literal rex nodes will be
    * translated to nulls.
    *
    * @param rexNodes actual parameters of the function
    * @return A Array of the Objects
    */
  private[table] def transformRexNodes(
      rexNodes: java.util.List[RexNode]): Array[AnyRef] = {
    rexNodes.map {
      case rexNode: RexLiteral =>
        val value = rexNode.getValue2
        rexNode.getType.getSqlTypeName match {
          case SqlTypeName.INTEGER =>
            value.asInstanceOf[Long].toInt.asInstanceOf[AnyRef]
          case SqlTypeName.SMALLINT =>
            value.asInstanceOf[Long].toShort.asInstanceOf[AnyRef]
          case SqlTypeName.TINYINT =>
            value.asInstanceOf[Long].toByte.asInstanceOf[AnyRef]
          case SqlTypeName.FLOAT =>
            value.asInstanceOf[Double].toFloat.asInstanceOf[AnyRef]
          case SqlTypeName.REAL =>
            value.asInstanceOf[Double].toFloat.asInstanceOf[AnyRef]
          case _ =>
            value.asInstanceOf[AnyRef]
        }
      case _ =>
        null
    }.toArray
  }

  private[table] def buildRelDataType(
      typeFactory: RelDataTypeFactory,
      resultType: InternalType,
      fieldNames: Array[String],
      fieldIndexes: Array[Int]): RelDataType = {

    if (fieldIndexes.length != fieldNames.length) {
      throw new TableException(
        "Number of field indexes and field names must be equal.")
    }

    // check uniqueness of field names
    if (fieldNames.length != fieldNames.toSet.size) {
      throw new TableException(
        "Table field names must be unique.")
    }

    val fieldTypes: Array[InternalType] =
      resultType match {
        case bt: RowType =>
          if (fieldNames.length != bt.getArity) {
            throw new TableException(
              s"Arity of type (" + bt.getFieldNames.deep + ") " +
                  "not equal to number of field names " + fieldNames.deep + ".")
          }
          fieldIndexes.map(i => bt.getInternalTypeAt(i).toInternalType)
        case _ =>
          if (fieldIndexes.length != 1 || fieldIndexes(0) != 0) {
            throw new TableException(
              "Non-composite input type may have only a single field and its index must be 0.")
          }
          Array(resultType)
      }

    val flinkTypeFactory = typeFactory.asInstanceOf[FlinkTypeFactory]
    val builder = flinkTypeFactory.builder
    fieldNames
        .zip(fieldTypes)
        .foreach { f =>
          builder.add(f._1, flinkTypeFactory.createTypeFromInternalType(f._2, isNullable = true))
        }
    builder.build
  }

  def createUserDefinedFunction(
      classLoader: ClassLoader,
      funcName: String,
      functionDef: String): Any = {

    var func: Any = Nil
    val javaClass = functionDef.contains(".")

    if (javaClass) {
      try {
        func = classLoader.loadClass(functionDef).newInstance()
      } catch {
        case e: Exception =>
          throw new RuntimeException(
            TableErrors.INST.sqlCreateUserDefinedFuncError(
              funcName,
              functionDef,
              e.getClass.getCanonicalName + " : " + e.getMessage),
            e)
      }
    } else {
      try {
        // try deserialize first
        func = InstantiationUtil
          .deserializeObject[Object](Base64.decodeBase64(hexString2String(functionDef)),
          classLoader)
      } catch {
        case e: Exception =>
          try {
            // It might be a java class without package name
            func = classLoader.loadClass(functionDef).newInstance()
          } catch {
            case e: Exception =>
              throw new RuntimeException(
                TableErrors.INST.sqlCreateUserDefinedFuncError(
                  funcName,
                  functionDef,
                  e.getClass.getCanonicalName + " : " + e.getMessage),
                e)
          }
      }
    }

    // Check hive and covert to flink udf
    func match {
      case _: UDF =>
        func = new HiveSimpleUDF(new HiveFunctionWrapper[UDF](functionDef))
      case _: GenericUDF =>
        func = new HiveGenericUDF(new HiveFunctionWrapper[GenericUDF](functionDef))
      case _: UDAF =>
        func = new HiveUDAFFunction(new HiveFunctionWrapper[UDAF](functionDef))
      case _: GenericUDAFResolver2 =>
        func = new HiveUDAFFunction(new HiveFunctionWrapper[GenericUDAFResolver2](functionDef))
      case _: GenericUDTF =>
        func = new HiveGenericUDTF(new HiveFunctionWrapper[GenericUDTF](functionDef))
      case _ =>
    }
    func
  }

  def getImplicitResultType[T](tf: TableFunction[T]) = {
    val implicitResultType = try {
      TypeExtractor.createTypeInfo(tf, classOf[TableFunction[_]], tf.getClass, 0)
    } catch {
      case e: InvalidTypesException =>
        // may be we should get type from getResultType
        new GenericType(classOf[AnyRef])
    }
    implicitResultType
  }

  /**
    * Get implicit type from table function through reflection, We will consider getResultType first
    * then this, see [[getResultTypeIgnoreException]] for details.
    * @return Inferred implicit [[TypeInformation]], if [[InvalidTypesException]] throws, return
    *         GenericTypeInfo(classOf[AnyRef]) as fallback
    */
  def getImplicitResultTypeInfo[T](tf: TableFunction[T]): TypeInformation[T] = {
    val implicitResultType = try {
      TypeExtractor.createTypeInfo(tf, classOf[TableFunction[_]], tf.getClass, 0)
    } catch {
      case _: InvalidTypesException =>
        new GenericTypeInfo(classOf[AnyRef])
    }
    implicitResultType.asInstanceOf[TypeInformation[T]]
  }

  /**
    * Get result type from [[TableFunction]] while sans any exception.
    * @return null if any exception throws
    */
  def getResultTypeIgnoreException[T](tf: TableFunction[T],
    arguments: Array[AnyRef] = null,
    argTypes: Array[Class[_]] = null): DataType = {
    try {
      tf.getResultType(arguments, argTypes)
    } catch {
      case _: Exception =>
        null
    }
  }

  private def hexString2String(str: String): String = {
    str.sliding(2, 2).map(s =>
      Integer.parseInt(s, 16).asInstanceOf[Char].toString).mkString("")
  }
}