From c6548a3d87716f3650bea9926ce198c01d4302f2 Mon Sep 17 00:00:00 2001 From: Pratham Manja Date: Mon, 24 Aug 2026 16:23:47 +0530 Subject: [PATCH] Respect inferSchema for variant ingestion in CSV parser --- docs/sql-data-sources-csv.md | 6 ++ .../spark/sql/catalyst/csv/CSVOptions.scala | 10 +++ .../sql/catalyst/csv/UnivocityParser.scala | 11 +++ .../apache/spark/sql/CsvFunctionsSuite.scala | 53 +++++++++++ .../execution/datasources/csv/CSVSuite.scala | 89 ++++++++++++++++++- 5 files changed, 168 insertions(+), 1 deletion(-) diff --git a/docs/sql-data-sources-csv.md b/docs/sql-data-sources-csv.md index bf63fe2be5d3e..9b42a7654088b 100644 --- a/docs/sql-data-sources-csv.md +++ b/docs/sql-data-sources-csv.md @@ -228,6 +228,12 @@ Data source options of CSV can be set via: If specified, the entire CSV record is parsed and stored as a single column of VariantType with the given column name, instead of being split into individual fields. read + + variantRespectInferSchema + false + 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). When false (the default), scalar types (long, decimal, date, timestamp, boolean) are always inferred regardless of the inferSchema setting, preserving existing behavior. + read + multiLine false diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala index 7db03a8a23231..0b563a7c04d3b 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/CSVOptions.scala +++ b/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. diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala index fa6bd19064f0c..6dbf3b5ee57a0 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala @@ -460,6 +460,9 @@ 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 { @@ -467,6 +470,8 @@ class UnivocityParser( // 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) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CsvFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CsvFunctionsSuite.scala index 455434b4de011..9e44e432bf712 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/CsvFunctionsSuite.scala +++ b/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. diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala index 3aefe25f6a08c..3533d44d72d25 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/csv/CSVSuite.scala +++ b/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