ScalarUDF called twice when using filter on UDF column
- Lingua principale
- Rust
- Stelle
- 9.3k
- Fork
- 2.4k
- Merge medio
- 3g 11h
- PR unite (30g)
- 360
Descrizione
**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)
Guida per i contributori
Apri la guida per i contributori
Direzione di ricerca
Riproduci il report usando l’esempio Rust con ctx.register_udf, filter ed explain, quindi esamina i piani logico e fisico mostrati. Traccia dove viene introdotta l’espressione UDF nella proiezione e nel filtraggio; l’issue è completata quando l’UDF viene valutata una sola volta per riga di input e i risultati filtrati corrispondono al predicato.
Scritto dal modello di indicizzazione a partire dal testo della issue.
Valutazione
- Stack tecnologico
- rust, sql
- Ambito
- databases
- Tipo di issue
- Bug
- Difficoltà
- 4/5
- Tempo stimato
- 3-5 giorni
- Stato di attività
- Ferma
- Chiarezza
- Abbastanza chiara
- Idoneità per principianti
- 35/100