/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.table.plan.nodes.datastream

import java.lang.{Boolean => JBoolean, Long => JLong}
import java.util

import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
import org.apache.calcite.rel.RelFieldCollation.Direction
import org.apache.calcite.rel._
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rex._
import org.apache.calcite.sql.SqlKind
import org.apache.calcite.sql.SqlMatchRecognize.AfterOption
import org.apache.calcite.sql.`type`.SqlTypeFamily
import org.apache.calcite.sql.fun.SqlStdOperatorTable._
import org.apache.flink.annotation.VisibleForTesting
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils.runtime.RowComparator
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy
import org.apache.flink.cep.nfa.compiler.NFACompiler
import org.apache.flink.cep.pattern.Pattern
import org.apache.flink.cep.pattern.Quantifier.QuantifierProperty
import org.apache.flink.cep.pattern.conditions.BooleanConditions
import org.apache.flink.cep.{CEP, PatternStream}
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.windowing.time.Time

import scala.collection.JavaConverters._
import org.apache.flink.table.api._
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.codegen.MatchCodeGenerator
import org.apache.flink.table.plan.logical.MatchRecognize
import org.apache.flink.table.plan.nodes.CommonMatchRecognize
import org.apache.flink.table.plan.rules.datastream.DataStreamRetractionRules
import org.apache.flink.table.plan.schema.RowSchema
import org.apache.flink.table.plan.util.RexDefaultVisitor
import org.apache.flink.table.runtime.`match`._
import org.apache.flink.table.runtime.aggregate.SortUtil
import org.apache.flink.table.runtime.conversion.CRowToRowMapFunction
import org.apache.flink.table.runtime.types.{CRow, CRowTypeInfo}
import org.apache.flink.table.runtime.{RowKeySelector, RowtimeProcessFunction}
import org.apache.flink.types.Row
import org.apache.flink.util.MathUtils

/**
  * Flink RelNode which matches along with LogicalMatch.
  */
