AggregatorUtil { public static CloseableIterable<Element> ingestAggregate(final Iterable<? extends Element> elements, final Schema schema) { if (null == schema) { throw new IllegalArgumentException("Schema 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 ToIngestElementKey(schema), Collectors.reducing(null, new IngestElementBinaryOperator(schema)))) .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 shouldThrowExceptionWhenIngestAggregatedIfSchemaIsNull() { final Schema schema = null; try { AggregatorUtil.ingestAggregate(Collections.emptyList(), schema); fail("Exception expected"); } catch (final IllegalArgumentException e) { assertNotNull(e.getMessage()); } }
@Test public void shouldIngestAggregateElementsWhenProvidedIterableCanOnlyBeConsumedOnce() { final Schema schema = Schema.fromJson(StreamUtil.openStreams(getClass(), "schema-groupby")); 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.ingestAggregate(onlyConsumingOnceIterable, schema); assertElementEquals(elements, aggregatedElements); }
@Test public void shouldIngestAggregateElementsWithNoGroupBy() { final Schema schema = Schema.fromJson(StreamUtil.openStreams(getClass(), "schema-groupby")); final List<Element> elements = Arrays.asList( new Entity.Builder() .group(TestGroups.NON_AGG_ENTITY) .vertex("vertex1") .property("count", 1) .build(), new Entity.Builder() .group(TestGroups.NON_AGG_ENTITY) .vertex("vertex1") .property("count", 2) .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 1) .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 2) .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex2") .property("count", 10) .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 100) .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 200) .build() ); final Set<Element> expected = Sets.newHashSet( new Entity.Builder() .group(TestGroups.NON_AGG_ENTITY) .vertex("vertex1") .property("count", 1) .build(), new Entity.Builder() .group(TestGroups.NON_AGG_ENTITY) .vertex("vertex1") .property("count", 2) .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 3) .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex2") .property("count", 10) .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 300) .build() ); final CloseableIterable<Element> aggregatedElements = AggregatorUtil.ingestAggregate(elements, schema); assertElementEquals(expected, aggregatedElements); }
@Test public void shouldIngestAggregateElementsWithGroupBy() { final Schema schema = Schema.fromJson(StreamUtil.openStreams(getClass(), "schema-groupby")); 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", 1) .property("property2", "value1") .property("visibility", "vis2") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 2) .property("property2", "value1") .property("visibility", "vis1") .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", 2) .property("property2", "value2") .property("visibility", "vis2") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 10) .property("property2", "value2") .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", 100) .property("property2", "value1") .property("visibility", "vis1") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 100) .property("property2", "value1") .property("visibility", "vis2") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 100) .property("property2", "value1") .property("visibility", "vis2") .build(), new Edge.Builder() .group(TestGroups.EDGE) .source("vertex2") .dest("vertex1") .property("count", 200) .property("property2", "value1") .property("visibility", "vis1") .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", "vis1") .build() ); final Set<Element> expected = Sets.newHashSet( new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 3) .property("property2", "value1") .property("visibility", "vis1") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 1) .property("property2", "value1") .property("visibility", "vis2") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 12) .property("property2", "value2") .property("visibility", "vis1") .build(), new Entity.Builder() .group(TestGroups.ENTITY) .vertex("vertex1") .property("count", 2) .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", 300) .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", 3000) .property("property2", "value2") .property("visibility", "vis1") .build() ); final CloseableIterable<Element> aggregatedElements = AggregatorUtil.ingestAggregate(elements, schema); assertElementEquals(expected, aggregatedElements); }