[SC-5504][REDSHIFT] Fix filter pushdown when no columns are selected

Escaping single quotes should be done during creation of the UNLOAD statement.

This also fixes a longstanding bug where backslashes were not escaped properly when they appeared in string literals in generated `WHERE` clauses.

Author: Josh Rosen <[email protected]>
Author: Juliusz Sompolski <[email protected]>
Author: Adrian Ionescu <[email protected]>

Closes apache#174 from juliuszsompolski/sc5504.

Author: adrian-ionescu

external/redshift-integration-tests/src/test/scala/com/databricks/spark/redshift/IntegrationSuiteBase.scala

@@ -160,7 +160,8 @@ trait IntegrationSuiteBase
          |(0, null, '2015-07-03', 0.0, -1.0, 4141214, 1239012341823719, null, 'f', '2015-07-03 00:00:00.000'),
          |(0, false, null, -1234152.12312498, 100000.0, null, 1239012341823719, 24, '___|_123', null),
          |(1, false, '2015-07-02', 0.0, 0.0, 42, 1239012341823719, -13, 'asdf', '2015-07-02 00:00:00.000'),
-         |(1, true, '2015-07-01', 1234152.12312498, 1.0, 42, 1239012341823719, 23, 'Unicode''s樂趣', '2015-07-01 00:00:00.001')
+         |(1, true, '2015-07-01', 1234152.12312498, 1.0, 42, 1239012341823719, 23, 'Unicode''s樂趣', '2015-07-01 00:00:00.001'),
+         |(null, null, null, null, null, null, null, null, 'Ba\\\\ckslash\\\\', null)
          """.stripMargin
     )
     // scalastyle:on

external/redshift-integration-tests/src/test/scala/com/databricks/spark/redshift/RedshiftReadSuite.scala

@@ -110,7 +110,7 @@ class RedshiftReadSuite extends IntegrationSuiteBase {
   test("Can load output of Redshift aggregation queries") {
     checkAnswer(
       read.option("query", s"select testbool, count(*) from $test_table group by testbool").load(),
-      Seq(Row(true, 1), Row(false, 2), Row(null, 2)))
+      Seq(Row(true, 1), Row(false, 2), Row(null, 3)))
   }
 
   test("multiple scans on same table") {
@@ -128,6 +128,7 @@ class RedshiftReadSuite extends IntegrationSuiteBase {
     checkAnswer(
       sqlContext.sql("select testbyte, testbool from test_table"),
       Seq(
+        Row(null, null),
         Row(null, null),
         Row(0.toByte, null),
         Row(0.toByte, false),
@@ -240,4 +241,72 @@ class RedshiftReadSuite extends IntegrationSuiteBase {
       .load()
     assert(df.schema.fields(0).dataType === LongType)
   }
+
+  test("properly escape literals in filter pushdown (SC-5504)") {
+    checkAnswer(
+      sqlContext.sql("select count(1) from test_table where testint = 4141214"),
+      Seq(Row(1))
+    )
+    checkAnswer(
+      sqlContext.sql("select count(1) from test_table where testint = 7"),
+      Seq(Row(0))
+    )
+    checkAnswer(
+      sqlContext.sql("select testint from test_table where testint = 42"),
+      Seq(Row(42), Row(42))
+    )
+
+    checkAnswer(
+      sqlContext.sql("select count(1) from test_table where teststring = 'asdf'"),
+      Seq(Row(1))
+    )
+    checkAnswer(
+      sqlContext.sql("select count(1) from test_table where teststring = 'alamakota'"),
+      Seq(Row(0))
+    )
+    checkAnswer(
+      sqlContext.sql("select teststring from test_table where teststring = 'asdf'"),
+      Seq(Row("asdf"))
+    )
+
+    checkAnswer(
+      sqlContext.sql("select count(1) from test_table where teststring = 'a\\'b'"),
+      Seq(Row(0))
+    )
+    checkAnswer(
+      sqlContext.sql("select teststring from test_table where teststring = 'a\\'b'"),
+      Seq()
+    )
+
+    // scalastyle:off
+    checkAnswer(
+      sqlContext.sql("select count(1) from test_table where teststring = 'Unicode\\'s樂趣'"),
+      Seq(Row(1))
+    )
+    checkAnswer(
+      sqlContext.sql("select teststring from test_table where teststring = \"Unicode's樂趣\""),
+      Seq(Row("Unicode's樂趣"))
+    )
+    // scalastyle:on
+
+    checkAnswer(
+      sqlContext.sql("select count(1) from test_table where teststring = 'a\\\\b'"),
+      Seq(Row(0))
+    )
+    checkAnswer(
+      sqlContext.sql("select teststring from test_table where teststring = 'a\\\\b'"),
+      Seq()
+    )
+
+    checkAnswer(
+      sqlContext.sql(
+        "select count(1) from test_table where teststring = 'Ba\\\\ckslash\\\\'"),
+      Seq(Row(1))
+    )
+    checkAnswer(
+      sqlContext.sql(
+        "select teststring from test_table where teststring = \"Ba\\\\ckslash\\\\\""),
+      Seq(Row("Ba\\ckslash\\"))
+    )
+  }
 }

external/redshift/src/main/scala/com/databricks/spark/redshift/FilterPushdown.scala

@@ -40,9 +40,9 @@ private[redshift] object FilterPushdown {
     def buildComparison(attr: String, value: Any, comparisonOp: String): Option[String] = {
       getTypeForAttribute(schema, attr).map { dataType =>
         val sqlEscapedValue: String = dataType match {
-          case StringType => s"\\'${value.toString.replace("'", "\\'\\'")}\\'"
-          case DateType => s"\\'${value.asInstanceOf[Date]}\\'"
-          case TimestampType => s"\\'${value.asInstanceOf[Timestamp]}\\'"
+          case StringType => s"'${value.toString.replace("'", "''").replace("\\", "\\\\")}'"
+          case DateType => s"'${value.asInstanceOf[Date]}'"
+          case TimestampType => s"'${value.asInstanceOf[Timestamp]}'"
           case _ => value.toString
         }
         s""""$attr" $comparisonOp $sqlEscapedValue"""

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

@@ -17,6 +17,7 @@ import scala.collection.JavaConverters._
 import com.amazonaws.auth.AWSCredentialsProvider
 import com.amazonaws.services.s3.AmazonS3Client
 import com.databricks.spark.redshift.Parameters.MergedParameters
+import com.databricks.spark.redshift.Utils.escapeJdbcString
 import com.eclipsesource.json.Json
 import org.slf4j.LoggerFactory
 
@@ -178,8 +179,8 @@ private[redshift] case class RedshiftRelation(
     val query = {
       // Since the query passed to UNLOAD will be enclosed in single quotes, we need to escape
       // any backslashes and single quotes that appear in the query itself
-      val escapedTableNameOrSubqury = tableNameOrSubquery.replace("\\", "\\\\").replace("'", "\\'")
-      s"SELECT $columnList FROM $escapedTableNameOrSubqury $whereClause"
+      s"SELECT $columnList FROM ${escapeJdbcString(tableNameOrSubquery)} " +
+        s"${escapeJdbcString(whereClause)}"
     }
     // We need to remove S3 credentials from the unload path URI because they will conflict with
     // the credentials passed via `credsString`.

external/redshift/src/main/scala/com/databricks/spark/redshift/Utils.scala

@@ -197,4 +197,11 @@ private[redshift] object Utils {
       case _ => None
     }
   }
+
+  /**
+   * Escapes a string, so that it can be passed as a JDBC string literal.
+   */
+  def escapeJdbcString(s: String): String = {
+    s.replace("\\", "\\\\").replace("'", "\\'")
+  }
 }

external/redshift/src/test/scala/com/databricks/spark/redshift/FilterPushdownSuite.scala

@@ -32,11 +32,8 @@ class FilterPushdownSuite extends SparkFunSuite {
   test("buildWhereClause with string literals that contain Unicode characters") {
     // scalastyle:off
     val whereClause = buildWhereClause(testSchema, Seq(EqualTo("test_string", "Unicode's樂趣")))
-    // Here, the apostrophe in the string needs to be replaced with two single quotes, '', but we
-    // also need to escape those quotes with backslashes because this WHERE clause is going to
-    // eventually be embedded inside of a single-quoted string that's embedded inside of a larger
-    // Redshift query.
-    assert(whereClause === """WHERE "test_string" = \'Unicode\'\'s樂趣\'""")
+    // Here, the apostrophe in the string needs to be replaced with two single quotes, ''.
+    assert(whereClause === """WHERE "test_string" = 'Unicode''s樂趣'""")
     // scalastyle:on
   }
 
@@ -57,7 +54,7 @@ class FilterPushdownSuite extends SparkFunSuite {
     val expectedWhereClause =
       """
         |WHERE "test_bool" = true
-        |AND "test_string" = \'Unicode是樂趣\'
+        |AND "test_string" = 'Unicode是樂趣'
         |AND "test_double" > 1000.0
         |AND "test_double" < 1.7976931348623157E308
         |AND "test_float" >= 1.0

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

@@ -144,6 +144,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|
+        |||||||||Ba\\ckslash\\|
         ||||||||||
       """.stripMargin.trim
     // scalastyle:on

external/redshift/src/test/scala/com/databricks/spark/redshift/TestUtils.scala

@@ -52,7 +52,9 @@ object TestUtils {
       1239012341823719L, null, "f", TestUtils.toTimestamp(2015, 6, 3, 0, 0, 0)),
     Row(0.toByte, false, null, -1234152.12312498, 100000.0f, null, 1239012341823719L, 24.toShort,
       "___|_123", null),
-    Row(List.fill(10)(null): _*))
+    Row(List.fill(10)(null): _*),
+    Row(null, null, null, null, null, null, null, null, "Ba\\ckslash\\", null)
+  )
   // scalastyle:on
 
   /**