gguptp commented on code in PR #206: URL: https://github.com/apache/flink-connector-aws/pull/206#discussion_r4107944385
########## 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: i am wondering is this fallback even necessary? if we wont find the database through lowercase search, this will also not find it right? We already create the glue database with lowercase(dbname), so it the DB exists, we will always find it via getDatabase call itself? -- 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]
