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

import org.apache.calcite.avatica.util.DateTimeUtils
import org.apache.commons.codec.binary.Base64
import org.apache.flink.runtime.io.disk.iomanager.IOManager
import org.apache.flink.runtime.memory.MemoryManager
import org.apache.flink.table.api.TableConfig
import org.apache.flink.table.api.functions.{FunctionContext, UserDefinedFunction}
import org.apache.flink.table.api.types.{DataTypes, InternalType, RowType}
import org.apache.flink.table.codegen.CodeGenUtils._
import org.apache.flink.table.codegen.CodeGeneratorContext._
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils
import org.apache.flink.table.dataformat._
import org.apache.flink.table.dataformat.util.BaseRowUtil
import org.apache.flink.table.runtime.AbstractStreamOperatorWithMetrics
import org.apache.flink.table.runtime.functions.FunctionContextImpl
import org.apache.flink.table.runtime.util.{ResettableExternalBuffer, RowIterator}
import org.apache.flink.table.typeutils.{AbstractRowSerializer, BaseRowSerializer, BinaryRowSerializer}
import org.apache.flink.table.util.collections._
import org.apache.flink.types.Row
import org.apache.flink.util.{InstantiationUtil, Preconditions}

import scala.collection.mutable

/**
 * The context for code generator, maintaining various reusable statements that could be insert
 * into different code sections in the final generated class.
 */
class CodeGeneratorContext(val tableConfig: TableConfig, val supportReference: Boolean) {

  // set of inner class definition statements that will be added only once
  private val reusableInnerClassDefinitionStatements: mutable.Map[String, String] =
    mutable.Map[String, String]()

  // set of member statements that will be added only once
  private val reusableMemberStatements: mutable.LinkedHashSet[String] =
    mutable.LinkedHashSet[String]()

  // Referenced objects of code generator classes to avoid serialization and base64.
  val references: mutable.ArrayBuffer[AnyRef] = new mutable.ArrayBuffer[AnyRef]()

  // set of constructor statements that will be added only once
  private val reusableInitStatements: mutable.LinkedHashSet[String] =
    mutable.LinkedHashSet[String]()

  // set of statements that will be added only once per record
  private val reusablePerRecordStatements: mutable.LinkedHashSet[String] =
    mutable.LinkedHashSet[String]()

  // set of open statements for RichFunction that will be added only once
  private val reusableOpenStatements: mutable.LinkedHashSet[String] =
    mutable.LinkedHashSet[String]()

  // set of close statements for RichFunction that will be added only once
  private val reusableCloseStatements: mutable.LinkedHashSet[String] =
    mutable.LinkedHashSet[String]()

  // set of endInput statements that will be added only once
  private val reusableEndInputStatements: mutable.LinkedHashSet[String] =
    mutable.LinkedHashSet[String]()

  // set of statements for cleanup dataview that will be added only once
  // we use a LinkedHashSet to keep the insertion order
  private val reusableCleanupStatements = mutable.LinkedHashSet[String]()

  // map of initial input unboxing expressions that will be added only once
  // (inputTerm, index) -> expr
  val reusableInputUnboxingExprs: mutable.Map[(String, Int), GeneratedExpression] =
    mutable.Map[(String, Int), GeneratedExpression]()

  // set of constructor statements that will be added only once
  private val reusableConstructorStatements: mutable.LinkedHashSet[(String, String)] =
    mutable.LinkedHashSet[(String, String)]()

  // map of type serializer expressions
  private val reusableTypeSerializerExprs: mutable.Map[InternalType, String] =
    mutable.Map[InternalType,  String]()

  // map of generic type expressions
  private val reusableTypeExprs: mutable.Map[InternalType, String] =
    mutable.Map[InternalType,  String]()

  // map of string constants
  private val reusableStringConstants: mutable.Map[String, String] = mutable.Map[String,  String]()

  /**
    * statements will be placed in the member area of the class if generated code is not split
    * or in local variables of method. This statements is mapping from method name to field
    * statements.
    */
  private val reusableMethodFieldStatements = mutable.Map[String, mutable.LinkedHashSet[String]]()

