Repository: apache/flink
File: flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/calcite/Sink.scala (Scala)

Code before:
import org.apache.flink.table.catalog.ContextResolvedTable
import org.apache.flink.table.connector.sink.DynamicTableSink
import org.apache.flink.table.planner.plan.utils.RelExplainUtil

import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
import org.apache.calcite.rel.hint.RelHint

import java.util

import scala.collection.JavaConversions._

/**
 * Relational expression that writes out data of input node into a [[DynamicTableSink]].
 *
 * @param cluster
 *   cluster that this relational expression belongs to
 * @param traitSet
 *   the traits of this rel
 * @param input
 *   input relational expression
 * @param hints
 *   the hints
 * @param targetColumns
 *   the specified target columns. @Deprecated(since = "2.2"), use
 *   [[org.apache.flink.table.planner.plan.abilities.sink.TargetColumnWritingSpec]] instead.
 * @param contextResolvedTable
 *   the table definition.
 * @param tableSink
 *   the [[DynamicTableSink]] for which to write into
 */
abstract class Sink(
    cluster: RelOptCluster,
    traitSet: RelTraitSet,
    input: RelNode,
    val hints: util.List[RelHint],
    @deprecated(since = "2.2")
    val targetColumns: Array[Array[Int]],
    val contextResolvedTable: ContextResolvedTable,
    val tableSink: DynamicTableSink)
  extends SingleRel(cluster, traitSet, input) {

  override def deriveRowType(): RelDataType = {
    input.getRowType
  }

  override def explainTerms(pw: RelWriter): RelWriter = {
    super
      .explainTerms(pw)
      .item("table", contextResolvedTable.getIdentifier.asSummaryString())

Review comment (suggestion):
```
the specified target columns. This field has been deprecated as part of FLIP-506 since 2.2. Please use
 *   [[TargetColumnWritingSpec]] in [[abilitySpecs]] instead.
```

Additionally, we should move all `abilitySpecs` from all the `Sink` subclasses up into the `Sink` base class. When a feature on the `Sink` is deprecated, we can provide a direct, compatible replacement for it.

Code after:
import org.apache.flink.table.catalog.ContextResolvedTable
import org.apache.flink.table.connector.sink.DynamicTableSink
import org.apache.flink.table.planner.plan.abilities.sink.SinkAbilitySpec
import org.apache.flink.table.planner.plan.abilities.sink.TargetColumnWritingSpec
import org.apache.flink.table.planner.plan.utils.RelExplainUtil

import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
import org.apache.calcite.rel.hint.RelHint

import java.util

import scala.collection.JavaConversions._

/**
 * Relational expression that writes out data of input node into a [[DynamicTableSink]].
 *
 * @param cluster
 *   cluster that this relational expression belongs to
 * @param traitSet
 *   the traits of this rel
 * @param input
 *   input relational expression
 * @param hints
 *   the hints
 * @param targetColumns
 *   the specified target columns. @Deprecated(since = "2.2"), use [[TargetColumnWritingSpec]]
 *   instead.
 * @param contextResolvedTable
 *   the table definition.
 * @param tableSink
 *   the [[DynamicTableSink]] for which to write into
 * @param abilitySpecs
 *   the [[SinkAbilitySpec]]s of this sink
 */
abstract class Sink(
    cluster: RelOptCluster,
    traitSet: RelTraitSet,
    input: RelNode,
    val hints: util.List[RelHint],
    @deprecated(since = "2.2")
    val targetColumns: Array[Array[Int]],
    val contextResolvedTable: ContextResolvedTable,
    val tableSink: DynamicTableSink,
    val abilitySpecs: Array[SinkAbilitySpec])
  extends SingleRel(cluster, traitSet, input) {

  override def deriveRowType(): RelDataType = {
    input.getRowType
  }