apache / apache/datafusion

External sort failing with non-spillable operators as input (RepartitionExec)

Open
#17,334 31 comments 1 reaction 0 assignees View on GitHub
bug
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### Describe the bug

I'm trying to sort some input (union of filtered Parquet files) with fixed parallelism, in order to do so I'm manually fiddling with a round-robin `RepartitionExec` operator before the sort. I reckon this is slightly hacky, they may be alternatives, the possibility of unions of large numbers of inputs makes partitioning trickier though.

When doing so, I get memory allocation failures on the `RepartitionExec` operator (I suspect this could happen with other non-spillable operators used as input of the Sort nodes though):

```
ResourcesExhausted(\"Additional allocation failed with top memory consumers (across reservations) as:
ExternalSorter[0] consumed 145849856 bytes,
ExternalSorterMerge[0] consumed 50000000 bytes,
RepartitionExec[0] consumed 2951744 bytes.

Error: Failed to allocate additional 2951744 bytes for RepartitionExec[0] with 2951744 bytes already allocated for this reservation - 1198400 bytes remain available for the total pool
```

If my understanding is correct, this feels expected given the current sort implementation: we trigger spill only after saturating the memory pool and getting allocation failure on `try_grow`:
https://github.com/apache/datafusion/blob/033fc9b0a124082c6b4b6c68a8445c16bdaa51a1/datafusion/physical-plan/src/sorts/sort.rs#L787

Which means that once the global memory pool is filled up with sort allocations, any operator used as input of the sort can fail if it uses non-spillable memory allocations.

### To Reproduce

My current repro setup cannot be shared as a snippet because it uses a bunch of production codepaths.

Roughly what I'm doing is:
- write a Parquet file with a couple millions rows (binary columns, 16 bytes each, UUIDs)
- read the Parquet file and run repartition + sort

```
// Memory allocator = FairSpillPool

let physical_plan = open_my_parquet_file(...);

// Note: I get failures with input partitions = 1 and sort_parallelism = 1
let partitioning = Partitioning::RoundRobinBatch(sort_parallelism as usize);
let repartitioned_exec = Arc::new(RepartitionExec::try_new(physical_plan, partitioning))
SortExec::new(sort_expression.clone(), repartitioned_exec)
```

Note we're currently on Datafusion `46.0.1` (will upgrade soon to pick up recent sort improvements)

### Expected behavior

_No response_

### Additional context

Main work-arounds/hacks that come to mind:
- register a fake spillable allocation to force `FairSpillPool` to spill earlier than before memory is full
- implement custom MemoryPool to special-case how different operators are tracked

Ideally I'd prefer a solution where we can precisely control how much memory is allocated to the sort compared to the rest of the operators, especially given the subtle memory allocations we faced during the sort merge phase - the solution "fake spillable allocation" is a bit coarse-grain with that regards.

Contributor guide

Open the contributing guide

Research direction

Start with datafusion/physical-plan/src/sorts/sort.rs around the linked try_grow and spill logic, then inspect how RepartitionExec and FairSpillPool account for memory. Reproduce the failure with the described Parquet input, round-robin repartitioning, and SortExec; done should mean non-spillable input operators no longer fail when the sort reaches its memory limit.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
backend, performance
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Needs clarification
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.