[joybestourous]

Motivation

A task executing ORDER BY ... LIMIT K can discover a useful cutoff while sibling tasks continue scanning weaker candidates. This adds opt-in sharing of those bounds so siblings can prune remaining work, while keeping each task's own Top-K state independent.

This PR remains draft, stacked on the independent benchmark prerequisite, based on upstream cb6c9c4. Feature-only diff.

Design

• Export default-off SiblingTopKConfig through the existing option-extension API. Existing remote, general dynamic-filter, and Top-K pushdown settings must also permit sharing.
• Reuse producer streams, the registry, transport, and expression remapping. Compatible fetched-sort predicates combine with AND; join-filter merge rules are unchanged.
• Give each worker a separate remote consumer. Received state never replaces or republishes the local producer. A snapshot may contain its recipient's older bound, redundant with the current local bound.
• Keep local filtering when a custom worker plan cannot accept the extra predicate without changing topology. Include focused tests, a guide, an example, and the benchmark's setup opt-in.

The final merge still chooses the result; sharing does not compute an exact global kth row early. DataFusion's NULL ordering and ordinary LIMIT tie semantics remain. Coordinator and workers must all support the feature before enabling it; unchanged messages do not make mixed-version opt-in safe.

Validation and measurements

Fresh rebased validation passed 496 tests, with 3 ignored, the example, formatting, and CI-scoped library/Iceberg Clippy with -D warnings. The public-path activation test proves remote Parquet pruning while the receiver's local bound remains unrestricted. Other coverage includes projection/multiple keys/NULLs/ties, disabled settings, under-K inputs, inner-fetch barriers, joins, and unsupported/delegating plans. Both release builds, four smoke cases per arm, and final benchmark Clippy passed. GitHub CI for this head failed before creating any jobs or logs; the cause is unavailable and rerunning requires a maintainer. CI is not green.

Measured baseline 5b80d0b45a6558df8d92e78863e4ee95929276f9 → feature 4f263e81eb3287d85ab1efeaba9da82768676d71, on 2026-11-10, ARM Neoverse-N1, Linux 6.8/aarch64, affinity 0–7, Rust 1.95 release builds and the same lockfile. Values are median paired candidate/baseline ratios [observed pair ranges]; RSS is instead the paired change in MiB. Five pairs per measurement type; effective fractions are from resource runs.

Case	Collection time	Instrumented CPU/query	Sampled peak RSS change (MiB)	Effective candidate queries
Pruning, 2 tasks	0.354 [0.335–0.369]	0.175 [0.173–0.175]	−1.777 [−3.176–+0.520]	150/150
Pruning, 8 tasks	0.873 [0.845–0.883]	0.866 [0.811–0.886]	+1.676 [−0.355–+3.496]	95/100
No scan pruning, 2 tasks	1.078 [1.062–1.093]	1.004 [1.001–1.006]	+0.215 [−0.570–+0.738]	0/100
No scan pruning, 8 tasks	1.075 [1.039–1.096]	1.064 [1.040–1.072]	+0.801 [−1.414–+2.828]	0/100

Pruning scan bytes fell from median 16.003 → 4.001 MiB/query and 65.386 → 56.385 MiB/query, paired ratios 0.250 and 0.861 [0.818–0.897]. Control bytes stayed identical at 32.511 and 130.042 MiB/query, with no group/range pruning. Useful bounds reached 150/150 and 100/100 pruning queries, and 100/100 in each control despite zero effective queries. Sharing reduced work in the pruning cases; controls paid about 7.8%/7.5% collection-time and 0.4%/6.4% instrumented CPU overhead. Every RSS range crosses zero, so no consistent memory change is established.

All 22 TPC-H SF1 queries completed with matching row counts and no errors. TOTAL, the sum of per-query elapsed-time medians, had paired ratio 0.993 [0.990–1.004] over three pairs. The option remains disabled by default: these synthetic/local results show conditional savings and costs, not a general production benefit.

Measurement scope, reproduction, validation limits, and scaling costs

The prerequisite uses two/eight actual scan tasks, two partitions per task, four Tokio threads, and ordinary localhost gRPC. Each task adds a Parquet file of 32 groups × 65,536 rows; eight tasks mean four times the input, not constant-work scaling. One layout allows early high bounds to prune low groups; the control has an excluded maximum-valued sentinel in every group. Row filtering and page indexes are disabled. No scan waits for a bound; correct queries remain included when remote effects are absent.

