External sorting not working for (maybe only for string columns??)

[alamb]

Describe the bug

Filing a ticket based on a conversation in discord: https://discord.com/channels/885562378132000778/1166447479609376850/1275728622224932959

Basically, I expect that when properly configured, DataFusion would be able to sort data that doesn't fit in RAM, but instead it results in an error like

failed to optimize table: Parquet { source: General("Z-order failed while scanning data: ResourcesExhausted(\"Failed to allocate additional 2348322032 bytes for ExternalSorter[0] with 0 bytes already allocated for this reservation - 1501294808 bytes remain available for the total pool\")") }

To Reproduce

Here is a reproducer: rust_playground.tar.gz

tar xf rust_playground.tar.gz
cd rust_playground/
cargo run

The code looks like this

    // how much data to sort
    let row_limit = 10 * 1000;
    let mem_limit = 10 * 1024 * 1024;
    let print_results = false;

    let pool = FairSpillPool::new(mem_limit);
    let config = RuntimeConfig::new()
        .with_memory_pool(Arc::new(pool))
        .with_disk_manager(DiskManagerConfig::new());

    let runtime_env = RuntimeEnv::new(config)?;
    let builder = SessionStateBuilder::new().with_runtime_env(Arc::new(runtime_env));

    let ctx = SessionContext::new_with_state(builder.build());

    let generator = AccessLogGenerator::new()
        .with_row_limit(row_limit)
        .with_max_batch_size(100); // 100 rows per batch

    // create a plan that simply sorts on the hostname
    let df = ctx
        .read_batches(generator)?
        .sort(vec![col("host").sort(true, true)])?;

    // execute the plan (it should succeed)
    let results: Vec<RecordBatch> = df.collect().await?;

Expected behavior

I expect the query to succeed (by spilling data to disk, etc)

Additional context

@westonpace notes #10073 may be related

Here is some of the commentary from discord:

I see four sites where an allocation failure doesn't lead to spilling:

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

Line 303 in a50aeef

self.reservation.try_grow(size)?

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

Line 434 in a50aeef

self.reservation.try_resize(size)?;

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

Line 513 in a50aeef

self.reservation.try_resize(batch.get_array_memory_size())?;

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

Line 72 in a50aeef

self.reservation.try_grow(batch.get_array_memory_size())?;

My main suspect is the fourth which is used by SortPreservingMergeStream when appending a batch, during the streaming merge of in-memory batches and spill files.

Guessed wrong. It's the second call site from the list above. Here's the relevant part of the backtrace:

  0x559ea91a9cef - core::result::Result<T,E>::unwrap::h6b8b66539b551ac9
                               at /build/rust/tmp/build/rustc-1.80.0-src/library/core/src/result.rs:1102:23
  13:     0x559ea91a9cef - datafusion_execution::memory_pool::MemoryReservation::try_grow::h6d435672ae31c4ca
                               at /home/claudio/Code/github.com/apache/datafusion/datafusion/execution/src/memory_pool/mod.rs:270:12
  14:     0x559ea91a9cef - datafusion_execution::memory_pool::MemoryReservation::try_resize::h80744f5d7b3aa70f
                               at /home/claudio/Code/github.com/apache/datafusion/datafusion/execution/src/memory_pool/mod.rs:252:34
  15:     0x559ea8b95ba9 - datafusion_physical_plan::sorts::sort::ExternalSorter::in_mem_sort::{{closure}}::h4686d81a505f0439
                               at /home/claudio/Code/github.com/apache/datafusion/datafusion/physical-plan/src/sorts/sort.rs:434:9
  16:     0x559ea8b6fda0 - datafusion_physical_plan::sorts::sort::ExternalSorter::insert_batch::{{closure}}::hdd00ee915381b03a
                               at /home/claudio/Code/github.com/apache/datafusion/datafusion/physical-plan/src/sorts/sort.rs:289:32
  17:     0x559ea8b6fda0 - <datafusion_physical_plan::sorts::sort::SortExec as datafusion_physical_plan::execution_plan::ExecutionPlan>::execute::{{closure}}::h256a01c17fe43713
                               at /home/claudio/Code/github.com/apache/datafusion/datafusion/physical-plan/src/sorts/sort.rs:910:52

I got this same backtrace in three worker threads at once:

thread 'tokio-runtime-worker' panicked at /home/claudio/Code/github.com/apache/datafusion/datafusion/execution/src/memory_pool/mod.rs:270:37:
called `Result::unwrap()` on an `Err` value: ResourcesExhausted("Failed to allocate additional 811511472 bytes for ExternalSorter[6] with 0 bytes already allocated for this reservation - 612881930 bytes remain available for the total pool")
stack backtrace:
thread 'tokio-runtime-worker' panicked at /home/claudio/Code/github.com/apache/datafusion/datafusion/execution/src/memory_pool/mod.rs:270:37:
called `Result::unwrap()` on an `Err` value: ResourcesExhausted("Failed to allocate additional 767287128 bytes for ExternalSorter[0] with 0 bytes already allocated for this reservation - 597992699 bytes remain available for the total pool")
thread 'tokio-runtime-worker' panicked at /home/claudio/Code/github.com/apache/datafusion/datafusion/execution/src/memory_pool/mod.rs:270:37:
called `Result::unwrap()` on an `Err` value: ResourcesExhausted("Failed to allocate additional 734693152 bytes for ExternalSorter[3] with 0 bytes already allocated for this reservation - 627555485 bytes remain available for the total pool")

Memory usage steadily grows to 1GiB until the crash but the pool size is 20 GiB

I've been able to avoid the allocation failures during sort with spill-to-disk by replacing try_grow with grow when we can't fall back to spilling. Specifically, I made that change in these places (all of which triggered allocation failures during earlier attempts):

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

