Skip to content

Add efficient partition-specific metrics access to ExecutionPlan #25582

Description

@viirya

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.

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

enhancementNew feature or request

Type

No type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions