AggregateHandler implements OutputOperationHandler<Aggregate, Iterable<? extends Element>> { @Override public Iterable<? extends Element> doOperation(final Aggregate operation, final Context context, final Store store) throws OperationException { return doOperation(operation, store.getSchema()); }  @Override Iterable<? extends Element> doOperation(final Aggregate operation, final Context context, final Store store); Iterable<? extends Element> doOperation(final Aggregate operation, final Schema schema);  }
@Test public void shouldAggregateElementsWhenNoGroupByInSchema() throws OperationException { final Schema schema = new Schema.Builder() .entity(TestGroups.ENTITY, new SchemaEntityDefinition.Builder() .build()) .build(); given(store.getSchema()).willReturn(schema); input.add(entity); input.add(entity1); input.add(entity2); final AggregatePair pair = new AggregatePair( new String[]{"timestamp"}, new ElementAggregator.Builder() .select("count") .execute(new Sum()) .build()); entities.put(TestGroups.ENTITY, pair); final Entity expectedEntity = new Entity.Builder() .group(TestGroups.ENTITY) .property("timestamp", 3L) .property("count", 5) .build(); expected.add(expectedEntity); expected.add(entity1); final Aggregate aggregate = new Aggregate.Builder() .input(input) .entities(entities) .build(); final Iterable<? extends Element> results = handler.doOperation(aggregate, context, store); final Set<Element> resultsSet = Sets.newHashSet(results); assertEquals(expected, resultsSet); }
@Test public void shouldAggregateElementsWhenAggregatorNotProvidedInSchema() throws OperationException { final Schema schema = new Schema.Builder() .entity(TestGroups.ENTITY, new SchemaEntityDefinition.Builder() .groupBy("timestamp") .build()) .build(); given(store.getSchema()).willReturn(schema); input.add(entity); input.add(entity1); input.add(entity2); input.add(entity3); final AggregatePair pair = new AggregatePair( new ElementAggregator.Builder() .select("count") .execute(new Sum()) .build()); entities.put(TestGroups.ENTITY, pair); final Entity expectedEntity = new Entity.Builder() .group(TestGroups.ENTITY) .property("timestamp", 3L) .property("count", 5) .build(); final Entity expectedEntity1 = new Entity.Builder() .group(TestGroups.ENTITY) .property("timestamp", 2L) .property("count", 6) .build(); expected.add(expectedEntity); expected.add(expectedEntity1); final Aggregate aggregate = new Aggregate.Builder() .input(input) .entities(entities) .build(); final Iterable<? extends Element> results = handler.doOperation(aggregate, context, store); final Set<Element> resultsSet = Sets.newHashSet(results); assertEquals(expected, resultsSet); }
@Test public void shouldAggregateTheSameFromSchemaOrOperation() throws OperationException { final Schema schema = new Schema.Builder() .edge(TestGroups.EDGE, new SchemaEdgeDefinition.Builder() .build()) .build(); final Store store1 = mock(Store.class); final List<Element> input1 = new ArrayList<>(); final Set<Element> expected1 = new HashSet<>(); final Map<String, AggregatePair> edges1 = new HashMap<>(); final Schema schema1 = new Schema.Builder() .edge(TestGroups.EDGE, new SchemaEdgeDefinition.Builder() .aggregator(new ElementAggregator.Builder() .select("count") .execute(new Sum()) .select("turns") .execute(new Sum()) .build()) .groupBy("timestamp") .build()) .build(); given(store.getSchema()).willReturn(schema); given(store1.getSchema()).willReturn(schema1); input.add(edge); input.add(edge1); input.add(edge2); final Edge localEdge = new Edge.Builder() .group(TestGroups.EDGE) .property("timestamp", 2L) .property("turns", 6) .property("count", 2L) .build(); final Edge localEdge1 = new Edge.Builder() .group(TestGroups.EDGE) .property("timestamp", 1L) .property("turns", 9) .property("count", 5L) .build(); final Edge localEdge2 = new Edge.Builder() .group(TestGroups.EDGE) .property("timestamp", 2L) .property("turns", 4) .property("count", 1L) .build(); input1.add(localEdge); input1.add(localEdge1); input1.add(localEdge2); final AggregatePair pair = new AggregatePair( new String[]{"timestamp"}, new ElementAggregator.Builder() .select("count") .execute(new Sum()) .select("turns") .execute(new Sum()) .build()); final AggregatePair otherPair = new AggregatePair(); edges.put(TestGroups.EDGE, pair); edges1.put(TestGroups.EDGE, otherPair); final Edge expectedEdge = new Edge.Builder() .group(TestGroups.EDGE) .property("timestamp", 2L) .property("turns", 10) .property("count", 3L) .build(); expected.add(expectedEdge); expected.add(edge1); expected1.add(expectedEdge); expected1.add(edge1); final Aggregate aggregate = new Aggregate.Builder() .input(input) .edges(edges) .build(); final Aggregate aggregate1 = new Aggregate.Builder() .input(input1) .edges(edges1) .build(); final Iterable<? extends Element> results = handler.doOperation(aggregate, context, store); final Set<Element> resultsSet = Sets.newHashSet(results); final Iterable<? extends Element> results1 = handler.doOperation(aggregate1, context, store1); final Set<Element> resultsSet1 = Sets.newHashSet(results1); assertEquals(expected, resultsSet); assertEquals(expected1, resultsSet1); }
@Test public void shouldAggregateAMixOfEdgesAndEntities() throws OperationException { final Schema schema = new Schema.Builder() .edge(TestGroups.EDGE, new SchemaEdgeDefinition.Builder() .aggregator(new ElementAggregator.Builder() .select("turns") .execute(new Max()) .build()) .build()) .entity(TestGroups.ENTITY, new SchemaEntityDefinition.Builder() .aggregator(new ElementAggregator.Builder() .select("count") .execute(new Sum()) .build()) .build()) .build(); given(store.getSchema()).willReturn(schema); input.add(edge); input.add(edge1); input.add(edge2); input.add(entity); input.add(entity1); input.add(entity2); input.add(entity3); final AggregatePair edgePair = new AggregatePair( new String[]{"timestamp"} ); final AggregatePair entityPair = new AggregatePair( new String[]{"timestamp"} ); edges.put(TestGroups.EDGE, edgePair); entities.put(TestGroups.ENTITY, entityPair); final Edge expectedEdge = new Edge.Builder() .group(TestGroups.EDGE) .property("timestamp", 2L) .property("turns", 6) .property("count", 2L) .build(); final Edge expectedEdge1 = new Edge.Builder() .group(TestGroups.EDGE) .property("timestamp", 1L) .property("turns", 9) .property("count", 5L) .build(); final Entity expectedEntity = new Entity.Builder() .group(TestGroups.ENTITY) .property("timestamp", 2L) .property("count", 6) .build(); final Entity expectedEntity1 = new Entity.Builder() .group(TestGroups.ENTITY) .property("timestamp", 3L) .property("count", 5) .build(); expected.add(expectedEdge); expected.add(expectedEdge1); expected.add(expectedEntity); expected.add(expectedEntity1); final Aggregate aggregate = new Aggregate.Builder() .input(input) .edges(edges) .entities(entities) .build(); final Iterable<? extends Element> results = handler.doOperation(aggregate, context, store); final Set<Element> resultsSet = Sets.newHashSet(results); assertEquals(expected, resultsSet); }
@Test public void shouldAggregateEdgesFromMultipleGroups() throws OperationException { final Schema schema = new Schema.Builder() .edge(TestGroups.EDGE, new SchemaEdgeDefinition.Builder() .aggregator(new ElementAggregator.Builder() .select("turns") .execute(new Max()) .build()) .groupBy("timestamp") .build()) .edge(TestGroups.EDGE_2, new SchemaEdgeDefinition.Builder() .aggregator(new ElementAggregator.Builder() .select("count") .execute(new Sum()) .select("length") .execute(new Sum()) .build()) .groupBy("timestamp") .build()) .build(); given(store.getSchema()).willReturn(schema); input.add(edge); input.add(edge1); input.add(edge2); input.add(edge3); input.add(edge4); final AggregatePair pair = new AggregatePair(); final AggregatePair pair1 = new AggregatePair(); edges.put(TestGroups.EDGE, pair); edges.put(TestGroups.EDGE_2, pair1); final Edge expectedEdge = new Edge.Builder() .group(TestGroups.EDGE) .property("timestamp", 2L) .property("turns", 6) .property("count", 2L) .build(); final Edge expectedEdge1 = new Edge.Builder() .group(TestGroups.EDGE) .property("timestamp", 1L) .property("turns", 9) .property("count", 5L) .build(); final Edge expectedEdge2 = new Edge.Builder() .group(TestGroups.EDGE_2) .property("timestamp", 4L) .property("length", 40) .property("count", 5L) .build(); expected.add(expectedEdge); expected.add(expectedEdge1); expected.add(expectedEdge2); final Aggregate aggregate = new Aggregate.Builder() .input(input) .edges(edges) .build(); final Iterable<? extends Element> results = handler.doOperation(aggregate, context, store); final Set<Element> resultsSet = Sets.newHashSet(results); assertEquals(expected, resultsSet); }
@Test public void shouldFailValidationWhenSchemaElementDefinitionsAreNull() { given(store.getSchema()).willReturn(new Schema()); input.add(edge); input.add(edge1); edges.put(TestGroups.EDGE, new AggregatePair()); final Aggregate aggregate = new Aggregate.Builder() .input(input) .edges(edges) .build(); try { final Iterable<? extends Element> results = handler.doOperation(aggregate, context, store); } catch (final OperationException e) { assertTrue(e.getMessage().contains("Edge group: " + TestGroups.EDGE + " does not exist in the schema.")); } }
@Test public void shouldFailValidationWhenElementAggregatorOperationIsNull() { final Schema schema = new Schema.Builder() .edge(TestGroups.EDGE, new SchemaEdgeDefinition.Builder() .aggregator(new ElementAggregator.Builder() .select("count") .execute(null) .build()) .groupBy("timestamp") .build()) .build(); given(store.getSchema()).willReturn(schema); input.add(edge); input.add(edge1); input.add(edge2); edges.put(TestGroups.EDGE, new AggregatePair()); final Aggregate aggregate = new Aggregate.Builder() .input(input) .edges(edges) .build(); try { final Iterable<? extends Element> results = handler.doOperation(aggregate, context, store); fail("Exception expected"); } catch (final OperationException e) { assertTrue(e.getMessage().contains("Schema contains an ElementAggregator with a null function.")); } }
@Test public void shouldFailValidationWhenTypeArgumentOfBinaryOperatorInFunctionIsIncorrect() { final Schema schema = new Schema.Builder() .entity(TestGroups.ENTITY, new SchemaEntityDefinition.Builder() .property(TestPropertyNames.TIMESTAMP, "timestamp.long") .build()) .type("timestamp.long", new TypeDefinition(Long.class)) .build(); given(store.getSchema()).willReturn(schema); input.add(entity); input.add(entity1); input.add(entity2); entities.put(TestGroups.ENTITY, new AggregatePair(new ElementAggregator.Builder() .select(TestPropertyNames.TIMESTAMP) .execute(new Or()) .build())); final Aggregate aggregate = new Aggregate.Builder() .input(input) .entities(entities) .build(); try { final Iterable<? extends Element> results = handler.doOperation(aggregate, context, store); fail("Exception expected"); } catch (final OperationException e) { assertTrue(e.getMessage().contains("Incompatible types.")); } }