Reactive: Consider adding OrderBy, OrderByDescending, ThenBy, ThenByDescending

[glopesdev]

I realize the idea of "sorting" is tricky to interpret in the reactive world. However, I would argue that the same holds true for enumerable sequences. Enumerable.OrderBy also needs to buffer the whole sequence before emitting the first value rather than emit values as they are produced, so I don't see why Observable.OrderBy could not do exactly the same. Is there something I'm missing?

As long as you know what to expect, it would help a lot for consistency between both LINQ frameworks and avoid having to convert observables to enumerables back-and-forth for a simple ordering; not to mention having to buffer the entire sequence twice (once to go from observable to enumerable; and another time to run OrderBy).

[Dorus]

And what is wrong with Observable.ToEnumerable().OrderBy?

I would argue that adding OrderBy to Observable directly would be completely pointless, because it gives the fake idea that there is _anything_ reactive about it, while it is not. At least with ToEnumerable you are explicit about that.

[glopesdev]

And what is wrong with Observable.ToEnumerable().OrderBy?

A couple of things actually.

First, it breaks down the ability to write fluent LINQ-style query expressions over observable sequences. OrderBy is a first class citizen fully supported in the query language syntax. Given that one of the tenets of Rx was originally to provide equivalent operators to IEnumerable, it should strike anyone as weird when the compiler complains about orderby field descending. This becomes even worse if you want to stay within the IObservable monad as you then have to do Observable.ToEnumerable().OrderBy.ToObservable().

Second, converting from IObservable to IEnumerable involves buffering all elements in the observable sequence _before_ sending them out. This is required because element production can be asynchronous to element consumption. This is true even if the observable is _cold_. OrderBy also requires buffering all elements in the sequence _before_ sorting them and this will be done _irrespective_ of the first conversion. If we could stay inside the Rx contract, we could dispense with the first buffer entirely.

I would argue that adding OrderBy to Observable directly would be completely pointless, because it gives the fake idea that there is anything reactive about it, while it is not.

I completely disagree.

First, OrderBy (sorting) is commonly understood to be an operator that requires knowledge about the whole sequence, the same as Max, Min, Aggregate, All or Any. It simply returns a sequence, rather than a single element. No one should be confused about the fact that elements can only be sent out _after_ the original sequence has terminated.

Second, even if I grant the possibility of confusion, I would argue that the exact same criticism applies even in the Enumerable world. Even though Enumerable follows a pull-model, it is still essentially a streaming interface where you _naively_ would expect the source sequence to move one element at a time as you iterate. By your reasoning, it should be equally surprising that you call MoveNext on an OrderBy and the whole source is unrolled. Should be especially fun if the sequence is infinite... again, this shows the idiosyncrasy belongs to the OrderBy operator, and not to any distinctions between _pull_ and _push_.

Third, the result of the OrderBy would still _most definitely_ be reactive _w.r.t._ the consumer of the sequence. The consumer doesn't know when OnCompleted will be called in the original sequence. He will get his ordered sequence when that event happens and that is completely asynchronous in the sense that he doesn't need to block waiting for it.

[headinthebox]

Do you have a real-world use case where you want to sort an event stream?

[glopesdev]

Yes, they show up all the time especially whenever you are slicing event streams into windows.

For example, in non-periodic time series analysis: _for every 1-minute window of data, take the 5 largest points (if any)._

