[HUDI-4093] If the value of the partition column is null or abnormal,…

… the insert operation throws NPE.
apache · Sep 17, 2022 · 6c89bfb · 6c89bfb
1 parent f1caa3a
commit 6c89bfb
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 3 deletions.
diff --git a/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java b/hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java
@@ -610,7 +610,7 @@ public static Option<String> getNullableValAsString(GenericRecord rec, String fi
    * @return field value either converted (for certain data types) or as it is.
    */
   public static Object convertValueForSpecificDataTypes(Schema fieldSchema, Object fieldValue, boolean consistentLogicalTimestampEnabled) {
-    if (fieldSchema == null) {
+    if (fieldSchema == null || fieldValue == null) {
       return fieldValue;
     }
 

diff --git a/.../src/main/scala/org/apache/spark/sql/execution/datasources/Spark3ParsePartitionUtil.scala b/.../src/main/scala/org/apache/spark/sql/execution/datasources/Spark3ParsePartitionUtil.scala
@@ -63,9 +63,16 @@ object Spark3ParsePartitionUtil extends SparkParsePartitionUtil {
       (dateFormatter, timestampFormatter)
     })
 
-    val (partitionValues, _) = parsePartition(path, typeInference, basePaths, userSpecifiedDataTypes,
-      validatePartitionValues, tz.toZoneId, dateFormatter, timestampFormatter)
+    var partitionStr  = path.toString
+    userSpecifiedDataTypes.keySet.foreach { name =>
+      val dataType = userSpecifiedDataTypes.get(name).getOrElse("")
+      if (!dataType.isInstanceOf[StringType]) {
+        partitionStr = partitionStr.replace(s"$name=default", s"$name=__HIVE_DEFAULT_PARTITION__")
+      }
+    }
 
+    val (partitionValues, _) = parsePartition(new Path(partitionStr), typeInference, basePaths, userSpecifiedDataTypes,
+      validatePartitionValues, tz.toZoneId, dateFormatter, timestampFormatter)
     partitionValues.map {
       case PartitionValues(columnNames: Seq[String], typedValues: Seq[TypedPartValue]) =>
         val rowValues = columnNames.zip(typedValues).map { case (columnName, typedValue) =>