Optimize the logic of jdbcCatalog's getTable method.

hdygxsj · hdygxsj · commit deac3d4db89f · 2025-03-12T20:07:27.000+08:00
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/PropertiesConverter.java
@@ -23,6 +23,7 @@
 import java.util.Map;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.catalog.CommonCatalogOptions;
+import org.apache.flink.table.catalog.ObjectPath;
 
 /**
  * PropertiesConverter is used to convert properties between Flink properties and Apache Gravitino
@@ -136,7 +137,8 @@ default Map<String, String> toFlinkDatabaseProperties(Map<String, String> gravit
    * @param gravitinoProperties The table properties provided by Gravitino.
    * @return The table properties for the Flink connector.
    */
-  default Map<String, String> toFlinkTableProperties(Map<String, String> gravitinoProperties) {
+  default Map<String, String> toFlinkTableProperties(
+      Map<String, String> gravitinoProperties, ObjectPath tablePath) {
     return gravitinoProperties;
   }
 
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
@@ -211,7 +211,7 @@ public CatalogBaseTable getTable(ObjectPath tablePath)
           catalog()
               .asTableCatalog()
               .loadTable(NameIdentifier.of(tablePath.getDatabaseName(), tablePath.getObjectName()));
-      return toFlinkTable(table);
+      return toFlinkTable(table, tablePath);
     } catch (NoSuchTableException e) {
       throw new TableNotExistException(catalogName(), tablePath, e);
     } catch (Exception e) {
@@ -550,7 +550,7 @@ public void alterPartitionColumnStatistics(
     throw new UnsupportedOperationException();
   }
 
-  protected CatalogBaseTable toFlinkTable(Table table) {
+  protected CatalogBaseTable toFlinkTable(Table table, ObjectPath tablePath) {
     org.apache.flink.table.api.Schema.Builder builder =
         org.apache.flink.table.api.Schema.newBuilder();
     for (Column column : table.columns()) {
@@ -562,7 +562,7 @@ protected CatalogBaseTable toFlinkTable(Table table) {
     Optional<List<String>> flinkPrimaryKey = getFlinkPrimaryKey(table);
     flinkPrimaryKey.ifPresent(builder::primaryKey);
     Map<String, String> flinkTableProperties =
-        propertiesConverter.toFlinkTableProperties(table.properties());
+        propertiesConverter.toFlinkTableProperties(table.properties(), tablePath);
     List<String> partitionKeys = partitionConverter.toFlinkPartitionKeys(table.partitioning());
     return CatalogTable.of(builder.build(), table.comment(), partitionKeys, flinkTableProperties);
   }
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
@@ -45,7 +45,7 @@ public interface BaseCatalogFactory extends CatalogFactory {
    *
    * @return The requested property converter.
    */
-  PropertiesConverter propertiesConverter();
+  PropertiesConverter propertiesConverter(Context context);
 
   /**
    * 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
@@ -63,7 +63,7 @@ public Catalog createCatalog(Context context) {
     return new GravitinoHiveCatalog(
         context.getName(),
         helper.getOptions().get(HiveCatalogFactoryOptions.DEFAULT_DATABASE),
-        propertiesConverter(),
+        propertiesConverter(context),
         partitionConverter(),
         hiveConf,
         helper.getOptions().get(HiveCatalogFactoryOptions.HIVE_VERSION));
@@ -113,7 +113,7 @@ public org.apache.gravitino.Catalog.Type gravitinoCatalogType() {
    * @return The requested property converter.
    */
   @Override
-  public PropertiesConverter propertiesConverter() {
+  public PropertiesConverter propertiesConverter(Context context) {
     return HivePropertiesConverter.INSTANCE;
   }
 
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
@@ -38,7 +38,7 @@ public Catalog createCatalog(Context context) {
     return new GravitinoIcebergCatalog(
         context.getName(),
         helper.getOptions().get(GravitinoIcebergCatalogFactoryOptions.DEFAULT_DATABASE),
-        propertiesConverter(),
+        propertiesConverter(context),
         partitionConverter(),
         context.getOptions());
   }
@@ -84,7 +84,7 @@ public org.apache.gravitino.Catalog.Type gravitinoCatalogType() {
    * @return
    */
   @Override
-  public PropertiesConverter propertiesConverter() {
+  public PropertiesConverter propertiesConverter(Context context) {
     return IcebergPropertiesConverter.INSTANCE;
   }
 
diff --git a/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalog.java b/flink-connector/flink/src/main/java/org/apache/gravitino/flink/connector/jdbc/GravitinoJdbcCatalog.java
@@ -19,16 +19,11 @@
 
 package org.apache.gravitino.flink.connector.jdbc;
 
-import java.util.Map;
 import java.util.Optional;
 import org.apache.flink.connector.jdbc.catalog.JdbcCatalog;
 import org.apache.flink.connector.jdbc.catalog.factory.JdbcCatalogFactory;
 import org.apache.flink.connector.jdbc.table.JdbcDynamicTableFactory;
 import org.apache.flink.table.catalog.AbstractCatalog;
-import org.apache.flink.table.catalog.CatalogBaseTable;
-import org.apache.flink.table.catalog.ObjectPath;
-import org.apache.flink.table.catalog.exceptions.CatalogException;
-import org.apache.flink.table.catalog.exceptions.TableNotExistException;
 import org.apache.flink.table.factories.CatalogFactory;
 import org.apache.flink.table.factories.Factory;
 import org.apache.gravitino.flink.connector.PartitionConverter;
@@ -65,22 +60,4 @@ protected AbstractCatalog realCatalog() {
   public Optional<Factory> getFactory() {
     return Optional.of(new JdbcDynamicTableFactory());
   }
-
-  @Override
-  public CatalogBaseTable getTable(ObjectPath tablePath)
-      throws TableNotExistException, CatalogException {
-    CatalogBaseTable table = super.getTable(tablePath);
-    Map<String, String> contextOptions = context.getOptions();
-    Map<String, String> tableOptions = table.getOptions();
-    tableOptions.remove("engine");
-    tableOptions.put(
-        "url",
-        contextOptions.get(JdbcPropertiesConstants.FLINK_JDBC_URL)
-            + "/"
-            + tablePath.getDatabaseName());
-    tableOptions.put("table-name", tablePath.getObjectName());
-    tableOptions.put("username", contextOptions.get(JdbcPropertiesConstants.FLINK_JDBC_USER));
-    tableOptions.put("password", contextOptions.get(JdbcPropertiesConstants.FLINK_JDBC_PASSWORD));
-    return table;
-  }
 }
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
@@ -29,8 +29,8 @@ public String gravitinoCatalogProvider() {
   }
 
   @Override
-  public PropertiesConverter propertiesConverter() {
-    return MysqlPropertiesConverter.INSTANCE;
+  public PropertiesConverter propertiesConverter(Context context) {
+    return new MysqlPropertiesConverter(context);
   }
 
   @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
@@ -29,8 +29,8 @@ public String gravitinoCatalogProvider() {
   }
 
   @Override
-  public PropertiesConverter propertiesConverter() {
-    return PostgresqlPropertiesConverter.INSTANCE;
+  public PropertiesConverter propertiesConverter(Context context) {
+    return new PostgresqlPropertiesConverter(context);
   }
 
   @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
@@ -19,13 +19,20 @@
 
 package org.apache.gravitino.flink.connector.jdbc;
 
+import java.util.HashMap;
 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 {
 
-  protected JdbcPropertiesConverter() {}
+  private final CatalogFactory.Context context;
+
+  protected JdbcPropertiesConverter(CatalogFactory.Context context) {
+    this.context = context;
+  }
 
   @Override
   public Map<String, String> toGravitinoCatalogProperties(Configuration flinkConf) {
@@ -45,5 +52,21 @@ public String transformPropertyToFlinkCatalog(String configKey) {
     return JdbcPropertiesConstants.gravitinoToFlinkMap.get(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",
+        catalogOptions.get(JdbcPropertiesConstants.FLINK_JDBC_URL)
+            + "/"
+            + tablePath.getDatabaseName());
+    tableOptions.put("table-name", tablePath.getObjectName());
+    tableOptions.put("username", catalogOptions.get(JdbcPropertiesConstants.FLINK_JDBC_USER));
+    tableOptions.put("password", catalogOptions.get(JdbcPropertiesConstants.FLINK_JDBC_PASSWORD));
+    return tableOptions;
+  }
+
   protected abstract String driverName();
 }
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,11 +19,13 @@
 
 package org.apache.gravitino.flink.connector.jdbc;
 
-public class MysqlPropertiesConverter extends JdbcPropertiesConverter {
+import org.apache.flink.table.factories.CatalogFactory;
 
-  public static final MysqlPropertiesConverter INSTANCE = new MysqlPropertiesConverter();
+public class MysqlPropertiesConverter extends JdbcPropertiesConverter {
 
-  private MysqlPropertiesConverter() {}
+  protected MysqlPropertiesConverter(CatalogFactory.Context context) {
+    super(context);
+  }
 
   @Override
   public String driverName() {
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,11 +19,13 @@
 
 package org.apache.gravitino.flink.connector.jdbc;
 
-public class PostgresqlPropertiesConverter extends JdbcPropertiesConverter {
+import org.apache.flink.table.factories.CatalogFactory;
 
-  public static final PostgresqlPropertiesConverter INSTANCE = new PostgresqlPropertiesConverter();
+public class PostgresqlPropertiesConverter extends JdbcPropertiesConverter {
 
-  private PostgresqlPropertiesConverter() {}
+  protected PostgresqlPropertiesConverter(CatalogFactory.Context context) {
+    super(context);
+  }
 
   @Override
   protected String driverName() {
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
@@ -43,7 +43,7 @@ public Catalog createCatalog(Context context) {
     String defaultDatabase =
         helper.getOptions().get(GravitinoPaimonCatalogFactoryOptions.DEFAULT_DATABASE);
     return new GravitinoPaimonCatalog(
-        context, defaultDatabase, propertiesConverter(), partitionConverter());
+        context, defaultDatabase, propertiesConverter(context), partitionConverter());
   }
 
   @Override
@@ -72,7 +72,7 @@ public org.apache.gravitino.Catalog.Type gravitinoCatalogType() {
   }
 
   @Override
-  public PropertiesConverter propertiesConverter() {
+  public PropertiesConverter propertiesConverter(Context context) {
     return PaimonPropertiesConverter.INSTANCE;
   }
 

Original file line number	Diff line number	Diff line change
`@@ -45,7 +45,7 @@ public interface BaseCatalogFactory extends CatalogFactory {`
`45`	`45`	`*`
`46`	`46`	`* @return The requested property converter.`
`47`	`47`	`*/`
`48`		`- PropertiesConverter propertiesConverter();`
	`48`	`+ PropertiesConverter propertiesConverter(Context context);`
`49`	`49`
`50`	`50`	`/**`
`51`	`51`	`* Define partition converter.`
Original file line number	Diff line number	Diff line change
`@@ -29,8 +29,8 @@ public String gravitinoCatalogProvider() {`
`29`	`29`	`}`
`30`	`30`
`31`	`31`	`@Override`
`32`		`- public PropertiesConverter propertiesConverter() {`
`33`		`- return MysqlPropertiesConverter.INSTANCE;`
	`32`	`+ public PropertiesConverter propertiesConverter(Context context) {`
	`33`	`+ return new MysqlPropertiesConverter(context);`
`34`	`34`	`}`
`35`	`35`
`36`	`36`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ public Catalog createCatalog(Context context) {`
`43`	`43`	`String defaultDatabase =`
`44`	`44`	`helper.getOptions().get(GravitinoPaimonCatalogFactoryOptions.DEFAULT_DATABASE);`
`45`	`45`	`return new GravitinoPaimonCatalog(`
`46`		`- context, defaultDatabase, propertiesConverter(), partitionConverter());`
	`46`	`+ context, defaultDatabase, propertiesConverter(context), partitionConverter());`
`47`	`47`	`}`
`48`	`48`
`49`	`49`	`@Override`
`@@ -72,7 +72,7 @@ public org.apache.gravitino.Catalog.Type gravitinoCatalogType() {`
`72`	`72`	`}`
`73`	`73`
`74`	`74`	`@Override`
`75`		`- public PropertiesConverter propertiesConverter() {`
	`75`	`+ public PropertiesConverter propertiesConverter(Context context) {`
`76`	`76`	`return PaimonPropertiesConverter.INSTANCE;`
`77`	`77`	`}`
`78`	`78`