AggregatorUtil { public static CloseableIterable<Element> queryAggregate(final Iterable<? extends Element> elements, final Schema schema, final View view) { if (null == schema) { throw new IllegalArgumentException("Schema is required"); } if (null == view) { throw new IllegalArgumentException("View is required"); } final Collection<String> aggregatedGroups = schema.getAggregatedGroups(); final List<Element> aggregatableElements = new ArrayList<>(); final List<Element> nonAggregatedElements = new ArrayList<>(); for (final Element element : elements) { if (null != element) { if (aggregatedGroups.contains(element.getGroup())) { aggregatableElements.add(element); } else { nonAggregatedElements.add(element); } } } final Iterable<Element> aggregatedElements = Streams.toStream(aggregatableElements) .collect(Collectors.groupingBy(new ToQueryElementKey(schema, view), Collectors.reducing(null, new QueryElementBinaryOperator(schema, view)))) .values(); return new ChainedIterable<>(aggregatedElements, nonAggregatedElements); } private  AggregatorUtil(); static CloseableIterable<Element> ingestAggregate(final Iterable<? extends Element> elements, final Schema schema); static CloseableIterable<Element> queryAggregate(final Iterable<? extends Element> elements, final Schema schema, final View view); static Map<String, Set<String>> getIngestGroupBys(final Schema schema); static Map<String, Set<String>> getQueryGroupBys(final Schema schema, final View view); static Set<String> getIngestGroupBy(final String group, final Schema schema); static Set<String> getQueryGroupBy(final String group, final Schema schema, final View view);  }
@Test public void shouldThrowExceptionWhenQueryAggregatedIfSchemaIsNull() { final Schema schema = null; final View view = new View(); try { AggregatorUtil.queryAggregate(Collections.emptyList(), schema, view); fail("Exception expected"); } catch (final IllegalArgumentException e) { assertTrue(e.getMessage().contains("Schema")); } }
@Test public void shouldThrowExceptionWhenQueryAggregatedIfViewIsNull() { final Schema schema = new Schema(); final View view = null; try { AggregatorUtil.queryAggregate(Collections.emptyList(), schema, view); fail("Exception expected"); } catch (final IllegalArgumentException e) { assertTrue(e.getMessage().contains("View")); } }
@Test public void shouldQueryAggregateElementsWithGroupBy() { final Schema schema = Schema.fromJson(StreamUtil.openStreams(getClass(), "schema-groupby")); final View view = new View.Builder() .entity(TestGroups.ENTITY, new ViewElementDefinition.Builder() .groupBy() .build()) .entity(TestGroups.EDGE, new ViewElementDefinition.Builder() .groupBy() .build()) .build(); final List<Element> elements = Arrays.asList( new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 1) .property("property2", "value1") .property("visibility", "vis1") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 2) .property("property2", "value1") .property("visibility", "vis2") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 2) .property("property2", "value2") .property("visibility", "vis1") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 10) .property("property2", "value2") .property("visibility", "vis2") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex2") .property("count", 20) .property("property2", "value10") .property("visibility", "vis1") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 100) .property("property2", "value1") .property("visibility", "vis1") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 200) .property("property2", "value1") .property("visibility", "vis2") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 1000) .property("property2", "value2") .property("visibility", "vis1") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 2000) .property("property2", "value2") .property("visibility", "vis2") .build() ); final Set<Element> expected = Sets.newHashSet( new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 15) .property("property2", "value1") .property("visibility", "vis1") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex2") .property("count", 20) .property("property2", "value10") .property("visibility", "vis1") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 3300) .property("property2", "value1") .property("visibility", "vis1") .build() ); final CloseableIterable<Element> aggregatedElements = AggregatorUtil.queryAggregate(elements, schema, view); assertElementEquals(expected, aggregatedElements); }
@Test public void shouldQueryAggregateElementsWithGroupByAndViewAggregator() { final Schema schema = Schema.fromJson(StreamUtil.openStreams(getClass(), "schema-groupby")); final View view = new View.Builder() .entity(TestGroups.ENTITY, new ViewElementDefinition.Builder() .groupBy() .aggregator(new ElementAggregator.Builder() .select("count") .execute(new Product()) .build()) .build()) .entity(TestGroups.EDGE, new ViewElementDefinition.Builder() .groupBy() .build()) .build(); final List<Element> elements = Arrays.asList( new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 1) .property("property2", "value1") .property("visibility", "vis1") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 2) .property("property2", "value1") .property("visibility", "vis2") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 2) .property("property2", "value2") .property("visibility", "vis1") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 10) .property("property2", "value2") .property("visibility", "vis2") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex2") .property("count", 20) .property("property2", "value10") .property("visibility", "vis1") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 100) .property("property2", "value1") .property("visibility", "vis1") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 200) .property("property2", "value1") .property("visibility", "vis2") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 1000) .property("property2", "value2") .property("visibility", "vis1") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 2000) .property("property2", "value2") .property("visibility", "vis2") .build() ); final Set<Element> expected = Sets.newHashSet( new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 40) .property("property2", "value1") .property("visibility", "vis1") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex2") .property("count", 20) .property("property2", "value10") .property("visibility", "vis1") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 3300) .property("property2", "value1") .property("visibility", "vis1") .build() ); final CloseableIterable<Element> aggregatedElements = AggregatorUtil.queryAggregate(elements, schema, view); assertElementEquals(expected, aggregatedElements); }
@Test public void shouldQueryAggregateElementsWhenProvidedIterableCanOnlyBeConsumedOnce() { final Schema schema = Schema.fromJson(StreamUtil.openStreams(getClass(), "schema-groupby")); final View view = new View.Builder() .entity(TestGroups.ENTITY, new ViewElementDefinition.Builder() .groupBy() .build()) .entity(TestGroups.EDGE, new ViewElementDefinition.Builder() .groupBy() .build()) .build(); final List<Element> elements = Arrays.asList( new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 1) .build(), new Entity.Builder() .group(TestGroups.NON_AGG_ENTITY) .vertex("vertex1") .property("count", 2) .build() ); final Iterable<Element> onlyConsumingOnceIterable = new Iterable<Element>() { private boolean firstTime = true; @Override public Iterator<Element> iterator() { if (firstTime) { firstTime = false; return elements.iterator(); } throw new NoSuchElementException(); } }; final CloseableIterable<Element> aggregatedElements = AggregatorUtil.queryAggregate(onlyConsumingOnceIterable, schema, view); assertElementEquals(elements, aggregatedElements); }