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:
[email protected]
With regards,
Apache Git Services
