/*
 * 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 java.lang.{Long => JLong}
import org.apache.calcite.rel.RelCollation
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.AggregateCall
import org.apache.calcite.rex._
import org.apache.calcite.sql.fun.SqlStdOperatorTable._
import org.apache.calcite.sql.SqlAggFunction
import org.apache.calcite.tools.RelBuilder
import org.apache.flink.api.common.functions.Function
import org.apache.flink.cep.pattern.conditions.{IterativeCondition, RichIterativeCondition}
import org.apache.flink.cep._
import org.apache.flink.configuration.Configuration
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{TableConfig, TableConfigOptions, TableException}
import org.apache.flink.table.api.types._
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.codegen.agg.AggsHandlerCodeGenerator
import org.apache.flink.table.codegen.CodeGenUtils._
import org.apache.flink.table.codegen.GeneratedExpression.{NEVER_NULL, NO_CODE}
import org.apache.flink.table.codegen.Indenter.toISC
import org.apache.flink.table.codegen.MatchCodeGenerator._
import org.apache.flink.table.plan.schema.BaseRowSchema
import org.apache.flink.table.plan.util.AggregateUtil
import org.apache.flink.table.dataformat.{BaseRow, GenericRow}
import org.apache.flink.table.functions.sql.ProctimeSqlFunction
import org.apache.flink.table.plan.util.MatchUtil.AggregationPatternVariableFinder
import org.apache.flink.table.runtime.conversion.DataStructureConverters.genToInternal
import org.apache.flink.table.runtime.functions.{AggsHandleFunction, ExecutionContextImpl}
import org.apache.flink.table.typeutils.TypeUtils
import org.apache.flink.table.utils.EncodingUtils
import org.apache.flink.util.Collector
import org.apache.flink.util.MathUtils.checkedDownCast

import _root_.scala.collection.JavaConversions._
import _root_.scala.collection.JavaConverters._
import _root_.scala.collection.mutable

/**
  * A code generator for generating CEP related functions.
  *
  * @param ctx the cotext of the code generator
  * @param nullableInput input(s) can be null.
  * @param nullCheck whether to do null check
  * @param patternNames sorted sequence of pattern variables
  * @param currentPattern if generating condition the name of pattern, which the condition will
  *                       be applied to
  */
