/*
 * 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.nodes.common

import java.lang.reflect.{Method, Modifier}
import java.util
import java.util.Collections
import com.google.common.primitives.Primitives
import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeField}
import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
import org.apache.calcite.rex._
import org.apache.calcite.sql.SqlKind
import org.apache.calcite.sql.fun.SqlStdOperatorTable
import org.apache.calcite.sql.validate.SqlValidatorUtil
import org.apache.calcite.tools.RelBuilder
import org.apache.calcite.util.mapping.IntPair
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils.{RowTypeInfo, TypeExtractor}
import org.apache.flink.streaming.api.datastream.AsyncDataStream.OutputMode
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.functions.async.ResultFuture
import org.apache.flink.streaming.api.operators.ProcessOperator
import org.apache.flink.streaming.api.operators.async.AsyncWaitOperator
import org.apache.flink.streaming.api.transformations.{OneInputTransformation, StreamTransformation}
import org.apache.flink.table.api.functions.{AsyncTableFunction, TableFunction, UserDefinedFunction}
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.types.{DataType, InternalType, TypeConverters}
import org.apache.flink.table.api.{TableConfig, TableException, ValidationException}
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.codegen.TemporalJoinCodeGenerator._
import org.apache.flink.table.codegen.{CodeGeneratorContext, TemporalJoinCodeGenerator}
import org.apache.flink.table.dataformat.{BaseRow, BinaryRow, GenericRow, JoinedRow}
import org.apache.flink.table.functions.sql.ScalarSqlFunctions
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils.{checkAndExtractMethods, signatureToString, signaturesToString}
import org.apache.flink.table.plan.nodes.FlinkRelNode
import org.apache.flink.table.plan.schema.{BaseRowSchema, TimeIndicatorRelDataType}
import org.apache.flink.table.plan.util.TemporalJoinUtil._
import org.apache.flink.table.plan.util.{CalcUtil, RexLiteralUtil}
import org.apache.flink.table.runtime.join.{TemporalTableJoinAsyncRunner, TemporalTableJoinProcessRunner, TemporalTableJoinWithCalcAsyncRunner, TemporalTableJoinWithCalcProcessRunner}
import org.apache.flink.table.sources.{IndexKey, LookupConfig, LookupableTableSource, TableSource}
import org.apache.flink.table.typeutils.{BaseRowTypeInfo, TypeUtils}
import org.apache.flink.table.util.TableConnectorUtil
import org.apache.flink.types.Row

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

/**
  * Common abstract RelNode for temporal table join which shares most methods.
  * @param input  input rel node
  * @param tableSource  the table source to be temporal joined
  * @param tableRowType  the row type of the table source
  * @param tableCalcProgram  the calc (projection&filter) after table scan before joining
  * @param period   the point in time to snapshot
  */
