feat(schema): Add read support for shredded

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

Author: voonhous

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)

hudi-hadoop-common/src/main/java/org/apache/hudi/avro/HoodieAvroWriteSupport.java

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.hudi.avro;
+
+import org.apache.hudi.common.schema.HoodieSchema;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+
+/**
+ * Interface for shredding variant values at write time.
+ * <p>
+ * Implementations parse variant binary data (value + metadata bytes) and produce
+ * a shredded {@link GenericRecord} with typed_value columns populated according
+ * to the shredding schema.
+ * <p>
+ * This interface allows the variant binary parsing logic (which may depend on
+ * engine-specific libraries like Spark's variant module) to be loaded via reflection,
+ * keeping the core write support free of engine-specific dependencies.
+ */
+public interface VariantShreddingProvider {
+
+  /**
+   * Transform an unshredded variant GenericRecord into a shredded one.
+   * <p>
+   * The input record is expected to have:
+   * <ul>
+   *   <li>{@code value}: ByteBuffer containing the variant value binary</li>
+   *   <li>{@code metadata}: ByteBuffer containing the variant metadata binary</li>
+   * </ul>
+   * <p>
+   * The output record should conform to {@code shreddedSchema} and have:
+   * <ul>
+   *   <li>{@code value}: ByteBuffer or null (null when typed_value captures the full value)</li>
+   *   <li>{@code metadata}: ByteBuffer (always present)</li>
+   *   <li>{@code typed_value}: the typed representation extracted from the variant binary,
+   *       or null if the variant type does not match the typed_value schema</li>
+   * </ul>
+   *
+   * @param unshreddedVariant GenericRecord with {value: ByteBuffer, metadata: ByteBuffer}
+   * @param shreddedSchema    target Avro schema with {value: nullable ByteBuffer, metadata: ByteBuffer, typed_value: type}
+   * @param variantSchema     HoodieSchema.Variant containing the shredding schema information
+   * @return a GenericRecord conforming to shreddedSchema with typed_value populated where possible
+   */
+  GenericRecord shredVariantRecord(
+      GenericRecord unshreddedVariant,
+      Schema shreddedSchema,
+      HoodieSchema.Variant variantSchema);
+}

hudi-hadoop-common/src/main/java/org/apache/hudi/avro/VariantShreddingProvider.java

@@ -49,6 +49,7 @@
 import java.util.Properties;
 
 import static org.apache.hudi.common.config.HoodieStorageConfig.HFILE_WRITER_TO_ALLOW_DUPLICATES;
+import static org.apache.hudi.common.config.HoodieStorageConfig.PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS;
 import static org.apache.parquet.avro.HoodieAvroParquetSchemaConverter.getAvroSchemaConverter;
 
 public class HoodieAvroFileWriterFactory extends HoodieFileWriterFactory {
@@ -123,9 +124,37 @@ protected HoodieFileWriter newOrcFileWriter(
   private HoodieAvroWriteSupport getHoodieAvroWriteSupport(HoodieSchema schema,
                                                            HoodieConfig config, boolean enableBloomFilter) {
     Option<BloomFilter> filter = enableBloomFilter ? Option.of(createBloomFilter(config)) : Option.empty();
+    HoodieSchema effectiveSchema = HoodieAvroWriteSupport.generateEffectiveSchema(schema, config);
+    Properties props = config.getProps();
+    // Auto-detect variant shredding provider from classpath if not explicitly configured
+    if (!props.containsKey(PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key())) {
+      String detected = detectShreddingProvider();
+      if (detected != null) {
+        props.setProperty(PARQUET_VARIANT_SHREDDING_PROVIDER_CLASS.key(), detected);
+      }
+    }
     return (HoodieAvroWriteSupport) ReflectionUtils.loadClass(
         config.getStringOrDefault(HoodieStorageConfig.HOODIE_AVRO_WRITE_SUPPORT_CLASS),
         new Class<?>[] {MessageType.class, HoodieSchema.class, Option.class, Properties.class},
-        getAvroSchemaConverter(storage.getConf().unwrapAs(Configuration.class)).convert(schema), schema, filter, config.getProps());
+        getAvroSchemaConverter(storage.getConf().unwrapAs(Configuration.class)).convert(effectiveSchema), schema, filter, props);
+  }
+
+  /**
+   * Auto-detect a {@link org.apache.hudi.avro.VariantShreddingProvider} implementation
+   * available on the classpath. Returns the fully-qualified class name if found, or null.
+   */
+  private static String detectShreddingProvider() {
+    String[] candidates = {
+        "org.apache.hudi.variant.Spark4VariantShreddingProvider"
+    };
+    for (String candidate : candidates) {
+      try {
+        Class.forName(candidate);
+        return candidate;
+      } catch (ClassNotFoundException e) {
+        // not on classpath, try next
+      }
+    }
+    return null;
   }
 }

hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/hadoop/HoodieAvroFileWriterFactory.java

@@ -360,7 +360,7 @@ class HoodieSparkSqlTestBase extends FunSuite with BeforeAndAfterAll {
     }
   }
 
-  protected def withRecordType(recordTypes: Seq[HoodieRecordType] = Seq(HoodieRecordType.SPARK),
+  protected def withRecordType(recordTypes: Seq[HoodieRecordType] = Seq(HoodieRecordType.AVRO, HoodieRecordType.SPARK),
                                recordConfig: Map[HoodieRecordType, Map[String, String]] = Map.empty)(f: => Unit) {
     // TODO HUDI-5264 Test parquet log with avro record in spark sql test
     recordTypes.foreach { recordType =>

hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/common/HoodieSparkSqlTestBase.scala

@@ -194,17 +194,24 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
            | tblproperties (
            |  primaryKey = 'id',
            |  type = 'cow',
-           |  preCombineField = 'ts',
-           |  'hoodie.parquet.variant.write.shredding.enabled' = 'true',
-           |  'hoodie.parquet.variant.force.shredding.schema.for.test' = 'a int, b string'
+           |  preCombineField = 'ts'
            | )
         """.stripMargin)
 
+      spark.sql("set hoodie.parquet.variant.write.shredding.enabled = true")
+      // TODO: We need to use hoodie.parquet.variant.allow.reading.shredded instead of line 204
+//      spark.sql("set hoodie.parquet.variant.allow.reading.shredded = true")
+      spark.sql("set spark.sql.variant.allowReadingShredded = true");
+      spark.sql("set hoodie.parquet.variant.force.shredding.schema.for.test = a int, b string")
+
       spark.sql(
         s"""
            |insert into $tableName values
            |  (1, 'row1', parse_json('{"a": 1, "b": "hello"}'), 1000)
         """.stripMargin)
+      checkAnswer(s"select id, name, cast(v as string), ts from $tableName order by id")(
+        Seq(1, "row1", "{\"a\":1,\"b\":\"hello\"}", 1000)
+      )
 
       // Verify parquet schema has shredded structure with typed_value
       val parquetFiles = listDataParquetFiles(tmp.getCanonicalPath)
@@ -253,6 +260,10 @@ class TestVariantDataType extends HoodieSparkSqlTestBase {
            |  (1, 'row1', parse_json('{"a": 1, "b": "hello"}'), 1000)
               """.stripMargin)
 
+      checkAnswer(s"select id, name, cast(v as string), ts from $tableName order by id")(
+        Seq(1, "row1", "{\"a\":1,\"b\":\"hello\"}", 1000)
+      )
+
       // Verify parquet schema does NOT have typed_value
       val parquetFiles = listDataParquetFiles(tmp.getCanonicalPath)
       assert(parquetFiles.nonEmpty, "Should have at least one data parquet file")