class DataStreamMatch(
    cluster: RelOptCluster,
    traitSet: RelTraitSet,
    inputNode: RelNode,
    logicalMatch: MatchRecognize,
    schema: RowSchema,
    inputSchema: RowSchema)
  extends SingleRel(cluster, traitSet, inputNode)
  with CommonMatchRecognize
  with DataStreamRel {

  override def needsUpdatesAsRetraction = true

  override def consumesRetractions = true

  override def deriveRowType(): RelDataType = schema.relDataType

  override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
    new DataStreamMatch(
      cluster,
      traitSet,
      inputs.get(0),
      logicalMatch,
      schema,
      inputSchema)
  }

  override def toString: String = {
    matchToString(logicalMatch, inputSchema.fieldNames, getExpressionString)
  }

  override def explainTerms(pw: RelWriter): RelWriter = {
    explainMatch(super.explainTerms(pw), logicalMatch, inputSchema.fieldNames, getExpressionString)
  }

  private def translateTimeBound(interval: RexNode): Time = {
    interval match {
      case x: RexLiteral if x.getTypeName.getFamily == SqlTypeFamily.INTERVAL_DAY_TIME =>
        Time.milliseconds(x.getValueAs(classOf[JLong]))
      case _ =>
        throw new TableException("Only constant intervals with millisecond resolution " +
          "are supported as time constraints of patterns.")
    }
  }

  @VisibleForTesting
  private[flink] def translatePattern(
    config: TableConfig,
    inputTypeInfo: TypeInformation[Row]
  ): (Pattern[Row, Row], Iterable[String]) = {
    val patternVisitor = new PatternVisitor(config, inputTypeInfo, logicalMatch)
    val cepPattern = if (logicalMatch.interval != null) {
      val interval = translateTimeBound(logicalMatch.interval)
      logicalMatch.pattern.accept(patternVisitor).within(interval)
    } else {
      logicalMatch.pattern.accept(patternVisitor)
    }
    (cepPattern, patternVisitor.names)
  }

  override def translateToPlan(
      tableEnv: StreamTableEnvironment,
      queryConfig: StreamQueryConfig)
    : DataStream[CRow] = {

    val inputIsAccRetract = DataStreamRetractionRules.isAccRetract(getInput)

    val config = tableEnv.config
    val inputTypeInfo = inputSchema.typeInfo

    val crowInput: DataStream[CRow] = getInput
      .asInstanceOf[DataStreamRel]
      .translateToPlan(tableEnv, queryConfig)

    if (inputIsAccRetract) {
      throw new TableException(
        "Retraction on match recognize is not supported. " +
          "Note: Match recognize should not follow a non-windowed GroupBy aggregation.")
    }

    val (timestampedInput, rowComparator) = translateOrder(tableEnv,
      crowInput,
      logicalMatch.orderKeys)

    val (cepPattern, patternNames) = translatePattern(config, inputTypeInfo)

    //TODO remove this once it is supported in CEP library
    if (NFACompiler.canProduceEmptyMatches(cepPattern)) {
      throw new TableException(
        "Patterns that can produce empty matches are not supported. There must be at least one " +
          "non-optional state.")
    }

    //TODO remove this once it is supported in CEP library
    if (cepPattern.getQuantifier.hasProperty(QuantifierProperty.GREEDY)) {
      throw new TableException(
        "Greedy quantifiers are not allowed as the last element of a Pattern yet. Finish your " +
          "pattern with either a simple variable or reluctant quantifier.")
    }

    val inputDS: DataStream[Row] = timestampedInput
      .map(new CRowToRowMapFunction)
      .setParallelism(timestampedInput.getParallelism)
      .name("ConvertToRow")
      .returns(inputTypeInfo)

    val partitionKeys = logicalMatch.partitionKeys
    val partitionedStream = applyPartitioning(partitionKeys, inputDS)

    val patternStream: PatternStream[Row] = if (rowComparator.isDefined) {
      CEP.pattern[Row](partitionedStream, cepPattern, new EventRowComparator(rowComparator.get))
    } else {
      CEP.pattern[Row](partitionedStream, cepPattern)
    }

    val measures = logicalMatch.measures
    val outTypeInfo = CRowTypeInfo(schema.typeInfo)
    if (logicalMatch.allRows) {
      throw new TableException("All rows per match mode is not supported yet.")
    } else {
      val generator = new MatchCodeGenerator(config, inputTypeInfo, patternNames.toSeq)
      val patternSelectFunction = generator.generateOneRowPerMatchExpression(
          schema,
          partitionKeys,
          measures)
      patternStream.process[CRow](patternSelectFunction, outTypeInfo)
    }
  }

  private def translateOrder(
      tableEnv: StreamTableEnvironment,
      crowInput: DataStream[CRow],
      orderKeys: RelCollation)
    : (DataStream[CRow], Option[RowComparator]) = {

    if (orderKeys.getFieldCollations.size() == 0) {
      throw new ValidationException("You must specify either rowtime or proctime for order by.")
    }

    // need to identify time between others order fields. Time needs to be first sort element
    val timeOrderField = SortUtil.getFirstSortField(orderKeys, inputSchema.relDataType)

    if (!FlinkTypeFactory.isTimeIndicatorType(timeOrderField.getType)) {
      throw new ValidationException(
        "You must specify either rowtime or proctime for order by as the first one.")
    }

    // time ordering needs to be ascending
    if (SortUtil.getFirstSortDirection(orderKeys) != Direction.ASCENDING) {
      throw new ValidationException(
        "Primary sort order of a streaming table must be ascending on time.")
    }

    val rowComparator = if (orderKeys.getFieldCollations.size() > 1) {
      Some(SortUtil
        .createRowComparator(inputSchema.relDataType,
          orderKeys.getFieldCollations.asScala.tail,
          tableEnv.execEnv.getConfig))
    } else {
      None
    }

    timeOrderField.getType match {
      case _ if FlinkTypeFactory.isRowtimeIndicatorType(timeOrderField.getType) =>
        (crowInput.process(
          new RowtimeProcessFunction(timeOrderField.getIndex, CRowTypeInfo(inputSchema.typeInfo))
        ).setParallelism(crowInput.getParallelism),
          rowComparator)
      case _ =>
        (crowInput, rowComparator)
    }
  }

  private def applyPartitioning(partitionKeys: util.List[RexNode], inputDs: DataStream[Row])
    : DataStream[Row] = {
    if (partitionKeys.size() > 0) {
      val keys = partitionKeys.asScala.map {
        case ref: RexInputRef => ref.getIndex
      }.toArray
      val keySelector = new RowKeySelector(keys, inputSchema.projectedTypeInfo(keys))
      inputDs.keyBy(keySelector)
    } else {
      inputDs
    }
  }
}