abstract class CommonTemporalTableJoin(
  cluster: RelOptCluster,
  traitSet: RelTraitSet,
  input: RelNode,
  val tableSource: TableSource,
  tableRowType: RelDataType,
  val tableCalcProgram: Option[RexProgram],
  period: RexNode,
  val joinInfo: JoinInfo,
  val joinType: JoinRelType)
  extends SingleRel(cluster, traitSet, input) 
  with FlinkRelNode {

  val joinKeyPairs: util.List[IntPair] = getTemporalTableJoinKeyPairs(joinInfo, tableCalcProgram)
  val indexKeys: util.List[IndexKey] = getTableIndexKeys(tableSource)
  // constant keys which maybe empty if calc program is None
  val constantLookupKeys: util.Map[Int, (InternalType, Object)] = analyzeConstantLookupKeys(
    cluster, 
    tableCalcProgram, 
    indexKeys)
  val joinedIndex: Option[IndexKey] = findMatchedIndex(indexKeys, joinKeyPairs, constantLookupKeys)

  override def deriveRowType(): RelDataType = {
    val flinkTypeFactory = cluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
    val rightType = if (tableCalcProgram.isDefined) {
      tableCalcProgram.get.getOutputRowType
    } else {
      tableRowType
    }
    SqlValidatorUtil.deriveJoinRowType(
      input.getRowType,
      rightType,
      joinType,
      flinkTypeFactory,
      null,
      Collections.emptyList[RelDataTypeField])
  }

  override def explainTerms(pw: RelWriter): RelWriter = {
    val remaining = joinInfo.getRemaining(cluster.getRexBuilder)
    val joinCondition = if (remaining.isAlwaysTrue) {
      None
    } else {
      Some(remaining)
    }

    joinExplainTerms(
      super.explainTerms(pw),
      tableSource,
      input.getRowType,
      getRowType,
      tableCalcProgram,
      joinInfo.pairs(),
      joinCondition,
      joinType,
      period,
      getExpressionString)
  }

  // ----------------------------------------------------------------------------------------
  //                             Physical Translation
  // ----------------------------------------------------------------------------------------

  def translateToPlanInternal(
    inputTransformation: StreamTransformation[BaseRow],
    env: StreamExecutionEnvironment,
    config: TableConfig,
    relBuilder: RelBuilder): StreamTransformation[BaseRow] = {

    val inputSchema = new BaseRowSchema(input.getRowType)
    val tableSchema = new BaseRowSchema(tableRowType)
    val resultSchema = new BaseRowSchema(getRowType)
    val inputBaseRowType = inputSchema.internalType()
    val tableBaseRowType = tableSchema.internalType()
    val resultBaseRowType = resultSchema.internalType()
    val resultBaseRowTypeInfo = resultSchema.typeInfo()
    val tableReturnTypeInfo =
      TypeConverters.createExternalTypeInfoFromDataType(tableSource.getReturnType)
    val tableReturnClass = CommonScan.extractTableSourceTypeClass(tableSource)

    // validate whether the node is valid and supported.
    validate(
      tableSource,
      period,
      inputSchema,
      tableSchema,
      joinKeyPairs,
      constantLookupKeys,
      indexKeys,
      joinedIndex,
      joinType)

    val checkedIndexInOrder = joinedIndex.get.getDefinedColumns.map(_.intValue()).toArray
    val indexFieldTypes = checkedIndexInOrder.map(tableSchema.fieldTypes(_))
    val remainingCondition = getRemainingJoinCondition(
      cluster.getRexBuilder,
      relBuilder,
      input.getRowType,
      tableRowType,
      tableCalcProgram,
      checkedIndexInOrder,
      joinKeyPairs,
      joinInfo,
      constantLookupKeys)

    val lookupKeysFromConstant: Map[Int, RexLiteral] = constantLookupKeys.toMap.map {
      case (i, (_, o)) => (i, relBuilder.literal(o).asInstanceOf[RexLiteral])
    }

    val lookupKeyPairs = joinKeyPairs.filter(p => checkedIndexInOrder.contains(p.target))
    // lookup key index -> input field index
    val lookupKey2InputFieldIndex: Map[Int, Int] = lookupKeyPairs
                                                   .map { k => (k.target, k.source) }
                                                   .toMap
    val lookupableTableSource = tableSource.asInstanceOf[LookupableTableSource[_]]
    val lookupConfig = if (lookupableTableSource.getLookupConfig != null) {
      lookupableTableSource.getLookupConfig
    } else {
      new LookupConfig
    }
    val leftOuterJoin = joinType == JoinRelType.LEFT

    val operator = if (lookupConfig.isAsyncEnabled) {
      val asyncBufferCapacity= lookupConfig.getAsyncBufferCapacity
      val asyncTimeout = lookupConfig.getAsyncTimeoutMs
      val asyncOutputMode = lookupConfig.getAsyncOutputMode

      val asyncTableFunction = lookupableTableSource.getAsyncLookupFunction(checkedIndexInOrder)
      val parameters = Array(classOf[ResultFuture[_]]) ++
        indexFieldTypes.map(TypeUtils.getInternalClassForType(_))
      val method = getSignatureMatchedEvalMethod(
        asyncTableFunction,
        parameters)
      // eval method valid check
      if (method.isEmpty) {
        val msg = s"Given parameter types of the async lookup TableFunction of TableSource " +
          s"'${tableSource.explainSource()}' do not match the expected signature.\n" +
          s"Expected: eval${signatureToString(parameters)} \n" +
          s"Actual: eval${signaturesToString(asyncTableFunction, "eval")}"
        throw new TableException(msg)
      }
      // return type valid check
      val udtfResultType = asyncTableFunction.getResultType(Array(), Array())
      val extractedResultTypeInfo = TypeExtractor.createTypeInfo(
        asyncTableFunction,
        classOf[AsyncTableFunction[_]],
        asyncTableFunction.getClass,
        0)
      checkUdtfReturnType(
        tableSource.explainSource(),
        tableReturnTypeInfo,
        udtfResultType,
        extractedResultTypeInfo)

      val generatedFetcher = TemporalJoinCodeGenerator.generateAsyncLookupFunction(
        config,
        relBuilder.getTypeFactory.asInstanceOf[FlinkTypeFactory],
        inputBaseRowType,
        resultBaseRowType,
        tableReturnTypeInfo,
        tableReturnClass,
        checkedIndexInOrder,
        lookupKey2InputFieldIndex,
        lookupKeysFromConstant,
        asyncTableFunction)

      val asyncFunc = if (tableCalcProgram.isDefined) {
        // a projection or filter after table source scan
        val calcSchema = new BaseRowSchema(tableCalcProgram.get.getOutputRowType)
        val rightTypeInfo = calcSchema.internalType
        val collector = generateAsyncCollector(
          config,
          inputBaseRowType,
          rightTypeInfo,
          remainingCondition)
        val calcMap = generateCalcMapFunction(config, tableCalcProgram, tableSchema)
        new TemporalTableJoinWithCalcAsyncRunner(
          generatedFetcher.name,
          generatedFetcher.code,
          calcMap.name,
          calcMap.code,
          collector.name,
          collector.code,
          asyncBufferCapacity,
          leftOuterJoin,
          inputSchema.fieldTypes.toArray,
          resultBaseRowTypeInfo)
      } else {
        val collector = generateAsyncCollector(
          config,
          inputBaseRowType,
          tableBaseRowType,
          remainingCondition)
        new TemporalTableJoinAsyncRunner(
          generatedFetcher.name,
          generatedFetcher.code,
          collector.name,
          collector.code,
          asyncBufferCapacity,
          leftOuterJoin,
          inputSchema.fieldTypes.toArray,
          resultBaseRowTypeInfo)
      }

      val mode = if (asyncOutputMode == LookupConfig.AsyncOutputMode.ORDERED) {
        OutputMode.ORDERED
      } else {
        OutputMode.UNORDERED
      }

      new AsyncWaitOperator(asyncFunc, asyncTimeout, asyncBufferCapacity, mode)
    } else {
      // sync join
      val lookupFunction = lookupableTableSource.getLookupFunction(checkedIndexInOrder)
      val parameters = indexFieldTypes.map(TypeUtils.getInternalClassForType(_))
      val method = getSignatureMatchedEvalMethod(
        lookupFunction,
        parameters)
      // valid check
      if (method.isEmpty) {
        val msg = s"Given parameter types of the lookup TableFunction of TableSource " +
          s"'${tableSource.explainSource()}' do not match the expected signature.\n" +
          s"Expected: eval${signatureToString(parameters)} \n" +
          s"Actual: eval${signaturesToString(lookupFunction, "eval")}"
        throw new TableException(msg)
      }
      // return type valid check
      val udtfResultType = lookupFunction.getResultType(Array(), Array())
      val extractedResultTypeInfo = TypeExtractor.createTypeInfo(
        lookupFunction,
        classOf[TableFunction[_]],
        lookupFunction.getClass,
        0)
      checkUdtfReturnType(
        tableSource.explainSource(),
        tableReturnTypeInfo,
        udtfResultType,
        extractedResultTypeInfo)

      val generatedFetcher = TemporalJoinCodeGenerator.generateLookupFunction(
        config,
        relBuilder.getTypeFactory.asInstanceOf[FlinkTypeFactory],
        inputBaseRowType,
        resultBaseRowType,
        tableReturnTypeInfo,
        tableReturnClass,
        checkedIndexInOrder,
        lookupKey2InputFieldIndex,
        lookupKeysFromConstant,
        lookupFunction,
        env.getConfig.isObjectReuseEnabled)

      val ctx = CodeGeneratorContext(config)
      val processFunc = if (tableCalcProgram.isDefined) {
        // a projection or filter after table source scan
        val calcSchema = new BaseRowSchema(tableCalcProgram.get.getOutputRowType)
        val rightTypeInfo = calcSchema.internalType()
        val collector = generateCollector(
          ctx,
          config,
          inputBaseRowType,
          rightTypeInfo,
          resultBaseRowType,
          remainingCondition,
          None)
        val calcMap = generateCalcMapFunction(config, tableCalcProgram, tableSchema)
        new TemporalTableJoinWithCalcProcessRunner(
          generatedFetcher.name,
          generatedFetcher.code,
          calcMap.name,
          calcMap.code,
          collector.name,
          collector.code,
          leftOuterJoin,
          inputSchema.fieldTypes.toArray,
          resultBaseRowTypeInfo)
      } else {
        val collector = generateCollector(
          ctx,
          config,
          inputBaseRowType,
          tableBaseRowType,
          resultBaseRowType,
          remainingCondition,
          None)
        new TemporalTableJoinProcessRunner(
          generatedFetcher.name,
          generatedFetcher.code,
          collector.name,
          collector.code,
          leftOuterJoin,
          inputSchema.fieldTypes.toArray,
          resultBaseRowTypeInfo)
      }
      new ProcessOperator(processFunc)
    }

    val operatorName = joinToString(
      lookupableTableSource,
      joinType,
      resultSchema,
      inputSchema,
      tableSchema,
      remainingCondition,
      constantLookupKeys,
      joinKeyPairs,
      getExpressionString)

    new OneInputTransformation(
      inputTransformation,
      operatorName,
      operator,
      TypeConverters.toBaseRowTypeInfo(resultBaseRowType),
      inputTransformation.getParallelism)
  }

  private def rowTypeEquals(expected: TypeInformation[_], actual: TypeInformation[_]): Boolean = {
    // check internal and external type, cause we will auto convert external class to internal
    // class (eg: Row => BaseRow).
    // check both type because GenericType<Row> and GenericType<BaseRow>.
    TypeUtils.getExternalClassForType(expected) == TypeUtils.getExternalClassForType(actual) ||
        TypeUtils.getInternalClassForType(expected) == TypeUtils.getInternalClassForType(actual)
  }

  private def getSignatureMatchedEvalMethod(
    function: UserDefinedFunction,
    methodSignature: Array[Class[_]]): Option[Method] = {

    val methods = checkAndExtractMethods(function, "eval")

    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
         // 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)
           }
       case cur if cur.isVarArgs =>
         val signatures = cur.getParameterTypes
         methodSignature.zipWithIndex.forall {
           // non-varargs
           case (clazz, i) if i < signatures.length - 1 =>
             parameterTypeEquals(clazz, signatures(i))
           // varargs
           case (clazz, i) if i >= signatures.length - 1 =>
             parameterTypeEquals(clazz, signatures.last.getComponentType)
         } || (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)
    }

    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 'eval' methods which match the signature.")
    }
    found.headOption
  }
  
  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.isArray &&
        expected.isArray &&
        candidate.getComponentType.isInstanceOf[Object] &&
        expected.getComponentType == classOf[Object])
  }

  private def getRemainingJoinCondition(
    rexBuilder: RexBuilder,
    relBuilder: RelBuilder,
    leftRowType: RelDataType,
    tableRowType: RelDataType,
    tableCalcProgram: Option[RexProgram],
    checkedIndexInOrder: Array[Int],
    joinKeyPairs: util.List[IntPair],
    joinInfo: JoinInfo,
    constantLookupKeys: util.Map[Int, (InternalType, Object)]): Option[RexNode] = {

    val remainingPairs = joinKeyPairs
                         .filter(p => !checkedIndexInOrder.contains(p.target))
    // convert remaining pairs to RexInputRef tuple for building sqlStdOperatorTable.EQUALS calls
    val remainingAnds = remainingPairs.map { p =>
      val leftInputRef = new RexInputRef(p.source, leftRowType.getFieldList.get(p.source).getType)
      val rightInputRef = tableCalcProgram match {
        case Some(program) =>
          val rightKeyIdx = program
                            .getOutputRowType.getFieldNames
                            .indexOf(program.getInputRowType.getFieldNames.get(p.target))
          new RexInputRef(
            leftRowType.getFieldCount + rightKeyIdx,
            program.getOutputRowType.getFieldList.get(rightKeyIdx).getType)

        case None =>
          new RexInputRef(
            leftRowType.getFieldCount + p.target,
            tableRowType.getFieldList.get(p.target).getType)
      }
      (leftInputRef, rightInputRef)
    }
    val equiAnds = relBuilder.and(remainingAnds.map(p => relBuilder.equals(p._1, p._2)): _*)
    val condition = relBuilder.and(equiAnds, joinInfo.getRemaining(rexBuilder))
    if (condition.isAlwaysTrue) {
      None
    } else {
      Some(condition)
    }
  }


  /**
    * Gets the join key pairs from left input key to temporal table key
    * @param joinInfo the join information of temporal table join
    * @param temporalTableCalcProgram the calc programs on temporal table
    */
  private def getTemporalTableJoinKeyPairs(
    joinInfo: JoinInfo,
    temporalTableCalcProgram: Option[RexProgram]): util.List[IntPair] = {
    temporalTableCalcProgram match {
      case Some(program) =>
        // the target key of joinInfo is the calc output fields, we have to remapping to table here
        val keyPairs: util.List[IntPair] = new util.ArrayList[IntPair]()
        joinInfo.pairs().map {
          p =>
            val calcSrcIdx = getIdenticalSourceField(program, p.target)
            if (calcSrcIdx != -1) {
              keyPairs.add(new IntPair(p.source, calcSrcIdx))
            }
        }
        keyPairs
      case None => joinInfo.pairs()
    }
  }

  /**
    * Analyze the constant lookup keys in the temporal table from the calc program on the temporal
    * table.
    */
  def analyzeConstantLookupKeys(
    cluster: RelOptCluster,
    temporalTableCalcProgram: Option[RexProgram],
    indexKeys: util.List[IndexKey]): util.Map[Int, (InternalType, Object)] = {

    val constantKeyMap: util.Map[Int, (InternalType, Object)] =
      new util.HashMap[Int, (InternalType, Object)]
    // all the columns in index keys
    val allKeys = mutable.HashSet.empty[Int]
    indexKeys.map(_.getDefinedColumns.map(allKeys += _))

    if (temporalTableCalcProgram.isDefined && null != temporalTableCalcProgram.get.getCondition) {
      val program = temporalTableCalcProgram.get
      val condition = RexUtil.toCnf(
        cluster.getRexBuilder,
        program.expandLocalRef(program.getCondition))
      // presume 'A = 1 AND A = 2' will be reduced to ALWAYS_FALSE
      extractConstantKeysFromEquiCondition(condition, allKeys.toArray, constantKeyMap)
    }
    constantKeyMap
  }

  private def findMatchedIndex(
    allIndexes: util.List[IndexKey],
    joinKeyPairs: util.List[IntPair],
    constantLookupKeys: util.Map[Int, (InternalType, Object)]): Option[IndexKey] = {
    val lookupKeyCandidates = joinKeyPairs.map(_.target) ++ constantLookupKeys.keySet()
    // do validation later due to unified ErrorCode
    allIndexes.find(_.isIndex(lookupKeyCandidates.toArray))
  }

  // ----------------------------------------------------------------------------------------
  //                             Physical Optimization Utilities
  // ----------------------------------------------------------------------------------------

  // this is highly inspired by Calcite's RexProgram#getSourceField(int)
  private def getIdenticalSourceField(rexProgram: RexProgram, outputOrdinal: Int): Int = {
    assert((outputOrdinal >= 0) && (outputOrdinal < rexProgram.getProjectList.size()))
    val project = rexProgram.getProjectList.get(outputOrdinal)
    var index = project.getIndex
    while (true) {
      var expr = rexProgram.getExprList.get(index)
      expr match {
        case call: RexCall if call.getOperator == SqlStdOperatorTable.IN_FENNEL =>
          // drill through identity function
          expr = call.getOperands.get(0)
        case call: RexCall if call.getOperator == SqlStdOperatorTable.CAST =>
          // drill through identity function
          expr = call.getOperands.get(0)
        case _ =>
      }
      expr match {
        case ref: RexLocalRef => index = ref.getIndex
        case ref: RexInputRef => return ref.getIndex
        case _ => return -1
      }
    }
    -1
  }

  private def extractConstantKeysFromEquiCondition(
    condition: RexNode,
    indexKeys: Array[Int],
    constantKeyMap: util.Map[Int, (InternalType, Object)]): Unit = {

    condition match {
      case c: RexCall if c.getKind == SqlKind.AND =>
        c.getOperands.foreach(r => extractConstantKeys(r, indexKeys, constantKeyMap))
      case rex: RexNode => extractConstantKeys(rex, indexKeys, constantKeyMap)
      case _ =>
    }
  }

  private def extractConstantKeys(
    pred: RexNode,
    keyIndexes: Array[Int],
    constantKeyMap: util.Map[Int, (InternalType, Object)])
  : util.Map[Int, (InternalType, Object)] = {

    pred match {
      case c: RexCall if c.getKind == SqlKind.EQUALS =>
        val leftTerm = c.getOperands.get(0)
        val rightTerm = c.getOperands.get(1)
        val t = FlinkTypeFactory.toInternalType(rightTerm.getType)
        leftTerm match {
          case rexLiteral: RexLiteral =>
            rightTerm match {
              case r: RexInputRef if keyIndexes.contains(r.getIndex) =>
                constantKeyMap.put(
                  r.getIndex,
                  (t, RexLiteralUtil.literalValue(rexLiteral)))
              case _ =>
            }
          case _ => rightTerm match {
            case rexLiteral: RexLiteral =>
              leftTerm match {
                case r: RexInputRef if keyIndexes.contains(r.getIndex) =>
                  constantKeyMap.put(
                    r.getIndex,
                    (t, RexLiteralUtil.literalValue(rexLiteral)))
                case _ =>
              }
            case _ =>
          }
        }
      case _ =>
    }
    constantKeyMap
  }

  // ----------------------------------------------------------------------------------------
  //                                       Validation
  // ----------------------------------------------------------------------------------------

  def validate(
    tableSource: TableSource,
    period: RexNode,
    inputSchema: BaseRowSchema,
    tableSourceSchema: BaseRowSchema,
    joinKeyPairs: util.List[IntPair],
    constantLookupKeys: util.Map[Int, (InternalType, Object)],
    allIndexKeys: util.List[IndexKey],
    joinedIndex: Option[IndexKey],
    joinType: JoinRelType): Unit = {

    if (joinKeyPairs.isEmpty && constantLookupKeys.isEmpty) {
      throw new TableException(
        "Temporal table join requires an equality condition on ALL of " +
          "temporal table's primary key(s) or unique key(s) or index field(s).")
    }

    // checked index never be null, so declared index also not null.
    if (allIndexKeys.isEmpty) {
      throw new TableException(
        "Temporal table require to define an primary key or unique key or index.")
    }

    // check a matched index exist
    if (joinedIndex.isEmpty) {
      throw new TableException(
        "Temporal table join requires an equality condition on ALL of " +
          "temporal table's primary key(s) or unique key(s) or index field(s).")
    }

    if (!tableSource.isInstanceOf[LookupableTableSource[_]]) {
      throw new TableException("TableSource must implement LookupableTableSource interface " +
                                 "if it is used as a temporal table.")
    }

    val checkedLookupKeys = joinedIndex.get.getDefinedColumns

    val lookupKeyPairs = joinKeyPairs.filter(p => checkedLookupKeys.contains(p.target))
    val leftKeys = lookupKeyPairs.map(_.source).toArray
    val rightKeys = lookupKeyPairs.map(_.target) ++ constantLookupKeys.keys
    val leftKeyTypes = leftKeys.map(inputSchema.fieldTypeInfos(_))
    // use original keyPair to validate key types (rigthKeys may include constant keys)
    val rightKeyTypes = lookupKeyPairs.map(p => tableSourceSchema.fieldTypeInfos(p.target))

    // check type
    leftKeyTypes.zip(rightKeyTypes).foreach(f => {
      if (f._1 != f._2) {
        val leftNames = leftKeys.map(inputSchema.fieldNames(_))
        val rightNames = rightKeys.map(tableSourceSchema.fieldNames(_))

        val leftNameTypes = leftKeyTypes
                            .zip(leftNames)
                            .map(f => s"${f._2}[${f._1.toString}]")

        val rightNameTypes = rightKeyTypes
                             .zip(rightNames)
                             .map(f => s"${f._2}[${f._1.toString}]")

        val condition = leftNameTypes
                        .zip(rightNameTypes)
                        .map(f => s"${f._1}=${f._2}")
                        .mkString(", ")
        throw new TableException("Join: Equality join predicate on incompatible types. " +
                                   s"And the condition is $condition")
      }
    })

    if (joinType != JoinRelType.LEFT && joinType != JoinRelType.INNER) {
      throw new TableException(
        "Temporal table join currently only support INNER JOIN and LEFT JOIN, " +
          "but was " + joinType.toString + " JOIN")
    }

    val tableReturnType = TypeConverters.createExternalTypeInfoFromDataType(
      tableSource.getReturnType)
    if (!tableReturnType.isInstanceOf[BaseRowTypeInfo] &&
      !tableReturnType.isInstanceOf[RowTypeInfo]) {
      throw new TableException(
        "Temporal table join only support Row or BaseRow type as return type of temporal table." +
          " But was " + tableReturnType)
    }

    // period specification check
    period.getType match {
      case t: TimeIndicatorRelDataType if !t.isEventTime => // ok
      case _ =>
        throw new TableException(
          "Currently only support join temporal table as of on left table's proctime field")
    }
    period match {
      case r: RexFieldAccess if r.getReferenceExpr.isInstanceOf[RexCorrelVariable] =>
      // it's left table's field, ok
      case call: RexCall if call.getOperator == ScalarSqlFunctions.PROCTIME =>
      // it is PROCTIME() call, ok
      case _ =>
        throw new TableException(
          "Currently only support join temporal table as of on left table's proctime field.")
    }

    // success
  }

  def checkUdtfReturnType(
    tableDesc: String,
    tableReturnTypeInfo: TypeInformation[_],
    udtfReturnType: DataType,
    extractedUdtfReturnTypeInfo: TypeInformation[_]): Unit = {
    if (udtfReturnType == null) {
      if (!rowTypeEquals(tableReturnTypeInfo, extractedUdtfReturnTypeInfo)) {
        throw new TableException(
          s"The TableSource [$tableDesc] return type $tableReturnTypeInfo " +
            s"do not match its lookup function extracted return type $extractedUdtfReturnTypeInfo")
      }
      if (extractedUdtfReturnTypeInfo.getTypeClass != classOf[BaseRow] &&
        extractedUdtfReturnTypeInfo.getTypeClass != classOf[Row]) {
        throw new TableException(
          "Result type of the async lookup TableFunction of TableSource " +
            s"'$tableDesc' is " +
            s"$extractedUdtfReturnTypeInfo type, " +
            s"currently only Row and BaseRow are supported.")
      }
    } else {
      val udtfReturnTypeInfo = TypeConverters.createExternalTypeInfoFromDataType(udtfReturnType)
      if (!rowTypeEquals(tableReturnTypeInfo, udtfReturnTypeInfo)) {
        throw new TableException(
          s"The TableSource [$tableDesc] return type $tableReturnTypeInfo " +
            s"do not match its lookup function return type $udtfReturnTypeInfo")
      }
      if (!udtfReturnTypeInfo.isInstanceOf[BaseRowTypeInfo] &&
        !udtfReturnTypeInfo.isInstanceOf[RowTypeInfo]) {
        throw new TableException(
          "Result type of the async lookup TableFunction of TableSource " +
            s"'$tableDesc' is $udtfReturnTypeInfo type, " +
            s"currently only Row and BaseRow are supported.")
      }
    }
  }

  // ----------------------------------------------------------------------------------------
  //                              toString Utilities
  // ----------------------------------------------------------------------------------------

  private def joinSelectionToString(inputType: RelDataType): String = {
    inputType.getFieldNames.toList.mkString(", ")
  }

  private def joinConditionToString(
    inputType: RelDataType,
    joinCondition: RexNode,
    expression: (RexNode, List[String], Option[List[RexNode]]) => String): String = {

    val inFields = inputType.getFieldNames.toList
    if (joinCondition != null) {
      expression(joinCondition, inFields, None)
    } else {
      null
    }
  }

  private def joinTypeToString(joinType: JoinRelType): String = joinType match {
    case JoinRelType.INNER => "InnerJoin"
    case JoinRelType.LEFT => "LeftOuterJoin"
    case JoinRelType.RIGHT => "RightOuterJoin"
    case JoinRelType.FULL => "FullOuterJoin"
  }

  private def joinToString(
    tableSource: LookupableTableSource[_],
    joinType: JoinRelType,
    joinResultSchema: BaseRowSchema,
    inputSchema: BaseRowSchema,
    tableSchema: BaseRowSchema,
    joinCondition: Option[RexNode],
    constantLookupKeys: util.Map[Int, (InternalType, Object)],
    lookupKeyPairs: util.List[IntPair],
    expression: (RexNode, List[String], Option[List[RexNode]]) => String): String = {

    val isAsyncEnabled = if (tableSource.getLookupConfig != null) {
      tableSource.getLookupConfig.isAsyncEnabled
    } else {
      (new LookupConfig).isAsyncEnabled
    }
    val prefix = if (isAsyncEnabled) {
      "AsyncJoinTable"
    } else {
      "JoinTable"
    }
    var str = s"$prefix(table: (${tableSource.explainSource()})" +
      s", joinType: ${joinTypeToString(joinType)}" +
      s", join: (${joinSelectionToString(inputSchema.relDataType)}), "

    val inputFieldNames = inputSchema.fieldNames
    val tableFieldNames = tableSchema.fieldNames
    val keyPairNames = lookupKeyPairs.map { p =>
      s"${inputFieldNames(p.source)}=${
        if (p.target > -1) tableFieldNames(p.target) else -1
      }"
    }
    str += s" on: (${keyPairNames.mkString(", ")}"
    str +=
      s"${constantLookupKeys.map(k => tableFieldNames(k._1) + " = " + k._2)
          .mkString(", ")})"
    if (joinCondition.isDefined) {
      val joinConditionString = joinConditionToString(
        joinResultSchema.relDataType,
        joinCondition.get,
        expression)
      str += s", where: ($joinConditionString)"
    }
    str += ")"
    str
  }

  def joinExplainTerms(
    pw: RelWriter,
    tableSource: TableSource,
    inputType: RelDataType,
    joinResultType: RelDataType,
    calcProgram: Option[RexProgram],
    lookupKeyPairs: util.List[IntPair],
    joinCondition: Option[RexNode],
    joinType: JoinRelType,
    period: RexNode,
    expression: (RexNode, List[String], Option[List[RexNode]]) => String): RelWriter = {

    val condition: String = if (calcProgram.isDefined) {
      CalcUtil.conditionToString(calcProgram.get, expression)
    } else {
      ""
    }
    var source = tableSource.explainSource()
    if (source == null || source.isEmpty) {
      source = TableConnectorUtil.generateRuntimeName(
        tableSource.getClass, tableSource.getTableSchema.getColumnNames)
    }

    val inputFieldNames = inputType.getFieldNames
    val tableFieldNames = tableSource.getTableSchema.getColumnNames
    val keyPairNames = lookupKeyPairs.map { p =>
      s"${inputFieldNames(p.source)}=${
        if (p.target >= 0 && p.target < tableFieldNames.length) tableFieldNames(p.target) else -1
      }"
    }

    pw.item("join", joinSelectionToString(joinResultType))
    .item("source", source)
    .item("on", keyPairNames.mkString(", "))
    .item("joinType", joinTypeToString(joinType))
    .itemIf("where", condition, !condition.isEmpty)
    .itemIf("joinCondition",
            joinConditionToString(joinResultType, joinCondition.orNull, expression),
            joinCondition.isDefined)
    .item("period", period)
  }
}