Performance issue when loading 6.5gb parquet file into memory

[mispp]

Is it expected that loading a ~6.5gb parquet file into memory has huge difference between polars and datafusion?
Datafusion's .cache() method takes ~2 minutes. Loading same data with polars takes ~15s.

polars start -> 2023-07-10T22:43:25.690623200+02:00
polars end -> 2023-07-10T22:43:40.854580400+02:00
datafusion start -> 2023-07-10T22:43:41.363312400+02:00
datafusion end -> 2023-07-10T22:45:32.949019300+02:00

Minimum working example is below.

Both are submitted with cargo run - if it makes a difference due to --release.

Code:

use std::io::Error;
use polars::prelude::*;
use datafusion::prelude::*;
use chrono;

#[tokio::main]
async fn main() -> Result<(), Error> {
    let _ = _dataframe2();

    let _ = _datafusion().await;

    Ok(())
}

pub async fn _datafusion() {
    let _ctx = SessionContext::new();

    let _read_options = ParquetReadOptions { file_extension: ".parquet", table_partition_cols: vec!(), parquet_pruning: None, skip_metadata: None };
    let _df = _ctx.read_parquet("/mnt/d/Projects/testdf/data/test_data.parquet", _read_options).await.unwrap();

    println!("datafusion start -> {:?}", chrono::offset::Local::now());

    let _cached = _df.cache().await;

    println!("datafusion end -> {:?}", chrono::offset::Local::now());   
}

pub fn _dataframe2() -> Result<String, PolarsError> {
    let mut file = std::fs::File::open("/mnt/d/Projects/testdf/data/test_data.parquet").unwrap();

    println!("polars start -> {:?}", chrono::offset::Local::now());

    let _df = ParquetReader::new(&mut file).finish().unwrap();

    println!("polars end -> {:?}", chrono::offset::Local::now());

    Ok("done".to_string())
}

Cargo.toml

[package]
name = "testdf"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
parquet = "40.0.0"
polars = { version = "0.30.0", features = ["lazy","temporal","describe","json","parquet","dtype-datetime","dtype-categorical", "sql", "streaming", "serde-lazy", "ipc", "dynamic_groupby", "sort_multiple", "rows", "dataframe_arithmetic", "partition_by"] }
serde = "1.0.163"
serde_json = "1.0.96"
connectorx = { version = "0.3.1", features = ["src_postgres", "dst_arrow", "dst_arrow2"] }
datafusion = "27.0.0"
tokio = "1.0"
chrono = "0.4.26"

[alamb]

🤔

I tried this with the ClickBench dataset (https://github.com/ClickHouse/ClickBench#data-loading) but Polars didn't seem to support that

polars start -> 2023-07-12T14:23:55.384025-04:00
thread 'main' panicked at 'called `Result::unwrap()` on an `Err` value: ComputeError(ErrString("cannot create series from Int16"))', src/main.rs:35:54
stack backtrace:

When I tried with a parquet file from TPCH scalefactor 10, I was able to replicate your results

I think this issue is that polars is reading the file in parallel while DataFusion is reading it with a single thread. I will file an issue about this

[mispp]

Thank you @alamb

If it helps I can add one observation: I was monitoring mem usage using Task Manager in Windows. Polaris has a consistent mem usage increase until completely loaded. Using data fusion, same speed (like Polars) is seen until about half the dataset being loaded, then it slows down and continues with this lower speed until fully loaded.

I can provide task manager pics if it matters.

[alamb]

I filed #6983 to track this issue

[alamb]

Here is a PR to fix it: #6984