private class PatternVisitor(
    config: TableConfig,
    inputTypeInfo: TypeInformation[Row],
    logicalMatch: MatchRecognize)
  extends RexDefaultVisitor[Pattern[Row, Row]] {

  private var pattern: Pattern[Row, Row] = _
  val names = new collection.mutable.LinkedHashSet[String]()

  override def visitLiteral(literal: RexLiteral): Pattern[Row, Row] = {
    val patternName = literal.getValueAs(classOf[String])
    pattern = translateSingleVariable(Option.apply(pattern), patternName)

    val patternDefinition = logicalMatch.patternDefinitions.get(patternName)
    if (patternDefinition != null) {
      val generator = new MatchCodeGenerator(config, inputTypeInfo, names.toSeq, Some(patternName))
      val condition = generator.generateIterativeCondition(patternDefinition)

      pattern.where(condition)
    } else {
      pattern.where(BooleanConditions.trueFunction())
    }
  }

  override def visitCall(call: RexCall): Pattern[Row, Row] = {
    call.getOperator match {
      case PATTERN_CONCAT =>
        val left = call.operands.get(0)
        val right = call.operands.get(1)

        pattern = left.accept(this)
        pattern = right.accept(this)
        pattern

      case PATTERN_QUANTIFIER =>
        val name = call.operands.get(0) match {
          case c: RexLiteral => c
          case x => throw new TableException(s"Expression not supported: $x Group patterns are " +
            s"not supported yet.")
        }
        pattern = name.accept(this)
        val startNum = MathUtils.checkedDownCast(call.operands.get(1).asInstanceOf[RexLiteral]
          .getValueAs(classOf[JLong]))
        val endNum = MathUtils.checkedDownCast(call.operands.get(2).asInstanceOf[RexLiteral]
          .getValueAs(classOf[JLong]))
        val isGreedy = !call.operands.get(3).asInstanceOf[RexLiteral]
          .getValueAs(classOf[JBoolean])

        applyQuantifier(pattern, startNum, endNum, isGreedy)

      case PATTERN_ALTER =>
        throw new TableException(
          s"Expression not supported: $call. Currently, CEP doesn't support branching patterns.")

      case PATTERN_PERMUTE =>
        throw new TableException(
          s"Expression not supported: $call. Currently, CEP doesn't support PERMUTE patterns.")

      case PATTERN_EXCLUDE =>
        throw new TableException(
          s"Expression not supported: $call. Currently, CEP doesn't support '{-' '-}' patterns.")
    }
  }

  override def visitNode(rexNode: RexNode): Pattern[Row, Row] = throw new TableException(
    s"Unsupported expression within Pattern: [$rexNode]")

  private def translateSkipStrategy = {
    val getPatternTarget = () => logicalMatch.after.asInstanceOf[RexCall].getOperands.get(0)
      .asInstanceOf[RexLiteral].getValueAs(classOf[String])

    logicalMatch.after.getKind match {
      case SqlKind.LITERAL =>
        logicalMatch.after.asInstanceOf[RexLiteral].getValueAs(classOf[AfterOption]) match {
          case AfterOption.SKIP_PAST_LAST_ROW => AfterMatchSkipStrategy.skipPastLastEvent()
          case AfterOption.SKIP_TO_NEXT_ROW => AfterMatchSkipStrategy.skipToNext()
        }
      case SqlKind.SKIP_TO_FIRST =>
        AfterMatchSkipStrategy.skipToFirst(getPatternTarget()).throwExceptionOnMiss()
      case SqlKind.SKIP_TO_LAST =>
        AfterMatchSkipStrategy.skipToLast(getPatternTarget()).throwExceptionOnMiss()
      case _ => throw new IllegalStateException(s"Corrupted query tree. Unexpected " +
        s"${logicalMatch.after} for after match strategy.")
    }
  }

  private def translateSingleVariable(
      previousPattern: Option[Pattern[Row, Row]],
      patternName: String)
    : Pattern[Row, Row] = {
    if (names.contains(patternName)) {
      throw new TableException("Pattern variables must be unique. That might change in the future.")
    } else {
      names.add(patternName)
    }

    previousPattern match {
      case Some(p) => p.next(patternName)
      case None =>
        Pattern.begin(patternName, translateSkipStrategy)
    }
  }

  private def applyQuantifier(
      pattern: Pattern[Row, Row],
      startNum: Int,
      endNum: Int,
      greedy: Boolean)
    : Pattern[Row, Row] = {
    val isOptional = startNum == 0 && endNum == 1

    val newPattern = if (startNum == 0 && endNum == -1) { // zero or more
      pattern.oneOrMore().optional().consecutive()
    } else if (startNum == 1 && endNum == -1) { // one or more
      pattern.oneOrMore().consecutive()
    } else if (isOptional) { // optional
      pattern.optional()
    } else if (endNum != -1) { // times
      pattern.times(startNum, endNum).consecutive()
    } else { // times or more
      pattern.timesOrMore(startNum).consecutive()
    }

    if (greedy && isOptional) {
      newPattern
    } else if (greedy) {
      newPattern.greedy()
    } else if (isOptional) {
      throw new TableException("Reluctant optional variables are not supported yet.")
    } else {
      newPattern
    }
  }
}