Every focused query checks complete ordered fixture-derived values and actual worker assignments. Effectiveness requires a useful remote predicate, pruning counters, and scan rows below a conservative local-only bound, covering row-group pruning/file-range stopping and discounting empty fragments. Scan bytes are separate Parquet-reader counters, not physical disk/network traffic. The filesystem is warm. A final useful bound alone is not proof that it helped execution.

Criterion and resources each use five alternating matched pairs without concurrent builds or other intentional benchmarks. Collection timing excludes setup, planning, and checks after collection; worker-completion observations can overlap collection. Criterion counts include calibration, warmup, and measured calls: candidate effects were 350/350, 100/115, 0/115, and 0/65 in table order, with useful bounds in every candidate query.

Fixed-work resources use fresh processes with 30 queries/run for pruning/2 and 20 for each other case, 900 total queries across both arms. The marked window includes planning, execution, observation, and checking. Process CPU includes feature-dependent observer work and is neither production-only overhead nor an upper bound on it. Sampled RSS can miss transient peaks; lifetime high-water RSS includes setup and is separate. All workers share one process, so results do not attribute resources to coordinator/individual workers. No resource pairs were rejected; no-effect queries were not discarded. Producer reports do not measure unique cutoffs or dynamic-control traffic bytes.

Measured executable source inputs match the signed revisions above. The baseline includes the independent benchmark on upstream cb6c9c4; the candidate includes this feature and the benchmark setup opt-in. Workload/observer code is otherwise identical. Both use lockfile SHA-256 8d9b2764248dcf2fed37f62659502a603a06ac8436b13cd4456ad444e30f4382, Rust 1.95.0, Clang 20.1.8 with LLD, release optimization, incremental compilation off, and four build jobs. Use separate build directories, build before measurement, and alternate the two arms:

export CARGO_INCREMENTAL=0 CARGO_BUILD_JOBS=4
export RUSTFLAGS='-C linker=clang -C link-arg=-fuse-ld=lld'
cargo bench --locked -p datafusion-distributed-benchmarks --bench sibling_topk --no-run
taskset -c 0-7 cargo bench --locked -p datafusion-distributed-benchmarks --bench sibling_topk -- --noplot

The benchmark README documents fixed-work mode. Use TOPK_RESOURCE_CASE=pruning/2 TOPK_RESOURCE_QUERIES=30, or 20 queries for the other cases, and sample the benchmark process only between resource_begin/resource_end.

TPC-H used a separate candidate-only local adapter, not a published runner change: in datafusion/distributed-benchmarks/src/run.rs, import SiblingTopKConfig from datafusion_distributed, then insert this in RunOpt::run_local immediately before backend.configure_session(builder).build():

let mut sharing = SiblingTopKConfig::default();
sharing.enabled = true;
builder.set_distributed_option_extension(sharing);

Leave the baseline unchanged. Build each dfbench separately; the focused observer is a different target and is not in dfbench. Generate one shared dataset and run each executable in three AB/BA/AB pairs:

cargo build --locked --release -p datafusion-distributed-benchmarks --bin dfbench
DATASET=$(target/release/dfbench prepare-tpch --scale-factor 1)
TOKIO_WORKER_THREADS=8 RUST_LOG=warn taskset -c 0-7 target/release/dfbench run --dataset "$DATASET" --workers 8 --partitions 2 --iterations 5

SF1 used 638,992,025 bytes with default 128-MiB maximum files, dictionary pages, and unsorted generation; available disk prevented SF10. All five iterations, including the first, enter each query median; TOTAL sums all 22 medians without integer truncation. Existing elapsed timing includes planning/collection and q15 setup. The standard local-worker shortcut and localhost topology are equal across arms. All 660 query iterations succeeded with matching row counts, but full values and sibling activation in each suite query were not checked. The suite has no CPU/RSS instrumentation and does not qualify SF10 or remote-host behavior.

Fresh tests were 349 unit + 128 integration + 19 doctest passes, with 3 ignored. The ordinary integration feature set does not enable every benchmark/chaos suite. Current CI-scoped library/Iceberg Clippy passed; the known unrelated whole-workspace remote-worker lint failure was not rerun. Historical sensitivity tests are not claimed as fresh rebase validation. No staging/production experiment or full-workspace CI parity is claimed.

The registry keeps the latest bound per producer, not every historical update. With N producers and N recipients, full-predicate fan-out can require O(N²) payload-copy work per merged update; receivers also decode/remap snapshots. Existing unbounded control queues can retain obsolete snapshots for slow recipients. Arrow's memory pool does not account for all these allocations. This patch adds no rate limiting, queue coalescing, strongest-bound reduction, or new telemetry.