# Spark sql pushdown conjunction filters **URL:** \ **Category:** Elasticsearch\ **Tags:** es-hadoop\ **Created:** [November 23, 2015, 3:54pm UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324 "2015-11-23T15:54:50Z")\ **Posts on this page:** 11\ **Page:** 1
**Author:** ![joris\_renkens](https://avatars.discourse-cdn.com/v4/letter/j/838e76/32.png) [@joris\_renkens](https://discuss.elastic.co/u/joris_renkens)\ **Post date:** [November 23, 2015, 3:54pm UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324/1 "2015-11-23T15:54:50Z")
Hi, I'm trying to use elasticsearch as a data source for a spark script. Everything works fine but when I use a conjunctive filter: data.filter(data("user").equalTo("foo").and(data("start").lt("2015-10-25 01:00:000")).and(data("end").gt("2015-10-25 00:00:000"))) The pushdown only seems to work for the first filter in the conjunction. When I remove both the filter on "start" and "end", the same amount of data is sent to the query node as in the case of the full filter. Is this a known issue and is there a fix? When I do a count aggregation on the result, I get huge differences so the date filters restrict the result a lot. Furthermore, it would be really convenient if aggregations where also pushed down to the elasticsearch cluster when possible. Is it known when this will be the case? Cheers, Joris ---
**Author:** ![costin](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/costin/32/44950_2.png) [@costin](https://discuss.elastic.co/u/costin)\ **Post date:** [November 23, 2015, 9:29pm UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324/2 "2015-11-23T21:29:20Z")
What is the resulting queryDSL vs what you are expecting? You can find this out by [enabling `TRACE` logging](https://www.elastic.co/guide/en/elasticsearch/hadoop/master/logging.html) on the `org.elasticsearch.spark` package. ---
**Author:** ![joris\_renkens](https://avatars.discourse-cdn.com/v4/letter/j/838e76/32.png) [@joris\_renkens](https://discuss.elastic.co/u/joris_renkens)\ **Post date:** [November 25, 2015, 8:40am UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324/3 "2015-11-25T08:40:23Z")
When I enable trace logging, I see the following two lines which I think are relevant: 15/11/25 08:34:50 DEBUG DataSource: Pushing down filters [EqualTo(user,foo)] 15/11/25 08:34:50 TRACE DataSource: Transformed filters into DSL $filterString To me, this seems to show that only the first filter is pushed down. Does this maybe have something to do with the date format of the date filters? On further testing I do see that: data("user").equalTo("foo1").or(data("user").equalTo("foo2")) does get properly pushed down: 15/11/25 08:43:02 DEBUG DataSource: Pushing down filters [Or(EqualTo(user,foo1),EqualTo(user,foo2))] ---
**Author:** ![joris\_renkens](https://avatars.discourse-cdn.com/v4/letter/j/838e76/32.png) [@joris\_renkens](https://discuss.elastic.co/u/joris_renkens)\ **Post date:** [November 25, 2015, 9:14am UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324/4 "2015-11-25T09:14:23Z")
When I change the lt filter to a equalTo filter, it does get pushed down. Are lt and gt filters not yet supported for pushdown? Is there a workaround? ---
**Author:** ![joris\_renkens](https://avatars.discourse-cdn.com/v4/letter/j/838e76/32.png) [@joris\_renkens](https://discuss.elastic.co/u/joris_renkens)\ **Post date:** [November 25, 2015, 10:30am UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324/5 "2015-11-25T10:30:54Z")
I've resolved the issue. It seems that the lt filter cannot be pushed down with a string as argument. Changing the argument to a java.sql.Timestamp allowed the filter to be pushed down. ---
**Author:** ![costin](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/costin/32/44950_2.png) [@costin](https://discuss.elastic.co/u/costin)\ **Post date:** [November 25, 2015, 9:24pm UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324/6 "2015-11-25T21:24:59Z")
@joris_renkens Sorry but I don't follow. For each step can you post the sample Spark query and the resulting queryDSL in the logs and what is that you expect? Also what version of Spark and ES-Hadoop are you using? It is unclear what works and what doesn't, whether the queryDSL is incomplete (it might be because Spark SQL doesn't pass in all the `Filter`s) or incorrect. In other words, whether the optimization is not what you would expect or if it's a bug. And the info request above would help with that greatly. Thanks, ---
**Author:** ![joris\_renkens](https://avatars.discourse-cdn.com/v4/letter/j/838e76/32.png) [@joris\_renkens](https://discuss.elastic.co/u/joris_renkens)\ **Post date:** [November 27, 2015, 8:18am UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324/7 "2015-11-27T08:18:20Z")
@costin I don't actually get to see the queryDSL, I get the following line: 15/11/25 08:34:50 TRACE DataSource: Transformed filters into DSL $filterString What I do see is: 15/11/25 08:34:50 DEBUG DataSource: Pushing down filters [EqualTo(user,foo)] for the query: data.filter(data("user").equalTo("foo") .and(data("start").lt("2015-10-25 01:00:000")) .and(data("end").gt("2015-10-25 00:00:000")) ) This tells me that when I pass timestamps as strings, the filters don't get pushed down. However, I get the following log message: 15/11/27 08:14:43 DEBUG DataSource: Pushing down filters [EqualTo(user,foo),LessThan(start,2015-10-25 01:00:00.0),GreaterThan(end,2015-10-25 00:00:00.0)] for the query: data.filter( data("user").equalTo("foo") .and(data("start").lt(new Timestamp(1445727600000))) .and(data("end").gt(new Timestamp(1445724000000))) ) Which tells me that the filters do get pushed down when the timestamp is provided as a Timestamp object. Is this enough info? Cheers, Joris ---
**Author:** ![costin](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/costin/32/44950_2.png) [@costin](https://discuss.elastic.co/u/costin)\ **Post date:** [December 8, 2015, 2:13pm UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324/8 "2015-12-08T14:13:41Z")
> [@joris\_renkens](#): > > I get the following line:15/11/25 08:34:50 TRACE DataSource: Transformed filters into DSL $filterString That looks like a bug. What version of ES-Hadoop are you using? > [@joris\_renkens](#): > > Is this enough info? What does your DataFrame look like? What about your ES mapping? Lastly, what version of Spark and ES are you using? ---
**Author:** ![joris\_renkens](https://avatars.discourse-cdn.com/v4/letter/j/838e76/32.png) [@joris\_renkens](https://discuss.elastic.co/u/joris_renkens)\ **Post date:** [January 4, 2016, 8:06am UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324/9 "2016-01-04T08:06:17Z")
Sorry for the late answer. My versions: "elasticsearch-spark" % "2.2.0-beta1" "spark-core" % "1.5.2" elasticsearch-2.0.0 my dataframe has 4 fields: title,recordStart,recordEnd,recordDuration ``` "recordEnd": { "type": "date", "format": "dateOptionalTime" }, "recordStart": { "type": "date", "format": "dateOptionalTime" }, "recordDuration": { "type": "long" }, "title": { "type": "string", "fields": { "raw": {"type":"string", "index":"not_analyzed"} } }, ``` ---
**Author:** ![costin](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/costin/32/44950_2.png) [@costin](https://discuss.elastic.co/u/costin)\ **Post date:** [January 9, 2016, 1:07pm UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324/10 "2016-01-09T13:07:56Z")
The issue was handling the dateType which requires a strict (aka value) querying not a terms one. This has been solved in ES-Hadoop 2.2-rc1 - can you please try it out? ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [July 6, 2017, 1:26pm UTC](https://discuss.elastic.co/t/spark-sql-pushdown-conjunction-filters/35324/11 "2017-07-06T13:26:42Z")