fix: Assertion fail in external sort

[2010YOUY01]

Which issue does this PR close?

related to #15372

Rationale for this change

One external sort query will panic due to an assertion failure

Reproducer

Compile and run datafusion-cli commit 46f4024

> set datafusion.execution.sort_spill_reservation_bytes = 0;
> select * from generate_series(1, 10000000) as t1(v1) order by v1;

thread 'tokio-runtime-worker' panicked at datafusion/physical-plan/src/sorts/sort.rs:406:9:
assertion failed: self.in_mem_batches_sorted
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

For background knowledge for external sort, check the doc in

datafusion/datafusion/physical-plan/src/sorts/sort.rs

Line 80 in 46f4024

/// Sorts an arbitrary sized, unsorted, stream of [`RecordBatch`]es to

Now a single field in_mem_batches is used to represent buffered data in different stage: during different time, it can be interpreted as either unordered input batch, or globally sorted batch, a state flag self.in_mem_batches_sorted is used to represent its sorted-ness and do extra sanity checks.
This approach has poor understandability and is also error-prone, and actually triggered one edge case assertion failure.
Note the execution logic is not wrong, the issue is this flag is set too late.

To also address #15372, this PR did a small refactor to make the related logic more understandable:

What changes are included in this PR?

refactor to let ExternalSorter.in_mem_batches to only hold unsorted input batches (instead of representing both sorted and unsorted batches during different execution stage, as it's doing before the PR)

Are these changes tested?

A unit test was added to cover the edge case that previously triggered an assertion failure.

Are there any user-facing changes?

No.

[Copilot]

Pull Request Overview

This PR fixes an assertion failure in the external sort by refactoring the state management of in-memory batches. It removes the in_mem_batches_sorted flag, uses a new locally scoped globally_sorted_batches vector to hold sorted batches, and updates the associated spill logic accordingly.

Reviewed Changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
datafusion/physical-plan/src/sorts/sort.rs	Refactors ExternalSorter to remove the redundant in_mem_batches_sorted flag and improve spill logic using a new globally_sorted_batches vector
datafusion/core/tests/memory_limit/mod.rs	Adds a new test to verify external sort behavior with a zero merge reservation

Comments suppressed due to low confidence (1)

datafusion/physical-plan/src/sorts/sort.rs:476

• [nitpick] Now that this function operates on a passed-in globally_sorted_batches vector rather than an instance field, a name like 'organize_sorted_batches' might more clearly convey its purpose.

fn organize_stringview_arrays(

[comphead]

perhaps we can get rid of assert and return Err? if the code regressed the users can face the app crash

[2010YOUY01]

Good point, addressed in 1f4eabb

[alamb]

Removing this field is an improvement 👍 thanks @2010YOUY01 and @comphead

Given there are still three fields whose relationship is tied, I wonder if it would improve the code if we encoded that relationship in an actual rust enum -- for example

enum SortState {
  /// All data is in memory
  Memory {
    batches: Vec<RecordBatch>,
  }, 
  /// intermediate data is spilling to disk
  Spilling {
    batches: Vec<RecordBatch>,
    in_progress_spill_file: InProgressSpillFile,
  }
...
}

🤔

[2010YOUY01]

👍🏼 will try it.

[alamb]

thank you for the name here -- much better