[HUDI-2538] persist some configs to hoodie.properties when the first write

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/upgrade/TwoToThreeUpgradeHandler.java

@@ -21,10 +21,11 @@
 
 import org.apache.hudi.common.config.ConfigProperty;
 import org.apache.hudi.common.engine.HoodieEngineContext;
+import org.apache.hudi.common.table.HoodieTableConfig;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.metadata.HoodieTableMetadataUtil;
 
-import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 
 /**
@@ -39,6 +40,10 @@ public Map<ConfigProperty, String> upgrade(HoodieWriteConfig config, HoodieEngin
       // table has been updated and is not backward compatible.
       HoodieTableMetadataUtil.deleteMetadataTable(config.getBasePath(), context);
     }
-    return Collections.emptyMap();
+    Map<ConfigProperty, String> tablePropsToAdd = new HashMap<>();
+    tablePropsToAdd.put(HoodieTableConfig.URL_ENCODE_PARTITIONING, config.getStringOrDefault(HoodieTableConfig.URL_ENCODE_PARTITIONING));
+    tablePropsToAdd.put(HoodieTableConfig.HIVE_STYLE_PARTITIONING_ENABLE, config.getStringOrDefault(HoodieTableConfig.HIVE_STYLE_PARTITIONING_ENABLE));
+    tablePropsToAdd.put(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME, config.getString(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME));
+    return tablePropsToAdd;
   }
 }

hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/keygen/factory/HoodieSparkKeyGeneratorFactory.java

@@ -19,14 +19,13 @@
 package org.apache.hudi.keygen.factory;
 
 import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.ReflectionUtils;
 import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.config.HoodieWriteConfig;
 import org.apache.hudi.exception.HoodieKeyGeneratorException;
-import org.apache.hudi.keygen.BuiltinKeyGenerator;
 import org.apache.hudi.keygen.ComplexKeyGenerator;
 import org.apache.hudi.keygen.CustomKeyGenerator;
 import org.apache.hudi.keygen.GlobalDeleteKeyGenerator;
-import org.apache.hudi.keygen.KeyGenUtils;
 import org.apache.hudi.keygen.KeyGenerator;
 import org.apache.hudi.keygen.NonpartitionedKeyGenerator;
 import org.apache.hudi.keygen.SimpleKeyGenerator;
@@ -37,8 +36,9 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.Locale;
-import java.util.Objects;
+import java.util.Map;
 
 /**
  * Factory help to create {@link org.apache.hudi.keygen.KeyGenerator}.
@@ -50,45 +50,73 @@ public class HoodieSparkKeyGeneratorFactory {
 
   private static final Logger LOG = LoggerFactory.getLogger(HoodieSparkKeyGeneratorFactory.class);
 
-  public static KeyGenerator createKeyGenerator(TypedProperties props) throws IOException {
-    // keyGenerator class name has higher priority
-    KeyGenerator keyGenerator = KeyGenUtils.createKeyGeneratorByClassName(props);
+  private static final Map<String, String> COMMON_TO_SPARK_KEYGENERATOR = new HashMap<>();
+  static {
+    COMMON_TO_SPARK_KEYGENERATOR.put("org.apache.hudi.keygen.ComplexAvroKeyGenerator",
+        "org.apache.hudi.keygen.ComplexKeyGenerator");
+    COMMON_TO_SPARK_KEYGENERATOR.put("org.apache.hudi.keygen.CustomAvroKeyGenerator",
+        "org.apache.hudi.keygen.CustomKeyGenerator");
+    COMMON_TO_SPARK_KEYGENERATOR.put("org.apache.hudi.keygen.GlobalAvroDeleteKeyGenerator",
+        "org.apache.hudi.keygen.GlobalDeleteKeyGenerator");
+    COMMON_TO_SPARK_KEYGENERATOR.put("org.apache.hudi.keygen.NonpartitionedAvroKeyGenerator",
+        "org.apache.hudi.keygen.NonpartitionedKeyGenerator");
+    COMMON_TO_SPARK_KEYGENERATOR.put("org.apache.hudi.keygen.SimpleAvroKeyGenerator",
+        "org.apache.hudi.keygen.SimpleKeyGenerator");
+    COMMON_TO_SPARK_KEYGENERATOR.put("org.apache.hudi.keygen.TimestampBasedAvroKeyGenerator",
+        "org.apache.hudi.keygen.TimestampBasedKeyGenerator");
+  }
 
-    return Objects.isNull(keyGenerator) ? createKeyGeneratorByType(props) : keyGenerator;
+  public static KeyGenerator createKeyGenerator(TypedProperties props) throws IOException {
+    String keyGeneratorClass = getKeyGeneratorClassName(props);
+    try {
+      return (KeyGenerator) ReflectionUtils.loadClass(keyGeneratorClass, props);
+    } catch (Throwable e) {
+      throw new IOException("Could not load key generator class " + keyGeneratorClass, e);
+    }
   }
 
-  private static BuiltinKeyGenerator createKeyGeneratorByType(TypedProperties props) throws IOException {
-    // Use KeyGeneratorType.SIMPLE as default keyGeneratorType
-    String keyGeneratorType =
-        props.getString(HoodieWriteConfig.KEYGENERATOR_TYPE.key(), null);
+  public static String getKeyGeneratorClassName(TypedProperties props) {
+    String keyGeneratorClass = props.getString(HoodieWriteConfig.KEYGENERATOR_CLASS_NAME.key(), null);
 
-    if (StringUtils.isNullOrEmpty(keyGeneratorType)) {
+    if (StringUtils.isNullOrEmpty(keyGeneratorClass)) {
+      String keyGeneratorType = props.getString(HoodieWriteConfig.KEYGENERATOR_TYPE.key(), KeyGeneratorType.SIMPLE.name());
       LOG.info("The value of {} is empty, use SIMPLE", HoodieWriteConfig.KEYGENERATOR_TYPE.key());
-      keyGeneratorType = KeyGeneratorType.SIMPLE.name();
-    }
-
-    KeyGeneratorType keyGeneratorTypeEnum;
-    try {
-      keyGeneratorTypeEnum = KeyGeneratorType.valueOf(keyGeneratorType.toUpperCase(Locale.ROOT));
-    } catch (IllegalArgumentException e) {
-      throw new HoodieKeyGeneratorException("Unsupported keyGenerator Type " + keyGeneratorType);
-    }
-    switch (keyGeneratorTypeEnum) {
-      case SIMPLE:
-        return new SimpleKeyGenerator(props);
-      case COMPLEX:
-        return new ComplexKeyGenerator(props);
-      case TIMESTAMP:
-        return new TimestampBasedKeyGenerator(props);
-      case CUSTOM:
-        return new CustomKeyGenerator(props);
-      case NON_PARTITION:
-        return new NonpartitionedKeyGenerator(props);
-      case GLOBAL_DELETE:
-        return new GlobalDeleteKeyGenerator(props);
-      default:
+      KeyGeneratorType keyGeneratorTypeEnum;
+      try {
+        keyGeneratorTypeEnum = KeyGeneratorType.valueOf(keyGeneratorType.toUpperCase(Locale.ROOT));
+      } catch (IllegalArgumentException e) {
         throw new HoodieKeyGeneratorException("Unsupported keyGenerator Type " + keyGeneratorType);
+      }
+      switch (keyGeneratorTypeEnum) {
+        case SIMPLE:
+          keyGeneratorClass = SimpleKeyGenerator.class.getName();
+          break;
+        case COMPLEX:
+          keyGeneratorClass = ComplexKeyGenerator.class.getName();
+          break;
+        case TIMESTAMP:
+          keyGeneratorClass = TimestampBasedKeyGenerator.class.getName();
+          break;
+        case CUSTOM:
+          keyGeneratorClass = CustomKeyGenerator.class.getName();
+          break;
+        case NON_PARTITION:
+          keyGeneratorClass = NonpartitionedKeyGenerator.class.getName();
+          break;
+        case GLOBAL_DELETE:
+          keyGeneratorClass = GlobalDeleteKeyGenerator.class.getName();
+          break;
+        default:
+          throw new HoodieKeyGeneratorException("Unsupported keyGenerator Type " + keyGeneratorType);
+      }
     }
+    return keyGeneratorClass;
   }
 
+  /**
+   * Convert hoodie-common KeyGenerator to SparkKeyGeneratorInterface implement.
+   */
+  public static String convertToSparkKeyGenerator(String keyGeneratorClassName) {
+    return COMMON_TO_SPARK_KEYGENERATOR.getOrDefault(keyGeneratorClassName, keyGeneratorClassName);
+  }
 }

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/functional/TestHoodieMetadataBase.java

