sparkKeyGenUtils

public class SparkKeyGenUtils {

  public static String getPartitionColumns(Map<String, String> parameters) throws IOException {
    TypedProperties props = new TypedProperties();
    props.putAll(parameters);
    return getPartitionColumns(props);
  }

  public static String getPartitionColumns(TypedProperties props) throws IOException {
    KeyGenerator keyGen = HoodieSparkKeyGeneratorFactory.createKeyGenerator(props);
    return getPartitionColumns(keyGen);
  }

  public static String getPartitionColumns(KeyGenerator keyGenerator) {
    if (keyGenerator instanceof BaseKeyGenerator) {
      if (keyGenerator instanceof CustomKeyGenerator || keyGenerator instanceof CustomAvroKeyGenerator) {
        BaseKeyGenerator baseKeyGenerator = (BaseKeyGenerator) keyGenerator;
        List<String> partitionCols = baseKeyGenerator.getPartitionPathFields().stream().map(partitionField ->  {
          String[] splits = partitionField.split(CustomAvroKeyGenerator.SPLIT_REGEX);
          if (splits.length != 2) {
            throw new IllegalArgumentException("Partition path config is not in right format for CustomKeyGenerator " + partitionField);
          } else {
            return splits[0];
          }
        }).collect(Collectors.toList());
        return String.join(",", partitionCols);
      } else {
        return String.join(",", ((BaseKeyGenerator)keyGenerator).getPartitionPathFields());
      }
    } else {
      return null;
    }
  }
}