[CHORE] Swordfish refactors #3256

colin-ho · 2024-11-11T02:04:43Z

There's a couple outstanding issues / inefficiencies / ugliness in the swordfish code. I originally intended of breaking these up into smaller PRs, but during the process of trying to split it up, I realized that all the changes are quite intertwined, and it may be easier on the reviewer to just see all of them in one. That being said, I'll try my best to explain all the changes and rationale in detail.

Problems

The PipelineChannel abstraction doesn't play well with streaming sinks, in fact, in only really works for intermediate ops. This is because the PipelineChannel assumes sending/receiving is round robin, but streaming sinks can send data from the workers, as well as after the finalize.
Pipeline nodes currently send across both data and probe tables in the same channel, which can be confusing. Furthermore, the logic of dispatching probe tables and data is also different, adding more complexity. Probe tables need to be broadcasted to all workers, while data should only be distributed to a single worker.
If an operator does not require input to be ordered, it shouldn't dispatch work to workers in round robin, it should dispatch to any available worker.
Some operators don't do work in their execute or finalize methods. These don't need to be spawned on the compute runtime.

Proposed Changes

Pipeline nodes should just send data across, via a simple Receiver<Arc<Micropartition>>.
Probe tables should be sent separately from data, via a ProbeStateBridge.
Implement an unordered dispatcher, an optimization for if ordering is not required. This uses loole channels, which are multi-producer multi-consumer. For consistency, use loole channels across all swordfish. In the future we can think about implementing our custom channels. https://users.rust-lang.org/t/loole-a-safe-async-sync-multi-producer-multi-consumer-channel/113785
Give the compute runtime to the operators, and let them decide whether they want to spawn.

codspeed-hq · 2024-11-11T02:12:19Z

CodSpeed Performance Report

Merging #3256 will improve performances by 45.45%

_{Comparing colin/probe-state-bridge (e751f61) with main (711e862)}

Summary

⚡ 2 improvements
✅ 15 untouched benchmarks

Benchmarks breakdown

	Benchmark	`main`	`colin/probe-state-bridge`	Change
⚡	`test_iter_rows_first_row[100 Small Files]`	269 ms	237.4 ms	+13.32%
⚡	`test_show[100 Small Files]`	22.3 ms	15.3 ms	+45.45%

codecov · 2024-11-11T02:30:14Z

Codecov Report

Attention: Patch coverage is 97.11316% with 25 lines in your changes missing coverage. Please review.

Project coverage is 77.59%. Comparing base (711e862) to head (e751f61).
Report is 10 commits behind head on main.

Files with missing lines	Patch %	Lines
src/daft-local-execution/src/dispatcher.rs	93.85%	7 Missing ⚠️
...local-execution/src/sinks/outer_hash_join_probe.rs	95.70%	7 Missing ⚠️
...ecution/src/intermediate_ops/actor_pool_project.rs	84.00%	4 Missing ⚠️
.../daft-local-execution/src/sinks/hash_join_build.rs	93.02%	3 Missing ⚠️
src/daft-local-execution/src/run.rs	84.61%	2 Missing ⚠️
.../src/intermediate_ops/anti_semi_hash_join_probe.rs	97.36%	1 Missing ⚠️
...-execution/src/intermediate_ops/intermediate_op.rs	98.41%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3256      +/-   ##
==========================================
+ Coverage   77.50%   77.59%   +0.08%     
==========================================
  Files         666      666              
  Lines       81335    81621     +286     
==========================================
+ Hits        63041    63331     +290     
+ Misses      18294    18290       -4

