[jira] [Commented] (FLINK-15579) Support UpsertStreamTableSink on Blink batch mode

["xingoo (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-15579?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17059996#comment-17059996 ] 

xingoo commented on FLINK-15579:
--------------------------------

I'm try the code in this pullrequest: [https://github.com/apache/flink/pull/11045/files/ab87a7aa2192c0129929f5600f3fcdfec9512e46]

and I found if use upsert, must close the primary key validation in org.apache.flink.table.planner.operations.SqlToOperationConverter(line 169)
{code:java}
//代码占位符
    private Operation convertCreateTable(SqlCreateTable sqlCreateTable) {
      // primary key and unique keys are not supported
//    if ((sqlCreateTable.getPrimaryKeyList().size() > 0)
//       || (sqlCreateTable.getUniqueKeysList().size() > 0)) {
//       throw new SqlConversionException("Primary key and unique key are not supported yet.");
//    }
{code}
if not, the tableKeys can not found
{code:java}
//代码占位符
case upsertSink: UpsertStreamTableSink[T] =>
  // check for append only table
  val isAppendOnlyTable = UpdatingPlanChecker.isAppendOnly(this)
  upsertSink.setIsAppendOnly(isAppendOnlyTable)
  val tableKeys = {
    val sinkFieldNames = upsertSink.getTableSchema.getFieldNames
    UpdatingPlanChecker.getUniqueKeyFields(getInput, planner, sinkFieldNames) match {
      case Some(keys) => keys.sortBy(_.length).headOption
      case None => None
    }
  }

{code}
 

> Support UpsertStreamTableSink on Blink batch mode
> -------------------------------------------------
>
>                 Key: FLINK-15579
>                 URL: https://issues.apache.org/jira/browse/FLINK-15579
>             Project: Flink
>          Issue Type: Improvement
>          Components: Table SQL / Planner
>    Affects Versions: 1.9.0, 1.9.1
>            Reporter: Shu Li Zheng
>            Assignee: Shu Li Zheng
>            Priority: Major
>              Labels: pull-request-available
>          Time Spent: 10m
>  Remaining Estimate: 0h
>
> Because JDBCTableSourceSinkFactory.createStreamTableSink() create JDBCUpsertTableSink. But BatchExecSink can not work with UpsertStreamTableSink.
> {code:scala}
>   override protected def translateToPlanInternal(
>       planner: BatchPlanner): Transformation[Any] = {
>     val resultTransformation = sink match {
>       case _: RetractStreamTableSink[T] | _: UpsertStreamTableSink[T] =>
>         throw new TableException("RetractStreamTableSink and UpsertStreamTableSink is not" +
>           " supported in Batch environment.")
> {code}
> DDL like:
> CREATE TABLE USER_RESULT(
> NAME VARCHAR,
> CITY VARCHAR,
> SCORE BIGINT
> ) WITH (
> 'connector.type' = 'jdbc',
> 'connector.url' = '',
> 'connector.table' = '',
> 'connector.driver' = 'com.mysql.cj.jdbc.Driver',
> 'connector.username' = 'root',
> 'connector.password' = '',
> 'connector.write.flush.interval' = '1s')



--
This message was sent by Atlassian Jira
(v8.3.4#803005)