[joybestourous]

Motivation

Sibling Top-K sharing (#861) needs a reproducible comparison that distinguishes remote pruning from the cost of publishing and distributing bounds. This prerequisite adds that workload independently of the feature, without changing library behavior.

Changes

• Add one Criterion target: pruning and no scan-pruning layouts, each with two or eight scan tasks, ordinary localhost gRPC, two partitions per task, and ORDER BY score DESC LIMIT 100.
• Check the complete ordered result and actual worker assignments. Report updates, useful received bounds, proven remote pruning, and scan rows/bytes. Keep correct queries in the samples when sharing has no demonstrated effect.
• Add a fixed-query mode for process CPU and resident-memory comparisons. The eight-task case has four times the input; it is not a constant-work fan-out experiment.

Validation and results

Both release builds and all four smoke cases per revision passed. Benchmark Clippy with -D warnings and formatting passed. The feature's fresh rebased suite passed 496 tests, with 3 ignored, its example, and CI-scoped library/Iceberg Clippy.

Measured baseline 5b80d0b45a6558df8d92e78863e4ee95929276f9 → feature 4f263e81eb3287d85ab1efeaba9da82768676d71, on 2026-11-10, ARM Neoverse-N1, Linux 6.8/aarch64, CPU affinity 0–7, Rust 1.95 release builds and the same lockfile. Candidate setup enables sharing; workload and checks are otherwise identical.

Each cell is the median paired candidate/baseline ratio [observed pair range], except RSS, which is the paired candidate-minus-baseline change in MiB. There are five pairs per measurement type; fractions are from the fixed-query 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 at two tasks and 65.386 → 56.385 MiB/query at eight: paired ratios 0.250 and 0.861 [0.818–0.897]. Controls retained identical scan bytes, 32.511 and 130.042 MiB/query, and zero group/range pruning. Both controls received useful bounds in 100/100 queries despite 0/100 proven effects; pruning cases received them in 150/150 and 100/100. All RSS ranges cross zero, so these runs do not establish a consistent memory change.

All 22 TPC-H SF1 queries completed with matching row counts and no errors. TOTAL, defined as the sum of per-query elapsed-time medians, had paired ratio 0.993 [0.990–1.004] over three pairs. These local workloads demonstrate conditional savings and measurable no-pruning costs, not a general production speedup.

Workloads, measurement boundaries, and reproduction

Each file has 32 groups of 65,536 rows. The pruning layout starts with high scores, followed by increasing lower scores. The control puts an excluded maximum-valued sentinel in every group, preventing statistics pruning. Row filtering and page indexes are disabled. The benchmark uses four Tokio threads; no scan waits for a remote bound.

Remote effectiveness requires an observed useful predicate and scan output below a conservative local-only bound, with actual pruning counters. The check covers row-group pruning and file-range stopping, uses actual file statistics, and discounts empty fragments. Scan bytes are separate Parquet-reader counters, not physical disk/network I/O. The filesystem is warm. Expected ordered values come from the fixture, not an independent query engine.

Five alternating matched pairs were run separately for Criterion and resources, with no concurrent builds or other intentional benchmarks. Criterion times complete collection: fixture setup, startup, planning, and checks after collection are excluded, but worker-completion observation can overlap collection. Its observed counts include calibration, warmup, and measured calls. Candidate proven-effect totals there were 350/350, 100/115, 0/115, and 0/65 in table order; every candidate query received a useful bound. Correct no-effect queries remained included.

Resources use five pairs of fresh processes, 30 queries/run for pruning/2 and 20 for each other case: 900 total queries across both arms. Process user+system CPU and sampled RSS cover the marked query window, including planning, execution, observation, and checking. Observation examines extra expressions when sharing is enabled: CPU measures instrumented-query cost, not production-only overhead or an upper bound on it. RSS can miss transient peaks; lifetime high-water RSS includes setup and is separate. All workers share one process, so this is not per-worker/coordinator attribution. Arrow's memory pool omits some expression/message allocations. No resource pairs were rejected; latency or lack of pruning was never a rejection criterion. Producer-report counts are not unique bounds or dynamic-filter network bytes.

Source inputs of the measured executables match the signed revisions above. The baseline is this prerequisite on upstream cb6c9c4; the candidate adds the feature and only its setup opt-in to the same benchmark. 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 for the two revisions, build before measuring, and alternate baseline/candidate order for five pairs:

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 execution and the resource_begin/resource_end markers. Use TOPK_RESOURCE_CASE=pruning/2 TOPK_RESOURCE_QUERIES=30; for the other three cases use 20 queries. Sample the benchmark process inside the marked window, not Cargo or whole-process startup.

For TPC-H, only the candidate received an additional local measurement adapter 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);

That adapter is not part of the published runner. Build dfbench separately in each revision; the benchmark observer is a separate target and is not part of dfbench. Generate one shared default-layout dataset and run the same command for each executable, three pairs in AB/BA/AB order:

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

The SF1 dataset is 638,992,025 bytes, with default 128-MiB maximum file size, dictionary pages, and unsorted generation. Available disk space prevented SF10. All five iterations, including the first, contribute to each query median; TOTAL sums all 22 medians without integer truncation. Elapsed time includes planning/collection and q15 setup. The standard runner's local-worker shortcut and localhost topology are held equal. All 660 query iterations succeeded with matching row counts; full values and per-query sibling activation were not checked. This suite has no CPU/RSS instrumentation and does not qualify SF10 or remote-host behavior.

The ordinary integration feature set does not enable every benchmark/chaos suite. CI-scoped library/Iceberg Clippy passed; the known unrelated whole-workspace remote-worker Clippy failure was not rerun. No full-workspace CI parity is claimed.