  /**
    * statements will be placed in the member area of the class if generated code is not split
    * or in local variables of method. This is the default reusable statements, if
    * [[startNewFieldStatements()]] is called, the default statements will be dropped and replaced
    * with a new statements.
    */
  private var reusableFieldStatements = mutable.LinkedHashSet[String]()

  /**
    * start a new field statements for a generated class with many methods.
    * @param methodName the method name which the fields will be placed into if code is not split.
    */
  def startNewFieldStatements(methodName: String): Unit = {
    reusableFieldStatements = mutable.LinkedHashSet[String]()
    reusableMethodFieldStatements(methodName) = reusableFieldStatements
  }

  /**
    * the class is used as the  generated operator code's base class.
    */
  private var operatorBaseClass: Class[_] = classOf[AbstractStreamOperatorWithMetrics[_]]

  /**
    * @return code block of statements that need to be placed in the member area of the class
    *         (e.g. inner class definition)
    */
  def reuseInnerClassDefinitionCode(): String = {
    reusableInnerClassDefinitionStatements.values.mkString("\n")
  }

  /**
   * @return code block of statements that need to be placed in the member area of the class
   *         (e.g. member variables and their initialization)
   */
  def reuseMemberCode(): String = {
    reusableMemberStatements.mkString("\n")
  }

  /**
    * @return code block of statements that will be placed in the member area of the class
    *         if generated code is split or in local variables of method
    */
  def reuseFieldCode(methodName: String = null): String = {
    if (methodName == null) {
      reusableFieldStatements.mkString("\n")
    } else {
      reusableMethodFieldStatements(methodName).mkString("\n")
    }
  }

  /**
   * @return code block of statements that need to be placed in the constructor
   */
  def reuseInitCode(): String = {
    reusableInitStatements.mkString("\n")
  }

  /**
   * @return code block of statements that need to be placed in the per recode process block
   *         (e.g. Function or StreamOperator's processElement)
   */
  def reusePerRecordCode(): String = {
    reusablePerRecordStatements.mkString("\n")
  }

  /**
   * @return code block of statements that need to be placed in the open() method
   *         (e.g. RichFunction or StreamOperator)
   */
  def reuseOpenCode(): String = {
    reusableOpenStatements.mkString("\n")
  }

  /**
   * @return code block of statements that need to be placed in the close() method
   *         (e.g. RichFunction or StreamOperator)
   */
  def reuseCloseCode(): String = {
    reusableCloseStatements.mkString("\n")
  }

  /**
    * @return code block of statements that need to be placed in the endInput() method
    *         (StreamOperator)
    */
  def reuseEndInputCode(): String = {
    reusableEndInputStatements.mkString("\n")
  }

  /**
    * @return code block of statements that need to be placed in the cleanup() method of
    *         [AggregationsFunction]
    */
  def reuseCleanupCode(): String = {
    reusableCleanupStatements.mkString("", "\n", "\n")
  }

  /**
   * @return code block of statements that unbox input variables to a primitive variable
   *         and a corresponding null flag variable
   */
  def reuseInputUnboxingCode(): String = {
    reusableInputUnboxingExprs.values.map(_.code).mkString("\n")
  }

  def reuseInputUnboxingCode(terms: Set[String]): String = {
    val exprs = reusableInputUnboxingExprs.filter { entry =>
      val ((term, _), _) = entry
      terms contains term
    }
    val codes = for (((_, _), expr) <- exprs) yield expr.code
    codes.mkString("\n").trim
  }

  /**
   * @return code block of constructor statements
   */
  def reuseConstructorCode(className: String): String = {
    reusableConstructorStatements.map { case (params, body) =>
      s"""
         |public $className($params) throws Exception {
         |  this();
         |  $body
         |}
         |""".stripMargin
    }.mkString("\n")
  }

  def addReusableInnerClass(className: String, statements: String): Unit = {
    reusableInnerClassDefinitionStatements(className) = statements
  }

  def addReusableMember(declareCode: String, initCode: String = ""): Unit = {
    reusableMemberStatements.add(declareCode)
    if (!initCode.isEmpty) reusableInitStatements.add(initCode)
  }

  def getTableConfig: TableConfig = this.tableConfig

