@Test void shouldNotPushDownFilterFarFromRelation() { when(settings.getSettingValue(Settings.Key.QUERY_SIZE_LIMIT)).thenReturn(200); ReferenceExpression field = ref("name", STRING); Expression filterExpr = dsl.equal(field, literal("John")); List<NamedExpression> groupByExprs = Arrays.asList(named("age", ref("age", INTEGER))); List<NamedAggregator> aggregators = Arrays.asList(named("avg(age)", new AvgAggregator(Arrays.asList(ref("age", INTEGER)), DOUBLE))); String indexName = "test"; ElasticsearchIndex index = new ElasticsearchIndex(client, settings, indexName); PhysicalPlan plan = index.implement( filter( aggregation( relation(indexName), aggregators, groupByExprs ), filterExpr)); assertTrue(plan instanceof FilterOperator); }
@Override public PhysicalPlan implement(LogicalPlan plan) { ElasticsearchIndexScan indexScan = new ElasticsearchIndexScan(client, settings, indexName, new ElasticsearchExprValueFactory(getFieldTypes())); return plan.accept(new DefaultImplementor<ElasticsearchIndexScan>() { @Override public PhysicalPlan visitFilter(LogicalFilter node, ElasticsearchIndexScan context) { if (!(node.getChild().get(0) instanceof LogicalRelation)) { return super.visitFilter(node, context); } FilterQueryBuilder queryBuilder = new FilterQueryBuilder(new DefaultExpressionSerializer()); QueryBuilder query = queryBuilder.build(node.getCondition()); context.pushDown(query); return visitChild(node, context); } @Override public PhysicalPlan visitAggregation(LogicalAggregation node, ElasticsearchIndexScan context) { if ((node.getChild().get(0) instanceof LogicalRelation) || (node.getChild().get(0) instanceof LogicalFilter && node.getChild().get(0) .getChild().get(0) instanceof LogicalRelation)) { AggregationQueryBuilder builder = new AggregationQueryBuilder(new DefaultExpressionSerializer()); List<AggregationBuilder> aggregationBuilder = builder.buildAggregationBuilder(node.getAggregatorList(), node.getGroupByList()); context.pushDownAggregation(aggregationBuilder); context.pushTypeMapping( builder.buildTypeMapping(node.getAggregatorList(), node.getGroupByList())); return visitChild(node, context); } else { return super.visitAggregation(node, context); } } @Override public PhysicalPlan visitRelation(LogicalRelation node, ElasticsearchIndexScan context) { return indexScan; } }, indexScan); }
ElasticsearchIndex implements Table { @Override public PhysicalPlan implement(LogicalPlan plan) { ElasticsearchIndexScan indexScan = new ElasticsearchIndexScan(client, settings, indexName, new ElasticsearchExprValueFactory(getFieldTypes())); return plan.accept(new DefaultImplementor<ElasticsearchIndexScan>() { @Override public PhysicalPlan visitFilter(LogicalFilter node, ElasticsearchIndexScan context) { if (!(node.getChild().get(0) instanceof LogicalRelation)) { return super.visitFilter(node, context); } FilterQueryBuilder queryBuilder = new FilterQueryBuilder(new DefaultExpressionSerializer()); QueryBuilder query = queryBuilder.build(node.getCondition()); context.pushDown(query); return visitChild(node, context); } @Override public PhysicalPlan visitAggregation(LogicalAggregation node, ElasticsearchIndexScan context) { if ((node.getChild().get(0) instanceof LogicalRelation) || (node.getChild().get(0) instanceof LogicalFilter && node.getChild().get(0) .getChild().get(0) instanceof LogicalRelation)) { AggregationQueryBuilder builder = new AggregationQueryBuilder(new DefaultExpressionSerializer()); List<AggregationBuilder> aggregationBuilder = builder.buildAggregationBuilder(node.getAggregatorList(), node.getGroupByList()); context.pushDownAggregation(aggregationBuilder); context.pushTypeMapping( builder.buildTypeMapping(node.getAggregatorList(), node.getGroupByList())); return visitChild(node, context); } else { return super.visitAggregation(node, context); } } @Override public PhysicalPlan visitRelation(LogicalRelation node, ElasticsearchIndexScan context) { return indexScan; } }, indexScan); } }
ElasticsearchIndex implements Table { @Override public PhysicalPlan implement(LogicalPlan plan) { ElasticsearchIndexScan indexScan = new ElasticsearchIndexScan(client, settings, indexName, new ElasticsearchExprValueFactory(getFieldTypes())); return plan.accept(new DefaultImplementor<ElasticsearchIndexScan>() { @Override public PhysicalPlan visitFilter(LogicalFilter node, ElasticsearchIndexScan context) { if (!(node.getChild().get(0) instanceof LogicalRelation)) { return super.visitFilter(node, context); } FilterQueryBuilder queryBuilder = new FilterQueryBuilder(new DefaultExpressionSerializer()); QueryBuilder query = queryBuilder.build(node.getCondition()); context.pushDown(query); return visitChild(node, context); } @Override public PhysicalPlan visitAggregation(LogicalAggregation node, ElasticsearchIndexScan context) { if ((node.getChild().get(0) instanceof LogicalRelation) || (node.getChild().get(0) instanceof LogicalFilter && node.getChild().get(0) .getChild().get(0) instanceof LogicalRelation)) { AggregationQueryBuilder builder = new AggregationQueryBuilder(new DefaultExpressionSerializer()); List<AggregationBuilder> aggregationBuilder = builder.buildAggregationBuilder(node.getAggregatorList(), node.getGroupByList()); context.pushDownAggregation(aggregationBuilder); context.pushTypeMapping( builder.buildTypeMapping(node.getAggregatorList(), node.getGroupByList())); return visitChild(node, context); } else { return super.visitAggregation(node, context); } } @Override public PhysicalPlan visitRelation(LogicalRelation node, ElasticsearchIndexScan context) { return indexScan; } }, indexScan); }  }
ElasticsearchIndex implements Table { @Override public PhysicalPlan implement(LogicalPlan plan) { ElasticsearchIndexScan indexScan = new ElasticsearchIndexScan(client, settings, indexName, new ElasticsearchExprValueFactory(getFieldTypes())); return plan.accept(new DefaultImplementor<ElasticsearchIndexScan>() { @Override public PhysicalPlan visitFilter(LogicalFilter node, ElasticsearchIndexScan context) { if (!(node.getChild().get(0) instanceof LogicalRelation)) { return super.visitFilter(node, context); } FilterQueryBuilder queryBuilder = new FilterQueryBuilder(new DefaultExpressionSerializer()); QueryBuilder query = queryBuilder.build(node.getCondition()); context.pushDown(query); return visitChild(node, context); } @Override public PhysicalPlan visitAggregation(LogicalAggregation node, ElasticsearchIndexScan context) { if ((node.getChild().get(0) instanceof LogicalRelation) || (node.getChild().get(0) instanceof LogicalFilter && node.getChild().get(0) .getChild().get(0) instanceof LogicalRelation)) { AggregationQueryBuilder builder = new AggregationQueryBuilder(new DefaultExpressionSerializer()); List<AggregationBuilder> aggregationBuilder = builder.buildAggregationBuilder(node.getAggregatorList(), node.getGroupByList()); context.pushDownAggregation(aggregationBuilder); context.pushTypeMapping( builder.buildTypeMapping(node.getAggregatorList(), node.getGroupByList())); return visitChild(node, context); } else { return super.visitAggregation(node, context); } } @Override public PhysicalPlan visitRelation(LogicalRelation node, ElasticsearchIndexScan context) { return indexScan; } }, indexScan); }  @Override Map<String, ExprType> getFieldTypes(); @Override PhysicalPlan implement(LogicalPlan plan); }
ElasticsearchIndex implements Table { @Override public PhysicalPlan implement(LogicalPlan plan) { ElasticsearchIndexScan indexScan = new ElasticsearchIndexScan(client, settings, indexName, new ElasticsearchExprValueFactory(getFieldTypes())); return plan.accept(new DefaultImplementor<ElasticsearchIndexScan>() { @Override public PhysicalPlan visitFilter(LogicalFilter node, ElasticsearchIndexScan context) { if (!(node.getChild().get(0) instanceof LogicalRelation)) { return super.visitFilter(node, context); } FilterQueryBuilder queryBuilder = new FilterQueryBuilder(new DefaultExpressionSerializer()); QueryBuilder query = queryBuilder.build(node.getCondition()); context.pushDown(query); return visitChild(node, context); } @Override public PhysicalPlan visitAggregation(LogicalAggregation node, ElasticsearchIndexScan context) { if ((node.getChild().get(0) instanceof LogicalRelation) || (node.getChild().get(0) instanceof LogicalFilter && node.getChild().get(0) .getChild().get(0) instanceof LogicalRelation)) { AggregationQueryBuilder builder = new AggregationQueryBuilder(new DefaultExpressionSerializer()); List<AggregationBuilder> aggregationBuilder = builder.buildAggregationBuilder(node.getAggregatorList(), node.getGroupByList()); context.pushDownAggregation(aggregationBuilder); context.pushTypeMapping( builder.buildTypeMapping(node.getAggregatorList(), node.getGroupByList())); return visitChild(node, context); } else { return super.visitAggregation(node, context); } } @Override public PhysicalPlan visitRelation(LogicalRelation node, ElasticsearchIndexScan context) { return indexScan; } }, indexScan); }  @Override Map<String, ExprType> getFieldTypes(); @Override PhysicalPlan implement(LogicalPlan plan);  }