[HUDI-4098]Support HMS for flink HudiCatalog

Author: cuibo01

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

@@ -798,6 +798,15 @@ private FlinkOptions() {
       .noDefaultValue()
       .withDescription("The hive configuration directory, where the hive-site.xml lies in, the file should be put on the client machine");
 
+  // ------------------------------------------------------------------------
+  //  Create Table Options With Catalog
+  // ------------------------------------------------------------------------
+  public static final ConfigOption<Boolean> HIVE_IS_EXTERNAL = ConfigOptions
+      .key("hive.is-external")
+      .booleanType()
+      .defaultValue(true)
+      .withDescription("Whether the table is external, default true");
+
   // -------------------------------------------------------------------------
   //  Utilities
   // -------------------------------------------------------------------------

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

@@ -23,9 +23,6 @@
 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;
 
 /**
@@ -57,45 +54,6 @@ 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/configuration/OptionsResolver.java

@@ -27,6 +27,7 @@
 import org.apache.flink.configuration.Configuration;
 
 import java.util.Locale;
+import java.util.Map;
 
 /**
  * Tool helping to resolve the flink options {@link FlinkOptions}.
@@ -66,6 +67,14 @@ public static boolean isMorTable(Configuration conf) {
         .equals(FlinkOptions.TABLE_TYPE_MERGE_ON_READ);
   }
 
+  /**
+   * Returns whether it is a MERGE_ON_READ table.
+   */
+  public static boolean isMorTable(Map<String, String> options) {
+    return options.getOrDefault(FlinkOptions.TABLE_TYPE.key(),
+        FlinkOptions.TABLE_TYPE.defaultValue()).equalsIgnoreCase(FlinkOptions.TABLE_TYPE_MERGE_ON_READ);
+  }
+
   /**
    * Returns whether it is a COPY_ON_WRITE table.
    */

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

@@ -29,6 +29,8 @@
  * Hoodie catalog options.
  */
 public class CatalogOptions {
+  public static final String HIVE_SITE_FILE = "hive-site.xml";
+  public static final String DEFAULT_DB = "default";
 
   public static final ConfigOption<String> CATALOG_PATH =
       ConfigOptions.key("catalog.path")
@@ -42,6 +44,16 @@ public class CatalogOptions {
           .stringType()
           .defaultValue("default");
 
+  public static final ConfigOption<String> HIVE_CONF_DIR = ConfigOptions
+      .key("hive.conf.dir")
+      .stringType()
+      .noDefaultValue();
+
+  public static final ConfigOption<String> MODE = ConfigOptions
+      .key("mode")
+      .stringType()
+      .defaultValue("dfs");
+
   /**
    * Returns all the common table options that can be shared.
    *

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

@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.catalog;
+
+import org.apache.hudi.common.util.StringUtils;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.serde2.typeinfo.CharTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.DecimalTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.ListTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.MapTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
+import org.apache.hadoop.hive.serde2.typeinfo.VarcharTypeInfo;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * hive table options.
+ */
+public class HiveTableOptions {
+  /** Get field names from field schemas. */
+  public static List<String> getFieldNames(List<FieldSchema> fieldSchemas) {
+    List<String> names = new ArrayList<>(fieldSchemas.size());
+    for (FieldSchema fs : fieldSchemas) {
+      names.add(fs.getName());
+    }
+    return names;
+  }
+
+  public static org.apache.flink.table.api.Schema convertTableSchema(Table hiveTable) {
+    List<FieldSchema> allCols = new ArrayList<>(hiveTable.getSd().getCols());
+    allCols.addAll(hiveTable.getPartitionKeys());
+
+    String pkConstraintName = hiveTable.getParameters().get(TableOptionProperties.PK_CONSTRAINT_NAME);
+    List<String> primaryColNames = StringUtils.isNullOrEmpty(pkConstraintName)
+        ? Collections.EMPTY_LIST
+        : StringUtils.split(hiveTable.getParameters().get(TableOptionProperties.PK_COLUMNS),",");
+
+    String[] colNames = new String[allCols.size()];
+    DataType[] colTypes = new DataType[allCols.size()];
+
+    for (int i = 0; i < allCols.size(); i++) {
+      FieldSchema fs = allCols.get(i);
+
+      colNames[i] = fs.getName();
+      colTypes[i] =
+          toFlinkType(TypeInfoUtils.getTypeInfoFromTypeString(fs.getType()));
+      if (primaryColNames.contains(colNames[i])) {
+        colTypes[i] = colTypes[i].notNull();
+      }
+    }
+
+    org.apache.flink.table.api.Schema.Builder builder = org.apache.flink.table.api.Schema.newBuilder().fromFields(colNames, colTypes);
+    if (!StringUtils.isNullOrEmpty(pkConstraintName)) {
+      builder.primaryKeyNamed(pkConstraintName, primaryColNames);
+    }
+
+    return builder.build();
+  }
+
+  /**
+   * Convert Hive data type to a Flink data type.
+   *
+   * @param hiveType a Hive data type
+   * @return the corresponding Flink data type
+   */
+  public static DataType toFlinkType(TypeInfo hiveType) {
+    checkNotNull(hiveType, "hiveType cannot be null");
+
+    switch (hiveType.getCategory()) {
+      case PRIMITIVE:
+        return toFlinkPrimitiveType((PrimitiveTypeInfo) hiveType);
+      case LIST:
+        ListTypeInfo listTypeInfo = (ListTypeInfo) hiveType;
+        return DataTypes.ARRAY(toFlinkType(listTypeInfo.getListElementTypeInfo()));
+      case MAP:
+        MapTypeInfo mapTypeInfo = (MapTypeInfo) hiveType;
+        return DataTypes.MAP(
+            toFlinkType(mapTypeInfo.getMapKeyTypeInfo()),
+            toFlinkType(mapTypeInfo.getMapValueTypeInfo()));
+      case STRUCT:
+        StructTypeInfo structTypeInfo = (StructTypeInfo) hiveType;
+
+        List<String> names = structTypeInfo.getAllStructFieldNames();
+        List<TypeInfo> typeInfos = structTypeInfo.getAllStructFieldTypeInfos();
+
+        DataTypes.Field[] fields = new DataTypes.Field[names.size()];
+
+        for (int i = 0; i < fields.length; i++) {
+          fields[i] = DataTypes.FIELD(names.get(i), toFlinkType(typeInfos.get(i)));
+        }
+
+        return DataTypes.ROW(fields);
+      default:
+        throw new UnsupportedOperationException(
+            String.format("Flink doesn't support Hive data type %s yet.", hiveType));
+    }
+  }
+
+  private static DataType toFlinkPrimitiveType(PrimitiveTypeInfo hiveType) {
+    checkNotNull(hiveType, "hiveType cannot be null");
+
+    switch (hiveType.getPrimitiveCategory()) {
+      case CHAR:
+        return DataTypes.CHAR(((CharTypeInfo) hiveType).getLength());
+      case VARCHAR:
+        return DataTypes.VARCHAR(((VarcharTypeInfo) hiveType).getLength());
+      case STRING:
+        return DataTypes.STRING();
+      case BOOLEAN:
+        return DataTypes.BOOLEAN();
+      case BYTE:
+        return DataTypes.TINYINT();
+      case SHORT:
+        return DataTypes.SMALLINT();
+      case INT:
+        return DataTypes.INT();
+      case LONG:
+        return DataTypes.BIGINT();
+      case FLOAT:
+        return DataTypes.FLOAT();
+      case DOUBLE:
+        return DataTypes.DOUBLE();
+      case DATE:
+        return DataTypes.DATE();
+      case TIMESTAMP:
+        return DataTypes.TIMESTAMP(9);
+      case BINARY:
+        return DataTypes.BYTES();
+      case DECIMAL:
+        DecimalTypeInfo decimalTypeInfo = (DecimalTypeInfo) hiveType;
+        return DataTypes.DECIMAL(
+            decimalTypeInfo.getPrecision(), decimalTypeInfo.getScale());
+      default:
+        throw new UnsupportedOperationException(
+            String.format(
+                "Flink doesn't support Hive primitive type %s yet", hiveType));
+    }
+  }
+
+  /** Create Hive columns from Flink TableSchema. */
+  public static List<FieldSchema> createHiveColumns(TableSchema schema) {
+    String[] fieldNames = schema.getFieldNames();
+    DataType[] fieldTypes = schema.getFieldDataTypes();
+
+    List<FieldSchema> columns = new ArrayList<>(fieldNames.length);
+
+    for (int i = 0; i < fieldNames.length; i++) {
+      columns.add(
+          new FieldSchema(
+              fieldNames[i],
+              toHiveTypeInfo(fieldTypes[i], true).getTypeName(),
+              null));
+    }
+
+    return columns;
+  }
+
+  /**
+   * Convert Flink DataType to Hive TypeInfo. For types with a precision parameter, e.g.
+   * timestamp, the supported precisions in Hive and Flink can be different. Therefore the
+   * conversion will fail for those types if the precision is not supported by Hive and
+   * checkPrecision is true.
+   *
+   * @param dataType a Flink DataType
+   * @param checkPrecision whether to fail the conversion if the precision of the DataType is not
+   *     supported by Hive
+   * @return the corresponding Hive data type
+   */
+  public static TypeInfo toHiveTypeInfo(DataType dataType, boolean checkPrecision) {
+    checkNotNull(dataType, "type cannot be null");
+    LogicalType logicalType = dataType.getLogicalType();
+    return logicalType.accept(new TypeInfoLogicalTypeVisitor(dataType, checkPrecision));
+  }
+}

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

@@ -30,6 +30,7 @@
 
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Locale;
 import java.util.Set;
 
 import static org.apache.flink.table.factories.FactoryUtil.PROPERTY_VERSION;
@@ -53,19 +54,19 @@ public Catalog createCatalog(Context context) {
     final FactoryUtil.CatalogFactoryHelper helper =
         FactoryUtil.createCatalogFactoryHelper(this, context);
     helper.validate();
-
-    if (helper.getOptions().get(HoodieCatalogFactoryOptions.MODE).equalsIgnoreCase("hms")) {
-      return new HoodieHiveCatalog(
-          context.getName(),
-          helper.getOptions().get(HoodieCatalogFactoryOptions.DEFAULT_DATABASE),
-          helper.getOptions().get(HoodieCatalogFactoryOptions.HIVE_CONF_DIR),
-          helper.getOptions().get(HoodieCatalogFactoryOptions.INIT_FS_TABLE));
-    } else if (helper.getOptions().get(HoodieCatalogFactoryOptions.MODE).equalsIgnoreCase("dfs")) {
-      return new HoodieCatalog(
-          context.getName(),
-          (Configuration) helper.getOptions());
-    } else {
-      throw new HoodieCatalogException("hoodie catalog supports only the hms and dfs modes.");
+    String mode = helper.getOptions().get(CatalogOptions.MODE);
+    switch (mode.toLowerCase(Locale.ROOT)) {
+      case "hms":
+        return new HoodieHiveCatalog(
+            context.getName(),
+            helper.getOptions().get(CatalogOptions.DEFAULT_DATABASE),
+            helper.getOptions().get(CatalogOptions.HIVE_CONF_DIR));
+      case "dfs":
+        return new HoodieCatalog(
+            context.getName(),
+            (Configuration) helper.getOptions());
+      default:
+        throw new HoodieCatalogException(String.format("Invalid catalog mode: %s, supported modes: [hms, dfs].", mode));
     }
   }
 
@@ -77,12 +78,11 @@ public Set<ConfigOption<?>> requiredOptions() {
   @Override
   public Set<ConfigOption<?>> optionalOptions() {
     final Set<ConfigOption<?>> options = new HashSet<>();
-    options.add(HoodieCatalogFactoryOptions.DEFAULT_DATABASE);
+    options.add(CatalogOptions.DEFAULT_DATABASE);
     options.add(PROPERTY_VERSION);
-    options.add(HoodieCatalogFactoryOptions.HIVE_CONF_DIR);
-    options.add(HoodieCatalogFactoryOptions.MODE);
+    options.add(CatalogOptions.HIVE_CONF_DIR);
+    options.add(CatalogOptions.MODE);
     options.add(CATALOG_PATH);
-    options.add(HoodieCatalogFactoryOptions.INIT_FS_TABLE);
     return options;
   }
 }