  /**
    * @param fields statements of local field like "boolean isNull;"
    * this fields may be put in class or in method
    */
  def addAllReusableFields(fields: Set[String]): Unit = {
    this.reusableFieldStatements ++= fields
  }

  def addPerRecordStatement(s: String): Unit = reusablePerRecordStatements.add(s)

  def addReusableOpenStatement(s: String): Unit = reusableOpenStatements.add(s)

  def addReusableCloseStatement(s: String): Unit = reusableCloseStatements.add(s)

  def addReusableEndInputStatement(s: String): Unit = reusableEndInputStatements.add(s)

  def addReusableCleanupStatement(s: String): Unit = reusableCleanupStatements.add(s)

  def getReusableInputUnboxingExprs(inputTerm: String, index: Int): Option[GeneratedExpression] =
    reusableInputUnboxingExprs.get((inputTerm, index))

  def addReusableInputUnboxingExprs(
      inputTerm: String,
      index: Int,
      expr: GeneratedExpression): Unit =
    reusableInputUnboxingExprs((inputTerm, index)) = expr

  /**
   * generator a output record..
   * The passed [[InternalType]] defines the type class to be instantiated.
   * @param reused If objects or variables can be reused, they will be added a reusable output
   * record to the member area of the generated class. If not they will be return directly.
   */
  def addOutputRecord(
      t: InternalType,
      clazz: Class[_],
      outRecordTerm: String,
      outRecordWriterTerm: Option[String] = None,
      reused: Boolean = true): String = {
    val statement = t match {
      case rt: RowType if clazz == classOf[BinaryRow] =>
        val writerTerm = outRecordWriterTerm.getOrElse(
          throw new CodeGenException("No writer is specified when writing BinaryRow record.")
        )
        val binaryRowWriter = classOf[BinaryRowWriter].getName
        val typeTerm = clazz.getCanonicalName
        s"""
           |final $typeTerm $outRecordTerm = new $typeTerm(${rt.getArity});
           |final $binaryRowWriter $writerTerm = new $binaryRowWriter($outRecordTerm);
           |""".stripMargin.trim
      case rt: RowType if classOf[ObjectArrayRow].isAssignableFrom(clazz) =>
        val typeTerm = clazz.getCanonicalName
        s"final $typeTerm $outRecordTerm = new $typeTerm(${rt.getArity});"
      case rt: RowType if clazz == classOf[JoinedRow] =>
        val typeTerm = clazz.getCanonicalName
        s"final $typeTerm $outRecordTerm = new $typeTerm();"
      case _ =>
        val typeTerm = boxedTypeTermForType(t)
        s"final $typeTerm $outRecordTerm = new $typeTerm();"
    }
    if (reused) {
      reusableMemberStatements.add(statement)
      ""
    } else {
      statement
    }
  }

  /**
    * Adds a reusable null [[org.apache.flink.table.dataformat.GenericRow]] to the member area.
    */
  def addReusableNullRow(rowTerm: String, arity: Int): String = {
    addOutputRecord(
      new RowType(
        (0 until arity).map(_ => DataTypes.INT): _*),
      classOf[GenericRow],
      rowTerm)
  }

  def addReusableTypeSerializer(t: InternalType): String = {
    // if type serializer has been used before, we can reuse the code that
    // has already been generated
    reusableTypeSerializerExprs.get(t) match {
      case Some(expr) => expr

      case None =>
        val term = newName("typeSerializer")
        val declareCode =
          s"org.apache.flink.api.common.typeutils.TypeSerializer $term;"
        val ser = DataTypes.createInternalSerializer(t)
        val initCode = if (supportReference) {
          s"$term = ${addReferenceObj(ser)};"
        } else {
          val byteArray = InstantiationUtil.serializeObject(ser)
          val serializedData = Base64.encodeBase64URLSafeString(byteArray)
          s"""
             |$term = (org.apache.flink.api.common.typeutils.TypeSerializer)
             | org.apache.flink.util.InstantiationUtil.deserializeObject(
             |  org.apache.commons.codec.binary.Base64.decodeBase64("$serializedData"),
             |    Thread.currentThread().getContextClassLoader());
           """.stripMargin
        }
        addReusableMember(declareCode, initCode)

        reusableTypeSerializerExprs(t) = term
        term
    }
  }

