Is your feature request related to a problem or challenge?
Consumers that execute many partitions of a retained physical plan may need to report metrics separately for each completed partition. Today, ExecutionPlan::metrics() returns a snapshot of all registered metrics, so callers first clone all metric handles and then filter by Metric::partition().
When completed partitions' metrics remain registered while other partitions are still running, reporting every partition can require O(N²) cloning and filtering work for N partitions with a fixed number of metrics each. This is relevant to Comet's opt-in shared physical plans and other consumers that retain a plan across partition executions.
Describe the solution you'd like
Add a partition-specific metrics API:
fn metrics_for_partition(&self, partition: usize) -> Option<MetricsSet>
Provide a default implementation on ExecutionPlan that filters the existing metrics() result, preserving compatibility for external implementations. Add indexed partition lookup to ExecutionPlanMetricsSet and override the method in container-backed operators so retrieval clones only the selected partition's metric handles, with expected O(1 + matching metrics) work excluding lock contention.
The API should:
- Select exactly metrics whose partition is
Some(partition); plan-wide metrics with None remain available through metrics().
- Preserve registration order, duplicate names and labels, and shared metric values.
- Observe later registrations on subsequent calls while keeping each snapshot's membership fixed.
- Return an empty snapshot for an unknown partition when metrics are supported, preserving
None for implementations without metrics.
- Treat partition IDs as operator-local, without traversing children or remapping IDs.
Datasource, sink, and FFI adapters should forward the request to the underlying implementation so callers can benefit from indexed lookup across these boundaries.
Describe alternatives you've considered
Filtering metrics() at the call site preserves the cost of cloning all registered handles. Caching a filtered snapshot can miss metrics registered later during execution. Maintaining a separate registry in each consumer duplicates bookkeeping and makes it harder to preserve operator metrics semantics.
An index adds memory and registration overhead, and constructing a registry from an existing MetricsSet requires building the index. Benchmarks should assess these costs alongside partition lookup and full snapshots.
Additional context
Related Comet discussions:
A useful regression test is a real streaming → filter → projection tree with multiple partitions, where one partition remains active while the others finish and report metrics. Benchmarks should keep the requested partition's metric count fixed while increasing the number of registered partitions.
The motivation is partition reporting overhead in retained plans; this is not a claim of a general DataFusion query performance regression.
Is your feature request related to a problem or challenge?
Consumers that execute many partitions of a retained physical plan may need to report metrics separately for each completed partition. Today,
ExecutionPlan::metrics()returns a snapshot of all registered metrics, so callers first clone all metric handles and then filter byMetric::partition().When completed partitions' metrics remain registered while other partitions are still running, reporting every partition can require O(N²) cloning and filtering work for N partitions with a fixed number of metrics each. This is relevant to Comet's opt-in shared physical plans and other consumers that retain a plan across partition executions.
Describe the solution you'd like
Add a partition-specific metrics API:
Provide a default implementation on
ExecutionPlanthat filters the existingmetrics()result, preserving compatibility for external implementations. Add indexed partition lookup toExecutionPlanMetricsSetand override the method in container-backed operators so retrieval clones only the selected partition's metric handles, with expected O(1 + matching metrics) work excluding lock contention.The API should:
Some(partition); plan-wide metrics withNoneremain available throughmetrics().Nonefor implementations without metrics.Datasource, sink, and FFI adapters should forward the request to the underlying implementation so callers can benefit from indexed lookup across these boundaries.
Describe alternatives you've considered
Filtering
metrics()at the call site preserves the cost of cloning all registered handles. Caching a filtered snapshot can miss metrics registered later during execution. Maintaining a separate registry in each consumer duplicates bookkeeping and makes it harder to preserve operator metrics semantics.An index adds memory and registration overhead, and constructing a registry from an existing
MetricsSetrequires building the index. Benchmarks should assess these costs alongside partition lookup and full snapshots.Additional context
Related Comet discussions:
A useful regression test is a real streaming → filter → projection tree with multiple partitions, where one partition remains active while the others finish and report metrics. Benchmarks should keep the requested partition's metric count fixed while increasing the number of registered partitions.
The motivation is partition reporting overhead in retained plans; this is not a claim of a general DataFusion query performance regression.