ScalarUDF called twice when using filter on UDF column
- 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
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