Can be solved with Buffer and Enumerable.OrderBy, but apart from the fact I have to unnecessarily buffer the whole sequence for every single window (see above), I often want to keep inside the IObservable monad for compositionality reasons (i.e. I don't want to end up with an IEnumerable).

But I really just pulled that example up randomly, the real answer is: all the cases in which you would normally want to order a data sequence, _whenever that sequence is generated asynchronously_.

I thought the point was simple: sorting is a key operation in any data language. It's usually expensive and non-trivial to implement flexibly, so whenever you can find language-integrated extensions that support it, you feel happy.

But hey, we can always disagree; for some reason someone thought MaxBy and MinBy were important and they're now in the framework, so I thought I would make the case that OrderBy is at least equally deserving.

Feel free to go ahead and close this issue. I'll probably have to implement it for my own framework anyway based on Jon Skeet's amazing Edulinq blog. When that happens I will try and offer up a pull-request for this issue, if it's still up.

[james-world]

I don't want to make this a jab at your example, since I think it's very interesting - but I do want to point out you actually don't need to cache all events in the whole window - you only need to cache at most 5 values in total - which you do need to keep sorted of course - whereas OrderBy would always cache the whole window.

I did an ordering implementation here which has an interesting discussion within the Q&A as well. Note the features added in this implementation to help ward off the potential problems (gap tolerance etc). Another similar one is here. Neither of these are general OrderBy solutions, but they might be of interest.

I do think the point you make about OrderBy being equally evil in the IEnumerable world is thought-provoking - however, I suspect in practice in the IEnumerable world you are far more likely to find these usages optimised at the source (e.g. through IQueryable ala EF etc).

An OrderBy over an entire stream is much less likely to be something you would want to consume reactively I think - much more likely to turn up over short intervals such as the example you cite. In such cases, I am personally happy to treat these as Enumerables as I think the behaviour is more intention-revealing that way and moving between these monads is quite simple.

On the topic of using LINQ comprehensions - I've never liked those for Rx since every application I've ever written using Rx would only be able to express a small fraction of it's queries in that style. As such I never used it since IMHO it's less readable to switch between the styles - I just stick to the fluent API. Actually, I usually end up doing that for IEnumerable stuff too, for much the same reason.

I can't articulate the theory as well as I am sure smarter people like Erik might be able to, but a few years of working with Rx has yet to provide any compelling real world cases for reactive OrderBy for me personally. I _have_ seen (such as in the earlier cited SO question) far more cases where OrderBy is sought for the wrong reasons. I think excluding it probably helps more people fall into the pit of success than it hinders those with genuine use cases.

[glopesdev]

Thanks for the structured feedback. I agree with basically everything you said and would keep only a couple of extra points.

1) You're correct that it's possible to hand-engineer an N-element sorted cache for an observable sequence that keeps the last N largest (or smallest) elements sorted. It would be great if this kind of buffer operator was included in the framework (LargestBy/SmallestBy with a capacity overload?)

Interestingly, such a cache sounds very useful also for the Enumerable world, where many times I've seen people sort huge lists just to get the 3 largest elements. So while I agree with you, again I fail to see the distinction between _pull_ and _push_; I've seen much suffering with OrderBy on both sides, would be nice to have more solutions, not less...

2) The transition from Observable to Enumerable may be simple from a syntactic perspective, but is definitely not so cheap. The fact that it is not known when the consumer will ask for the data forces us to have a synchronized queue in the middle which can become annoying if you start needing to transition between the two worlds a lot. This has definitely started happening for me and most of the issues could be mitigated if there was simply a full implementation of all of LINQ on the Rx side (another example is Reverse).

I have to admit I am a bit bemused that I was unable to convince you of the importance of having a more complete implementation, especially because for me this all started because I wanted to use Rx more, not less...

Maybe I will reconsider moving everything back to Enumerable, which will mean effectively splitting my growing code base of useful combinators into both worlds, which will be a pain to maintain. I had become convinced you could do pretty much everything with IObservable (and much faster by the way for truly streaming scenarios) and that has definitely turned out true except for these kinds of gaps...

[Dorus]

1) LargestBy is easy enough to make:

C# public static IObservable<TSource> LargestBy<TSource>(this IObservable<TSource> source, Func<TSource, int> keySelector, int size) { return source.Aggregate(new List<KeyValuePair<int, TSource>>(), (list, el) => { list.Add(new KeyValuePair<int, TSource>(keySelector(el), el)); list.Sort((x, y) => x.Key.CompareTo(y.Key)); if (list.Count > size) { list.RemoveAt(0); } return list; }) .Select(list => list.Select(el => el.Value).ToObservable()) .Merge(); }

