Drop velocity-based estimation in favor of driver path completion estimation by gabotechs · Pull Request #552 · datafusion-contrib/datafusion-distributed

gabotechs · 2026-07-13T12:28:57Z

Summary

Replaces the velocity-based runtime-statistics estimate used by the dynamic task-count planner with a driver-path completion estimate.

The sampler on top of each stage feeds statistics (final row/byte counts) back to the coordinator, which uses them to size the stage above. The old estimate extrapolated those final sizes from a sampled throughput multiplied by a fixed, assumed query duration, which systematically overestimated — often by orders of magnitude. The new estimate instead scales what a stage has already produced by how far it is through its known total work.

How it worked before

The SamplerExec reported, per partition:

*_ready — rows / bytes already materialized at sampling time, and
*_per_second — a sampled production velocity (rows/bytes per second).

gather_runtime_statistics extrapolated the final size as:

total_rows  = rows_ready  + rows_per_second  * ESTIMATED_QUERY_TIME_S   // = 10
total_bytes = bytes_ready + bytes_per_second * ESTIMATED_QUERY_TIME_S

i.e. "what's ready now, plus the sampled rate projected over 10 more seconds of execution."

While doing some benchmarks, I've seen that this approach heavily over-estimates row count.

How the new estimate avoids it

Instead of projecting a velocity over a made-up duration, it computes an actual completion fraction and divides by it:

pct_completed = rows_pulled_from_leaf / estimated_driver_path_leaf_rows
total_rows    = rows_ready  / pct_completed
total_bytes   = bytes_ready / pct_completed

rows_pulled_from_leaf — how many rows the stage has actually pulled from its leaf scans so far (now reported by the sampler in place of the velocity fields).
estimated_driver_path_leaf_rows — the total number of rows the stage must pull to finish, obtained by summing the leaf num_rows statistics along the driver path: the probe-side chain that actually streams to the output. Join build sides are excluded because they are one-off setup work, not output progress (see the new apply_driver_path helper). These counts come from real source statistics (e.g. parquet row counts).

Caveats

If a stage's leaves expose no row-count statistics, it falls back to a fixed FALLBACK_ESTIMATED_PCT_SAMPLED = 0.2 (assume ~20 % sampled).

Benchmarks

TPCH-SF1: 1.02 faster
TPCH-SF10: 1.04 faster
TPCH-SF100: 1.01 faster
TPC-DS-SF1: 1.01 slower
ClickBench: 1.01 slower

Performance is not impact because both in main and this PR max out on tasks for a cluster of 12 workers.

Regarding q-error on stats estimation, here are the results:

TPCH-SF1:
QERR P50: prev=1.80x, new=1.80x
QERR P95: prev=6.75x, new=2.29x

TPCH-SF10:
QERR P50: prev=1.80x, new=1.45x
QERR P95: prev=8.94x, new=2.12x

TPCH-SF100:
QERR P50: prev=2.12x, new=1.73x
QERR P95: prev=9.68x, new=2.16x

TPC-DS-SF1:
QERR P50: prev=1.54x, new=1.45x
QERR P95: prev=4.62x, new=4.01x

ClickBench:
QERR P50: prev=1.42x, new=1.45x
QERR P95: prev=1.42x, new=1.45x

gabotechs · 2026-07-13T15:14:28Z

-/// truncated to this max value, as it's assumed that [READY_CHUNK_LIMIT] was hit and no useful
-/// measurement can actually be emitted.
-const MAX_ROWS_PER_SECOND: usize = 1024 * 1024;
+const READY_CHUNK_LIMIT: usize = 1024 * 1024;


Increased this value on purpose. The idea is that we should be able to yield as much as possible here syncrhonously, as these are record batches that are anyways already in-memory, so we prefer to sample as much as possible, given that they are already there.

asolimando

LGTM, I have a few questions on the recursion (jump notation and chasing only the right child for joins), the rest is solid and I agree that removing the temporal notion from the equation (especially with an unknown total duration) will provide more stable estimates.

asolimando · 2026-07-16T09:25:38Z

+                // mapping from SamplerExec to leaf node.
+                //
+                // Just dividing by `partition_count` gives a good approximation, but it's still
+                // not perfect as it will not account for data skews.


This is a perfectly reasonable start, I agree.

As a future improvement, I wonder if we can apply sampling similarly to get an idea of the degree of data skew we have across partitions. wdyt?

asolimando · 2026-07-16T09:28:47Z

+                if Arc::ptr_eq(p, &child) { TreeNodeRecursion::Jump } else { TreeNodeRecursion::Continue }
+            }),
+            @r"
+        Single


Nit: how hard would it be to have an indentation level for children? it's not a big deal as I can read the plan declaration to see what are the children, but if not hard it would have better readability

