Predicate push-down on smart data sources

Smart data sources are those that support data processing directly in their own engine--where the data resides--by preventing unnecessary data to be sent to Apache Spark.

On example is a relational SQL database with a smart data source. Consider a table with three columns: column1, column2, and column3, where the third column contains a timestamp. In addition, consider an ApacheSparkSQL query using this JDBC data source but only accessing a subset of columns and rows based using projection and selection. The following SQL query is an example of such a task:

select column2,column3 from tab where column3>1418812500

Running on a smart data source, data locality is made use of by letting the SQL database do the filtering of rows based on timestamp and removal of column1.

Let's have a look at a practical example on how this is implemented in the Apache Spark MongoDB connector. First, we'll take a look at the class definition:

private[spark] case class MongoRelation(mongoRDD: MongoRDD[BsonDocument], _schema: Option[StructType])(@transient val sqlContext: SQLContext)
  extends BaseRelation
  with PrunedFilteredScan
  with InsertableRelation
 with LoggingTrait {

As you can see, the MongoRelation class extends BaseRelation. This is all that is needed to create a new plugin to the DataSource API in order to support an additional data source. However, this class also implemented the PrunedFilteredScan trait adding the buildScan method in order to support filtering on the data source itself. So let's take a look at the implementation of this method:

override def buildScan(requiredColumns: Array[String], filters: Array[Filter]): RDD[Row] = {
  // Fields that explicitly aren't nullable must also be added to the filters
  val pipelineFilters = schema
                        .fields
                        .filter(!_.nullable)
                        .map(_.name)
                        .map(IsNotNull) 
                        ++ filters

  if (requiredColumns.nonEmpty || pipelineFilters.nonEmpty) {
      logInfo(s"requiredColumns: ${requiredColumns.mkString(", ")},
          filters: ${pipelineFilters.mkString(", ")}")
  }
  mongoRDD.appendPipeline(createPipeline(requiredColumns, pipelineFilters))
                          .map(doc => documentToRow(doc, schema, requiredColumns))
}

It is not necessary to understand the complete code snippet, but you can see that two parameters are passed to the buildScan method: requiredColumns and filters. This means that the code can use this information to remove columns and rows directly using the MongoDB API.