  def addReusableGenericType(t: InternalType): String = {
    // if type information has been used before, we can reuse the code that
    // has already been generated
    reusableTypeExprs.get(t) match {
      case Some(expr) => expr

      case None =>
        val term = newName("typeInfo")
        val typeTerm = classOf[InternalType].getName
        val declareCode = s"transient $typeTerm $term;"
        val initCode = if (supportReference) {
          s"$term = ${addReferenceObj(t)};"
        } else {
          val byteArray = InstantiationUtil.serializeObject(t)
          val serializedData = Base64.encodeBase64URLSafeString(byteArray)
          s"""
             |$term = ($typeTerm) org.apache.flink.util.InstantiationUtil.deserializeObject(
             |  org.apache.commons.codec.binary.Base64.decodeBase64("$serializedData"),
             |    Thread.currentThread().getContextClassLoader());
           """.stripMargin
        }
        addReusableMember(declareCode, initCode)

        reusableTypeExprs(t) = term
        term
    }
  }

  /**
   * Adds a reusable [[java.lang.reflect.Field]] to the member area of the generated class.
   * The field can be used for accessing POJO fields more efficiently during runtime, however,
   * the field does not have to be public.
   */
  def addReusablePrivateFieldAccess(clazz: Class[_], fieldName: String): String = {
    val fieldTerm = s"field_${clazz.getCanonicalName.replace('.', '$')}_$fieldName"
    val fieldExtraction =
      s"""
         |final java.lang.reflect.Field $fieldTerm =
         |  org.apache.flink.api.java.typeutils.TypeExtractor.getDeclaredField(
         |    ${clazz.getCanonicalName}.class, "$fieldName");
         |""".stripMargin
    val fieldAccessibility =
      s"""
         |$fieldTerm.setAccessible(true);
         |""".stripMargin

    addReusableMember(fieldExtraction, fieldAccessibility)
    fieldTerm
  }

  /**
   * Adds a reusable set to the member area of the generated class.
   */
  def addReusableSet(elements: Seq[GeneratedExpression], t: InternalType): String = {
    val fieldTerm = newName("set")

    val setTerm = (t match {
      case DataTypes.BYTE => classOf[ByteSet]
      case DataTypes.SHORT => classOf[ShortSet]
      case DataTypes.INT => classOf[IntSet]
      case DataTypes.LONG => classOf[LongSet]
      case DataTypes.FLOAT => classOf[FloatSet]
      case DataTypes.DOUBLE => classOf[DoubleSet]
      case _ => classOf[ObjectHashSet[_]]
    }).getCanonicalName

    addReusableMember(
      s"final $setTerm $fieldTerm;",
      s"$fieldTerm = new $setTerm(${elements.size});")

    elements.foreach { element =>
      val content =
        s"""
           |${element.code}
           |if (${element.nullTerm}) {
           |  $fieldTerm.addNull();
           |} else {
           |  $fieldTerm.add(${element.resultTerm});
           |}
           |""".stripMargin
      reusableInitStatements.add(content)
    }
    reusableInitStatements.add(s"$fieldTerm.optimize();")

    fieldTerm
  }

  /**
   * Adds a reusable timestamp to the beginning of the SAM of the generated class.
   */
  def addReusableTimestamp(): String = {
    val fieldTerm = s"timestamp"
    val field =
      s"""
         |final long $fieldTerm = java.lang.System.currentTimeMillis();
         |""".stripMargin
    reusablePerRecordStatements.add(field)
    fieldTerm
  }

  /**
   * Adds a reusable local timestamp to the beginning of the SAM of the generated class.
   */
  def addReusableLocalTimestamp(): String = {
    addReusableTimestamp()
  }

