feat: Create datafusion-distributed crate with shuffle reader/writer

[andygrove]

Which issue does this PR close?

N/A

Rationale for this change

DataFusion is a great foundation for distributed systems, so let's make it easier to build distributed systems with DataFusion.

What changes are included in this PR?

New datafusion-distributed crate containing the shuffle reader and writer from Ballista. I would also like to add the distributed query planner into this crate in a future PR.

We could also consider making improvements to the shuffle mechanism based on work happening in Comet.

Are these changes tested?

There will be unit tests once this PR is ready for review.

Are there any user-facing changes?

No

[andygrove]

@thinkharderdev @Dandandan @avantgardnerio Just fyi and wanted to get your opinion on whether this is useful for you

[thinkharderdev]

@thinkharderdev @Dandandan @avantgardnerio Just fyi and wanted to get your opinion on whether this is useful for you

This seems like a good idea although I'm not sure we would use it directly as we have some fairly specific customizations we've added to ShuffleReader/ShuffleWriter internally

Edit: Adding the distributed scheduler to this create would be great though and something we'd definitely be interested in using and contributing to, especially if if can abstract out the concrete implementation of actually shuffling data between stages

[andygrove]

Edit: Adding the distributed scheduler to this create would be great though and something we'd definitely be interested in using and contributing to, especially if if can abstract out the concrete implementation of actually shuffling data between stages

I am also interested in this type of abstraction. I was thinking along the lines of having the planner insert QueryStageExec instead of ShuffleWriterExec. QueryStageExec would not have any execution logic. Responsibility for execution can be left to users of this planner.

[alamb]

FWIW we (InfluxData) would likely (never) end up using the shuffle reader / writer, nor a distributed query planner (we would have our own).

From my perspective if there is more than one user of this code (e.g more than Ballista) then it makes sense to put it in datafusion. If there is realistically only one user of this code it probably doesn't belong here

[andygrove]

From my perspective if there is more than one user of this code (e.g more than Ballista) then it makes sense to put it in datafusion. If there is realistically only one user of this code it probably doesn't belong here

That makes sense, but there is also a chicken and egg situation with this. Let's see if anyone comments in favor of this, and if not, I will close the PR.

[Epicism]

I'm late to the party, but this would be helpful to build distributed services based on different Datafusion variants (e.g., LanceDB vs vanilla Datafusion.).

[alamb]

Another potential option is to make a datafusion-shuffle repo in https://github.com/datafusion-contrib that we could prototype / collaborate what this would look like (and is lower overhead / commitment than a new crate in the core)

We could revisit the question of bringing the shuffle into the core once the exact shape and scope of the code was known