[SPARK-15143][SPARK-15144][SQL] Add CSV tests with HadoopFsRelationTest and support for nullValue for other types

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/csv/CSVInferSchema.scala

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.execution.datasources.csv
 
 import java.math.BigDecimal
-import java.text.{NumberFormat, SimpleDateFormat}
+import java.text.NumberFormat
 import java.util.Locale
 
 import scala.util.control.Exception._
@@ -192,59 +192,59 @@ private[csv] object CSVTypeCast {
       nullable: Boolean = true,
       options: CSVOptions = CSVOptions()): Any = {
 
-    castType match {
-      case _: ByteType => if (datum == options.nullValue && nullable) null else datum.toByte
-      case _: ShortType => if (datum == options.nullValue && nullable) null else datum.toShort
-      case _: IntegerType => if (datum == options.nullValue && nullable) null else datum.toInt
-      case _: LongType => if (datum == options.nullValue && nullable) null else datum.toLong
-      case _: FloatType =>
-        if (datum == options.nullValue && nullable) {
-          null
-        } else if (datum == options.nanValue) {
-          Float.NaN
-        } else if (datum == options.negativeInf) {
-          Float.NegativeInfinity
-        } else if (datum == options.positiveInf) {
-          Float.PositiveInfinity
-        } else {
-          Try(datum.toFloat)
-            .getOrElse(NumberFormat.getInstance(Locale.getDefault).parse(datum).floatValue())
-        }
-      case _: DoubleType =>
-        if (datum == options.nullValue && nullable) {
-          null
-        } else if (datum == options.nanValue) {
-          Double.NaN
-        } else if (datum == options.negativeInf) {
-          Double.NegativeInfinity
-        } else if (datum == options.positiveInf) {
-          Double.PositiveInfinity
-        } else {
-          Try(datum.toDouble)
-            .getOrElse(NumberFormat.getInstance(Locale.getDefault).parse(datum).doubleValue())
-        }
-      case _: BooleanType => datum.toBoolean
-      case dt: DecimalType =>
-        if (datum == options.nullValue && nullable) {
-          null
-        } else {
-          val value = new BigDecimal(datum.replaceAll(",", ""))
-          Decimal(value, dt.precision, dt.scale)
-        }
-      case _: TimestampType if options.dateFormat != null =>
-        // This one will lose microseconds parts.
-        // See https://issues.apache.org/jira/browse/SPARK-10681.
-        options.dateFormat.parse(datum).getTime * 1000L
-      case _: TimestampType =>
-        // This one will lose microseconds parts.
-        // See https://issues.apache.org/jira/browse/SPARK-10681.
-        DateTimeUtils.stringToTime(datum).getTime  * 1000L
-      case _: DateType if options.dateFormat != null =>
-        DateTimeUtils.millisToDays(options.dateFormat.parse(datum).getTime)
-      case _: DateType =>
-        DateTimeUtils.millisToDays(DateTimeUtils.stringToTime(datum).getTime)
-      case _: StringType => UTF8String.fromString(datum)
-      case _ => throw new RuntimeException(s"Unsupported type: ${castType.typeName}")
+    if (datum == null || (datum == options.nullValue && nullable)) {
+      null
+    } else {
+      castType match {
+        case _: ByteType => datum.toByte
+        case _: ShortType => datum.toShort
+        case _: IntegerType => datum.toInt
+        case _: LongType => datum.toLong
+        case _: FloatType =>
+          if (datum == options.nanValue) {
+            Float.NaN
+          } else if (datum == options.negativeInf) {
+            Float.NegativeInfinity
+          } else if (datum == options.positiveInf) {
+            Float.PositiveInfinity
+          } else {
+            Try(datum.toFloat)
+              .getOrElse(NumberFormat.getInstance(Locale.getDefault).parse(datum).floatValue())
+          }
+        case _: DoubleType =>
+          if (datum == options.nanValue) {
+            Double.NaN
+          } else if (datum == options.negativeInf) {
+            Double.NegativeInfinity
+          } else if (datum == options.positiveInf) {
+            Double.PositiveInfinity
+          } else {
+            Try(datum.toDouble)
+              .getOrElse(NumberFormat.getInstance(Locale.getDefault).parse(datum).doubleValue())
+          }
+        case _: BooleanType => datum.toBoolean
+        case dt: DecimalType =>
+          if (datum == options.nullValue && nullable) {
+            null
+          } else {
+            val value = new BigDecimal(datum.replaceAll(",", ""))
+            Decimal(value, dt.precision, dt.scale)
+          }
+        case _: TimestampType if options.dateFormat != null =>
+          // This one will lose microseconds parts.
+          // See https://issues.apache.org/jira/browse/SPARK-10681.
+          options.dateFormat.parse(datum).getTime * 1000L
+        case _: TimestampType =>
+          // This one will lose microseconds parts.
+          // See https://issues.apache.org/jira/browse/SPARK-10681.
+          DateTimeUtils.stringToTime(datum).getTime  * 1000L
+        case _: DateType if options.dateFormat != null =>
+          DateTimeUtils.millisToDays(options.dateFormat.parse(datum).getTime)
+        case _: DateType =>
+          DateTimeUtils.millisToDays(DateTimeUtils.stringToTime(datum).getTime)
+        case _: StringType => UTF8String.fromString(datum)
+        case _ => throw new RuntimeException(s"Unsupported type: ${castType.typeName}")
+      }
     }
   }
 

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

@@ -431,7 +431,6 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
   }
 
   test("nullable fields with user defined null value of \"null\"") {
-
     // year,make,model,comment,blank
     val dataSchema = StructType(List(
       StructField("year", IntegerType, nullable = true),
@@ -447,7 +446,7 @@ class CSVSuite extends QueryTest with SharedSQLContext with SQLTestUtils {
 
     verifyCars(cars, withHeader = true, checkValues = false)
     val results = cars.collect()
-    assert(results(0).toSeq === Array(2012, "Tesla", "S", "null", "null"))
+    assert(results(0).toSeq === Array(2012, "Tesla", "S", null, null))
     assert(results(2).toSeq === Array(null, "Chevy", "Volt", null, null))
   }
 

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

@@ -73,10 +73,10 @@ class CSVTypeCastSuite extends SparkFunSuite {
 
   test("String type should always return the same as the input") {
     assert(
-      CSVTypeCast.castTo("", StringType, nullable = true, CSVOptions()) ==
+      CSVTypeCast.castTo("", StringType, nullable = true, CSVOptions("nullValue", null)) ==
         UTF8String.fromString(""))
     assert(
-      CSVTypeCast.castTo("", StringType, nullable = false, CSVOptions()) ==
+      CSVTypeCast.castTo("", StringType, nullable = false, CSVOptions("nullValue", null)) ==
         UTF8String.fromString(""))
   }
 
@@ -180,5 +180,13 @@ class CSVTypeCastSuite extends SparkFunSuite {
       CSVTypeCast.castTo("-", DoubleType, nullable = true, CSVOptions("nullValue", "-")))
     assertNull(
       CSVTypeCast.castTo("-", DecimalType.DoubleDecimal, true, CSVOptions("nullValue", "-")))
+    assertNull(
+      CSVTypeCast.castTo("-", StringType, nullable = true, CSVOptions("nullValue", "-")))
+    assertNull(
+      CSVTypeCast.castTo("-", TimestampType, nullable = true, CSVOptions("nullValue", "-")))
+    assertNull(
+      CSVTypeCast.castTo("-", DateType, nullable = true, CSVOptions("nullValue", "-")))
+    assertNull(
+      CSVTypeCast.castTo("-", BooleanType, nullable = true, CSVOptions("nullValue", "-")))
   }
 }

sql/hive/src/test/scala/org/apache/spark/sql/sources/CSVHadoopFsRelationSuite.scala

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.sources
+
+import org.apache.hadoop.fs.Path
+
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.sql.types._
+
+class CSVHadoopFsRelationSuite extends HadoopFsRelationTest {
+  override val dataSourceName: String = "csv"
+
+  override val extraReadOptions: Map[String, String] =
+    Map("header" -> "true", "inferSchema" -> "true")
+
+  override val extraWriteOptions: Map[String, String] = Map("header" -> "true")
+
+  override protected def supportsDataType(dataType: DataType): Boolean = dataType match {
+    case _: NullType => false
+    // `StringType` test is too flaky. Seems random generate data affects delimiter
+    // for writing and CSV parse does not recognize this.
+    case _: StringType => false
+    case _: BinaryType => false
+    case _: CalendarIntervalType => false
+    case _: ArrayType => false
+    case _: MapType => false
+    case _: StructType => false
+    // Currently, this writes `DateType` and `TimestampType` as a long value.
+    // Since `dateFormat` is not yet supported for writing, this is disabled for now.
+    case _: DateType => false
+    case _: TimestampType => false
+    case _: UserDefinedType[_] => false
+    case _ => true
+  }
+
+  test("save()/load() - partitioned table - simple queries - partition columns in data") {
+    withTempDir { file =>
+      val basePath = new Path(file.getCanonicalPath)
+      val fs = basePath.getFileSystem(SparkHadoopUtil.get.conf)
+      val qualifiedBasePath = fs.makeQualified(basePath)
+
+      for (p1 <- 1 to 2; p2 <- Seq("foo", "bar")) {
+        val partitionDir = new Path(qualifiedBasePath, s"p1=$p1/p2=$p2")
+        val header = Seq("a,b")
+        val data = (1 to 3).map(i => s"""$i,val_$i""")
+        sparkContext
+          .parallelize(header ++ data)
+          .saveAsTextFile(partitionDir.toString)
+      }
+
+      val dataSchemaWithPartition =
+        StructType(dataSchema.fields :+ StructField("p1", IntegerType, nullable = true))
+
+      checkQueries(
+        hiveContext.read.format(dataSourceName)
+          .option("dataSchema", dataSchemaWithPartition.json)
+          .option("inferSchema", "true")
+          .option("header", "true")
+          .load(file.getCanonicalPath))
+    }
+  }
+}