Using Spark predicate push down in Spark SQL queries

Restrictions on column filters

Partition key columns can be pushed down as long as:

• All partition key columns are included in the filter.
• No more than one equivalence predicate per column.

Use an IN clause to specify multiple restrictions for a particular column:

val primaryColors = List("red", "yellow", "blue")

val df = spark.read.cassandraFormat("cars", "inventory").load
df.filter(df("car_color").isin(primaryColors: _*))

Clustering key columns can be pushed down with the following rules:

• Only the last predicate in the filter can be a non equivalence predicate.
• If there is more than one predicate for a column, the predicates cannot be equivalence predicates.

When predicate push down occurs

When a Dataset has no push down filters, all requests on the Dataset do a full unfiltered table scan. Adding predicate filters on the Dataset for eligible database columns modifies the underlying query to narrow its scope.

Determining if predicate push down is being used in queries

By using the explain method on a Dataset (or EXPLAIN in Spark SQL), queries can be analyzed to see if the predicates need to be cast to the correct data type. For example, create the following CQL table:

CREATE KEYSPACE test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1 };
USE test;
CREATE table words (
    user  TEXT, 
    word  TEXT, 
    count INT, 
    PRIMARY KEY (user, word));

INSERT INTO words (user, word, count ) VALUES ( 'Russ', 'dino', 10 );
INSERT INTO words (user, word, count ) VALUES ( 'Russ', 'fad', 5 );
INSERT INTO words (user, word, count ) VALUES ( 'Sam', 'alpha', 3 );
INSERT INTO words (user, word, count ) VALUES ( 'Zebra', 'zed', 100 );

Then create a Spark Dataset in the Spark console using that table and look for PushedFilters in the output after issuing the EXPLAIN command:

val df = spark.read.cassandraFormat("words", "test").load
df.explain

== Physical Plan ==
*Scan org.apache.spark.sql.cassandra.CassandraSourceRelation [user#0,word#1,count#2] ReadSchema: struct<user:string,word:string,count:int>

Because this query doesn't filter on columns capable of being pushed down, there are no PushedFilters in the physical plan.

Adding a filter, however, does change the physical plan to include PushedFilters:

val dfWithPushdown = df.filter(df("word") > "ham")
dfWithPushdown.explain

== Physical Plan ==
*Scan org.apache.spark.sql.cassandra.CassandraSourceRelation [user#0,word#1,count#2] PushedFilters: [*GreaterThan(word,ham)], ReadSchema: struct<user:string,word:string,count:int>

The PushedFilters section of the physical plan includes the GreaterThan push down filter. The asterisk indicates that push down filter will be handled only at the datasource level.

Troubleshooting predicate push down

When creating Spark SQL queries that use comparison operators, making sure the predicates are pushed down to the database correctly is critical to retrieving the correct data with the best performance.

For example, given a CQL table with the following schema:

CREATE TABLE test.common (
    year int,
    birthday timestamp,
    userid uuid,
    likes text,
    name text,
    PRIMARY KEY (year, birthday, userid)
)

Suppose you want to write a query that selects all entries where the birthday is earlier than a given date:

SELECT * FROM test.common WHERE birthday < '2001-1-1';

Use the EXPLAIN command to see the query plan:

EXPLAIN SELECT * FROM test.common WHERE birthday < '2001-1-1';

== Physical Plan ==
*Filter (cast(birthday#1 as string) < 2001-1-1)
+- *Scan org.apache.spark.sql.cassandra.CassandraSourceRelation [year#0,birthday#1,userid#2,likes#3,name#4] ReadSchema: struct<year:int,birthday:timestamp,userid:string,likes:string,name:string>
Time taken: 0.72 seconds, Fetched 1 row(s)

Note that the Filter directive is treating the birthday column, a CQL TIMESTAMP, as a string. The query optimizer looks at this comparison and needs to make the types match before generating a predicate. In this case the optimizer decides to cast the birthday column as a string to match the string '2001-1-1', but cast functions cannot be pushed down. The predicate isn't pushed down, and it doesn't appear in PushedFilters. A full table scan will be performed at the database layer, with the results returned to Spark for further processing.

To push down the correct predicate for this query, use the cast method to specify that the predicate is comparing the birthday column to a TIMESTAMP, so the types match and the optimizer can generate the correct predicate.

EXPLAIN SELECT * FROM test.common WHERE birthday < cast('2001-1-1' as TIMESTAMP);

== Physical Plan ==
*Scan org.apache.spark.sql.cassandra.CassandraSourceRelation [year#0,birthday#1,userid#2,likes#3,name#4] 
PushedFilters: [*LessThan(birthday,2001-01-01 00:00:00.0)], 
ReadSchema: struct<year:int,birthday:timestamp,userid:string,likes:string,name:string>
Time taken: 0.034 seconds, Fetched 1 row(s)

Note the PushedFilters indicating that the LessThan predicate will be pushed down for the column data in birthday. This should speed up the query as a full table scan will be avoided.