api/odsadapter/src/main/java/org/eclipse/mdm/api/odsadapter/search/MergedSearchQuery.java - mdmbl/org.eclipse.mdm - Git at Google

 /********************************************************************************
  * Copyright (c) 2015-2020 Contributors to the Eclipse Foundation
  *
  * See the NOTICE file(s) distributed with this work for additional
  * information regarding copyright ownership.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0 which is available at
  * http://www.eclipse.org/legal/epl-2.0.
  *
  * SPDX-License-Identifier: EPL-2.0
  *
  ********************************************************************************/

 package org.eclipse.mdm.api.odsadapter.search;

 import static java.util.stream.Collectors.groupingBy;
 import static java.util.stream.Collectors.reducing;

 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;

 import org.eclipse.mdm.api.base.adapter.Attribute;
 import org.eclipse.mdm.api.base.adapter.EntityType;
 import org.eclipse.mdm.api.base.adapter.Relation;
 import org.eclipse.mdm.api.base.model.ContextType;
 import org.eclipse.mdm.api.base.model.Value;
 import org.eclipse.mdm.api.base.query.Aggregation;
 import org.eclipse.mdm.api.base.query.DataAccessException;
 import org.eclipse.mdm.api.base.query.Filter;
 import org.eclipse.mdm.api.base.query.Record;
 import org.eclipse.mdm.api.base.query.Result;
 import org.eclipse.mdm.api.base.search.ContextState;
 import org.eclipse.mdm.api.base.search.SearchQuery;
 import org.eclipse.mdm.api.base.search.Searchable;

 /**
  * Merges 2 distinct search queries, where one queries context data as ordered
  * and the other context as measured.
  *
  * @since 1.0.0
  * @author Viktor Stoehr, Gigatronik Ingolstadt GmbH
  */
 final class MergedSearchQuery implements SearchQuery {

 	// ======================================================================
 	// Instance variables
 	// ======================================================================

 	private final EntityType entityType;

 	private final BaseEntitySearchQuery byResult;
 	private final BaseEntitySearchQuery byOrder;

 	// ======================================================================
 	// Constructors
 	// ======================================================================

 	MergedSearchQuery(EntityType entityType, Function<ContextState, BaseEntitySearchQuery> factory) {
 		this.entityType = entityType;

 		byResult = factory.apply(ContextState.MEASURED);
 		byOrder = factory.apply(ContextState.ORDERED);
 	}

 	// ======================================================================
 	// Public methods
 	// ======================================================================

 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public List<EntityType> listEntityTypes() {
 		return byOrder.listEntityTypes();
 	}

 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public Searchable getSearchableRoot() {
 		return byOrder.getSearchableRoot();
 	}

 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public List<Value> getFilterValues(Attribute attribute, Filter filter) throws DataAccessException {
 		List<Value> orderValues = byOrder.getFilterValues(attribute, filter);
 		List<Value> resultValues = byResult.getFilterValues(attribute, filter);

 		return Stream.concat(orderValues.stream(), resultValues.stream())
 				// group by value and merge values
 				.collect(groupingBy(Value::extract, reducing((v1, v2) -> v1)))
 				// collect merged results
 				.values().stream().filter(Optional::isPresent).map(Optional::get).collect(Collectors.toList());
 	}

 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public List<Result> getFilterResults(List<Attribute> attributes, Filter filter, ContextState contextState)
 			throws DataAccessException {
 		ContextState filterContext = filter.getContext() != null ? filter.getContext() : contextState;
 		List<Result> results;

 		Set<EntityType> entityTypesFromAttributes = getTypesFromAttributes(attributes);
 		if (entityTypesFromAttributes.size() > 1) {
 			throw new DataAccessException("Cannot handle multiple types here");
 		}

 		if (ContextState.ORDERED.equals(filterContext)) {
 			results = byOrder.getFilterResults(attributes, filter, contextState);
 			swapValues(attributes, results, Aggregation.DISTINCT);
 		} else {
 			results = byResult.getFilterResults(attributes, filter, contextState);
 		}

 		return results;

 	}

 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public List<Result> fetchComplete(List<EntityType> entityTypes, Filter filter) throws DataAccessException {
 		return mergeResults(byOrder.fetchComplete(entityTypes, filter), byResult.fetchComplete(entityTypes, filter));
 	}

 	/**
 	 * {@inheritDoc}
 	 */
 	@Override
 	public List<Result> fetch(List<Attribute> attributes, Filter filter) throws DataAccessException {
 		ContextState contextState = filter.getContext();
 		if (ContextState.ORDERED.equals(contextState)) {
 			return fetch(contextState, byOrder, byResult, attributes, filter, results -> createIdFilter(results, entityType, Aggregation.NONE, contextState));
 		} else if (ContextState.MEASURED.equals(contextState)) {
 			return fetch(contextState, byResult, byOrder, attributes, filter, results -> createIdFilter(results, entityType, Aggregation.NONE, contextState));
 		}
 		return fetch(contextState, byResult, byOrder, attributes, filter, list -> filter);
 	}

 	// ======================================================================
 	// Private methods
 	// ======================================================================

 	/**
 	 * Will fetch the values based on the given primary and secondary
 	 * {@link BaseEntitySearchQuery} with the provided {@link Filter}. Will swap
 	 * ordered/measured values if necessary.
 	 *
 	 * @param contextState   the {@link ContextState} were the values came from.
 	 * @param primary        the primary {@link BaseEntitySearchQuery}
 	 * @param secondary      the secondary {@link BaseEntitySearchQuery}
 	 * @param attributes     selected attributes as list of {@link Attribute}
 	 * @param filter         the current {@link Filter}
 	 * @param filterFunction this function provides the filter for the secondary
 	 *                       query
 	 *
 	 * @return list of merged {@link Result}
 	 */
 	private List<Result> fetch(ContextState contextState, BaseEntitySearchQuery primary,
 			BaseEntitySearchQuery secondary, List<Attribute> attributes, Filter filter,
 			Function<List<Result>, Filter> filterFunction) {
 		List<Result> primaryResults = primary.fetch(attributes, filter);
 		if (primaryResults.isEmpty()) {
 			// return early as secondaryResults do not matter, if no primaryResults are
 			// found.
 			return Collections.emptyList();
 		}
 		Filter additionalFilter = filterFunction.apply(primaryResults);
 		List<Result> secondaryResults = secondary.fetch(attributes, additionalFilter);
 		if (ContextState.ORDERED.equals(contextState)) {
 			swapValues(attributes, primaryResults, Aggregation.NONE);
 		} else if (ContextState.MEASURED.equals(contextState)) {
 			swapValues(attributes, secondaryResults, Aggregation.NONE);
 		}
 		return mergeResults(primaryResults, secondaryResults);
 	}

 	/**
 	 * Swaps the ordered/measured values inside the {@link Value} object related to
 	 * the given attributes.
 	 *
 	 * @param attributes selected attributes as list of {@link Attribute}
 	 * @param results    the list of results where the values should be swapped
 	 */
 	private void swapValues(List<Attribute> attributes, List<Result> results, Aggregation aggregation) {
 		Set<String> relevantNames = attributes.stream()
 				.filter(this::filterContextAttribute)
 				.map(Attribute::getName)
 				.map(name -> getAggregatedName(name, aggregation))
 				.collect(Collectors.toSet());

 		results.stream()
 				.flatMap(Result::stream)
 				.map(Record::getValues)
 				.map(Map::values)
 				.flatMap(Collection::stream)
 				.filter(value -> relevantNames.contains(value.getName()))
 				.forEach(Value::swapContext);
 	}

 	/**
 	 * Check whether an attribute which belong to entities from context.
 	 * e.g entities created as "UnitUnderTestPart", which then have a parent relation to "UnitUnderTest"
 	 *
 	 * @param attribute the attribute to check
 	 * @return <code>true<</code> if the attribute belongs to a context, <code>false</code> otherwise
 	 */
 	private boolean filterContextAttribute(Attribute attribute) {
 		return attribute.getEntityType().getParentRelations().stream()
 				.map(Relation::getName)
 				.anyMatch(name -> ContextType.UNITUNDERTEST.typeName().equals(name) ||
 								ContextType.TESTEQUIPMENT.typeName().equals(name) ||
 								ContextType.TESTSEQUENCE.typeName().equals(name));
 	}

 	/**
 	 * Creates a {@link Filter} based on the IDs of the given results.
 	 *
 	 * @param results the results where the IDs should be extracted from
 	 *
 	 * @return a new filter with the IDs
 	 */
 	private Filter createIdFilter(List<Result> results, EntityType currentEntityType, Aggregation aggregation, ContextState contextState) {
 		Filter idFilter = Filter.or();
 		results.stream()
 				.flatMap(Result::stream)
 				.filter(record -> hasID(record, currentEntityType, aggregation))
 				.collect(Collectors.groupingBy(Record::getEntityType,
 						Collectors.mapping(record -> getID(record, currentEntityType, aggregation, contextState), Collectors.toList())))
 				.forEach(idFilter::ids);
 		return idFilter;
 	}

 	/**
 	 * Checks if a given {@link Record} contains a valid ID
 	 *
 	 * @param record the record which should be checked
 	 *
 	 * @return <code>true</code> if the record has a valid ID, <code>false</code>
 	 *         otherwise
 	 */
 	private boolean hasID(Record record, EntityType entityType, Aggregation aggregation) {
 		if (record != null) {
 			try {
 				String idKey =  getAggregatedName(entityType.getIDAttribute().getName(), aggregation);
 				Value idValue = record.getValues().get(idKey);
 				return idValue != null;
 			} catch (IllegalStateException exception) {
 				return false;
 			}
 		} else {
 			return false;
 		}
 	}

 	/**
 	 * Returns the ID of a {@ink Record} and takes {@link EntityType}, {@link Aggregation} and {@link ContextState}
 	 * into account.
 	 *
 	 * @param record the record to get the ID from
 	 * @param entityType the related entity type
 	 * @param aggregation the aggregation which was used to create the record
 	 * @param contextState the context state
 	 *
 	 * @return the ID
 	 */
 	private String getID(Record record, EntityType entityType, Aggregation aggregation, ContextState contextState) {
 		if (record != null) {
 			String idKey =  getAggregatedName(entityType.getIDAttribute().getName(), aggregation);
 			Value idValue = record.getValues().get(idKey);
 			if (idValue == null) {
 				throw new IllegalStateException("ID attribute was not selected.");
 			}
 			return idValue.extract(contextState);
 		} else {
 			return null;
 		}
 	}

 	/**
 	 * Returns the name and takes the given {@link Aggregation} into account.
 	 *
 	 * @param name the name to use the aggregation with
 	 * @param aggregation the aggregation
 	 *
 	 * @return the updated name
 	 */
 	private String getAggregatedName(String name, Aggregation aggregation) {
 		String key;
 		if (Aggregation.NONE == aggregation) {
 			key = name;
 		} else {
 			key = String.format("%s(%s)", aggregation.name(), name);
 		}
 		return key;
 	}

 	/**
 	 * Merges given {@link Result}s to one using the root entity type of this search
 	 * query.
 	 *
 	 * @param results1 The first {@code Result}.
 	 * @param results2 The second {@code Result}.
 	 * @return The merged {@link Result} is returned.
 	 */
 	private List<Result> mergeResults(List<Result> results1, List<Result> results2) {
 		return mergeResults(entityType, Aggregation.NONE, results1, results2);
 	}

 	private List<Result> mergeResults(EntityType mergeType, Aggregation aggregation, List<Result> results1, List<Result> results2) {
 		return Stream.concat(results1.stream(), results2.stream())
 				// group by instance ID and merge grouped results
 				.collect(groupingBy(r -> getMergeId(mergeType, r.getRecord(mergeType), aggregation), reducing(Result::merge)))
 				// collect merged results
 				.values().stream().filter(Optional::isPresent).map(Optional::get).collect(Collectors.toList());
 	}

 	/**
 	 * Returns the ID of the record, if ID of ContextState.MEASURED is not set,
 	 * returns ID of ContextState.ORDERED.
 	 *
 	 * @param rec {@link Record} which ID is requested.
 	 * @return Returns the ID of the record, if ID of ContextState.MEASURED is not
 	 *         set, returns ID of ContextState.ORDERED.
 	 */
 	private String getMergeId(EntityType idType, Record rec, Aggregation aggregation) {

 		String idKey;
 		if (Aggregation.NONE == aggregation) {
 			idKey = idType.getIDAttribute().getName();
 		} else {
 			idKey = String.format("%s(%s)", aggregation.name(), idType.getIDAttribute().getName());
 		}

 		Value idValue = rec.getValues().get(idKey);
 		String id =  idValue.extract();

 		if (id == null) {
 			// MEASURED (default) ContextState is null, so we use the ID stored in ORDERED
 			return rec.getValues().get(idKey).extract(ContextState.ORDERED);
 		} else {
 			return id;
 		}

 	}

 	/**
 	 * Returns a set of {@link EntityType} used in the given attributes.
 	 *
 	 * @param attributes the attributes to collect the entity types from
 	 *
 	 * @return a set ot entity types
 	 */
 	private Set<EntityType> getTypesFromAttributes(List<Attribute> attributes) {
 		return attributes.stream()
 				.map(Attribute::getEntityType)
 				.collect(Collectors.toSet());
 	}
 }
	/********************************************************************************
	* Copyright (c) 2015-2020 Contributors to the Eclipse Foundation
	*
	* See the NOTICE file(s) distributed with this work for additional
	* information regarding copyright ownership.
	*
	* This program and the accompanying materials are made available under the
	* terms of the Eclipse Public License v. 2.0 which is available at
	* http://www.eclipse.org/legal/epl-2.0.
	*
	* SPDX-License-Identifier: EPL-2.0
	*
	********************************************************************************/

	package org.eclipse.mdm.api.odsadapter.search;

	import static java.util.stream.Collectors.groupingBy;
	import static java.util.stream.Collectors.reducing;

	import java.util.Collection;
	import java.util.Collections;
	import java.util.List;
	import java.util.Map;
	import java.util.Optional;
	import java.util.Set;
	import java.util.function.Function;
	import java.util.stream.Collectors;
	import java.util.stream.Stream;

	import org.eclipse.mdm.api.base.adapter.Attribute;
	import org.eclipse.mdm.api.base.adapter.EntityType;
	import org.eclipse.mdm.api.base.adapter.Relation;
	import org.eclipse.mdm.api.base.model.ContextType;
	import org.eclipse.mdm.api.base.model.Value;
	import org.eclipse.mdm.api.base.query.Aggregation;
	import org.eclipse.mdm.api.base.query.DataAccessException;
	import org.eclipse.mdm.api.base.query.Filter;
	import org.eclipse.mdm.api.base.query.Record;
	import org.eclipse.mdm.api.base.query.Result;
	import org.eclipse.mdm.api.base.search.ContextState;
	import org.eclipse.mdm.api.base.search.SearchQuery;
	import org.eclipse.mdm.api.base.search.Searchable;

	/**
	* Merges 2 distinct search queries, where one queries context data as ordered
	* and the other context as measured.
	*
	* @since 1.0.0
	* @author Viktor Stoehr, Gigatronik Ingolstadt GmbH
	*/
	final class MergedSearchQuery implements SearchQuery {

	// ======================================================================
	// Instance variables
	// ======================================================================

	private final EntityType entityType;

	private final BaseEntitySearchQuery byResult;
	private final BaseEntitySearchQuery byOrder;

	// ======================================================================
	// Constructors
	// ======================================================================

	MergedSearchQuery(EntityType entityType, Function<ContextState, BaseEntitySearchQuery> factory) {
	this.entityType = entityType;

	byResult = factory.apply(ContextState.MEASURED);
	byOrder = factory.apply(ContextState.ORDERED);
	}

	// ======================================================================
	// Public methods
	// ======================================================================

	/**
	* {@inheritDoc}
	*/
	@Override
	public List<EntityType> listEntityTypes() {
	return byOrder.listEntityTypes();
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public Searchable getSearchableRoot() {
	return byOrder.getSearchableRoot();
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public List<Value> getFilterValues(Attribute attribute, Filter filter) throws DataAccessException {
	List<Value> orderValues = byOrder.getFilterValues(attribute, filter);
	List<Value> resultValues = byResult.getFilterValues(attribute, filter);

	return Stream.concat(orderValues.stream(), resultValues.stream())
	// group by value and merge values
	.collect(groupingBy(Value::extract, reducing((v1, v2) -> v1)))
	// collect merged results
	.values().stream().filter(Optional::isPresent).map(Optional::get).collect(Collectors.toList());
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public List<Result> getFilterResults(List<Attribute> attributes, Filter filter, ContextState contextState)
	throws DataAccessException {
	ContextState filterContext = filter.getContext() != null ? filter.getContext() : contextState;
	List<Result> results;

	Set<EntityType> entityTypesFromAttributes = getTypesFromAttributes(attributes);
	if (entityTypesFromAttributes.size() > 1) {
	throw new DataAccessException("Cannot handle multiple types here");
	}

	if (ContextState.ORDERED.equals(filterContext)) {
	results = byOrder.getFilterResults(attributes, filter, contextState);
	swapValues(attributes, results, Aggregation.DISTINCT);
	} else {
	results = byResult.getFilterResults(attributes, filter, contextState);
	}

	return results;

	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public List<Result> fetchComplete(List<EntityType> entityTypes, Filter filter) throws DataAccessException {
	return mergeResults(byOrder.fetchComplete(entityTypes, filter), byResult.fetchComplete(entityTypes, filter));
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public List<Result> fetch(List<Attribute> attributes, Filter filter) throws DataAccessException {
	ContextState contextState = filter.getContext();
	if (ContextState.ORDERED.equals(contextState)) {
	return fetch(contextState, byOrder, byResult, attributes, filter, results -> createIdFilter(results, entityType, Aggregation.NONE, contextState));
	} else if (ContextState.MEASURED.equals(contextState)) {
	return fetch(contextState, byResult, byOrder, attributes, filter, results -> createIdFilter(results, entityType, Aggregation.NONE, contextState));
	}
	return fetch(contextState, byResult, byOrder, attributes, filter, list -> filter);
	}

	// ======================================================================
	// Private methods
	// ======================================================================

	/**
	* Will fetch the values based on the given primary and secondary
	* {@link BaseEntitySearchQuery} with the provided {@link Filter}. Will swap
	* ordered/measured values if necessary.
	*
	* @param contextState the {@link ContextState} were the values came from.
	* @param primary the primary {@link BaseEntitySearchQuery}
	* @param secondary the secondary {@link BaseEntitySearchQuery}
	* @param attributes selected attributes as list of {@link Attribute}
	* @param filter the current {@link Filter}
	* @param filterFunction this function provides the filter for the secondary
	* query
	*
	* @return list of merged {@link Result}
	*/
	private List<Result> fetch(ContextState contextState, BaseEntitySearchQuery primary,
	BaseEntitySearchQuery secondary, List<Attribute> attributes, Filter filter,
	Function<List<Result>, Filter> filterFunction) {
	List<Result> primaryResults = primary.fetch(attributes, filter);
	if (primaryResults.isEmpty()) {
	// return early as secondaryResults do not matter, if no primaryResults are
	// found.
	return Collections.emptyList();
	}
	Filter additionalFilter = filterFunction.apply(primaryResults);
	List<Result> secondaryResults = secondary.fetch(attributes, additionalFilter);
	if (ContextState.ORDERED.equals(contextState)) {
	swapValues(attributes, primaryResults, Aggregation.NONE);
	} else if (ContextState.MEASURED.equals(contextState)) {
	swapValues(attributes, secondaryResults, Aggregation.NONE);
	}
	return mergeResults(primaryResults, secondaryResults);
	}

	/**
	* Swaps the ordered/measured values inside the {@link Value} object related to
	* the given attributes.
	*
	* @param attributes selected attributes as list of {@link Attribute}
	* @param results the list of results where the values should be swapped
	*/
	private void swapValues(List<Attribute> attributes, List<Result> results, Aggregation aggregation) {
	Set<String> relevantNames = attributes.stream()
	.filter(this::filterContextAttribute)
	.map(Attribute::getName)
	.map(name -> getAggregatedName(name, aggregation))
	.collect(Collectors.toSet());

	results.stream()
	.flatMap(Result::stream)
	.map(Record::getValues)
	.map(Map::values)
	.flatMap(Collection::stream)
	.filter(value -> relevantNames.contains(value.getName()))
	.forEach(Value::swapContext);
	}

	/**
	* Check whether an attribute which belong to entities from context.
	* e.g entities created as "UnitUnderTestPart", which then have a parent relation to "UnitUnderTest"
	*
	* @param attribute the attribute to check
	* @return <code>true<</code> if the attribute belongs to a context, <code>false</code> otherwise
	*/
	private boolean filterContextAttribute(Attribute attribute) {
	return attribute.getEntityType().getParentRelations().stream()
	.map(Relation::getName)
	.anyMatch(name -> ContextType.UNITUNDERTEST.typeName().equals(name) \|\|
	ContextType.TESTEQUIPMENT.typeName().equals(name) \|\|
	ContextType.TESTSEQUENCE.typeName().equals(name));
	}

	/**
	* Creates a {@link Filter} based on the IDs of the given results.
	*
	* @param results the results where the IDs should be extracted from
	*
	* @return a new filter with the IDs
	*/
	private Filter createIdFilter(List<Result> results, EntityType currentEntityType, Aggregation aggregation, ContextState contextState) {
	Filter idFilter = Filter.or();
	results.stream()
	.flatMap(Result::stream)
	.filter(record -> hasID(record, currentEntityType, aggregation))
	.collect(Collectors.groupingBy(Record::getEntityType,
	Collectors.mapping(record -> getID(record, currentEntityType, aggregation, contextState), Collectors.toList())))
	.forEach(idFilter::ids);
	return idFilter;
	}

	/**
	* Checks if a given {@link Record} contains a valid ID
	*
	* @param record the record which should be checked
	*
	* @return <code>true</code> if the record has a valid ID, <code>false</code>
	* otherwise
	*/
	private boolean hasID(Record record, EntityType entityType, Aggregation aggregation) {
	if (record != null) {
	try {
	String idKey = getAggregatedName(entityType.getIDAttribute().getName(), aggregation);
	Value idValue = record.getValues().get(idKey);
	return idValue != null;
	} catch (IllegalStateException exception) {
	return false;
	}
	} else {
	return false;
	}
	}

	/**
	* Returns the ID of a {@ink Record} and takes {@link EntityType}, {@link Aggregation} and {@link ContextState}
	* into account.
	*
	* @param record the record to get the ID from
	* @param entityType the related entity type
	* @param aggregation the aggregation which was used to create the record
	* @param contextState the context state
	*
	* @return the ID
	*/
	private String getID(Record record, EntityType entityType, Aggregation aggregation, ContextState contextState) {
	if (record != null) {
	String idKey = getAggregatedName(entityType.getIDAttribute().getName(), aggregation);
	Value idValue = record.getValues().get(idKey);
	if (idValue == null) {
	throw new IllegalStateException("ID attribute was not selected.");
	}
	return idValue.extract(contextState);
	} else {
	return null;
	}
	}

	/**
	* Returns the name and takes the given {@link Aggregation} into account.
	*
	* @param name the name to use the aggregation with
	* @param aggregation the aggregation
	*
	* @return the updated name
	*/
	private String getAggregatedName(String name, Aggregation aggregation) {
	String key;
	if (Aggregation.NONE == aggregation) {
	key = name;
	} else {
	key = String.format("%s(%s)", aggregation.name(), name);
	}
	return key;
	}

	/**
	* Merges given {@link Result}s to one using the root entity type of this search
	* query.
	*
	* @param results1 The first {@code Result}.
	* @param results2 The second {@code Result}.
	* @return The merged {@link Result} is returned.
	*/
	private List<Result> mergeResults(List<Result> results1, List<Result> results2) {
	return mergeResults(entityType, Aggregation.NONE, results1, results2);
	}

	private List<Result> mergeResults(EntityType mergeType, Aggregation aggregation, List<Result> results1, List<Result> results2) {
	return Stream.concat(results1.stream(), results2.stream())
	// group by instance ID and merge grouped results
	.collect(groupingBy(r -> getMergeId(mergeType, r.getRecord(mergeType), aggregation), reducing(Result::merge)))
	// collect merged results
	.values().stream().filter(Optional::isPresent).map(Optional::get).collect(Collectors.toList());
	}

	/**
	* Returns the ID of the record, if ID of ContextState.MEASURED is not set,
	* returns ID of ContextState.ORDERED.
	*
	* @param rec {@link Record} which ID is requested.
	* @return Returns the ID of the record, if ID of ContextState.MEASURED is not
	* set, returns ID of ContextState.ORDERED.
	*/
	private String getMergeId(EntityType idType, Record rec, Aggregation aggregation) {

	String idKey;
	if (Aggregation.NONE == aggregation) {
	idKey = idType.getIDAttribute().getName();
	} else {
	idKey = String.format("%s(%s)", aggregation.name(), idType.getIDAttribute().getName());
	}

	Value idValue = rec.getValues().get(idKey);
	String id = idValue.extract();

	if (id == null) {
	// MEASURED (default) ContextState is null, so we use the ID stored in ORDERED
	return rec.getValues().get(idKey).extract(ContextState.ORDERED);
	} else {
	return id;
	}

	}

	/**
	* Returns a set of {@link EntityType} used in the given attributes.
	*
	* @param attributes the attributes to collect the entity types from
	*
	* @return a set ot entity types
	*/
	private Set<EntityType> getTypesFromAttributes(List<Attribute> attributes) {
	return attributes.stream()
	.map(Attribute::getEntityType)
	.collect(Collectors.toSet());
	}
	}