[
https://issues.apache.org/jira/browse/FLINK-6442?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16087607#comment-16087607
]
ASF GitHub Bot commented on FLINK-6442:
---------------------------------------
Github user lincoln-lil commented on the issue:
https://github.com/apache/flink/pull/3829
@fhueske, thanks for reviewing this. I read the design doc again, the
original idea to support this functionality was doing minimal changes in the
current state.
Yes, currently a TableSink's schema is always derived from the query
result, and before it is configured the schema is null. Practically, I found
that we can overwrite the getOutputType method to declare a concrete sink
table's schema, and this can be used in a dml sql validation.
So without adding an new SinkTable or modifying TableSink interface, we
can choose keeping writeToSink() method in TableAPI as a usage of derive table
and tell users they can implement a custom type TableSink with output type
declaration as a predefined schema.
I'm not sure if we should add a new method like 'sqlInsert' for dml insert
sql, and uncertain about the return type of an insert Operatoin in
Flink(Calcite's TableModify defines output for DML(INSERT,DELETE,UPDATE) is a
RowCount column of BigInt type). I'll discuss this with @shaoxuan-wang @wuchong
and @sunjincheng121 offline.
After that I'll update this PR according to your comments.
Thanks, Lincoln
> Extend TableAPI Support Sink Table Registration and ‘insert into’ Clause in
> SQL
> -------------------------------------------------------------------------------
>
> Key: FLINK-6442
> URL: https://issues.apache.org/jira/browse/FLINK-6442
> Project: Flink
> Issue Type: New Feature
> Components: Table API & SQL
> Reporter: lincoln.lee
> Assignee: lincoln.lee
> Priority: Minor
>
> Currently in TableAPI there’s only registration method for source table,
> when we use SQL writing a streaming job, we should add additional part for
> the sink, like TableAPI does:
> {code}
> val sqlQuery = "SELECT * FROM MyTable WHERE _1 = 3"
> val t = StreamTestData.getSmall3TupleDataStream(env)
> tEnv.registerDataStream("MyTable", t)
> // one way: invoke tableAPI’s writeToSink method directly
> val result = tEnv.sql(sqlQuery)
> result.writeToSink(new YourStreamSink)
> // another way: convert to datastream first and then invoke addSink
> val result = tEnv.sql(sqlQuery).toDataStream[Row]
> result.addSink(new StreamITCase.StringSink)
> {code}
> From the api we can see the sink table always be a derived table because its
> 'schema' is inferred from the result type of upstream query.
> Compare to traditional RDBMS which support DML syntax, a query with a target
> output could be written like this:
> {code}
> insert into table target_table_name
> [(column_name [ ,...n ])]
> query
> {code}
> The equivalent form of the example above is as follows:
> {code}
> tEnv.registerTableSink("targetTable", new YourSink)
> val sql = "INSERT INTO targetTable SELECT a, b, c FROM sourceTable"
> val result = tEnv.sql(sql)
> {code}
> It is supported by Calcite’s grammar:
> {code}
> insert:( INSERT | UPSERT ) INTO tablePrimary
> [ '(' column [, column ]* ')' ]
> query
> {code}
> I'd like to extend Flink TableAPI to support such feature. see design doc:
> https://goo.gl/n3phK5
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
