nit: Possibly make this explicit with a return as well?

done

It appears matched >= this.perKeyElements.size() should be its own "Got more elements than expected" error case and throw an exception.

done.

nit: I don't think this assert is going to work, it is the case above?

the case above is denied (!...?)

Sorry if my comment wasn't clear, here is what I'm thinking: the case above is mached < this.perKeyElements.size() && !... the assert triggers on !..., so the assert actually triggers on matched >= this.perKeyElements.size().

After the latest update you should never be able to hit this assert, you can leave it as is.

Java assert is, I think, less preferable than a library that generates more elaborate error messages. Also, failures within a user DoFn may be swallowed by a runner. Can the tests be described in terms of a dead-letter output and PAssert?

Should we test types other than Integer?

hmmm ... I could try to parameterize the test with different types -though IDK if the Parameterized JUnit runner integrates well with the ValidatesRunner framework. Do you have a suggestion? Should I try adding a more complex type instead of just ints?

I don't think you'll be able to parameterize this, you'll need different test methods for each type.

I believe runners see these as byte[] today, so the behavior here actually depends on coders, so changes in the coder will change the ordering. (Future relational work will likely make some runners aware of the coder contents as well.) The byte coder is the trivial passthrough type, so Byte might actually be the best for a basic "does the runner produce deterministic ordering" test.

Sorry for the late addition: You might want to test some more extreme values. VarIntCoder is variable sized, so around those size boundaries might be good to test.

I tried that. All good.

"Stage" is not a concept in the Beam model, unless you also define it elsewhere in this PR. (commenting as I go)

renamed this to bundle. WDYT?

"shuffle" is also not a model concept. For example runners can choose to move shuffles around so then the ordered delivery might change. Unless you can demonstrate/require that allowable optimizations preserve the ordered delivery you are looking for.

renamed to bundle and across bundle boundaries. WDYT?

Seems unrelated.

Java assert is, I think, less preferable than a library that generates more elaborate error messages. Also, failures within a user DoFn may be swallowed by a runner. Can the tests be described in terms of a dead-letter output and PAssert?

Although it seems to be mentioned I think it would be good to restate here explicitly, that this holds if and only if the two PTransforms are directly connected - or at least there is no intermediate grouping transform in between. "A downstream PCollection" in the above paragraph might be interpreted as any downstream PCollection, which does not hold.

I was thinking about this comment and it occurs to me that you have to have key-limited parallelism in the producer of course. I don't think this is required anywhere in the model (state/timers require key+window limited parallelism) and also not mentioned here.

What I meant was that if we we have a chain of transforms as follows:

A -> B -> C

if all these transforms are stateful transforms, then there is no guarantee for ordering of elements emitted from A arriving at C. The only exception would be when there is no change in key, because then we can prove that the ordering will be preserved due to transitivity. If key between A and B changes, then there is no guarantee for ordering at C (even if the key changes back to the same as emitted from A).

Agree with both the above points. Should it explicitly say the following?

• Key-ordered deliver is only guaranteed between immediately connected elements.
• To enforce key ordering both producer and consumer require key limited parallelism.

Or are these statements not accurate?

Additionally, should it refer to stages or elements? When building a pipeline you don't necessarily know what stages will result from any fusion that occurs, so would you specify a requirement for key-ordered delivery between elements rather than stages?

Hi all! Sorry about the delay, but I've finally tried to address these comments. Can you please take a look and LMK what you think>

Is there any reference that this is true? It is something i am trying to figure out at the moment.

hm we have not stated this in Dataflow documentation (I'm working on that), but it's true for streaming.

Ok, i guess it makes sense to define the concept here first.

These definitions are slightly different to the glossary ones for fusion and stage. Maybe just have a link in the last bullet point above saying runners might use a strategy called fusion which combines elements into stages?

done