@@ -22,6 +22,7 @@
 import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy;
 import org.apache.hudi.common.model.HoodieTableType;
 import org.apache.hudi.common.model.WriteOperationType;
+import org.apache.hudi.common.table.HoodieTableConfig;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
 import org.apache.hudi.common.table.view.FileSystemViewStorageConfig;
 import org.apache.hudi.common.testutils.HoodieMetadataTestTable;
@@ -33,6 +34,7 @@
 import org.apache.hudi.config.metrics.HoodieMetricsConfig;
 import org.apache.hudi.config.metrics.HoodieMetricsGraphiteConfig;
 import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.keygen.SimpleKeyGenerator;
 import org.apache.hudi.metadata.HoodieTableMetadata;
 import org.apache.hudi.metadata.HoodieTableMetadataWriter;
 import org.apache.hudi.metadata.SparkHoodieBackedTableMetadataWriter;
@@ -50,6 +52,7 @@
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Properties;
 
 import static java.util.Arrays.asList;
 import static java.util.Collections.emptyList;
@@ -268,6 +271,8 @@ protected HoodieWriteConfig.Builder getWriteConfigBuilder(HoodieFailedWritesClea
 
   protected HoodieWriteConfig.Builder getWriteConfigBuilder(HoodieFailedWritesCleaningPolicy policy, boolean autoCommit, boolean useFileListingMetadata,
                                                             boolean enableMetrics, boolean enableFullScan) {
+    Properties properties = new Properties();
+    properties.put(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME.key(), SimpleKeyGenerator.class.getName());
     return HoodieWriteConfig.newBuilder().withPath(basePath).withSchema(TRIP_EXAMPLE_SCHEMA)
         .withParallelism(2, 2).withDeleteParallelism(2).withRollbackParallelism(2).withFinalizeWriteParallelism(2)
         .withAutoCommit(autoCommit)
@@ -287,7 +292,8 @@ protected HoodieWriteConfig.Builder getWriteConfigBuilder(HoodieFailedWritesClea
         .withMetricsConfig(HoodieMetricsConfig.newBuilder().on(enableMetrics)
             .withExecutorMetrics(true).build())
         .withMetricsGraphiteConfig(HoodieMetricsGraphiteConfig.newBuilder()
-            .usePrefix("unit-test").build());
+            .usePrefix("unit-test").build())
+        .withProperties(properties);
   }
 
 }

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/keygen/TestCustomKeyGenerator.java

