fmorillo7694 commented on code in PR #206: URL: https://github.com/apache/flink-connector-aws/pull/206#discussion_r4125187505
########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/operator/GlueDatabaseOperator.java: ########## @@ -0,0 +1,391 @@ +/* + * 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.catalog.glue.operator; + +import org.apache.flink.table.catalog.CatalogDatabase; +import org.apache.flink.table.catalog.CatalogDatabaseImpl; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.glue.util.GlueCatalogConstants; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.model.AlreadyExistsException; +import software.amazon.awssdk.services.glue.model.Database; +import software.amazon.awssdk.services.glue.model.DeleteDatabaseRequest; +import software.amazon.awssdk.services.glue.model.EntityNotFoundException; +import software.amazon.awssdk.services.glue.model.GetDatabaseRequest; +import software.amazon.awssdk.services.glue.model.GetDatabasesRequest; +import software.amazon.awssdk.services.glue.model.GetDatabasesResponse; +import software.amazon.awssdk.services.glue.model.GlueException; +import software.amazon.awssdk.services.glue.model.InvalidInputException; +import software.amazon.awssdk.services.glue.model.OperationTimeoutException; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.regex.Pattern; + +/** + * Handles all database-related operations for the Glue catalog. Provides functionality for listing, + * retrieving, creating, and deleting databases in AWS Glue. + */ +public class GlueDatabaseOperator extends GlueOperator { + + /** Logger for logging database operations. */ + private static final Logger LOG = LoggerFactory.getLogger(GlueDatabaseOperator.class); + + /** + * Pattern for validating database names. AWS Glue supports alphanumeric characters and + * underscores. We preserve original case in metadata while storing lowercase in Glue. + */ + private static final Pattern VALID_NAME_PATTERN = Pattern.compile("^[a-zA-Z0-9_]+$"); + + /** + * Constructor for GlueDatabaseOperations. Initializes the Glue client and catalog name. + * + * @param glueClient The Glue client to interact with AWS Glue. + * @param catalogName The name of the catalog. + */ + public GlueDatabaseOperator(GlueClient glueClient, String catalogName) { + super(glueClient, catalogName); + } + + /** + * Validates that a database name contains only allowed characters. AWS Glue supports + * alphanumeric characters and underscores. Case is preserved in metadata while Glue stores + * lowercase internally. + * + * @param databaseName The database name to validate + * @throws CatalogException if the database name contains invalid characters + */ + private void validateDatabaseName(String databaseName) { + if (databaseName == null || databaseName.isEmpty()) { + throw new CatalogException("Database name cannot be null or empty"); + } + + if (!VALID_NAME_PATTERN.matcher(databaseName).matches()) { + throw new CatalogException( + "Database name can only contain letters, numbers, and underscores. " + + "Original case is preserved in metadata while AWS Glue stores lowercase internally."); + } + } + + /** + * Lists all the databases in the Glue catalog. Returns the original database names as specified + * by users, not the lowercase names stored in Glue. + * + * @return A list of database names with original case preserved. + * @throws CatalogException if there is an error fetching the list of databases. + */ + public List<String> listDatabases() throws CatalogException { + try { + List<String> databaseNames = new ArrayList<>(); + String nextToken = null; + while (true) { + GetDatabasesRequest.Builder requestBuilder = GetDatabasesRequest.builder(); + if (nextToken != null) { + requestBuilder.nextToken(nextToken); + } + GetDatabasesResponse response = glueClient.getDatabases(requestBuilder.build()); + + // Extract original names from database metadata + for (Database database : response.databaseList()) { + String originalName = getOriginalDatabaseName(database); + databaseNames.add(originalName); + } + + nextToken = response.nextToken(); + if (nextToken == null) { + break; + } + } + return databaseNames; + } catch (GlueException e) { + LOG.error("Failed to list databases in Glue", e); + throw new CatalogException("Failed to list databases: " + e.getMessage(), e); + } + } + + /** + * Extracts the original database name from a Glue database object. Falls back to the stored + * name if no original name is found. + * + * @param database The Glue database object + * @return The original database name with case preserved + */ + private String getOriginalDatabaseName(Database database) { + if (database.parameters() != null + && database.parameters().containsKey(GlueCatalogConstants.ORIGINAL_DATABASE_NAME)) { + return database.parameters().get(GlueCatalogConstants.ORIGINAL_DATABASE_NAME); + } + // Fallback to stored name for backward compatibility + return database.name(); + } + + /** + * Converts a user-provided database name to the name used for storage in Glue. Glue requires + * lowercase names, so we store in lowercase but preserve original in metadata. + * + * @param originalDatabaseName The original database name as specified by the user + * @return The database name to use for Glue storage (lowercase) + */ + private String toGlueDatabaseName(String originalDatabaseName) { + return originalDatabaseName.toLowerCase(); + } + + /** + * Finds the Glue storage name for a given original database name. This is needed because users + * may specify names with different casing than stored in Glue. + * + * @param originalDatabaseName The original database name to find + * @return The Glue storage name if found, null if not found + * @throws CatalogException if there's an error searching + */ + public String findGlueDatabaseName(String originalDatabaseName) throws CatalogException { + Database database = findGlueDatabase(originalDatabaseName); + return database == null ? null : database.name(); + } + + /** + * Finds the Glue database for a given original database name, returning its full metadata. The + * common case (lowercase match) resolves with a single GetDatabase call, which both proves + * existence and provides the parameters needed to verify the stored original name. + * + * @param originalDatabaseName The original database name to find + * @return The Glue database if found, null if not found + * @throws CatalogException if there's an error searching + */ + private Database findGlueDatabase(String originalDatabaseName) throws CatalogException { + try { + // First try the direct lowercase match (most common case) with a single call. + String glueName = toGlueDatabaseName(originalDatabaseName); + try { + Database database = + glueClient + .getDatabase(GetDatabaseRequest.builder().name(glueName).build()) + .database(); + // SQL identifiers are case-insensitive: any case variation of the stored + // original name resolves to the same database (Glue prevents two databases + // from sharing the same lowercase storage name). + if (database != null + && getOriginalDatabaseName(database) + .equalsIgnoreCase(originalDatabaseName)) { + return database; + } + } catch (EntityNotFoundException e) { + // Fall through to the full search below. + } + + // If direct match failed, search all databases (for backward compatibility or edge + // cases) + String nextToken = null; Review Comment: You are right — since we always store `lowercase(name)` and Glue enforces unique storage names, the lowercase `GetDatabase` is definitive. `findGlueDatabase` is now a single `GetDatabase` call (`getGlueDatabaseOrNull`); the scan fallback and `glueDatabaseExistsByGlueName` are gone. Same reasoning applied on the table side (`getGlueTableOrNull`), which also removed the unreachable case-insensitive conflict checks. ########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/operator/GlueFunctionOperator.java: ########## @@ -0,0 +1,315 @@ +/* + * 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.catalog.glue.operator; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.catalog.CatalogFunction; +import org.apache.flink.table.catalog.CatalogFunctionImpl; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.glue.util.GlueCatalogConstants; +import org.apache.flink.table.catalog.glue.util.GlueFunctionsUtil; +import org.apache.flink.table.resource.ResourceUri; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.model.AlreadyExistsException; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.EntityNotFoundException; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsResponse; +import software.amazon.awssdk.services.glue.model.GlueException; +import software.amazon.awssdk.services.glue.model.PrincipalType; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.UserDefinedFunction; +import software.amazon.awssdk.services.glue.model.UserDefinedFunctionInput; + +import java.util.Collection; +import java.util.LinkedList; +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; + +/** Utilities for Glue catalog Function related operations. */ +@Internal +public class GlueFunctionOperator extends GlueOperator { + + private static final Logger LOG = LoggerFactory.getLogger(GlueFunctionOperator.class); + + /** + * Constructor to initialize the shared fields. + * + * @param glueClient The Glue client used for interacting with the AWS Glue service. + * @param catalogName The catalog name associated with the Glue operations. + */ + public GlueFunctionOperator(GlueClient glueClient, String catalogName) { + super(glueClient, catalogName); + } + + /** + * Create a function. Function name should be handled in a case-insensitive way. + * + * @param functionPath path of the function + * @param function Flink function to be created + * @throws CatalogException in case of any runtime exception + */ + public void createGlueFunction(ObjectPath functionPath, CatalogFunction function) + throws CatalogException, FunctionAlreadyExistException { + UserDefinedFunctionInput functionInput = createFunctionInput(functionPath, function); + CreateUserDefinedFunctionRequest.Builder createUDFRequest = + CreateUserDefinedFunctionRequest.builder() + .databaseName(functionPath.getDatabaseName()) + .functionInput(functionInput); + try { + CreateUserDefinedFunctionResponse response = + glueClient.createUserDefinedFunction(createUDFRequest.build()); + if (response == null + || (response.sdkHttpResponse() != null + && !response.sdkHttpResponse().isSuccessful())) { + throw new CatalogException( + "Error creating function: " + functionPath.getFullName()); + } Review Comment: Removed. All function operations now rely on the SDK's typed exceptions only (`AlreadyExistsException` → `FunctionAlreadyExistException`, `EntityNotFoundException` → `FunctionNotExistException`, `InvalidInputException`/`GlueException` → `CatalogException` with the function name). ########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/operator/GlueFunctionOperator.java: ########## @@ -0,0 +1,315 @@ +/* + * 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.catalog.glue.operator; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.catalog.CatalogFunction; +import org.apache.flink.table.catalog.CatalogFunctionImpl; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.glue.util.GlueCatalogConstants; +import org.apache.flink.table.catalog.glue.util.GlueFunctionsUtil; +import org.apache.flink.table.resource.ResourceUri; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.model.AlreadyExistsException; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.EntityNotFoundException; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsResponse; +import software.amazon.awssdk.services.glue.model.GlueException; +import software.amazon.awssdk.services.glue.model.PrincipalType; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.UserDefinedFunction; +import software.amazon.awssdk.services.glue.model.UserDefinedFunctionInput; + +import java.util.Collection; +import java.util.LinkedList; +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; + +/** Utilities for Glue catalog Function related operations. */ +@Internal +public class GlueFunctionOperator extends GlueOperator { + + private static final Logger LOG = LoggerFactory.getLogger(GlueFunctionOperator.class); + + /** + * Constructor to initialize the shared fields. + * + * @param glueClient The Glue client used for interacting with the AWS Glue service. + * @param catalogName The catalog name associated with the Glue operations. + */ + public GlueFunctionOperator(GlueClient glueClient, String catalogName) { + super(glueClient, catalogName); + } + + /** + * Create a function. Function name should be handled in a case-insensitive way. + * + * @param functionPath path of the function + * @param function Flink function to be created + * @throws CatalogException in case of any runtime exception + */ + public void createGlueFunction(ObjectPath functionPath, CatalogFunction function) + throws CatalogException, FunctionAlreadyExistException { + UserDefinedFunctionInput functionInput = createFunctionInput(functionPath, function); + CreateUserDefinedFunctionRequest.Builder createUDFRequest = + CreateUserDefinedFunctionRequest.builder() + .databaseName(functionPath.getDatabaseName()) + .functionInput(functionInput); + try { + CreateUserDefinedFunctionResponse response = + glueClient.createUserDefinedFunction(createUDFRequest.build()); + if (response == null + || (response.sdkHttpResponse() != null + && !response.sdkHttpResponse().isSuccessful())) { + throw new CatalogException( + "Error creating function: " + functionPath.getFullName()); + } + LOG.info("Created Function: {}", functionPath.getFullName()); + } catch (AlreadyExistsException e) { + LOG.error( + String.format( + "%s already Exists. Function language of type: %s. \n%s", + functionPath.getFullName(), function.getFunctionLanguage(), e)); + throw new FunctionAlreadyExistException(catalogName, functionPath, e); + } catch (GlueException e) { + LOG.error("Error creating glue function: {}\n{}", functionPath.getFullName(), e); + throw new CatalogException(GlueCatalogConstants.GLUE_EXCEPTION_MSG_IDENTIFIER, e); + } + } + + /** + * Modify an existing function. Function name should be handled in a case-insensitive way. + * + * @param functionPath path of function. + * @param newFunction modified function. + * @throws CatalogException on runtime errors. + * @throws FunctionNotExistException if the function doesn't exist. + */ + public void alterGlueFunction(ObjectPath functionPath, CatalogFunction newFunction) + throws CatalogException, FunctionNotExistException { + + UserDefinedFunctionInput functionInput = createFunctionInput(functionPath, newFunction); + + UpdateUserDefinedFunctionRequest updateUserDefinedFunctionRequest = + UpdateUserDefinedFunctionRequest.builder() + .functionName(functionPath.getObjectName()) + .databaseName(functionPath.getDatabaseName()) + .functionInput(functionInput) + .build(); + try { + UpdateUserDefinedFunctionResponse response = + glueClient.updateUserDefinedFunction(updateUserDefinedFunctionRequest); + if (response == null + || (response.sdkHttpResponse() != null + && !response.sdkHttpResponse().isSuccessful())) { + throw new CatalogException( + "Error altering function: " + functionPath.getFullName()); Review Comment: Done — `alterGlueFunction` catches `EntityNotFoundException` → `FunctionNotExistException`, `InvalidInputException` → `CatalogException("Invalid function definition for db.fn: ...")`, and `GlueException` last. ########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/operator/GlueFunctionOperator.java: ########## @@ -0,0 +1,315 @@ +/* + * 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.catalog.glue.operator; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.catalog.CatalogFunction; +import org.apache.flink.table.catalog.CatalogFunctionImpl; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.glue.util.GlueCatalogConstants; +import org.apache.flink.table.catalog.glue.util.GlueFunctionsUtil; +import org.apache.flink.table.resource.ResourceUri; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.model.AlreadyExistsException; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.EntityNotFoundException; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsResponse; +import software.amazon.awssdk.services.glue.model.GlueException; +import software.amazon.awssdk.services.glue.model.PrincipalType; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.UserDefinedFunction; +import software.amazon.awssdk.services.glue.model.UserDefinedFunctionInput; + +import java.util.Collection; +import java.util.LinkedList; +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; + +/** Utilities for Glue catalog Function related operations. */ +@Internal +public class GlueFunctionOperator extends GlueOperator { + + private static final Logger LOG = LoggerFactory.getLogger(GlueFunctionOperator.class); + + /** + * Constructor to initialize the shared fields. + * + * @param glueClient The Glue client used for interacting with the AWS Glue service. + * @param catalogName The catalog name associated with the Glue operations. + */ + public GlueFunctionOperator(GlueClient glueClient, String catalogName) { + super(glueClient, catalogName); + } + + /** + * Create a function. Function name should be handled in a case-insensitive way. + * + * @param functionPath path of the function + * @param function Flink function to be created + * @throws CatalogException in case of any runtime exception + */ + public void createGlueFunction(ObjectPath functionPath, CatalogFunction function) + throws CatalogException, FunctionAlreadyExistException { + UserDefinedFunctionInput functionInput = createFunctionInput(functionPath, function); + CreateUserDefinedFunctionRequest.Builder createUDFRequest = + CreateUserDefinedFunctionRequest.builder() + .databaseName(functionPath.getDatabaseName()) + .functionInput(functionInput); + try { + CreateUserDefinedFunctionResponse response = + glueClient.createUserDefinedFunction(createUDFRequest.build()); + if (response == null + || (response.sdkHttpResponse() != null + && !response.sdkHttpResponse().isSuccessful())) { + throw new CatalogException( + "Error creating function: " + functionPath.getFullName()); + } + LOG.info("Created Function: {}", functionPath.getFullName()); + } catch (AlreadyExistsException e) { + LOG.error( + String.format( + "%s already Exists. Function language of type: %s. \n%s", + functionPath.getFullName(), function.getFunctionLanguage(), e)); + throw new FunctionAlreadyExistException(catalogName, functionPath, e); + } catch (GlueException e) { + LOG.error("Error creating glue function: {}\n{}", functionPath.getFullName(), e); + throw new CatalogException(GlueCatalogConstants.GLUE_EXCEPTION_MSG_IDENTIFIER, e); + } + } + + /** + * Modify an existing function. Function name should be handled in a case-insensitive way. + * + * @param functionPath path of function. + * @param newFunction modified function. + * @throws CatalogException on runtime errors. + * @throws FunctionNotExistException if the function doesn't exist. + */ + public void alterGlueFunction(ObjectPath functionPath, CatalogFunction newFunction) + throws CatalogException, FunctionNotExistException { + + UserDefinedFunctionInput functionInput = createFunctionInput(functionPath, newFunction); + + UpdateUserDefinedFunctionRequest updateUserDefinedFunctionRequest = + UpdateUserDefinedFunctionRequest.builder() + .functionName(functionPath.getObjectName()) + .databaseName(functionPath.getDatabaseName()) + .functionInput(functionInput) + .build(); + try { + UpdateUserDefinedFunctionResponse response = + glueClient.updateUserDefinedFunction(updateUserDefinedFunctionRequest); + if (response == null + || (response.sdkHttpResponse() != null + && !response.sdkHttpResponse().isSuccessful())) { + throw new CatalogException( + "Error altering function: " + functionPath.getFullName()); + } + LOG.info("Altered Function: {}", functionPath.getFullName()); + } catch (EntityNotFoundException e) { + LOG.error("Function not found: {}", functionPath.getFullName()); + throw new FunctionNotExistException(catalogName, functionPath, e); + } catch (GlueException e) { + LOG.error("Error altering glue function: {}\n{}", functionPath.getFullName(), e); + throw new CatalogException(GlueCatalogConstants.GLUE_EXCEPTION_MSG_IDENTIFIER, e); Review Comment: Done across create/alter/get/drop/list: messages now read e.g. `Invalid function definition for db.fn: <glue message>` / `Error dropping function db.fn: <glue message>`, and the Glue exception is kept as the cause. `GlueFunctionOperatorTest` pins the exact messages. ########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/operator/GlueFunctionOperator.java: ########## @@ -0,0 +1,315 @@ +/* + * 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.catalog.glue.operator; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.catalog.CatalogFunction; +import org.apache.flink.table.catalog.CatalogFunctionImpl; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.glue.util.GlueCatalogConstants; +import org.apache.flink.table.catalog.glue.util.GlueFunctionsUtil; +import org.apache.flink.table.resource.ResourceUri; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.model.AlreadyExistsException; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.EntityNotFoundException; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsResponse; +import software.amazon.awssdk.services.glue.model.GlueException; +import software.amazon.awssdk.services.glue.model.PrincipalType; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.UserDefinedFunction; +import software.amazon.awssdk.services.glue.model.UserDefinedFunctionInput; + +import java.util.Collection; +import java.util.LinkedList; +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; + +/** Utilities for Glue catalog Function related operations. */ +@Internal +public class GlueFunctionOperator extends GlueOperator { + + private static final Logger LOG = LoggerFactory.getLogger(GlueFunctionOperator.class); + + /** + * Constructor to initialize the shared fields. + * + * @param glueClient The Glue client used for interacting with the AWS Glue service. + * @param catalogName The catalog name associated with the Glue operations. + */ + public GlueFunctionOperator(GlueClient glueClient, String catalogName) { + super(glueClient, catalogName); + } + + /** + * Create a function. Function name should be handled in a case-insensitive way. + * + * @param functionPath path of the function + * @param function Flink function to be created + * @throws CatalogException in case of any runtime exception + */ + public void createGlueFunction(ObjectPath functionPath, CatalogFunction function) + throws CatalogException, FunctionAlreadyExistException { + UserDefinedFunctionInput functionInput = createFunctionInput(functionPath, function); + CreateUserDefinedFunctionRequest.Builder createUDFRequest = + CreateUserDefinedFunctionRequest.builder() + .databaseName(functionPath.getDatabaseName()) + .functionInput(functionInput); + try { + CreateUserDefinedFunctionResponse response = + glueClient.createUserDefinedFunction(createUDFRequest.build()); + if (response == null + || (response.sdkHttpResponse() != null + && !response.sdkHttpResponse().isSuccessful())) { + throw new CatalogException( + "Error creating function: " + functionPath.getFullName()); + } + LOG.info("Created Function: {}", functionPath.getFullName()); + } catch (AlreadyExistsException e) { + LOG.error( + String.format( + "%s already Exists. Function language of type: %s. \n%s", + functionPath.getFullName(), function.getFunctionLanguage(), e)); + throw new FunctionAlreadyExistException(catalogName, functionPath, e); + } catch (GlueException e) { + LOG.error("Error creating glue function: {}\n{}", functionPath.getFullName(), e); + throw new CatalogException(GlueCatalogConstants.GLUE_EXCEPTION_MSG_IDENTIFIER, e); + } + } + + /** + * Modify an existing function. Function name should be handled in a case-insensitive way. + * + * @param functionPath path of function. + * @param newFunction modified function. + * @throws CatalogException on runtime errors. + * @throws FunctionNotExistException if the function doesn't exist. + */ + public void alterGlueFunction(ObjectPath functionPath, CatalogFunction newFunction) + throws CatalogException, FunctionNotExistException { + + UserDefinedFunctionInput functionInput = createFunctionInput(functionPath, newFunction); + + UpdateUserDefinedFunctionRequest updateUserDefinedFunctionRequest = + UpdateUserDefinedFunctionRequest.builder() + .functionName(functionPath.getObjectName()) + .databaseName(functionPath.getDatabaseName()) + .functionInput(functionInput) + .build(); + try { + UpdateUserDefinedFunctionResponse response = + glueClient.updateUserDefinedFunction(updateUserDefinedFunctionRequest); + if (response == null + || (response.sdkHttpResponse() != null + && !response.sdkHttpResponse().isSuccessful())) { + throw new CatalogException( + "Error altering function: " + functionPath.getFullName()); + } + LOG.info("Altered Function: {}", functionPath.getFullName()); + } catch (EntityNotFoundException e) { + LOG.error("Function not found: {}", functionPath.getFullName()); + throw new FunctionNotExistException(catalogName, functionPath, e); + } catch (GlueException e) { + LOG.error("Error altering glue function: {}\n{}", functionPath.getFullName(), e); + throw new CatalogException(GlueCatalogConstants.GLUE_EXCEPTION_MSG_IDENTIFIER, e); + } + } + + /** + * Get the user defined function from glue Catalog. Function name should be handled in a + * case-insensitive way. + * + * @param functionPath path of the function + * @return the requested function + * @throws CatalogException in case of any runtime exception + * @throws FunctionNotExistException if the function doesn't exist + */ + public CatalogFunction getGlueFunction(ObjectPath functionPath) + throws CatalogException, FunctionNotExistException { + GetUserDefinedFunctionRequest request = + GetUserDefinedFunctionRequest.builder() + .databaseName(functionPath.getDatabaseName()) + .functionName(functionPath.getObjectName()) + .build(); + try { + GetUserDefinedFunctionResponse response = glueClient.getUserDefinedFunction(request); + UserDefinedFunction udf = response.userDefinedFunction(); Review Comment: It cannot: the SDK throws `EntityNotFoundException` rather than returning a null `userDefinedFunction`. The method is now `getGlueFunctionOrNull` (null only on `EntityNotFoundException`) and `getGlueFunction` throws `FunctionNotExistException` on null, so there is a single well-defined null. ########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/operator/GlueFunctionOperator.java: ########## @@ -0,0 +1,315 @@ +/* + * 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.catalog.glue.operator; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.catalog.CatalogFunction; +import org.apache.flink.table.catalog.CatalogFunctionImpl; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.glue.util.GlueCatalogConstants; +import org.apache.flink.table.catalog.glue.util.GlueFunctionsUtil; +import org.apache.flink.table.resource.ResourceUri; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.services.glue.GlueClient; +import software.amazon.awssdk.services.glue.model.AlreadyExistsException; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.CreateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.DeleteUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.EntityNotFoundException; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsRequest; +import software.amazon.awssdk.services.glue.model.GetUserDefinedFunctionsResponse; +import software.amazon.awssdk.services.glue.model.GlueException; +import software.amazon.awssdk.services.glue.model.PrincipalType; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionRequest; +import software.amazon.awssdk.services.glue.model.UpdateUserDefinedFunctionResponse; +import software.amazon.awssdk.services.glue.model.UserDefinedFunction; +import software.amazon.awssdk.services.glue.model.UserDefinedFunctionInput; + +import java.util.Collection; +import java.util.LinkedList; +import java.util.List; +import java.util.Optional; +import java.util.stream.Collectors; + +/** Utilities for Glue catalog Function related operations. */ +@Internal +public class GlueFunctionOperator extends GlueOperator { + + private static final Logger LOG = LoggerFactory.getLogger(GlueFunctionOperator.class); + + /** + * Constructor to initialize the shared fields. + * + * @param glueClient The Glue client used for interacting with the AWS Glue service. + * @param catalogName The catalog name associated with the Glue operations. + */ + public GlueFunctionOperator(GlueClient glueClient, String catalogName) { + super(glueClient, catalogName); + } + + /** + * Create a function. Function name should be handled in a case-insensitive way. + * + * @param functionPath path of the function + * @param function Flink function to be created + * @throws CatalogException in case of any runtime exception + */ + public void createGlueFunction(ObjectPath functionPath, CatalogFunction function) + throws CatalogException, FunctionAlreadyExistException { + UserDefinedFunctionInput functionInput = createFunctionInput(functionPath, function); + CreateUserDefinedFunctionRequest.Builder createUDFRequest = + CreateUserDefinedFunctionRequest.builder() + .databaseName(functionPath.getDatabaseName()) + .functionInput(functionInput); + try { + CreateUserDefinedFunctionResponse response = + glueClient.createUserDefinedFunction(createUDFRequest.build()); + if (response == null + || (response.sdkHttpResponse() != null + && !response.sdkHttpResponse().isSuccessful())) { + throw new CatalogException( + "Error creating function: " + functionPath.getFullName()); + } + LOG.info("Created Function: {}", functionPath.getFullName()); + } catch (AlreadyExistsException e) { + LOG.error( + String.format( + "%s already Exists. Function language of type: %s. \n%s", + functionPath.getFullName(), function.getFunctionLanguage(), e)); + throw new FunctionAlreadyExistException(catalogName, functionPath, e); + } catch (GlueException e) { + LOG.error("Error creating glue function: {}\n{}", functionPath.getFullName(), e); + throw new CatalogException(GlueCatalogConstants.GLUE_EXCEPTION_MSG_IDENTIFIER, e); + } + } + + /** + * Modify an existing function. Function name should be handled in a case-insensitive way. + * + * @param functionPath path of function. + * @param newFunction modified function. + * @throws CatalogException on runtime errors. + * @throws FunctionNotExistException if the function doesn't exist. + */ + public void alterGlueFunction(ObjectPath functionPath, CatalogFunction newFunction) + throws CatalogException, FunctionNotExistException { + + UserDefinedFunctionInput functionInput = createFunctionInput(functionPath, newFunction); + + UpdateUserDefinedFunctionRequest updateUserDefinedFunctionRequest = + UpdateUserDefinedFunctionRequest.builder() + .functionName(functionPath.getObjectName()) + .databaseName(functionPath.getDatabaseName()) + .functionInput(functionInput) + .build(); + try { + UpdateUserDefinedFunctionResponse response = + glueClient.updateUserDefinedFunction(updateUserDefinedFunctionRequest); + if (response == null + || (response.sdkHttpResponse() != null + && !response.sdkHttpResponse().isSuccessful())) { + throw new CatalogException( + "Error altering function: " + functionPath.getFullName()); + } + LOG.info("Altered Function: {}", functionPath.getFullName()); + } catch (EntityNotFoundException e) { + LOG.error("Function not found: {}", functionPath.getFullName()); + throw new FunctionNotExistException(catalogName, functionPath, e); + } catch (GlueException e) { + LOG.error("Error altering glue function: {}\n{}", functionPath.getFullName(), e); + throw new CatalogException(GlueCatalogConstants.GLUE_EXCEPTION_MSG_IDENTIFIER, e); + } + } + + /** + * Get the user defined function from glue Catalog. Function name should be handled in a + * case-insensitive way. + * + * @param functionPath path of the function + * @return the requested function + * @throws CatalogException in case of any runtime exception + * @throws FunctionNotExistException if the function doesn't exist + */ + public CatalogFunction getGlueFunction(ObjectPath functionPath) + throws CatalogException, FunctionNotExistException { + GetUserDefinedFunctionRequest request = + GetUserDefinedFunctionRequest.builder() + .databaseName(functionPath.getDatabaseName()) + .functionName(functionPath.getObjectName()) + .build(); + try { + GetUserDefinedFunctionResponse response = glueClient.getUserDefinedFunction(request); + UserDefinedFunction udf = response.userDefinedFunction(); + List<ResourceUri> resourceUris = + udf.resourceUris().stream() + .map( + resourceUri -> + new org.apache.flink.table.resource.ResourceUri( + org.apache.flink.table.resource.ResourceType + .valueOf( + resourceUri Review Comment: Confirmed it can: the SDK enum maps any resource type it does not know to `UNKNOWN_TO_SDK_VERSION`, and `ResourceType.valueOf("UNKNOWN_TO_SDK_VERSION")` would have thrown an opaque `IllegalArgumentException`. `toCatalogFunction` now reads `resourceTypeAsString()` and throws a `CatalogException` naming the function, the type string and the URI ("…that Flink does not support; supported types are JAR, FILE and ARCHIVE"). -- 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. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
