Join two "ordered" streams (streamsorter)

[th0ger]

In continuation of the "ordered streams" workaround discussed in #2395 (by setting readBatchSize:1 and scale.max: 1).

I have two ordered source streams that I want to combine into a single ordered stream using a join vertex. Which vertex type could support me writing a UDF that sorts the output events?

[vigith]

is there an ID to join by? if yes, try session-window with a timeout?

[th0ger]

@vigith I've read it briefly and overall looks good. I'll dive into the details later.
Can you include an example udf (Python SDK) server+handler that implements my original use case?
That will better highlight the split between inbuild/udf, and may also act as unit test.

[vigith]

Yes, we will add an example. Need to think through a bit more before we can implement it in the Numaflow core

[th0ger]

Some comments for the current phrasing of specification for #2440

• "strict but eventual N to N mapping." Is this statement accurate, if we decide that late events are dropped? Dropping seems most consistent with how source and reducers work. (In principle we one could also emit late events out of order.)
• "Non-Requirement: Dropping on reordering could be supported." I think this has to be a requirement for late events.
• "the Datum should be kept intact." I think it will be more accurate to state "a Datum must be returned", since this specifies the interface.
  It's another discussion whether Datum is immutable and, if so, whether it's mutable by core or user.
  I'm recalling that Datum includes keys, value, event_time, watermark and headers. I guess the outbound Datum can have a different watermark than inbound(?), so technically it's not "intact".
  Whether keys, value, headers should be mutable, I'm not sure. For event_time see below.
• "Reorder can be on any user-defined attribute." Besides datum.event_time, what does this mean? Parsed attributes coming from datum.value? Maybe reordering by a sequence numbers is a nice feature, for example if there are duplicates in event_time (caused by low time resolution), then a sequence number could serve as a secondary sort key. Could we require the need for multiple sort keys? (Think SQL ORDER BY event_time ASC, seqno ASC.) If this "user-defined attribute" is a datum.value.foo_timestamp, should foo_timestamp then be set as the outbound event_time?
• We must specify keyed stream behaviour. I think the obvious approach is keys streams are reordered independently. (Alternative is to reorder the global stream.)

[vigith]

I think there is an ambiguity in the word "Drop" (ctx: Dropping on reordering could be supported). What I meant here is that user could do "msg.Drop()" if they want to (kind of a filter) not an explicit drop due to lateness, etc.

"strict but eventual N to N mapping." Is this statement accurate, if we decide that late events are dropped? Dropping seems most consistent with how source and reducers work. (In principle we one could also emit late events out of order.)

yes, that is totally fine. lateness is assigned at source. sorter will not drop late data unless the UDF does "msg.Drop()".

"Non-Requirement: Dropping on reordering could be supported." I think this has to be a requirement for late events.

same as above.

"the Datum should be kept intact." I think it will be more accurate to state "a Datum must be returned", since this specifies the interface.
It's another discussion whether Datum is immutable and, if so, whether it's mutable by core or user.
I'm recalling that Datum includes keys, value, event_time, watermark and headers. I guess the outbound Datum can have a different watermark than inbound(?), so technically it's not "intact".
Whether keys, value, headers should be mutable, I'm not sure. For event_time see below.

I do not want users to mutate the watermark and event_time; if they want, the could do at source using user-defined-transformer. We can do a follow-up later if this doesn't work for some reason.

"Reorder can be on any user-defined attribute." Besides datum.event_time, what does this mean? Parsed attributes coming from datum.value? Maybe reordering by a sequence numbers is a nice feature, for example if there are duplicates in event_time (caused by low time resolution), then a sequence number could serve as a secondary sort key. Could we require the need for multiple sort keys? (Think SQL ORDER BY event_time ASC, seqno ASC.) If this "user-defined attribute" is a datum.value.foo_timestamp, should foo_timestamp then be set as the outbound event_time?

I was thinking perhaps they could do based on some attribute in "datum.Value()". Sorting on User-defined attribute might get removed if "correctness" is being violated.

We must specify keyed stream behaviour. I think the obvious approach is keys streams are reordered independently. (Alternative is to reorder the global stream.)

Solution should work for keyed streams too but ordering will be limited to keys, the stream will not be globally ordered. This is because if we add keys, then there will be an udf invocation per keys and they might get executed in out-of-order fashion. Generalization of this can be solved via #1677.

[th0ger]

I'm good with this.

[th0ger]

There is an additional requirement we need to discuss for #2440: Correct handling of "late streams".

By "late stream" I mean the scenario where we have two input sources that are individually ordered (or almost ordered, say, max 1s out of order). But the latencies of the two source streams are significantly different. Say stream A has latency of 10s and stream B can sometimes have allowed latency of up to 60s. So stream B events are allowed to be late of stream A events by 50s.

My concerns is, that when joined in a vertex, stream A will progress the watermark to an extent that all B events become late and are dropped.
The concern actually applies equally to reducer vertices and the "accumulator/reorder" vertex that we're spec'ing. So let's first conclude whether a joined-reducer vertex can handle late streams correctly?

With above example values, if we set maxDelay: 1s on pipeline level, will the reducer get stream B events in its windows? Or what is the optimal maxDelay.

This should of course generalize to joins with more than 2 inputs.
The mockup code #2427 (reply in thread) does not handle this issue.

[vigith]

(source A) ---+
               \
                +-- (Sorter S) ...
               /
(source B) ---+

The watermark at Sorter S will be min(WM(A), WM(B)). The watermark will progress only when the slower vertex moves.

NOTE: We will not drop late data. Only in a reduce vertex, the late data is ignored because the windows have been closed.

[th0ger]

So late data will also be ignored in the Sorter, similar to reducer?
(Otherwise you could never progress with output order guaranteed.)

[vigith]

yes, late will will not be sorted. Perhaps we can expose IsLate() in the Datum so user-defined sorter can also make a decision?

[th0ger]

I think that's an odd default, but if it's configurable, that's okay.

It would be equivalent to a reducer that would receive late data opens a duplicate window to handle it.

You could also use an yaml flag droplate:bool, that would be very user friendly.

[vigith]

we can make it drop by default