[HUDI-4098] Support HMS for flink HoodieCatalog

Author: cuibo01

hudi-common/src/main/java/org/apache/hudi/common/table/TableSchemaResolver.java

@@ -262,7 +262,7 @@ private MessageType getTableParquetSchemaFromDataFile() {
     }
   }
 
-  public static MessageType convertAvroSchemaToParquetMessageType(Schema schema, Configuration hadoopConf) {
+  public static MessageType convertAvroSchemaToParquet(Schema schema, Configuration hadoopConf) {
     AvroSchemaConverter avroSchemaConverter = new AvroSchemaConverter(hadoopConf);
     return avroSchemaConverter.convert(schema);
   }

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/HadoopConfigurations.java

@@ -23,6 +23,9 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.hadoop.fs.Path;
 
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -54,6 +57,45 @@ public static org.apache.hadoop.conf.Configuration getHadoopConf(Configuration c
     return hadoopConf;
   }
 
+  /**
+   * Returns a new hadoop configuration that is initialized with the given hadoopConfDir.
+   *
+   * @param hadoopConfDir Hadoop conf directory path.
+   * @return A Hadoop configuration instance.
+   */
+  public static org.apache.hadoop.conf.Configuration getHadoopConfiguration(String hadoopConfDir) {
+    if (new File(hadoopConfDir).exists()) {
+      List<File> possiableConfFiles = new ArrayList<File>();
+      File coreSite = new File(hadoopConfDir, "core-site.xml");
+      if (coreSite.exists()) {
+        possiableConfFiles.add(coreSite);
+      }
+      File hdfsSite = new File(hadoopConfDir, "hdfs-site.xml");
+      if (hdfsSite.exists()) {
+        possiableConfFiles.add(hdfsSite);
+      }
+      File yarnSite = new File(hadoopConfDir, "yarn-site.xml");
+      if (yarnSite.exists()) {
+        possiableConfFiles.add(yarnSite);
+      }
+      // Add mapred-site.xml. We need to read configurations like compression codec.
+      File mapredSite = new File(hadoopConfDir, "mapred-site.xml");
+      if (mapredSite.exists()) {
+        possiableConfFiles.add(mapredSite);
+      }
+      if (possiableConfFiles.isEmpty()) {
+        return null;
+      } else {
+        org.apache.hadoop.conf.Configuration hadoopConfiguration = new org.apache.hadoop.conf.Configuration();
+        for (File confFile : possiableConfFiles) {
+          hadoopConfiguration.addResource(new Path(confFile.getAbsolutePath()));
+        }
+        return hadoopConfiguration;
+      }
+    }
+    return null;
+  }
+
   /**
    * Creates a Hive configuration with configured dir path or empty if no Hive conf dir is set.
    */

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

@@ -18,6 +18,8 @@
 
 package org.apache.hudi.table.catalog;
 
+import org.apache.hudi.exception.HoodieCatalogException;
+
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.catalog.Catalog;
@@ -26,12 +28,11 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
 
+import static org.apache.flink.table.factories.FactoryUtil.PROPERTY_VERSION;
 import static org.apache.hudi.table.catalog.CatalogOptions.CATALOG_PATH;
-import static org.apache.hudi.table.catalog.CatalogOptions.DEFAULT_DATABASE;
 
 /**
  * A catalog factory impl that creates {@link HoodieCatalog}.
@@ -52,21 +53,36 @@ public Catalog createCatalog(Context context) {
         FactoryUtil.createCatalogFactoryHelper(this, context);
     helper.validate();
 
-    return new HoodieCatalog(
-        context.getName(),
-        (Configuration) helper.getOptions());
+    if (helper.getOptions().get(HoodieCatalogFactoryOptions.TYPE).equalsIgnoreCase("hms")) {
+      return new HoodieHiveCatalog(
+          context.getName(),
+          helper.getOptions().get(HoodieCatalogFactoryOptions.DEFAULT_DATABASE),
+          helper.getOptions().get(HoodieCatalogFactoryOptions.HIVE_CONF_DIR),
+          helper.getOptions().get(HoodieCatalogFactoryOptions.HADOOP_CONF_DIR));
+    } else if (helper.getOptions().get(HoodieCatalogFactoryOptions.TYPE).equalsIgnoreCase("dfs")) {
+      return new HoodieCatalog(
+          context.getName(),
+          (Configuration) helper.getOptions());
+    } else {
+      throw new HoodieCatalogException("hoodie catalog supports only the hms and dfs modes.");
+    }
   }
 
   @Override
   public Set<ConfigOption<?>> requiredOptions() {
     Set<ConfigOption<?>> options = new HashSet<>();
     options.add(CATALOG_PATH);
-    options.add(DEFAULT_DATABASE);
     return options;
   }
 
   @Override
   public Set<ConfigOption<?>> optionalOptions() {
-    return Collections.emptySet();
+    final Set<ConfigOption<?>> options = new HashSet<>();
+    options.add(HoodieCatalogFactoryOptions.DEFAULT_DATABASE);
+    options.add(PROPERTY_VERSION);
+    options.add(HoodieCatalogFactoryOptions.HIVE_CONF_DIR);
+    options.add(HoodieCatalogFactoryOptions.HADOOP_CONF_DIR);
+    options.add(HoodieCatalogFactoryOptions.TYPE);
+    return options;
   }
 }

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

@@ -24,7 +24,6 @@
 
 /** {@link ConfigOption}s for {@link HoodieHiveCatalog}. */
 public class HoodieCatalogFactoryOptions {
-  public static final String IDENTIFIER = "hudi-hive";
   public static final String DEFAULT_DB = "default";
   public static final String HIVE_SITE_FILE = "hive-site.xml";
 
@@ -39,5 +38,8 @@ public class HoodieCatalogFactoryOptions {
   public static final ConfigOption<String> HADOOP_CONF_DIR =
       ConfigOptions.key("hadoop-conf-dir").stringType().noDefaultValue();
 
+  public static final ConfigOption<String> TYPE =
+      ConfigOptions.key("type").stringType().defaultValue("dfs");
+
   private HoodieCatalogFactoryOptions() {}
 }

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

@@ -18,7 +18,8 @@
 
 package org.apache.hudi.table.catalog;
 
-import org.apache.flink.api.java.hadoop.mapred.utils.HadoopUtils;
+import org.apache.hudi.configuration.HadoopConfigurations;
+
 import org.apache.flink.table.catalog.exceptions.CatalogException;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
@@ -33,8 +34,6 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URL;
-import java.util.ArrayList;
-import java.util.List;
 
 import static org.apache.flink.util.StringUtils.isNullOrWhitespaceOnly;
 import static org.apache.hudi.table.catalog.HoodieCatalogFactoryOptions.HIVE_SITE_FILE;
@@ -46,7 +45,7 @@ public class HoodieCatalogUtil {
   private static final Logger LOG = LoggerFactory.getLogger(HoodieCatalogUtil.class);
 
   /**
-   * Returns a new hiveConfig.
+   * Returns a new {@code HiveConf}.
    *
    * @param hiveConfDir Hive conf directory path.
    * @param hadoopConfDir Hadoop conf directory path.
@@ -56,16 +55,9 @@ public static HiveConf createHiveConf(@Nullable String hiveConfDir, @Nullable St
     // create HiveConf from hadoop configuration with hadoop conf directory configured.
     Configuration hadoopConf = null;
     if (isNullOrWhitespaceOnly(hadoopConfDir)) {
-      for (String possibleHadoopConfPath :
-          HadoopUtils.possibleHadoopConfPaths(
-              new org.apache.flink.configuration.Configuration())) {
-        hadoopConf = getHadoopConfiguration(possibleHadoopConfPath);
-        if (hadoopConf != null) {
-          break;
-        }
-      }
+      hadoopConf = HadoopConfigurations.getHadoopConf(new org.apache.flink.configuration.Configuration());
     } else {
-      hadoopConf = getHadoopConfiguration(hadoopConfDir);
+      hadoopConf = HadoopConfigurations.getHadoopConfiguration(hadoopConfDir);
       if (hadoopConf == null) {
         String possiableUsedConfFiles =
             "core-site.xml | hdfs-site.xml | yarn-site.xml | mapred-site.xml";
@@ -77,9 +69,6 @@ public static HiveConf createHiveConf(@Nullable String hiveConfDir, @Nullable St
                     + ") exist in the folder."));
       }
     }
-    if (hadoopConf == null) {
-      hadoopConf = new Configuration();
-    }
     // ignore all the static conf file URLs that HiveConf may have set
     HiveConf.setHiveSiteLocation(null);
     HiveConf.setLoadMetastoreConfig(false);
@@ -120,43 +109,4 @@ public static HiveConf createHiveConf(@Nullable String hiveConfDir, @Nullable St
   public static boolean isEmbeddedMetastore(HiveConf hiveConf) {
     return isNullOrWhitespaceOnly(hiveConf.getVar(HiveConf.ConfVars.METASTOREURIS));
   }
-
-  /**
-   * Returns a new Hadoop Configuration object using the path to the hadoop conf configured.
-   *
-   * @param hadoopConfDir Hadoop conf directory path.
-   * @return A Hadoop configuration instance.
-   */
-  public static Configuration getHadoopConfiguration(String hadoopConfDir) {
-    if (new File(hadoopConfDir).exists()) {
-      List<File> possiableConfFiles = new ArrayList<File>();
-      File coreSite = new File(hadoopConfDir, "core-site.xml");
-      if (coreSite.exists()) {
-        possiableConfFiles.add(coreSite);
-      }
-      File hdfsSite = new File(hadoopConfDir, "hdfs-site.xml");
-      if (hdfsSite.exists()) {
-        possiableConfFiles.add(hdfsSite);
-      }
-      File yarnSite = new File(hadoopConfDir, "yarn-site.xml");
-      if (yarnSite.exists()) {
-        possiableConfFiles.add(yarnSite);
-      }
-      // Add mapred-site.xml. We need to read configurations like compression codec.
-      File mapredSite = new File(hadoopConfDir, "mapred-site.xml");
-      if (mapredSite.exists()) {
-        possiableConfFiles.add(mapredSite);
-      }
-      if (possiableConfFiles.isEmpty()) {
-        return null;
-      } else {
-        Configuration hadoopConfiguration = new Configuration();
-        for (File confFile : possiableConfFiles) {
-          hadoopConfiguration.addResource(new Path(confFile.getAbsolutePath()));
-        }
-        return hadoopConfiguration;
-      }
-    }
-    return null;
-  }
 }

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

@@ -127,7 +127,7 @@ public HoodieHiveCatalog(String catalogName, String defaultDatabase, HiveConf hi
           "Embedded metastore is not allowed. Make sure you have set a valid value for "
               + HiveConf.ConfVars.METASTOREURIS.toString());
     }
-    LOG.info("Created HiveCatalog '{}'", catalogName);
+    LOG.info("Created Hoodie Catalog '{}' in hms mode", catalogName);
   }
 
   @Override