  /**
   * Adds a reusable time to the beginning of the SAM of the generated class.
   */
  def addReusableTime(): String = {
    val fieldTerm = s"time"
    val timestamp = addReusableTimestamp()
    // adopted from org.apache.calcite.runtime.SqlFunctions.currentTime()
    val field =
      s"""
         |final int $fieldTerm = (int) ($timestamp % ${DateTimeUtils.MILLIS_PER_DAY});
         |if (time < 0) {
         |  time += ${DateTimeUtils.MILLIS_PER_DAY};
         |}
         |""".stripMargin
    reusablePerRecordStatements.add(field)
    fieldTerm
  }

  /**
   * Adds a reusable local time to the beginning of the SAM of the generated class.
   */
  def addReusableLocalTime(): String = {
    val fieldTerm = s"localtime"
    val timeZone = addReusableTimeZone()
    val localtimestamp = addReusableLocalTimestamp()
    // adopted from org.apache.calcite.runtime.SqlFunctions.localTime()
    val field =
      s"""
         |final int $fieldTerm = (int) ( ($localtimestamp + $timeZone.getOffset($localtimestamp))
         |                              % ${DateTimeUtils.MILLIS_PER_DAY});
         |""".stripMargin
    reusablePerRecordStatements.add(field)
    fieldTerm
  }

  /**
   * Adds a reusable date to the beginning of the SAM of the generated class.
   */
  def addReusableDate(): String = {
    val fieldTerm = s"date"
    val timestamp = addReusableTimestamp()
    val time = addReusableTime()
    val timeZone = addReusableTimeZone()

    // adopted from org.apache.calcite.runtime.SqlFunctions.currentDate()
    val field =
      s"""
         |final int $fieldTerm = (int) (($timestamp + $timeZone.getOffset($timestamp))
         |                              / ${DateTimeUtils.MILLIS_PER_DAY});
         |if ($time < 0) {
         |  $fieldTerm -= 1;
         |}
         |""".stripMargin
    reusablePerRecordStatements.add(field)
    fieldTerm
  }

  /**
   * Adds a reusable [[java.util.Random]] to the member area of the generated class.
   *
   * The seed parameter must be a literal/constant expression.
   *
   * @return member variable term
   */
  def addReusableRandom(seedExpr: Option[GeneratedExpression], nullCheck: Boolean): String = {
    val fieldTerm = newName("random")

    val field =
      s"""
         |final java.util.Random $fieldTerm;
         |""".stripMargin

    val fieldInit = seedExpr match {
      case Some(s) if nullCheck =>
        s"""
           |${s.code}
           |if (!${s.nullTerm}) {
           |  $fieldTerm = new java.util.Random(${s.resultTerm});
           |}
           |else {
           |  $fieldTerm = new java.util.Random();
           |}
           |""".stripMargin
      case Some(s) =>
        s"""
           |${s.code}
           |$fieldTerm = new java.util.Random(${s.resultTerm});
           |""".stripMargin
      case _ =>
        s"""
           |$fieldTerm = new java.util.Random();
           |""".stripMargin
    }

    addReusableMember(field, fieldInit)
    fieldTerm
  }

  def addReferenceObj(obj: AnyRef, className: String = null): String = {
    val idx = references.length

    // make a deep copy of the object
    val byteArray = InstantiationUtil.serializeObject(obj)
    val objCopy: AnyRef = InstantiationUtil.deserializeObject(
      byteArray,
      Thread.currentThread().getContextClassLoader)
    references += objCopy

    val clsName = Option(className).getOrElse(obj.getClass.getName)
    s"(($clsName) references[$idx])"
  }

  /**
    * Adds a reusable Object to the member area of the generated class
    */
  def addReusableObject(obj: AnyRef, termPrefix: String, className: String = null): String = {
    val fieldTerm = newName(termPrefix)
    val typeTerm = obj.getClass.getCanonicalName

    val initCode = if (supportReference) {
      s"$fieldTerm = ${addReferenceObj(obj, className)};"
    } else {
      val byteArray = InstantiationUtil.serializeObject(obj)
      val serializedData = Base64.encodeBase64URLSafeString(byteArray)
      s"""
         | $fieldTerm = ($typeTerm)
         |      ${classOf[InstantiationUtil].getCanonicalName}.deserializeObject(
         |         ${classOf[Base64].getCanonicalName}.decodeBase64("$serializedData"),
         |         Thread.currentThread().getContextClassLoader());
           """.stripMargin
    }

    addReusableMember(s"private $typeTerm $fieldTerm;", initCode)
    fieldTerm
  }

