Repository navigation
feat: balance whole-file Iceberg work across tasks - #788
alexanderbianchi wants to merge 1 commit into
Conversation
fa1fd87 to
a72c891
Compare
a72c891 to
839fece
Compare
839fece to
8b3e3e9
Compare
8b3e3e9 to
07abef7
Compare
a513d32 to
8c32533
Compare
| /// Assign whole files largest-first to the least-loaded task. Requires eager_file_planning. | ||
| pub greedy_file_assignment: bool, default = false | ||
| } |
There was a problem hiding this comment.
🤔 If greedy file assignment proves to be more performant, why making it an option? why not just have that be the only way of working with eager planned files?
Also, why would this not be compatible with execution-time file discovery (eager_file_planning:false)? At the moment of discovering the files, I imagine all the information relevant for routing to an appropriate partition from an appropriate worker is available equally on both execution modes.
| task_count: usize, | ||
| partitions_per_task: usize, | ||
| ) -> Self { | ||
| let mut order: BinaryHeap<FilePriority<'_>> = BinaryHeap::with_capacity(files.len()); |
There was a problem hiding this comment.
Damn... LeetCode intensifies here.
| /// Destination partitions in file-stream order. The dispatcher validates nonzero | ||
| /// counts and their product before constructing either routing policy. | ||
| #[derive(Debug)] | ||
| pub(crate) enum FilePartitions { |
There was a problem hiding this comment.
Pretty good abstraction here 👍 but I wonder if why might just want to go all-in with the Greedy approach instead. If it proves to be more performant, I don't see why we'd want to keep the RoundRobin one at all.
| let result: Result<()> = async { | ||
| let mut destinations = if greedy { | ||
| let files = planned.as_ref().ok_or_else(|| { |
There was a problem hiding this comment.
🤔 This is starting to have a good level of nesting. I don't have a good suggestion on how to structure it differently though... If you have ideas they are welcome, otherwise, it's good as it is.
| /// Uses the harness session and runtime, with the requested distributed fan-out. | ||
| pub async fn scan_file_tasks(&self, task_count: usize) -> Result<Vec<Vec<FileScanTask>>> { |
There was a problem hiding this comment.
Isn't this exposing the internals a bit?
The order and distribution in which FileScanTasks are yielded is not part of the public API contract, and I'd consider this an implementation detail over which the integration tests should be abstracted.
I'm trying to think about alternatives, for example: Surfacing how FileScanTask are yielded somehow as metrics in the scan node.
Alternatively, if you think it's worth still surfacing how FileScanTasks are yielded to the integration tests, having a text-based representation and using insta for asserting that output can be very useful, as it would result in pretty readable tests, and it'd be very easy to qualify changes in the FileScanTask distribution algorithm.
|
benchmarks run tpch/sf100_iceberg --config iceberg.greedy_file_assignment=true --config iceberg.eager_file_planning=true --base main |
|
Requested by this comment. Benchmark resultsCompared: Main === Comparing tpch/sf100_iceberg results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] === TASKS: prev=1320.0, new=1320.0, diff=no change (sum of per-query averages) TOTAL: prev=99718 ms, new=92100 ms, diff=1.08 faster ✔ Show full query output q1: prev=4752 ms, new=4769 ms, diff=1.00 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
q2: prev=2218 ms, new=2059 ms, diff=1.08 faster ✔, tasks: prev=96.0, new=96.0, diff=no change
q3: prev=6477 ms, new=5426 ms, diff=1.19 faster ✔, tasks: prev=58.0, new=58.0, diff=no change
q4: prev=3217 ms, new=2949 ms, diff=1.09 faster ✔, tasks: prev=48.0, new=48.0, diff=no change
q5: prev=4422 ms, new=4130 ms, diff=1.07 faster ✔, tasks: prev=85.0, new=85.0, diff=no change
q6: prev=3220 ms, new=2778 ms, diff=1.16 faster ✔, tasks: prev=12.0, new=12.0, diff=no change
q7: prev=5793 ms, new=5150 ms, diff=1.12 faster ✔, tasks: prev=85.0, new=85.0, diff=no change
q8: prev=6489 ms, new=5999 ms, diff=1.08 faster ✔, tasks: prev=104.0, new=104.0, diff=no change
q9: prev=7437 ms, new=6912 ms, diff=1.08 faster ✔, tasks: prev=104.0, new=104.0, diff=no change
q10: prev=6427 ms, new=5865 ms, diff=1.10 faster ✔, tasks: prev=71.0, new=71.0, diff=no change
q11: prev=1430 ms, new=1363 ms, diff=1.05 faster ✔, tasks: prev=65.0, new=65.0, diff=no change
q12: prev=3844 ms, new=3394 ms, diff=1.13 faster ✔, tasks: prev=48.0, new=48.0, diff=no change
q13: prev=3128 ms, new=2899 ms, diff=1.08 faster ✔, tasks: prev=46.0, new=46.0, diff=no change
q14: prev=3875 ms, new=3450 ms, diff=1.12 faster ✔, tasks: prev=30.0, new=30.0, diff=no change
q15: prev=7096 ms, new=6293 ms, diff=1.13 faster ✔, tasks: prev=50.0, new=50.0, diff=no change
q16: prev=1294 ms, new=1180 ms, diff=1.10 faster ✔, tasks: prev=55.0, new=55.0, diff=no change
q17: prev=5013 ms, new=4317 ms, diff=1.16 faster ✔, tasks: prev=42.0, new=42.0, diff=no change
q18: prev=5955 ms, new=6451 ms, diff=1.08 slower ✖, tasks: prev=70.0, new=70.0, diff=no change
q19: prev=4166 ms, new=3798 ms, diff=1.10 faster ✔, tasks: prev=30.0, new=30.0, diff=no change
q20: prev=5256 ms, new=5292 ms, diff=1.01 slower ✖, tasks: prev=67.0, new=67.0, diff=no change
q21: prev=7230 ms, new=6542 ms, diff=1.11 faster ✔, tasks: prev=86.0, new=86.0, diff=no change
q22: prev= 979 ms, new=1084 ms, diff=1.11 slower ✖, tasks: prev=44.0, new=44.0, diff=no change
Verification and run detailsJob
Workload: Capacity: 12 PR-head configs: Other timings: Queue 1s · Dataset validation 0s · Total 24m 26s How to use the benchmark botPost a comment whose first non-empty line is:
For example:
Currently available datasets:
Supported instances and per-worker limits:
Each node runs one worker. The worker allocation reserves 1 vCPU and 4 GiB for Kubernetes and system processes, then makes the rest of the selected instance available to the benchmark. Limits: Only authorized users can enqueue jobs. Jobs run serially, the queue holds 20 active jobs, and each requester may have 3. Query selection is not supported. By default, every query uses 1 warmup and 5 measured iterations for both revisions. |
|
benchmarks run tpch/sf1_iceberg tpch/sf10_iceberg --config iceberg.greedy_file_assignment=true --config iceberg.eager_file_planning=true --base main |
|
Requested by this comment. Benchmark resultsCompared: Main === Comparing tpch/sf1_iceberg results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] === TASKS: prev=151.0, new=150.2, diff=0.8 fewer (0.5%) (sum of per-query averages) TOTAL: prev=14316 ms, new=15697 ms, diff=1.10 slower ✖ Show full query output q1: prev= 414 ms, new= 344 ms, diff=1.20 faster ✔, tasks: prev=4.0, new=3.2, diff=0.8 fewer (20.0%)
q2: prev= 687 ms, new= 716 ms, diff=1.04 slower ✖, tasks: prev=0.0, new=0.0, diff=no change
q3: prev= 699 ms, new= 724 ms, diff=1.04 slower ✖, tasks: prev=5.0, new=5.0, diff=no change
q4: prev= 506 ms, new= 463 ms, diff=1.09 faster ✔, tasks: prev=7.0, new=7.0, diff=no change
q5: prev= 768 ms, new= 822 ms, diff=1.07 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
q6: prev= 277 ms, new= 341 ms, diff=1.23 slower ✖, tasks: prev=2.0, new=2.0, diff=no change
q7: prev= 839 ms, new= 933 ms, diff=1.11 slower ✖, tasks: prev=15.0, new=13.0, diff=2.0 fewer (13.3%)
q8: prev= 916 ms, new=1267 ms, diff=1.38 slower ✖, tasks: prev=17.0, new=17.0, diff=no change
q9: prev=1045 ms, new=1296 ms, diff=1.24 slower ✖, tasks: prev=15.0, new=15.0, diff=no change
q10: prev= 720 ms, new= 975 ms, diff=1.35 slower ✖, tasks: prev=6.0, new=8.0, diff=2.0 more (33.3%)
q11: prev= 648 ms, new= 600 ms, diff=1.08 faster ✔, tasks: prev=0.0, new=0.0, diff=no change
q12: prev= 519 ms, new= 487 ms, diff=1.07 faster ✔, tasks: prev=7.0, new=7.0, diff=no change
q13: prev= 362 ms, new= 355 ms, diff=1.02 faster ✔, tasks: prev=0.0, new=0.0, diff=no change
q14: prev= 533 ms, new= 459 ms, diff=1.16 faster ✔, tasks: prev=5.0, new=5.0, diff=no change
q15: prev= 732 ms, new= 772 ms, diff=1.05 slower ✖, tasks: prev=10.0, new=10.0, diff=no change
q16: prev= 553 ms, new= 464 ms, diff=1.19 faster ✔, tasks: prev=0.0, new=0.0, diff=no change
q17: prev= 664 ms, new= 937 ms, diff=1.41 slower ✖, tasks: prev=7.0, new=7.0, diff=no change
q18: prev= 750 ms, new= 886 ms, diff=1.18 slower ✖, tasks: prev=7.0, new=7.0, diff=no change
q19: prev= 487 ms, new= 492 ms, diff=1.01 slower ✖, tasks: prev=5.0, new=5.0, diff=no change
q20: prev= 721 ms, new= 828 ms, diff=1.15 slower ✖, tasks: prev=8.0, new=8.0, diff=no change
q21: prev=1104 ms, new=1171 ms, diff=1.06 slower ✖, tasks: prev=19.0, new=19.0, diff=no change
q22: prev= 372 ms, new= 365 ms, diff=1.02 faster ✔, tasks: prev=0.0, new=0.0, diff=no change
=== Comparing tpch/sf10_iceberg results 'datafusion-benchmark-base' [prev] with 'datafusion-benchmark-head' [new] === TASKS: prev=942.0, new=942.0, diff=no change (sum of per-query averages) TOTAL: prev=35827 ms, new=37205 ms, diff=1.04 slower ✖ Show full query output q1: prev=1384 ms, new=1806 ms, diff=1.30 slower ✖, tasks: prev=24.0, new=24.0, diff=no change
q2: prev=1316 ms, new=1500 ms, diff=1.14 slower ✖, tasks: prev=42.0, new=42.0, diff=no change
q3: prev=1939 ms, new=1764 ms, diff=1.10 faster ✔, tasks: prev=36.0, new=36.0, diff=no change
q4: prev=1102 ms, new=1110 ms, diff=1.01 slower ✖, tasks: prev=41.0, new=41.0, diff=no change
q5: prev=1724 ms, new=1767 ms, diff=1.02 slower ✖, tasks: prev=63.0, new=63.0, diff=no change
q6: prev= 917 ms, new= 950 ms, diff=1.04 slower ✖, tasks: prev=12.0, new=12.0, diff=no change
q7: prev=1902 ms, new=1841 ms, diff=1.03 faster ✔, tasks: prev=70.0, new=70.0, diff=no change
q8: prev=2416 ms, new=2408 ms, diff=1.00 faster ✔, tasks: prev=84.0, new=84.0, diff=no change
q9: prev=3215 ms, new=2847 ms, diff=1.13 faster ✔, tasks: prev=85.0, new=85.0, diff=no change
q10: prev=2535 ms, new=2002 ms, diff=1.27 faster ✔, tasks: prev=49.0, new=49.0, diff=no change
q11: prev=1173 ms, new=1341 ms, diff=1.14 slower ✖, tasks: prev=30.0, new=30.0, diff=no change
q12: prev=1309 ms, new=1448 ms, diff=1.11 slower ✖, tasks: prev=41.0, new=41.0, diff=no change
q13: prev=1121 ms, new=1016 ms, diff=1.10 faster ✔, tasks: prev=17.0, new=17.0, diff=no change
q14: prev=1305 ms, new=1452 ms, diff=1.11 slower ✖, tasks: prev=25.0, new=25.0, diff=no change
q15: prev=1905 ms, new=2198 ms, diff=1.15 slower ✖, tasks: prev=50.0, new=50.0, diff=no change
q16: prev= 744 ms, new= 741 ms, diff=1.00 faster ✔, tasks: prev=22.0, new=22.0, diff=no change
q17: prev=1589 ms, new=1562 ms, diff=1.02 faster ✔, tasks: prev=37.0, new=37.0, diff=no change
q18: prev=2023 ms, new=1925 ms, diff=1.05 faster ✔, tasks: prev=48.0, new=48.0, diff=no change
q19: prev=1448 ms, new=1741 ms, diff=1.20 slower ✖, tasks: prev=25.0, new=25.0, diff=no change
q20: prev=1782 ms, new=2401 ms, diff=1.35 slower ✖, tasks: prev=48.0, new=48.0, diff=no change
q21: prev=2402 ms, new=2792 ms, diff=1.16 slower ✖, tasks: prev=79.0, new=79.0, diff=no change
q22: prev= 576 ms, new= 593 ms, diff=1.03 slower ✖, tasks: prev=14.0, new=14.0, diff=no change
Verification and run detailsJob
Workload: Capacity: 12 PR-head configs: Other timings: Queue 1s · Dataset validation 1s · Total 15m 20s How to use the benchmark botPost a comment whose first non-empty line is:
For example:
Currently available datasets:
Supported instances and per-worker limits:
Each node runs one worker. The worker allocation reserves 1 vCPU and 4 GiB for Kubernetes and system processes, then makes the rest of the selected instance available to the benchmark. Limits: Only authorized users can enqueue jobs. Jobs run serially, the queue holds 20 active jobs, and each requester may have 3. Query selection is not supported. By default, every query uses 1 warmup and 5 measured iterations for both revisions. |
2783b8f to
5b4cce9
Compare
What changes
Make byte-balanced assignment the policy for eager Iceberg scans. There is no
greedy_file_assignmentflag or eager assignment-policy enum. Lazy work-unit feeds keep their current task-first round-robin assignment.One function assigns selected files largest-first to the distributed task with the fewest assigned bytes, then cycles through that task's local partitions. File path and range offset break equal-size ties; file count and task index break equal-load ties. The same function creates local eager groups and specializes them across distributed tasks.
Assignment now happens during planning on the source introduced by #803. Routing inspects those groups and the existing eager codec transports them unchanged. There is no feed dispatcher modification, assignment cache, additional codec, or separate accounting for index arrays and heaps. Retained descriptors remain charged to the query memory pool.
This assigns whole files to tasks, not workers. Candidate file bytes are scheduling estimates, not decoded memory or post-filter output. With one task, largest-first ordering is distributed round-robin across its local partitions; local partitions are not independently byte-balanced. Assignment runs synchronously in the existing scale-up hook. No performance claim.
Stack and review
Based on #787, on top of #803. Review only the base-to-head diff: the assignment function, its two source-construction call sites, and integration coverage.
The earlier feed-based implementation on #773 is preserved at 2783b8f, retained by the
archive/pr-788-on-773tag. Its #787 base is aa21dbe. Both histories remain available if we continue with #773 instead.Validation
Integration snapshots inspect the public assigned sources produced by the physical planner, covering skewed sizes, reversed manifest order, equal-size ties, local assignment, and empty scans. Existing routing tests verify that assignments survive serialization and execution; query results are compared with lazy scans. No new harness API exposes the assignment algorithm.
Passed: full Iceberg integration suite (64 tests and one doctest), all-target Clippy with warnings denied, formatting, and
git diff --check. Snapshots were reviewed and the full suite ran with snapshot updates disabled.