@@ -33,6 +33,8 @@
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
+import java.io.PrintWriter;
+import java.io.StringWriter;
 
 public class TestCustomKeyGenerator extends KeyGeneratorTestUtilities {
 
@@ -122,6 +124,13 @@ private TypedProperties getPropertiesForNonPartitionedKeyGen(boolean useKeyGener
     return properties;
   }
 
+  private String stackTraceToString(Throwable e) {
+    StringWriter sw = new StringWriter();
+    PrintWriter pw = new PrintWriter(sw);
+    e.printStackTrace(pw);
+    return sw.toString();
+  }
+
   @Test
   public void testSimpleKeyGeneratorWithKeyGeneratorClass() throws IOException {
     testSimpleKeyGenerator(getPropertiesForSimpleKeyGen(true));
@@ -259,7 +268,7 @@ public void testNoRecordKeyFieldProp(boolean useKeyGeneratorClassName) {
             .getMessage()
             .contains("Property hoodie.datasource.write.recordkey.field not found"));
       } else {
-        Assertions.assertTrue(e.getMessage().contains("Property hoodie.datasource.write.recordkey.field not found"));
+        Assertions.assertTrue(stackTraceToString(e).contains("Property hoodie.datasource.write.recordkey.field not found"));
       }
 
     }
@@ -282,7 +291,7 @@ public void testNoRecordKeyFieldProp(boolean useKeyGeneratorClassName) {
             .getMessage()
             .contains("Property hoodie.datasource.write.recordkey.field not found"));
       } else {
-        Assertions.assertTrue(e.getMessage().contains("Property hoodie.datasource.write.recordkey.field not found"));
+        Assertions.assertTrue(stackTraceToString(e).contains("Property hoodie.datasource.write.recordkey.field not found"));
       }
     }
   }

hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/HoodieClientTestHarness.java

@@ -249,7 +249,15 @@ protected void initMetaClient() throws IOException {
     initMetaClient(getTableType());
   }
 
+  protected void initMetaClient(Properties properties) throws IOException {
+    initMetaClient(getTableType(), properties);
+  }
+
   protected void initMetaClient(HoodieTableType tableType) throws IOException {
+    initMetaClient(tableType, new Properties());
+  }
+
+  protected void initMetaClient(HoodieTableType tableType, Properties properties) throws IOException {
     if (basePath == null) {
       throw new IllegalStateException("The base path has not been initialized.");
     }
@@ -258,7 +266,10 @@ protected void initMetaClient(HoodieTableType tableType) throws IOException {
       throw new IllegalStateException("The Spark context has not been initialized.");
     }
 
-    metaClient = HoodieTestUtils.init(hadoopConf, basePath, tableType);
+    if (tableName != null && !tableName.isEmpty()) {
+      properties.put(HoodieTableConfig.NAME.key(), tableName);
+    }
+    metaClient = HoodieTestUtils.init(hadoopConf, basePath, tableType, properties);
   }
 
   protected Properties getPropertiesForKeyGen() {

hudi-common/src/main/java/org/apache/hudi/common/config/HoodieConfig.java

@@ -74,6 +74,10 @@ public <T> void setDefaultValue(ConfigProperty<T> configProperty, T defaultVal)
     }
   }
 
