yjshen commented on code in PR #7179:
URL: https://github.com/apache/arrow-datafusion/pull/7179#discussion_r1282527910
##########
datafusion/core/src/physical_plan/sorts/sort.rs:
##########
@@ -168,7 +288,11 @@ impl ExternalSorter {
!self.spills.is_empty()
}
- /// MergeSort in mem batches as well as spills into total order with
`SortPreservingMergeStream`.
+ /// Returns the final sorted output of all batches inserted via
+ /// [`Self::insert_batch`] as a stream of [`RecordBatch`]es.
+ ///
+ /// This may be an in memory sort/merge if all input fit into memory, or
+ /// a streaming merge from spill files on disk.
Review Comment:
This process could either be an in-memory sort/merge if all the input fits
into memory or a combined streaming merge incorporating both in-memory batches
and data from spill files on disk.
##########
datafusion/core/src/physical_plan/sorts/sort.rs:
##########
@@ -76,27 +76,147 @@ impl ExternalSorterMetrics {
}
}
-/// Sort arbitrary size of data to get a total order (may spill several times
during sorting based on free memory available).
+/// Sorts an arbitrary sized, unsorted, stream of [`RecordBatch`]es to
+/// a total order. Depending on the input size and memory manager
+/// configuration, writes intermediate results to disk ("spills")
+/// using Arrow IPC format.
+///
+/// # Algorithm
///
-/// The basic architecture of the algorithm:
/// 1. get a non-empty new batch from input
-/// 2. check with the memory manager if we could buffer the batch in memory
-/// 2.1 if memory sufficient, then buffer batch in memory, go to 1.
-/// 2.2 if the memory threshold is reached, sort all buffered batches and
spill to file.
-/// buffer the batch in memory, go to 1.
-/// 3. when input is exhausted, merge all in memory batches and spills to get
a total order.
+///
+/// 2. check with the memory manager there is sufficient space to
+/// buffer the batch in memory 2.1 if memory sufficient, buffer
+/// batch in memory, go to 1.
+///
+/// 2.2 if no more memory is available, sort all buffered batches and
+/// spill to file. buffer the next batch in memory, go to 1.
+///
+/// 3. when input is exhausted, merge all in memory batches and spills
+/// to get a total order.
+///
+/// # When data fits in available memory
+///
+/// If there is sufficient memory, data is sorted in memory to produce the
output
+///
+/// ```text
+/// ┌─────┐
+/// │ 2 │
+/// │ 3 │
+/// │ 1 │─ ─ ─ ─ ─ ─ ─ ─ ─ ┐
+/// │ 4 │
+/// │ 2 │ │
+/// └─────┘ ▼
+/// ┌─────┐
+/// │ 1 │ In memory
+/// │ 4 │─ ─ ─ ─ ─ ─▶ sort/merge ─ ─ ─ ─ ─▶ total sorted output
+/// │ 1 │
+/// └─────┘ ▲
+/// ... │
+///
+/// ┌─────┐ │
+/// │ 4 │
+/// │ 3 │─ ─ ─ ─ ─ ─ ─ ─ ─ ┘
+/// └─────┘
+///
+/// in_mem_batches
+///
+/// ```
+///
+/// # When data does not fit in available memory
+///
+/// When memory is exhausted, data is first sorted and written to one
+/// or more spill files on disk:
+///
+/// ```text
+/// ┌─────┐ .─────────────────.
+/// │ 2 │ ( )
+/// │ 3 │ │`─────────────────'│
+/// │ 1 │─ ─ ─ ─ ─ ─ ─ │ ┌────┐ │
+/// │ 4 │ │ │ │ 1 │░ │
+/// │ 2 │ │ │... │░ │
+/// └─────┘ ▼ │ │ 4 │░ ┌ ─ ─ │
+/// ┌─────┐ │ └────┘░ 1 │░ │
+/// │ 1 │ In memory │ ░░░░░░ │ ░░ │
+/// │ 4 │─ ─ ▶ sort/merge ─ ─ ─ ─ ┼ ─ ─ ─ ─ ─▶ ... │░ │
+/// │ 1 │ and write to file │ │ ░░ │
+/// └─────┘ │ 4 │░ │
+/// ... ▲ │ └░─░─░░ │
+/// │ │ ░░░░░░ │
+/// ┌─────┐ │.─────────────────.│
+/// │ 4 │ │ ( )
+/// │ 3 │─ ─ ─ ─ ─ ─ ─ `─────────────────'
+/// └─────┘
+///
+/// in_mem_batches spills
+/// (file on disk in Arrow
+/// IPC format)
+/// ```
+///
+/// Once the input is completely read, the spill files are read and
+/// merged with any in memory batches to produce a single total sorted
+/// output:
+///
+/// ```text
+/// .─────────────────.
+/// ( )
+/// │`─────────────────'│
+/// │ ┌────┐ │
+/// │ │ 1 │░ │
+/// │ │... │─ ─ ─ ─ ─ ─│─ ─ ─ ─ ─ ─
+/// │ │ 4 │░ ┌────┐ │ │
+/// │ └────┘░ │ 1 │░ │ ▼
+/// │ ░░░░░░ │ │░ │
+/// │ │... │─ ─│─ ─ ─ ▶ merge ─ ─ ─▶ total sorted output
+/// │ │ │░ │
+/// │ │ 4 │░ │ ▲
+/// │ └────┘░ │ │
+/// │ ░░░░░░ │
+/// │.─────────────────.│ │
+/// ( )
+/// `─────────────────' │
+/// spills
+/// │
+///
+/// │
+///
+/// ┌─────┐ │
+/// │ 1 │
+/// │ 4 │─ ─ ─ ─ │
+/// └─────┘ │
+/// ... In memory
+/// └ ─ ─ ─▶ sort/merge
+/// ┌─────┐
+/// │ 4 │ ▲
+/// │ 3 │─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ┘
+/// └─────┘
+///
+/// in_mem_batches
+/// ```
struct ExternalSorter {
+ /// schema of the output (and the input)
schema: SchemaRef,
+ /// Potentially unsorted in memory buffer
in_mem_batches: Vec<RecordBatch>,
+ /// if `Self::in_mem_batches` are srted
Review Comment:
```suggestion
/// if `Self::in_mem_batches` are sorted
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