  /**
   * Adds a reusable [[UserDefinedFunction]] to the member area of the generated class.
   */
  def addReusableFunction(
      function: UserDefinedFunction,
      functionContextClass: Class[_ <: FunctionContext] = classOf[FunctionContextImpl],
      constructorTerm: String = null): String = {
    Preconditions.checkNotNull(functionContextClass)
    val classQualifier = function.getClass.getCanonicalName
    val fieldTerm = udfFieldName(function)

    val fieldFunction =
      s"""
         |final $classQualifier $fieldTerm;
         |""".stripMargin
    reusableMemberStatements.add(fieldFunction)

    val initCode = if (supportReference) {
      s"$fieldTerm = (${addReferenceObj(function)});"
    } else {
      val functionSerializedData = UserDefinedFunctionUtils.serialize(function)
      s"""
         |$fieldTerm = ($classQualifier)
         |${UserDefinedFunctionUtils.getClass.getName.stripSuffix("$")}
         |.deserialize("$functionSerializedData");
       """.stripMargin
    }

    reusableInitStatements.add(initCode)

    val openFunction = if (constructorTerm != null) {
      s"""
         |$fieldTerm.open(new ${functionContextClass.getCanonicalName}($constructorTerm));
       """.stripMargin
    } else {
      s"""
         |$fieldTerm.open(new ${functionContextClass.getCanonicalName}(getRuntimeContext()));
       """.stripMargin
    }
    reusableOpenStatements.add(openFunction)

    val closeFunction =
      s"""
         |$fieldTerm.close();
       """.stripMargin
    reusableCloseStatements.add(closeFunction)

    fieldTerm
  }

  def addReusableFunction(funcName: String, funcCode: String, classQualifier: String): String = {
    val fieldTerm = newName(s"function_$funcName")

    val base64FuncCode = Base64.encodeBase64URLSafeString(funcCode.getBytes("UTF-8"))

    val fieldFunction =
      s"""
         |$classQualifier $fieldTerm;
         |
         |""".stripMargin
    reusableMemberStatements.add(fieldFunction)

    val openFunction =
      s"""
         |$fieldTerm = ($classQualifier)
         |  org.apache.flink.table.codegen.CodeGenUtils.compile(
         |    getRuntimeContext().getUserCodeClassLoader(),
         |    "$funcName",
         |    new String(org.apache.commons.codec.binary.Base64.decodeBase64("$base64FuncCode"),
         |      "UTF-8")).newInstance();
         |""".stripMargin
    reusableOpenStatements.add(openFunction)

    val closeFunction =
      s"""
         |$fieldTerm.close();
         |""".stripMargin
    reusableCloseStatements.add(closeFunction)

    fieldTerm
  }

  /**
   * Adds a reusable constructor statement with the given parameter types.
   */
  def addReusableConstructor(parameterTypes: Class[_]*): Array[String] = {
    val parameters = mutable.ListBuffer[String]()
    val fieldTerms = mutable.ListBuffer[String]()
    val body = mutable.ListBuffer[String]()

    parameterTypes.zipWithIndex.foreach { case (t, index) =>
      val classQualifier = t.getCanonicalName
      val fieldTerm = newName(s"instance_${classQualifier.replace('.', '$')}")
      val field = s"final $classQualifier $fieldTerm;"
      reusableMemberStatements.add(field)
      fieldTerms += fieldTerm
      parameters += s"$classQualifier arg$index"
      body += s"$fieldTerm = arg$index;"
    }

    reusableConstructorStatements.add((parameters.mkString(","), body.mkString("", "\n", "\n")))

    fieldTerms.toArray
  }

  /**
   * Adds a reusable [[java.util.ArrayList]] to the member area.
   */
  def addReusableList(fieldTerm: String): Unit = {
    val field =
      s"""
         |transient java.util.List $fieldTerm = null;
         |""".stripMargin
    val init =
      s"""
         |$fieldTerm = new java.util.ArrayList();
         |""".stripMargin
    addReusableMember(field, init)
  }

