Optimize the logic of jdbcCatalog's getTable method.

hdygxsj · hdygxsj · commit 0e4808609206 · 2025-03-12T20:25:16.000+08:00
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalogFactory.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalogFactory.java
@@ -19,6 +19,7 @@
 
 package org.apache.gravitino.flink.connector.catalog;
 
+import java.util.Map;
 import org.apache.flink.table.factories.CatalogFactory;
 import org.apache.gravitino.Catalog;
 import org.apache.gravitino.flink.connector.PartitionConverter;
@@ -45,7 +46,7 @@ public interface BaseCatalogFactory extends CatalogFactory {
    *
    * @return The requested property converter.
    */
-  PropertiesConverter propertiesConverter(Context context);
+  PropertiesConverter propertiesConverter(Map<String, String> catalogOptions);
 
   /**
    * Define partition converter.
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFactory.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalogFactory.java
@@ -22,6 +22,7 @@
 import static org.apache.gravitino.flink.connector.hive.GravitinoHiveCatalogFactoryOptions.IDENTIFIER;
 
 import com.google.common.collect.ImmutableSet;
+import java.util.Map;
 import java.util.Set;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.table.catalog.Catalog;
@@ -63,7 +64,7 @@ public Catalog createCatalog(Context context) {
     return new GravitinoHiveCatalog(
         context.getName(),
         helper.getOptions().get(HiveCatalogFactoryOptions.DEFAULT_DATABASE),
-        propertiesConverter(context),
+        propertiesConverter(context.getOptions()),
         partitionConverter(),
         hiveConf,
         helper.getOptions().get(HiveCatalogFactoryOptions.HIVE_VERSION));
@@ -113,7 +114,7 @@ public org.apache.gravitino.Catalog.Type gravitinoCatalogType() {
    * @return The requested property converter.
    */
   @Override
-  public PropertiesConverter propertiesConverter(Context context) {
+  public PropertiesConverter propertiesConverter(Map<String, String> catalogOptions) {
     return HivePropertiesConverter.INSTANCE;
   }
 
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/hive/HivePropertiesConverter.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/hive/HivePropertiesConverter.java
@@ -22,6 +22,7 @@
 import com.google.common.collect.ImmutableMap;
 import java.util.Map;
 import java.util.stream.Collectors;
+import org.apache.flink.table.catalog.ObjectPath;
 import org.apache.gravitino.catalog.hive.HiveConstants;
 import org.apache.gravitino.flink.connector.PropertiesConverter;
 import org.apache.hadoop.hive.conf.HiveConf;
@@ -47,7 +48,8 @@ public String transformPropertyToFlinkCatalog(String configKey) {
   }
 
   @Override
-  public Map<String, String> toFlinkTableProperties(Map<String, String> gravitinoProperties) {
+  public Map<String, String> toFlinkTableProperties(
+      Map<String, String> gravitinoProperties, ObjectPath tablePath) {
     Map<String, String> properties =
         gravitinoProperties.entrySet().stream()
             .collect(
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactory.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/iceberg/GravitinoIcebergCatalogFactory.java
@@ -19,6 +19,7 @@
 package org.apache.gravitino.flink.connector.iceberg;
 
 import java.util.Collections;
+import java.util.Map;
 import java.util.Set;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.table.catalog.Catalog;
@@ -38,7 +39,7 @@ public Catalog createCatalog(Context context) {
     return new GravitinoIcebergCatalog(
         context.getName(),
         helper.getOptions().get(GravitinoIcebergCatalogFactoryOptions.DEFAULT_DATABASE),
-        propertiesConverter(context),
+        propertiesConverter(context.getOptions()),
         partitionConverter(),
         context.getOptions());
   }
@@ -84,7 +85,7 @@ public org.apache.gravitino.Catalog.Type gravitinoCatalogType() {
    * @return
    */
   @Override
-  public PropertiesConverter propertiesConverter(Context context) {
+  public PropertiesConverter propertiesConverter(Map<String, String> catalogOptions) {
     return IcebergPropertiesConverter.INSTANCE;
   }
 
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFactory.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalogFactory.java
@@ -46,7 +46,7 @@ public org.apache.flink.table.catalog.Catalog createCatalog(Context context) {
         defaultDatabase,
         GravitinoJdbcCatalogFactoryOptions.DEFAULT_DATABASE.key() + " should not be null.");
     return new GravitinoJdbcCatalog(
-        context, defaultDatabase, propertiesConverter(), partitionConverter());
+        context, defaultDatabase, propertiesConverter(context.getOptions()), partitionConverter());
   }
 
   @Override
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoMysqlJdbcCatalogFactory.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoMysqlJdbcCatalogFactory.java
@@ -19,6 +19,7 @@
 
 package org.apache.gravitino.flink.connector.jdbc;
 
+import java.util.Map;
 import org.apache.gravitino.flink.connector.PropertiesConverter;
 
 public class GravitinoMysqlJdbcCatalogFactory extends GravitinoJdbcCatalogFactory {
@@ -29,8 +30,8 @@ public String gravitinoCatalogProvider() {
   }
 
   @Override
-  public PropertiesConverter propertiesConverter(Context context) {
-    return new MysqlPropertiesConverter(context);
+  public PropertiesConverter propertiesConverter(Map<String, String> catalogOptions) {
+    return new MysqlPropertiesConverter(catalogOptions);
   }
 
   @Override
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoPostgresJdbcCatalogFactory.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoPostgresJdbcCatalogFactory.java
@@ -19,6 +19,7 @@
 
 package org.apache.gravitino.flink.connector.jdbc;
 
+import java.util.Map;
 import org.apache.gravitino.flink.connector.PropertiesConverter;
 
 public class GravitinoPostgresJdbcCatalogFactory extends GravitinoJdbcCatalogFactory {
@@ -29,8 +30,8 @@ public String gravitinoCatalogProvider() {
   }
 
   @Override
-  public PropertiesConverter propertiesConverter(Context context) {
-    return new PostgresqlPropertiesConverter(context);
+  public PropertiesConverter propertiesConverter(Map<String, String> catalogOptions) {
+    return new PostgresqlPropertiesConverter(catalogOptions);
   }
 
   @Override
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConverter.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/JdbcPropertiesConverter.java
@@ -23,15 +23,14 @@
 import java.util.Map;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.factories.CatalogFactory;
 import org.apache.gravitino.flink.connector.PropertiesConverter;
 
 public abstract class JdbcPropertiesConverter implements PropertiesConverter {
 
-  private final CatalogFactory.Context context;
+  private final Map<String, String> catalogOptions;
 
-  protected JdbcPropertiesConverter(CatalogFactory.Context context) {
-    this.context = context;
+  protected JdbcPropertiesConverter(Map<String, String> catalogOptions) {
+    this.catalogOptions = catalogOptions;
   }
 
   @Override
@@ -55,7 +54,6 @@ public String transformPropertyToFlinkCatalog(String configKey) {
   @Override
   public Map<String, String> toFlinkTableProperties(
       Map<String, String> gravitinoProperties, ObjectPath tablePath) {
-    Map<String, String> catalogOptions = context.getOptions();
     Map<String, String> tableOptions = new HashMap<>();
     tableOptions.put(
         "url",
@@ -69,4 +67,8 @@ public Map<String, String> toFlinkTableProperties(
   }
 
   protected abstract String driverName();
+
+  public Map<String, String> getCatalogOptions() {
+    return catalogOptions;
+  }
 }
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/MysqlPropertiesConverter.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/MysqlPropertiesConverter.java
@@ -19,12 +19,12 @@
 
 package org.apache.gravitino.flink.connector.jdbc;
 
-import org.apache.flink.table.factories.CatalogFactory;
+import java.util.Map;
 
 public class MysqlPropertiesConverter extends JdbcPropertiesConverter {
 
-  protected MysqlPropertiesConverter(CatalogFactory.Context context) {
-    super(context);
+  protected MysqlPropertiesConverter(Map<String, String> catalogOptions) {
+    super(catalogOptions);
   }
 
   @Override
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/PostgresqlPropertiesConverter.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/PostgresqlPropertiesConverter.java
@@ -19,12 +19,12 @@
 
 package org.apache.gravitino.flink.connector.jdbc;
 
-import org.apache.flink.table.factories.CatalogFactory;
+import java.util.Map;
 
 public class PostgresqlPropertiesConverter extends JdbcPropertiesConverter {
 
-  protected PostgresqlPropertiesConverter(CatalogFactory.Context context) {
-    super(context);
+  protected PostgresqlPropertiesConverter(Map<String, String> catalogOptions) {
+    super(catalogOptions);
   }
 
   @Override
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFactory.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalogFactory.java
@@ -20,6 +20,7 @@
 package org.apache.gravitino.flink.connector.paimon;
 
 import java.util.Collections;
+import java.util.Map;
 import java.util.Set;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.table.catalog.Catalog;
@@ -43,7 +44,7 @@ public Catalog createCatalog(Context context) {
     String defaultDatabase =
         helper.getOptions().get(GravitinoPaimonCatalogFactoryOptions.DEFAULT_DATABASE);
     return new GravitinoPaimonCatalog(
-        context, defaultDatabase, propertiesConverter(context), partitionConverter());
+        context, defaultDatabase, propertiesConverter(context.getOptions()), partitionConverter());
   }
 
   @Override
@@ -72,7 +73,7 @@ public org.apache.gravitino.Catalog.Type gravitinoCatalogType() {
   }
 
   @Override
-  public PropertiesConverter propertiesConverter(Context context) {
+  public PropertiesConverter propertiesConverter(Map<String, String> catalogOptions) {
     return PaimonPropertiesConverter.INSTANCE;
   }
 
diff --git a/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/jdbc/AbstractJdbcPropertiesConverter.java b/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/jdbc/AbstractJdbcPropertiesConverter.java
@@ -33,21 +33,24 @@ public abstract class AbstractJdbcPropertiesConverter {
   String url = "testUrl";
   String defaultDatabase = "test";
 
-  protected abstract JdbcPropertiesConverter getConverter();
+  Map<String, String> catalogProperties =
+      ImmutableMap.of(
+          JdbcPropertiesConstants.GRAVITINO_JDBC_USER,
+          username,
+          JdbcPropertiesConstants.GRAVITINO_JDBC_PASSWORD,
+          password,
+          JdbcPropertiesConstants.GRAVITINO_JDBC_URL,
+          url,
+          JdbcPropertiesConstants.GRAVITINO_JDBC_DEFAULT_DATABASE,
+          defaultDatabase);
+
+  protected abstract JdbcPropertiesConverter getConverter(Map<String, String> catalogOptions);
 
   @Test
   public void testToPaimonFileSystemCatalog() {
-    Map<String, String> catalogProperties =
-        ImmutableMap.of(
-            JdbcPropertiesConstants.GRAVITINO_JDBC_USER,
-            username,
-            JdbcPropertiesConstants.GRAVITINO_JDBC_PASSWORD,
-            password,
-            JdbcPropertiesConstants.GRAVITINO_JDBC_URL,
-            url,
-            JdbcPropertiesConstants.GRAVITINO_JDBC_DEFAULT_DATABASE,
-            defaultDatabase);
-    Map<String, String> properties = getConverter().toFlinkCatalogProperties(catalogProperties);
+
+    Map<String, String> properties =
+        getConverter(catalogProperties).toFlinkCatalogProperties(catalogProperties);
     Assertions.assertEquals(username, properties.get(JdbcPropertiesConstants.FLINK_JDBC_USER));
     Assertions.assertEquals(password, properties.get(JdbcPropertiesConstants.FLINK_JDBC_PASSWORD));
     Assertions.assertEquals(url, properties.get(JdbcPropertiesConstants.FLINK_JDBC_URL));
@@ -68,7 +71,8 @@ public void testToGravitinoCatalogProperties() {
                 url,
                 JdbcPropertiesConstants.FLINK_JDBC_DEFAULT_DATABASE,
                 defaultDatabase));
-    Map<String, String> properties = getConverter().toGravitinoCatalogProperties(configuration);
+    Map<String, String> properties =
+        getConverter(catalogProperties).toGravitinoCatalogProperties(configuration);
 
     Assertions.assertEquals(username, properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_USER));
     Assertions.assertEquals(
@@ -77,6 +81,7 @@ public void testToGravitinoCatalogProperties() {
     Assertions.assertEquals(
         defaultDatabase, properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_DEFAULT_DATABASE));
     Assertions.assertEquals(
-        getConverter().driverName(), properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_DRIVER));
+        getConverter(catalogProperties).driverName(),
+        properties.get(JdbcPropertiesConstants.GRAVITINO_JDBC_DRIVER));
   }
 }
diff --git a/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestMysqlPropertiesConverter.java b/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestMysqlPropertiesConverter.java
@@ -19,10 +19,12 @@
 
 package org.apache.gravitino.flink.connector.jdbc;
 
+import java.util.Map;
+
 public class TestMysqlPropertiesConverter extends AbstractJdbcPropertiesConverter {
 
   @Override
-  protected JdbcPropertiesConverter getConverter() {
-    return MysqlPropertiesConverter.INSTANCE;
+  protected JdbcPropertiesConverter getConverter(Map<String, String> catalogOptions) {
+    return new MysqlPropertiesConverter(catalogOptions);
   }
 }
diff --git a/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestPostgresqlPropertiesConverter.java b/flink-connector/flink/src/test/java/org/apache/gravitino/flink/connector/jdbc/TestPostgresqlPropertiesConverter.java
@@ -19,10 +19,12 @@
 
 package org.apache.gravitino.flink.connector.jdbc;
 
+import java.util.Map;
+
 public class TestPostgresqlPropertiesConverter extends AbstractJdbcPropertiesConverter {
 
   @Override
-  protected JdbcPropertiesConverter getConverter() {
-    return PostgresqlPropertiesConverter.INSTANCE;
+  protected JdbcPropertiesConverter getConverter(Map<String, String> catalogOptions) {
+    return new PostgresqlPropertiesConverter(catalogOptions);
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ public org.apache.flink.table.catalog.Catalog createCatalog(Context context) {`
`46`	`46`	`defaultDatabase,`
`47`	`47`	`GravitinoJdbcCatalogFactoryOptions.DEFAULT_DATABASE.key() + " should not be null.");`
`48`	`48`	`return new GravitinoJdbcCatalog(`
`49`		`- context, defaultDatabase, propertiesConverter(), partitionConverter());`
	`49`	`+ context, defaultDatabase, propertiesConverter(context.getOptions()), partitionConverter());`
`50`	`50`	`}`
`51`	`51`
`52`	`52`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -19,6 +19,7 @@`
`19`	`19`
`20`	`20`	`package org.apache.gravitino.flink.connector.jdbc;`
`21`	`21`
	`22`	`+import java.util.Map;`
`22`	`23`	`import org.apache.gravitino.flink.connector.PropertiesConverter;`
`23`	`24`
`24`	`25`	`public class GravitinoMysqlJdbcCatalogFactory extends GravitinoJdbcCatalogFactory {`
`@@ -29,8 +30,8 @@ public String gravitinoCatalogProvider() {`
`29`	`30`	`}`
`30`	`31`
`31`	`32`	`@Override`
`32`		`- public PropertiesConverter propertiesConverter(Context context) {`
`33`		`- return new MysqlPropertiesConverter(context);`
	`33`	`+ public PropertiesConverter propertiesConverter(Map<String, String> catalogOptions) {`
	`34`	`+ return new MysqlPropertiesConverter(catalogOptions);`
`34`	`35`	`}`
`35`	`36`
`36`	`37`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -19,10 +19,12 @@`
`19`	`19`
`20`	`20`	`package org.apache.gravitino.flink.connector.jdbc;`
`21`	`21`
	`22`	`+import java.util.Map;`
	`23`	`+`
`22`	`24`	`public class TestMysqlPropertiesConverter extends AbstractJdbcPropertiesConverter {`
`23`	`25`
`24`	`26`	`@Override`
`25`		`- protected JdbcPropertiesConverter getConverter() {`
`26`		`- return MysqlPropertiesConverter.INSTANCE;`
	`27`	`+ protected JdbcPropertiesConverter getConverter(Map<String, String> catalogOptions) {`
	`28`	`+ return new MysqlPropertiesConverter(catalogOptions);`
`27`	`29`	`}`
`28`	`30`	`}`