Regression: `DataFrameWriteOptions::with_single_file_output` produces a directory

[sergiimk]

Describe the bug

Consider a snippet like this:

df.write_parquet(
  "dir/data",
  DataFrameWriteOptions::new().with_single_file_output(true),
  None
).await

Before v43 this would write a single file called data, but in v43 this is creating data as a directory with a randomly named file(s) in it.

This seems to be related to #13079 (cc @dhegberg) that added an extension-based heuristic.

I see this as a regression, as single file output is requested explicitly, and I don't want a heuristics to be applied.

We are using Parquet files with a content-addressable file system and our files don't have extensions.

To Reproduce

See above

Expected behavior

Considering the introduction of the extension-based heuristic I would suggest the following behavior:

• with_single_file_output is not called (single_file_output == None) - apply the heuristic
• with_single_file_output(true) - produce a single file at the exact path specified
• with_single_file_output(false) - create directory under specified path if doesn't exist and write one or many files

Additional context

[alamb]

I agree this is a regression. Thank you for the callout @sergiimk

I think this is a pretty good first issue for someone as the description is clear and the need is well defined.

[irenjj]

take

[irenjj]

It seems hard to control the behavior of write_parquet by single_file_output(and I've noticed that It's never used), what really controls whether to generate a single file output is determining the suffix(in start_demuxer_task()), there are several methods I can think of to handle this issue:

1. We can add a suffix like .single to the paths that require generating a single file, and then recognize this suffix in start_demuxer_task().
2. Give up single_file_output in DataFrameWriteOptions, use FileSinkConfig instead to control single file behavior.

cc @alamb @sergiimk @dhegberg

[sergiimk]

Did some digging and found this old PR #9041 (cc @yyy1000) that seems to have removed single_file_output flag from FileSinkConfig - worth looking into it to understand the reasoning, not to undo the changes.

Looking at v42 code it does indeed seem that DataFrameWriteOptions::single_file_output is not read anywhere and the logic relies on just this condition in demux:

let single_file_output = !base_output_path.is_collection();

which in v43 became:

let single_file_output = !base_output_path.is_collection() && base_output_path.file_extension().is_some();

The DataFrameWriteOptions::new().with_single_file_output(true) is used in a bunch of tests though, so it's just a lucky coincidence that all tests give file a proper test.parquet name and not just test.

Personally I think that all kinds of extension-based heuristics don't belong in such low level code like start_demuxer_task and perhaps better left at the DataFrame level.

Whichever heuristic version (pre v36, pre v43, or post v43) is the right one - I don't really mind, but I think there should be a way to skip it and specify explicitly.

[irenjj]

It seems that the previous PR intentionally removed the single_file_output option, but later introduced a heuristic.

[alamb]

I feel like i reviewed a PR recently related to this issue but could not find it. I wonder if it is still valid

• I am collecting regressions we should try and fix for the next release Release DataFusion 45.0.0 #14008

[nssalian]

@alamb , is this issue still valid?

[alamb]

@alamb , is this issue still valid?

I am not sure @nssalian -- could you try reproducing the description and see if it still happens?

[hknlof]

This still is happening with DataFusion 49. Using .parquet suffix in output str, aligns to expected behavior of this issue.

use datafusion::{
    dataframe::DataFrameWriteOptions,
    prelude::{ParquetReadOptions, SessionContext},
};

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    let ctx = SessionContext::new();
    let options = ParquetReadOptions::new();
    let df = ctx
        .read_parquet("./data/yellow_tripdata_2025-01.parquet", options)
        .await?;

    let clone_df = df.clone();

    df.write_parquet(
        "./data/new_yellow_caps.parquet", // pre-serve the shape of the issue
        DataFrameWriteOptions::new().with_single_file_output(true),
        None,
    )
    .await?;

    clone_df
        .write_parquet(
            "./data/new_yellow_caps", // pre-serve the shape of the issue
            DataFrameWriteOptions::new().with_single_file_output(true),
            None,
        )
        .await?;

    Ok(())
}

hknlof@localhost data_samples % find data -type f | sort
data/new_yellow_caps.parquet
data/new_yellow_caps/kWLvaFapgf6OoXk5_0.parquet
data/new_yellow_caps/kWLvaFapgf6OoXk5_1.parquet
data/new_yellow_caps/kWLvaFapgf6OoXk5_2.parquet
data/new_yellow_caps/kWLvaFapgf6OoXk5_3.parquet
data/yellow_tripdata_2025-01.parquet

# Cargo.toml
[package]
name = "data_samples"
version = "0.1.0"
edition = "2024"

[dependencies]
datafusion = "49.0.0"
tokio = { version = "1.47.0", features = ["rt-multi-thread", "tokio-macros"] }