Empty string is converted to null

src/it/scala/com/databricks/spark/redshift/RedshiftReadSuite.scala

@@ -206,6 +206,21 @@ class RedshiftReadSuite extends IntegrationSuiteBase {
     }
   }
 
+  test("test empty string and null") {
+    withTempRedshiftTable("records_with_empty_and_null_characters") { tableName =>
+      conn.createStatement().executeUpdate(
+        s"CREATE TABLE $tableName (x varchar(256))")
+      conn.createStatement().executeUpdate(
+        s"INSERT INTO $tableName VALUES ('null'), (''), (null)")
+      conn.commit()
+      assert(DefaultJDBCWrapper.tableExists(conn, tableName))
+      checkAnswer(
+        read.option("dbtable", tableName).load(),
+        Seq("null", "", null).map(x => Row.apply(x)))
+    }
+  }
+
+
   test("read special double values (regression test for #261)") {
     val tableName = s"roundtrip_special_double_values_$randomSuffix"
     try {

src/main/scala/com/databricks/spark/redshift/Conversions.scala

@@ -78,7 +78,7 @@ private[redshift] object Conversions {
    *
    * Note that instances of this function are NOT thread-safe.
    */
-  def createRowConverter(schema: StructType): Array[String] => InternalRow = {
+  def createRowConverter(schema: StructType, nullString: String): Array[String] => InternalRow = {
     val dateFormat = createRedshiftDateFormat()
     val decimalFormat = createRedshiftDecimalFormat()
     val conversionFunctions: Array[String => Any] = schema.fields.map { field =>
@@ -116,7 +116,16 @@ private[redshift] object Conversions {
       var i = 0
       while (i < schema.length) {
         val data = inputRow(i)
-        converted(i) = if (data == null || data.isEmpty) null else conversionFunctions(i)(data)
+        converted(i) = if ((data == null || data == nullString) ||
+          (data.isEmpty && schema.fields(i).dataType != StringType)) {
+          null
+        }
+        else if (data.isEmpty) {
+          ""
+        }
+        else {
+          conversionFunctions(i)(data)
+        }
         i += 1
       }
       encoder.toRow(externalRow)

src/main/scala/com/databricks/spark/redshift/RedshiftFileFormat.scala

@@ -95,7 +95,8 @@ private[redshift] class RedshiftFileFormat extends FileFormat {
       // be closed once it is completely iterated, but this is necessary to guard against
       // resource leaks in case the task fails or is interrupted.
       Option(TaskContext.get()).foreach(_.addTaskCompletionListener(_ => iter.close()))
-      val converter = Conversions.createRowConverter(requiredSchema)
+      val converter = Conversions.createRowConverter(requiredSchema,
+        options.getOrElse("nullString", Parameters.DEFAULT_PARAMETERS("csvnullstring")))
       iter.map(converter)
     }
   }

src/main/scala/com/databricks/spark/redshift/RedshiftRelation.scala

@@ -164,6 +164,7 @@ private[redshift] case class RedshiftRelation(
       sqlContext.read
         .format(classOf[RedshiftFileFormat].getName)
         .schema(prunedSchema)
+        .option("nullString", params.nullString)
         .load(filesToRead: _*)
         .queryExecution.executedPlan.execute().asInstanceOf[RDD[Row]]
     }
@@ -193,7 +194,8 @@ private[redshift] case class RedshiftRelation(
     // the credentials passed via `credsString`.
     val fixedUrl = Utils.fixS3Url(Utils.removeCredentialsFromURI(new URI(tempDir)).toString)
 
-    s"UNLOAD ('$query') TO '$fixedUrl' WITH CREDENTIALS '$credsString' ESCAPE MANIFEST"
+    s"UNLOAD ('$query') TO '$fixedUrl' WITH CREDENTIALS '$credsString'" +
+      s" ESCAPE MANIFEST NULL AS '${params.nullString}'"
   }
 
   private def pruneSchema(schema: StructType, columns: Array[String]): StructType = {

src/test/scala/com/databricks/spark/redshift/ConversionsSuite.scala

@@ -31,7 +31,8 @@ import org.apache.spark.sql.types._
 class ConversionsSuite extends FunSuite {
 
   private def createRowConverter(schema: StructType) = {
-    Conversions.createRowConverter(schema).andThen(RowEncoder(schema).resolveAndBind().fromRow)
+    Conversions.createRowConverter(schema, Parameters.DEFAULT_PARAMETERS("csvnullstring"))
+      .andThen(RowEncoder(schema).resolveAndBind().fromRow)
   }
 
   test("Data should be correctly converted") {

src/test/scala/com/databricks/spark/redshift/RedshiftSourceSuite.scala

@@ -150,7 +150,7 @@ class RedshiftSourceSuite
         |1|f|2015-07-02|0|0.0|42|1239012341823719|-13|asdf|2015-07-02 00:00:00.0
         |0||2015-07-03|0.0|-1.0|4141214|1239012341823719||f|2015-07-03 00:00:00
         |0|f||-1234152.12312498|100000.0||1239012341823719|24|___\|_123|
-        ||||||||||
+        |||||||||@NULL@|
       """.stripMargin.trim
     // scalastyle:on
     val expectedQuery = (