这里说的 PK 是定义在你结果表 DDL 上的PK，最开始的报错信息应该是你结果表上没声明PK吧。
你自定义的 connector 支持 upsert 的话，参考下 HBaseUpsertTableSink 的实现，你的 sink 获取到 Factory
Context 中schema 的 pk 后，需要按照 upsert 语义处理下。
祝好，
Leonard
> On May 14, 2021, at 15:39, automths <autom...@163.com> wrote:
>
> Hi:
>
> 该问题有进一步的进展了。
>
>
> 我把cdc对应的表在创建表时设置了primary key，然后该问题解决了。
>
>
> 现在有一点弄不明白，在查找primary key的时候，是去查找source表的primary key吗？
> 源码位置(org.apache.flink.table.planner.plan.utils.UpdatingPlanChecker#getUniqueKeyForUpsertSink)如下：
>
>
> def getUniqueKeyForUpsertSink(
> sinkNode: LegacySink,
> planner: PlannerBase,
> sink: UpsertStreamTableSink[_]): Option[Array[String]] = {
> // extract unique key fields
> // Now we pick shortest one to sink
> // TODO UpsertStreamTableSink setKeyFields interface should be
> Array[Array[String]]
> val sinkFieldNames = sink.getTableSchema.getFieldNames
> /** Extracts the unique keys of the table produced by the plan. */
> val fmq = FlinkRelMetadataQuery.reuseOrCreate(
> planner.getRelBuilder.getCluster.getMetadataQuery)
> val uniqueKeys = fmq.getUniqueKeys(sinkNode.getInput) // 此处是查找source的primary
> key吗？
> if (uniqueKeys != null && uniqueKeys.size() > 0) {
> uniqueKeys
> .filter(_.nonEmpty)
> .map(_.toArray.map(sinkFieldNames))
> .toSeq
> .sortBy(_.length)
> .headOption
> } else {
> None
> }
> }
>
>
>
>
> flink 版本为1.12.0。
> 在1.13.0上也有出现。
>
>
> 请知道的大佬告知。
>
>
>
>
> 祝好！
> automths
>
>
>
>
> On 05/14/2021 11:00，automths<autom...@163.com> wrote：
> Hi:
>
> 我自定义了一个UpsertStreamTableSink，在接入cdc数据向我的这个Sink写数据的时候，抛出了下面的异常：
> Exception in thread "main" org.apache.flink.table.api.TableException:
> UpsertStreamTableSink requires that Table has a full primary keys if it is
> updated.
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:93)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlanInternal(StreamExecLegacySink.scala:48)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:59)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecLegacySink.translateToPlan(StreamExecLegacySink.scala:48)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:66)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:65)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at scala.collection.Iterator$class.foreach(Iterator.scala:891)
>
>
> 我在构建tableSchema的时候，已经设置了primary key了，但依旧抛出这个错误。
> 我的flink版本是flink-1.12.0的。
>
>
> 请教一下，这个问题，该怎么解决？
>
>
>
>
> 祝好！
> automths
>
>
>
