/*
 * 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.plan.RelOptCluster
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rex._
import org.apache.calcite.sql.SqlKind
import org.apache.flink.api.common.functions.Function
import org.apache.flink.streaming.api.transformations.StreamTransformation
import org.apache.flink.table.api.types.{DataTypes, RowType, TypeConverters}
import org.apache.flink.table.api.{TableConfig, TableConfigOptions, TableException}
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.codegen.CodeGenUtils.{boxedTypeTermForType, newNames}
import org.apache.flink.table.codegen.operator.OperatorCodeGenerator
import org.apache.flink.table.dataformat.{BaseRow, BoxedWrapperRow}
import org.apache.flink.table.runtime.OneInputSubstituteStreamOperator
import org.apache.flink.table.typeutils.BaseRowTypeInfo

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

object CalcCodeGenerator {

  private[flink] def generateCalcOperator(
    ctx: CodeGeneratorContext,
    cluster: RelOptCluster,
    inputDataType: RelDataType,
    inputTransform: StreamTransformation[BaseRow],
    outRowType: RelDataType,
    config: TableConfig,
    calcProgram: RexProgram,
    condition: Option[RexNode],
    retainHeader: Boolean = false,
    ruleDescription: String
  ): OneInputSubstituteStreamOperator[BaseRow, BaseRow] = {
    val inputType = TypeConverters.createInternalTypeFromTypeInfo(
      inputTransform.getOutputType).asInstanceOf[RowType]
    // filter out time attributes
    val inputTerm = CodeGeneratorContext.DEFAULT_INPUT1_TERM
    val outputType = FlinkTypeFactory.toInternalRowType(outRowType)
    val (processCode, codeSplit, filterCodeSplit) = generateProcessCode(
      ctx,
      inputType,
      outputType,
      classOf[BoxedWrapperRow],
      outputType.getFieldNames,
      config,
      calcProgram,
      condition,
      retainHeader = retainHeader)

    val genOperatorExpression =
      OperatorCodeGenerator.generateOneInputStreamOperator[BaseRow, BaseRow](
        ctx,
        ruleDescription,
        processCode,
        "",
        inputType,
        config,
        inputTerm = inputTerm,
        splitFunc = codeSplit,
        filterSplitFunc = filterCodeSplit,
        lazyInputUnboxingCode = true)

    new OneInputSubstituteStreamOperator[BaseRow, BaseRow](
      genOperatorExpression.name,
      genOperatorExpression.code,
      references = ctx.references)
  }

  private[flink] def generateFunction[T <: Function](
      inputType: RowType,
      ruleDescription: String,
      returnType: RowType,
      outRowClass: Class[_ <: BaseRow],
      calcProjection: RexProgram,
      calcCondition: Option[RexNode],
      config: TableConfig,
      functionClass: Class[T]): GeneratedFunction[T, BaseRow] = {
    val ctx = CodeGeneratorContext(config)
    val inputTerm = CodeGeneratorContext.DEFAULT_INPUT1_TERM
    val collectorTerm = CodeGeneratorContext.DEFAULT_OPERATOR_COLLECTOR_TERM
    val (processCode, codeSplit, filterCodeSplit) = generateProcessCode(
      ctx,
      inputType,
      returnType,
      outRowClass,
      returnType.getFieldNames,
      config,
      calcProjection,
      calcCondition,
      collectorTerm = collectorTerm,
      outputDirectly = true
    )
    FunctionCodeGenerator.generateFunction(
      ctx,
      ruleDescription,
      functionClass,
      processCode,
      returnType,
      inputType,
      config,
      input1Term = inputTerm,
      collectorTerm = collectorTerm,
      codeSplit = codeSplit,
      filterCodeSplit = filterCodeSplit
    )
  }

  private[flink] def generateProcessCode(
      ctx: CodeGeneratorContext,
      inputType: RowType,
      outRowType: RowType,
      outRowClass: Class[_ <: BaseRow],
      resultFieldNames: Seq[String],
      config: TableConfig,
      calcProgram: RexProgram,
      condition: Option[RexNode],
      inputTerm: String = CodeGeneratorContext.DEFAULT_INPUT1_TERM,
      collectorTerm: String = CodeGeneratorContext.DEFAULT_OPERATOR_COLLECTOR_TERM,
      retainHeader: Boolean = false,
      outputDirectly: Boolean = false
  ): (String, GeneratedSplittableExpression, GeneratedSplittableExpression) = {

    val projection = calcProgram.getProjectList.map(calcProgram.expandLocalRef)
    val exprGenerator = new ExprCodeGenerator(ctx, false, config.getNullCheck)
      .bindInput(inputType, inputTerm = inputTerm)

    val onlyFilter = projection.lengthCompare(inputType.getArity) == 0 &&
      projection.zipWithIndex.forall { case (rexNode, index) =>
        rexNode.isInstanceOf[RexInputRef] && rexNode.asInstanceOf[RexInputRef].getIndex == index
      }
    var splitFunc: GeneratedSplittableExpression =
      GeneratedSplittableExpression.UNSPLIT_EXPRESSION
    var filterSplitFunc: GeneratedSplittableExpression =
      GeneratedSplittableExpression.UNSPLIT_EXPRESSION

    def produceOutputCode(resultTerm: String) = if (outputDirectly) {
      s"$collectorTerm.collect($resultTerm);"
    } else {
      s"${OperatorCodeGenerator.generatorCollect(resultTerm)}"
    }

    def produceProjectionCode = {
      val projectionExprs = projection.map(exprGenerator.generateExpression)
      val projectionExpression = exprGenerator.generateResultExpression(
        projectionExprs,
        outRowType,
        outRowClass)

      val inputTypeTerm = boxedTypeTermForType(inputType)

      splitFunc = CodeGenUtils.generateSplitFunctionCalls(
        projectionExpression.codeBuffer,
        config.getConf.getInteger(TableConfigOptions.SQL_CODEGEN_LENGTH_MAX),
        "calcProjectApply",
        "private final void",
        ctx.reuseFieldCode().length,
        defineParams = s"$inputTypeTerm $inputTerm",
        callingParams = inputTerm)

      val projectionExpressionCode = if (splitFunc.isSplit) {
        s"""
           |${projectionExpression.preceding}
           |${splitFunc.callings.mkString("\n")}
           |${projectionExpression.flowing}
       """.stripMargin
      } else {
        projectionExpression.code
      }
      val header = if (retainHeader) {
        s"${projectionExpression.resultTerm}.setHeader($inputTerm.getHeader());"
      } else {
        ""
      }

      s"""
         |$header
         |$projectionExpressionCode
         |${produceOutputCode(projectionExpression.resultTerm)}
         |""".stripMargin
    }

    def produceFilterCode: GeneratedExpression = {
      condition.get match {
        case call: RexCall if call.getKind == SqlKind.OR && call.operands.length > 15 =>
          val subConditions = call.getOperands.asScala.map(exprGenerator.generateExpression)
          filterSplitFunc = CodeGenUtils.generateSplitFunctionCalls(
            subConditions.map(sf => sf.code + "\n" + s"return ${sf.resultTerm};"),
            1, //split every sub condition to single method
            "filterApply",
            "private final boolean",
            0)
          val Seq(conditionResultTerm, nullTerm) = newNames(Seq("result", "isNull"))
          val callCode =
            s"""
               |boolean $nullTerm = false;
               |boolean $conditionResultTerm =
               (${filterSplitFunc.callings.map{
              c => c.substring(0, c.length - 1)
            }.mkString(" || ")});
             """.stripMargin
          GeneratedExpression(conditionResultTerm, nullTerm, callCode, DataTypes.BOOLEAN)
        case call: RexCall if call.getKind == SqlKind.AND
          && call.operands.forall(_.isA(SqlKind.OR)) =>
          val andPostions = call.operands.map(_.asInstanceOf[RexCall].operands.size())
          val subOps = call.operands.flatMap(_.asInstanceOf[RexCall].operands.asScala)
          if (subOps.length > 15) {
            val subConditions = subOps.map(exprGenerator.generateExpression)
            filterSplitFunc = CodeGenUtils.generateSplitFunctionCalls(
              subConditions.map(sf => sf.code + "\n" + s"return ${sf.resultTerm};"),
              1, //split every sub condition to single method
              "filterApply",
              "private final boolean",
              0)
            val Seq(conditionResultTerm, nullTerm) = newNames(Seq("result", "isNull"))
            var index = 0
            var subCalls = filterSplitFunc.callings.map(c => c.substring(0, c.length - 1))
            val resBuffer = new ListBuffer[Seq[String]]()
            while (subCalls.nonEmpty) {
              val len = andPostions(index)
              val subRes = subCalls.take(len)
              subCalls = subCalls.drop(len)
              index += 1
              resBuffer.add(subRes)
            }

            val callCode =
              s"""
                 |boolean $nullTerm = false;
                 |boolean $conditionResultTerm =
               (${resBuffer.map(_.mkString("(", " || ", ")")).mkString(" && ")});
             """.stripMargin

            GeneratedExpression(conditionResultTerm, nullTerm, callCode, DataTypes.BOOLEAN)
          } else {
            exprGenerator.generateExpression(condition.get)
          }
        case _ => exprGenerator.generateExpression(condition.get)
      }
    }

    val processCode = if (condition.isEmpty && onlyFilter) {
      throw new TableException("This calc has no useful projection and no filter. " +
        "It should be removed by CalcRemoveRule.")
    } else if (condition.isEmpty) { // only projection
    val projectionCode = produceProjectionCode
      s"""
         |${ctx.reuseInputUnboxingCode()}
         |$projectionCode
         |""".stripMargin
    } else {
      val filterCondition = produceFilterCode
      // only filter
      if (onlyFilter) {
        s"""
           |${ctx.reuseInputUnboxingCode()}
           |${filterCondition.code}
           |if (${filterCondition.resultTerm}) {
           |  ${produceOutputCode(inputTerm)}
           |}
           |""".stripMargin
      } else { // both filter and projection
      val filterInputCode = ctx.reuseInputUnboxingCode()
        val filterInputSet = Set(ctx.reusableInputUnboxingExprs.keySet.toSeq: _*)

        // if any filter conditions, projection code will enter an new scope
        val projectionCode = produceProjectionCode

        val projectionInputCode = ctx.reusableInputUnboxingExprs
          .filter((entry) => !filterInputSet.contains(entry._1))
          .values.map(_.code).mkString("\n")
        s"""
           |$filterInputCode
           |${filterCondition.code}
           |if (${filterCondition.resultTerm}) {
           |  $projectionInputCode
           |  $projectionCode
           |}
           |""".stripMargin
      }
    }
    (processCode, splitFunc, filterSplitFunc)
  }
}