[GitHub] walterddr commented on a change in pull request #6508: [Flink-10079] [table] Automatically register sink table from external catalogs

[GitBox <gi...@apache.org>]

walterddr 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_r210052280
 
 

 ##########
 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:
   I am not exactly sure but this seems problematic to me. is there a guarantee that the first element of the path will always be `catalog` name? This also depends on whether the path requires to be fully-qualify or not, 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:
users@infra.apache.org


With regards,
Apache Git Services