You might even want to replace ToObservable() with GetEnumerator(), or just return the List.

And this does make sense, other than OrderBy, this is just a variant of Aggregate. Notice this has a constant space requirement other than a naive .OrderBy().Take(size) implementation that grows with window size.

(This is probably still not very optimal, you could avoid adding elements when they are smaller then everything in the list, and who know, maybe removing from the end of the list is faster. And it reordered the elements).

And you can write exactly the same extension method for IEnumerable.

2) Nothing wrong with switching back to Enumerables when it suits, as soon as you have to pass on a complete list of values, you might as well do so with IEnumerable. While you should avoid gathering all values in a Observable, there is no reason to avoid Enumrables once you've done so.

It is unclear to me what kind of magic you expect from a native Reverse that it wont have to wait for OnComplete() and buffer all values until then.

[glopesdev]

1) I like the LargestBy idea, it would be a nice compromise. Probably you don't want to use SortedList though, because of its lack of support for duplicate entries. It feels like this combinator should be able to handle different values with the same key, similar to MaxBy. I'm not sure if there is a sorted collection built into .NET that would easily allow for this other than just doing Array.Sort all the time...

2) Reverse was just another example of a situation where you have to pay double the cost in space and time if you go from Observable to Enumerable rather than doing it directly in the reactive world. I don't expect any kind of magic, I just don't want to pay the cost of going from Observable to Enumerable in situations where I could clearly avoid it.

Anyway, I get the point that you guys are not convinced, it's fine. I've gone through all my arguments (consistency, performance, utility), and if they are not enough, I don't want to force it. As I said, for my own peace of mind I'll try and push for an implementation anyway, it's a good learning exercise and will help me to familiarize with the Rx code base. I'll follow all of the design guidelines of current framework operators and submit a pull request later along with unit tests. Then you can decide whether to reject it or not with some code in hand.

Should I consider opening a feature request issue for LargestBy/SmallestBy? Sounds like it would be an extremely useful set of operators to have and it doesn't look like a trivial thing to get right with the current building blocks.

[Dorus]

Woops you are right, I'm not overly familiar with SortedList. I've fixed that.

And i want to say, you give good feedback, and by no means I'm here to accept/reject any ideas, just sharing my thoughts. I would certainly like to see a feature request issue for LargestBy/SmallestBy.

[quinmars]

Interestingly, the OrderBy operator could be implement in a way that OrderBy(...).Take(5) would have a _O(n)_ time complexity:

1. The operator buffers all incoming elements (_O(n)_)
2. The operator builds a binary heap (_O(n)_)
3. The operator extract the minimum element and forward it to the observer (_O(log n)_)

Step 3 is repeat as long as there are elements and the operator is not disposed.

It'd probably be fun to implement it and I kinda like the idea that you already observe the elements while it's sorting. On the other hand I'm not sure if the operator is useful enough to be added to the mainline Rx.NET.

[glopesdev]

If you have to repeat the removal operation n times, wouldn't that make the sorting operation O(n log(n))? That wouldn't be necessarily bad, mind you, since most in-memory sorting algorithms are O(n log(n)) anyway.

More fundamentally regarding this issue, I never understood why everyone seems to think that implementing OrderBy is not useful for Rx. The main argument seems to be that there is no way to implement sorting on a data stream. However, doesn't this reasoning apply exactly the same to IEnumerable? You can have infinite enumerable sequences, sorting blows up there exactly the same way. You can have sorted IQueryable sequences optimized at the source, and exactly the same could happen in the reactive world using IQbservable.

For example, I could legitimately submit a query to an online database and expect an IObservable instead of an IEnumerable. The sorting would be done at the source, and the reactive component here is simply that the database will stream the elements to me as soon as they are ready, such that I don't have to rely on polling an enumerable sequence.