  def addReusableLogger(logTerm: String, clazzTerm: String): Unit = {
    val field =
      s"""
         |private static final org.slf4j.Logger $logTerm =
         |  org.slf4j.LoggerFactory.getLogger("$clazzTerm");
         |""".stripMargin
    addReusableMember(field)
  }

  /**
    * Adds a reusable constant to the member area of the generated [[Function]].
    *
    * @param constant constant expression
    * @return member variable term
    */
  def addReusableConstant(
      constant: GeneratedExpression,
      nullCheck: Boolean): GeneratedExpression = {
    require(constant.literal, "Literal expected")

    val fieldTerm = newName("constant")
    val nullTerm = fieldTerm + "isNull"

    val fieldType = primitiveTypeTermForType(constant.resultType)

    val field =
      s"""
         |private final $fieldType $fieldTerm;
         |private final boolean $nullTerm;
         |""".stripMargin
    reusableMemberStatements.add(field)

    val init =
      s"""
         |${constant.code}
         |$fieldTerm = ${constant.resultTerm};
         |$nullTerm = ${constant.nullTerm};
         |""".stripMargin
    reusableInitStatements.add(init)

    GeneratedExpression(fieldTerm, nullTerm, "", constant.resultType)
  }

  def newReusableField(name: String, fieldTypeTerm: String): String = {
    val fieldName = newName(name)
    reusableFieldStatements.add(s"$fieldTypeTerm $fieldName;")
    fieldName
  }

  def newReusableFields(names: Seq[String], fieldTypeTerms: Seq[String]): Seq[String] = {
    require(names.length == fieldTypeTerms.length)
    require(names.toSet.size == names.length, "Duplicated names")
    val fieldNames = newNames(names)
    fieldTypeTerms.zip(fieldNames).foreach(x => reusableFieldStatements.add(s"${x._1} ${x._2};"))
    fieldNames
  }

  def setOperatorBaseClass(operatorBaseClass: Class[_]): CodeGeneratorContext = {
    this.operatorBaseClass = operatorBaseClass
    this
  }

  def getOperatorBaseClass: Class[_] = this.operatorBaseClass

  def addReusableStringConstants(value: String): String =
    reusableStringConstants.get(value) match {
      case Some(field) => field
      case None =>
        val field = newName("str")
        addReusableMember(
          s"$BINARY_STRING $field = $BINARY_STRING.fromString(${"\""}$value${"\""});")
        reusableStringConstants(value) = field
        field
    }

  def addReusableTimeZone(): String = {
    val memberTerm = s"timeZone"
    val zoneID = tableConfig.getTimeZone.getID
    val stmt =
      s"""private static final java.util.TimeZone $DEFAULT_TIMEZONE_TERM =
         |                 java.util.TimeZone.getTimeZone("$zoneID");""".stripMargin

    addReusableMember(stmt)
    memberTerm
  }

  def addReusableResettableExternalBuffer(
    fieldTerm: String, memSize: Long, serializer: String): Unit = {
    val memManager = CodeGenUtils.newName("memManager")
    val ioManager = CodeGenUtils.newName("ioManager")
    addReusableOpenStatement(
      s"$MEMORY_MANAGER $memManager = getContainingTask().getEnvironment().getMemoryManager();")
    addReusableOpenStatement(
      s"$IO_MANAGER $ioManager = getContainingTask().getEnvironment().getIOManager();")

    val field =
      s"""
         |${CodeGeneratorContext.RESETTABLE_EXTERNAL_BUFFER} $fieldTerm = null;
         |""".stripMargin
    val open =
      s"""
         |$fieldTerm = new ${CodeGeneratorContext.RESETTABLE_EXTERNAL_BUFFER}(
         |    $memManager, $ioManager, $memManager
         |    .allocatePages(
         |        getContainingTask(), ((int) $memSize) / $memManager.getPageSize()), $serializer);
         |""".stripMargin
    addReusableMember(field)
    addReusableOpenStatement(open)
  }

  /**
    * Adds a reusable MessageDigest to the member area of the generated [[Function]].
    *
    * @return member variable term
    */
  def addReusableMessageDigest(algorithm: String): String = {
    val fieldTerm = newName("messageDigest")

    val field = s"final java.security.MessageDigest $fieldTerm;"
    reusableMemberStatements.add(field)

    val fieldInit =
      s"""
         |try {
         |  $fieldTerm = java.security.MessageDigest.getInstance("$algorithm");
         |} catch (java.security.NoSuchAlgorithmException e) {
         |  throw new RuntimeException("Algorithm for '$algorithm' is not available.", e);
         |}
         |""".stripMargin
    reusableInitStatements.add(fieldInit)

    fieldTerm
  }

  /**
    * Adds a constant SHA2 reusable MessageDigest to the member area of the generated [[Function]].
    *
    * @return member variable term
    */
  def addReusableSha2MessageDigest(constant: GeneratedExpression, nullCheck: Boolean): String = {
    require(constant.literal, "Literal expected")
    val fieldTerm = newName("messageDigest")

    val field =
      s"final java.security.MessageDigest $fieldTerm;"
    reusableMemberStatements.add(field)

    val bitLen = constant.resultTerm
    val init =
      s"""
         |if ($bitLen == 224 || $bitLen == 256 || $bitLen == 384 || $bitLen == 512) {
         |  try {
         |    $fieldTerm = java.security.MessageDigest.getInstance("SHA-" + $bitLen);
         |  } catch (java.security.NoSuchAlgorithmException e) {
         |    throw new RuntimeException(
         |      "Algorithm for 'SHA-" + $bitLen + "' is not available.", e);
         |  }
         |} else {
         |  throw new RuntimeException("Unsupported algorithm.");
         |}
         |""".stripMargin
    val nullableInit = if (nullCheck) {
      s"""
         |${constant.code}
         |if (${constant.nullTerm}) {
         |  $fieldTerm = null;
         |} else {
         |  $init
         |}
         |""".stripMargin
    } else {
      s"""
         |${constant.code}
         |$init
         |""".stripMargin
    }
    reusableInitStatements.add(nullableInit)

    fieldTerm
  }
}

