apache / apache/datafusion

ScalarUDF called twice when using filter on UDF column

Open
#4,030 1 comment 0 reactions 0 assignees View on GitHub
bug
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

**Describe the bug**
Filtering results from a ScalarUDF results in it being called twice.

**To Reproduce**

``` rust

ctx.register_csv("csv", "/test.csv", CsvReadOptions::new()).await.unwrap();

let udf = {
create_udf(
"rand_bool",
vec![DataType::Float32],
Arc::new(DataType::Boolean),
Volatility::Stable,
make_scalar_function(|a| {
const BOOLS: [bool; 4] = [true, true, false, false];

let x = a.first().unwrap();
println!("udf in: {x:?}");

Ok(Arc::new(BooleanArray::from(Vec::from(&BOOLS[..x.len()]))) as ArrayRef)
}),
)
};

ctx.register_udf(udf.clone());

let query = ctx.table("csv").unwrap()
.select(vec![
Expr::Wildcard,
udf.call(vec![col("num")]).alias("rand"),
]).unwrap()
.filter(col("rand").eq(lit(false))).unwrap();

query.show_limit(10).await.unwrap();
query.explain(false, false).unwrap().show().await.unwrap();

```
Same happens with SQL:
``` sql
SELECT * FROM (SELECT *, rand_bool(num) AS rand FROM csv) WHERE NOT rand
```

The UDF is not so "stable". Regardless it should not be called twice (prints `udf in: PrimitiveArray ...` twice). And the results can actually return `true` when the filter is false.

Output + Test files

Output:
```
udf in: PrimitiveArray
[
100.0,
200.0,
150.0,
300.0,
]
udf in: PrimitiveArray
[
150.0,
300.0,
]
+--------+-----+------+
| name_1 | num | rand |
+--------+-----+------+
| andy | 150 | true |
| paul | 300 | true |
+--------+-----+------+
+---------------+------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | Projection: csv.name_1, csv.num, rand_bool(CAST(csv.num AS Float32)) AS rand |
| | Filter: NOT rand_bool(CAST(csv.num AS Float32)) |
| | TableScan: csv projection=[name_1, num], partial_filters=[NOT rand_bool(CAST(csv.num AS Float32))] |
| physical_plan | ProjectionExec: expr=[name_1@0 as name_1, num@1 as num, rand_bool(CAST(num@1 AS Float32)) as rand] |
| | CoalesceBatchesExec: target_batch_size=4096 |
| | FilterExec: NOT rand_bool(CAST(num@1 AS Float32)) |
| | RepartitionExec: partitioning=RoundRobinBatch(12) |
| | CsvExec: files=[<>/test.csv], has_header=true, limit=None, projection=[name_1, num] |
| | |
+---------------+------------------------------------------------------------------------------------------------------------------------------------------------+

```

test.csv
``` csv
name_1,num
andrew,100
jorge,200
andy,150
paul,300
```

**Expected behavior**
udf should be projected first then filtered.

**Additional context**
Running @ master (d391b859c44e1c366eb4da5e8cabd199336f4243)

Contributor guide

Open the contributing guide

Research direction

Reproduce the report using the Rust example with ctx.register_udf, filter, and explain, then inspect the shown logical and physical plans. Trace where the UDF expression is introduced in projection and filtering; the issue is done when the UDF is evaluated once per input row and filtered results match the predicate.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust, sql
Domain
databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.