adriangb opened a new issue, #26566:
URL: https://github.com/apache/datafusion/issues/26566
### Is your feature request related to a problem or challenge?
When a query fails with `ResourcesExhausted`, `TrackConsumersPool` lists the
largest consumers. `TrackConsumersPool::metrics()` (#21146) returns all of
them. But a consumer name only gives the operator type and the partition, for
example `FinalHashAggregateStream[4]`. If a plan has two operators of the same
type, a user or a tool cannot tell which operator owns which consumer.
Example. This query has two hash aggregations:
```sql
WITH hourly AS (
SELECT date_trunc('hour', ts) AS bucket, service, count(*) AS n
FROM events
GROUP BY 1, 2
)
SELECT service, sum(n) FROM hourly GROUP BY service;
```
Both aggregations register `PartialHashAggregateStream[i]` and
`FinalHashAggregateStream[i]`. If the query runs out of memory, the error and
`metrics()` give names that do not identify one of the two aggregations. A tool
that wants to show memory use on each node of the physical plan (similar to
`EXPLAIN ANALYZE`) cannot join the consumers to the plan.
### Describe the solution you'd like
Add an optional plan-node identity to `MemoryConsumer`, and expose it in
`MemoryConsumerMetrics`:
```rust
impl MemoryConsumer {
/// Records the plan node that owns this consumer.
pub fn with_plan_node(self, node: &dyn Any) -> Self;
pub fn plan_node(&self) -> Option<usize>;
}
pub struct MemoryConsumerMetrics {
// existing fields
pub plan_node: Option<usize>,
}
```
The value is the address of the `ExecutionPlan` instance that called
`execute`. A caller that holds the physical plan can compare it with
`Arc::as_ptr` for each node. While the caller holds the plan, no node is freed,
so an address cannot be used again during the comparison.
`datafusion-execution` does not depend on `datafusion-physical-plan`, so the
field cannot hold a typed reference to the plan. Also, `execute` takes `&self`,
so an operator has no `Arc` to itself.
The built-in operators set the identity where they create their consumers,
for example:
```rust
MemoryConsumer::new(format!("FinalHashAggregateStream[{partition}]"))
.with_plan_node(self)
```
The field is optional. Custom operators that do not set it work as before.
### Describe alternatives you've considered
- Put a node identifier into the consumer name. This needs a stable node id,
which DataFusion does not have. It also turns the name into an encoding that
callers must parse.
- Store an `Arc` or `Weak` to the plan node. This needs a change to the
`ExecutionPlan::execute` signature and an `Any` type that breaks the crate
dependency order. It gives no more than an identity comparison.
- Guess the node from the operator type, the partition count and the
registration order. This is not reliable.
### Additional context
I can open a PR. If you prefer smaller changes, I can split it into the API
and pool change first, and the operator call sites second. A later change could
also print the node in the `report_top` error message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
