fhueske 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_r217384955
##########
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) {
+ var externalCatalog = getRegisteredExternalCatalog(paths(0))
+ for (i <- 1 to (paths.length - 2)) {
+ externalCatalog = externalCatalog.getSubCatalog(paths(i))
+ }
+ val externalTable = externalCatalog.getTable(paths(paths.length - 1))
+ if (externalTable.isTableSink) {
+ registerTableSink(sinkTableName,
Review comment:
We should not register the sink table again in the root catalog.
----------------------------------------------------------------
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:
us...@infra.apache.org
With regards,
Apache Git Services
