[Data] Fix Union operator to avoid blocking when preserve order

[owenowenisme]

Description

Make the Union operator not blocking when preserve_order is enabled if _add_input_inner is called with the input in the front.

[gemini-code-assist]

Code Review

This pull request fixes a blocking issue in the Union operator when preserve_order is enabled. The change introduces a streaming approach, flushing data from input operators as they complete, which is a good improvement. My review identifies a critical potential for deadlock because the flushing logic isn't triggered when an input stream finishes without sending a final block. I've also included a couple of suggestions to improve code readability and maintainability by reducing duplication and simplifying expressions. Addressing the critical issue is necessary to prevent hangs in the data processing pipeline.

[alexeykudinkin]

@owenowenisme the problem is that we still we're exhausting 1 input before we move on to the next one, which means we're gonna be accumulating remaining outputs before producing.

Instead, please take a look how we achieve determinism in make_async_gen and apply the same technique here:

• Iterate over inputs in the same order
• Always deque 1 block and never skip an input!
• Once op completes you can start skipping it

[owenowenisme]

@alexeykudinkin
Wait, are you saying that we round-robin the input ops of Union? I think Union should not behave like make_async_gen, because Union's documented behavior is concatenation (ds1.union(ds2) → all of ds1, then all of ds2), while round-robin would produce interleaved output. These are different orderings.