upmerge

integration_tests/src/main/python/csv_test.py

@@ -324,6 +324,7 @@ def test_csv_fallback(spark_tmp_path, read_func, disable_conf, spark_tmp_table_f
             cpu_fallback_class_name=get_non_gpu_allowed()[0],
             conf=updated_conf)
 
+# todo add '' to represent no date format specified
 csv_supported_date_formats = ['yyyy-MM-dd', 'yyyy/MM/dd', 'yyyy-MM', 'yyyy/MM',
         'MM-yyyy', 'MM/yyyy', 'MM-dd-yyyy', 'MM/dd/yyyy', 'dd-MM-yyyy', 'dd/MM/yyyy']
 @pytest.mark.parametrize('date_format', csv_supported_date_formats, ids=idfn)

integration_tests/src/test/resources/dates.json

@@ -1,5 +1,6 @@
 { "number": "2020-09-16" }
-{ "number": " 2020-09-16" }
-{ "number": "2020-09-16 " }
+{ "number": " 2020-09-17" }
+{ "number": "2020-09-18 " }
+{ "number": " 2020-09-19 " }
 { "number": "1581-01-01" }
 { "number": "1583-01-01" }

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuCSVScan.scala

@@ -359,7 +359,7 @@ abstract class CSVPartitionReaderBase[BUFF <: LineBufferer, FACT <: LineBufferer
     }
   }
 
-  override def dateFormat: String = GpuCsvUtils.dateFormatInRead(parsedOptions)
+  override def dateFormat: Option[String] = Some(GpuCsvUtils.dateFormatInRead(parsedOptions))
   override def timestampFormat: String = GpuCsvUtils.timestampFormatInRead(parsedOptions)
 }
 

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuCast.scala

@@ -22,7 +22,7 @@ import java.util.Optional
 
 import scala.collection.mutable.ArrayBuffer
 
-import ai.rapids.cudf.{BinaryOp, CaptureGroups, ColumnVector, ColumnView, DecimalUtils, DType, RegexProgram, Scalar}
+import ai.rapids.cudf.{BinaryOp, CaptureGroups, ColumnVector, ColumnView, DType, DecimalUtils, RegexProgram, Scalar}
 import ai.rapids.cudf
 import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource}
 import com.nvidia.spark.rapids.RapidsPluginImplicits._
@@ -1376,7 +1376,7 @@ object GpuCast {
     }
   }
 
-  private def castStringToDateAnsi(input: ColumnVector, ansiMode: Boolean): ColumnVector = {
+  def castStringToDateAnsi(input: ColumnVector, ansiMode: Boolean): ColumnVector = {
     val result = castStringToDate(input)
     if (ansiMode) {
       // When ANSI mode is enabled, we need to throw an exception if any values could not be

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuTextBasedPartitionReader.scala

@@ -16,7 +16,6 @@
 
 package com.nvidia.spark.rapids
 
-import java.time.DateTimeException
 import java.util.Optional
 
 import scala.collection.mutable.ListBuffer
@@ -36,7 +35,7 @@ import org.apache.spark.sql.connector.read.PartitionReader
 import org.apache.spark.sql.execution.QueryExecutionException
 import org.apache.spark.sql.execution.datasources.{HadoopFileLinesReader, PartitionedFile}
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.rapids.{ExceptionTimeParserPolicy, GpuToTimestamp, LegacyTimeParserPolicy}
+import org.apache.spark.sql.rapids.{GpuToTimestamp, LegacyTimeParserPolicy}
 import org.apache.spark.sql.types.{DataTypes, DecimalType, StructField, StructType}
 import org.apache.spark.sql.vectorized.ColumnarBatch
 
@@ -372,30 +371,40 @@ abstract class GpuTextBasedPartitionReader[BUFF <: LineBufferer, FACT <: LineBuf
     }
   }
 
-  def dateFormat: String
+  def dateFormat: Option[String]
   def timestampFormat: String
 
   def castStringToDate(input: ColumnVector, dt: DType): ColumnVector = {
     castStringToDate(input, dt, failOnInvalid = true)
   }
 
   def castStringToDate(input: ColumnVector, dt: DType, failOnInvalid: Boolean): ColumnVector = {
-    val cudfFormat = DateUtils.toStrf(dateFormat, parseString = true)
-    withResource(input.strip()) { stripped =>
-      withResource(stripped.isTimestamp(cudfFormat)) { isDate =>
-        if (failOnInvalid && GpuOverrides.getTimeParserPolicy == ExceptionTimeParserPolicy) {
-          withResource(isDate.all()) { all =>
-            if (all.isValid && !all.getBoolean) {
-              throw new DateTimeException("One or more values is not a valid date")
-            }
-          }
-        }
-        withResource(stripped.asTimestamp(dt, cudfFormat)) { asDate =>
-          withResource(Scalar.fromNull(dt)) { nullScalar =>
-            isDate.ifElse(asDate, nullScalar)
+
+    // TODO make these same changes for timestamps and add tests
+
+    val dateFormatPattern = dateFormat.getOrElse("yyyy-MM-dd")
+
+    val cudfFormat = DateUtils.toStrf(dateFormatPattern, parseString = true)
+
+    dateFormat match {
+      case Some(_) =>
+        val twoDigits = raw"\d{2}"
+        val fourDigits = raw"\d{4}"
+
+        val regexRoot = dateFormatPattern
+          .replace("yyyy", fourDigits)
+          .replace("MM", twoDigits)
+          .replace("dd", twoDigits)
+        GpuCast.convertDateOrNull(input, "^" + regexRoot + "$", cudfFormat)
+      case _ =>
+        // legacy behavior
+        // TODO this is similar to, but different from  GpuJsonToStructsShim
+//        withResource(Scalar.fromString(" ")) { space =>
+          withResource(input.strip()) { trimmed =>
+            // TODO add tests for EXCEPTION policy handling
+            GpuCast.castStringToDateAnsi(trimmed, ansiMode = false) // TODO
           }
-        }
-      }
+//        }
     }
   }
 

sql-plugin/src/main/scala/org/apache/spark/sql/catalyst/json/rapids/GpuJsonScan.scala

@@ -490,6 +490,6 @@ class JsonPartitionReader(
     }
   }
 
-  override def dateFormat: String = GpuJsonUtils.dateFormatInRead(parsedOptions)
+  override def dateFormat: Option[String] = GpuJsonUtils.optionalDateFormatInRead(parsedOptions)
   override def timestampFormat: String = GpuJsonUtils.timestampFormatInRead(parsedOptions)
 }