allow legacy parser with inferDate

Author: Jonathancui123

core/src/main/resources/error/error-classes.json

@@ -23,12 +23,6 @@
     ],
     "sqlState" : "22005"
   },
-  "CANNOT_INFER_DATE" : {
-    "message" : [
-      "Cannot infer date in schema inference when LegacyTimeParserPolicy is \"LEGACY\". Legacy Date formatter does not support strict date format matching which is required to avoid inferring timestamps and other non-date entries to date."
-    ],
-    "sqlState" : "22007"
-  },
   "CANNOT_PARSE_DECIMAL" : {
     "message" : [
       "Cannot parse decimal"

docs/sql-data-sources-csv.md

@@ -111,7 +111,7 @@ Data source options of CSV can be set via:
   <tr>
     <td><code>inferDate</code></td> 
     <td>false</td>
-    <td>Whether or not to infer columns that satisfy the <code>dateFormat</code> option as <code>Date</code>. Requires <code>inferSchema</code> to be <code>true</code>. When <code>false</code>, columns with dates will be inferred as <code>String</code> (or as <code>Timestamp</code> if it fits the <code>timestampFormat</code>).</td>
+    <td>Whether or not to infer columns that satisfy the <code>dateFormat</code> option as <code>Date</code>. Requires <code>inferSchema</code> to be <code>true</code> and cannot infer from legacy formats. When <code>false</code>, columns with dates will be inferred as <code>String</code> (or as <code>Timestamp</code> if it fits the <code>timestampFormat</code>).</td>
     <td>read</td>
   </tr>
   <tr>

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

@@ -50,7 +50,8 @@ class CSVInferSchema(val options: CSVOptions) extends Serializable {
     options.dateFormatInRead,
     options.locale,
     legacyFormat = FAST_DATE_FORMAT,
-    isParsing = true)
+    isParsing = true,
+    isInferring = true)
 
   private val decimalParser = if (options.locale == Locale.US) {
     // Special handling the default locale for backward compatibility

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

@@ -150,17 +150,11 @@ class CSVOptions(
 
   /**
    * Infer columns with all valid date entries as date type (otherwise inferred as timestamp type).
-   * Disabled by default for backwards compatibility and performance. When enabled, date entries in
-   * timestamp columns will be cast to timestamp upon parsing. Not compatible with
-   * legacyTimeParserPolicy == LEGACY since legacy date parser will accept extra trailing characters
+   * Disabled by default for performance. When enabled, date entries in timestamp columns
+   * will be cast to timestamp upon parsing. Cannot infer legacy date formats since the legacy date
+   * parser will accept extra trailing characters
    */
-  val inferDate = {
-    val inferDateFlag = getBool("inferDate")
-    if (SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY && inferDateFlag) {
-      throw QueryExecutionErrors.inferDateWithLegacyTimeParserError()
-    }
-    inferDateFlag
-  }
+  val inferDate = getBool("inferDate")
 
   // Provide a default value for dateFormatInRead when inferDate. This ensures that the
   // Iso8601DateFormatter (with strict date parsing) is used for date inference

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

@@ -112,7 +112,8 @@ class UnivocityParser(
     options.dateFormatInRead,
     options.locale,
     legacyFormat = FAST_DATE_FORMAT,
-    isParsing = true)
+    isParsing = true,
+    isInferring = false)
 
   private val csvFilters = if (SQLConf.get.csvFilterPushDown) {
     new OrderedFilters(filters, requiredSchema)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala

@@ -76,7 +76,8 @@ class JacksonParser(
     options.dateFormatInRead,
     options.locale,
     legacyFormat = FAST_DATE_FORMAT,
-    isParsing = true)
+    isParsing = true,
+    isInferring = false)
 
   /**
    * Create a converter which converts the JSON documents held by the `JsonParser`

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateFormatter.scala

@@ -175,8 +175,9 @@ object DateFormatter {
       format: Option[String],
       locale: Locale = defaultLocale,
       legacyFormat: LegacyDateFormat = LENIENT_SIMPLE_DATE_FORMAT,
-      isParsing: Boolean): DateFormatter = {
-    if (SQLConf.get.legacyTimeParserPolicy == LEGACY) {
+      isParsing: Boolean,
+      isInference: Boolean = false): DateFormatter = {
+    if (SQLConf.get.legacyTimeParserPolicy == LEGACY && !isInference) {
       getLegacyFormatter(format.getOrElse(defaultPattern), locale, legacyFormat)
     } else {
       val df = format
@@ -203,8 +204,9 @@ object DateFormatter {
       format: Option[String],
       locale: Locale,
       legacyFormat: LegacyDateFormat,
-      isParsing: Boolean): DateFormatter = {
-    getFormatter(format, locale, legacyFormat, isParsing)
+      isParsing: Boolean,
+      isInferring: Boolean): DateFormatter = {
+    getFormatter(format, locale, legacyFormat, isParsing, isInferring)
   }
 
   def apply(

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala

@@ -34,7 +34,7 @@ import org.apache.hadoop.fs.permission.FsPermission
 import org.codehaus.commons.compiler.CompileException
 import org.codehaus.janino.InternalCompilerException
 
-import org.apache.spark.{Partition, SparkArithmeticException, SparkArrayIndexOutOfBoundsException, SparkClassNotFoundException, SparkConcurrentModificationException, SparkDateTimeException, SparkException, SparkFileAlreadyExistsException, SparkFileNotFoundException, SparkIllegalArgumentException, SparkIndexOutOfBoundsException, SparkNoSuchElementException, SparkNoSuchMethodException, SparkNumberFormatException, SparkRuntimeException, SparkSecurityException, SparkSQLException, SparkSQLFeatureNotSupportedException, SparkThrowable, SparkUnsupportedOperationException, SparkUpgradeException}
+import org.apache.spark.{Partition, SparkArithmeticException, SparkArrayIndexOutOfBoundsException, SparkClassNotFoundException, SparkConcurrentModificationException, SparkDateTimeException, SparkException, SparkFileAlreadyExistsException, SparkFileNotFoundException, SparkIllegalArgumentException, SparkIndexOutOfBoundsException, SparkNoSuchElementException, SparkNoSuchMethodException, SparkNumberFormatException, SparkRuntimeException, SparkSecurityException, SparkSQLException, SparkSQLFeatureNotSupportedException, SparkUnsupportedOperationException, SparkUpgradeException}
 import org.apache.spark.executor.CommitDeniedException
 import org.apache.spark.launcher.SparkLauncher
 import org.apache.spark.memory.SparkOutOfMemoryError
@@ -529,12 +529,6 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase {
        """.stripMargin)
   }
 
-  def inferDateWithLegacyTimeParserError(): Throwable with SparkThrowable = {
-    new SparkIllegalArgumentException(errorClass = "CANNOT_INFER_DATE",
-      messageParameters = Array()
-    )
-  }
-
   def streamedOperatorUnsupportedByDataSourceError(
       className: String, operator: String): Throwable = {
     new UnsupportedOperationException(

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

@@ -2803,41 +2803,35 @@ abstract class CSVSuite
       "inferSchema" -> "true",
       "inferDate" -> "true")
 
-    // Error should be thrown when attempting to inferDate with Legacy parser
+    // We should still be able to parse legacy formats when the schema is provided
     if (SQLConf.get.legacyTimeParserPolicy == LegacyBehaviorPolicy.LEGACY) {
-      val msg = intercept[IllegalArgumentException] {
-        spark.read
-          .format("csv")
-          .options(options1)
-          .load(testFile(dateInferSchemaFile))
-      }.getMessage
-      assert(msg.contains("CANNOT_INFER_DATE"))
-    } else {
-      // 1. Specify date format and timestamp format
-      // 2. Date inference should work with default date format when dateFormat is not provided
-      Seq(options1, options2).foreach {options =>
-        val results = spark.read
-          .format("csv")
-          .options(options)
-          .load(testFile(dateInferSchemaFile))
-
-        val expectedSchema = StructType(List(StructField("date", DateType),
-          StructField("timestamp-date", TimestampType),
-          StructField("date-timestamp", TimestampType)))
-        assert(results.schema == expectedSchema)
+      val ds = Seq("1500-02-29").toDS()
+      val csv = spark.read.option("header", false).schema("d date").csv(ds)
+      checkAnswer(csv, Row(Date.valueOf("1500-03-01")))
+    }
+    // 1. Specify date format and timestamp format
+    // 2. Date inference should work with default date format when dateFormat is not provided
+    Seq(options1, options2).foreach {options =>
+      val results = spark.read
+        .format("csv")
+        .options(options)
+        .load(testFile(dateInferSchemaFile))
 
-        val expected =
-          Seq(
-            Seq(Date.valueOf("2001-9-8"), Timestamp.valueOf("2014-10-27 18:30:0.0"),
-              Timestamp.valueOf("1765-03-28 00:00:0.0")),
-            Seq(Date.valueOf("1941-1-2"), Timestamp.valueOf("2000-09-14 01:01:0.0"),
-              Timestamp.valueOf("1423-11-12 23:41:0.0")),
-            Seq(Date.valueOf("0293-11-7"), Timestamp.valueOf("1995-06-25 00:00:00.0"),
-              Timestamp.valueOf("2016-01-28 20:00:00.0"))
-          )
-        assert(results.collect().toSeq.map(_.toSeq) == expected)
-      }
+      val expectedSchema = StructType(List(StructField("date", DateType),
+        StructField("timestamp-date", TimestampType),
+        StructField("date-timestamp", TimestampType)))
+      assert(results.schema == expectedSchema)
 
+      val expected =
+        Seq(
+          Seq(Date.valueOf("2001-9-8"), Timestamp.valueOf("2014-10-27 18:30:0.0"),
+            Timestamp.valueOf("1765-03-28 00:00:0.0")),
+          Seq(Date.valueOf("1941-1-2"), Timestamp.valueOf("2000-09-14 01:01:0.0"),
+            Timestamp.valueOf("1423-11-12 23:41:0.0")),
+          Seq(Date.valueOf("0293-11-7"), Timestamp.valueOf("1995-06-25 00:00:00.0"),
+            Timestamp.valueOf("2016-01-28 20:00:00.0"))
+        )
+      assert(results.collect().toSeq.map(_.toSeq) == expected)
     }
   }
 }