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

import org.apache.flink.annotation.{InterfaceStability, VisibleForTesting}
import org.apache.flink.api.common.accumulators.SerializedListAccumulator
import org.apache.flink.api.common.typeutils.TypeSerializer
import org.apache.flink.api.common.{ExecutionMode, JobExecutionResult}
import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.client.JobExecutionException
import org.apache.flink.runtime.schedule.VertexInputTracker.{InputDependencyConstraint, VertexInputTrackerOptions}
import org.apache.flink.streaming.api.datastream.DataStream
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.graph.{StreamGraph, StreamGraphGenerator}
import org.apache.flink.streaming.api.transformations.StreamTransformation
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.types.{DataType, DataTypes, RowType}
import org.apache.flink.table.calcite.{FlinkRelBuilder, FlinkTypeFactory}
import org.apache.flink.table.descriptors.{BatchTableDescriptor, ConnectorDescriptor}
import org.apache.flink.table.errorcode.TableErrors
import org.apache.flink.table.expressions.{Expression, TimeAttribute}
import org.apache.flink.table.plan.`trait`.FlinkRelDistributionTraitDef
import org.apache.flink.table.plan.cost.{FlinkBatchCost, FlinkCostFactory}
import org.apache.flink.table.plan.logical.{LogicalNode, SinkNode}
import org.apache.flink.table.plan.nodes.calcite.LogicalSink
import org.apache.flink.table.plan.nodes.exec.{BatchExecNode, ExecNode}
import org.apache.flink.table.plan.nodes.physical.batch.{BatchExecSink, BatchPhysicalRel}
import org.apache.flink.table.plan.nodes.process.DAGProcessContext
import org.apache.flink.table.plan.optimize.{BatchOptimizeContext, FlinkBatchPrograms}
import org.apache.flink.table.plan.schema._
import org.apache.flink.table.plan.stats.FlinkStatistic
import org.apache.flink.table.plan.subplan.BatchDAGOptimizer
import org.apache.flink.table.plan.util.{DeadlockBreakupProcessor, FlinkNodeOptUtil, FlinkRelOptUtil, SameRelObjectShuttle, SubplanReuseUtil}
import org.apache.flink.table.resource.batch.RunningUnitKeeper
import org.apache.flink.table.runtime.AbstractStreamOperatorWithMetrics
import org.apache.flink.table.sinks._
import org.apache.flink.table.sources.{BatchTableSource, _}
import org.apache.flink.table.temptable.TableServiceException
import org.apache.flink.table.util.PlanUtil._
import org.apache.flink.table.util._
import org.apache.flink.util.{AbstractID, ExceptionUtils, Preconditions}
import org.apache.calcite.plan.{Context, ConventionTraitDef, RelOptPlanner}
import org.apache.calcite.rel.{RelCollationTraitDef, RelNode}
import org.apache.calcite.sql.SqlExplainLevel
import org.apache.calcite.sql2rel.SqlToRelConverter
import org.apache.calcite.sql2rel.SqlToRelConverter.Config
import _root_.java.util.{ArrayList => JArrayList, List => JList, Map => JMap, Set => JSet}

import _root_.scala.collection.JavaConversions._
import _root_.scala.collection.JavaConverters._
import _root_.scala.collection.mutable
import _root_.scala.collection.mutable.ArrayBuffer
import _root_.scala.util.{Failure, Success, Try}

/**
 *  A session to construct between [[Table]] and [[DataStream]], its main function is:
  *
  *  1. Get a table from [[DataStream]], or through registering a [[TableSource]];
  *  2. Transform current already construct table to [[DataStream]];
  *  3. Add [[TableSink]] to the [[Table]].
 * @param config The [[TableConfig]] of this [[BatchTableEnvironment]].
 */
