[HUDI-4098] Support HMS for flink HoodieCatalog

Author: cuibo01

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/catalog/HoodieHiveCatalog.java

@@ -499,13 +499,13 @@ private String inferTablePath(ObjectPath tablePath, CatalogBaseTable table) {
 
   private Map<String, String> applyOptionsHook(Map<String, String> options) {
     Map<String, String> properties = new HashMap<>(options);
-    if (!options.containsKey(FlinkOptions.RECORD_KEY_FIELD)) {
+    if (!options.containsKey(FlinkOptions.RECORD_KEY_FIELD.key())) {
       properties.put(FlinkOptions.RECORD_KEY_FIELD.key(), FlinkOptions.RECORD_KEY_FIELD.defaultValue());
     }
-    if (!options.containsKey(FlinkOptions.PRECOMBINE_FIELD)) {
+    if (!options.containsKey(FlinkOptions.PRECOMBINE_FIELD.key())) {
       properties.put(FlinkOptions.PRECOMBINE_FIELD.key(), FlinkOptions.PRECOMBINE_FIELD.defaultValue());
     }
-    if (!options.containsKey(FlinkOptions.TABLE_TYPE)) {
+    if (!options.containsKey(FlinkOptions.TABLE_TYPE.key())) {
       properties.put(FlinkOptions.TABLE_TYPE.key(), FlinkOptions.TABLE_TYPE.defaultValue());
     }
     return properties;

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieCatalogUtils.java

@@ -18,12 +18,10 @@
 
 package org.apache.hudi.table.catalog;
 
+import org.apache.flink.table.catalog.exceptions.CatalogException;
 import org.apache.hadoop.hive.conf.HiveConf;
-import org.junit.jupiter.api.io.TempDir;
 
-import java.io.File;
-
-import static org.apache.hudi.table.catalog.HoodieCatalogFactoryOptions.HIVE_SITE_FILE;
+import java.io.IOException;
 
 /** Test utils for Hoodie catalog. */
 public class TestHoodieCatalogUtils {
@@ -32,8 +30,7 @@ public class TestHoodieCatalogUtils {
 
   private static final String TEST_CATALOG_NAME = "test-catalog";
 
-  @TempDir
-  static File tempFile;
+  private static final org.junit.rules.TemporaryFolder TEMPORARY_FOLDER = new org.junit.rules.TemporaryFolder();
 
   /** Create a HiveCatalog with an embedded Hive Metastore. */
   public static HoodieHiveCatalog createHiveCatalog() {
@@ -50,18 +47,20 @@ public static HoodieHiveCatalog createHiveCatalog(String name) {
 
   public static HiveConf createHiveConf() {
     ClassLoader classLoader = TestHoodieCatalogUtils.class.getClassLoader();
+    try {
+      TEMPORARY_FOLDER.create();
+      String warehouseDir = TEMPORARY_FOLDER.newFolder().getAbsolutePath() + "/metastore_db";
+      String warehouseUri = String.format(HIVE_WAREHOUSE_URI_FORMAT, warehouseDir);
 
-    new File(tempFile,"db").mkdir();
-    File metastore_db = new File(tempFile,"/db/metastore_db");
-    String warehouseUri = String.format(HIVE_WAREHOUSE_URI_FORMAT, metastore_db.getAbsolutePath());
-
-    HiveConf.setHiveSiteLocation(classLoader.getResource(HIVE_SITE_FILE));
-
-    File hive_warehouse = new File(tempFile,"/hive_warehouse");
-    hive_warehouse.mkdir() ;
-    HiveConf hiveConf = new HiveConf();
-    hiveConf.setVar(HiveConf.ConfVars.METASTOREWAREHOUSE, hive_warehouse.getAbsolutePath());
-    hiveConf.setVar(HiveConf.ConfVars.METASTORECONNECTURLKEY, warehouseUri);
-    return hiveConf;
+      HiveConf.setHiveSiteLocation(classLoader.getResource(HoodieCatalogFactoryOptions.HIVE_SITE_FILE));
+      HiveConf hiveConf = new HiveConf();
+      hiveConf.setVar(
+          HiveConf.ConfVars.METASTOREWAREHOUSE,
+          TEMPORARY_FOLDER.newFolder("hive_warehouse").getAbsolutePath());
+      hiveConf.setVar(HiveConf.ConfVars.METASTORECONNECTURLKEY, warehouseUri);
+      return hiveConf;
+    } catch (IOException e) {
+      throw new CatalogException("Failed to create test HiveConf to HiveCatalog.", e);
+    }
   }
 }

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/table/catalog/TestHoodieHiveCatalog.java

@@ -18,6 +18,9 @@
 
 package org.apache.hudi.table.catalog;
 
+import org.apache.hudi.configuration.FlinkOptions;
+import org.apache.hudi.exception.HoodieCatalogException;
+
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.catalog.CatalogBaseTable;
@@ -29,13 +32,13 @@
 import org.apache.flink.table.catalog.exceptions.TableNotExistException;
 import org.apache.flink.table.factories.FactoryUtil;
 import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.hudi.exception.HoodieCatalogException;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 
 import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR;
@@ -71,6 +74,19 @@ public static void closeCatalog() {
     }
   }
 
+  @Test
+  public void testCreateAndGetHoodieMORTable() throws Exception {
+    Map<String, String> originOptions = new HashMap<>();
+    originOptions.put(FactoryUtil.CONNECTOR.key(), "hudi");
+    originOptions.put(FlinkOptions.TABLE_TYPE.key(), FlinkOptions.TABLE_TYPE_MERGE_ON_READ);
+    CatalogTable table =
+        new CatalogTableImpl(schema, originOptions, "hudi table");
+    hoodieCatalog.createTable(tablePath, table, false);
+    CatalogBaseTable table1 = hoodieCatalog.getTable(tablePath);
+    assertEquals(table1.getOptions().get(CONNECTOR.key()), "hudi");
+    assertEquals(table1.getOptions().get(FlinkOptions.TABLE_TYPE.key()), FlinkOptions.TABLE_TYPE_MERGE_ON_READ);
+  }
+
   @Test
   public void testCreateAndGetHoodieTable() throws Exception {
     Map<String, String> originOptions =