[SPARK-58917][SQL] Respect inferSchema for variant ingestion in CSV parser

docs/sql-data-sources-csv.md

@@ -228,6 +228,12 @@ Data source options of CSV can be set via:
     <td>If specified, the entire CSV record is parsed and stored as a single column of <code>VariantType</code> with the given column name, instead of being split into individual fields.</td>
     <td>read</td>
   </tr>
+  <tr>
+    <td><code>variantRespectInferSchema</code></td>
+    <td>false</td>
+    <td>When true and <code>inferSchema</code> is false, the CSV to Variant parser preserves scalar CSV values as strings inside the Variant instead of inferring their types. This only affects variant ingestion (<code>singleVariantColumn</code> mode or explicit <code>VariantType</code> columns). When false (the default), scalar types (long, decimal, date, timestamp, boolean) are always inferred regardless of the <code>inferSchema</code> setting, preserving existing behavior.</td>
+    <td>read</td>
+  </tr>
   <tr>
     <td><code>multiLine</code></td>
     <td>false</td>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala

@@ -340,6 +340,15 @@ class CSVOptions(
   def needHeaderForSingleVariantColumn: Boolean =
     singleVariantColumn.isDefined && headerFlag
 
+  /**
+   * When true and inferSchema is false, the CSV to Variant parser preserves scalar CSV values
+   * as strings inside the Variant instead of inferring their types. This only affects variant
+   * ingestion (singleVariantColumn mode or explicit VariantType columns).
+   *
+   * Defaults to false to preserve existing behavior where scalar types are always inferred.
+   */
+  val variantRespectInferSchema: Boolean = getBool(VARIANT_RESPECT_INFER_SCHEMA, default = false)
+
   def asWriterSettings: CsvWriterSettings = {
     val writerSettings = new CsvWriterSettings()
     val format = writerSettings.getFormat
@@ -443,6 +452,7 @@ object CSVOptions extends DataSourceOptions {
   newOption(SEP, DELIMITER)
   val COLUMN_PRUNING = newOption("columnPruning")
   val SINGLE_VARIANT_COLUMN = newOption(DataSourceOptions.SINGLE_VARIANT_COLUMN)
+  val VARIANT_RESPECT_INFER_SCHEMA = newOption("variantRespectInferSchema")
 
   // Max error content length in CSV parser/writer exception messages, and the bound on the bad
   // record embedded in MALFORMED_CSV_RECORD errors.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala

@@ -460,13 +460,18 @@ class UnivocityParser(
    * input as a specific scalar type, or fails after trying all the types and defaults to the string
    * type. The state is reset for every input file.
    *
+   * When variantRespectInferSchema is true and inferSchema is false, scalar type inference is
+   * skipped and all non-null values are stored as strings.
+   *
    * Floating point types (double, float) are not considered to avoid precision loss.
    */
   private final class VariantValueConverter extends ValueConverter {
     private var currentType: DataType = LongType
     // Keep consistent with `CSVInferSchema`: only produce TimestampNTZ when the default timestamp
     // type is TimestampNTZ.
     private val isDefaultNTZ = SQLConf.get.timestampType == TimestampNTZType
+    // When variantRespectInferSchema is true and inferSchema is false, skip type inference
+    private val shouldInferTypes = !options.variantRespectInferSchema || options.inferSchemaFlag
 
     override def apply(s: String): Any = {
       val builder = new VariantBuilder(false)
@@ -481,6 +486,12 @@ class UnivocityParser(
         return
       }
 
+      // If variantRespectInferSchema is true and inferSchema is false, store as string
+      if (!shouldInferTypes) {
+        builder.appendString(s)
+        return
+      }
+
       def parseLong(): DataType = {
         try {
           builder.appendLong(s.toLong)

sql/core/src/test/scala/org/apache/spark/sql/CsvFunctionsSuite.scala

@@ -869,6 +869,59 @@ class CsvFunctionsSuite extends SharedSparkSession {
       Seq(Row(s"""{null, $largeInput}""")))
   }
 
+  test("from_csv with variant: variantRespectInferSchema option") {
+    val df = Seq("0001,100,1.1,true", "0002,200,2.2,false").toDF("value")
+
+    // Default behavior: types are inferred (0001 becomes 1, 0002 becomes 2)
+    checkAnswer(
+      df.select(
+        from_csv(
+          $"value",
+          StructType.fromDDL("a variant, b variant, c variant, d variant"),
+          Map.empty[String, String]
+        ).cast("string")),
+      Seq(
+        Row("""{1, 100, 1.1, true}"""),
+        Row("""{2, 200, 2.2, false}""")))
+
+    // With variantRespectInferSchema=true and inferSchema=false: preserve as strings
+    checkAnswer(
+      df.select(
+        from_csv(
+          $"value",
+          StructType.fromDDL("a variant, b variant, c variant, d variant"),
+          Map("variantRespectInferSchema" -> "true", "inferSchema" -> "false")
+        ).cast("string")),
+      Seq(
+        Row("""{"0001", "100", "1.1", "true"}"""),
+        Row("""{"0002", "200", "2.2", "false"}""")))
+
+    // With variantRespectInferSchema=true and inferSchema=true: still infer types
+    checkAnswer(
+      df.select(
+        from_csv(
+          $"value",
+          StructType.fromDDL("a variant, b variant, c variant, d variant"),
+          Map("variantRespectInferSchema" -> "true", "inferSchema" -> "true")
+        ).cast("string")),
+      Seq(
+        Row("""{1, 100, 1.1, true}"""),
+        Row("""{2, 200, 2.2, false}""")))
+
+    // Test with singleVariantColumn mode
+    checkAnswer(
+      df.select(
+        from_csv(
+          $"value",
+          StructType.fromDDL("v variant"),
+          Map("singleVariantColumn" -> "v", "variantRespectInferSchema" -> "true", 
+              "inferSchema" -> "false")
+        ).cast("string")),
+      Seq(
+        Row("""{{"_c0":"0001","_c1":"100","_c2":"1.1","_c3":"true"}}"""),
+        Row("""{{"_c0":"0002","_c1":"200","_c2":"2.2","_c3":"false"}}""")))
+  }
+
   test("from_csv with variant: extreme negative scale decimal does not hang") {
     // A value like "1E99999" parses to a BigDecimal with scale=-99999.
     // Calling setScale(0) on it would hang, so it should fall through to string.

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala

@@ -3342,10 +3342,11 @@ abstract class CSVSuite
   }
 
   test("validate CSV Options") {
-    assert(CSVOptions.getAllOptions.size == 42)
+    assert(CSVOptions.getAllOptions.size == 43)
     // Please add validation on any new CSV options here
     assert(CSVOptions.isValidOption("header"))
     assert(CSVOptions.isValidOption("inferSchema"))
+    assert(CSVOptions.isValidOption("variantRespectInferSchema"))
     assert(CSVOptions.isValidOption("ignoreLeadingWhiteSpace"))
     assert(CSVOptions.isValidOption("ignoreTrailingWhiteSpace"))
     assert(CSVOptions.isValidOption("preferDate"))
@@ -3917,6 +3918,92 @@ abstract class CSVSuite
     )
   }
 
+  test("variantRespectInferSchema option") {
+    withTempPath { path =>
+      val data =
+        """0001,100,1.1,true
+          |0002,2000-01-01,2000-01-01 01:02:03,false
+          |0003,1e9,hello,extra
+          |""".stripMargin
+      Files.write(path.toPath, data.getBytes(StandardCharsets.UTF_8))
+
+      // Default behavior: inferSchema is always applied for variant ingestion
+      checkAnswer(
+        spark.read.option("singleVariantColumn", "v")
+          .csv(path.getCanonicalPath).selectExpr("cast(v as string)"),
+        Seq(
+          Row("""{"_c0":1,"_c1":100,"_c2":1.1,"_c3":true}"""),
+          Row("""{"_c0":2,"_c1":"2000-01-01","_c2":"2000-01-01 01:02:03-08:00","_c3":false}"""),
+          Row("""{"_c0":3,"_c1":"1e9","_c2":"hello","_c3":"extra"}""")
+        )
+      )
+
+      // With variantRespectInferSchema=true and inferSchema=false: preserve as strings
+      checkAnswer(
+        spark.read
+          .option("singleVariantColumn", "v")
+          .option("variantRespectInferSchema", "true")
+          .option("inferSchema", "false")
+          .csv(path.getCanonicalPath).selectExpr("cast(v as string)"),
+        Seq(
+          Row("""{"_c0":"0001","_c1":"100","_c2":"1.1","_c3":"true"}"""),
+          Row("""{"_c0":"0002","_c1":"2000-01-01","_c2":"2000-01-01 01:02:03","_c3":"false"}"""),
+          Row("""{"_c0":"0003","_c1":"1e9","_c2":"hello","_c3":"extra"}""")
+        )
+      )
+
+      // With variantRespectInferSchema=true and inferSchema=true: still infer types
+      checkAnswer(
+        spark.read
+          .option("singleVariantColumn", "v")
+          .option("variantRespectInferSchema", "true")
+          .option("inferSchema", "true")
+          .csv(path.getCanonicalPath).selectExpr("cast(v as string)"),
+        Seq(
+          Row("""{"_c0":1,"_c1":100,"_c2":1.1,"_c3":true}"""),
+          Row("""{"_c0":2,"_c1":"2000-01-01","_c2":"2000-01-01 01:02:03-08:00","_c3":false}"""),
+          Row("""{"_c0":3,"_c1":"1e9","_c2":"hello","_c3":"extra"}""")
+        )
+      )
+
+      // Test with explicit variant schema columns
+      checkAnswer(
+        spark.read
+          .option("variantRespectInferSchema", "true")
+          .option("inferSchema", "false")
+          .schema("c0 variant, c1 variant, c2 variant, c3 variant")
+          .csv(path.getCanonicalPath)
+          .selectExpr("cast(c0 as string)", "cast(c1 as string)", "cast(c2 as string)", "cast(c3 as string)"),
+        Seq(
+          Row("0001", "100", "1.1", "true"),
+          Row("0002", "2000-01-01", "2000-01-01 01:02:03", "false"),
+          Row("0003", "1e9", "hello", "extra")
+        )
+      )
+
+      // Test with header
+      val dataWithHeader =
+        """id,num,dec,bool
+          |0001,100,1.1,true
+          |0002,200,2.2,false
+          |""".stripMargin
+      Files.write(path.toPath, dataWithHeader.getBytes(StandardCharsets.UTF_8))
+
+      checkAnswer(
+        spark.read
+          .option("singleVariantColumn", "v")
+          .option("variantRespectInferSchema", "true")
+          .option("inferSchema", "false")
+          .option("header", "true")
+          .csv(path.getCanonicalPath).selectExpr("cast(v as string)"),
+        Seq(
+          Row("""{"bool":"true","dec":"1.1","id":"0001","num":"100"}"""),
+          Row("""{"bool":"false","dec":"2.2","id":"0002","num":"200"}""")
+        )
+      )
+    }
+  }
+
   private def createTestFiles(dir: File, fileFormatWriter: Boolean,
                               header: Boolean): Seq[Row] = {
     val numRecord = 100