/*
 * 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.reflect.Modifier
import java.lang.{Iterable => JIterable}
import java.util.{List => JList}

import org.apache.calcite.rex.RexLiteral
import org.apache.flink.api.common.state.{ListStateDescriptor, MapStateDescriptor, State, StateDescriptor}
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils.TypeExtractionUtils.{extractTypeArgument, getRawClass}
import org.apache.flink.table.api.TableConfig
import org.apache.flink.table.api.dataview._
import org.apache.flink.table.codegen.CodeGenUtils.{newName, reflectiveFieldWriteAccess}
import org.apache.flink.table.codegen.Indenter.toISC
import org.apache.flink.table.dataview.{StateListView, StateMapView}
import org.apache.flink.table.functions.AggregateFunction
import org.apache.flink.table.functions.aggfunctions.DistinctAccumulator
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils
import org.apache.flink.table.functions.utils.UserDefinedFunctionUtils.{getUserDefinedMethod, signatureToString}
import org.apache.flink.table.runtime.aggregate.{GeneratedAggregations, SingleElementIterable}
import org.apache.flink.table.utils.EncodingUtils
import org.apache.flink.types.Row

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

/**
  * A code generator for generating [[GeneratedAggregations]].
  *
  * @param config configuration that determines runtime behavior
  * @param nullableInput input(s) can be null.
  * @param input type information about the input of the Function
  * @param constants constant expressions that act like a second input in the parameter indices.
  */
