Avoid per-file metric registration in `ParquetSource`

[mitchsw]

Is your feature request related to a problem or challenge?

I am benchmarking a query plan with many partitions and a scan over >12,000 Parquet files. Each file opened by the default ParquetSource registers 45 metrics into the one ExecutionPlanMetricsSet shared by the whole DataSourceExec. This results in half a million synchronous mutex acquisitions and metric registrations.

Dial9 schedule/off-CPU profiling of high-file-count scans consistently identified this mutex as the dominant application-level blocking source. The captured stacks ran from Parquet file opening through MetricBuilder::build and ExecutionPlanMetricsSet::register, where workers entered futex_wait while acquiring the shared parking_lot::Mutex. Because registration uses a synchronous mutex on Tokio runtime threads, concurrent file opens create a lock convoy: worker threads park, runnable tasks accumulate in the Tokio queue, and CPU remains underutilized.

This per-file detail is not used by normal queries or EXPLAIN ANALYZE, which aggregates metrics by name and discards filename labels. The filename-labelled metrics are only exposed through EXPLAIN ANALYZE VERBOSE or direct inspection of unaggregated metrics.

Describe the solution you'd like

Add a metrics cardinality ConfigOption:

datafusion.execution.metrics_cardinality = compact | verbose

Default to compact.

With compact cardinality, the ParquetSource would:

• Register one metric set per execution partition
• Omit filename labels
• Share the handles across files and replacement readers

With verbose cardinality, the ParquetSource would preserve per-file filename-labelled metrics.

This changes metric registration growth from O(files) to O(partitions).

To maintain existing behavior, AnalyzeExec will handle EXPLAIN ANALYZE VERBOSE by mutating the child TaskContext enabling verbose cardinality (no other ExecutionPlan edits its ConfigOptions, but it felt appropriate here). Other use cases that directly consume per-file metrics could also explicitly configure it.

Describe alternatives you've considered

A larger metrics-system redesign could introduce lock-free or sharded registration, typed metric scopes, cardinality limits, or execution-local registries. That may be worthwhile, but it is substantially broader than the Parquet contention problem.

Additional context

No response

[mitchsw]

I have a draft PR I am working on, if this solution shape sounds acceptable

[mitchsw]

take

[mitchsw]

Here is an illustrative dial9 trace of a query on an ~idle server. At the start of the query, we see the workers on long polls, including blocking calls. All the blocking calls are all futex_wait on stacks containing this frame: https://github.com/apache/datafusion/blob/54.0.0/datafusion/physical-expr-common/src/metrics/mod.rs#L492-L494.

[2010YOUY01]

This changes metric registration growth from O(files) to O(partitions).

If we try to do O(partitions), without this new config, what are the blockers? I understand there might be some larger refactor, but it might be doable, and we don't have to introduce extra complexity (new config) as an intermediate step.

[mitchsw]

Thanks for the suggestion! I'm not 100% sure I understand your proposal, but attempting to answer:

1. We can remove per-file metrics entirely, making registration O(partitions) without adding a new config. However, that would break EXPLAIN ANALYZE VERBOSE, which currently outputs per-file metrics.
2. We can retain the O(files) registrations but replace the query-wide ExecutionPlanMetricsSet map and mutex with a two-tier structure: ExecutionPlan -> partition -> file. Each partition would have its own metric set and lock, which should eliminate the query-wide contention that dominated my profile. However, we would still allocate and maintain per-file metrics that are never read in the common case.

If we want compact metrics by default while retaining per-file metrics for verbose analysis, we need some execution-time signal to select between the two behaviors. I am happy to explore alternative 2 if the config is undesirable.

[2010YOUY01]

I see. I think the issue of adding a new metric verbosity config is that, other metrics implementation would have to understand a new global config, and decide how to put metrics to different tier.

How about adding a data source (or parquet) specific config like

datafusion.execution.parquet.show_per_file_metrics