pmoust commented on issue #24704:
URL: https://github.com/apache/datafusion/issues/24704#issuecomment-6117989998
@alamb you were right to be surprised. Most of what I posted above does not
stand, so here is the correction and what stock `datafusion-cli` shows.
**Corrections to my earlier comment**
1. The 54.1.0 vs 55.2.0 contrast compared two plans, not two versions. On
54.1.0 our session planned `AggregateExec mode=FinalPartitioned` over a hash
repartition. On 55.2.0 our integration ran the same statement with
`datafusion.optimizer.repartition_aggregations = false`, which gives one
single-partition `AggregateExec mode=Final` feeding `RepartitionExec:
RoundRobinBatch(32)`. That difference was on our side, not in DataFusion. What
tripped us: in 55.2.0 `length()` (also `octet_length` and `character_length`)
over a `Dictionary(Int32, Utf8)` argument returns `Dictionary(Int32, Int32)`,
where 54.1.0 cast the argument to `Utf8` first and returned `Int32` (`SELECT
arrow_typeof(length(arrow_cast('a', 'Dictionary(Int32, Utf8)')));` in
`datafusion-cli` on each version). It looks consistent with the dictionary
encoding preservation the 55 upgrade guide describes for `Coercion`, though I
have not traced it in the source; the guide presents that as something a
function opts into and doe
s not list the built-in length functions as affected, which is what caught us.
Our own average's type coercion admitted only plain integers, so it stopped
taking its exact path and our planner settings followed. We are fixing that on
our side.
2. Please disregard the source paragraph. I cannot find the
`MaterializedOutput::next_batch` I named in the published 55.2.0 crate, and the
two things I described as new are in 54.1.0 too: the hash aggregate hands its
output out as `batch.slice` chunks (`aggregates/row_hash.rs:878-879`), and
`RepartitionExec` reserves `batch.get_array_memory_size()` per batch
(`repartition/mod.rs:181`).
3. Stock `datafusion-cli` does not reproduce the `ResourcesExhausted` we
reported, on either version (below). That refusal came from our own memory pool
and accounting around the plan in point 1.
**What stock `datafusion-cli` shows**
ClickBench Q29 over the single-file `hits.parquet`, c6a.4xlarge (16 vCPU, 32
GB), `binary_as_string = true`, `target_partitions = 32`, greedy pool, one
warm-up then three runs, median wall, max RSS from `/usr/bin/time -v`, spill
counts from `EXPLAIN ANALYZE`.
Default settings (`repartition_aggregations = true`), `-m 10g`: 4.00 s and
2.75 GiB on 54.1.0, 3.97 s and 2.42 GiB on 55.2.0, no spills.
With `SET datafusion.optimizer.repartition_aggregations = false` both
versions plan
```
SortPreservingMergeExec <- ProjectionExec <- SortExec(TopK) <- FilterExec
<- RepartitionExec: RoundRobinBatch(32), input_partitions=1
<- AggregateExec: mode=Final <- CoalescePartitionsExec <- AggregateExec:
mode=Partial
```
| `-m` | version | result | median wall | max RSS | `RepartitionExec` spills
| `AggregateExec mode=Final` spills |
|---|---|---|---|---|---|---|
| 10g | 54.1.0 | ok | 4.44 s | 2.79 GiB | 0 | 0 |
| 10g | 55.2.0 | ok | 4.18 s | 2.40 GiB | 0 | 0 |
| 4g | 54.1.0 | ok | 4.37 s | 2.73 GiB | 1 (1.4 MB) | 0 |
| 4g | 55.2.0 | ok | 4.42 s | 2.36 GiB | 20 (48.0 MB) | 0 |
| 3g | 54.1.0 | ok | 4.46 s | 2.88 GiB | 1 (2.3 MB) | 0 |
| 3g | 55.2.0 | ok | 4.43 s | 2.43 GiB | 67 (164.6 MB) | 0 |
| 2g | 54.1.0 | ok | 12.88 s | 2.90 GiB | 0 | 2 (929.0 MB) |
| 2g | 55.2.0 | ok | 4.51 s | 2.37 GiB | 184 (449.8 MB) | 0 |
- Nothing fails and 55.2.0 is not slower in any cell.
- With `--disk-limit 0`, so that nothing can spill, both versions complete
at `-m 10g` and both fail at `-m 4g` as soon as a spill is attempted. I did not
run the fair pool with no disk. The repro reads `Referer` as plain `Utf8`, so
it does not go through the dictionary path in point 1.
- The one difference between the versions is in the last two columns: under
a tight pool the round-robin exchange spills 20, 67 and 184 times on 55.2.0
against 1, 1 and 0 on 54.1.0, where at 2g the final aggregate spills instead. I
do not know why and will not guess a second time.
**Where that leaves this**
Our report does not show a regression in 55 and says nothing about the
blocked-emission work here; I am sorry for the noise on this thread. If the
exchange spill counts above are worth their own issue, say so and I will open
one with this repro; if they are expected, there is nothing further from us.
```sql
SET datafusion.execution.parquet.binary_as_string = true;
SET datafusion.optimizer.repartition_aggregations = false;
SET datafusion.execution.target_partitions = 32;
CREATE EXTERNAL TABLE hits STORED AS PARQUET LOCATION 'hits.parquet';
EXPLAIN ANALYZE SELECT REGEXP_REPLACE("Referer",
'^https?://(?:www\.)?([^/]+)/.*$', '\1') AS k,
AVG(length("Referer")) AS l, COUNT(*) AS c, MIN("Referer")
FROM hits WHERE "Referer" <> '' GROUP BY k HAVING COUNT(*) > 100000 ORDER BY
l DESC LIMIT 25;
```
run as `datafusion-cli -m 3g --mem-pool-type greedy -f q.sql`; read
`spill_count` on the `RepartitionExec: partitioning=RoundRobinBatch(32)` line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
