@Test void testFilteredMappedJoin() { BiPredicate<DataPoint, DataPoint> filter = (dp1, dp2) -> { assertEquals(dp1, dp2, "Joining unrelated datapoints"); return true; }; DataSet result = INSTANCE.filteredMappedJoin(STRUCTURE, INSTANCE, filter, (a, b) -> a); assertEquals(STRUCTURE, result.getMetadata()); assertEquals(new HashSet<>(Arrays.asList(DATAPOINTS)), result.stream().collect(toSet())); }
@Override public DataSet filteredMappedJoin(DataSetMetadata metadata, DataSet other, BiPredicate<DataPoint,DataPoint> predicate, BinaryOperator<DataPoint> mergeOp) { Set<DataStructureComponent<Identifier, ?, ?>> commonIds = getMetadata().getComponents(Identifier.class); commonIds.retainAll(other.getComponents(Identifier.class)); Map<Map<DataStructureComponent<Identifier, ?, ?>, ScalarValue<?, ?, ?>>, List<DataPoint>> index; try (Stream<DataPoint> stream = other.stream()) { if (commonIds.equals(other.getComponents(Identifier.class))) index = stream.collect(toConcurrentMap(dp -> dp.getValues(commonIds, Identifier.class), Collections::singletonList)); else index = stream.collect(groupingByConcurrent(dp -> dp.getValues(commonIds, Identifier.class))); } return new LightFDataSet<>(metadata, d -> d.stream() .map(dpThis -> { List<DataPoint> otherSubGroup = index.get(dpThis.getValues(commonIds, Identifier.class)); if (otherSubGroup == null) return Stream.<DataPoint>empty(); else return otherSubGroup.stream() .filter(dpOther -> predicate.test(dpThis, dpOther)) .map(dpOther -> mergeOp.apply(dpThis, dpOther)); }).reduce(Stream::concat) .orElse(Stream.empty()), this); }
AbstractDataSet implements DataSet { @Override public DataSet filteredMappedJoin(DataSetMetadata metadata, DataSet other, BiPredicate<DataPoint,DataPoint> predicate, BinaryOperator<DataPoint> mergeOp) { Set<DataStructureComponent<Identifier, ?, ?>> commonIds = getMetadata().getComponents(Identifier.class); commonIds.retainAll(other.getComponents(Identifier.class)); Map<Map<DataStructureComponent<Identifier, ?, ?>, ScalarValue<?, ?, ?>>, List<DataPoint>> index; try (Stream<DataPoint> stream = other.stream()) { if (commonIds.equals(other.getComponents(Identifier.class))) index = stream.collect(toConcurrentMap(dp -> dp.getValues(commonIds, Identifier.class), Collections::singletonList)); else index = stream.collect(groupingByConcurrent(dp -> dp.getValues(commonIds, Identifier.class))); } return new LightFDataSet<>(metadata, d -> d.stream() .map(dpThis -> { List<DataPoint> otherSubGroup = index.get(dpThis.getValues(commonIds, Identifier.class)); if (otherSubGroup == null) return Stream.<DataPoint>empty(); else return otherSubGroup.stream() .filter(dpOther -> predicate.test(dpThis, dpOther)) .map(dpOther -> mergeOp.apply(dpThis, dpOther)); }).reduce(Stream::concat) .orElse(Stream.empty()), this); } }
AbstractDataSet implements DataSet { @Override public DataSet filteredMappedJoin(DataSetMetadata metadata, DataSet other, BiPredicate<DataPoint,DataPoint> predicate, BinaryOperator<DataPoint> mergeOp) { Set<DataStructureComponent<Identifier, ?, ?>> commonIds = getMetadata().getComponents(Identifier.class); commonIds.retainAll(other.getComponents(Identifier.class)); Map<Map<DataStructureComponent<Identifier, ?, ?>, ScalarValue<?, ?, ?>>, List<DataPoint>> index; try (Stream<DataPoint> stream = other.stream()) { if (commonIds.equals(other.getComponents(Identifier.class))) index = stream.collect(toConcurrentMap(dp -> dp.getValues(commonIds, Identifier.class), Collections::singletonList)); else index = stream.collect(groupingByConcurrent(dp -> dp.getValues(commonIds, Identifier.class))); } return new LightFDataSet<>(metadata, d -> d.stream() .map(dpThis -> { List<DataPoint> otherSubGroup = index.get(dpThis.getValues(commonIds, Identifier.class)); if (otherSubGroup == null) return Stream.<DataPoint>empty(); else return otherSubGroup.stream() .filter(dpOther -> predicate.test(dpThis, dpOther)) .map(dpOther -> mergeOp.apply(dpThis, dpOther)); }).reduce(Stream::concat) .orElse(Stream.empty()), this); } protected  AbstractDataSet(DataSetMetadata dataStructure); }
AbstractDataSet implements DataSet { @Override public DataSet filteredMappedJoin(DataSetMetadata metadata, DataSet other, BiPredicate<DataPoint,DataPoint> predicate, BinaryOperator<DataPoint> mergeOp) { Set<DataStructureComponent<Identifier, ?, ?>> commonIds = getMetadata().getComponents(Identifier.class); commonIds.retainAll(other.getComponents(Identifier.class)); Map<Map<DataStructureComponent<Identifier, ?, ?>, ScalarValue<?, ?, ?>>, List<DataPoint>> index; try (Stream<DataPoint> stream = other.stream()) { if (commonIds.equals(other.getComponents(Identifier.class))) index = stream.collect(toConcurrentMap(dp -> dp.getValues(commonIds, Identifier.class), Collections::singletonList)); else index = stream.collect(groupingByConcurrent(dp -> dp.getValues(commonIds, Identifier.class))); } return new LightFDataSet<>(metadata, d -> d.stream() .map(dpThis -> { List<DataPoint> otherSubGroup = index.get(dpThis.getValues(commonIds, Identifier.class)); if (otherSubGroup == null) return Stream.<DataPoint>empty(); else return otherSubGroup.stream() .filter(dpOther -> predicate.test(dpThis, dpOther)) .map(dpOther -> mergeOp.apply(dpThis, dpOther)); }).reduce(Stream::concat) .orElse(Stream.empty()), this); } protected  AbstractDataSet(DataSetMetadata dataStructure); @Override DataSet membership(String componentName); @Override Optional<DataStructureComponent<?, ?, ?>> getComponent(String name); @Override DataSetMetadata getMetadata(); @Override DataSet filteredMappedJoin(DataSetMetadata metadata, DataSet other, BiPredicate<DataPoint,DataPoint> predicate, BinaryOperator<DataPoint> mergeOp); @Override DataSet mapKeepingKeys(DataSetMetadata metadata,
			Function<? super DataPoint, ? extends Map<? extends DataStructureComponent<? extends NonIdentifier, ?, ?>, ? extends ScalarValue<?, ?, ?>>> operator); @Override Stream<T> streamByKeys(Set<DataStructureComponent<Identifier, ?, ?>> keys, 
			Map<DataStructureComponent<Identifier, ?, ?>, ScalarValue<?, ?, ?>> filter,
			Collector<DataPoint, A, TT> groupCollector,
			BiFunction<TT, Map<DataStructureComponent<Identifier, ?, ?>, ScalarValue<?, ?, ?>>, T> finisher); @Override DataSet filter(Predicate<DataPoint> predicate); @Override final Stream<DataPoint> stream(); String toString(); }
