[GLUTEN-11550][VL][UT] Enable Variant test suites

Enable GlutenVariantEndToEndSuite, GlutenVariantShreddingSuite, and
GlutenParquetVariantShreddingSuite for both spark40 and spark41.

Fixes:
1. VeloxValidatorApi: Detect variant shredded structs (produced by
   Spark's PushVariantIntoScan) by checking __VARIANT_METADATA_KEY
   metadata. Triggers fallback to Spark's native Parquet reader.

2. Spark41Shims: Detect Parquet variant logical type annotations and
   fall back to vanilla Spark when PARQUET_IGNORE_VARIANT_ANNOTATION
   is not set, since Velox native reader does not check variant
   annotations.

3. pom.xml: Add -Dfile.encoding=UTF-8 to test JVM args.

   On JDK 17 and earlier, java.nio.charset.Charset.defaultCharset()
   is determined by the OS locale. On CI containers (centos-8/9)
   where LANG=C, the default charset is US-ASCII (ANSI_X3.4-1968).
   JDK 18+ changed this via JEP 400 (https://openjdk.org/jeps/400)
   to always default to UTF-8 regardless of locale.

   Spark's VariantUtil.getString() uses new String(byte[], offset,
   length) without specifying charset, which decodes using the JVM
   default charset. With JDK 17 + LANG=C, UTF-8 encoded multi-byte
   characters (e.g. Chinese) are decoded as ASCII, producing garbled
   output.

   Call chain:
     VariantEndToEndSuite.check("\"你好，世界...\"")
     -> to_json(parse_json(col("v")))
     -> StructsToJsonEvaluator.evaluate()
     -> JacksonGenerator.write(VariantVal)
     -> VariantVal.toJson()
     -> Variant.toJsonImpl()
     -> VariantUtil.getString(byte[], pos)
     -> new String(value, start, length)  // no charset specified
   https://github.com/apache/spark/blob/v4.0.1/common/variant/src/main/java/org/apache/spark/types/variant/VariantUtil.java#L508
   https://github.com/apache/spark/blob/v4.1.0/common/variant/src/main/java/org/apache/spark/types/variant/VariantUtil.java#L509

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: baibaichen

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala

@@ -208,6 +208,14 @@ object VeloxBackendSettings extends BackendSettingsApi {
         return None
       }
       val fileLimit = GlutenConfig.get.parquetMetadataFallbackFileLimit
+      // Variant annotation check is always performed (not gated by
+      // parquetMetadataValidationEnabled) because it is a correctness issue.
+      val variantAnnotationResult =
+        ParquetMetadataUtils.validateVariantAnnotation(rootPaths, hadoopConf, fileLimit)
+      if (variantAnnotationResult.isDefined) {
+        return variantAnnotationResult.map(
+          reason => s"Detected unsupported metadata in parquet files: $reason")
+      }
       val parquetOptions = new ParquetOptions(CaseInsensitiveMap(properties), SQLConf.get)
       val parquetMetadataValidationResult =
         ParquetMetadataUtils.validateMetadata(rootPaths, hadoopConf, parquetOptions, fileLimit)

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxValidatorApi.scala

@@ -121,6 +121,15 @@ object VeloxValidatorApi {
       case map: MapType =>
         validateSchema(map.keyType).orElse(validateSchema(map.valueType))
       case struct: StructType =>
+        // Detect variant shredded struct produced by Spark's PushVariantIntoScan.
+        // These structs have all fields annotated with __VARIANT_METADATA_KEY metadata.
+        // Velox cannot read the variant shredding encoding in Parquet files.
+        if (
+          struct.fields.nonEmpty &&
+          struct.fields.forall(_.metadata.contains("__VARIANT_METADATA_KEY"))
+        ) {
+          return Some(s"Variant shredded struct is not supported: $struct")
+        }
         struct.foreach {
           field =>
             val reason = validateSchema(field.dataType)

backends-velox/src/main/scala/org/apache/gluten/utils/ParquetMetadataUtils.scala

@@ -162,6 +162,48 @@ object ParquetMetadataUtils extends Logging {
     None
   }
 
+  /**
+   * Checks whether Parquet files contain variant logical type annotations that require fallback to
+   * vanilla Spark. This check is always performed (not gated by parquetMetadataValidationEnabled)
+   * because it is a correctness issue: Velox native reader does not check variant annotations.
+   */
+  def validateVariantAnnotation(
+      rootPaths: Seq[String],
+      hadoopConf: Configuration,
+      fileLimit: Int
+  ): Option[String] = {
+    rootPaths.foreach {
+      rootPath =>
+        val fs = new Path(rootPath).getFileSystem(hadoopConf)
+        try {
+          val filesIterator = fs.listFiles(new Path(rootPath), true)
+          var checkedFileCount = 0
+          while (filesIterator.hasNext && checkedFileCount < fileLimit) {
+            val fileStatus = filesIterator.next()
+            checkedFileCount += 1
+            try {
+              val footer = ParquetFooterReaderShim.readFooter(
+                hadoopConf,
+                fileStatus,
+                ParquetMetadataConverter.NO_FILTER)
+              if (
+                SparkShimLoader.getSparkShims
+                  .shouldFallbackForParquetVariantAnnotation(footer)
+              ) {
+                return Some("Variant annotation detected in Parquet file.")
+              }
+            } catch {
+              case _: Exception => // ignore
+            }
+          }
+        } catch {
+          case e: Exception =>
+            logWarning("Catch exception when checking variant annotation", e)
+        }
+    }
+    None
+  }
+
   private def isTimezoneFoundInMetadata(
       footer: ParquetMetadata,
       parquetOptions: ParquetOptions): Option[String] = {

gluten-ut/spark40/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -853,8 +853,8 @@ class VeloxTestSettings extends BackendTestSettings {
   enableSuite[GlutenUDTRegistrationSuite]
   enableSuite[GlutenUnsafeRowSuite]
   enableSuite[GlutenUserDefinedTypeSuite]
-  // TODO: 4.x enableSuite[GlutenVariantEndToEndSuite]  // 3 failures
-  // TODO: 4.x enableSuite[GlutenVariantShreddingSuite]  // 8 failures
+  enableSuite[GlutenVariantEndToEndSuite]
+  enableSuite[GlutenVariantShreddingSuite]
   enableSuite[GlutenVariantSuite]
   enableSuite[GlutenVariantWriteShreddingSuite]
   // TODO: 4.x enableSuite[GlutenXmlFunctionsSuite]  // 10 failures

gluten-ut/spark41/src/test/scala/org/apache/gluten/utils/velox/VeloxTestSettings.scala

@@ -397,7 +397,7 @@ class VeloxTestSettings extends BackendTestSettings {
     .exclude("parquet widening conversion ShortType -> DecimalType(20,0)")
     .exclude("parquet widening conversion ShortType -> DecimalType(38,0)")
     .exclude("parquet widening conversion ShortType -> DoubleType")
-  // TODO: 4.x enableSuite[GlutenParquetVariantShreddingSuite]  // 1 failure
+  enableSuite[GlutenParquetVariantShreddingSuite]
   // Generated suites for org.apache.spark.sql.execution.datasources.text
   // TODO: 4.x enableSuite[GlutenWholeTextFileV1Suite]  // 1 failure
   // TODO: 4.x enableSuite[GlutenWholeTextFileV2Suite]  // 1 failure
@@ -818,8 +818,8 @@ class VeloxTestSettings extends BackendTestSettings {
   enableSuite[GlutenUDTRegistrationSuite]
   enableSuite[GlutenUnsafeRowSuite]
   enableSuite[GlutenUserDefinedTypeSuite]
-  // TODO: 4.x enableSuite[GlutenVariantEndToEndSuite]  // 3 failures
-  // TODO: 4.x enableSuite[GlutenVariantShreddingSuite]  // 8 failures
+  enableSuite[GlutenVariantEndToEndSuite]
+  enableSuite[GlutenVariantShreddingSuite]
   enableSuite[GlutenVariantSuite]
   enableSuite[GlutenVariantWriteShreddingSuite]
   // TODO: 4.x enableSuite[GlutenXmlFunctionsSuite]  // 10 failures

pom.xml

@@ -167,6 +167,7 @@
       --add-opens=java.base/sun.util.calendar=ALL-UNNAMED
       -Djdk.reflect.useDirectMethodHandle=false
       -Dio.netty.tryReflectionSetAccessible=true
+      -Dfile.encoding=UTF-8
     </extraJavaTestArgs>
     <log4j.conf>file:src/test/resources/log4j2.properties</log4j.conf>
   </properties>

shims/common/src/main/scala/org/apache/gluten/sql/shims/SparkShims.scala

@@ -237,6 +237,8 @@ trait SparkShims {
 
   def isParquetFileEncrypted(footer: ParquetMetadata): Boolean
 
+  def shouldFallbackForParquetVariantAnnotation(footer: ParquetMetadata): Boolean = false
+
   def getOtherConstantMetadataColumnValues(file: PartitionedFile): JMap[String, Object] =
     Map.empty[String, Any].asJava.asInstanceOf[JMap[String, Object]]
 

shims/spark41/src/main/scala/org/apache/gluten/sql/shims/spark41/Spark41Shims.scala

@@ -53,7 +53,7 @@ import org.apache.spark.sql.types._
 import org.apache.hadoop.fs.{FileStatus, Path}
 import org.apache.parquet.hadoop.metadata.{CompressionCodecName, ParquetMetadata}
 import org.apache.parquet.hadoop.metadata.FileMetaData.EncryptionType
-import org.apache.parquet.schema.MessageType
+import org.apache.parquet.schema.{GroupType, LogicalTypeAnnotation, MessageType}
 
 import java.time.ZoneOffset
 import java.util.{Map => JMap}
@@ -571,6 +571,22 @@ class Spark41Shims extends SparkShims {
     }
   }
 
+  override def shouldFallbackForParquetVariantAnnotation(footer: ParquetMetadata): Boolean = {
+    if (SQLConf.get.getConf(SQLConf.PARQUET_IGNORE_VARIANT_ANNOTATION)) {
+      return false
+    }
+    containsVariantAnnotation(footer.getFileMetaData.getSchema)
+  }
+
+  private def containsVariantAnnotation(groupType: GroupType): Boolean = {
+    groupType.getFields.asScala.exists {
+      field =>
+        Option(field.getLogicalTypeAnnotation)
+          .exists(_.isInstanceOf[LogicalTypeAnnotation.VariantLogicalTypeAnnotation]) ||
+        (!field.isPrimitive && containsVariantAnnotation(field.asGroupType()))
+    }
+  }
+
   override def getOtherConstantMetadataColumnValues(file: PartitionedFile): JMap[String, Object] =
     file.otherConstantMetadataColumnValues.asJava.asInstanceOf[JMap[String, Object]]