+  public Boolean contains(String key) {
+    return props.containsKey(key);
+  }
+
   public <T> boolean contains(ConfigProperty<T> configProperty) {
     if (props.containsKey(configProperty.key())) {
       return true;
@@ -135,7 +139,7 @@ public <T> Boolean getBoolean(ConfigProperty<T> configProperty) {
   public <T> boolean getBooleanOrDefault(ConfigProperty<T> configProperty) {
     Option<Object> rawValue = getRawValue(configProperty);
     return rawValue.map(v -> Boolean.parseBoolean(v.toString()))
-            .orElse((Boolean) configProperty.defaultValue());
+            .orElse(Boolean.parseBoolean(configProperty.defaultValue().toString()));
   }
 
   public <T> Long getLong(ConfigProperty<T> configProperty) {

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

@@ -31,6 +31,7 @@
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.exception.HoodieIOException;
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
 
 import org.apache.avro.Schema;
 import org.apache.hadoop.fs.FSDataInputStream;
@@ -161,6 +162,9 @@ public class HoodieTableConfig extends HoodieConfig {
       .noDefaultValue()
       .withDocumentation("Key Generator class property for the hoodie table");
 
+  public static final ConfigProperty<String> URL_ENCODE_PARTITIONING = KeyGeneratorOptions.URL_ENCODE_PARTITIONING;
+  public static final ConfigProperty<String> HIVE_STYLE_PARTITIONING_ENABLE = KeyGeneratorOptions.HIVE_STYLE_PARTITIONING_ENABLE;
+
   public static final String NO_OP_BOOTSTRAP_INDEX_CLASS = NoOpBootstrapIndex.class.getName();
 
   public HoodieTableConfig(FileSystem fs, String metaPath, String payloadClassName) {
@@ -363,6 +367,18 @@ public String getRecordKeyFieldProp() {
     return getString(RECORDKEY_FIELDS);
   }
 
+  public String getKeyGeneratorClassName() {
+    return getString(KEY_GENERATOR_CLASS_NAME);
+  }
+
+  public String getHiveStylePartitioningEnable() {
+    return getString(HIVE_STYLE_PARTITIONING_ENABLE);
+  }
+
+  public String getUrlEncodePartitoning() {
+    return getString(URL_ENCODE_PARTITIONING);
+  }
+
   public Map<String, String> propsMap() {
     return props.entrySet().stream()
         .collect(Collectors.toMap(e -> String.valueOf(e.getKey()), e -> String.valueOf(e.getValue())));

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

@@ -629,6 +629,8 @@ public static class PropertyBuilder {
     private Boolean bootstrapIndexEnable;
     private Boolean populateMetaFields;
     private String keyGeneratorClassProp;
+    private Boolean hiveStylePartitioningEnable;
+    private Boolean urlEncodePartitioning;
 
     private PropertyBuilder() {
 
@@ -717,6 +719,16 @@ public PropertyBuilder setKeyGeneratorClassProp(String keyGeneratorClassProp) {
       return this;
     }
 
+    public PropertyBuilder setHiveStylePartitioningEnable(Boolean hiveStylePartitioningEnable) {
+      this.hiveStylePartitioningEnable = hiveStylePartitioningEnable;
+      return this;
+    }
+
+    public PropertyBuilder setUrlEncodePartitioning(Boolean urlEncodePartitioning) {
+      this.urlEncodePartitioning = urlEncodePartitioning;
+      return this;
+    }
+
     public PropertyBuilder fromMetaClient(HoodieTableMetaClient metaClient) {
       return setTableType(metaClient.getTableType())
         .setTableName(metaClient.getTableConfig().getTableName())
@@ -778,6 +790,12 @@ public PropertyBuilder fromProperties(Properties properties) {
       if (hoodieConfig.contains(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME)) {
         setKeyGeneratorClassProp(hoodieConfig.getString(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME));
       }
+      if (hoodieConfig.contains(HoodieTableConfig.HIVE_STYLE_PARTITIONING_ENABLE)) {
+        setHiveStylePartitioningEnable(hoodieConfig.getBoolean(HoodieTableConfig.HIVE_STYLE_PARTITIONING_ENABLE));
+      }
+      if (hoodieConfig.contains(HoodieTableConfig.URL_ENCODE_PARTITIONING)) {
+        setUrlEncodePartitioning(hoodieConfig.getBoolean(HoodieTableConfig.URL_ENCODE_PARTITIONING));
+      }
       return this;
     }
 
@@ -841,6 +859,12 @@ public Properties build() {
       if (null != keyGeneratorClassProp) {
         tableConfig.setValue(HoodieTableConfig.KEY_GENERATOR_CLASS_NAME, keyGeneratorClassProp);
       }
+      if (null != hiveStylePartitioningEnable) {
+        tableConfig.setValue(HoodieTableConfig.HIVE_STYLE_PARTITIONING_ENABLE, Boolean.toString(hiveStylePartitioningEnable));
+      }
+      if (null != urlEncodePartitioning) {
+        tableConfig.setValue(HoodieTableConfig.URL_ENCODE_PARTITIONING, Boolean.toString(urlEncodePartitioning));
+      }
       return tableConfig.getProps();
     }
 

hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieCommonTestHarness.java

@@ -34,12 +34,17 @@
  */
 public class HoodieCommonTestHarness {
 
+  protected String tableName = null;
   protected String basePath = null;
   protected transient HoodieTestDataGenerator dataGen = null;
   protected transient HoodieTableMetaClient metaClient;
   @TempDir
   public java.nio.file.Path tempDir;
 
+  protected void setTableName(String tableName) {
+    this.tableName = tableName;
+  }
+
   /**
    * Initializes basePath.
    */