tragicjun commented on a change in pull request #6508: [Flink-10079] [table]
Automatically register sink table from external catalogs
URL: https://github.com/apache/flink/pull/6508#discussion_r210471287
##########
File path:
flink-libraries/flink-table/src/main/scala/org/apache/flink/table/api/TableEnvironment.scala
##########
@@ -750,7 +751,28 @@ abstract class TableEnvironment(val config: TableConfig) {
if (null == sinkTableName) throw TableException("Name of TableSink must
not be null.")
if (sinkTableName.isEmpty) throw TableException("Name of TableSink must
not be empty.")
if (!isRegistered(sinkTableName)) {
- throw TableException(s"No table was registered under the name
$sinkTableName.")
+ // try resolving and registering sink table from registered external
catalogs
+ try {
+ val paths = sinkTableName.split("\\.")
+ if (paths.length > 1) {
Review comment:
@walterddr thanks for reviewing. I assume full-path is required for both
source table and sink table, as currently there is no "**USE** <db>" style of
switching catalog/database supported. If fully qualified name is enforced, the
first element of the path must be `catalog` name, right?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