@InterfaceStability.Evolving
class BatchTableEnvironment(
    val streamEnv: StreamExecutionEnvironment,
    config: TableConfig)
    extends TableEnvironment(streamEnv, config) {

  private val ruKeeper = new RunningUnitKeeper(this)

  /** Fetch [[RunningUnitKeeper]] bond with this table env. */
  private[table] def getRUKeeper: RunningUnitKeeper = ruKeeper

  // the builder for Calcite RelNodes, Calcite's representation of a relational expression tree.
  override protected def createRelBuilder: FlinkRelBuilder = FlinkRelBuilder.create(
    frameworkConfig,
    config,
    getTypeFactory,
    Array(
      ConventionTraitDef.INSTANCE,
      FlinkRelDistributionTraitDef.INSTANCE,
      RelCollationTraitDef.INSTANCE),
    catalogManager
  )

  // prefix for unique table names.
  override private[flink] val tableNamePrefix = "_DataStreamTable_"

  /**
   * `expand` is set as false, and each sub-query becomes a [[org.apache.calcite.rex.RexSubQuery]].
   */
  override protected def getSqlToRelConverterConfig: Config =
    SqlToRelConverter.configBuilder()
        .withTrimUnusedFields(false)
        .withConvertTableAccess(false)
        .withExpand(false)
        .withInSubQueryThreshold(Int.MaxValue)
        .build()

  /**
   * Returns specific FlinkCostFactory of this Environment.
   */
  override protected def getFlinkCostFactory: FlinkCostFactory = FlinkBatchCost.FACTORY

  /**
    * Triggers the program execution with specific job name.
    * @param jobName name for the job
    */
  override def execute(jobName: String): JobExecutionResult = {
    tableServiceManager.startTableServiceJob()
    val sinkNodesBak = new mutable.MutableList[SinkNode]
    sinkNodesBak ++= sinkNodes
    Try {
      val streamGraph = generateStreamGraph(jobName)
      val res = executeStreamGraph(streamGraph)
      tableServiceManager.markAllTablesCached()
      res
    } match {
      case Success(value) => value
      case Failure(ex) => ex match {
        case je: JobExecutionException
          if ExceptionUtils.findThrowable(je, classOf[TableServiceException]).isPresent =>
          sinkNodes ++= sinkNodesBak
          tableServiceManager.invalidateCachedTable()
          val streamGraph = generateStreamGraph(jobName)
          executeStreamGraph(streamGraph)
        case _ => throw ex
      }
    }
  }

  private[flink] override def collect[T](
      table: Table,
      sink: CollectTableSink[T],
      jobName: Option[String]): Seq[T] = {
    val outType = sink.getOutputType
    val typeSerializer = DataTypes.createExternalSerializer(outType)
        .asInstanceOf[TypeSerializer[T]]
    val id = new AbstractID().toString
    sink.init(typeSerializer, id)
    writeToSink(table, sink)
    val res = execute(jobName.getOrElse(DEFAULT_JOB_NAME))
    val accResult: JArrayList[Array[Byte]] = res.getAccumulatorResult(id)
    SerializedListAccumulator.deserializeList(accResult, typeSerializer).asScala
  }

  protected override def translateStreamGraph(
      streamingTransformations: ArrayBuffer[StreamTransformation[_]],
      jobName: Option[String]): StreamGraph = {
    mergeParameters()
    val context = StreamGraphGenerator.Context.buildBatchProperties(streamEnv)
    if (getConfig.getConf.getBoolean(TableConfigOptions.SQL_EXEC_DATA_EXCHANGE_MODE_ALL_BATCH)) {
      val constraint = getConfig.getConf.getValue(
        VertexInputTrackerOptions.INPUT_DEPENDENCY_CONSTRAINT)
      // When the user does not set this value,
      // in batch mode, use ALL to avoid deadlock and resource utilization rate.
      if (constraint == null) {
        getConfig.getConf.setString(
          VertexInputTrackerOptions.INPUT_DEPENDENCY_CONSTRAINT,
          InputDependencyConstraint.ALL.toString)
      }
      context.getExecutionConfig.setExecutionMode(ExecutionMode.BATCH)
    } else {
      context.getExecutionConfig.setExecutionMode(ExecutionMode.PIPELINED)
    }
    context.setSlotSharingEnabled(false)

    ruKeeper.setScheduleConfig(context)
    jobName match {
      case Some(jn) => context.setJobName(jn)
      case None => context.setJobName(DEFAULT_JOB_NAME)
    }
    val streamGraph = StreamGraphGenerator.generate(context, streamingTransformations)

    setupOperatorMetricCollect()
    ruKeeper.clear()

    streamingTransformations.clear()
    streamGraph
  }

  private def executeStreamGraph(streamGraph: StreamGraph): JobExecutionResult = {
    val result = streamEnv.execute(streamGraph)
    dumpPlanWithMetricsIfNeed(streamGraph, result)
    result
  }

  /**
    * Set up operator metric collect to be true.
    */
  @VisibleForTesting
  private[flink] def setupOperatorMetricCollect(): Unit = {
    if (streamEnv != null &&
        streamEnv.getConfig != null &&
        config.getConf.getBoolean(TableConfigOptions.SQL_EXEC_OPERATOR_METRIC_DUMP_ENABLED)) {
      val parameters = new Configuration()
      Option(streamEnv.getConfig.getGlobalJobParameters).foreach(gb =>
        gb.toMap.foreach(kv => parameters.setString(kv._1, kv._2))
      )
      parameters.setString(
        AbstractStreamOperatorWithMetrics.METRICS_CONF_KEY,
        AbstractStreamOperatorWithMetrics.METRICS_CONF_VALUE)
      streamEnv.getConfig.setGlobalJobParameters(parameters)
    }
  }

  override private[table] def writeToSink[T](
      table: Table,
      sink: TableSink[T],
      sinkName: String): Unit = {
    val sinkNode = SinkNode(table.logicalPlan, sink, sinkName)
    if (config.getSubsectionOptimization) {
      sinkNodes += sinkNode
    } else {
      val transformation = translate(table, sink, sinkName)
      transformations.add(transformation)
    }
  }

  /**
   * Registers a [[DataStream]] as a table under a given name in the [[TableEnvironment]]'s
   * catalog.
   *
   * @param name     The name under which the table is registered in the catalog.
   * @param boundedStream The [[DataStream]] to register as table in the catalog.
   * @tparam T the type of the [[DataStream]].
   */
  protected def registerBoundedStreamInternal[T](
      name: String, boundedStream: DataStream[T], replace: Boolean): Unit = {
    val (fieldNames, fieldIdxs) = getFieldInfo(boundedStream.getTransformation.getOutputType)
    val boundedStreamTable = new DataStreamTable[T](boundedStream, fieldIdxs, fieldNames)
    registerTableInternal(name, boundedStreamTable, replace)
  }

  /**
   * Registers a [[DataStream]] as a table under a given name in the [[TableEnvironment]]'s
   * catalog.
   *
   * @param name           The name under which the table is registered in the catalog.
   * @param boundedStream       The [[DataStream]] to register as table in the catalog.
   * @param fieldNullables The field isNullables attributes of boundedStream.
   * @tparam T the type of the [[DataStream]].
   */
  protected def registerBoundedStreamInternal[T](
      name: String,
      boundedStream: DataStream[T],
      fieldNullables: Array[Boolean],
      replace: Boolean): Unit = {
    val dataType = boundedStream.getTransformation.getOutputType
    val (fieldNames, fieldIndexes) = getFieldInfo(dataType)
    val fieldTypes = TableEnvironment.getFieldTypes(dataType)
    val relDataType = getTypeFactory.buildRelDataType(fieldNames, fieldTypes, fieldNullables)
    val boundedStreamTable = new IntermediateBoundedStreamTable[T](
      relDataType, boundedStream, fieldIndexes, fieldNames)
    registerTableInternal(name, boundedStreamTable, replace)
  }

  /**
   * Registers a [[DataStream]] as a table under a given name with field names as specified by
   * field expressions in the [[TableEnvironment]]'s catalog.
   *
   * @param name     The name under which the table is registered in the catalog.
   * @param boundedStream The [[DataStream]] to register as table in the catalog.
   * @param fields   The field expressions to define the field names of the table.
   * @tparam T The type of the [[DataStream]].
   */
  protected def registerBoundedStreamInternal[T](
      name: String, boundedStream: DataStream[T],
      fields: Array[Expression], replace: Boolean): Unit = {

    if (fields.exists(_.isInstanceOf[TimeAttribute])) {
      throw new ValidationException(
        ".rowtime and .proctime time indicators are not allowed in a batch exec environment.")
    }

    val dataType = boundedStream.getTransformation.getOutputType
    val (fieldNames, fieldIndexes) = getFieldInfo[T](dataType, fields)
    val boundedStreamTable = new DataStreamTable[T](boundedStream, fieldIndexes, fieldNames)
    registerTableInternal(name, boundedStreamTable, replace)
  }

  /**
   * Registers a [[DataStream]] as a table under a given name with field names as specified by
   * field expressions in the [[TableEnvironment]]'s catalog.
   *
   * @param name           The name under which the table is registered in the catalog.
   * @param boundedStream       The [[DataStream]] to register as table in the catalog.
   * @param fields         The field expressions to define the field names of the table.
   * @param fieldNullables The field isNullables attributes of boundedStream.
   * @tparam T The type of the [[DataStream]].
   */
  protected def registerBoundedStreamInternal[T](
      name: String,
      boundedStream: DataStream[T],
      fields: Array[Expression],
      fieldNullables: Array[Boolean],
      replace: Boolean): Unit = {

    if (fields.exists(_.isInstanceOf[TimeAttribute])) {
      throw new ValidationException(
        ".rowtime and .proctime time indicators are not allowed in a batch exec environment.")
    }
    val dataType = boundedStream.getTransformation.getOutputType
    val (fieldNames, fieldIndexes) = getFieldInfo(dataType, fields)
    val physicalFieldTypes = TableEnvironment.getFieldTypes(dataType)
    val fieldTypes = fieldIndexes.map(physicalFieldTypes.apply)
    val relDataType = getTypeFactory.buildRelDataType(fieldNames, fieldTypes, fieldNullables)
    val boundedStreamTable = new IntermediateBoundedStreamTable[T](
      relDataType, boundedStream, fieldIndexes, fieldNames)
    registerTableInternal(name, boundedStreamTable, replace)
  }

  private def translate[A](
      table: Table,
      sink: TableSink[A],
      sinkName: String,
      dagOptimizeEnabled: Boolean = false): StreamTransformation[_] = {
    val sinkNode = SinkNode(table.logicalPlan, sink, sinkName)
    val nodeDag = optimizeAndTranslateNodeDag(dagOptimizeEnabled, sinkNode)
    nodeDag.head match {
      case batchExecSink: BatchExecSink[A] => translate(batchExecSink, sink.getOutputType)
      case _ => throw new TableException(TableErrors.INST.sqlCompileSinkNodeRequired())
    }
  }

  /**
    * Translates a [[ExecNode]] DAG into a [[StreamTransformation]] DAG.
    *
    * @param sinks The node DAG to translate.
    * @return The [[StreamTransformation]] DAG that corresponds to the node DAG.
    */
  override protected def translate(sinks: Seq[ExecNode[_, _]]): Seq[StreamTransformation[_]] = {
    sinks.map {
      case n: BatchExecSink[_] => translate(n, n.sink.getOutputType)
      case _ => throw new TableException(TableErrors.INST.sqlCompileSinkNodeRequired())
    }
  }

  /**
   * Translates a [[BatchExecNode]] plan into a [[StreamTransformation]].
   * Converts to target type if necessary.
   *
   * @param node        The plan to translate.
   * @param resultType  The [[DataType]] of the elements that result from resulting
   *                    [[StreamTransformation]].
   * @return The [[StreamTransformation]] that corresponds to the given node.
   */
  private def translate[OUT](
      node: ExecNode[_, OUT],
      resultType: DataType): StreamTransformation[OUT] = {
    TableEnvironment.validateType(resultType)

    node match {
      case node: BatchExecNode[OUT] => node.translateToPlan(this)
      case _ =>
        throw new TableException("Cannot generate BoundedStream due to an invalid logical plan. " +
          "This is a bug and should not happen. Please file an issue.")
    }
  }

  /**
   * Generates the optimized [[RelNode]] tree from the original relational node tree.
   *
   * @param relNode The original [[RelNode]] tree
   * @return The optimized [[RelNode]] tree
   */
  private[flink] def optimize(relNode: RelNode): RelNode = {
    val programs = config.getCalciteConfig.getBatchPrograms
      .getOrElse(FlinkBatchPrograms.buildPrograms(config.getConf))
    Preconditions.checkNotNull(programs)

    val optimizedPlan = programs.optimize(relNode, new BatchOptimizeContext {
      override def getContext: Context = getFrameworkConfig.getContext

      override def getRelOptPlanner: RelOptPlanner = getPlanner
    })

    // Rewrite same rel object to different rel objects
    // in order to get the correct dag (dag reuse is based on object not digest)
    optimizedPlan.accept(new SameRelObjectShuttle())
  }

  /**
    * Translates a [[Table]] into a [[DataStream]].
    *
    * The transformation involves optimizing the relational expression tree as defined by
    * Table API calls and / or SQL queries and generating corresponding [[DataStream]] operators.
    *
    * @param table The root node of the relational expression tree.
    * @param resultType The [[DataType]] of the resulting [[DataStream]].
    * @tparam T The type of the resulting [[DataStream]].
    * @return The [[DataStream]] that corresponds to the translated [[Table]].
    */
  protected def translateToDataStream[T](
    table: Table,
    resultType: DataType): DataStream[T] = {
    val sink = new DataStreamTableSink[T](table, resultType, false, false)
    val sinkName = createUniqueTableName()
    val sinkNode = LogicalSink.create(table.getRelNode, sink, sinkName)
    val optimizedPlan = optimize(sinkNode)
    val optimizedNodes = translateNodeDag(Seq(optimizedPlan))
    require(optimizedNodes.size() == 1)
    val transformation = translate(optimizedNodes.head, resultType)
    new DataStream(execEnv, transformation).asInstanceOf[DataStream[T]]
  }

  /**
    * Convert [[BatchPhysicalRel]] DAG to [[BatchExecNode]] DAG and translate them.
    */
  @VisibleForTesting
  private[flink] def translateNodeDag(rels: Seq[RelNode]): Seq[BatchExecNode[_]] = {
    require(rels.nonEmpty && rels.forall(_.isInstanceOf[BatchPhysicalRel]))
    // reuse subplan
    val reusedPlan = SubplanReuseUtil.reuseSubplan(rels, config)
    // convert BatchPhysicalRel DAG to BatchExecNode DAG
    val nodeDag = reusedPlan.map(_.asInstanceOf[BatchExecNode[_]])
    // breakup deadlock
    // TODO move DeadlockBreakupProcessor into batch DAGProcessors
    val nodeDagWithoutDeadlock = new DeadlockBreakupProcessor().process(
      nodeDag, new DAGProcessContext(this))
    // build running units
    nodeDagWithoutDeadlock.foreach(n => ruKeeper.buildRUs(n.asInstanceOf[BatchExecNode[_]]))
    // call processors
    val dagProcessors = getConfig.getBatchDAGProcessors
    require(dagProcessors != null)
    val postNodeDag = dagProcessors.process(
      nodeDagWithoutDeadlock, new DAGProcessContext(this, ruKeeper.getRunningUnitMap))

    dumpOptimizedPlanIfNeed(postNodeDag)
    postNodeDag.map(_.asInstanceOf[BatchExecNode[_]])
  }

  /**
    * Optimize the RelNode tree (or DAG), and translate the result to ExecNode tree (or DAG).
    */
  @VisibleForTesting
  private[flink] override def optimizeAndTranslateNodeDag(
      dagOptimizeEnabled: Boolean,
      logicalNodes: LogicalNode*): Seq[ExecNode[_, _]] = {
    if (logicalNodes.isEmpty) {
      throw new TableException(TableErrors.INST.sqlCompileNoSinkTblError())
    }
    val nodeDag = if (dagOptimizeEnabled) {
      val optLogicalNodes = tableServiceManager.cachePlanBuilder.buildPlanIfNeeded(logicalNodes)
      // optimize dag
      val optRelNodes = BatchDAGOptimizer.optimize(optLogicalNodes, this)
      // translate node dag
      translateNodeDag(optRelNodes)
    } else {
      require(logicalNodes.size == 1)
      val sinkTable = new Table(this, logicalNodes.head)
      val originTree = sinkTable.getRelNode
      // optimize tree
      val optimizedTree = optimize(originTree)
      // translate node tree
      translateNodeDag(Seq(optimizedTree))
    }
    require(nodeDag.size() == logicalNodes.size)
    nodeDag
  }

  /**
    * Registers an internal [[BatchTableSource]] in this [[TableEnvironment]]'s catalog without
    * name checking. Registered tables can be referenced in SQL queries.
    *
    * @param name        The name under which the [[TableSource]] is registered.
    * @param tableSource The [[TableSource]] to register.
    * @param replace     Whether to replace the registered table.
    */
  override protected def registerTableSourceInternal(
    name: String,
    tableSource: TableSource,
    statistic: FlinkStatistic,
    replace: Boolean = false): Unit = {

    tableSource match {

      // check for proper batch table source
      case batchTableSource: BatchTableSource[_] =>
        // check if a table (source or sink) is registered
        getTable(name) match {

          // table source and/or sink is registered
          case Some(table: TableSourceSinkTable[_]) => table.tableSourceTable match {

            // wrapper contains source
            case Some(_: TableSourceTable) if !replace =>
              throw new TableException(s"Table '$name' already exists. " +
                s"Please choose a different name.")

            // wrapper contains only sink (not source)
            case _ =>
              val enrichedTable = new TableSourceSinkTable(
                Some(new BatchTableSourceTable(batchTableSource, statistic)),
                table.tableSinkTable)
              replaceRegisteredTable(name, enrichedTable)
          }

          // no table is registered
          case _ =>
            val newTable = new TableSourceSinkTable(
              Some(new BatchTableSourceTable(batchTableSource, statistic)),
              None)
            registerTableInternal(name, newTable)
        }

      // not a batch table source
      case _ =>
        throw new TableException("Only BatchTableSource can be " +
          "registered in BatchTableEnvironment.")
    }
  }

  /**
    * Creates a table source and/or table sink from a descriptor.
    *
    * Descriptors allow for declaring the communication to external systems in an
    * implementation-agnostic way. The classpath is scanned for suitable table factories that match
    * the desired configuration.
    *
    * The following example shows how to read from a connector using a JSON format and
    * registering a table source as "MyTable":
    *
    * {{{
    *
    * tableEnv
    *   .connect(
    *     new ExternalSystemXYZ()
    *       .version("0.11"))
    *   .withFormat(
    *     new Json()
    *       .jsonSchema("{...}")
    *       .failOnMissingField(false))
    *   .withSchema(
    *     new Schema()
    *       .field("user-name", "VARCHAR").from("u_name")
    *       .field("count", "DECIMAL")
    *   .registerSource("MyTable")
    * }}}
    *
    * @param connectorDescriptor connector descriptor describing the external system
    */
  def connect(connectorDescriptor: ConnectorDescriptor): BatchTableDescriptor = {
    new BatchTableDescriptor(this, connectorDescriptor)
  }

  override def registerTableSinkInternal(
      name: String,
      fieldNames: Array[String],
      fieldTypes: Array[DataType],
      tableSink: TableSink[_],
      replace: Boolean): Unit = {
    checkValidTableName(name)
    if (fieldNames == null) throw new TableException("fieldNames must not be null.")
    if (fieldTypes == null) throw new TableException("fieldTypes must not be null.")
    if (fieldNames.length == 0) throw new TableException("fieldNames must not be empty.")
    if (fieldNames.length != fieldTypes.length) {
      throw new TableException("Same number of field names and types required.")
    }

    // configure and register
    val configuredSink = tableSink.configure(fieldNames, fieldTypes)
    registerTableSinkInternal(name, configuredSink, replace)
  }

  protected def registerTableSinkInternal(
      name: String,
      configuredSink: TableSink[_],
      replace: Boolean): Unit = {
    // validate
    checkValidTableName(name)
    if (configuredSink.getFieldNames == null || configuredSink.getFieldTypes == null) {
      throw new TableException("Table sink is not configured.")
    }
    if (configuredSink.getFieldNames.length == 0) {
      throw new TableException("Field names must not be empty.")
    }
    if (configuredSink.getFieldNames.length != configuredSink.getFieldTypes.length) {
      throw new TableException("Same number of field names and types required.")
    }

    // register
    configuredSink match {

      // check for proper batch table sink
      case _: BatchTableSink[_] | _: BatchCompatibleStreamTableSink[_] =>

        // check if a table (source or sink) is registered
        getTable(name) match {

          // table source and/or sink is registered
          case Some(table: TableSourceSinkTable[_]) => table.tableSinkTable match {

            // wrapper contains sink
            case Some(_: TableSinkTable[_]) if !replace =>
              throw new TableException(s"Table '$name' already exists. " +
                s"Please choose a different name.")

            // wrapper contains only source (not sink)
            case _ =>
              val enrichedTable = new TableSourceSinkTable(
                table.tableSourceTable,
                Some(new TableSinkTable(configuredSink)))
              replaceRegisteredTable(name, enrichedTable)
          }

          // no table is registered
          case _ =>
            val newTable = new TableSourceSinkTable(
              None,
              Some(new TableSinkTable(configuredSink)))
            registerTableInternal(name, newTable)
        }

      // not a batch table sink
      case _ =>
        throw new TableException("Only BatchTableSink can be registered in BatchTableEnvironment.")
    }
  }

  /**
    * Merge global job parameters and table config parameters,
    * and set the merged result to GlobalJobParameters
    */
  private def mergeParameters(): Unit = {
    if (streamEnv != null && streamEnv.getConfig != null) {
      val parameters = new Configuration()
      if (config != null && config.getConf != null) {
        parameters.addAll(config.getConf)
      }

      if (streamEnv.getConfig.getGlobalJobParameters != null) {
        streamEnv.getConfig.getGlobalJobParameters.toMap.asScala.foreach {
          kv => parameters.setString(kv._1, kv._2)
        }
      }

      streamEnv.getConfig.setGlobalJobParameters(parameters)
    }
  }

  /**
   * Returns the AST of the specified Table API and SQL queries and the execution plan to compute
   * the result of the given [[Table]].
   *
   * @param table    The table for which the AST and execution plan will be returned.
   * @param extended Flag to include detailed optimizer estimates.
   */
  private[flink] def explain(table: Table, extended: Boolean): String = {
    val ast = table.getRelNode
    // explain as simple tree, ignore dag optimization if it's enabled
    val optimizedNode = optimizeAndTranslateNodeDag(false, table.logicalPlan).head

    val fieldTypes = ast.getRowType.getFieldList.asScala
      .map(field => FlinkTypeFactory.toInternalType(field.getType))
    val transformation = translate(optimizedNode, new RowType(fieldTypes: _*))
    val streamGraph = translateStreamGraph(ArrayBuffer(transformation), None)

    val executionPlan = PlanUtil.explainPlan(streamGraph)

    val (explainLevel, withResource, withMemCost) = if (extended) {
      (SqlExplainLevel.ALL_ATTRIBUTES, true, true)
    } else {
      (SqlExplainLevel.EXPPLAN_ATTRIBUTES, false, false)
    }

    s"== Abstract Syntax Tree ==" +
      System.lineSeparator +
      s"${FlinkRelOptUtil.toString(ast)}" +
      System.lineSeparator +
      s"== Optimized Logical Plan ==" +
      System.lineSeparator +
      s"${FlinkNodeOptUtil.treeToString(
        optimizedNode, explainLevel, withResource, withMemCost)}" +
      System.lineSeparator +
      s"== Physical Execution Plan ==" +
      System.lineSeparator +
      s"$executionPlan"
  }

  def explain(table: Table): String = explain(table: Table, extended = false)

  def explain(extended: Boolean = false): String = {
    if (!config.getSubsectionOptimization) {
      throw new TableException("Can not explain due to subsection optimization is not supported, " +
        "please check your TableConfig.")
    }

    val sinkExecNodes = optimizeAndTranslateNodeDag(true, sinkNodes: _*)

    val sb = new StringBuilder
    sb.append("== Abstract Syntax Tree ==")
    sb.append(System.lineSeparator)
    sinkNodes.foreach { sink =>
      val table = new Table(this, sink.children.head)
      val ast = table.getRelNode
      sb.append(FlinkRelOptUtil.toString(ast))
      sb.append(System.lineSeparator)
    }

    sb.append("== Optimized Logical Plan ==")
    sb.append(System.lineSeparator)
    val (explainLevel, withResource, withMemCost) = if (extended) {
      (SqlExplainLevel.ALL_ATTRIBUTES, true, true)
    } else {
      (SqlExplainLevel.EXPPLAN_ATTRIBUTES, false, false)
    }
    sb.append(FlinkNodeOptUtil.dagToString(sinkExecNodes, explainLevel, withResource, withMemCost))

    val sinkTransformations = translate(sinkExecNodes)
    val streamGraph = StreamGraphGenerator.generate(
      StreamGraphGenerator.Context.buildBatchProperties(streamEnv), sinkTransformations)
    val sqlPlan = PlanUtil.explainPlan(streamGraph)
    sb.append("== Physical Execution Plan ==")
    sb.append(System.lineSeparator)
    sb.append(sqlPlan)
    sb.toString()
  }

  /**
   * Dump stream graph plan with accumulated operator metrics if config enabled.
   *
   * @param streamGraph streamGraph
   * @param jobResult   job result of stream graph
   */
  private[this] def dumpPlanWithMetricsIfNeed(
      streamGraph: StreamGraph,
      jobResult: JobExecutionResult): Unit = {
    val dumpFilePath = config.getConf.getString(
      TableConfigOptions.SQL_EXEC_OPERATOR_METRIC_DUMP_PATH)
    if (config.getConf.getBoolean(TableConfigOptions.SQL_EXEC_OPERATOR_METRIC_DUMP_ENABLED)
        && dumpFilePath != null) {
      streamGraph.dumpPlanWithMetrics(dumpFilePath, jobResult)
    }
  }

  /**
    * Dump optimized plan if config enabled.
    *
    * @param optimizedNodes optimized plan
    */
  private[this] def dumpOptimizedPlanIfNeed(optimizedNodes: Seq[ExecNode[_, _]]): Unit = {
    val dumpFilePath = config.getConf.getString(TableConfigOptions.SQL_OPTIMIZER_PLAN_DUMP_PATH)
    val planDump = config.getConf.getBoolean(TableConfigOptions.SQL_OPTIMIZER_PLAN_DUMP_ENABLED)

    if (planDump && dumpFilePath != null) {
      dumpExecNodes(optimizedNodes, dumpFilePath)
    }
  }
}