class AggregationCodeGenerator(
    config: TableConfig,
    nullableInput: Boolean,
    input: TypeInformation[_ <: Any],
    constants: Option[Seq[RexLiteral]])
  extends CodeGenerator(config, nullableInput, input) {

  // 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]()

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

  /**
    * Generates a [[org.apache.flink.table.runtime.aggregate.GeneratedAggregations]] that can be
    * passed to a Java compiler.
    *
    * @param name        Class name of the function.
    *                    Does not need to be unique but has to be a valid Java class identifier.
    * @param physicalInputTypes Physical input row types
    * @param aggregates  All aggregate functions
    * @param aggFields   Indexes of the input fields for all aggregate functions
    * @param aggMapping  The mapping of aggregates to output fields
    * @param distinctAccMapping The mapping of the distinct accumulator index to the
    *                           corresponding aggregates.
    * @param isStateBackedDataViews a flag to indicate if distinct filter uses state backend.
    * @param partialResults A flag defining whether final or partial results (accumulators) are set
    *                       to the output row.
    * @param fwdMapping  The mapping of input fields to output fields
    * @param mergeMapping An optional mapping to specify the accumulators to merge. If not set, we
    *                     assume that both rows have the accumulators at the same position.
    * @param outputArity The number of fields in the output row.
    * @param needRetract a flag to indicate if the aggregate needs the retract method
    * @param needMerge a flag to indicate if the aggregate needs the merge method
    * @param needReset a flag to indicate if the aggregate needs the resetAccumulator method
    * @param accConfig Data view specification for accumulators
    *
    * @return A GeneratedAggregationsFunction
    */
  def generateAggregations(
      name: String,
      physicalInputTypes: Seq[TypeInformation[_]],
      aggregates: Array[AggregateFunction[_ <: Any, _ <: Any]],
      aggFields: Array[Array[Int]],
      aggMapping: Array[Int],
      distinctAccMapping: Array[(Integer, JList[Integer])],
      isStateBackedDataViews: Boolean,
      partialResults: Boolean,
      fwdMapping: Array[Int],
      mergeMapping: Option[Array[Int]],
      outputArity: Int,
      needRetract: Boolean,
      needMerge: Boolean,
      needReset: Boolean,
      accConfig: Option[Array[Seq[DataViewSpec[_]]]])
    : GeneratedAggregationsFunction = {

    // get unique function name
    val funcName = newName(name)
    // register UDAGGs
    val aggs = aggregates.map(a => addReusableFunction(a, contextTerm))

    // get java types of accumulators
    val accTypeClasses = aggregates.map { a =>
      a.getClass.getMethod("createAccumulator").getReturnType
    }
    val accTypes = accTypeClasses.map(_.getCanonicalName)

    // create constants
    val constantExprs = constants.map(_.map(generateExpression)).getOrElse(Seq())
    val constantTypes = constantExprs.map(_.resultType)
    val constantFields = constantExprs.map(addReusableBoxedConstant)

    // get parameter lists for aggregation functions
    val parametersCode = aggFields.map { inFields =>
      val fields = inFields.filter(_ > -1).map { f =>
        // index to constant
        if (f >= physicalInputTypes.length) {
          constantFields(f - physicalInputTypes.length)
        }
        // index to input field
        else {
          s"(${CodeGenUtils.boxedTypeTermForTypeInfo(physicalInputTypes(f))}) input.getField($f)"
        }
      }

      fields.mkString(", ")
    }

    // get parameter lists for distinct acc, constant fields are not necessary
    val parametersCodeForDistinctAcc = aggFields.map { inFields =>
      val fields = inFields.filter(i => i > -1 && i < physicalInputTypes.length).map { f =>
        // index to input field
        s"(${CodeGenUtils.boxedTypeTermForTypeInfo(physicalInputTypes(f))}) input.getField($f)"
      }

      fields.mkString(", ")
    }

    val parametersCodeForDistinctMerge = aggFields.map { inFields =>
      // transform inFields to pairs of (inField, index in acc) firstly,
      // e.g. (4, 2, 3, 2) will be transformed to ((4,2), (2,0), (3,1), (2,0))
      val fields = inFields.filter(_ > -1).groupBy(identity).toSeq.sortBy(_._1).zipWithIndex
        .flatMap { case (a, i) => a._2.map((_, i)) }
        .map { case (f, i) =>
          // index to constant
          if (f >= physicalInputTypes.length) {
            constantFields(f - physicalInputTypes.length)
          }
          // index to input field
          else {
            s"(${CodeGenUtils.boxedTypeTermForTypeInfo(physicalInputTypes(f))}) k.getField($i)"
          }
        }

      fields.mkString(", ")
    }

    // get method signatures
    val classes = UserDefinedFunctionUtils.typeInfoToClass(physicalInputTypes)
    val constantClasses = UserDefinedFunctionUtils.typeInfoToClass(constantTypes)
    val methodSignaturesList = aggFields.map { inFields =>
      inFields.filter(_ > -1).map { f =>
        // index to constant
        if (f >= physicalInputTypes.length) {
          constantClasses(f - physicalInputTypes.length)
        }
        // index to input field
        else {
          classes(f)
        }
      }
    }

    // get distinct filter of acc fields for each aggregate functions
    val distinctAccType = s"${classOf[DistinctAccumulator].getName}"

    val distinctAccCount = distinctAccMapping.count(_._1 >= 0)

    if (distinctAccCount > 0 && partialResults && isStateBackedDataViews) {
      // should not happen, but add an error message just in case.
      throw new CodeGenException(
        s"Cannot emit partial results if DISTINCT values are tracked in state-backed maps. " +
          s"Please report this bug."
      )
    }

    // initialize and create data views for accumulators & distinct filters
    addAccumulatorDataViews()

    // check and validate the needed methods
    aggregates.zipWithIndex.map {
      case (a, i) =>
        getUserDefinedMethod(a, "accumulate", Array(accTypeClasses(i)) ++ methodSignaturesList(i))
        .getOrElse(
          throw new CodeGenException(
            s"No matching accumulate method found for AggregateFunction " +
              s"'${a.getClass.getCanonicalName}'" +
              s"with parameters '${signatureToString(methodSignaturesList(i))}'.")
        )

        if (needRetract) {
          getUserDefinedMethod(a, "retract", Array(accTypeClasses(i)) ++ methodSignaturesList(i))
          .getOrElse(
            throw new CodeGenException(
              s"No matching retract method found for AggregateFunction " +
                s"'${a.getClass.getCanonicalName}'" +
                s"with parameters '${signatureToString(methodSignaturesList(i))}'.")
          )
        }

        if (needMerge) {
          val method =
            getUserDefinedMethod(a, "merge", Array(accTypeClasses(i), classOf[JIterable[Any]]))
            .getOrElse(
              throw new CodeGenException(
                s"No matching merge method found for AggregateFunction " +
                  s"${a.getClass.getCanonicalName}'.")
            )

          // use the TypeExtractionUtils here to support nested GenericArrayTypes and
          // other complex types
          val iterableGenericType = extractTypeArgument(method.getGenericParameterTypes()(1), 0)
          val iterableTypeClass = getRawClass(iterableGenericType)

          if (iterableTypeClass != accTypeClasses(i)) {
            throw new CodeGenException(
              s"Merge method in AggregateFunction ${a.getClass.getCanonicalName} does not have " +
                s"the correct Iterable type. Actually: ${iterableTypeClass.toString}. " +
                s"Expected: ${accTypeClasses(i).toString}")
          }
        }

        if (needReset) {
          getUserDefinedMethod(a, "resetAccumulator", Array(accTypeClasses(i)))
          .getOrElse(
            throw new CodeGenException(
              s"No matching resetAccumulator method found for " +
                s"aggregate ${a.getClass.getCanonicalName}'.")
          )
        }
    }

    /**
      * Add all data views for all field accumulators and distinct filters defined by
      * aggregation functions.
      */
    def addAccumulatorDataViews(): Unit = {
      if (accConfig.isDefined) {
        // create state handles for DataView backed accumulator fields.
        val descMapping: Map[String, StateDescriptor[_, _]] = accConfig.get
          .flatMap(specs => specs.map(s => (s.stateId, s.toStateDescriptor)))
          .toMap[String, StateDescriptor[_ <: State, _]]

        for (i <- 0 until aggs.length + distinctAccCount) yield {
          for (spec <- accConfig.get(i)) yield {
            // Check if stat descriptor exists.
            val desc: StateDescriptor[_, _] = descMapping.getOrElse(spec.stateId,
              throw new CodeGenException(
                s"Can not find DataView in accumulator by id: ${spec.stateId}"))

            addReusableDataView(spec, desc, i)
          }
        }
      }
    }

    /**
      * Create DataView Term, for example, acc1_map_dataview.
      *
      * @param aggIndex index of aggregate function
      * @param fieldName field name of DataView
      * @return term to access [[MapView]] or [[ListView]]
      */
    def createDataViewTerm(aggIndex: Int, fieldName: String): String = {
      s"acc${aggIndex}_${fieldName}_dataview"
    }

    /**
      * Adds a reusable [[org.apache.flink.table.api.dataview.DataView]] to the open, cleanup,
      * close and member area of the generated function.
      * @param spec the [[DataViewSpec]] of the desired data view term.
      * @param desc the [[StateDescriptor]] of the desired data view term.
      * @param aggIndex the aggregation function index associate with the data view.
      */
    def addReusableDataView(
        spec: DataViewSpec[_],
        desc: StateDescriptor[_, _],
        aggIndex: Int): Unit = {
      val dataViewField = spec.field
      val dataViewTypeTerm = dataViewField.getType.getCanonicalName

      // define the DataView variables
      val serializedData = EncodingUtils.encodeObjectToString(desc)
      val dataViewFieldTerm = createDataViewTerm(aggIndex, dataViewField.getName)
      val field =
        s"""
           |    final $dataViewTypeTerm $dataViewFieldTerm;
           |""".stripMargin
      reusableMemberStatements.add(field)

      // create DataViews
      val descFieldTerm = s"${dataViewFieldTerm}_desc"
      val descClassQualifier = classOf[StateDescriptor[_, _]].getCanonicalName
      val descDeserializeCode =
        s"""
           |    $descClassQualifier $descFieldTerm = ($descClassQualifier)
           |      ${classOf[EncodingUtils].getCanonicalName}.decodeStringToObject(
           |        "$serializedData",
           |        $descClassQualifier.class,
           |        $contextTerm.getUserCodeClassLoader());
           |""".stripMargin
      val createDataView = if (dataViewField.getType == classOf[MapView[_, _]]) {
        s"""
           |    $descDeserializeCode
           |    $dataViewFieldTerm = new ${classOf[StateMapView[_, _]].getCanonicalName}(
           |      $contextTerm.getMapState(
           |        (${classOf[MapStateDescriptor[_, _]].getCanonicalName}) $descFieldTerm));
           |""".stripMargin
      } else if (dataViewField.getType == classOf[ListView[_]]) {
        s"""
           |    $descDeserializeCode
           |    $dataViewFieldTerm = new ${classOf[StateListView[_]].getCanonicalName}(
           |      $contextTerm.getListState(
           |        (${classOf[ListStateDescriptor[_]].getCanonicalName}) $descFieldTerm));
           |""".stripMargin
      } else {
        throw new CodeGenException(s"Unsupported dataview type: $dataViewTypeTerm")
      }
      reusableOpenStatements.add(createDataView)

      // cleanup DataViews
      val cleanup =
        s"""
           |    $dataViewFieldTerm.clear();
           |""".stripMargin
      reusableCleanupStatements.add(cleanup)
    }

    def genAccDataViewFieldSetter(str: String, i: Int): String = {
      if (accConfig.isDefined) {
        genDataViewFieldSetter(accConfig.get(i), str, i)
      } else {
        ""
      }
    }

    /**
      * Generate statements to set data view field when use state backend.
      *
      * @param specs aggregation [[DataViewSpec]]s for this aggregation term.
      * @param accTerm aggregation term
      * @param aggIndex index of aggregation
      * @return data view field set statements
      */
    def genDataViewFieldSetter(
        specs: Seq[DataViewSpec[_]],
        accTerm: String,
        aggIndex: Int): String = {
      val setters = for (spec <- specs) yield {
        val field = spec.field
        val dataViewTerm = createDataViewTerm(aggIndex, field.getName)
        val fieldSetter = if (Modifier.isPublic(field.getModifiers)) {
          s"$accTerm.${field.getName} = $dataViewTerm;"
        } else {
          val fieldTerm = addReusablePrivateFieldAccess(field.getDeclaringClass, field.getName)
          s"${reflectiveFieldWriteAccess(fieldTerm, field, accTerm, dataViewTerm)};"
        }

        s"""
           |    $fieldSetter
          """.stripMargin
      }
      setters.mkString("\n")
    }

    def genSetAggregationResults: String = {

      val sig: String =
        j"""
           |  public final void setAggregationResults(
           |    org.apache.flink.types.Row accs,
           |    org.apache.flink.types.Row output) throws Exception """.stripMargin

      val setAggs: String = {
        for ((i, aggIndexes) <- distinctAccMapping) yield {
          if (partialResults) {
            def setAggs(aggIndexes: JList[Integer]) = {
              for (i <- aggIndexes) yield {
                j"""
                   |output.setField(
                   |  ${aggMapping(i)},
                   |  (${accTypes(i)}) accs.getField($i));
                 """.stripMargin
              }
            }.mkString("\n")

            if (i >= 0) {
              j"""
                 |    output.setField(
                 |      ${aggMapping(i)},
                 |      ($distinctAccType) accs.getField($i));
                 |    ${setAggs(aggIndexes)}
                 """.stripMargin
            } else {
              j"""
                 |    ${setAggs(aggIndexes)}
                 """.stripMargin
            }
          } else {
            def setAggs(aggIndexes: JList[Integer]) = {
              for (i <- aggIndexes) yield {
                val setAccOutput =
                  j"""
                     |${genAccDataViewFieldSetter(s"acc$i", i)}
                     |output.setField(
                     |  ${aggMapping(i)},
                     |  baseClass$i.getValue(acc$i));
                     """.stripMargin

                j"""
                   |org.apache.flink.table.functions.AggregateFunction baseClass$i =
                   |  (org.apache.flink.table.functions.AggregateFunction) ${aggs(i)};
                   |${accTypes(i)} acc$i = (${accTypes(i)}) accs.getField($i);
                   |$setAccOutput
                   """.stripMargin
              }
            }.mkString("\n")

            j"""
               |    ${setAggs(aggIndexes)}
               """.stripMargin
          }
        }
      }.mkString("\n")

      j"""
         |$sig {
         |$setAggs
         |  }""".stripMargin
    }

    def genAccumulate: String = {

      val sig: String =
        j"""
           |  public final void accumulate(
           |    org.apache.flink.types.Row accs,
           |    org.apache.flink.types.Row input) throws Exception """.stripMargin

      val accumulate: String = {
        def accumulateAcc(aggIndexes: JList[Integer]) = {
          for (i <- aggIndexes) yield {
            j"""
               |${accTypes(i)} acc$i = (${accTypes(i)}) accs.getField($i);
               |${genAccDataViewFieldSetter(s"acc$i", i)}
               |${aggs(i)}.accumulate(acc$i
               |  ${if (!parametersCode(i).isEmpty) "," else ""} ${parametersCode(i)});
               """.stripMargin
          }
        }.mkString("\n")

        for ((i, aggIndexes) <- distinctAccMapping) yield {
          if (i >= 0) {
            j"""
               |    $distinctAccType distinctAcc$i = ($distinctAccType) accs.getField($i);
               |    ${genAccDataViewFieldSetter(s"distinctAcc$i", i)}
               |    if (distinctAcc$i.add(${classOf[Row].getCanonicalName}.of(
               |        ${parametersCodeForDistinctAcc(aggIndexes.get(0))}))) {
               |      ${accumulateAcc(aggIndexes)}
               |    }
               """.stripMargin
          } else {
            j"""
               |    ${accumulateAcc(aggIndexes)}
               """.stripMargin
          }
        }
      }.mkString("\n")

      j"""$sig {
         |$accumulate
         |  }""".stripMargin
    }

    def genRetract: String = {

      val sig: String =
        j"""
           |  public final void retract(
           |    org.apache.flink.types.Row accs,
           |    org.apache.flink.types.Row input) throws Exception """.stripMargin

      val retract: String = {
        def retractAcc(aggIndexes: JList[Integer]) = {
          for (i <- aggIndexes) yield {
            j"""
               |${accTypes(i)} acc$i = (${accTypes(i)}) accs.getField($i);
               |${genAccDataViewFieldSetter(s"acc$i", i)}
               |${aggs(i)}.retract(acc$i
               |  ${if (!parametersCode(i).isEmpty) "," else ""} ${parametersCode(i)});
               """.stripMargin
          }
        }.mkString("\n")

        for ((i, aggIndexes) <- distinctAccMapping) yield {
          if (i >= 0) {
            j"""
               |    $distinctAccType distinctAcc$i = ($distinctAccType) accs.getField($i);
               |    ${genAccDataViewFieldSetter(s"distinctAcc$i", i)}
               |    if (distinctAcc$i.remove(${classOf[Row].getCanonicalName}.of(
               |        ${parametersCodeForDistinctAcc(aggIndexes.get(0))}))) {
               |      ${retractAcc(aggIndexes)}
               |    }
               """.stripMargin
          } else {
            j"""
               |    ${retractAcc(aggIndexes)}
               """.stripMargin
          }
        }
      }.mkString("\n")

      if (needRetract) {
        j"""
           |$sig {
           |$retract
           |  }""".stripMargin
      } else {
        j"""
           |$sig {
           |  }""".stripMargin
      }
    }

    def genCreateAccumulators: String = {

      val sig: String =
        j"""
           |  public final org.apache.flink.types.Row createAccumulators() throws Exception
           |    """.stripMargin
      val init: String =
        j"""
           |      org.apache.flink.types.Row accs =
           |          new org.apache.flink.types.Row(${aggs.length + distinctAccCount});"""
        .stripMargin
      val create: String = {
        def createAcc(aggIndexes: JList[Integer]) = {
          for (i <- aggIndexes) yield {
            j"""
               |${accTypes(i)} acc$i = (${accTypes(i)}) ${aggs(i)}.createAccumulator();
               |accs.setField(
               |  $i,
               |  acc$i);
               """.stripMargin
          }
        }.mkString("\n")

        for ((i, aggIndexes) <- distinctAccMapping) yield {
          if (i >= 0) {
            j"""
               |    $distinctAccType distinctAcc$i = ($distinctAccType)
               |      new ${classOf[DistinctAccumulator].getCanonicalName}();
               |    accs.setField(
               |      $i,
               |      distinctAcc$i);
               |    ${createAcc(aggIndexes)}
               """.stripMargin
          } else {
            j"""
               |    ${createAcc(aggIndexes)}
               """.stripMargin
          }
        }
      }.mkString("\n")
      val ret: String =
        j"""
           |      return accs;"""
        .stripMargin

      j"""$sig {
         |$init
         |$create
         |$ret
         |  }""".stripMargin
    }

    def genSetForwardedFields: String = {

      val sig: String =
        j"""
           |  public final void setForwardedFields(
           |    org.apache.flink.types.Row input,
           |    org.apache.flink.types.Row output)
           |    """.stripMargin

      val forward: String = {
        for (i <- fwdMapping.indices if fwdMapping(i) >= 0) yield
          {
            j"""
               |    output.setField(
               |      $i,
               |      input.getField(${fwdMapping(i)}));"""
            .stripMargin
          }
      }.mkString("\n")

      j"""$sig {
         |$forward
         |  }""".stripMargin
    }

    def genCreateOutputRow: String = {
      j"""
         |  public final org.apache.flink.types.Row createOutputRow() {
         |    return new org.apache.flink.types.Row($outputArity);
         |  }""".stripMargin
    }

    def genMergeAccumulatorsPair: String = {
      val mapping = mergeMapping.getOrElse((0 until aggs.length + distinctAccCount).toArray)

      val sig: String =
        j"""
           |  public final org.apache.flink.types.Row mergeAccumulatorsPair(
           |    org.apache.flink.types.Row a,
           |    org.apache.flink.types.Row b)
           """.stripMargin
      val merge: String = {
        def accumulateAcc(aggIndexes: JList[Integer]) = {
          for (i <- aggIndexes) yield {
            j"""
               |${accTypes(i)} aAcc$i = (${accTypes(i)}) a.getField($i);
               |${aggs(i)}.accumulate(aAcc$i, ${parametersCodeForDistinctMerge(i)});
               |a.setField($i, aAcc$i);
               """.stripMargin
          }
        }.mkString("\n")

        def mergeAcc(aggIndexes: JList[Integer]) = {
          for (i <- aggIndexes) yield {
            j"""
               |${accTypes(i)} aAcc$i = (${accTypes(i)}) a.getField($i);
               |${accTypes(i)} bAcc$i = (${accTypes(i)}) b.getField(${mapping(i)});
               |accIt$i.setElement(bAcc$i);
               |${aggs(i)}.merge(aAcc$i, accIt$i);
               |a.setField($i, aAcc$i);
               """.stripMargin
          }
        }.mkString("\n")

        for ((i, aggIndexes) <- distinctAccMapping) yield {
          if (i >= 0) {
            j"""
               |    $distinctAccType aDistinctAcc$i = ($distinctAccType) a.getField($i);
               |    $distinctAccType bDistinctAcc$i = ($distinctAccType) b.getField(${mapping(i)});
               |    java.util.Iterator<java.util.Map.Entry> mergeIt$i =
               |        bDistinctAcc$i.elements().iterator();
               |
               |    while (mergeIt$i.hasNext()) {
               |      java.util.Map.Entry entry = (java.util.Map.Entry) mergeIt$i.next();
               |      ${classOf[Row].getCanonicalName} k =
               |          (${classOf[Row].getCanonicalName}) entry.getKey();
               |      Long v = (Long) entry.getValue();
               |      if (aDistinctAcc$i.add(k, v)) {
               |        ${accumulateAcc(aggIndexes)}
               |      }
               |    }
               |    a.setField($i, aDistinctAcc$i);
               """.stripMargin
          } else {
            j"""
               |    ${mergeAcc(aggIndexes)}
               """.stripMargin
          }
        }
      }.mkString("\n")
      val ret: String =
        j"""
           |      return a;
           """.stripMargin

      if (needMerge) {
        if (accConfig.isDefined) {
          throw new CodeGenException("DataView doesn't support merge when the backend uses " +
            s"state when generate aggregation for $funcName.")
        }
        j"""
           |$sig {
           |$merge
           |$ret
           |  }""".stripMargin
      } else {
        j"""
           |$sig {
           |$ret
           |  }""".stripMargin
      }
    }

    def genMergeList: String = {
      {
        val singleIterableClass = classOf[SingleElementIterable[_]].getCanonicalName
        for (i <- accTypes.indices) yield
          j"""
             |    private final $singleIterableClass<${accTypes(i)}> accIt$i =
             |      new $singleIterableClass<${accTypes(i)}>();
             """.stripMargin
      }.mkString("\n")
    }

    def genResetAccumulator: String = {

      val sig: String =
        j"""
           |  public final void resetAccumulator(
           |    org.apache.flink.types.Row accs) throws Exception """.stripMargin

      val reset: String = {
        def resetAcc(aggIndexes: JList[Integer]) = {
          for (i <- aggIndexes) yield {
            j"""
               |${accTypes(i)} acc$i = (${accTypes(i)}) accs.getField($i);
               |${genAccDataViewFieldSetter(s"acc$i", i)}
               |${aggs(i)}.resetAccumulator(acc$i);
               """.stripMargin
          }
        }.mkString("\n")

        for ((i, aggIndexes) <- distinctAccMapping) yield {
          if (i >= 0) {
            j"""
               |    $distinctAccType distinctAcc$i = ($distinctAccType) accs.getField($i);
               |    ${genAccDataViewFieldSetter(s"distinctAcc$i", i)}
               |    distinctAcc$i.reset();
               |    ${resetAcc(aggIndexes)}
               """.stripMargin
          } else {
            j"""
               |    ${resetAcc(aggIndexes)}
               """.stripMargin
          }
        }
      }.mkString("\n")

      if (needReset) {
        j"""$sig {
           |$reset
           |  }""".stripMargin
      } else {
        j"""$sig {
           |  }""".stripMargin
      }
    }

    val aggFuncCode = Seq(
      genSetAggregationResults,
      genAccumulate,
      genRetract,
      genCreateAccumulators,
      genSetForwardedFields,
      genCreateOutputRow,
      genMergeAccumulatorsPair,
      genResetAccumulator).mkString("\n")

    val generatedAggregationsClass = classOf[GeneratedAggregations].getCanonicalName
    val funcCode =
      j"""
         |public final class $funcName extends $generatedAggregationsClass {
         |
         |  ${reuseMemberCode()}
         |  $genMergeList
         |  public $funcName() throws Exception {
         |    ${reuseInitCode()}
         |  }
         |  ${reuseConstructorCode(funcName)}
         |
         |  public final void open(
         |    org.apache.flink.api.common.functions.RuntimeContext $contextTerm) throws Exception {
         |    ${reuseOpenCode()}
         |  }
         |
         |  $aggFuncCode
         |
         |  public final void cleanup() throws Exception {
         |    ${reuseCleanupCode()}
         |  }
         |
         |  public final void close() throws Exception {
         |    ${reuseCloseCode()}
         |  }
         |}
         """.stripMargin

    GeneratedAggregationsFunction(funcName, funcCode)
  }
}