minor updates, add test

Author: sadikovi

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

@@ -40,7 +40,7 @@ class CSVInferSchema(val options: CSVOptions) extends Serializable {
     isParsing = true)
 
   private val timestampNTZFormatter = TimestampFormatter(
-    options.timestampNTZFormat,
+    options.timestampNTZFormatInRead,
     options.zoneId,
     legacyFormat = FAST_DATE_FORMAT,
     isParsing = true,

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

@@ -164,7 +164,9 @@ class CSVOptions(
       s"${DateFormatter.defaultPattern}'T'HH:mm:ss[.SSS][XXX]"
     })
 
-  val timestampNTZFormat: Option[String] = parameters.get("timestampNTZFormat")
+  val timestampNTZFormatInRead: Option[String] = parameters.get("timestampNTZFormat")
+  val timestampNTZFormatInWrite: String = parameters.getOrElse("timestampNTZFormat",
+    s"${DateFormatter.defaultPattern}'T'HH:mm:ss[.SSS]")
 
   val multiLine = parameters.get("multiLine").map(_.toBoolean).getOrElse(false)
 

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

@@ -49,7 +49,7 @@ class UnivocityGenerator(
     legacyFormat = FAST_DATE_FORMAT,
     isParsing = false)
   private val timestampNTZFormatter = TimestampFormatter(
-    options.timestampNTZFormat,
+    options.timestampNTZFormatInWrite,
     options.zoneId,
     legacyFormat = FAST_DATE_FORMAT,
     isParsing = false,

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

@@ -94,7 +94,7 @@ class UnivocityParser(
     legacyFormat = FAST_DATE_FORMAT,
     isParsing = true)
   private lazy val timestampNTZFormatter = TimestampFormatter(
-    options.timestampNTZFormat,
+    options.timestampNTZFormatInRead,
     options.zoneId,
     legacyFormat = FAST_DATE_FORMAT,
     isParsing = true,

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

@@ -368,4 +368,15 @@ class CsvFunctionsSuite extends QueryTest with SharedSparkSession {
       .selectExpr("value.a")
     checkAnswer(fromCsvDF, Row(localDT))
   }
+
+  test("SPARK-36490: Handle incorrectly formatted timestamp_ntz values in from_csv") {
+    val fromCsvDF = Seq("2021-08-12T15:16:23.000+11:00").toDF("csv")
+      .select(
+        from_csv(
+          $"csv",
+          StructType(StructField("a", TimestampNTZType) :: Nil),
+          Map.empty[String, String]) as "value")
+      .selectExpr("value.a")
+    checkAnswer(fromCsvDF, Row(null))
+  }
 }

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

@@ -343,11 +343,11 @@ object CSVBenchmark extends SqlBasedBenchmark {
   override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
     runBenchmark("Benchmark to measure CSV read/write performance") {
       val numIters = 3
-      quotedValuesBenchmark(rowsNum = 50 * 1000, numIters)
-      multiColumnsBenchmark(rowsNum = 1000 * 1000, numIters)
-      countBenchmark(rowsNum = 10 * 1000 * 1000, numIters)
+      // quotedValuesBenchmark(rowsNum = 50 * 1000, numIters)
+      // multiColumnsBenchmark(rowsNum = 1000 * 1000, numIters)
+      // countBenchmark(rowsNum = 10 * 1000 * 1000, numIters)
       datetimeBenchmark(rowsNum = 10 * 1000 * 1000, numIters)
-      filtersPushdownBenchmark(rowsNum = 100 * 1000, numIters)
+      // filtersPushdownBenchmark(rowsNum = 100 * 1000, numIters)
     }
   }
 }

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

@@ -1130,19 +1130,21 @@ abstract class CSVSuite
         .option("header", "true")
         .load(path)
 
-      if (spark.conf.get(SQLConf.LEGACY_TIME_PARSER_POLICY.key) == "legacy") {
-        // Timestamps without timezone are parsed as strings, so the col0 type would be
-        // StringType which is similar to reading without schema inference.
-        val exp = spark.read.format("csv").option("header", "true").load(path)
-        checkAnswer(res, exp)
-      } else {
-        val exp = spark.sql("""
-          select timestamp_ltz'2020-12-12T12:12:12.000' as col0 union all
-          select timestamp_ltz'2020-12-12T17:12:12.000Z' as col0 union all
-          select timestamp_ltz'2020-12-12T17:12:12.000+05:00' as col0 union all
-          select timestamp_ltz'2020-12-12T12:12:12.000' as col0
-          """)
-        checkAnswer(res, exp)
+      for (policy <- Seq("exception", "corrected", "legacy")) {
+        if (spark.conf.get(SQLConf.LEGACY_TIME_PARSER_POLICY.key) == "legacy") {
+          // Timestamps without timezone are parsed as strings, so the col0 type would be
+          // StringType which is similar to reading without schema inference.
+          val exp = spark.read.format("csv").option("header", "true").load(path)
+          checkAnswer(res, exp)
+        } else {
+          val exp = spark.sql("""
+            select timestamp_ltz'2020-12-12T12:12:12.000' as col0 union all
+            select timestamp_ltz'2020-12-12T17:12:12.000Z' as col0 union all
+            select timestamp_ltz'2020-12-12T17:12:12.000+05:00' as col0 union all
+            select timestamp_ltz'2020-12-12T12:12:12.000' as col0
+            """)
+          checkAnswer(res, exp)
+        }
       }
     }
   }