@@ -375,7 +375,7 @@ private Table translateSparkTable2Flink(ObjectPath tablePath, Table hiveTable) {
           boolean hiveStyle = Arrays.stream(FSUtils.getFs(hoodieTablePath, hiveConf).listStatus(hoodieTablePath))
               .map(fileStatus -> fileStatus.getPath().getName())
               .filter(f -> !f.equals(".hoodie") && !f.equals("default"))
-              .anyMatch(FilePathUtils::hiveStylePartitionMath);
+              .anyMatch(FilePathUtils::isHiveStylePartitioning);
           parameters.put(FlinkOptions.HIVE_STYLE_PARTITIONING.key(), String.valueOf(hiveStyle));
         }
         client.alter_table(tablePath.getDatabaseName(), tablePath.getObjectName(), hiveTable);
@@ -404,7 +404,7 @@ public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistExcep
       }
       schema = builder.build();
     } else {
-      LOG.warn("{} does not have any hoodie schema, and use hive table to covert the catalogBaseTable", tablePath);
+      LOG.warn("{} does not have any hoodie schema, and use hive table schema to infer the table schema", tablePath);
       schema = TableOptionProperties.convertTableSchema(hiveTable);
     }
     return CatalogTable.of(schema, parameters.get(COMMENT),

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

@@ -176,7 +176,7 @@ public static Map<String, String> getTableOptions(Map<String, String> options) {
 
   public static Map<String, String> translateFlinkTableProperties2Spark(CatalogTable catalogTable, Configuration hadoopConf) {
     Schema schema = AvroSchemaConverter.convertToSchema(catalogTable.getSchema().toPhysicalRowDataType().getLogicalType());
-    MessageType messageType = TableSchemaResolver.convertAvroSchemaToParquetMessageType(schema, hadoopConf);
+    MessageType messageType = TableSchemaResolver.convertAvroSchemaToParquet(schema, hadoopConf);
     String sparkVersion = catalogTable.getOptions().getOrDefault(SPARK_VERSION, DEFAULT_SPARK_VERSION);
     return SparkDataSourceTableUtils.getSparkTableProperties(catalogTable.getPartitionKeys(), sparkVersion, 4000, messageType);
   }

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

@@ -443,7 +443,7 @@ public static String[] extractHivePartitionFields(org.apache.flink.configuration
     return conf.getString(FlinkOptions.HIVE_SYNC_PARTITION_FIELDS).split(",");
   }
 
-  public static boolean hiveStylePartitionMath(String path) {
+  public static boolean isHiveStylePartitioning(String path) {
     return HIVE_PARTITION_NAME_PATTERN.matcher(path).matches();
   }
 }

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/FilePathUtils.java

@@ -15,5 +15,4 @@
 # limitations under the License.
 
 org.apache.hudi.table.HoodieTableFactory
-org.apache.hudi.table.catalog.HoodieCatalogFactory
-org.apache.hudi.table.catalog.HoodieHiveCatalogFactory
+org.apache.hudi.table.catalog.HoodieCatalogFactory