[Spark] Parallel stats collection within each partition

[fred-db]

Which Delta project/connector is this regarding?

• Spark
• Standalone
• Flink
• Kernel
• Other (fill in here)

Description

• Do stats collection in parallel within a partition instead of sequentially to reduce the time spent idle waiting for network requests to go through while fetching the file status or the parquet footers from cloud store
• Using global threadpool on each executor to do parallel stats collection
• Code to partition the dataset before collecting stats to increase the achievable throughput

How was this patch tested?

• Existing UTs should cover the correct collection of statistics.

Does this PR introduce any user-facing changes?

No

[felipepessoto]

Is this only improving the scenario where user recompute stats for existing files?

I'm asking because generating stats takes considerable amount of time when inserting data, in my experiments around 20% of overhead in comparison when stats.collect is disabled.

[vkorukanti]

Is this only improving the scenario where user recompute stats for existing files?

I'm asking because generating stats takes considerable amount of time when inserting data, in my experiments around 20% of overhead in comparison when stats.collect is disabled.

AFAIK the method computeStats is used only when updating/deleting table with deletion vectors (DVs). It is a must to recompute stats for files that don't have stats when the AddFile contains a DV.

[vkorukanti]

The package name seems to be wrong. Are you missing a change?

[fred-db]

Ah yes, I am depending on this PR #2154 being merged first. I didn't want to include the changes of it in this PR, as it would make the code review harder. But once the other PR is merged, this should match up.

[vkorukanti]

The thread count is a bit high. Have you seen any issues with your testing?

[fred-db]

I didn't see any issues when trying it out on a 4 worker cluster. I think you could run into some throttling issues if the cluster you use is very large and you have to collect stats for DVs for many files. But then you can always increase the number of files per partition for stats collection, that should reduce throttling issues.

[vkorukanti]

lgtm.