feat(schema): Add read + write support for shredded for AVRO

- Added support to write shredded types for HoodieRecordType.AVRO
- Added functional tests for testing newly added configs

Author: voonhous

hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/HoodieInternalRowUtils.scala

@@ -22,6 +22,7 @@ import org.apache.hudi.avro.HoodieAvroUtils.{createFullName, toJavaDate}
 import org.apache.hudi.exception.HoodieException
 import org.apache.hudi.common.schema.HoodieSchema
 import org.apache.hudi.HoodieSchemaConversionUtils.convertHoodieSchemaToStructType
+import org.apache.hudi.HoodieSparkUtils.sparkAdapter
 import org.apache.spark.sql.HoodieCatalystExpressionUtils.generateUnsafeProjection
 import org.apache.spark.sql.HoodieUnsafeRowUtils.{NestedFieldPath, composeNestedFieldPath}
 import org.apache.spark.sql.catalyst.expressions.{SpecificInternalRow, UnsafeArrayData, UnsafeProjection, UnsafeRow}
@@ -398,11 +399,47 @@ object HoodieInternalRowUtils {
         (fieldUpdater, ordinal, value) =>
           fieldUpdater.set(ordinal, CatalystTypeConverters.convertToCatalyst(java.sql.Date.valueOf(value.toString)))
 
+      // Handle conversion from VariantType to variant struct representation
+      case (newStructType: StructType, _) if sparkAdapter.isVariantType(prevDataType) && looksLikeVariantStruct(newStructType) =>
+        (fieldUpdater, ordinal, value) => {
+          if (value == null) {
+            fieldUpdater.setNullAt(ordinal)
+          } else {
+            val row = sparkAdapter.convertVariantToStruct(value, newStructType)
+            fieldUpdater.set(ordinal, row)
+          }
+        }
+
+      // Handle conversion from variant struct representation to VariantType
+      case (_, prevStructType: StructType) if sparkAdapter.isVariantType(newDataType) && looksLikeVariantStruct(prevStructType) =>
+        (fieldUpdater, ordinal, value) => {
+          if (value == null) {
+            fieldUpdater.setNullAt(ordinal)
+          } else {
+            val row = value.asInstanceOf[InternalRow]
+            val variant = sparkAdapter.convertStructToVariant(row, prevStructType)
+            fieldUpdater.set(ordinal, variant)
+          }
+        }
+
       case (_, _) =>
         throw new IllegalArgumentException(s"$prevDataType and $newDataType are incompatible")
     }
   }
 
+  /**
+   * Checks if a StructType looks like a variant representation (has value and metadata binary fields).
+   * This is a structural check that doesn't rely on metadata, useful during schema reconciliation
+   * when toggling between shredded/unshredded formats or merging data with different representations.
+   */
+  private def looksLikeVariantStruct(structType: StructType): Boolean = {
+    structType.fields.length >= 2 &&
+      structType.fieldNames.contains("value") &&
+      structType.fieldNames.contains("metadata") &&
+      structType("value").dataType == BinaryType &&
+      structType("metadata").dataType == BinaryType
+  }
+
   private def lookupRenamedField(newFieldName: String,
                                  newFieldQualifiedName: String,
                                  renamedColumnsMap: JMap[String, String]): String = {

hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala

@@ -483,4 +483,30 @@ trait SparkAdapter extends Serializable {
     shreddedStructType: StructType,
     writeStruct: Consumer[InternalRow]
   ): BiConsumer[SpecializedGetters, Integer]
+
+  /**
+   * Converts a Variant value to a struct representation (value and metadata binary fields).
+   * This is used during schema reconciliation when converting from VariantType to StructType.
+   *
+   * For Spark 4.x, extracts value and metadata bytes from VariantVal and creates an InternalRow.
+   * For Spark 3.x, this throws UnsupportedOperationException.
+   *
+   * @param variantValue The variant value object
+   * @param structType The target StructType with value and metadata fields
+   * @return InternalRow with value and metadata fields populated
+   */
+  def convertVariantToStruct(variantValue: Any, structType: StructType): InternalRow
+
+  /**
+   * Converts a struct representation (value and metadata binary fields) to a Variant value.
+   * This is used during schema reconciliation when converting from StructType to VariantType.
+   *
+   * For Spark 4.x, extracts value and metadata bytes from InternalRow and creates a VariantVal.
+   * For Spark 3.x, this throws UnsupportedOperationException.
+   *
+   * @param structRow The InternalRow with value and metadata fields
+   * @param structType The StructType of the struct
+   * @return Variant value object
+   */
+  def convertStructToVariant(structRow: InternalRow, structType: StructType): Any
 }

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

@@ -198,6 +198,16 @@ public class HoodieStorageConfig extends HoodieConfig {
           + "When disabled, only unshredded variant data can be read. "
           + "Equivalent to Spark's spark.sql.variant.allowReadingShredded.");
 
+  public static final ConfigProperty<String> PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS = ConfigProperty
+      .key("hoodie.parquet.variant.shredding.provider.class")
+      .noDefaultValue()
+      .markAdvanced()
+      .sinceVersion("1.1.0")
+      .withDocumentation("Fully-qualified class name of the VariantShreddingProvider implementation "
+          + "used to shred variant values at write time in the Avro record path. "
+          + "The provider parses variant binary data and populates typed_value columns. "
+          + "When not set, the provider is auto-detected from the classpath.");
+
   public static final ConfigProperty<Boolean> WRITE_UTC_TIMEZONE = ConfigProperty
       .key("hoodie.parquet.write.utc-timezone.enabled")
       .defaultValue(true)