object CodeGeneratorContext {

  def apply(tableConfig: TableConfig, supportReference: Boolean = false): CodeGeneratorContext =
    new CodeGeneratorContext(tableConfig, supportReference)

  val DEFAULT_INPUT1_TERM = "in1"

  val DEFAULT_INPUT2_TERM = "in2"

  val DEFAULT_COLLECTOR_TERM = "c"

  val DEFAULT_OUT_RECORD_TERM = "out"

  val DEFAULT_OPERATOR_COLLECTOR_TERM = "output"

  val DEFAULT_OUT_RECORD_WRITER_TERM = "outWriter"

  val DEFAULT_CONTEXT_TERM = "ctx"

  val ROW: String = classOf[Row].getCanonicalName

  val ROW_ITERATOR: String = classOf[RowIterator[_]].getCanonicalName

  val BINARY_ROW: String = classOf[BinaryRow].getCanonicalName

  val BINARY_STRING: String = classOf[BinaryString].getCanonicalName

  val BASE_ROW: String = classOf[BaseRow].getCanonicalName

  val JOINED_ROW: String = classOf[JoinedRow].getCanonicalName

  val GENERIC_ROW: String = classOf[GenericRow].getCanonicalName

  val MEMORY_MANAGER: String = classOf[MemoryManager].getCanonicalName

  val IO_MANAGER: String = classOf[IOManager].getCanonicalName

  val RESETTABLE_EXTERNAL_BUFFER: String = classOf[ResettableExternalBuffer].getCanonicalName

  val ABSTRACT_ROW_SERIALIZER: String = classOf[AbstractRowSerializer[_]].getCanonicalName

  val BASE_ROW_SERIALIZER: String = classOf[BaseRowSerializer[_]].getCanonicalName

  val BINARY_ROW_SERIALIZER: String = classOf[BinaryRowSerializer].getCanonicalName

  val BASE_ROW_UTIL: String = classOf[BaseRowUtil].getCanonicalName

  val DEFAULT_TIMEZONE_TERM = "timeZone"

  def udfFieldName(udf: UserDefinedFunction): String = s"function_${udf.functionIdentifier}"
}