Files with missing lines	Coverage Δ
src/common/runtime/src/lib.rs	`90.78% <100.00%> (ø)`
src/daft-local-execution/src/channel.rs	`98.14% <100.00%> (-0.36%)`	⬇️
...-local-execution/src/intermediate_ops/aggregate.rs	`100.00% <100.00%> (ø)`
...ft-local-execution/src/intermediate_ops/explode.rs	`100.00% <100.00%> (ø)`
...aft-local-execution/src/intermediate_ops/filter.rs	`100.00% <100.00%> (ø)`
...tion/src/intermediate_ops/inner_hash_join_probe.rs	`99.23% <100.00%> (+3.46%)`	⬆️
...ft-local-execution/src/intermediate_ops/project.rs	`100.00% <100.00%> (ø)`
...aft-local-execution/src/intermediate_ops/sample.rs	`100.00% <100.00%> (ø)`
...ft-local-execution/src/intermediate_ops/unpivot.rs	`100.00% <100.00%> (ø)`
src/daft-local-execution/src/lib.rs	`95.29% <100.00%> (+2.43%)`	⬆️
... and 18 more

... and 3 files with indirect coverage changes

colin-ho · 2024-11-11T07:53:19Z

src/daft-local-execution/src/dispatcher.rs

+
+/// A dispatcher that distributes morsels to workers in an unordered fashion.
+/// Used if the operator does not require maintaining the order of the input.
+pub(crate) struct UnorderedDispatcher {


Script that tests for uneven runtimes across tasks. With the UnorderedDispatcher, performance is considerably increased.

daft.context.set_execution_config(default_morsel_size=1) @daft.udf(return_dtype=daft.DataType.int64()) def do_work(x): x = x.to_pylist() # my laptop has 12 cores, so 12 workers will be spawned to run this udf. if x[0] % 12 == 0: print("doing a lot of work on ", x) time.sleep(1) else: print("doing a little work on ", x) time.sleep(0.1) return x df = daft.from_pydict({"a": [i for i in range(48)]}).with_column("b", do_work(col("a"))).agg(col("b").sum()) df.collect()

colin-ho · 2024-11-12T10:58:02Z

src/daft-local-execution/src/lib.rs

+/// The `OperatorOutput` enum represents the output of an operator.
+/// It can be either `Ready` or `Pending`.
+/// If the output is `Ready`, the value is immediately available.
+/// If the output is `Pending`, the value is not yet available and a `RuntimeTask` is returned.
+pub(crate) enum OperatorOutput<T> {
+    Ready(T),
+    Pending(RuntimeTask<T>),
+}


This OperatorOutput allows operators to spawn their own tasks on the compute runtime if needed, or if no compute is necessary, return whatever result they have. This also allows hash join probes to await on the probe tables within the execute method.

colin-ho · 2024-11-12T11:00:37Z

src/daft-local-execution/src/run.rs

@@ -46,6 +47,13 @@ impl LocalPartitionIterator {
 #[cfg_attr(feature = "python", pyclass(module = "daft.daft"))]
 pub struct NativeExecutor {
    local_physical_plan: Arc<LocalPhysicalPlan>,
+    cancel: CancellationToken,


The tasks spawned on the execution runtime handle currently only abort if they try a send and see that the channel is closed (this is something we did in the native runner pr). However, they should also just abort if the native executor gets dropped.

colin-ho · 2024-11-12T11:02:15Z

src/daft-local-execution/src/sinks/hash_join_build.rs

+pub(crate) struct ProbeStateBridge {
+    inner: OnceLock<Arc<ProbeState>>,
+    notify: tokio::sync::Notify,
+}


https://docs.rs/tokio/latest/tokio/sync/struct.OnceCell.html doesn't work cuz there's no get method that is async.

Opted for just a manual implementation.

src/daft-local-execution/src/channel.rs

src/daft-local-execution/src/intermediate_ops/actor_pool_project.rs

src/daft-local-execution/src/lib.rs

samster25 · 2024-11-13T05:43:36Z

src/daft-local-execution/src/dispatcher.rs

+///
+/// It returns a vector of receivers (one per worker) that will receive the morsels.
+///
+/// Implementations must spawn a task on the runtime handle that reads from the


I think it would make more sense to name this:

trait DispatcherBuilder { fn spawn_dispatcher(...) -> SpawnedDispatcherResult; } struct SpawnedDispatcherResult { receivers: Vec<Receiver<Arc<MicroPartition>>>, dispatch_handle: SpawnedTask }

I think my key confusion was that dispatch sounds like an action that would happen on every tick rather than something that happens once.

Gotcha, made it more explicit that it should spawn a dispatching task. Also implemented the SpawnedDispatcherResult, then the caller can await on it themselves.

…tor output

samster25

🔥

samster25 · 2024-11-14T06:30:50Z

src/daft-local-execution/src/lib.rs

+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.get_mut();
+        Pin::new(&mut this.0).poll(cx).map(|r| r.context(JoinSnafu))


actually this looks a bit sketchy! Why do we have to Pin::new this?

The poll method on a JoinHandle accepts Pin<&mut Self>, but self.0 is not pinned even though self is pinned, so i pinned it.

I did more reading, and I found pin_project which allows you to project the pin on a struct into it's fields, and don't need to Pin::new. Technically i think it was fine to Pin::new it since the joinhandle is Unpin? Note to self: Need to read up more on pin and unpin though

Colin Ho and others added 5 commits November 7, 2024 16:20

remove pipeline channel

6b364c2

switch to loole in separate pr

e78ff42

use dispatcher in streaming sink and intermediate op

4bfddff

probe state bridge

3943e0f

cleanup and add comments

a2a1339

github-actions bot added the chore label Nov 11, 2024

cap scan tasks

8508af4

colin-ho commented Nov 11, 2024

View reviewed changes

undo for now

a5537f4

colin-ho marked this pull request as ready for review November 11, 2024 08:48

colin-ho requested a review from samster25 November 11, 2024 14:15

colin-ho commented Nov 12, 2024

View reviewed changes

samster25 reviewed Nov 13, 2024

View reviewed changes

Colin Ho added 7 commits November 13, 2024 13:53

Merge branch main into colin/probe-state-bridge

18a9e97

operators accept value, no clone bound on channel, impl fut for opera…

e20bf09

…tor output

dispatcher_spawner

d5032be

dispatcher enum

b506c6b

dispatcher comments

922c133

dispatch spawner

c6f0bda

spawned dispatcher result

6583ef8

colin-ho requested a review from samster25 November 14, 2024 06:26

samster25 approved these changes Nov 14, 2024

View reviewed changes

samster25 reviewed Nov 14, 2024

View reviewed changes

Colin Ho added 2 commits November 14, 2024 16:21

Merge branch 'main' into colin/probe-state-bridge

27b025d

pin project

e751f61

colin-ho merged commit ca8f25d into main Nov 15, 2024
44 checks passed

colin-ho deleted the colin/probe-state-bridge branch November 15, 2024 16:33

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[CHORE] Swordfish refactors #3256

[CHORE] Swordfish refactors #3256

colin-ho commented Nov 11, 2024 •

edited

Loading

codspeed-hq bot commented Nov 11, 2024 •

edited

Loading

codecov bot commented Nov 11, 2024 •

edited

Loading

colin-ho Nov 11, 2024

colin-ho Nov 12, 2024

colin-ho Nov 12, 2024

colin-ho Nov 12, 2024

samster25 Nov 13, 2024

colin-ho Nov 13, 2024

samster25 left a comment

samster25 Nov 14, 2024

colin-ho Nov 14, 2024 •

edited

Loading

[CHORE] Swordfish refactors #3256

[CHORE] Swordfish refactors #3256

Conversation

colin-ho commented Nov 11, 2024 • edited Loading

Problems

Proposed Changes

codspeed-hq bot commented Nov 11, 2024 • edited Loading

Merging #3256 will improve performances by 45.45%

Summary

Benchmarks breakdown

codecov bot commented Nov 11, 2024 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

samster25 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

colin-ho Nov 14, 2024 • edited Loading

Choose a reason for hiding this comment

colin-ho commented Nov 11, 2024 •

edited

Loading

codspeed-hq bot commented Nov 11, 2024 •

edited

Loading

codecov bot commented Nov 11, 2024 •

edited

Loading

colin-ho Nov 14, 2024 •

edited

Loading