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);  }
@Test void implementRelationOperatorOnly() { when(settings.getSettingValue(Settings.Key.QUERY_SIZE_LIMIT)).thenReturn(200); String indexName = "test"; LogicalPlan plan = relation(indexName); Table index = new ElasticsearchIndex(client, settings, indexName); assertEquals( new ElasticsearchIndexScan(client, settings, indexName, exprValueFactory), index.implement(plan)); }
@Test void implementOtherLogicalOperators() { when(settings.getSettingValue(Settings.Key.QUERY_SIZE_LIMIT)).thenReturn(200); String indexName = "test"; NamedExpression include = named("age", ref("age", INTEGER)); ReferenceExpression exclude = ref("name", STRING); ReferenceExpression dedupeField = ref("name", STRING); Expression filterExpr = literal(ExprBooleanValue.of(true)); 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))); Map<ReferenceExpression, ReferenceExpression> mappings = ImmutableMap.of(ref("name", STRING), ref("lastname", STRING)); Pair<ReferenceExpression, Expression> newEvalField = ImmutablePair.of(ref("name1", STRING), ref("name", STRING)); Integer sortCount = 100; Pair<Sort.SortOption, Expression> sortField = ImmutablePair.of(Sort.SortOption.PPL_ASC, ref("name1", STRING)); LogicalPlan plan = project( LogicalPlanDSL.dedupe( sort( eval( remove( rename( relation(indexName), mappings), exclude), newEvalField), sortCount, sortField), dedupeField), include); Table index = new ElasticsearchIndex(client, settings, indexName); assertEquals( PhysicalPlanDSL.project( PhysicalPlanDSL.dedupe( PhysicalPlanDSL.sort( PhysicalPlanDSL.eval( PhysicalPlanDSL.remove( PhysicalPlanDSL.rename( new ElasticsearchIndexScan(client, settings, indexName, exprValueFactory), mappings), exclude), newEvalField), sortCount, sortField), dedupeField), include), index.implement(plan)); }
@Test void shouldDiscardPhysicalFilterIfConditionPushedDown() { when(settings.getSettingValue(Settings.Key.QUERY_SIZE_LIMIT)).thenReturn(200); ReferenceExpression field = ref("name", STRING); NamedExpression named = named("n", field); Expression filterExpr = dsl.equal(field, literal("John")); String indexName = "test"; ElasticsearchIndex index = new ElasticsearchIndex(client, settings, indexName); PhysicalPlan plan = index.implement( project( filter( relation(indexName), filterExpr ), named)); assertTrue(plan instanceof ProjectOperator); assertTrue(((ProjectOperator) plan).getInput() instanceof ElasticsearchIndexScan); }
@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); }
@Test void shouldPushDownAggregation() { 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.getChild().get(0) instanceof ElasticsearchIndexScan); plan = index.implement( aggregation( filter( relation(indexName), filterExpr), aggregators, groupByExprs)); assertTrue(plan instanceof ElasticsearchIndexScan); }
@Test void shouldNotPushDownAggregationFarFromRelation() { 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( aggregation( filter(filter( relation(indexName), filterExpr), filterExpr), aggregators, groupByExprs)); assertTrue(plan instanceof AggregationOperator); }