fmorillo7694 commented on code in PR #206: URL: https://github.com/apache/flink-connector-aws/pull/206#discussion_r4125185700
########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/factory/GlueCatalogFactory.java: ########## @@ -0,0 +1,107 @@ +/* + * 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.factory; + +import org.apache.flink.annotation.PublicEvolving; +import org.apache.flink.configuration.ConfigOption; +import org.apache.flink.configuration.ConfigOptions; +import org.apache.flink.connector.aws.table.util.AWSOptionUtils; +import org.apache.flink.connector.aws.table.util.HttpClientOptionUtils; +import org.apache.flink.table.catalog.Catalog; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.glue.GlueCatalog; +import org.apache.flink.table.factories.CatalogFactory; +import org.apache.flink.table.factories.FactoryUtil; + +import java.util.HashSet; +import java.util.Map; +import java.util.Properties; +import java.util.Set; + +/** Factory for creating GlueCatalog instances. */ +@PublicEvolving +public class GlueCatalogFactory implements CatalogFactory { + + /** HTTP client implementations selectable via {@code http-client.type}. */ + private static final String[] ALLOWED_HTTP_CLIENT_TYPES = new String[] {"APACHE"}; Review Comment: Done: the factory now uses `AWSConfigConstants.CLIENT_TYPE_APACHE`. While fixing it I found the check itself never ran: `HttpClientOptionUtils` validated `http-client.type` *after* `translateClientKeys` had renamed it to `aws.http-client.type`, so no value was ever rejected (and the `"APACHE"` allow-list would have rejected the only valid one if it had). It now validates on the raw key, case-insensitively; `GlueCatalogFactoryTest` covers reject + case-insensitive accept, `HttpClientOptionUtilsTest` covers the gate directly. ########## 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. Review Comment: Added (`LOG.debug` on the not-found path in `getGlueDatabaseOrNull`). ########## 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; + while (true) { + GetDatabasesRequest.Builder requestBuilder = GetDatabasesRequest.builder(); + if (nextToken != null) { + requestBuilder.nextToken(nextToken); + } + GetDatabasesResponse response = glueClient.getDatabases(requestBuilder.build()); + + for (Database database : response.databaseList()) { + String storedOriginalName = getOriginalDatabaseName(database); + if (storedOriginalName.equalsIgnoreCase(originalDatabaseName)) { + return database; + } + } + + nextToken = response.nextToken(); + if (nextToken == null) { + break; + } + } + + return null; // Database not found + } catch (GlueException e) { + throw new CatalogException("Error searching for database: " + originalDatabaseName, e); + } + } + + /** + * Retrieves the specified database from the Glue catalog. + * + * @param originalDatabaseName The original name of the database to fetch. + * @return The CatalogDatabase object representing the Glue database. + * @throws DatabaseNotExistException If the database does not exist in the Glue catalog. + * @throws CatalogException If there is any error retrieving the database. + */ + public CatalogDatabase getDatabase(String originalDatabaseName) + throws DatabaseNotExistException, CatalogException { + try { + // A single resolution pass both finds the database and returns its metadata, + // so no second GetDatabase call is needed. + Database glueDatabase = findGlueDatabase(originalDatabaseName); + if (glueDatabase == null) { + throw new DatabaseNotExistException(catalogName, originalDatabaseName); + } + return convertGlueDatabase(glueDatabase); + } catch (EntityNotFoundException e) { + throw new DatabaseNotExistException(catalogName, originalDatabaseName); + } catch (InvalidInputException e) { + LOG.error("Invalid input while getting database: {}", originalDatabaseName, e); + throw new CatalogException("Invalid database name: " + originalDatabaseName, e); + } catch (OperationTimeoutException e) { + LOG.error("Timeout while getting database: {}", originalDatabaseName, e); + throw new CatalogException( + "Timeout while getting database: " + originalDatabaseName, e); + } catch (GlueException e) { + LOG.error("Error getting database: {}", originalDatabaseName, e); + throw new CatalogException("Error getting database: " + originalDatabaseName, e); + } + } + + /** + * Converts the Glue database model to a Flink CatalogDatabase. Preserves original database name + * metadata in the properties. + * + * @param glueDatabase The Glue database model. + * @return A CatalogDatabase representing the Glue database. + */ + private CatalogDatabase convertGlueDatabase(Database glueDatabase) { + Map<String, String> properties = new HashMap<>(); + + // Copy all existing parameters + if (glueDatabase.parameters() != null) { + properties.putAll(glueDatabase.parameters()); + } + + return new CatalogDatabaseImpl(properties, glueDatabase.description()); + } + + /** + * Checks whether a database exists in Glue by original name. + * + * @param originalDatabaseName The original name of the database to check. + * @return true if the database exists, false otherwise. + */ + public boolean glueDatabaseExists(String originalDatabaseName) { + try { + String glueDatabaseName = findGlueDatabaseName(originalDatabaseName); + return glueDatabaseName != null; + } catch (CatalogException e) { + LOG.warn("Error checking database existence for: {}", originalDatabaseName, e); + return false; + } + } + + /** + * Direct check if a database exists in Glue by Glue storage name. This method directly calls + * Glue API without going through higher-level search functions. + * + * @param glueDatabaseName The Glue storage name of the database to check. + * @return true if the database exists, false otherwise. + */ + private boolean glueDatabaseExistsByGlueName(String glueDatabaseName) { Review Comment: Removed, together with the full-scan fallback it supported (see the thread on the fallback). ########## 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; + while (true) { + GetDatabasesRequest.Builder requestBuilder = GetDatabasesRequest.builder(); + if (nextToken != null) { + requestBuilder.nextToken(nextToken); + } + GetDatabasesResponse response = glueClient.getDatabases(requestBuilder.build()); + + for (Database database : response.databaseList()) { + String storedOriginalName = getOriginalDatabaseName(database); + if (storedOriginalName.equalsIgnoreCase(originalDatabaseName)) { + return database; + } + } + + nextToken = response.nextToken(); + if (nextToken == null) { + break; + } + } + + return null; // Database not found + } catch (GlueException e) { Review Comment: Agreed. `getGlueDatabaseOrNull` now handles `EntityNotFoundException` (→ null), `InvalidInputException`, `OperationTimeoutException` and `AccessDeniedException` separately, each with a message naming the database, and a Glue timeout is no longer reported as "the database does not exist" (`glueDatabaseExists` throws instead of returning false; two tests flipped accordingly). ########## 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; + while (true) { + GetDatabasesRequest.Builder requestBuilder = GetDatabasesRequest.builder(); + if (nextToken != null) { + requestBuilder.nextToken(nextToken); + } + GetDatabasesResponse response = glueClient.getDatabases(requestBuilder.build()); + + for (Database database : response.databaseList()) { + String storedOriginalName = getOriginalDatabaseName(database); + if (storedOriginalName.equalsIgnoreCase(originalDatabaseName)) { + return database; + } + } + + nextToken = response.nextToken(); + if (nextToken == null) { + break; + } + } + + return null; // Database not found + } catch (GlueException e) { + throw new CatalogException("Error searching for database: " + originalDatabaseName, e); + } + } + + /** + * Retrieves the specified database from the Glue catalog. + * + * @param originalDatabaseName The original name of the database to fetch. + * @return The CatalogDatabase object representing the Glue database. + * @throws DatabaseNotExistException If the database does not exist in the Glue catalog. + * @throws CatalogException If there is any error retrieving the database. + */ + public CatalogDatabase getDatabase(String originalDatabaseName) + throws DatabaseNotExistException, CatalogException { + try { + // A single resolution pass both finds the database and returns its metadata, + // so no second GetDatabase call is needed. + Database glueDatabase = findGlueDatabase(originalDatabaseName); + if (glueDatabase == null) { + throw new DatabaseNotExistException(catalogName, originalDatabaseName); + } + return convertGlueDatabase(glueDatabase); + } catch (EntityNotFoundException e) { + throw new DatabaseNotExistException(catalogName, originalDatabaseName); + } catch (InvalidInputException e) { + LOG.error("Invalid input while getting database: {}", originalDatabaseName, e); + throw new CatalogException("Invalid database name: " + originalDatabaseName, e); + } catch (OperationTimeoutException e) { + LOG.error("Timeout while getting database: {}", originalDatabaseName, e); + throw new CatalogException( + "Timeout while getting database: " + originalDatabaseName, e); + } catch (GlueException e) { + LOG.error("Error getting database: {}", originalDatabaseName, e); + throw new CatalogException("Error getting database: " + originalDatabaseName, e); + } + } + + /** + * Converts the Glue database model to a Flink CatalogDatabase. Preserves original database name + * metadata in the properties. + * + * @param glueDatabase The Glue database model. + * @return A CatalogDatabase representing the Glue database. + */ + private CatalogDatabase convertGlueDatabase(Database glueDatabase) { + Map<String, String> properties = new HashMap<>(); + + // Copy all existing parameters + if (glueDatabase.parameters() != null) { Review Comment: It already does: `convertGlueDatabase` strips `flink.original-database-name` before building the `CatalogDatabase` — that is the only internal parameter on databases. -- 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]
