Out of memory when sorting

[andygrove]

Describe the bug
Original bug was filed against Python bindings: apache/datafusion-python#157

Describe the bug
try a sort and export a parquet file using Colab generate an Out of memory error

To Reproduce

!curl -L 'https://drive.google.com/uc?export=download&id=18gv0Yd_a-Zc7CSolol8qeYVAAzSthnSN&confirm=t' > lineitem.parquet
from datafusion import SessionContext
ctx = SessionContext()
ctx.register_parquet('lineitem', 'lineitem.parquet')
df = ctx.sql("select * from lineitem order by l_shipdate")
df.write_parquet("lineitem_Datafusion.parquet")

Expected behavior
I expected to use only the available memory

here is the link comparing the same using Polars and DuckDB
https://colab.research.google.com/drive/1pfAPpIG7jpvGB_aHj-PXX66vRaRT0xlj#scrollTo=O8-lyg1y6RT2

[comphead]

@andygrove any idea how this can be reproduced?
DF test is below

#[tokio::test]
async fn test_huge_sort() {
    let ctx = SessionContext::new();
    ctx.register_parquet(
        "lineitem",
        "/Users/a/lineitem.parquet",
        ParquetReadOptions::default(),
    )
    .await
    .unwrap();
    let sql = "select * from lineitem order by l_shipdate";
    let dataframe = ctx.sql(sql).await.unwrap();
    dataframe.write_parquet("/Users/a/lineitem_sorted.parquet", None).await;
}

Any idea how to limit mem for the process? I have only idea to bring up a VM with some limited memory or setting it up with ulimit

[tustvold]

If running on Linux you could use cgroups to artificially limit the memory available to the process.

FWIW I think this relates to the default configuration of the RuntimeEnv, which has no limit, if you configure the session with a limiting MemoryPool it should enforce a limit. If you additional configure a DiskManager it should spill

[comphead]

Thanks @tustvold , I ran the test with configured mem max set up and spill enabled.

    #[tokio::test]
    async fn test_huge_sort() -> Result<()> {
        let runtime_config = crate::execution::runtime_env::RuntimeConfig::new()
            .with_memory_pool(Arc::new(crate::execution::memory_pool::GreedyMemoryPool::new(1024*1024*1024)))
            .with_disk_manager(crate::execution::disk_manager::DiskManagerConfig::new_specified(vec!["/Users/a/spill/".into()]));
        let runtime = Arc::new(crate::execution::runtime_env::RuntimeEnv::new(runtime_config).unwrap());
        let ctx = SessionContext::with_config_rt(SessionConfig::new(), runtime);
        ctx.register_parquet(
            "lineitem",
            "/Users/a/lineitem.parquet",
            ParquetReadOptions::default(),
        )
        .await
        .unwrap();
        let sql = "select * from lineitem order by l_shipdate";
        let dataframe = ctx.sql(sql).await.unwrap();
        dataframe.show_limit(10).await?;
        //dataframe.write_parquet("/Users/a/lineitem_sorted.parquet", None).await?;
        Ok(())
    }

@andygrove Seems the test still trying to consume only available memory, without exhausting all machine memory

Error: External(ResourcesExhausted("Failed to allocate additional 1419104 bytes for RepartitionExec[3] with 0 bytes already allocated - maximum available is 496736"))
test dataframe::tests::test_huge_sort ... FAILED

@tustvold However diskmanager doesn't spill anything into the folder, is it expected?

[tustvold]

You likely need to give it more than 1MB, not all operators can spill

[djouallah]

sorry, how do you pass those config using Python API

[andygrove]

sorry, how do you pass those config using Python API

We will need to expose them there. Should be trivial. I will add notes on the Python issue

[DDtKey]

Was able to reproduce similar behaviour against the latest version (17.0.0) and wasn't able against 16.0.0 - it returned correct error about allocation ( Resources exhausted: Failed to allocate additional 12372192 bytes ...).

So looks like regression to me. Gonna use bisect to find commit that has introduced it and report it here with details.

My case in two words: with latest version memory limits (FairSpillPool - 4GiB) were ignored and memory consuming up to OOM (join of 4 files with ~1gb of data in each one, without explicit ordering)

But when I set prefer_hash_join = false it also returned error with resources exhausting (by 17.0.0), but it looked weird 🤔

[DDtKey]

This behavior(at least my case described above) were introduced here (a9ddcd3, PR link).

Before - it returned Resources exhausted when I used memory-pool and currently the memory usage grows up to OOM.

It could be reproduced with similar code:
UPD: a bit more easier way to repro this is described in #5162

   let ctx = SessionContext::with_config_rt(
        SessionConfig::default(),
        Arc::new(
            RuntimeEnv::new(
                RuntimeConfig::new()
                    .with_memory_pool(Arc::new(FairSpillPool::new(4 * 1024 * 1024 * 1024))),
            )
            .unwrap(),
        ),
    );
    
    // I can share the file - it's kind of random data, but not sure what I can use to do that. 
    // However, it's reproducible for any files which joins could lead to a large result file (> memory pool limit).
    ctx.register_csv("hr", file_path, CsvReadOptions::default())
        .await?;
        
    // 4 joins - just to represent a problem
    let data_frame = ctx
        .sql(
            r#"
        SELECT hr1."Emp_ID"
        from hr hr1 
        left join hr hr2 on hr1."Emp_ID" = hr2."Emp_ID" 
        left join hr hr3 on hr2."Emp_ID" = hr3."Emp_ID" 
        left join hr hr4 on hr3."Emp_ID" = hr4."Emp_ID"
    "#,
        )
        .await?;
        
        data_frame
        .write_csv(output_path)
        .await?;

Additional notes

It's could be useful to mention - memory consumption are jumping for 3rd join. For 2 it works totally fine and even don't return an error.
This query working fine with optimizer.repartition_joins = false, it just don't consume so much memory without repartitions for joins (but, it will take much more time for sure).

[comphead]

HI @DDtKey I have checked both implementations (GreedyMemoryPool and FairSpillPool) for sort problem above #5108 (comment)

Both works correctly and returns ExhaustionError. I'll try to reproduce the same for repartition.

in the meantime you may want also to check OOM test for repartition https://github.com/apache/arrow-datafusion/blob/master/datafusion/core/src/physical_plan/repartition/mod.rs#L1150

[DDtKey]

@comphead likely it's different issues. Probably make sense to create another issue with this description

JFY: in my example there is no explicit ordering at all, it's only about repartitions.

It doesn't respect memory pool for sure in my cases 🤔
And probably related to usage of unbounded channels.

Thanks for the reference, I'll check

[alamb]

And probably related to usage of unbounded channels.

Interestingly, the point of #4867 was in fact to remove the unbounded channels. I am looking more into this issue