Finally, for small subsets of reactive streams (e.g. windows), reactive sorting absolutely makes sense. If you are slicing the original stream reactively, then the sorted output of each sliced stream is emitted reactively, am I missing something?

OrderBy is built into LINQ for a reason, and it makes exactly the same sense in the reactive world as it does in the enumerable world.

In the end, my conclusion has been that there must be some kind of confusion in the way I have been explaining myself. I've been so many times through this argument and I cannot understand what I could be fundamentally missing that makes people think these two cases have nothing to do with each other.

[quinmars]

Of course it'd become _O(n log n )_ if you consume all elements, that's the theoretical lower bound of general sorting. But if you only take 5 elements it'd be _O(n)_.

I haven't said that I'm against that operator, just that I'm unsure if it is useful enough. I think there are some plausible arguments to include it, the parity to IEnumerable and IQueryable, to name one. I wonder why the initial Rx team didn't include it, although they stress in every talk, that I have seen so far, the parity between IObservable and IEnumerable.

Maybe we can collect a pros/cons list.

Pros:

• Parity to IEnumerable
• More support for LINQ, i.e. orderby, thenby, etc.
• it's obvious what it does (it's not of the type OnXxxWithYyyLatestMapBy)
• it can be very useful for IQbservable

Cons:

• not needed very often
• there are work arounds: .ToEnumerable().OrderBy().ToObservable()

Seeing that list I tend to support an addition. Sure it isn't needed very often, esspecially in pure event processing, on the other hand it does not feel to bloat the API since everybody knows what OrderBy does.

[glopesdev]

@quinmars i think listing the pros/cons more succinctly definitely helps, thanks.

Just one comment regarding your last con: .ToEnumerable().OrderBy().ToObservable() is not such an interesting workaround, since as I mentioned you will be buffering the data twice.

ToEnumerable buffers all the elements of the observable sequence and then OrderBy buffers all the elements again for ordering. By adding a reactive OrderBy you actually remove one of the intermediate buffers entirely.

[quinmars]

@glopesdev I wouldn't care much about that supposed extra buffering. The OrderBy buffers also all results of the selector. The fastest you can get is probably .ToList().SelectMany(l => { l.Sort(); return l;})

Anyway I implemented just four fun all for operators:
https://github.com/quinmars/reactive-extensions/commit/9ca5358217958c0bf5efc5bb3bc96601c2894ed6

I haven't done any benchmarking so far, but I guess it's slower than the Enumerable implementation if you sort the full sequence.

[akarnokd]

This sounds like an uncommon operator for Rx.NET proper and perhaps could live in a 3rd party library. Otherwise closing due to inactivity.

[glopesdev]

I disagree that this is an uncommon operator, but all clarifications, with examples, were presented above.

I understand most people in this discussion are not used to dealing with event-triggered time series analysis where it shows up very frequently. However, that's why you just have to rely on the mathematics, which clearly shows this is an equivalence gap in Rx.NET and until its closed, its not possible to claim parity between IObservable and IEnumerable.

I feel that by closing this issue, we are just sweeping it under the rug and keeping people from thinking about it.

[shtse8]

Just discover this discussion and stand on the support side as there are also GroupBy (which waits for the completion) and GroupByUntil (which waits for another observable) implemented and able to be used now. So why don't we add OrderBy and OrderByUntil as well?

btw, it's easy to implement, so I think anyone who needs it can implement by himself.
```C#

    public static IObservable<T> OrderBy<T, TKey>(this IObservable<T> source, Func<T, TKey> keySelector)
    {
        return source
            .ToArray()
            .SelectMany(x => x.OrderBy(keySelector));
    }
    public static IObservable<T> OrderByUntil<T, TKey, TDuration>(this IObservable<T> source, Func<T, TKey> keySelector, IObservable<TDuration> durationSelector)
    {
        return source
            .Buffer(durationSelector)
            .SelectMany(x => x.OrderBy(keySelector));
    }

```