UDFs are no longer executed in a thread

[orf]

Describe the bug

In Datafusion 28 and below, UDFs where executed in a separate thread when writing to parquet. The example code below does not fail the assertion in version 28 but does in version 30 and git main.

If you have a UDF that expects to be running in a thread, or does some form of blocking computation then this change means your previously parallel dataframe plan becomes serial.

I couldn't spot anything in the release notes about this.

To Reproduce

deps:

[dependencies]
tokio = { version = "^1.0", features = ["rt-multi-thread", "full"] }
datafusion = { version = "=30", default-features = false, features = ["encoding__expressions", "zstd"] }

code:

#[tokio::main(flavor = "multi_thread", worker_threads = 10)]
async fn main() {
    use std::sync::Arc;
    use datafusion::arrow::array::{ArrayRef, BooleanArray};
    use datafusion::arrow::datatypes::DataType;
    use datafusion::logical_expr::Volatility;
    use datafusion::parquet::file::properties::WriterProperties;
    use datafusion::physical_plan::functions::make_scalar_function;
    use datafusion::prelude::*;

    let config = SessionConfig::default();
    let ctx = SessionContext::with_config(config);
    let df = ctx
        .read_parquet("input.parquet", ParquetReadOptions::default())
        .await
        .unwrap();

    let main_thread_id = std::thread::current().id();

    let func = create_udf(
        "some_func",
        vec![DataType::Binary],
        Arc::new(DataType::Boolean),
        Volatility::Immutable,
        make_scalar_function(move |args: &[ArrayRef]| {
            let func_thread_id = std::thread::current().id();
            assert_ne!(main_thread_id, func_thread_id);
            return Ok(Arc::new(BooleanArray::from(vec![true; args[0].len()])));
        }));

    let df = df
        .select(vec![col("hash"), func.call(vec![col("hash")])])
        .unwrap();

    let props = WriterProperties::builder()
        .build();
    df.write_parquet("some_dir/", Some(props)).await.unwrap();
}

Expected behavior

Sync UDF functions should be executed in a blocking thread pool.

Additional context

I thought this might be related to #7205, but it doesn't appear to be culprit.

[alamb]

hi @orf -- I am not sure that DataFusion guarantees that udfs will be running in a separate thread or that they can do blocking operations without stopping DataFusion

What is your blocking udf doing? Is it doing network operations or something where having async udfs (like in #6518 ) would help?

[alamb]

Given the age of this ticket I am going to close it as not a bug. please reopen if you disagree - otherwise let's continue the conversation on

• Async User Defined Functions (UDF) #6518