Line 434 in a50aeef

self.reservation.try_resize(size)?;

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

Line 572 in a50aeef

self.merge_reservation.try_resize(size)?;

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

Line 72 in a50aeef

self.reservation.try_grow(batch.get_array_memory_size())?;

The first two happen during in-memory sorting, for the spillable and unspillable reservations, respectively. The second one happens during the merge phase.

Memory slightly overshot the 20GiB pool size but stayed close initially. Eventually it grew to 30GiB (during the merge phase, maybe?).

I couldn't get the sort to complete because the pod froze when filesystem usage grew to 1TB. That seems a bit much for sorting a 100GiB partition, I'll be looking into that next.

[cjolowicz]

This is probably unrelated, but there's something that puzzles me about the fair spill pool logic. The try_grow function checks that the consumer isn't allocating more than its fair share. So it needs to determine how much memory the consumer has already allocated. But it only looks at the reservation that's passed, not at all reservations of that consumer.

Consider a single spillable consumer that allocates the entire pool. It shouldn't be able to grow the resulting reservation. But if it splits the reservation into two, it can now grow it - even when the other reservation remains allocated.

Here's a failing test showing this:
https://github.com/apache/datafusion/pull/12170/files

---- memory_pool::pool::tests::test_fair_with_split_reservation stdout ----
thread 'memory_pool::pool::tests::test_fair_with_split_reservation' panicked at datafusion/execution/src/memory_pool/pool.rs:417:34:
called `Result::unwrap_err()` on an `Ok` value: ()

Let me know if I'm misunderstanding something here.

[cjolowicz]

I couldn't get the sort to complete because the pod froze when filesystem usage grew to 1TB.

I could reproduce the pod freezing up at 1TB disk usage. This happened within a minute of spawning 500 threads. All threads were blocked on a mutex (futex_wait) except one in epoll with an inode fd. Maybe deadlock at the start of the merge phase?

1TB seems a bit much for sorting a 100GiB partition.

It actually makes sense because spill files use Apache IPC format without compression, while the partition uses Parquet files with Snappy compression.

[16pierre]

Also hitting this bug, is there any update on a fix ?

[alamb]

Also hitting this bug, is there any update on a fix ?

I don't know of anyone working explicitly working to make external sorting better. Some recent work maybe would make it better

• Fix record batch memory size double counting #13377
• Replace record_batch.get_array_memory_size() in spilling operators #13430

But realistically I think there is significant additional room for improvement. Any help welcome

[Omega359]

I don't believe that adding an option for compression in DiskManager would be all that difficult - the IPCWriter and reader already has support for LZ4. It would of course incur the cost of compression/decompression but would likely result in 50+% space savings on disk. It's a nice to have though - I think of it more as a band-aid than a solution.

[2010YOUY01]

For the given reproducer, I got the error

Error: ResourcesExhausted("Failed to allocate additional 117568 bytes for ExternalSorter[0] with 0 bytes already allocated for this reservation - 0 bytes remain available for the total pool

Configure datafusion.execution.sort_spill_reservation_bytes to 1MB can let it run successfully. ( I don't know whether the parquet related error message is caused by the same issue)

// Reproducer: place in datafusion/core/tests/memory_limit/mod.rs
#[tokio::test]
async fn test_sort_with_memory_limit() -> Result<()> {
    // initialize logging to see DataFusion's internal logging
    let _ = env_logger::try_init();

    // how much data to sort
    let row_limit = 10 * 1000;
    let mem_limit = 10 * 1024 * 1024; // 10 MB
    let sort_spill_reservation_bytes = 1024 * 1024; // 1 MB

    let generator = AccessLogGenerator::new()
        .with_row_limit(row_limit)
        .with_max_batch_size(100); // 100 rows per batch

    let pool = Arc::new(GreedyMemoryPool::new(mem_limit));
    let runtime = RuntimeEnvBuilder::new()
        .with_memory_pool(pool)
        .with_disk_manager(DiskManagerConfig::new())
        .build()?;
    let session_config = SessionConfig::new()
        .with_sort_spill_reservation_bytes(sort_spill_reservation_bytes);
    let state = SessionStateBuilder::new()
        .with_config(session_config)
        .with_runtime_env(Arc::new(runtime))
        .build();

    let ctx = SessionContext::new_with_state(state);

    // create a plan that simply sorts on the hostname
    let df = ctx
        .read_batches(generator)?
        .sort(vec![col("host").sort(true, true)])?;

    // execute the plan (it should succeed)
    let _results: Vec<RecordBatch> = df.collect().await?;

    Ok(())
}

Reasons:

• How sort_spill_reservation_bytes works: For example there is a sort query with memory limit 10MB, and sort_spill_reservation_bytes is 1M, it will accumulate batches in memory until it reaches 9M, then do in-memory sort. The sort will do in-place sort for all individual batches, and finally do a sort-preserving merge to get the final one sorted run. 1MB is reserved for internal data structures of the merging phase.
• The query in reproducer only has 10 MB memory budget, and all 10MB is reserved for later merge usage, then the execution can fail

Thoughts:
Note I'm not 100% sure about the implementation detail, if that's the case, I think we can get rid of this configuration option, and figure out sort_spill_reservation_bytes according to the available memory budget, to prevent similar failures.

[westonpace]

@2010YOUY01 Your solution does not work for me. I did play around with sort_spill_reservation_bytes when I originally encountered this but couldn't find a way to get it to work. Here is a reproducer that uses a 100MB memory budget and fails for all values of sort_spill_reservation_bytes between 1 and 100 MB. It might be that I am configuring things or using something improperly however.