class MatchCodeGenerator(
    ctx: CodeGeneratorContext,
    relBuilder: RelBuilder,
    nullableInput: Boolean,
    nullCheck: Boolean,
    patternNames: Seq[String],
    currentPattern: Option[String] = None,
    collectorTerm: String = CodeGeneratorContext.DEFAULT_COLLECTOR_TERM)
  extends ExprCodeGenerator(ctx, nullableInput, nullCheck) {

  private case class GeneratedPatternList(resultTerm: String, code: String, filled: Boolean = true)

  private case class GeneratedClassifierList(
      resultTerm: String, code: String, filled: Boolean = true)

  /**
    * Used to assign unique names for list of events per pattern variable name. Those lists
    * are treated as inputs and are needed by input access code.
    */
  private val reusablePatternLists: mutable.HashMap[(Boolean, String), GeneratedPatternList] =
    mutable.HashMap[(Boolean, String), GeneratedPatternList]()

  private val reusableClassiferList: mutable.HashMap[Boolean, GeneratedClassifierList] =
    mutable.HashMap[Boolean, GeneratedClassifierList]()

  private val reusableInputUnboxingExprs: mutable.Map[(String, Int), GeneratedExpression] =
    mutable.Map[(String, Int), GeneratedExpression]()

  private val reusablePerRecordStatements: mutable.LinkedHashSet[String] =
    mutable.LinkedHashSet[String]()

  /**
    * Used to deduplicate aggregations calculation. The deduplication is performed by
    * [[RexNode#toString]]. Those expressions needs to be accessible from splits, if such exists.
    */
  private val reusableAggregationExpr = new mutable.HashMap[String, GeneratedExpression]()

  /**
    * Context information used by Pattern reference variable to index rows mapped to it.
    * Indexes element at offset either from beginning or the end based on the value of first.
    */
  private var offset: Int = 0
  private var first : Boolean = false

  private var oneRowPerMatch: Boolean = false
  private var running: Boolean = false

  /**
    * Flags that tells if we generate expressions inside an aggregate. It tells how to access input
    * row.
    */
  private var isWithinAggExprState: Boolean = false

  /**
    * Used to collect all aggregates per pattern variable.
    */
  private val aggregatesPerVariable = new mutable.HashMap[(Boolean, String), AggBuilder]

  /**
    * Name of term in function used to transform input row into aggregate input row.
    */
  private val inputAggRowTerm = "inAgg"

  private val keyRowTerm = "keyRow"

  /**
    * @return term of pattern names
    */
  private val patternNamesTerm = newName("patternNames")

  private lazy val eventTypeTerm = boxedTypeTermForType(input1Type)

  /**
    * Sets the new reference variable indexing context. This should be used when resolving logical
    * offsets = LAST/FIRST
    *
    * @param first  true if indexing from the beginning, false otherwise
    * @param offset offset from either beginning or the end
    */
  private def updateOffsets(first: Boolean, offset: Int): Unit = {
    this.first = first
    this.offset = offset
  }

  /** Resets indexing context of Pattern variable. */
  private def resetOffsets(): Unit = {
    first = false
    offset = 0
  }

  private def setRunning(running: Boolean): Unit = {
    this.running = running
  }

  private def resetRunning(): Unit = {
    this.running = false
  }

  private def setOneRowPerMatch(oneRowPerMatch: Boolean): Unit = {
    this.oneRowPerMatch = oneRowPerMatch
  }

  private def reusePatternLists(): String = {
    reusablePatternLists.values.map(_.code).mkString("\n")
  }

  private def reuseClassifierLists(): String = {
    reusableClassiferList.values.map(_.code).mkString("\n")
  }

  private def reuseInputUnboxingCode(): String = {
    reusableInputUnboxingExprs.values.map(_.code).mkString("\n")
  }

  private def reusePerRecordCode(): String = {
    reusablePerRecordStatements.mkString("\n")
  }

  private def addReusablePatternNames(): Unit = {
    ctx.addReusableMember(s"private String[] $patternNamesTerm = new String[] { ${
        patternNames.map(p => s""""${EncodingUtils.escapeJava(p)}"""").mkString(", ")
      } };")
  }

  /**
    * Generates a [[org.apache.flink.api.common.functions.Function]] that can be passed to Java
    * compiler.
    *
    * @param name Class name of the Function. Must not be unique but has to be a valid Java class
    *             identifier.
    * @param clazz Flink Function to be generated.
    * @param bodyCode code contents of the SAM (Single Abstract Method). Inputs, collector, or
    *                 output record can be accessed via the given term methods.
    * @tparam F Flink Function to be generated.
    * @tparam T Return type of the Flink Function.
    * @return instance of GeneratedFunction
    */
  def generateMatchFunction[F <: Function, T <: Any](
      name: String,
      config: TableConfig,
      clazz: Class[F],
      bodyCode: String)
  : GeneratedClass[_] = {
    val funcName = newName(name)
    val collectorTypeTerm = classOf[Collector[Any]].getCanonicalName
    val inputTypeTerm = boxedTypeTermForType(input1Type)
    val (functionClass, signature, inputStatements, unboxingCodeSplit) =
      if (clazz == classOf[RichIterativeCondition[_]]) {
        val baseClass = classOf[RichIterativeCondition[_]]
        val contextType = classOf[IterativeCondition.Context[_]].getCanonicalName
        val unboxingCodeSplit = generateSplitFunctionCalls(
          ctx.reusableInputUnboxingExprs.values.map(_.code).toSeq,
          config.getConf.getInteger(TableConfigOptions.SQL_CODEGEN_LENGTH_MAX),
          "inputUnbox",
          "private final void",
          ctx.reuseFieldCode().length,
          defineParams = s"$inputTypeTerm $input1Term, " +
            s"${classOf[IterativeCondition.Context[_]].getCanonicalName} $contextTerm",
          callingParams = s"$input1Term, $contextTerm"
        )

        (baseClass,
          s"boolean filter(Object _in1, $contextType $contextTerm)",
          List(s"$inputTypeTerm $input1Term = ($inputTypeTerm) _in1;"),
          unboxingCodeSplit)
      } else if (clazz == classOf[RichPatternSelectFunction[_, _]]) {
        val baseClass = classOf[RichPatternSelectFunction[_, _]]
        val unboxingCodeSplit = generateSplitFunctionCalls(
          ctx.reusableInputUnboxingExprs.values.map(_.code).toSeq,
          config.getConf.getInteger(TableConfigOptions.SQL_CODEGEN_LENGTH_MAX),
          "inputUnbox",
          "private final void",
          ctx.reuseFieldCode().length,
          defineParams = s"java.util.Map<String, java.util.List<$inputTypeTerm>> $input1Term",
          callingParams = input1Term
        )

        (baseClass,
          s"Object select(java.util.Map<String, java.util.List<$inputTypeTerm>> $input1Term)",
          List(),
          unboxingCodeSplit)
      } else if (clazz == classOf[RichPatternFlatSelectFunction[_, _]]) {
        val baseClass = classOf[RichPatternFlatSelectFunction[_, _]]
        val unboxingCodeSplit = generateSplitFunctionCalls(
          ctx.reusableInputUnboxingExprs.values.map(_.code).toSeq,
          config.getConf.getInteger(TableConfigOptions.SQL_CODEGEN_LENGTH_MAX),
          "inputUnbox",
          "private final void",
          ctx.reuseFieldCode().length,
          defineParams = s"java.util.Map<String, java.util.List<$inputTypeTerm>> $input1Term, " +
            s"$collectorTypeTerm $collectorTerm",
          callingParams = s"$input1Term, $collectorTerm"
        )

        (baseClass,
          s"void flatSelect(java.util.Map<String, java.util.List<$inputTypeTerm>> $input1Term, " +
            s"$collectorTypeTerm $collectorTerm)",
          List(),
          unboxingCodeSplit)
      } else if (clazz == classOf[RichPatternTimeoutFunction[_, _]]) {
        val baseClass = classOf[RichPatternTimeoutFunction[_, _]]
        val unboxingCodeSplit = generateSplitFunctionCalls(
          ctx.reusableInputUnboxingExprs.values.map(_.code).toSeq,
          config.getConf.getInteger(TableConfigOptions.SQL_CODEGEN_LENGTH_MAX),
          "inputUnbox",
          "private final void",
          ctx.reuseFieldCode().length,
          defineParams = s"java.util.Map<String, java.util.List<$inputTypeTerm>> $input1Term, " +
            s"long timeoutTimestamp",
          callingParams = s"$input1Term, timeoutTimestamp"
        )

        (baseClass,
          s"Object timeout(java.util.Map<String, java.util.List<Object>> $input1Term, " +
            "long timeoutTimestamp)",
          List(),
          unboxingCodeSplit)
      } else if (clazz == classOf[RichPatternFlatTimeoutFunction[_, _]]) {
        val baseClass = classOf[RichPatternFlatTimeoutFunction[_, _]]
        val unboxingCodeSplit = generateSplitFunctionCalls(
          ctx.reusableInputUnboxingExprs.values.map(_.code).toSeq,
          config.getConf.getInteger(TableConfigOptions.SQL_CODEGEN_LENGTH_MAX),
          "inputUnbox",
          "private final void",
          ctx.reuseFieldCode().length,
          defineParams = s"java.util.Map<String, java.util.List<$inputTypeTerm>> $input1Term, " +
            s"$collectorTypeTerm $collectorTerm",
          callingParams = s"$input1Term, $collectorTerm"
        )

        (baseClass,
          s"void timeout(java.util.Map<String, java.util.List<Object>> $input1Term, " +
            s"long timeoutTimestamp, $collectorTypeTerm $collectorTerm)",
          List(),
          unboxingCodeSplit)
      } else {
        throw new CodeGenException("Unsupported Function.")
      }

    val funcCode = if (unboxingCodeSplit.isSplit) {
      j"""
        public class $funcName extends ${functionClass.getCanonicalName} {
          ${ctx.reuseMemberCode()}
          ${ctx.reuseFieldCode()}

          public $funcName(Object[] references) throws Exception {
            ${ctx.reuseInitCode()}
          }

          @Override
          public void open(${classOf[Configuration].getCanonicalName} parameters) throws Exception {
            ${ctx.reuseOpenCode()}
          }

          @Override
          public $signature throws Exception {
            ${inputStatements.mkString("\n")}
            ${reusePatternLists()}
            ${reuseClassifierLists()}
            ${unboxingCodeSplit.callings.mkString("\n")}
            ${ctx.reusePerRecordCode()}
            $bodyCode
          }

          ${
            unboxingCodeSplit.definitions.zip(unboxingCodeSplit.bodies) map {
              case (define, body) =>
                s"""
                   |$define throws Exception {
                   | ${ctx.reusePerRecordCode()}
                   | $body
                   |}
                   """.stripMargin
            } mkString "\n"
          }

          @Override
          public void close() throws Exception {
            ${ctx.reuseCloseCode()}
          }
        }
      """.stripMargin
    } else {
      j"""
        public class $funcName extends ${functionClass.getCanonicalName} {
          ${ctx.reuseMemberCode()}

          public $funcName(Object[] references) throws Exception {
            ${ctx.reuseInitCode()}
          }

          @Override
          public void open(${classOf[Configuration].getCanonicalName} parameters) throws Exception {
            ${ctx.reuseOpenCode()}
          }

          @Override
          public $signature throws Exception {
            ${inputStatements.mkString("\n")}
            ${reusePatternLists()}
            ${reuseClassifierLists()}
            ${ctx.reusePerRecordCode()}
            ${ctx.reuseFieldCode()}
            ${ctx.reuseInputUnboxingCode()}
            $bodyCode
          }

          @Override
          public void close() throws Exception {
            ${ctx.reuseCloseCode()}
          }
        }
      """.stripMargin
    }

    if (clazz == classOf[RichIterativeCondition[_]]) {
      GeneratedIterativeCondition(funcName, funcCode, ctx.references.toArray)
    } else if (clazz == classOf[RichPatternSelectFunction[_, _]]) {
      GeneratedPatternSelectFunction(funcName, funcCode, ctx.references.toArray)
    } else if (clazz == classOf[RichPatternFlatSelectFunction[_, _]]) {
      GeneratedPatternFlatSelectFunction(funcName, funcCode, ctx.references.toArray)
    } else if (clazz == classOf[RichPatternTimeoutFunction[_, _]]) {
      GeneratedPatternTimeoutFunction(funcName, funcCode, ctx.references.toArray)
    } else if (clazz == classOf[RichPatternFlatTimeoutFunction[_, _]]) {
      GeneratedPatternFlatTimeoutFunction(funcName, funcCode, ctx.references.toArray)
    } else {
      throw new CodeGenException("Unsupported Function.")
    }
  }

  def generateOneRowPerMatchExpression(
      partitionKeys: java.util.List[RexNode],
      measures: java.util.Map[String, RexNode],
      returnSchema: BaseRowSchema): GeneratedExpression = {
    setOneRowPerMatch(oneRowPerMatch = true)

    // For "ONE ROW PER MATCH", the output columns include:
    // 1) the partition columns;
    // 2) the columns defined in the measures clause.
    val resultExprs =
      partitionKeys.asScala.map { case inputRef: RexInputRef =>
        generatePartitionKeyAccess(inputRef)
      } ++ returnSchema.fieldNames.filter(measures.containsKey(_)).map { fieldName =>
        generateExpression(measures.get(fieldName))
      }

    val resultCodeGenerator = new ExprCodeGenerator(ctx, nullableInput, nullCheck)
        .bindInput(input1Type, inputTerm = input1Term)
    val resultExpression = resultCodeGenerator.generateResultExpression(
      resultExprs,
      new RowType(
        returnSchema.fieldTypeInfos,
        returnSchema.fieldNames.toArray),
      classOf[GenericRow])

    aggregatesPerVariable.values.foreach(_.generateAggFunction())

    resultExpression
  }

  def generateAllRowsPerMatchExpression(
      partitionKeys: java.util.List[RexNode],
      orderKeys: RelCollation,
      measures: java.util.Map[String, RexNode],
      returnSchema: BaseRowSchema): GeneratedExpression = {

    val patternNameTerm = newName("patternName")
    val eventNameTerm = newName("event")
    val eventNameListTerm = newName("eventList")
    val listTypeTerm = classOf[java.util.List[_]].getCanonicalName

    setOneRowPerMatch(oneRowPerMatch = false)

    // For "ALL ROWS PER MATCH", the output columns include:
    // 1) the partition columns;
    // 2) the ordering columns;
    // 3) the columns defined in the measures clause;
    // 4) any remaining columns defined of the input.
    val fieldsAccessed = mutable.Set[Int]()
    val resultExprs =
      partitionKeys.asScala.map { case inputRef: RexInputRef =>
        fieldsAccessed += inputRef.getIndex
        generateFieldAccess(ctx, input1Type, eventNameTerm, inputRef.getIndex, nullCheck)
      } ++ orderKeys.getFieldCollations.asScala.map { fieldCollation =>
        fieldsAccessed += fieldCollation.getFieldIndex
        generateFieldAccess(ctx, input1Type, eventNameTerm, fieldCollation.getFieldIndex, nullCheck)
      } ++ (0 until TypeUtils.getArity(input1Type)).filterNot(fieldsAccessed.contains).map { idx =>
        generateFieldAccess(ctx, input1Type, eventNameTerm, idx, nullCheck)
      } ++ returnSchema.fieldNames.filter(measures.containsKey(_)).map { fieldName =>
        generateExpression(measures.get(fieldName))
      }

    val resultCodeGenerator = new ExprCodeGenerator(ctx, nullableInput, nullCheck)
        .bindInput(input1Type, inputTerm = input1Term)
    val resultExpression = resultCodeGenerator.generateResultExpression(
      resultExprs,
      new RowType(
        returnSchema.fieldTypeInfos,
        returnSchema.fieldNames.toArray),
      classOf[GenericRow])

    val resultCode = {
      addReusablePatternNames()

      def fillPatternLists(): String = {
        reusablePatternLists.filterNot(_._2.filled).map { patternList =>
          val patternName = patternList._1._2
          val listName = patternList._2.resultTerm
          if (patternName == ALL_PATTERN_VARIABLE) {
            j"""
               |$listName.add($eventNameTerm);
               |""".stripMargin
          } else {
            val escapedPatternName = EncodingUtils.escapeJava(patternName)

            j"""
               |if ($patternNameTerm.equals("$escapedPatternName")) {
               |  $listName.add($eventNameTerm);
               |}
               |""".stripMargin
          }
        }.mkString("\n")
      }

      def fillClassifierLists(): String = {
        reusableClassiferList.filterNot(_._2.filled).map { classiferList =>
          val listName = classiferList._2.resultTerm
          j"""
             |$listName.add($patternNameTerm);
             |""".stripMargin
        }.mkString("\n")
      }

      j"""
         |for (String $patternNameTerm : $patternNamesTerm) {
         |  $listTypeTerm $eventNameListTerm = ($listTypeTerm) $input1Term.get($patternNameTerm);
         |  if ($eventNameListTerm != null) {
         |    for ($eventTypeTerm $eventNameTerm : $eventNameListTerm) {
         |      ${fillPatternLists()}
         |      ${fillClassifierLists()}
         |      ${reuseInputUnboxingCode()}
         |      ${reusePerRecordCode()}
         |      ${resultExpression.code}
         |      $collectorTerm.collect(${resultExpression.resultTerm});
         |    }
         |  }
         |}
         |""".stripMargin
    }

    aggregatesPerVariable.values.foreach(_.generateAggFunction())

    GeneratedExpression("", "false", resultCode, null)
  }

  def generateCondition(call: RexNode): GeneratedExpression = {
    val exp = call.accept(this)
    aggregatesPerVariable.values.foreach(_.generateAggFunction())
    exp
  }

  override def visitCall(call: RexCall): GeneratedExpression = {
    call.getOperator match {
      case PREV | NEXT =>
        val countLiteral = call.getOperands.get(1).asInstanceOf[RexLiteral]
        val count = checkedDownCast(countLiteral.getValueAs(classOf[JLong]))
        if (count != 0) {
          throw new TableException("Flink does not support physical offsets within partition.")
        } else {
          updateOffsets(first = false, 0)
          val exp = call.getOperands.get(0).accept(this)
          resetOffsets()
          exp
        }

      case FIRST | LAST =>
        val countLiteral = call.getOperands.get(1).asInstanceOf[RexLiteral]
        val offset = checkedDownCast(countLiteral.getValueAs(classOf[JLong]))
        updateOffsets(call.getOperator == FIRST, offset)
        val expr = call.operands.get(0).accept(this)
        resetOffsets()
        expr

      case CLASSIFIER => findClassifierByLogicalPosition()

      case RUNNING =>
        if (oneRowPerMatch) {
          // running is the same as final
          call.getOperands.get(0).accept(this)
        } else {
          setRunning(true)
          val expr = call.operands.get(0).accept(this)
          resetRunning()
          expr
        }

      case FINAL => call.getOperands.get(0).accept(this)

      case _: SqlAggFunction =>
        val variable = call.accept(new AggregationPatternVariableFinder)
          .getOrElse(throw new TableException("No pattern variable specified in aggregate"))

        val matchAgg = aggregatesPerVariable.get((running, variable)) match {
          case Some(agg) => agg
          case None =>
            val agg = new AggBuilder(variable)
            aggregatesPerVariable((running, variable)) = agg
            agg
        }

        matchAgg.generateDeduplicatedAggAccess(call)

      case ProctimeSqlFunction =>
          MatchCodeGenerator.generateProctimeTimestamp()

      case _ => super.visitCall(call)
    }
  }

  /**
    * Extracts partition keys from any element of the match
    *
    * @param partitionKey partition key to be extracted
    * @return generated code for the given key
    */
  private def generatePartitionKeyAccess(partitionKey: RexInputRef): GeneratedExpression = {
    val keyRow = generateKeyRow()
    generateFieldAccess(ctx, keyRow.resultType, keyRow.resultTerm, partitionKey.getIndex, nullCheck)
  }

  private def generateKeyRow(): GeneratedExpression = {
    val exp = ctx.getReusableInputUnboxingExprs(keyRowTerm, 0) match {
      case Some(expr) =>
        expr

      case None =>
        val nullTerm = newName("isNull")

        ctx.addReusableMember(s"$eventTypeTerm $keyRowTerm;")

        val keyCode =
          j"""
             |boolean $nullTerm = true;
             |for (java.util.Map.Entry entry : $input1Term.entrySet()) {
             |  java.util.List value = (java.util.List) entry.getValue();
             |  if (value != null && value.size() > 0) {
             |    $keyRowTerm = ($eventTypeTerm) value.get(0);
             |    $nullTerm = false;
             |    break;
             |  }
             |}
             |""".stripMargin

        val exp = GeneratedExpression(keyRowTerm, nullTerm, keyCode, input1Type)
        ctx.addReusableInputUnboxingExprs(keyRowTerm, 0, exp)
        exp
    }
    exp.copy(code = NO_CODE)
  }

  override def visitPatternFieldRef(fieldRef: RexPatternFieldRef): GeneratedExpression = {
    if (isWithinAggExprState) {
      generateFieldAccess(ctx, input1Type, inputAggRowTerm, fieldRef.getIndex, nullCheck)
    } else {
      if (fieldRef.getAlpha.equals(ALL_PATTERN_VARIABLE) &&
        currentPattern.isDefined && offset == 0 && !first) {
        generateInputAccess(
          ctx, input1Type, input1Term, fieldRef.getIndex, nullableInput, nullCheck)
      } else {
        generatePatternFieldRef(fieldRef)
      }
    }
  }

  private def generateDefinePatternVariableExp(
      patternName: String,
      currentPattern: String)
  : GeneratedPatternList = {
    val Seq(listName, eventNameTerm) = newNames(Seq("patternEvents", "event"))

    ctx.addReusableMember(s"java.util.List $listName;")

    val addCurrent = if (currentPattern == patternName || patternName == ALL_PATTERN_VARIABLE) {
      j"""
         |$listName.add($input1Term);
         |""".stripMargin
    } else {
      ""
    }

    val listCode = if (patternName == ALL_PATTERN_VARIABLE) {
      addReusablePatternNames()
      val patternTerm = newName("pattern")

      j"""
         |$listName = new java.util.ArrayList();
         |for (String $patternTerm : $patternNamesTerm) {
         |  for ($eventTypeTerm $eventNameTerm :
         |  $contextTerm.getEventsForPattern($patternTerm)) {
         |    $listName.add($eventNameTerm);
         |  }
         |}
         |""".stripMargin
    } else {
      val escapedPatternName = EncodingUtils.escapeJava(patternName)
      j"""
         |$listName = new java.util.ArrayList();
         |for ($eventTypeTerm $eventNameTerm :
         |  $contextTerm.getEventsForPattern("$escapedPatternName")) {
         |    $listName.add($eventNameTerm);
         |}
         |""".stripMargin
    }

    val code =
      j"""
         |$listCode
         |$addCurrent
         |""".stripMargin

    GeneratedPatternList(listName, code)
  }

  private def generateMeasurePatternVariableExp(patternName: String): GeneratedPatternList = {
    val Seq(listName, patternTerm) = newNames(Seq("patternEvents", "pattern"))

    ctx.addReusableMember(s"java.util.List $listName;")

    val (code, filled) = if (running) {
      val code =
        j"""
           |$listName = new java.util.ArrayList();
           |""".stripMargin
      (code, false)
    } else if (patternName == ALL_PATTERN_VARIABLE) {
      addReusablePatternNames()

      val code =
        j"""
           |$listName = new java.util.ArrayList();
           |for (String $patternTerm : $patternNamesTerm) {
           |  java.util.List rows = (java.util.List) $input1Term.get($patternTerm);
           |  if (rows != null) {
           |    $listName.addAll(rows);
           |  }
           |}
           |""".stripMargin
      (code, true)
    } else {
      val escapedPatternName = EncodingUtils.escapeJava(patternName)

      val code =
        j"""
           |$listName = (java.util.List) $input1Term.get("$escapedPatternName");
           |if ($listName == null) {
           |  $listName = java.util.Collections.emptyList();
           |}
           |""".stripMargin
      (code, true)
    }

    GeneratedPatternList(listName, code, filled)
  }

  private def findEventByLogicalPosition(
      patternFieldAlpha: String)
  : GeneratedExpression = {
    val rowNameTerm = newName("row")

    val listName = findEventsByPatternName(patternFieldAlpha).resultTerm
    val resultIndex = if (first) {
      j"""$offset"""
    } else {
      j"""$listName.size() - $offset - 1"""
    }

    ctx.addReusableMember(s"$eventTypeTerm $rowNameTerm;")

    val funcCode =
      j"""
         |$rowNameTerm = null;
         |if ($listName.size() > $offset) {
         |  $rowNameTerm = (($eventTypeTerm) $listName.get($resultIndex));
         |}
         |""".stripMargin

    GeneratedExpression(rowNameTerm, "", funcCode, input1Type)
  }

  private def findEventsByPatternName(
      patternFieldAlpha: String): GeneratedPatternList = {
    reusablePatternLists.get((running, patternFieldAlpha)) match {
      case Some(expr) =>
        expr

      case None =>
        val exp = currentPattern match {
          case Some(p) => generateDefinePatternVariableExp(patternFieldAlpha, p)
          case None => generateMeasurePatternVariableExp(patternFieldAlpha)
        }
        reusablePatternLists((running, patternFieldAlpha)) = exp
        exp
    }
  }

  private def generateDefineClassifierVariableExp(currentPattern: String)
      : GeneratedClassifierList = {
    val Seq(listName, eventNameTerm) = newNames(Seq("classifiers", "event"))

    ctx.addReusableMember(s"java.util.List $listName;")

    val addCurrent =
      j"""
         |$listName.add("$currentPattern");
         |""".stripMargin

    val listCode = {
      val patternNamesToVisit = patternNames.take(patternNames.indexOf(currentPattern) + 1)

      for (patternName <- patternNamesToVisit) yield {
        val escapedPatternName = EncodingUtils.escapeJava(patternName)
        j"""
           |for ($eventTypeTerm $eventNameTerm :
           |  $contextTerm.getEventsForPattern("$escapedPatternName")) {
           |    $listName.add("${EncodingUtils.escapeJava(patternName)}");
           |}
           |""".stripMargin
      }
    }.mkString("\n")

    val code =
      j"""
         |$listName = new java.util.ArrayList();
         |$listCode
         |$addCurrent
         |""".stripMargin

    GeneratedClassifierList(listName, code)
  }

  private def generateMeasureClassifierVariableExp(): GeneratedClassifierList = {
    val Seq(listName, patternTerm, eventNameTerm) = newNames(Seq("classifiers", "pattern", "event"))

    ctx.addReusableMember(s"java.util.List $listName;")

    val (code, filled) = if (!oneRowPerMatch && running) {
      val code =
        j"""
           |$listName = new java.util.ArrayList();
           |""".stripMargin
      (code, false)
    } else {
      addReusablePatternNames()

      val code =
        j"""
           |$listName = new java.util.ArrayList();
           |for (String $patternTerm : $patternNamesTerm) {
           |  java.util.List rows = (java.util.List)
           |    $input1Term.get(${EncodingUtils.escapeJava(patternTerm)});
           |  if (rows != null) {
           |    for ($eventTypeTerm $eventNameTerm : rows) {
           |      $listName.add($patternTerm);
           |    }
           |  }
           |}
           |""".stripMargin
      (code, true)
    }

    GeneratedClassifierList(listName, code, filled)
  }

  private def findClassifierByLogicalPosition(): GeneratedExpression = {
    val Seq(nullTerm, classifierTerm) = newNames(Seq("isNull", "classifier"))

    val listName = findClassifiers().resultTerm
    val resultIndex = if (first) {
      j"""$offset"""
    } else {
      j"""$listName.size() - $offset - 1"""
    }

    val resultType = StringType.INSTANCE
    val resultTypeTerm = primitiveTypeTermForType(resultType)
    val resultInternal = genToInternal(ctx, resultType, s"$listName.get($resultIndex)")

    ctx.addReusableMember(s"$resultTypeTerm $classifierTerm;")

    val funcCode =
      j"""
         |boolean $nullTerm = true;
         |$classifierTerm = null;
         |if ($listName.size() > $offset) {
         |  $classifierTerm = ($resultTypeTerm) $resultInternal;
         |  $nullTerm = false;
         |}
         |""".stripMargin

    GeneratedExpression(classifierTerm, nullTerm, funcCode, resultType)
  }

  private def findClassifiers(): GeneratedClassifierList = {
    reusableClassiferList.get(running) match {
      case Some(expr) =>
        expr

      case None =>
        val exp = currentPattern match {
          case Some(p) => generateDefineClassifierVariableExp(p)
          case None => generateMeasureClassifierVariableExp()
        }
        reusableClassiferList(running) = exp
        exp
    }
  }

  private def generatePatternFieldRef(fieldRef: RexPatternFieldRef): GeneratedExpression = {
    val escapedAlpha = EncodingUtils.escapeJava(fieldRef.getAlpha)

    val patternVariableRef = getReusableInputUnboxingExprs(s"$escapedAlpha#$first", offset) match {
      case Some(expr) =>
        expr

      case None =>
        val exp = findEventByLogicalPosition(fieldRef.getAlpha)
        addReusableInputUnboxingExprs(s"$escapedAlpha#$first", offset, exp)
        exp
    }

    generateNullableInputFieldAccess(
      ctx,
      patternVariableRef.resultType,
      patternVariableRef.resultTerm,
      fieldRef.getIndex,
      nullCheck)
  }

  private def addReusablePerRecordStatement(s: String): Unit = {
    if (running) {
      reusablePerRecordStatements.add(s)
    } else {
      ctx.addPerRecordStatement(s)
    }
  }

  private def getReusableInputUnboxingExprs(inputTerm: String, index: Int)
      : Option[GeneratedExpression] =
    if (running) {
      reusableInputUnboxingExprs.get((inputTerm, index))
    } else {
      ctx.getReusableInputUnboxingExprs(inputTerm, index)
    }

  private def addReusableInputUnboxingExprs(
      inputTerm: String, index: Int, expr: GeneratedExpression): Unit =
    if (running) {
      reusableInputUnboxingExprs((inputTerm, index)) = expr
    } else {
      ctx.addReusableInputUnboxingExprs(inputTerm, index, expr)
    }

  class AggBuilder(variable: String) {

    private val aggregates = new mutable.ListBuffer[RexCall]()

    private val variableUID = newName("variable")

    private val calculateAggFuncName = s"calculateAgg_$variableUID"

    def generateDeduplicatedAggAccess(aggCall: RexCall): GeneratedExpression = {
      reusableAggregationExpr.get(s"${aggCall.toString}#$running") match  {
        case Some(expr) =>
          expr

        case None =>
          val exp: GeneratedExpression = generateAggAccess(aggCall)
          aggregates += aggCall
          reusableAggregationExpr(s"${aggCall.toString}#$running") = exp
          addReusablePerRecordStatement(exp.code)
          exp.copy(code = NO_CODE)
      }
    }

    private def generateAggAccess(aggCall: RexCall): GeneratedExpression = {
      val singleAggResultTerm = newName("result")
      val singleAggNullTerm = newName("nullTerm")
      val singleAggResultType = FlinkTypeFactory.toTypeInfo(aggCall.`type`).toInternalType
      val primitiveSingleAggResultTypeTerm = primitiveTypeTermForType(singleAggResultType)
      val boxedSingleAggResultTypeTerm = boxedTypeTermForType(singleAggResultType)

      val allAggRowTerm = s"aggRow_$variableUID"

      val rowsForVariableCode = findEventsByPatternName(variable)
      val codeForAgg =
        j"""
           |$GENERIC_ROW $allAggRowTerm = $calculateAggFuncName(${rowsForVariableCode.resultTerm});
           |""".stripMargin

      addReusablePerRecordStatement(codeForAgg)

      val defaultValue = primitiveDefaultValue(singleAggResultType)
      val codeForSingleAgg = if (nullCheck) {
        j"""
           |boolean $singleAggNullTerm;
           |$primitiveSingleAggResultTypeTerm $singleAggResultTerm;
           |if ($allAggRowTerm.getField(${aggregates.size}) != null) {
           |  $singleAggResultTerm = ($boxedSingleAggResultTypeTerm) $allAggRowTerm
           |    .getField(${aggregates.size});
           |  $singleAggNullTerm = false;
           |} else {
           |  $singleAggNullTerm = true;
           |  $singleAggResultTerm = $defaultValue;
           |}
           |""".stripMargin
      } else {
        j"""
           |$primitiveSingleAggResultTypeTerm $singleAggResultTerm =
           |    ($boxedSingleAggResultTypeTerm) $allAggRowTerm.getField(${aggregates.size});
           |""".stripMargin
      }

      addReusablePerRecordStatement(codeForSingleAgg)

      GeneratedExpression(singleAggResultTerm, singleAggNullTerm, NO_CODE, singleAggResultType)
    }

    def generateAggFunction(): Unit = {
      val matchAgg = extractAggregatesAndExpressions

      val aggCalls = matchAgg.aggregations.map(a => AggregateCall.create(
        a.sqlAggFunction,
        false,
        false,
        a.exprIndices,
        -1,
        a.resultType,
        a.sqlAggFunction.getName))

      val needRetraction = matchAgg.aggregations.map(_ => false).toArray

      val typeFactory = relBuilder.getTypeFactory.asInstanceOf[FlinkTypeFactory]
      val inputRelType = typeFactory.createStructType(
        matchAgg.inputExprs.map(_.getType),
        matchAgg.inputExprs.indices.map(i => s"TMP$i"))

      val aggInfoList = AggregateUtil.transformToStreamAggregateInfoList(
        aggCalls,
        inputRelType,
        needRetraction,
        needInputCount = false,
        isStateBackendDataViews = false,
        needDistinctInfo = false)

      val inputFieldTypes = matchAgg.inputExprs
        .map(expr => FlinkTypeFactory.toInternalType(expr.getType))

      val aggsHandlerCodeGenerator = new AggsHandlerCodeGenerator(
        CodeGeneratorContext(new TableConfig, supportReference = true),
        relBuilder,
        inputFieldTypes,
        needRetract = false,
        needMerge = false,
        nullCheck = true,
        copyInputField = false)
      val generatedAggsHandler = aggsHandlerCodeGenerator.generateAggsHandler(
        s"AggFunction_$variableUID",
        aggInfoList)

      val generatedTerm = ctx.addReusableObject(generatedAggsHandler, "generatedAggHandler")
      val functionTerm = s"aggregator_$variableUID"

      val declareCode = s"private $AGGS_HANDLE_FUNCTION $functionTerm;"
      val initCode = s"$functionTerm = ($AGGS_HANDLE_FUNCTION) " +
        s"$generatedTerm.newInstance($CURRENT_CLASS_LOADER);"
      ctx.addReusableMember(declareCode, initCode)

      val transformFuncName = s"transformRowForAgg_$variableUID"
      val inputTransform: String = generateAggInputExprEvaluation(
        matchAgg.inputExprs,
        transformFuncName)

      generateAggCalculation(functionTerm, transformFuncName, inputTransform)
    }

    private def extractAggregatesAndExpressions: MatchAgg = {
      val inputRows = new mutable.LinkedHashMap[String, (RexNode, Int)]

      val singleAggregates = aggregates.map { aggCall =>
        val callsWithIndices = aggCall.operands.asScala.map(innerCall => {
          inputRows.get(innerCall.toString) match {
            case Some(x) =>
              x

            case None =>
              val callWithIndex = (innerCall, inputRows.size)
              inputRows(innerCall.toString) = callWithIndex
              callWithIndex
          }
        })

        SingleAggCall(
          aggCall.getOperator.asInstanceOf[SqlAggFunction],
          aggCall.`type`,
          callsWithIndices.map(callsWithIndice => Integer.valueOf(callsWithIndice._2)))
      }

      MatchAgg(singleAggregates, inputRows.values.map(_._1).toSeq)
    }

    private def generateAggCalculation(
        functionTerm: String,
        transformFuncName: String,
        inputTransformFunc: String): Unit = {
      val code =
        j"""
           |$inputTransformFunc
           |
           |private $GENERIC_ROW $calculateAggFuncName(java.util.List input)
           |    throws Exception {
           |  $functionTerm.setAccumulators($functionTerm.createAccumulators());
           |  for ($BASE_ROW row : input) {
           |    $functionTerm.accumulate($transformFuncName(row));
           |  }
           |  $GENERIC_ROW result = ($GENERIC_ROW) $functionTerm.getValue();
           |  return result;
           |}
           |""".stripMargin

      ctx.addReusableMember(code)

      ctx.addReusableOpenStatement(
        s"$functionTerm.open(new $EXECUTION_CONTEXT_IMPL(null, getRuntimeContext()));")
      ctx.addReusableCloseStatement(s"$functionTerm.close();")
    }

    private def generateAggInputExprEvaluation(
        inputExprs: Seq[RexNode],
        funcName: String): String = {
      isWithinAggExprState = true
      val resultTerm = newName("result")
      val exprs = inputExprs.zipWithIndex.map {
        case (inputExpr, outputIndex) =>
          val expr = generateExpression(inputExpr)
          s"""
             |  ${expr.code}
             |  if (${expr.nullTerm}) {
             |    $resultTerm.update($outputIndex, null);
             |  } else {
             |    $resultTerm.update($outputIndex, ${expr.resultTerm});
             |  }
         """.stripMargin
      }.mkString("\n")
      isWithinAggExprState = false

      j"""
         |private $GENERIC_ROW $funcName($BASE_ROW $inputAggRowTerm) {
         |  $GENERIC_ROW $resultTerm = new $GENERIC_ROW(${inputExprs.size});
         |  $exprs
         |  return $resultTerm;
         |}
         |""".stripMargin
    }

    private case class SingleAggCall(
        sqlAggFunction: SqlAggFunction,
        resultType: RelDataType,
        exprIndices: Seq[Integer]
    )

    private case class MatchAgg(
        aggregations: Seq[SingleAggCall],
        inputExprs: Seq[RexNode]
    )
  }
}

object MatchCodeGenerator {
  val ALL_PATTERN_VARIABLE = "*"

  val EXECUTION_CONTEXT_IMPL: String = className[ExecutionContextImpl]

  val AGGS_HANDLE_FUNCTION: String = className[AggsHandleFunction]

  val GENERIC_ROW: String = className[GenericRow]

  val BASE_ROW: String = className[BaseRow]

  val CURRENT_CLASS_LOADER = "Thread.currentThread().getContextClassLoader()"

  def generateProctimeTimestamp(): GeneratedExpression = {
    val resultTerm = newName("result")
    val resultCode =
      s"""
         |long $resultTerm = System.currentTimeMillis();
         |""".stripMargin.trim
    GeneratedExpression(resultTerm, NEVER_NULL, resultCode, DataTypes.TIMESTAMP)
  }
}