fmorillo7694 commented on code in PR #206: URL: https://github.com/apache/flink-connector-aws/pull/206#discussion_r4102958213
########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/constants/GlueCatalogConstants.java: ########## @@ -0,0 +1,63 @@ +/* + * 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.constants; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.table.catalog.glue.GlueCatalog; + +import java.util.regex.Pattern; + +/** Constants and Defined Values used for {@link GlueCatalog}. */ +@Internal +public class GlueCatalogConstants { Review Comment: Removed in 56f13dd - `constants/GlueCatalogConstants` was fully unused (everything imports the util one). Also removed `constants/AWSGlueConfigConstants` and its test: it was dead as well, and its endpoint use case is now served by the standard `aws.endpoint` option through the shared client path. ########## docs/content/docs/connectors/table/glue.md: ########## @@ -0,0 +1,440 @@ +--- +title: "AWS Glue Catalog" +weight: 11 +type: docs +aliases: + - /dev/table/connectors/glue.html +--- +<!-- +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. +--> + +# AWS Glue Catalog + +The AWS Glue Catalog provides a way to use [AWS Glue](https://aws.amazon.com/glue) as a catalog for Apache Flink. +This allows users to access Glue's metadata store directly from Flink SQL and Table API. + +## Features + +- Register AWS Glue as a catalog in Flink applications +- Access Glue databases and tables through Flink SQL +- Support for various AWS data sources (S3, Kinesis, MSK) +- Mapping between Flink and AWS Glue data types +- Compatibility with Flink's Table API and SQL interface + +The Glue Catalog is registered through the Table API / SQL. DataStream applications can also +use it by converting between DataStreams and Tables with the +[DataStream API integration]({{< ref "docs/dev/table/data_stream_api" >}}), so tables backed by +Glue metadata are accessible from DataStream programs through a `StreamTableEnvironment`. + +## Dependencies + +{{< sql_download_table "glue" >}} Review Comment: Yes - fixed to `sql_connector_download_table` in 37f62bc (matches kinesis.md/dynamodb.md). ########## flink-catalog-aws/flink-catalog-aws-glue/src/test/java/org/apache/flink/table/catalog/glue/GlueCatalogTest.java: ########## @@ -0,0 +1,1131 @@ +/* + * 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; + +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.CatalogBaseTable; +import org.apache.flink.table.catalog.CatalogDatabase; +import org.apache.flink.table.catalog.CatalogDatabaseImpl; +import org.apache.flink.table.catalog.CatalogFunction; +import org.apache.flink.table.catalog.CatalogFunctionImpl; +import org.apache.flink.table.catalog.CatalogPartition; +import org.apache.flink.table.catalog.CatalogPartitionImpl; +import org.apache.flink.table.catalog.CatalogPartitionSpec; +import org.apache.flink.table.catalog.CatalogTable; +import org.apache.flink.table.catalog.CatalogView; +import org.apache.flink.table.catalog.FunctionLanguage; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.catalog.ResolvedCatalogTable; +import org.apache.flink.table.catalog.ResolvedCatalogView; +import org.apache.flink.table.catalog.ResolvedSchema; +import org.apache.flink.table.catalog.exceptions.CatalogException; +import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotEmptyException; +import org.apache.flink.table.catalog.exceptions.DatabaseNotExistException; +import org.apache.flink.table.catalog.exceptions.FunctionAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.FunctionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionAlreadyExistsException; +import org.apache.flink.table.catalog.exceptions.PartitionNotExistException; +import org.apache.flink.table.catalog.exceptions.PartitionSpecInvalidException; +import org.apache.flink.table.catalog.exceptions.TableAlreadyExistException; +import org.apache.flink.table.catalog.exceptions.TableNotExistException; +import org.apache.flink.table.catalog.exceptions.TableNotPartitionedException; +import org.apache.flink.table.catalog.glue.operator.GlueDatabaseOperator; +import org.apache.flink.table.catalog.glue.operator.GlueTableOperator; +import org.apache.flink.table.catalog.glue.util.GlueTestClientFactory; +import org.apache.flink.table.catalog.glue.util.RealGlueCleanupExtension; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import software.amazon.awssdk.services.glue.GlueClient; + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Comprehensive tests for GlueCatalog. Covers basic operations, advanced features, and edge cases. + */ +@ExtendWith(RealGlueCleanupExtension.class) +public class GlueCatalogTest { + + private GlueClient glueClient; + private GlueCatalog glueCatalog; + private GlueTableOperator glueTableOperations; + private GlueDatabaseOperator glueDatabaseOperations; + + @BeforeEach + void setUp() { + // In-memory FakeGlueClient by default (CI); real AWS Glue when credentials are + // supplied - see GlueTestClientFactory. + String region = "us-east-1"; + String defaultDB = "default"; + glueClient = GlueTestClientFactory.createClient(); + glueTableOperations = new GlueTableOperator(glueClient, "testCatalog"); + glueDatabaseOperations = new GlueDatabaseOperator(glueClient, "testCatalog"); + + glueCatalog = new GlueCatalog("glueCatalog", defaultDB, region, glueClient); + } + + @AfterEach + void tearDown() { + // Close the catalog to release resources + if (glueCatalog != null) { + glueCatalog.close(); + } + } + + // ------------------------------------------------------------------------- + // Constructor, Open, Close Tests + // ------------------------------------------------------------------------- + + /** Test constructor without explicit GlueClient. */ + @Test + public void testConstructorWithoutGlueClient() { + // Instead of testing the actual AWS client creation which causes + // ConcurrentModificationException in tests, we'll verify the class can be + // instantiated and used properly with parameters + assertThatCode( + () -> { + // Create catalog with parameters but no client + GlueCatalog catalog = + new GlueCatalog( + "glueCatalog", "default", "us-east-1", glueClient); + // Use our fake client to avoid AWS SDK issues + catalog.open(); + catalog.close(); + }) + .doesNotThrowAnyException(); + } + + /** Test open and close methods. */ + @Test + public void testOpenAndClose() { + // Act & Assert + assertThatCode( + () -> { + glueCatalog.open(); + glueCatalog.close(); + }) + .doesNotThrowAnyException(); + } + + // ------------------------------------------------------------------------- + // Database Operations Tests + // ------------------------------------------------------------------------- + + /** Test creating a database. */ + @Test + public void testCreateDatabase() throws CatalogException, DatabaseAlreadyExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + + // Act + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Assert + assertThat(glueDatabaseOperations.glueDatabaseExists(databaseName)).isTrue(); + } + + /** Test database exists. */ + @Test + public void testDatabaseExists() throws DatabaseAlreadyExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Act & Assert + assertThat(glueCatalog.databaseExists(databaseName)).isTrue(); + assertThat(glueCatalog.databaseExists("nonexistingdatabase")).isFalse(); + } + + /** Test create database with ifNotExists=true. */ + @Test + public void testCreateDatabaseIfNotExists() throws DatabaseAlreadyExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + + // Create database first time + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Act - Create again with ifNotExists=true should not throw exception + assertThatCode( + () -> { + glueCatalog.createDatabase(databaseName, catalogDatabase, true); + }) + .doesNotThrowAnyException(); + + // Assert + assertThat(glueCatalog.databaseExists(databaseName)).isTrue(); + } + + /** Test drop database. */ + @Test + public void testDropDatabase() + throws DatabaseAlreadyExistException, + DatabaseNotExistException, + DatabaseNotEmptyException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Act + glueCatalog.dropDatabase(databaseName, false, false); + + // Assert + assertThat(glueCatalog.databaseExists(databaseName)).isFalse(); + } + + /** Test drop database with ignoreIfNotExists=true. */ + @Test + public void testDropDatabaseIgnoreIfNotExists() { + // Act & Assert - should not throw exception with ignoreIfNotExists=true + assertThatCode( + () -> { + glueCatalog.dropDatabase("nonexistingdatabase", true, false); + }) + .doesNotThrowAnyException(); + } + + /** Test drop database with ignoreIfNotExists=false. */ + @Test + public void testDropDatabaseFailIfNotExists() { + // Act & Assert - should throw exception with ignoreIfNotExists=false + assertThatThrownBy( + () -> { + glueCatalog.dropDatabase("nonexistingdatabase", false, false); + }) + .isInstanceOf(DatabaseNotExistException.class); + } + + /** Test drop non-empty database with cascade=false should throw DatabaseNotEmptyException. */ + @Test + public void testDropNonEmptyDatabaseWithoutCascade() + throws DatabaseAlreadyExistException, + TableAlreadyExistException, + DatabaseNotExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String tableName = "testtable"; + + // Create database + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Create table in database + CatalogTable catalogTable = + CatalogTable.newBuilder() + .schema(Schema.newBuilder().build()) + .comment("test table") + .partitionKeys(Collections.emptyList()) + .options(Collections.emptyMap()) + .build(); + ResolvedSchema resolvedSchema = ResolvedSchema.of(); + ResolvedCatalogTable resolvedCatalogTable = + new ResolvedCatalogTable(catalogTable, resolvedSchema); + glueCatalog.createTable( + new ObjectPath(databaseName, tableName), resolvedCatalogTable, false); + + // Act & Assert - should throw DatabaseNotEmptyException with cascade=false + assertThatThrownBy( + () -> { + glueCatalog.dropDatabase(databaseName, false, false); + }) + .isInstanceOf(DatabaseNotEmptyException.class); + + // Verify database and table still exist + assertThat(glueCatalog.databaseExists(databaseName)).isTrue(); + assertThat(glueCatalog.tableExists(new ObjectPath(databaseName, tableName))).isTrue(); + } + + /** Test drop non-empty database with cascade=true should succeed and delete all objects. */ + @Test + public void testDropNonEmptyDatabaseWithCascade() + throws DatabaseAlreadyExistException, + TableAlreadyExistException, + DatabaseNotExistException, + DatabaseNotEmptyException, + FunctionAlreadyExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String tableName = "testtable"; + String viewName = "testview"; + String functionName = "testfunction"; + + // Create database + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Create table in database + CatalogTable catalogTable = + CatalogTable.newBuilder() + .schema(Schema.newBuilder().build()) + .comment("test table") + .partitionKeys(Collections.emptyList()) + .options(Collections.emptyMap()) + .build(); + ResolvedSchema resolvedSchema = ResolvedSchema.of(); + ResolvedCatalogTable resolvedCatalogTable = + new ResolvedCatalogTable(catalogTable, resolvedSchema); + glueCatalog.createTable( + new ObjectPath(databaseName, tableName), resolvedCatalogTable, false); + + // Create view in database + CatalogView catalogView = + CatalogView.of( + Schema.newBuilder().build(), + "test view", + "SELECT * FROM " + tableName, + "SELECT * FROM " + tableName, + Collections.emptyMap()); + ResolvedCatalogView resolvedCatalogView = + new ResolvedCatalogView(catalogView, resolvedSchema); + glueCatalog.createTable(new ObjectPath(databaseName, viewName), resolvedCatalogView, false); + + // Create function in database + CatalogFunction catalogFunction = + new CatalogFunctionImpl("com.example.TestFunction", FunctionLanguage.JAVA); + glueCatalog.createFunction( + new ObjectPath(databaseName, functionName), catalogFunction, false); + + // Verify objects exist before cascade drop + assertThat(glueCatalog.databaseExists(databaseName)).isTrue(); + assertThat(glueCatalog.tableExists(new ObjectPath(databaseName, tableName))).isTrue(); + assertThat(glueCatalog.tableExists(new ObjectPath(databaseName, viewName))).isTrue(); + assertThat(glueCatalog.functionExists(new ObjectPath(databaseName, functionName))).isTrue(); + + // Act - drop database with cascade=true + glueCatalog.dropDatabase(databaseName, false, true); + + // Assert - database and all objects should be gone + assertThat(glueCatalog.databaseExists(databaseName)).isFalse(); + } + + /** Test drop empty database with cascade=false should succeed. */ + @Test + public void testDropEmptyDatabaseWithoutCascade() + throws DatabaseAlreadyExistException, + DatabaseNotExistException, + DatabaseNotEmptyException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Act - drop empty database with cascade=false + glueCatalog.dropDatabase(databaseName, false, false); + + // Assert + assertThat(glueCatalog.databaseExists(databaseName)).isFalse(); + } + + /** Test drop empty database with cascade=true should succeed. */ + @Test + public void testDropEmptyDatabaseWithCascade() + throws DatabaseAlreadyExistException, + DatabaseNotExistException, + DatabaseNotEmptyException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Act - drop empty database with cascade=true + glueCatalog.dropDatabase(databaseName, false, true); + + // Assert + assertThat(glueCatalog.databaseExists(databaseName)).isFalse(); + } + + /** Test cascade drop with only tables (no views or functions). */ + @Test + public void testDropDatabaseCascadeWithTablesOnly() + throws DatabaseAlreadyExistException, + TableAlreadyExistException, + DatabaseNotExistException, + DatabaseNotEmptyException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String tableName1 = "testtable1"; + String tableName2 = "testtable2"; + + // Create database + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Create multiple tables + CatalogTable catalogTable = + CatalogTable.newBuilder() + .schema(Schema.newBuilder().build()) + .comment("test table") + .partitionKeys(Collections.emptyList()) + .options(Collections.emptyMap()) + .build(); + ResolvedSchema resolvedSchema = ResolvedSchema.of(); + ResolvedCatalogTable resolvedCatalogTable = + new ResolvedCatalogTable(catalogTable, resolvedSchema); + + glueCatalog.createTable( + new ObjectPath(databaseName, tableName1), resolvedCatalogTable, false); + glueCatalog.createTable( + new ObjectPath(databaseName, tableName2), resolvedCatalogTable, false); + + // Verify tables exist + assertThat(glueCatalog.tableExists(new ObjectPath(databaseName, tableName1))).isTrue(); + assertThat(glueCatalog.tableExists(new ObjectPath(databaseName, tableName2))).isTrue(); + + // Act - drop database with cascade + glueCatalog.dropDatabase(databaseName, false, true); + + // Assert + assertThat(glueCatalog.databaseExists(databaseName)).isFalse(); + } + + // ------------------------------------------------------------------------- + // Table Operations Tests + // ------------------------------------------------------------------------- + + /** Test create table. */ + @Test + public void testCreateTable() + throws CatalogException, + DatabaseAlreadyExistException, + TableAlreadyExistException, + DatabaseNotExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String tableName = "testtable"; + + CatalogTable catalogTable = + CatalogTable.newBuilder() + .schema(Schema.newBuilder().build()) + .comment("test table") + .partitionKeys(Collections.emptyList()) + .options(Collections.emptyMap()) + .build(); + ResolvedSchema resolvedSchema = ResolvedSchema.of(); + ResolvedCatalogTable resolvedCatalogTable = + new ResolvedCatalogTable(catalogTable, resolvedSchema); + + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Act + glueCatalog.createTable( + new ObjectPath(databaseName, tableName), resolvedCatalogTable, false); + + // Assert + assertThat(glueTableOperations.glueTableExists(databaseName, tableName)).isTrue(); + } + + /** Test create table with ifNotExists=true. */ + @Test + public void testCreateTableIfNotExists() + throws DatabaseAlreadyExistException, + TableAlreadyExistException, + DatabaseNotExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String tableName = "testtable"; + + CatalogTable catalogTable = + CatalogTable.newBuilder() + .schema(Schema.newBuilder().build()) + .comment("test table") + .partitionKeys(Collections.emptyList()) + .options(Collections.emptyMap()) + .build(); + ResolvedSchema resolvedSchema = ResolvedSchema.of(); + ResolvedCatalogTable resolvedCatalogTable = + new ResolvedCatalogTable(catalogTable, resolvedSchema); + + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Create table first time + glueCatalog.createTable( + new ObjectPath(databaseName, tableName), resolvedCatalogTable, false); + + // Act - Create again with ifNotExists=true + assertThatCode( + () -> { + glueCatalog.createTable( + new ObjectPath(databaseName, tableName), + resolvedCatalogTable, + true); + }) + .doesNotThrowAnyException(); + } + + /** Test get table. */ + @Test + public void testGetTable() + throws CatalogException, + DatabaseAlreadyExistException, + TableAlreadyExistException, + DatabaseNotExistException, + TableNotExistException { + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String tableName = "testtable"; + + CatalogTable catalogTable = + CatalogTable.newBuilder() + .schema(Schema.newBuilder().build()) + .comment("test table") + .partitionKeys(Collections.emptyList()) + .options(Collections.emptyMap()) + .build(); + ResolvedSchema resolvedSchema = ResolvedSchema.of(); + ResolvedCatalogTable resolvedCatalogTable = + new ResolvedCatalogTable(catalogTable, resolvedSchema); + + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Act + glueCatalog.createTable( + new ObjectPath(databaseName, tableName), resolvedCatalogTable, false); + + // Act + CatalogTable retrievedTable = + (CatalogTable) glueCatalog.getTable(new ObjectPath(databaseName, tableName)); + + // Assert + assertThat(retrievedTable).isNotNull(); + } + + /** Test table not exist check. */ + @Test + public void testTableNotExist() { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String tableName = "testtable"; + + // Act & Assert + assertThatThrownBy( + () -> { + glueCatalog.getTable(new ObjectPath(databaseName, tableName)); + }) + .isInstanceOf(TableNotExistException.class); + } + + /** Test drop table operation. */ + @Test + public void testDropTable() + throws CatalogException, + DatabaseAlreadyExistException, + TableAlreadyExistException, + DatabaseNotExistException, + TableNotExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String tableName = "testtable"; + + CatalogTable catalogTable = + CatalogTable.newBuilder() + .schema(Schema.newBuilder().build()) + .comment("test table") + .partitionKeys(Collections.emptyList()) + .options(Collections.emptyMap()) + .build(); + ResolvedSchema resolvedSchema = ResolvedSchema.of(); + ResolvedCatalogTable resolvedCatalogTable = + new ResolvedCatalogTable(catalogTable, resolvedSchema); + + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Act + glueCatalog.createTable( + new ObjectPath(databaseName, tableName), resolvedCatalogTable, false); + + // Act + glueCatalog.dropTable(new ObjectPath(databaseName, tableName), false); + + // Assert + assertThat(glueTableOperations.glueTableExists(databaseName, tableName)).isFalse(); + } + + /** Test drop table with ifExists=true for non-existing table. */ + @Test + public void testDropTableWithIfExists() throws DatabaseAlreadyExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Act & Assert - should not throw exception with ifExists=true + assertThatCode( + () -> { + glueCatalog.dropTable( + new ObjectPath(databaseName, "nonExistingTable"), true); + }) + .doesNotThrowAnyException(); + } + + /** Test create table with non-existing database. */ + @Test + public void testCreateTableNonExistingDatabase() { + // Arrange + String databaseName = "nonexistingdatabase"; + String tableName = "testtable"; + + CatalogTable catalogTable = + CatalogTable.newBuilder() + .schema(Schema.newBuilder().build()) + .comment("test table") + .partitionKeys(Collections.emptyList()) + .options(Collections.emptyMap()) + .build(); + ResolvedSchema resolvedSchema = ResolvedSchema.of(); + ResolvedCatalogTable resolvedCatalogTable = + new ResolvedCatalogTable(catalogTable, resolvedSchema); + + // Act & Assert + assertThatThrownBy( + () -> { + glueCatalog.createTable( + new ObjectPath(databaseName, tableName), + resolvedCatalogTable, + false); + }) + .isInstanceOf(DatabaseNotExistException.class); + } + + /** Test listing tables for non-existing database. */ + @Test + public void testListTablesNonExistingDatabase() { + // Act & Assert + assertThatThrownBy( + () -> { + glueCatalog.listTables("nonexistingdatabase"); + }) + .isInstanceOf(DatabaseNotExistException.class); + } + + // ------------------------------------------------------------------------- + // View Operations Tests + // ------------------------------------------------------------------------- + + /** Test creating and listing views. */ + @Test + public void testCreatingAndListingViews() + throws DatabaseAlreadyExistException, + DatabaseNotExistException, + TableAlreadyExistException, + TableNotExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String viewName = "testview"; + + // Create database + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Create view + CatalogView view = + CatalogView.of( + Schema.newBuilder().build(), + "This is a test view", + "SELECT * FROM testtable", + "SELECT * FROM testtable", + Collections.emptyMap()); + + ResolvedSchema resolvedSchema = ResolvedSchema.of(); + ResolvedCatalogView resolvedView = new ResolvedCatalogView(view, resolvedSchema); + // Act + glueCatalog.createTable(new ObjectPath(databaseName, viewName), resolvedView, false); + + // Get the view + CatalogBaseTable retrievedView = + glueCatalog.getTable(new ObjectPath(databaseName, viewName)); + assertThat(retrievedView.getTableKind()).isEqualTo(CatalogBaseTable.TableKind.VIEW); + + // Assert view is listed in listViews + List<String> views = glueCatalog.listViews(databaseName); + assertThat(views).contains(viewName); + } + + /** Test listing views for non-existing database. */ + @Test + public void testListViewsNonExistingDatabase() { + // Act & Assert + assertThatThrownBy( + () -> { + glueCatalog.listViews("nonexistingdatabase"); + }) + .isInstanceOf(DatabaseNotExistException.class); + } + + // ------------------------------------------------------------------------- + // Function Operations Tests + // ------------------------------------------------------------------------- + + /** + * Regression test: the planner's function resolution probes {@code getFunction} on the + * session's current database for every SQL expression and only falls back to built-in functions + * on {@link FunctionNotExistException}. A function lookup against a non-existent database must + * therefore report FunctionNotExistException, not CatalogException - otherwise any expression + * query fails SQL validation whenever the current database does not exist in Glue. + */ + @Test + public void testGetFunctionInNonExistentDatabaseThrowsFunctionNotExist() { + ObjectPath functionPath = new ObjectPath("nonexistentdb", "somefunction"); + + assertThatThrownBy(() -> glueCatalog.getFunction(functionPath)) + .isInstanceOf(FunctionNotExistException.class); + } + + /** Test function operations. */ + @Test + public void testFunctionOperations() + throws DatabaseAlreadyExistException, + DatabaseNotExistException, + FunctionAlreadyExistException, + FunctionNotExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String functionName = "testfunction"; + ObjectPath functionPath = new ObjectPath(databaseName, functionName); + + // Create database + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Create function + CatalogFunction function = + new CatalogFunctionImpl( + "org.apache.flink.table.functions.BuiltInFunctions", FunctionLanguage.JAVA); + + // Act & Assert + // Create function + glueCatalog.createFunction(functionPath, function, false); + + // Check if function exists + assertThat(glueCatalog.functionExists(functionPath)).isTrue(); + + // List functions + List<String> functions = glueCatalog.listFunctions(databaseName); + assertThat(functions).contains(functionName.toLowerCase()); + } + + /** Test function operations with ignore flags. */ + @Test + public void testFunctionOperationsWithIgnoreFlags() + throws DatabaseAlreadyExistException, + DatabaseNotExistException, + FunctionAlreadyExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String functionName = "testfunction"; + ObjectPath functionPath = new ObjectPath(databaseName, functionName); + + // Create database + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Create function + CatalogFunction function = + new CatalogFunctionImpl( + "org.apache.flink.table.functions.BuiltInFunctions", FunctionLanguage.JAVA); + glueCatalog.createFunction(functionPath, function, false); + + // Test createFunction with ignoreIfExists=true + assertThatCode( + () -> { + glueCatalog.createFunction(functionPath, function, true); + }) + .doesNotThrowAnyException(); + } + + /** Test alter function. */ + @Test + public void testAlterFunction() + throws DatabaseAlreadyExistException, + DatabaseNotExistException, + FunctionAlreadyExistException, + FunctionNotExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String functionName = "testfunction"; + ObjectPath functionPath = new ObjectPath(databaseName, functionName); + + // Create database + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Create function + CatalogFunction function = + new CatalogFunctionImpl( + "org.apache.flink.table.functions.BuiltInFunctions", FunctionLanguage.JAVA); + glueCatalog.createFunction(functionPath, function, false); + + // Create a new function definition + CatalogFunction newFunction = + new CatalogFunctionImpl( + "org.apache.flink.table.functions.ScalarFunction", FunctionLanguage.JAVA); + + // Act + glueCatalog.alterFunction(functionPath, newFunction, false); + + // Assert + CatalogFunction retrievedFunction = glueCatalog.getFunction(functionPath); + assertThat(retrievedFunction.getClassName()).isEqualTo(newFunction.getClassName()); + } + + /** Test alter function with ignore if not exists flag. */ + @Test + public void testAlterFunctionIgnoreIfNotExists() + throws DatabaseAlreadyExistException, DatabaseNotExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Create a function definition + CatalogFunction newFunction = + new CatalogFunctionImpl( + "org.apache.flink.table.functions.ScalarFunction", FunctionLanguage.JAVA); + + // Manually handle the exception since the implementation may not be properly + // checking ignoreIfNotExists flag internally + try { + glueCatalog.alterFunction( + new ObjectPath(databaseName, "nonExistingFunction"), newFunction, true); + // If no exception is thrown, the test passes + } catch (FunctionNotExistException e) { + // We expect this exception to be thrown but it should be handled internally + // when ignoreIfNotExists=true + assertThat(e).isInstanceOf(FunctionNotExistException.class); + } + } + + /** Test drop function. */ + @Test + public void testDropFunction() + throws DatabaseAlreadyExistException, + DatabaseNotExistException, + FunctionAlreadyExistException, + FunctionNotExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + String functionName = "testfunction"; + ObjectPath functionPath = new ObjectPath(databaseName, functionName); + + // Create database + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Create function + CatalogFunction function = + new CatalogFunctionImpl( + "org.apache.flink.table.functions.BuiltInFunctions", FunctionLanguage.JAVA); + glueCatalog.createFunction(functionPath, function, false); + + // Drop function + glueCatalog.dropFunction(functionPath, false); + + // Check function no longer exists + assertThat(glueCatalog.functionExists(functionPath)).isFalse(); + } + + /** Test drop function with ignore flag. */ + @Test + public void testDropFunctionWithIgnoreFlag() + throws DatabaseAlreadyExistException, DatabaseNotExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Test dropFunction with ignoreIfNotExists=true + assertThatCode( + () -> { + glueCatalog.dropFunction( + new ObjectPath(databaseName, "nonExistingFunction"), true); + }) + .doesNotThrowAnyException(); + } + + /** Test function exists edge cases. */ + @Test + public void testFunctionExistsEdgeCases() throws DatabaseAlreadyExistException { + // Arrange + String databaseName = GlueTestClientFactory.uniqueName("testdatabase"); + CatalogDatabase catalogDatabase = new CatalogDatabaseImpl(Collections.emptyMap(), "test"); + glueCatalog.createDatabase(databaseName, catalogDatabase, false); + + // Act & Assert + // Function in non-existing database + assertThat(glueCatalog.functionExists(new ObjectPath("nonExistingDb", "testFunction"))) + .isFalse(); + } + + // ------------------------------------------------------------------------- + // Error Handling Tests + // ------------------------------------------------------------------------- + + /** Test null parameter handling. */ + @Test + public void testNullParameterHandling() { + // Act & Assert + assertThatThrownBy( + () -> { + glueCatalog.createTable(null, null, false); + }) + .isInstanceOf(NullPointerException.class); + + assertThatThrownBy( + () -> { + glueCatalog.createTable(new ObjectPath("db", "table"), null, false); + }) + .isInstanceOf(NullPointerException.class); + } + + @Test Review Comment: Added in eee3aa0 - `GlueCatalogSchemaFidelityTest.testWatermarkPrimaryKeyAndNonPhysicalColumnsRoundTrip` does a full create-then-read round-trip (mixed-case names, computed + metadata columns, watermark, primary key) asserting column order, kinds, watermark, PK, and that only physical columns land as Glue columns. SQL-path round-trips were also added to the moto ITCase and the e2e module. Worth it: these tests caught two real bugs (NOT NULL and TIMESTAMP(3) precision do not survive Glue's type strings - both now persisted and restored). ########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/util/GlueFunctionsUtil.java: ########## @@ -0,0 +1,96 @@ +/* + * 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.util; + +import org.apache.flink.table.catalog.CatalogFunction; +import org.apache.flink.table.catalog.FunctionLanguage; +import org.apache.flink.table.catalog.exceptions.CatalogException; + +import software.amazon.awssdk.services.glue.model.UserDefinedFunction; + +import java.util.Arrays; +import java.util.stream.Collectors; + +/** + * Utility class for handling Functions in AWS Glue Catalog integration. Provides methods for + * converting between Flink and Glue function representation. + */ +public class GlueFunctionsUtil { + + /** + * Extracts the class name from a Glue UserDefinedFunction. + * + * @param udf The Glue UserDefinedFunction + * @return The extracted class name + */ + public static String getCatalogFunctionClassName(final UserDefinedFunction udf) { + String[] splitName = udf.className().split(GlueCatalogConstants.DEFAULT_SEPARATOR); + return splitName[splitName.length - 1]; + } + + /** + * Determines the function language from a Glue UserDefinedFunction. + * + * @param glueFunction The Glue UserDefinedFunction + * @return The corresponding Flink FunctionLanguage + * @throws CatalogException if the function language cannot be determined + */ + public static FunctionLanguage getFunctionalLanguage(final UserDefinedFunction glueFunction) { + if (glueFunction.className().startsWith(GlueCatalogConstants.FLINK_JAVA_FUNCTION_PREFIX)) { + return FunctionLanguage.JAVA; + } else if (glueFunction + .className() + .startsWith(GlueCatalogConstants.FLINK_PYTHON_FUNCTION_PREFIX)) { + return FunctionLanguage.PYTHON; + } else if (glueFunction + .className() + .startsWith(GlueCatalogConstants.FLINK_SCALA_FUNCTION_PREFIX)) { + return FunctionLanguage.SCALA; + } else { + throw new CatalogException( Review Comment: Flink's `FunctionLanguage` enum has exactly these three, so for Flink-created functions the three prefixes are exhaustive. The real gap was functions created by other engines (Hive/Spark UDFs have no Flink prefix): those used to fail the whole lookup with a CatalogException. Since 56f13dd they fall back to JAVA with a warning - both are Java classes under the hood, and whether they are loadable depends on the user's classpath, which is the same behavior Hive-catalog users get. Covered in the new GlueFunctionsUtilTest. ########## flink-catalog-aws/flink-catalog-aws-glue/src/main/java/org/apache/flink/table/catalog/glue/util/GlueTableUtils.java: ########## @@ -0,0 +1,218 @@ +/* + * 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.util; + +import org.apache.flink.table.api.Schema; +import org.apache.flink.table.catalog.ObjectPath; +import org.apache.flink.table.types.DataType; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.services.glue.model.Column; +import software.amazon.awssdk.services.glue.model.StorageDescriptor; +import software.amazon.awssdk.services.glue.model.Table; + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +/** + * Utility class for working with Glue tables, including transforming Glue-specific metadata into + * Flink-compatible objects. + */ +public class GlueTableUtils { + + /** Logger for logging Glue table operations. */ + private static final Logger LOG = LoggerFactory.getLogger(GlueTableUtils.class); + + /** Glue type converter for type conversions between Flink and Glue types. */ + private final GlueTypeConverter glueTypeConverter; + + /** + * Constructor to initialize GlueTableUtils with a GlueTypeConverter. + * + * @param glueTypeConverter The GlueTypeConverter instance for type mapping. + */ + public GlueTableUtils(GlueTypeConverter glueTypeConverter) { + this.glueTypeConverter = glueTypeConverter; + } + + /** + * Builds a Glue StorageDescriptor from the given table properties, columns, and location. + * + * @param tableProperties Table properties for the Glue table. + * @param glueColumns Columns to be included in the StorageDescriptor. + * @param tableLocation Location of the Glue table. + * @return A newly built StorageDescriptor object. + */ + public StorageDescriptor buildStorageDescriptor( + Map<String, String> tableProperties, List<Column> glueColumns, String tableLocation) { + + return StorageDescriptor.builder().columns(glueColumns).location(tableLocation).build(); + } + + /** + * Extracts the table location based on the table properties and the table path. First, it + * checks for a location key from the connector registry. If no such key is found, it uses a + * default path based on the table path. + * + * @param tableProperties Table properties containing the connector and location. + * @param tablePath The Flink ObjectPath representing the table. + * @return The location of the Glue table. + */ + public String extractTableLocation(Map<String, String> tableProperties, ObjectPath tablePath) { + String connectorType = tableProperties.get("connector"); + if (connectorType != null) { + String locationKey = ConnectorRegistry.getLocationKey(connectorType); + if (locationKey != null && tableProperties.containsKey(locationKey)) { + String location = tableProperties.get(locationKey); + return location; + } + } + + String defaultLocation = + tablePath.getDatabaseName() + "/tables/" + tablePath.getObjectName(); + return defaultLocation; + } + + /** + * Converts a Flink column to a Glue column. The column's data type is converted using the + * GlueTypeConverter. + * + * @param flinkColumn The Flink column to be converted. + * @return The corresponding Glue column. + */ + public Column mapFlinkColumnToGlueColumn(org.apache.flink.table.catalog.Column flinkColumn) { Review Comment: You were right, and it was worse than watermarks and primary keys: computed and metadata columns were silently persisted as physical columns (schema corruption on read-back). Fixed in eee3aa0: physical columns stay native Glue columns (for cross-engine readability), and everything Glue cannot represent - watermarks, primary key, computed columns, metadata columns, declared NOT NULL constraints, and lossy data types like TIMESTAMP(3) which Glue's type strings cannot carry - round-trips through `flink.schema.*` table parameters, restored at the declared positions on read and hidden from the exposed options. Verified at all three tiers: unit round-trip, moto SQL path, and a SHOW CREATE TABLE round-trip against real AWS Glue in the e2e module. -- 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]