asolimando · 2026-07-16T09:30:39Z

+            let rec = decide(p);
+            let suffix = match rec {
+                TreeNodeRecursion::Continue => "",
+                TreeNodeRecursion::Jump => " [->jump]",


Q: I am not sure I got the "jump" notation, what does this correspond to in a distributed plan? A recursive step to model DAGs?

asolimando · 2026-07-16T09:32:24Z

+                    recurse(hash_join.right(), f)
+                } else if let Some(nested_loop_join) = plan.downcast_ref::<NestedLoopJoinExec>() {
+                    recurse(nested_loop_join.right(), f)
+                } else if let Some(cross_join) = plan.downcast_ref::<CrossJoinExec>() {
+                    recurse(cross_join.right(), f)


Q: why do we only follow the right children for these join nodes?

This PR enhances the benchmarks by comparing estimated bytes vs actual bytes in network boundaries during dynamic planninng: - Compare sampled bytes with actual network-boundary output bytes using q-error. - Store it per iteration in local and remote benchmarks. - Report median q-error (`QERR P50`) in benchmark comparisons. Preliminary step for #552. --- Sample output: ``` === Comparing tpch_sf10 results from branch 'stats-accuracy-benches' [prev] with 'stats-accuracy-benches' [new] === os: macos arch: aarch64 cpu cores: 16 threads: 2 -> 2 workers: 8 -> 8 =================================================================================================================== tpch_sf10 q1: prev= 327 ms, new= 355 ms, diff=1.09 slower ✖ tpch_sf10 q2: prev= 271 ms, new= 278 ms, diff=1.03 slower ✖ tpch_sf10 q3: prev= 378 ms, new= 346 ms, diff=1.09 faster ✔ tpch_sf10 q4: prev= 132 ms, new= 129 ms, diff=1.02 faster ✔ tpch_sf10 q5: prev= 617 ms, new= 610 ms, diff=1.01 faster ✔ tpch_sf10 q6: prev= 115 ms, new= 109 ms, diff=1.06 faster ✔ tpch_sf10 q7: prev= 765 ms, new= 743 ms, diff=1.03 faster ✔ tpch_sf10 q8: prev= 781 ms, new= 889 ms, diff=1.14 slower ✖ tpch_sf10 q9: prev=1040 ms, new=1010 ms, diff=1.03 faster ✔ tpch_sf10 q10: prev= 448 ms, new= 455 ms, diff=1.02 slower ✖ tpch_sf10 q11: prev= 132 ms, new= 121 ms, diff=1.09 faster ✔ tpch_sf10 q12: prev= 200 ms, new= 205 ms, diff=1.02 slower ✖ tpch_sf10 q13: prev= 233 ms, new= 232 ms, diff=1.00 faster ✔ tpch_sf10 q14: prev= 158 ms, new= 171 ms, diff=1.08 slower ✖ tpch_sf10 q15: prev= 262 ms, new= 241 ms, diff=1.09 faster ✔ tpch_sf10 q16: prev= 92 ms, new= 94 ms, diff=1.02 slower ✖ tpch_sf10 q17: prev= 833 ms, new= 816 ms, diff=1.02 faster ✔ tpch_sf10 q18: prev= 917 ms, new= 946 ms, diff=1.03 slower ✖ tpch_sf10 q19: prev= 187 ms, new= 199 ms, diff=1.06 slower ✖ tpch_sf10 q20: prev= 422 ms, new= 420 ms, diff=1.00 faster ✔ tpch_sf10 q21: prev=1044 ms, new=1022 ms, diff=1.02 faster ✔ tpch_sf10 q22: prev= 101 ms, new= 103 ms, diff=1.02 slower ✖ TOTAL: prev=9455 ms, new=9494 ms, diff=1.00 slower ✖ QERR P50: prev=n/a, new=19.47x ```

…imation

gabotechs force-pushed the gabrielmusat/leaf-sampler branch 2 times, most recently from c358e53 to f53c5ba Compare July 13, 2026 12:41

gabotechs commented Jul 13, 2026

View reviewed changes

Comment thread benchmarks/src/datasets/common.rs

Comment thread benchmarks/src/run.rs Outdated

gabotechs force-pushed the gabrielmusat/leaf-sampler branch from f53c5ba to 1ed6245 Compare July 13, 2026 14:51

gabotechs marked this pull request as ready for review July 13, 2026 15:12

gabotechs force-pushed the gabrielmusat/leaf-sampler branch from 1ed6245 to 64528a4 Compare July 13, 2026 15:13

gabotechs commented Jul 13, 2026

View reviewed changes

gabotechs force-pushed the gabrielmusat/leaf-sampler branch 2 times, most recently from 6c3a271 to 5da24a0 Compare July 16, 2026 07:21

gabotechs mentioned this pull request Jul 16, 2026

Add stats estimation accuracy to benchmarks #556

Merged

gabotechs changed the base branch from main to gabrielmusat/stats-accuracy-benches July 16, 2026 07:21

gabotechs force-pushed the gabrielmusat/leaf-sampler branch from 5da24a0 to 0ed7511 Compare July 16, 2026 07:43

gabotechs force-pushed the gabrielmusat/stats-accuracy-benches branch from c191f38 to 96a4ec4 Compare July 16, 2026 07:43

gabotechs force-pushed the gabrielmusat/leaf-sampler branch from 0ed7511 to d605408 Compare July 16, 2026 07:57

gabotechs force-pushed the gabrielmusat/stats-accuracy-benches branch 3 times, most recently from 0c9855b to ada6559 Compare July 16, 2026 08:14

gabotechs force-pushed the gabrielmusat/leaf-sampler branch from d605408 to 674fd1c Compare July 16, 2026 08:14

gabotechs force-pushed the gabrielmusat/stats-accuracy-benches branch from ada6559 to 53809fb Compare July 16, 2026 08:16

gabotechs force-pushed the gabrielmusat/leaf-sampler branch 2 times, most recently from 2ef9809 to 5bebc68 Compare July 16, 2026 08:21

gabotechs force-pushed the gabrielmusat/stats-accuracy-benches branch 2 times, most recently from 050416b to aac68dd Compare July 16, 2026 08:31

gabotechs force-pushed the gabrielmusat/leaf-sampler branch 2 times, most recently from c27c116 to 563d8a4 Compare July 16, 2026 09:11

gabotechs force-pushed the gabrielmusat/stats-accuracy-benches branch 2 times, most recently from 0a25d08 to 71c347f Compare July 16, 2026 09:19

gabotechs force-pushed the gabrielmusat/leaf-sampler branch 2 times, most recently from 0404839 to f1db5c4 Compare July 16, 2026 09:30

gabotechs force-pushed the gabrielmusat/stats-accuracy-benches branch from 71c347f to e7ee621 Compare July 16, 2026 09:30

asolimando reviewed Jul 16, 2026

View reviewed changes

gabotechs force-pushed the gabrielmusat/leaf-sampler branch from f1db5c4 to f4096f2 Compare July 16, 2026 10:26

gabotechs force-pushed the gabrielmusat/stats-accuracy-benches branch from e7ee621 to 82e6199 Compare July 16, 2026 10:26

gabotechs force-pushed the gabrielmusat/leaf-sampler branch from f4096f2 to b6b6c77 Compare July 16, 2026 10:47

gabotechs force-pushed the gabrielmusat/stats-accuracy-benches branch 2 times, most recently from 1610648 to 94eec9c Compare July 16, 2026 11:00

gabotechs force-pushed the gabrielmusat/leaf-sampler branch from b6b6c77 to 6bcf962 Compare July 16, 2026 11:00

Base automatically changed from gabrielmusat/stats-accuracy-benches to main July 16, 2026 13:09

gabotechs force-pushed the gabrielmusat/leaf-sampler branch from 6bcf962 to 1d003f1 Compare July 16, 2026 13:10

Drop velocity-based estimation in favor of driver path completion est…

958a6f7

…imation

gabotechs force-pushed the gabrielmusat/leaf-sampler branch from 1d003f1 to 958a6f7 Compare July 17, 2026 10:11

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Drop velocity-based estimation in favor of driver path completion estimation#552

Drop velocity-based estimation in favor of driver path completion estimation#552
gabotechs wants to merge 1 commit into
mainfrom
gabrielmusat/leaf-sampler

gabotechs commented Jul 13, 2026 •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

gabotechs Jul 13, 2026 •

edited

Loading

Uh oh!

asolimando left a comment

Uh oh!

asolimando Jul 16, 2026

Uh oh!

asolimando Jul 16, 2026

Uh oh!

asolimando Jul 16, 2026

Uh oh!

asolimando Jul 16, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

gabotechs commented Jul 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

How it worked before

How the new estimate avoids it

Caveats

Benchmarks

Uh oh!

Uh oh!

Uh oh!

gabotechs Jul 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

asolimando left a comment

Choose a reason for hiding this comment

Uh oh!

asolimando Jul 16, 2026

Choose a reason for hiding this comment

Uh oh!

asolimando Jul 16, 2026

Choose a reason for hiding this comment

Uh oh!

asolimando Jul 16, 2026

Choose a reason for hiding this comment

Uh oh!

asolimando Jul 16, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

gabotechs commented Jul 13, 2026 •

edited

Loading

gabotechs Jul 13, 2026 •

edited

Loading