ElasticsearchIndexScan extends TableScanOperator { public void pushDown(QueryBuilder query) { SearchSourceBuilder source = request.getSourceBuilder(); QueryBuilder current = source.query(); if (current == null) { source.query(query); } else { if (isBoolFilterQuery(current)) { ((BoolQueryBuilder) current).filter(query); } else { source.query(QueryBuilders.boolQuery() .filter(current) .filter(query)); } } if (source.sorts() == null) { source.sort(DOC_FIELD_NAME, ASC); } } ElasticsearchIndexScan(ElasticsearchClient client,
                                Settings settings, String indexName,
                                ElasticsearchExprValueFactory exprValueFactory); @Override void open(); @Override boolean hasNext(); @Override ExprValue next(); void pushDown(QueryBuilder query); void pushDownAggregation(List<AggregationBuilder> aggregationBuilderList); void pushTypeMapping(Map<String, ExprType> typeMapping); @Override void close();  }
@Test void pushDownFilters() { assertThat() .pushDown(QueryBuilders.termQuery("name", "John")) .shouldQuery(QueryBuilders.termQuery("name", "John")) .pushDown(QueryBuilders.termQuery("age", 30)) .shouldQuery( QueryBuilders.boolQuery() .filter(QueryBuilders.termQuery("name", "John")) .filter(QueryBuilders.termQuery("age", 30))) .pushDown(QueryBuilders.rangeQuery("balance").gte(10000)) .shouldQuery( QueryBuilders.boolQuery() .filter(QueryBuilders.termQuery("name", "John")) .filter(QueryBuilders.termQuery("age", 30)) .filter(QueryBuilders.rangeQuery("balance").gte(10000))); }