feat(variant): Add support to write shredded variants for HoodieRecordType.AVRO (apache#18065)

Adds write support for shredded Variant on the AVRO record path. Variants are
decomposed into a typed_value column for better compression and column pruning.
Shredding is delegated to a VariantShreddingProvider (in hudi-common, loaded
reflectively), with a Spark 4 implementation built on Spark's
VariantShreddingWriter.

- Config: new advanced hoodie.parquet.variant.shredding.provider.class
  (sinceVersion 1.3.0), auto-detected from the classpath when unset. Gated by
  the existing write.shredding.enabled; reads governed by allow.reading.shredded.
- Write path: HoodieAvroFileWriterFactory builds the Parquet schema from the
  effective (shredded) schema so value is nullable and typed_value is present;
  injects the provider on a copied Properties so the shared config is never
  mutated. Fails fast when shredding is required but no provider is available,
  and on the not-yet-supported read-then-reshred path (apache#18931).
- Spark read: BaseSpark4Adapter recognizes both unshredded (2-field) and
  shredded (3-field) physical layouts as Variant. Spark 4.0.x keeps a reorder
  workaround for its hardcoded [value, metadata] converter; 4.1+ reads by name
  (SPARK-54410), removable later (apache#18935). No session SQLConf mutation - Spark's
  converter reads the flag via SQLConf.get.
- DDL parsing: shared StringUtils.splitTopLevelCommas (paren-aware) keeps
  decimal(15, 1) intact, reused by both the shredding DDL parser and the vector
  column metadata parser.
- Tests: Spark integration (TestVariantDataType) plus unit coverage for the
  forced-shredding DDL (incl. decimal), shredded schema shape, the
  read-then-reshred guard, and splitTopLevelCommas.

Limitations: shredded writes require Spark 4.0+; reading shredded data back
requires Spark 4.1+ (Spark 4.0 read deferred, apache#18931).

Closes apache#18066
Closes apache#17748

Author: voonhous

hudi-common/src/main/java/org/apache/hudi/avro/VariantShreddingProvider.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-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java

@@ -235,11 +235,12 @@ public class HoodieStorageConfig extends HoodieConfig {
       .noDefaultValue()
       .markAdvanced()
       .sinceVersion("1.1.0")
-      .withDocumentation("Forces a specific shredding schema for all variant columns, intended for testing. "
+      .withDocumentation("Test-only: forces a specific shredding schema for all variant columns. Not intended "
+          + "for production use; it exists solely to exercise the shredding write path in tests, mirroring "
+          + "Spark's internal spark.sql.variant.forceShreddingSchemaForTest. "
           + "The value should be a DDL-format schema string (e.g., 'a int, b string, c decimal(15, 1)'). "
           + "When set and write shredding is enabled, this schema overrides the schema-driven shredding "
-          + "configuration for all variant columns. "
-          + "Equivalent to Spark's spark.sql.variant.forceShreddingSchemaForTest.");
+          + "configuration for all variant columns.");
 
   public static final ConfigProperty<Boolean> PARQUET_VARIANT_ALLOW_READING_SHREDDED = ConfigProperty
       .key("hoodie.parquet.variant.allow.reading.shredded")
@@ -250,6 +251,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.3.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)
@@ -596,11 +607,6 @@ public Builder parquetVariantWriteShreddingEnabled(boolean enabled) {
       return this;
     }
 
-    public Builder parquetVariantForceShreddingSchemaForTest(String schemaString) {
-      storageConfig.setValue(PARQUET_VARIANT_FORCE_SHREDDING_SCHEMA_FOR_TEST, schemaString);
-      return this;
-    }
-
     public Builder parquetVariantAllowReadingShredded(boolean allowed) {
       storageConfig.setValue(PARQUET_VARIANT_ALLOW_READING_SHREDDED, String.valueOf(allowed));
       return this;

hudi-common/src/main/java/org/apache/hudi/common/schema/HoodieSchema.java

@@ -20,6 +20,7 @@
 
 import org.apache.hudi.avro.AvroSchemaUtils;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
 import org.apache.hudi.common.util.ValidationUtils;
 import org.apache.hudi.common.util.collection.Pair;
 import org.apache.hudi.exception.HoodieAvroSchemaException;
@@ -282,35 +283,18 @@ public static String buildVectorColumnsMetadataValue(HoodieSchema schema) {
    * @return set of field names (preserves insertion order), or empty set if input is null / empty
    */
   public static Set<String> parseVectorColumnNames(String footerValue) {
-    if (footerValue == null || footerValue.isEmpty()) {
-      return Collections.emptySet();
-    }
     LinkedHashSet<String> names = new LinkedHashSet<>();
-    int depth = 0;
-    int start = 0;
-    for (int i = 0; i < footerValue.length(); i++) {
-      char c = footerValue.charAt(i);
-      if (c == '(') {
-        depth++;
-      } else if (c == ')') {
-        depth--;
-      } else if (c == ',' && depth == 0) {
-        addVectorColumnName(footerValue, start, i, names);
-        start = i + 1;
+    // Split on top-level commas only so the comma inside a VECTOR(dim, elemType) descriptor is
+    // not mistaken for a column separator.
+    for (String entry : StringUtils.splitTopLevelCommas(footerValue)) {
+      int colon = entry.indexOf(':');
+      if (colon > 0) {
+        names.add(entry.substring(0, colon).trim());
       }
     }
-    addVectorColumnName(footerValue, start, footerValue.length(), names);
     return names;
   }
 
-  private static void addVectorColumnName(String s, int start, int end, Set<String> names) {
-    int colon = s.indexOf(':', start);
-    if (colon > start && colon < end) {
-      names.add(s.substring(start, colon).trim());
-    }
-  }
-
-
   private Schema avroSchema;
   private HoodieSchemaType type;
   private transient List<HoodieSchemaField> fields;