AbstractDataSet implements DataSet { @Override public DataSet filteredMappedJoin(DataSetMetadata metadata, DataSet other, BiPredicate<DataPoint,DataPoint> predicate, BinaryOperator<DataPoint> mergeOp) { Set<DataStructureComponent<Identifier, ?, ?>> commonIds = getMetadata().getComponents(Identifier.class); commonIds.retainAll(other.getComponents(Identifier.class)); Map<Map<DataStructureComponent<Identifier, ?, ?>, ScalarValue<?, ?, ?>>, List<DataPoint>> index; try (Stream<DataPoint> stream = other.stream()) { if (commonIds.equals(other.getComponents(Identifier.class))) index = stream.collect(toConcurrentMap(dp -> dp.getValues(commonIds, Identifier.class), Collections::singletonList)); else index = stream.collect(groupingByConcurrent(dp -> dp.getValues(commonIds, Identifier.class))); } return new LightFDataSet<>(metadata, d -> d.stream() .map(dpThis -> { List<DataPoint> otherSubGroup = index.get(dpThis.getValues(commonIds, Identifier.class)); if (otherSubGroup == null) return Stream.<DataPoint>empty(); else return otherSubGroup.stream() .filter(dpOther -> predicate.test(dpThis, dpOther)) .map(dpOther -> mergeOp.apply(dpThis, dpOther)); }).reduce(Stream::concat) .orElse(Stream.empty()), this); } protected  AbstractDataSet(DataSetMetadata dataStructure); @Override DataSet membership(String componentName); @Override Optional<DataStructureComponent<?, ?, ?>> getComponent(String name); @Override DataSetMetadata getMetadata(); @Override DataSet filteredMappedJoin(DataSetMetadata metadata, DataSet other, BiPredicate<DataPoint,DataPoint> predicate, BinaryOperator<DataPoint> mergeOp); @Override DataSet mapKeepingKeys(DataSetMetadata metadata,
			Function<? super DataPoint, ? extends Map<? extends DataStructureComponent<? extends NonIdentifier, ?, ?>, ? extends ScalarValue<?, ?, ?>>> operator); @Override Stream<T> streamByKeys(Set<DataStructureComponent<Identifier, ?, ?>> keys, 
			Map<DataStructureComponent<Identifier, ?, ?>, ScalarValue<?, ?, ?>> filter,
			Collector<DataPoint, A, TT> groupCollector,
			BiFunction<TT, Map<DataStructureComponent<Identifier, ?, ?>, ScalarValue<?, ?, ?>>, T> finisher); @Override DataSet filter(Predicate<DataPoint> predicate); @Override final Stream<DataPoint> stream(); String toString();  }