Implement Spark 1.6's new unhandledFilters API

.travis.yml

@@ -23,6 +23,10 @@ matrix:
     - jdk: openjdk7
       scala: 2.10.5
       env: HADOOP_VERSION="2.2.0" SPARK_VERSION="1.5.0" SPARK_AVRO_VERSION="2.0.1"
+    # Tests against Spark 1.6.0-SNAPHSOT, to be updated once 1.6.0 is released:
+    - jdk: openjdk7
+      scala: 2.10.5
+      env: HADOOP_VERSION="2.2.0" SPARK_VERSION="1.6.0-SNAPSHOT" SPARK_AVRO_VERSION="2.0.1"
     # Configuration corresponding to DBC 1.4.x driver package as of DBC 2.4,
     # which uses spark-avro 1.0.0. We use Hadoop 2.2.0 here, while DBC uses
     # 1.2.1, because the 1.4.1 published to Maven Central is a Hadoop 2.x build.

project/SparkRedshiftBuild.scala

@@ -53,8 +53,10 @@ object SparkRedshiftBuild extends Build {
       spIgnoreProvided := true,
       licenses += "Apache-2.0" -> url("http://opensource.org/licenses/Apache-2.0"),
       credentials += Credentials(Path.userHome / ".ivy2" / ".credentials"),
-      resolvers +=
+      resolvers ++= Seq(
         "Sonatype OSS Snapshots" at "https://oss.sonatype.org/content/repositories/snapshots",
+        "ASF Snapshots" at "http://repository.apache.org/snapshots/"
+      ),
       scalacOptions ++= Seq("-target:jvm-1.6"),
       javacOptions ++= Seq("-source", "1.6", "-target", "1.6"),
       libraryDependencies ++= Seq(

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

@@ -18,7 +18,7 @@ package com.databricks.spark.redshift
 
 import java.sql.SQLException
 
-import org.apache.spark.sql.{AnalysisException, Row, SQLContext, SaveMode}
+import org.apache.spark.sql.{execution, AnalysisException, Row, SaveMode}
 import org.apache.spark.sql.types._
 
 /**
@@ -268,6 +268,15 @@ class RedshiftIntegrationSuite extends IntegrationSuiteBase {
     // scalastyle:on
   }
 
+  test("RedshiftRelation implements Spark 1.6+'s unhandledFilters API") {
+    assume(org.apache.spark.SPARK_VERSION.take(3) >= "1.6")
+    val df = sqlContext.sql("select testbool from test_table where testbool = true")
+    val physicalPlan = df.queryExecution.sparkPlan
+    physicalPlan.collectFirst { case f: execution.Filter => f }.foreach { filter =>
+      fail(s"Filter should have been eliminated; plan is:\n$physicalPlan")
+    }
+  }
+
   test("roundtrip save and load") {
     // This test can be simplified once #98 is fixed.
     val tableName = s"roundtrip_save_and_load_$randomSuffix"

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

@@ -41,7 +41,7 @@ private[redshift] object FilterPushdown {
    * Attempt to convert the given filter into a SQL expression. Returns None if the expression
    * could not be converted.
    */
-  private def buildFilterExpression(schema: StructType, filter: Filter): Option[String] = {
+  def buildFilterExpression(schema: StructType, filter: Filter): Option[String] = {
     def buildComparison(attr: String, value: Any, comparisonOp: String): Option[String] = {
      getTypeForAttribute(schema, attr).map { dataType =>
        val sqlEscapedValue: String = dataType match {

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

@@ -77,6 +77,13 @@ private[redshift] case class RedshiftRelation(
     writer.saveToRedshift(sqlContext, data, saveMode, params)
   }
 
+  // In Spark 1.6+, this method allows a data source to declare which filters it handles, allowing
+  // Spark to skip its own defensive filtering. See SPARK-10978 for more details. As long as we
+  // compile against Spark 1.4, we cannot use the `override` modifier here.
+  def unhandledFilters(filters: Array[Filter]): Array[Filter] = {
+    filters.filterNot(filter => FilterPushdown.buildFilterExpression(schema, filter).isDefined)
+  }
+
   override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
     val creds =
       AWSCredentialsUtils.load(params.rootTempDir, sqlContext.sparkContext.hadoopConfiguration)