[flink] Remove useless classloader in FlinkCatalog

Author: JingsongLi

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalog.java

@@ -93,6 +93,7 @@
 import org.apache.flink.table.catalog.TableChange.ModifyWatermark;
 import org.apache.flink.table.catalog.TableChange.ResetOption;
 import org.apache.flink.table.catalog.TableChange.SetOption;
+import org.apache.flink.table.catalog.UniqueConstraint;
 import org.apache.flink.table.catalog.WatermarkSpec;
 import org.apache.flink.table.catalog.exceptions.CatalogException;
 import org.apache.flink.table.catalog.exceptions.DatabaseAlreadyExistException;
@@ -179,23 +180,16 @@ public class FlinkCatalog extends AbstractCatalog {
 
     private static final Logger LOG = LoggerFactory.getLogger(FlinkCatalog.class);
 
-    private final ClassLoader classLoader;
     private final Catalog catalog;
     private final String name;
 
     private final boolean disableCreateTableInDefaultDatabase;
 
-    public FlinkCatalog(
-            Catalog catalog,
-            String name,
-            String defaultDatabase,
-            ClassLoader classLoader,
-            Options options) {
+    public FlinkCatalog(Catalog catalog, String name, String defaultDatabase, Options options) {
         super(name, defaultDatabase);
         LOG.info("Creating Flink catalog: metastore={}", options.get(CatalogOptions.METASTORE));
         this.catalog = catalog;
         this.name = name;
-        this.classLoader = classLoader;
         this.disableCreateTableInDefaultDatabase = options.get(DISABLE_CREATE_TABLE_IN_DEFAULT_DB);
         if (!disableCreateTableInDefaultDatabase) {
             try {
@@ -248,9 +242,9 @@ public void createDatabase(String name, CatalogDatabase database, boolean ignore
         Map<String, String> properties;
         if (database != null) {
             properties = new HashMap<>(database.getProperties());
-            if (database.getDescription().isPresent()
-                    && !database.getDescription().get().equals("")) {
-                properties.put(COMMENT_PROP, database.getDescription().get());
+            Optional<String> description = database.getDescription();
+            if (description.isPresent() && !description.get().isEmpty()) {
+                properties.put(COMMENT_PROP, description.get());
             }
         } else {
             properties = Collections.emptyMap();
@@ -941,7 +935,7 @@ private CatalogBaseTable toCatalogTable(Table table) {
         }
 
         // add primary keys
-        if (table.primaryKeys().size() > 0) {
+        if (!table.primaryKeys().isEmpty()) {
             builder.primaryKey(table.primaryKeys());
         }
 
@@ -1012,6 +1006,7 @@ private byte[] decodeRefreshHandlerBytes(String refreshHandlerBytes) {
     }
 
     public static Schema fromCatalogTable(CatalogBaseTable catalogTable) {
+        @SuppressWarnings("unchecked")
         ResolvedSchema schema =
                 ((ResolvedCatalogBaseTable<CatalogBaseTable>) catalogTable).getResolvedSchema();
         RowType rowType = (RowType) schema.toPhysicalRowDataType().getLogicalType();
@@ -1037,7 +1032,7 @@ public static Schema fromCatalogTable(CatalogBaseTable catalogTable) {
                         .options(options)
                         .primaryKey(
                                 schema.getPrimaryKey()
-                                        .map(pk -> pk.getColumns())
+                                        .map(UniqueConstraint::getColumns)
                                         .orElse(Collections.emptyList()))
                         .partitionKeys(getPartitionKeys(catalogTable));
         Map<String, String> columnComments = getColumnComments(catalogTable);
@@ -1185,7 +1180,7 @@ private Table getPaimonTable(ObjectPath tablePath) throws TableNotExistException
     private List<PartitionEntry> getPartitionEntries(
             Table table, ObjectPath tablePath, @Nullable CatalogPartitionSpec partitionSpec)
             throws TableNotPartitionedException {
-        if (table.partitionKeys() == null || table.partitionKeys().size() == 0) {
+        if (table.partitionKeys() == null || table.partitionKeys().isEmpty()) {
             throw new TableNotPartitionedException(getName(), tablePath);
         }
 
@@ -1267,7 +1262,7 @@ public final boolean partitionExists(ObjectPath tablePath, CatalogPartitionSpec
             throws CatalogException {
         try {
             List<CatalogPartitionSpec> partitionSpecs = getPartitionSpecs(tablePath, partitionSpec);
-            return partitionSpecs.size() > 0;
+            return !partitionSpecs.isEmpty();
         } catch (TableNotPartitionedException | TableNotExistException e) {
             throw new CatalogException(e);
         }
@@ -1444,9 +1439,7 @@ public final void alterFunction(
             }
         } catch (Catalog.FunctionNotExistException e) {
             throw new FunctionNotExistException(getName(), functionPath);
-        } catch (Catalog.DefinitionAlreadyExistException e) {
-            throw new RuntimeException(e);
-        } catch (Catalog.DefinitionNotExistException e) {
+        } catch (Catalog.DefinitionAlreadyExistException | Catalog.DefinitionNotExistException e) {
             throw new RuntimeException(e);
         }
     }

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkCatalogFactory.java

@@ -63,17 +63,11 @@ public static FlinkCatalog createCatalog(
                 CatalogFactory.createCatalog(context, classLoader),
                 catalogName,
                 context.options().get(DEFAULT_DATABASE),
-                classLoader,
                 context.options());
     }
 
     public static FlinkCatalog createCatalog(String catalogName, Catalog catalog, Options options) {
-        return new FlinkCatalog(
-                catalog,
-                catalogName,
-                Catalog.DEFAULT_DATABASE,
-                FlinkCatalogFactory.class.getClassLoader(),
-                options);
+        return new FlinkCatalog(catalog, catalogName, Catalog.DEFAULT_DATABASE, options);
     }
 
     public static Catalog createPaimonCatalog(Options catalogOptions) {

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkGenericCatalogFactory.java

@@ -96,7 +96,6 @@ public static FlinkGenericCatalog createCatalog(
                                 CatalogContext.create(options, new FlinkFileIOLoader()), cl),
                         name,
                         options.get(DEFAULT_DATABASE),
-                        cl,
                         options);
 
         return new FlinkGenericCatalog(paimon, flinkCatalog);