Introduce morsel-driven Parquet scan

[Dandandan]

Is your feature request related to a problem or challenge?

• Part of [EPIC] Make DataFusion the top of the ClickBench Parquet leaderboard #18489

Current parelllization of Parquet scan is bounded by the thread that has the most data / is the slowest to execute, which means in the case of data skew (driven by either larger partitions or less selective filters during pruning / filter pushdown..., variable object store latency), the parallelism will be significantly limited.

Describe the solution you'd like

We can change the strategy by morsel-driven parallelism like described in https://db.in.tum.de/~leis/papers/morsels.pdf.

Doing so is faster for a lot of queries, when there is an amount of skew (such as clickbench) and we have enough row filters to spread out the work.
For clickbench_partitioned / clickbench_pushdown it seems up to ~2x as fast for some queries, on a 10 core machine.

Describe alternatives you've considered

No response

Additional context

No response

[alamb]

Here are some thoughts (it is getting hard to keep track of what is going on on the PR #20481). I have been looking at #20481 in order to figure out how we can most smoothly structure the code / ideas in this PR into the existing code of DataFusion

Here is my summary of the architectural changes to #20481:

• The FileStream is updated to know about "Morsels"
• A new parallel API is added to the FileOpener API : "morselize" that takes a single PartitionedFile and breaks it into Morsels (which are smaller PartitionedFiles)

Challenges I see with this design (all can be overcome with some more code):

1. As there is a parallel API in FileOpener that has a parallel code path this may be hard to test
2. It may also be hard to apply the morsel idea to non parquet paths (even though the idea is absolutely applicable). However, it might also be ok
3. It uses the "extensions" field to stash the morsels, which I think will break some downstream users who use that for other things (like trace context, for example)
4. It is not clear to me how we will add other features like configurable IO prefetching without hard coding more into FileOpener

Ideas going forward:

1. Rather than add a parallel API to the FileOpener API, I think we should try and make Morsels explicit in the pipeline somehow (perhaps via a Morselizer and rename FileOpener toMorselOpener trait 🤔 )
2. Generalize the existing code that tries to statically split files based on byte offsets -- perhaps create morsels there as well.
3. Isolate the work stealing to a new structure / trait (as we will need to turn it off for some plans, e.g. those that require sortedness)

I am going to try and prototype what a more explicit "morselizer" / "FileOpener' might look like

[alamb]

Here is a document that @Dandandan has been working on about morsels

https://docs.google.com/document/d/1yZvEXkgkIdOEZU87T8GzanQu8udbsaRq2yqetVmVDHk/edit?tab=t.0#heading=h.nvj05uxaozsr

[matthewmturner]

Here is a document that @Dandandan has been working on about morsels

@alamb Would you be able to make the document public? thank you @Dandandan

[Dandandan]

Given the interest, I am thinking of creating a formal proposal for this - would that make sense?

[alamb]

Given the interest, I am thinking of creating a formal proposal for this - would that make sense?

Yes, I think so

I personally thought you did a good job motivating the project in your meetup slides last week in stockholm: #18429 (comment)

Such a writeup would also be a great way to get a head start on a blog 😆

[Dandandan]

Ok - I'll try write something up (based on the slides and current insights)!