KurtYoung commented on a change in pull request #12076: URL: https://github.com/apache/flink/pull/12076#discussion_r425547315
########## File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/utils/TableEnvUtils.java ########## @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.utils; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.internal.TableEnvironmentInternal; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogManager; +import org.apache.flink.table.catalog.ConnectorCatalogTable; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.catalog.UnresolvedIdentifier; +import org.apache.flink.table.sinks.TableSink; +import org.apache.flink.table.sources.TableSource; + +import java.util.Optional; + +/** + * Utility methods to register table source/sink as a Table in {@link TableEnvironment}'s catalog. + */ +@Internal +public class TableEnvUtils { + /** + * Register a table source as a Table under the given name in this + * {@link TableEnvironment}'s catalog. + */ + public static void registerTableSource( + TableEnvironment tEnv, + String name, + TableSource<?> tableSource, + boolean isBatch) { + if (tEnv instanceof TableEnvironmentInternal) { Review comment: check and throw exception first, and do other normal logic without this big brace ########## File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/utils/TableEnvUtils.java ########## @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.utils; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.internal.TableEnvironmentInternal; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogManager; +import org.apache.flink.table.catalog.ConnectorCatalogTable; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.catalog.UnresolvedIdentifier; +import org.apache.flink.table.sinks.TableSink; +import org.apache.flink.table.sources.TableSource; + +import java.util.Optional; + +/** + * Utility methods to register table source/sink as a Table in {@link TableEnvironment}'s catalog. + */ +@Internal +public class TableEnvUtils { + /** + * Register a table source as a Table under the given name in this + * {@link TableEnvironment}'s catalog. + */ + public static void registerTableSource( Review comment: registerTableSourceInternal ########## File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/utils/TableEnvUtils.java ########## @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.utils; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.internal.TableEnvironmentInternal; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogManager; +import org.apache.flink.table.catalog.ConnectorCatalogTable; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.catalog.UnresolvedIdentifier; +import org.apache.flink.table.sinks.TableSink; +import org.apache.flink.table.sources.TableSource; + +import java.util.Optional; + +/** + * Utility methods to register table source/sink as a Table in {@link TableEnvironment}'s catalog. + */ +@Internal +public class TableEnvUtils { Review comment: How about adding these two methods to `TableEnvironmentInternal`? ########## File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/utils/TableEnvUtils.java ########## @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.utils; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.internal.TableEnvironmentInternal; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogManager; +import org.apache.flink.table.catalog.ConnectorCatalogTable; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.catalog.UnresolvedIdentifier; +import org.apache.flink.table.sinks.TableSink; +import org.apache.flink.table.sources.TableSource; + +import java.util.Optional; + +/** + * Utility methods to register table source/sink as a Table in {@link TableEnvironment}'s catalog. + */ +@Internal +public class TableEnvUtils { + /** + * Register a table source as a Table under the given name in this + * {@link TableEnvironment}'s catalog. + */ + public static void registerTableSource( + TableEnvironment tEnv, + String name, + TableSource<?> tableSource, + boolean isBatch) { + if (tEnv instanceof TableEnvironmentInternal) { + CatalogManager catalogManager = ((TableEnvironmentInternal) tEnv).getCatalogManager(); Review comment: You can keep all the logic of original `registerTableSourceInternal`, e.g. it will call `validateTableSource` when register ########## File path: flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/utils/TableEnvUtils.java ########## @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.table.utils; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.api.ValidationException; +import org.apache.flink.table.api.internal.TableEnvironmentInternal; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogManager; +import org.apache.flink.table.catalog.ConnectorCatalogTable; +import org.apache.flink.table.catalog.ObjectIdentifier; +import org.apache.flink.table.catalog.UnresolvedIdentifier; +import org.apache.flink.table.sinks.TableSink; +import org.apache.flink.table.sources.TableSource; + +import java.util.Optional; + +/** + * Utility methods to register table source/sink as a Table in {@link TableEnvironment}'s catalog. + */ +@Internal +public class TableEnvUtils { + /** + * Register a table source as a Table under the given name in this + * {@link TableEnvironment}'s catalog. + */ + public static void registerTableSource( + TableEnvironment tEnv, + String name, + TableSource<?> tableSource, + boolean isBatch) { + if (tEnv instanceof TableEnvironmentInternal) { + CatalogManager catalogManager = ((TableEnvironmentInternal) tEnv).getCatalogManager(); + ObjectIdentifier objectIdentifier = catalogManager.qualifyIdentifier(UnresolvedIdentifier.of(name)); + Optional<CatalogBaseTable> table = getTemporaryTable(catalogManager, objectIdentifier); + + if (table.isPresent()) { + if (table.get() instanceof ConnectorCatalogTable<?, ?>) { + ConnectorCatalogTable<?, ?> sourceSinkTable = (ConnectorCatalogTable<?, ?>) table.get(); + if (sourceSinkTable.getTableSource().isPresent()) { + throw new ValidationException(String.format( + "Table '%s' already exists. Please choose a different name.", name)); + } else { + ConnectorCatalogTable sourceAndSink = ConnectorCatalogTable.sourceAndSink( + tableSource, + sourceSinkTable.getTableSink().get(), + isBatch); + catalogManager.dropTemporaryTable(objectIdentifier, false); + catalogManager.createTemporaryTable(sourceAndSink, objectIdentifier, false); + } + } else { + throw new ValidationException(String.format( + "Table '%s' already exists. Please choose a different name.", name)); + } + } else { + ConnectorCatalogTable source = ConnectorCatalogTable.source(tableSource, isBatch); + catalogManager.createTemporaryTable(source, objectIdentifier, false); + } + } else { + throw new ValidationException(String.format( + "This is a workaround to remove registration of TableSource in Table Env. We " + + "need TableEnvironmentInternal#getCatalogManager to register a table source " + + "as a Table under the given name '%s' in this TableEnvironment's catalog.", + name)); + } + } + + /** + * Register a table sink as a Table under the given name in this + * {@link TableEnvironment}'s catalog. + */ + public static void registerTableSink( Review comment: registerTableSinkInternal ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected]
