How to zip 2 sequences based on property (zip, join) (2024)

I don't like either of the posted answers. Both of them are variations on the same theme: Keep all members of both sequences in memory indefinitely and iterate over the entire right sequence whenever a new left element comes in, and incrementally check the left key whenever a new right element comes in. Both answers you O(L + R) memory indefinitely and are O(R * L) time complexity (where L and R are the sizes of the left and right sequences).

If we were dealing with collections (or enumerables), that would be a sufficient answer. But we're not: We're dealing with observables, and the answers should acknowledge that. There could be large time gaps in between the actual use case. The question is posed as a test case stemming from an enumerable. If it were simply an enumerable, the right answer is to convert back to Enumerable and use Linq's Join. If there's a possibility of a long running process with time gaps, the answer should acknowledge that you may want to only join on elements that have happened within some period of time, releasing memory in the process.

This satisfies the test answer, while allowing for a time box:

var sequence = tasks.Join(people, _ => Observable.Timer(TimeSpan.FromSeconds(.5)), _ => Observable.Timer(TimeSpan.FromSeconds(.5)), (t, p) => (task: t, person: p) ) .Where(t => t.person.Timestamp == t.task.Timestamp) .Select(t => new Assignment { Task = t.task.Value, Person = t.person.Value });

This creates a window for each element of .5 seconds, meaning a left element and right element will match if they pop out within .5 seconds of each other. After .5 seconds, each element is released from memory. If, for whatever reason, you didn't want to release from memory and hold all objects in memory indefinitely, this would suffice:

var sequence = tasks.Join(people, _ => Observable.Never<Unit>(), _ => Observable.Never<Unit>(), (t, p) => (task: t, person: p) ) .Where(t => t.person.Timestamp == t.task.Timestamp) .Select(t => new Assignment { Task = t.task.Value, Person = t.person.Value });

Here is a custom Join operator that could be used in order to solve this problem. It is based on the Merge, GroupByUntil and SelectMany operators:

/// <summary>/// Correlates the elements of two sequences based on matching keys. Results are/// produced for all combinations of correlated elements that have an overlapping/// duration./// </summary>public static IObservable<TResult> Join<TLeft, TRight, TKey, TResult>( this IObservable<TLeft> left, IObservable<TRight> right, Func<TLeft, TKey> leftKeySelector, Func<TRight, TKey> rightKeySelector, Func<TLeft, TRight, TResult> resultSelector, TimeSpan? keyDuration = null, IEqualityComparer<TKey> keyComparer = null){ // Arguments validation omitted keyComparer ??= EqualityComparer<TKey>.Default; var groupDuration = keyDuration.HasValue ? Observable.Timer(keyDuration.Value) : Observable.Never<long>(); return left .Select(x => (x, (TRight)default, Type: 1, Key: leftKeySelector(x))) .Merge(right.Select(x => ((TLeft)default, x, Type: 2, Key: rightKeySelector(x)))) .GroupByUntil(e => e.Key, _ => groupDuration, keyComparer) .Select(g => ( g.Where(e => e.Type == 1).Select(e => e.Item1), g.Where(e => e.Type == 2).Select(e => e.Item2).Replay().AutoConnect(0) )) .SelectMany(g => g.Item1.SelectMany(_ => g.Item2, resultSelector));}

Usage example:

IObservable<Assignment> sequence = tasks .Join(people, t => t.Timestamp, p => p.Timestamp, (t, p) => new Assignment { Task = t.Value, Person = p.Value });

It should be noted that this problem cannot be solved with guaranteed 100% correctness without buffering all the elements that the two source sequences produce. Obviously this is not going to scale well in case the sequences contain infinite elements.

In case sacrificing the absolute correctness in favor of scalability is acceptable, the optional keyDuration argument can be used to configure the maximum duration that a stored key (and its associated elements) can be preserved in memory. An expired key can potentially be reborn, in case new elements having this key are produced by the left or right sequences.

The above implementation performs reasonably well with sequences containing large number of elements. Joining two same-sized sequences, each having 100,000 elements, takes ~8 seconds in my PC.

The observable Zip operator works just the same as the enumerable version. You didn't use that in the first test so it's not like to be the operator you need here.

What you need is simply the SelectMany operator.

Try this query:

var sequence = from t in tasks from p in people where t.Timestamp == p.Timestamp select new Assignment { Task = t.Value, Person = p.Value };

That works with your test.