GH-46454: [C++][Dataset][Acero] Preserve order when writting with TeeNode

[gitmodimo]

Rationale for this change

TeeNode needs to sequence batches when implicit order within processed dataset.

What changes are included in this PR?

Conditionally sequence batches when preserve_order=true

Are these changes tested?

I tested it in my use case. No CI tests AFAIK.

Are there any user-facing changes?

Dataset will now be ordered as expected.

• GitHub Issue: [C++][Dataset] Preserve order when writing dataset using TeeNode #46454

[github-actions]

⚠️ GitHub issue #46454 has been automatically assigned in GitHub to PR creator.

[gitmodimo]

@pitrou Can I ping you when PR is ready?
This change is TeeNode equivalent to @EnricoMi WriteNode #44470

[pitrou]

@gitmodimo Well, the expert is really @zanmato1984 on this.

[gitmodimo]

@gitmodimo Well, the expert is really @zanmato1984 on this.

@pitrou is there a way I can request review from @zanmato1984? Github automatically assigned you for this.

[pitrou]

Well, he receives these messages, so he will know about this PR :)

[zanmato1984]

Thanks for working on this. Having TeeNode honor preserve_order seems reasonable: TeeNode performs the dataset write itself, so sequencing further downstream cannot restore the order of data that TeeNode has already written.

The overall direction looks good. I found one correctness issue that should be addressed, plus two non-blocking comments about the test.

[zanmato1984]

Could we add the same input-ordering validation used by SinkNode and ConsumingSinkNode?

With preserve_order=true, TeeNode creates a SerialSequencingQueue even when inputs_[0]->ordering() is unordered. In that case, batches may retain kUnsequencedIndex. SerialSequencingQueue::InsertBatch DCHECKs this in debug builds; in release builds, such batches never match the queue's initial next_index_ == 0, so they can remain queued and prevent the plan from finishing.

TeeNode should reject this configuration in Validate(), with a regression test for unordered input plus preserve_order=true.

[gitmodimo]

I added validation and test for it

[zanmato1984]

Non-blocking: Is this output-order check necessary for testing TeeNode?

The default SinkNode already sequences input with a meaningful ordering, so the collected output should be ordered even if TeeNode forwards batches out of order. As a result, this assertion does not directly exercise TeeNode's preserve_order behavior. The checks against the datasets written by the unordered and ordered TeeNodes below appear to provide the relevant coverage.

The sink is still needed to consume the plan, but the conversion to output_table and its ordering assertion may be unnecessary.

[gitmodimo]

removed

[zanmato1984]

Non-blocking: Could this use the existing test JitterNode with a fixed seed instead of relying on the scalar function's wall-clock delay?

JitterNode is designed to resequence physical batch delivery while preserving batch indices, and is already used by the order-by tests for this purpose. Reusing it here would make the test intent clearer and avoid depending on thread timing and sleeps to produce out-of-order arrival.

[gitmodimo]

Yes this is cleaner. I also migrated write node to use jitter node stimulus.

[zanmato1984]

Could we keep the existing MultiThreadedWritePersistsOrder test using FileSystemDataset::Write(write_options, scanner)? My understanding is that it provides regression coverage for GH-26818, including scan sequencing and implicit ordering across fragments. Replacing it with table_source -> jitter -> write seems to test only the WriteNode path directly. Would it be better to retain the existing integration test and use JitterNode only for the new TeeNode test?

[gitmodimo]

Oh I see it also validates scanner gets the require_sequenced_output.

[zanmato1984]

+1

Thanks for working on this.