apache / apache/datafusion

ScalarUDF called twice when using filter on UDF column

Aperta
#4,030 1 commento 0 reazioni 0 assegnatari Vedi su GitHub
bug
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

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.