Fix update tests

Author: voonhous

hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala

@@ -215,9 +215,7 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
     })
   }
 
-  // TODO: Re-enable once variant merge/update support is fully implemented #18069
-  // Currently fails with NullPointerException in AvroDeserializer during merge operations
-  ignore("Test Variant Shredding with Update Operation") {
+  test("Test Variant Shredding with Update Operation") {
     assume(HoodieSparkUtils.gteqSpark4_0, "Variant type requires Spark 4.0 or higher")
 
     withRecordType()(withTempDir { tmp =>
@@ -277,9 +275,7 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
     })
   }
 
-  // TODO: Re-enable once variant merge/update support is fully implemented #18069
-  // Currently fails with NullPointerException in AvroDeserializer during merge operations
-  ignore("Test Variant Shredding with Merge Operation") {
+  test("Test Variant Shredding with Merge Operation") {
     assume(HoodieSparkUtils.gteqSpark4_0, "Variant type requires Spark 4.0 or higher")
 
     withRecordType()(withTempDir { tmp =>

hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala

@@ -237,17 +237,22 @@ private[sql] class AvroDeserializer(rootAvroType: Schema,
           val record = value.asInstanceOf[IndexedRecord]
 
           val valueBuffer = record.get(valueIdx).asInstanceOf[ByteBuffer]
-          val valueBytes = new Array[Byte](valueBuffer.remaining)
-          valueBuffer.get(valueBytes)
-          valueBuffer.rewind()
+          // value field can be null during merging
+          if (valueBuffer == null) {
+            updater.setNullAt(ordinal)
+          } else {
+            val valueBytes = new Array[Byte](valueBuffer.remaining)
+            valueBuffer.get(valueBytes)
+            valueBuffer.rewind()
 
-          val metadataBuffer = record.get(metadataIdx).asInstanceOf[ByteBuffer]
-          val metadataBytes = new Array[Byte](metadataBuffer.remaining)
-          metadataBuffer.get(metadataBytes)
-          metadataBuffer.rewind()
+            val metadataBuffer = record.get(metadataIdx).asInstanceOf[ByteBuffer]
+            val metadataBytes = new Array[Byte](metadataBuffer.remaining)
+            metadataBuffer.get(metadataBytes)
+            metadataBuffer.rewind()
 
-          val variant = new VariantVal(valueBytes, metadataBytes)
-          updater.set(ordinal, variant)
+            val variant = new VariantVal(valueBytes, metadataBytes)
+            updater.set(ordinal, variant)
+          }
 
       case (RECORD, st: StructType) =>
         // Avro datasource doesn't accept filters with nested attributes. See SPARK-32328.