You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "ShreyeshArangath (via GitHub)" <gi...@apache.org> on 2026/02/15 01:52:13 UTC
[PR] feat: add Python bindings for accessing ExecutionMetrics [datafusion-python]
ShreyeshArangath opened a new pull request, #1381:
URL: https://github.com/apache/datafusion-python/pull/1381
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #1379
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
Today, DataFusion Python only exposes execution metrics through formatted console output via `explain(analyze=True)`.
This makes it difficult to programmatically inspect execution behavior.
There is currently no structured API to access per-operator metrics such as output_rows, elapsed_compute, spill_count and other runtime metrics collected during execution.
This PR introduces a structured Python interface for execution metrics, mirroring the Rust API in `datafusion::physical_plan::metrics`.
# What changes are included in this PR?
- Added plan caching to `PyDataFrame` so the physical plan used during execution is retained and available for metrics access.
- Kept the `metrics()` method and added `collect_metrics()` helper to walk the execution plan tree and aggregate metrics from all operators.
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
Users can now programmatically access execution metrics
```python
df = ctx.sql("SELECT * FROM t WHERE x > 1")
df.collect()
plan = df.execution_plan()
metrics = plan.collect_metrics()
for operator_name, metrics_set in metrics:
print(f"{operator_name}: {metrics_set.output_rows} rows")
```
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer merged PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#discussion_r2855537956
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
+ """
+
+ def __init__(self, raw: df_internal.MetricsSet) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ def metrics(self) -> list[Metric]:
+ """Return all individual metrics in this set."""
+ return [Metric(m) for m in self._raw.metrics()]
+
+ @property
+ def output_rows(self) -> int | None:
+ """Sum of output_rows across all partitions."""
+ return self._raw.output_rows()
+
+ @property
+ def elapsed_compute(self) -> int | None:
+ """Sum of elapsed_compute across all partitions, in nanoseconds."""
+ return self._raw.elapsed_compute()
+
+ @property
+ def spill_count(self) -> int | None:
+ """Sum of spill_count across all partitions."""
+ return self._raw.spill_count()
+
+ @property
+ def spilled_bytes(self) -> int | None:
+ """Sum of spilled_bytes across all partitions."""
+ return self._raw.spilled_bytes()
+
+ @property
+ def spilled_rows(self) -> int | None:
+ """Sum of spilled_rows across all partitions."""
+ return self._raw.spilled_rows()
+
+ def sum_by_name(self, name: str) -> int | None:
+ """Return the sum of metrics matching the given name."""
+ return self._raw.sum_by_name(name)
+
+ def __repr__(self) -> str:
+ """Return a string representation of the metrics set."""
+ return repr(self._raw)
+
+
+class Metric:
+ """A single execution metric with name, value, partition, and labels."""
+
+ def __init__(self, raw: df_internal.Metric) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ @property
+ def name(self) -> str:
+ """The name of this metric (e.g. ``output_rows``)."""
+ return self._raw.name
+
+ @property
+ def value(self) -> int | None:
+ """The numeric value of this metric, or None for non-numeric types."""
+ return self._raw.value
+
+ @property
+ def partition(self) -> int | None:
+ """The partition this metric applies to, or None if global."""
Review Comment:
Do you know if this partition is a hash or an index or something else? I am trying to figure out how a user makes use of this information.
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
+ """
+
+ def __init__(self, raw: df_internal.MetricsSet) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ def metrics(self) -> list[Metric]:
+ """Return all individual metrics in this set."""
+ return [Metric(m) for m in self._raw.metrics()]
+
+ @property
+ def output_rows(self) -> int | None:
+ """Sum of output_rows across all partitions."""
+ return self._raw.output_rows()
+
+ @property
+ def elapsed_compute(self) -> int | None:
+ """Sum of elapsed_compute across all partitions, in nanoseconds."""
+ return self._raw.elapsed_compute()
+
+ @property
+ def spill_count(self) -> int | None:
+ """Sum of spill_count across all partitions."""
+ return self._raw.spill_count()
+
+ @property
+ def spilled_bytes(self) -> int | None:
+ """Sum of spilled_bytes across all partitions."""
+ return self._raw.spilled_bytes()
+
+ @property
+ def spilled_rows(self) -> int | None:
+ """Sum of spilled_rows across all partitions."""
+ return self._raw.spilled_rows()
+
+ def sum_by_name(self, name: str) -> int | None:
+ """Return the sum of metrics matching the given name."""
+ return self._raw.sum_by_name(name)
+
+ def __repr__(self) -> str:
+ """Return a string representation of the metrics set."""
+ return repr(self._raw)
+
+
+class Metric:
+ """A single execution metric with name, value, partition, and labels."""
+
+ def __init__(self, raw: df_internal.Metric) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ @property
+ def name(self) -> str:
+ """The name of this metric (e.g. ``output_rows``)."""
+ return self._raw.name
+
+ @property
+ def value(self) -> int | None:
+ """The numeric value of this metric, or None for non-numeric types."""
+ return self._raw.value
+
+ @property
+ def partition(self) -> int | None:
+ """The partition this metric applies to, or None if global."""
+ return self._raw.partition
+
+ def labels(self) -> dict[str, str]:
+ """Return the labels associated with this metric."""
Review Comment:
Something to include in user documentation is an example of these labels.
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
+ """
+
+ def __init__(self, raw: df_internal.MetricsSet) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ def metrics(self) -> list[Metric]:
+ """Return all individual metrics in this set."""
+ return [Metric(m) for m in self._raw.metrics()]
+
+ @property
+ def output_rows(self) -> int | None:
+ """Sum of output_rows across all partitions."""
+ return self._raw.output_rows()
+
+ @property
+ def elapsed_compute(self) -> int | None:
+ """Sum of elapsed_compute across all partitions, in nanoseconds."""
+ return self._raw.elapsed_compute()
+
+ @property
+ def spill_count(self) -> int | None:
+ """Sum of spill_count across all partitions."""
+ return self._raw.spill_count()
+
+ @property
+ def spilled_bytes(self) -> int | None:
+ """Sum of spilled_bytes across all partitions."""
+ return self._raw.spilled_bytes()
+
+ @property
+ def spilled_rows(self) -> int | None:
+ """Sum of spilled_rows across all partitions."""
+ return self._raw.spilled_rows()
+
+ def sum_by_name(self, name: str) -> int | None:
+ """Return the sum of metrics matching the given name."""
+ return self._raw.sum_by_name(name)
+
+ def __repr__(self) -> str:
+ """Return a string representation of the metrics set."""
+ return repr(self._raw)
+
+
+class Metric:
+ """A single execution metric with name, value, partition, and labels."""
+
+ def __init__(self, raw: df_internal.Metric) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ @property
+ def name(self) -> str:
+ """The name of this metric (e.g. ``output_rows``)."""
+ return self._raw.name
+
+ @property
+ def value(self) -> int | None:
+ """The numeric value of this metric, or None for non-numeric types."""
+ return self._raw.value
Review Comment:
Is it not possible to give values for non-numeric metrics?
##########
src/dataframe.rs:
##########
@@ -802,7 +823,13 @@ impl PyDataFrame {
}
/// Get the execution plan for this `DataFrame`
+ ///
+ /// If the DataFrame has already been executed (e.g. via `collect()`),
+ /// returns the cached plan which includes populated metrics.
fn execution_plan(&self, py: Python) -> PyDataFusionResult<PyExecutionPlan> {
+ if let Some(plan) = self.last_plan.lock().as_ref() {
+ return Ok(PyExecutionPlan::new(Arc::clone(plan)));
+ }
let plan = wait_for_future(py, self.df.as_ref().clone().create_physical_plan())??;
Ok(plan.into())
Review Comment:
If you go the route of using the existing `last_plan` for `collect()` like in my other comment then I think you could set it here just like you do in collect().
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
+ """
+
+ def __init__(self, raw: df_internal.MetricsSet) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ def metrics(self) -> list[Metric]:
+ """Return all individual metrics in this set."""
+ return [Metric(m) for m in self._raw.metrics()]
+
+ @property
+ def output_rows(self) -> int | None:
+ """Sum of output_rows across all partitions."""
+ return self._raw.output_rows()
+
+ @property
+ def elapsed_compute(self) -> int | None:
+ """Sum of elapsed_compute across all partitions, in nanoseconds."""
+ return self._raw.elapsed_compute()
+
+ @property
+ def spill_count(self) -> int | None:
+ """Sum of spill_count across all partitions."""
Review Comment:
Same with spill count. Do you know what units it has?
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
Review Comment:
This is leading me to think we should have some high level documentation, probably in the DataFrame page (or a subpage under it). Some of the things it would be good to do are to explain to a user what kinds of information they could find under these metrics and why that data are not available until after the DataFrame has been executed.
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
+ """
+
+ def __init__(self, raw: df_internal.MetricsSet) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ def metrics(self) -> list[Metric]:
+ """Return all individual metrics in this set."""
+ return [Metric(m) for m in self._raw.metrics()]
+
+ @property
+ def output_rows(self) -> int | None:
+ """Sum of output_rows across all partitions."""
+ return self._raw.output_rows()
+
+ @property
+ def elapsed_compute(self) -> int | None:
+ """Sum of elapsed_compute across all partitions, in nanoseconds."""
Review Comment:
We probably want to describe what `elapsed_compute` is rather than assume user knowledge.
##########
src/metrics.rs:
##########
@@ -0,0 +1,143 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use datafusion::physical_plan::metrics::{MetricValue, MetricsSet, Metric};
+use pyo3::prelude::*;
+
+#[pyclass(frozen, name = "MetricsSet", module = "datafusion")]
+#[derive(Debug, Clone)]
+pub struct PyMetricsSet {
+ metrics: MetricsSet,
+}
+
+impl PyMetricsSet {
+ pub fn new(metrics: MetricsSet) -> Self {
+ Self { metrics }
+ }
+}
+
+#[pymethods]
+impl PyMetricsSet {
+ /// Returns all individual metrics in this set.
+ fn metrics(&self) -> Vec<PyMetric> {
+ self.metrics
+ .iter()
+ .map(|m| PyMetric::new(Arc::clone(m)))
+ .collect()
+ }
+
+ /// Returns the sum of all `output_rows` metrics, or None if not present.
+ fn output_rows(&self) -> Option<usize> {
+ self.metrics.output_rows()
+ }
+
+ /// Returns the sum of all `elapsed_compute` metrics in nanoseconds, or None if not present.
+ fn elapsed_compute(&self) -> Option<usize> {
+ self.metrics.elapsed_compute()
+ }
+
+ /// Returns the sum of all `spill_count` metrics, or None if not present.
+ fn spill_count(&self) -> Option<usize> {
+ self.metrics.spill_count()
+ }
+
+ /// Returns the sum of all `spilled_bytes` metrics, or None if not present.
+ fn spilled_bytes(&self) -> Option<usize> {
+ self.metrics.spilled_bytes()
+ }
+
+ /// Returns the sum of all `spilled_rows` metrics, or None if not present.
+ fn spilled_rows(&self) -> Option<usize> {
+ self.metrics.spilled_rows()
+ }
+
+ /// Returns the sum of metrics matching the given name.
+ fn sum_by_name(&self, name: &str) -> Option<usize> {
+ self.metrics.sum_by_name(name).map(|v| v.as_usize())
+ }
+
+ fn __repr__(&self) -> String {
+ format!("{}", self.metrics)
+ }
+}
+
+#[pyclass(frozen, name = "Metric", module = "datafusion")]
+#[derive(Debug, Clone)]
+pub struct PyMetric {
+ metric: Arc<Metric>,
+}
+
+impl PyMetric {
+ pub fn new(metric: Arc<Metric>) -> Self {
+ Self { metric }
+ }
+}
+
+#[pymethods]
+impl PyMetric {
+ /// Returns the name of this metric.
+ #[getter]
+ fn name(&self) -> String {
+ self.metric.value().name().to_string()
+ }
+
+ /// Returns the numeric value of this metric, or None for non-numeric types.
+ #[getter]
+ fn value(&self) -> Option<usize> {
Review Comment:
It feels like we could return `Option<Py<PyAny>>` and try casting the value appropriately.
##########
src/dataframe.rs:
##########
@@ -1127,13 +1154,22 @@ impl PyDataFrame {
fn execute_stream(&self, py: Python) -> PyDataFusionResult<PyRecordBatchStream> {
let df = self.df.as_ref().clone();
- let stream = spawn_future(py, async move { df.execute_stream().await })?;
+ let plan = wait_for_future(py, df.create_physical_plan())??;
+ *self.last_plan.lock() = Some(Arc::clone(&plan));
+ let task_ctx = Arc::new(self.df.as_ref().task_ctx());
+ let stream = spawn_future(py, async move { df_execute_stream(plan, task_ctx) })?;
Ok(PyRecordBatchStream::new(stream))
}
fn execute_stream_partitioned(&self, py: Python) -> PyResult<Vec<PyRecordBatchStream>> {
let df = self.df.as_ref().clone();
- let streams = spawn_future(py, async move { df.execute_stream_partitioned().await })?;
+ let plan = wait_for_future(py, df.create_physical_plan())?
+ .map_err(PyDataFusionError::from)?;
+ *self.last_plan.lock() = Some(Arc::clone(&plan));
+ let task_ctx = Arc::new(self.df.as_ref().task_ctx());
Review Comment:
It *feels* like we're doing this in a bunch of places, so maybe make a private helper function.
##########
src/dataframe.rs:
##########
@@ -626,7 +637,12 @@ impl PyDataFrame {
/// Unless some order is specified in the plan, there is no
/// guarantee of the order of the result.
fn collect<'py>(&self, py: Python<'py>) -> PyResult<Vec<Bound<'py, PyAny>>> {
- let batches = wait_for_future(py, self.df.as_ref().clone().collect())?
+ let df = self.df.as_ref().clone();
+ let plan = wait_for_future(py, df.create_physical_plan())?
+ .map_err(PyDataFusionError::from)?;
+ *self.last_plan.lock() = Some(Arc::clone(&plan));
+ let task_ctx = Arc::new(self.df.as_ref().task_ctx());
+ let batches = wait_for_future(py, df_collect(plan, task_ctx))?
Review Comment:
If I run `collect()` twice on a DF, should we instead just do the lock on the last plan and clone it? I suspect there's not a huge performance difference the vast majority of the time as opposed to how you have it.
##########
src/metrics.rs:
##########
@@ -0,0 +1,143 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use datafusion::physical_plan::metrics::{MetricValue, MetricsSet, Metric};
+use pyo3::prelude::*;
+
+#[pyclass(frozen, name = "MetricsSet", module = "datafusion")]
+#[derive(Debug, Clone)]
+pub struct PyMetricsSet {
+ metrics: MetricsSet,
+}
+
+impl PyMetricsSet {
+ pub fn new(metrics: MetricsSet) -> Self {
+ Self { metrics }
+ }
+}
+
+#[pymethods]
+impl PyMetricsSet {
+ /// Returns all individual metrics in this set.
+ fn metrics(&self) -> Vec<PyMetric> {
+ self.metrics
+ .iter()
+ .map(|m| PyMetric::new(Arc::clone(m)))
+ .collect()
+ }
+
+ /// Returns the sum of all `output_rows` metrics, or None if not present.
+ fn output_rows(&self) -> Option<usize> {
+ self.metrics.output_rows()
+ }
+
+ /// Returns the sum of all `elapsed_compute` metrics in nanoseconds, or None if not present.
Review Comment:
There is a lot of boiler plate comments like this where the function is self explanatory and not exposed to the end user.
##########
python/tests/test_plans.py:
##########
@@ -40,3 +46,101 @@ def test_logical_plan_to_proto(ctx, df) -> None:
execution_plan = ExecutionPlan.from_proto(ctx, execution_plan_bytes)
assert str(original_execution_plan) == str(execution_plan)
+
+
+def test_metrics_tree_walk() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ results = plan.collect_metrics()
+ assert len(results) >= 1
+ found_metrics = False
+ for name, ms in results:
+ assert isinstance(name, str)
+ assert isinstance(ms, MetricsSet)
+ if ms.output_rows is not None and ms.output_rows > 0:
+ found_metrics = True
+ assert found_metrics
+
+
+def test_metric_properties() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ for _, ms in plan.collect_metrics():
+ r = repr(ms)
+ assert isinstance(r, str)
+ for metric in ms.metrics():
+ assert isinstance(metric, Metric)
+ assert isinstance(metric.name, str)
+ assert len(metric.name) > 0
+ assert metric.partition is None or isinstance(metric.partition, int)
+ assert isinstance(metric.labels(), dict)
+ mr = repr(metric)
+ assert isinstance(mr, str)
+ assert len(mr) > 0
+ return
+ pytest.skip("No metrics found")
+
+
+def test_no_metrics_before_execution() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1), (2), (3)")
+ df = ctx.sql("SELECT * FROM t")
+ plan = df.execution_plan()
+ ms = plan.metrics()
+ assert ms is None or ms.output_rows is None or ms.output_rows == 0
Review Comment:
We should know exactly which condition should be hit. Do you know when the metrics is None vs returns no useful output?
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
Review Comment:
On second read I now see this is aggregating across partitions. So does that mean the `metrics()` fn is returning per partition metrics for one ExecutionPlan? Asking for my understanding mostly.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#discussion_r3053323656
##########
crates/core/src/metrics.rs:
##########
@@ -0,0 +1,166 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use datafusion::physical_plan::metrics::{Metric, MetricValue, MetricsSet, Timestamp};
+use pyo3::prelude::*;
+
+#[pyclass(from_py_object, frozen, name = "MetricsSet", module = "datafusion")]
+#[derive(Debug, Clone)]
+pub struct PyMetricsSet {
+ metrics: MetricsSet,
+}
+
+impl PyMetricsSet {
+ pub fn new(metrics: MetricsSet) -> Self {
+ Self { metrics }
+ }
+}
+
+#[pymethods]
+impl PyMetricsSet {
+ fn metrics(&self) -> Vec<PyMetric> {
+ self.metrics
+ .iter()
+ .map(|m| PyMetric::new(Arc::clone(m)))
+ .collect()
+ }
+
+ fn output_rows(&self) -> Option<usize> {
+ self.metrics.output_rows()
+ }
+
+ fn elapsed_compute(&self) -> Option<usize> {
+ self.metrics.elapsed_compute()
+ }
+
+ fn spill_count(&self) -> Option<usize> {
+ self.metrics.spill_count()
+ }
+
+ fn spilled_bytes(&self) -> Option<usize> {
+ self.metrics.spilled_bytes()
+ }
+
+ fn spilled_rows(&self) -> Option<usize> {
+ self.metrics.spilled_rows()
+ }
+
+ fn sum_by_name(&self, name: &str) -> Option<usize> {
+ self.metrics.sum_by_name(name).map(|v| v.as_usize())
+ }
+
+ fn __repr__(&self) -> String {
+ format!("{}", self.metrics)
+ }
+}
+
+#[pyclass(from_py_object, frozen, name = "Metric", module = "datafusion")]
+#[derive(Debug, Clone)]
+pub struct PyMetric {
+ metric: Arc<Metric>,
+}
+
+impl PyMetric {
+ pub fn new(metric: Arc<Metric>) -> Self {
+ Self { metric }
+ }
+
+ fn timestamp_to_pyobject<'py>(
+ py: Python<'py>,
+ ts: &Timestamp,
+ ) -> PyResult<Option<Bound<'py, PyAny>>> {
+ match ts.value() {
+ Some(dt) => {
+ let nanos = dt.timestamp_nanos_opt().ok_or_else(|| {
+ PyErr::new::<pyo3::exceptions::PyOverflowError, _>("timestamp out of range")
+ })?;
+ let datetime_mod = py.import("datetime")?;
+ let datetime_cls = datetime_mod.getattr("datetime")?;
+ let tz_utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
+ let secs = nanos / 1_000_000_000;
+ let micros = (nanos % 1_000_000_000) / 1_000;
+ let result = datetime_cls.call_method1(
+ "fromtimestamp",
+ (secs as f64 + micros as f64 / 1_000_000.0, tz_utc),
+ )?;
+ Ok(Some(result))
+ }
+ None => Ok(None),
+ }
+ }
+}
+
+#[pymethods]
+impl PyMetric {
+ #[getter]
+ fn name(&self) -> String {
+ self.metric.value().name().to_string()
+ }
+
+ #[getter]
+ fn value<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyAny>>> {
+ match self.metric.value() {
+ MetricValue::OutputRows(c) => Ok(Some(c.value().into_pyobject(py)?.into_any())),
+ MetricValue::OutputBytes(c) => Ok(Some(c.value().into_pyobject(py)?.into_any())),
+ MetricValue::ElapsedCompute(t) => Ok(Some(t.value().into_pyobject(py)?.into_any())),
+ MetricValue::SpillCount(c) => Ok(Some(c.value().into_pyobject(py)?.into_any())),
+ MetricValue::SpilledBytes(c) => Ok(Some(c.value().into_pyobject(py)?.into_any())),
+ MetricValue::SpilledRows(c) => Ok(Some(c.value().into_pyobject(py)?.into_any())),
+ MetricValue::CurrentMemoryUsage(g) => Ok(Some(g.value().into_pyobject(py)?.into_any())),
+ MetricValue::Count { count, .. } => {
+ Ok(Some(count.value().into_pyobject(py)?.into_any()))
+ }
+ MetricValue::Gauge { gauge, .. } => {
+ Ok(Some(gauge.value().into_pyobject(py)?.into_any()))
+ }
+ MetricValue::Time { time, .. } => Ok(Some(time.value().into_pyobject(py)?.into_any())),
+ MetricValue::StartTimestamp(ts) | MetricValue::EndTimestamp(ts) => {
+ Self::timestamp_to_pyobject(py, ts)
+ }
+ _ => Ok(None),
+ }
+ }
+
+ fn value_as_datetime<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyAny>>> {
Review Comment:
This is a @property on the python side so should we have `#[getter]`?
##########
python/datafusion/plan.py:
##########
@@ -151,3 +155,176 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node, or None if this node has no MetricsSet.
Review Comment:
```suggestion
"""Return metrics for this plan node, or None if this plan has no MetricsSet.
```
##########
docs/source/user-guide/dataframe/execution-metrics.rst:
##########
@@ -0,0 +1,212 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+.. or more contributor license agreements. See the NOTICE file
+.. distributed with this work for additional information
+.. regarding copyright ownership. The ASF licenses this file
+.. to you under the Apache License, Version 2.0 (the
+.. "License"); you may not use this file except in compliance
+.. with the License. You may obtain a copy of the License at
+
+.. http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+.. software distributed under the License is distributed on an
+.. "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+.. KIND, either express or implied. See the License for the
+.. specific language governing permissions and limitations
+.. under the License.
+
+.. _execution_metrics:
+
+Execution Metrics
+=================
+
+Overview
+--------
+
+When DataFusion executes a query it compiles the logical plan into a tree of
+*physical plan operators* (e.g. ``FilterExec``, ``ProjectionExec``,
+``HashAggregateExec``). Each operator can record runtime statistics while it
+runs. These statistics are called **execution metrics**.
+
+Typical metrics include:
+
+- **output_rows** – number of rows produced by the operator
+- **elapsed_compute** – total CPU time (nanoseconds) spent inside the operator
+- **spill_count** – number of times the operator spilled data to disk
+- **spilled_bytes** – total bytes written to disk during spills
+- **spilled_rows** – total rows written to disk during spills
+
+Metrics are collected *per-partition*: DataFusion may execute each operator
+in parallel across several partitions. The convenience properties on
+:py:class:`~datafusion.MetricsSet` (e.g. ``output_rows``, ``elapsed_compute``)
+automatically sum the named metric across **all** partitions, giving a single
+aggregate value for the operator as a whole. You can also access the raw
+per-partition :py:class:`~datafusion.Metric` objects via
+:py:meth:`~datafusion.MetricsSet.metrics`.
+
+When Are Metrics Available?
+---------------------------
+
+Metrics are populated only **after** the DataFrame has been executed.
+Execution is triggered by any of the terminal operations:
Review Comment:
It's actually up to exec executor to decide when the metrics are calculated, right? Aren't there some that may populate metrics ahead of time? I think that's part of what I found when looking at https://github.com/apache/datafusion-python/pull/1381#issuecomment-4144415233
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "ShreyeshArangath (via GitHub)" <gi...@apache.org>.
ShreyeshArangath commented on PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#issuecomment-4093199310
Apologies for the delayed response on this one 😅
> One area I am concerned about is that when we do a display() we do bypass all of this mechanism. That is good and bad. The good is that the metrics are definitely going to be different between the smaller collection that happens when we display because it ends early. The bad is that as a user it's probably confusing to see the the data but then be told that we don't have the metrics for the data in front of them. What do you think?
That's a totally fair concern, and I think this is worth addressing. It's definitely going to trip people up. I'm still thinking through this, but a high-level idea could be that we could possibly have display() also cache a plan so users can at least inspect metrics from the display execution (haven't looked into what might be required to support this). The metrics would reflect the truncated run but timestamps and compute times would still be meaningful. WDYT?
For documentation, I've already started a little bit of work in this PR, please do lmk what you think (you likely have a lot more context on what a user might expect).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#issuecomment-4192072018
How is this coming along? We've got a short list of PRs to merge before we start the release on 53 and I'm wondering if you think this is close to review to see if we should include in this release or hold for the next.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#discussion_r2855554863
##########
python/tests/test_plans.py:
##########
@@ -40,3 +46,101 @@ def test_logical_plan_to_proto(ctx, df) -> None:
execution_plan = ExecutionPlan.from_proto(ctx, execution_plan_bytes)
assert str(original_execution_plan) == str(execution_plan)
+
+
+def test_metrics_tree_walk() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ results = plan.collect_metrics()
+ assert len(results) >= 1
+ found_metrics = False
+ for name, ms in results:
+ assert isinstance(name, str)
+ assert isinstance(ms, MetricsSet)
+ if ms.output_rows is not None and ms.output_rows > 0:
+ found_metrics = True
+ assert found_metrics
+
+
+def test_metric_properties() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ for _, ms in plan.collect_metrics():
+ r = repr(ms)
+ assert isinstance(r, str)
+ for metric in ms.metrics():
+ assert isinstance(metric, Metric)
+ assert isinstance(metric.name, str)
+ assert len(metric.name) > 0
+ assert metric.partition is None or isinstance(metric.partition, int)
+ assert isinstance(metric.labels(), dict)
+ mr = repr(metric)
+ assert isinstance(mr, str)
+ assert len(mr) > 0
+ return
+ pytest.skip("No metrics found")
+
+
+def test_no_metrics_before_execution() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1), (2), (3)")
+ df = ctx.sql("SELECT * FROM t")
+ plan = df.execution_plan()
+ ms = plan.metrics()
+ assert ms is None or ms.output_rows is None or ms.output_rows == 0
+
+
+def test_collect_partitioned_metrics() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+
+ df.collect_partitioned()
+ plan = df.execution_plan()
+
+ found_metrics = False
+ for _, ms in plan.collect_metrics():
+ if ms.output_rows is not None and ms.output_rows > 0:
Review Comment:
We should know exactly how many output rows to expect
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
+ """
+
+ def __init__(self, raw: df_internal.MetricsSet) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ def metrics(self) -> list[Metric]:
+ """Return all individual metrics in this set."""
+ return [Metric(m) for m in self._raw.metrics()]
+
+ @property
+ def output_rows(self) -> int | None:
+ """Sum of output_rows across all partitions."""
+ return self._raw.output_rows()
+
+ @property
+ def elapsed_compute(self) -> int | None:
+ """Sum of elapsed_compute across all partitions, in nanoseconds."""
+ return self._raw.elapsed_compute()
+
+ @property
+ def spill_count(self) -> int | None:
+ """Sum of spill_count across all partitions."""
+ return self._raw.spill_count()
+
+ @property
+ def spilled_bytes(self) -> int | None:
+ """Sum of spilled_bytes across all partitions."""
+ return self._raw.spilled_bytes()
+
+ @property
+ def spilled_rows(self) -> int | None:
+ """Sum of spilled_rows across all partitions."""
+ return self._raw.spilled_rows()
+
+ def sum_by_name(self, name: str) -> int | None:
+ """Return the sum of metrics matching the given name."""
Review Comment:
Sum across partitions, I presume. This is different because it works for an arbitrary metric, right?
##########
python/tests/test_plans.py:
##########
@@ -40,3 +46,101 @@ def test_logical_plan_to_proto(ctx, df) -> None:
execution_plan = ExecutionPlan.from_proto(ctx, execution_plan_bytes)
assert str(original_execution_plan) == str(execution_plan)
+
+
+def test_metrics_tree_walk() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ results = plan.collect_metrics()
+ assert len(results) >= 1
+ found_metrics = False
+ for name, ms in results:
+ assert isinstance(name, str)
+ assert isinstance(ms, MetricsSet)
+ if ms.output_rows is not None and ms.output_rows > 0:
+ found_metrics = True
+ assert found_metrics
+
+
+def test_metric_properties() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ for _, ms in plan.collect_metrics():
+ r = repr(ms)
+ assert isinstance(r, str)
+ for metric in ms.metrics():
+ assert isinstance(metric, Metric)
+ assert isinstance(metric.name, str)
+ assert len(metric.name) > 0
+ assert metric.partition is None or isinstance(metric.partition, int)
+ assert isinstance(metric.labels(), dict)
+ mr = repr(metric)
+ assert isinstance(mr, str)
+ assert len(mr) > 0
+ return
+ pytest.skip("No metrics found")
Review Comment:
I don't think we want a skip here. I think we want to ensure in our unit test that we do generate metrics. I think we should know a priori what metrics we're getting in this test. While some values we can't use, some parts of it we should be able to test directly, like the metric name and labels.
##########
python/tests/test_plans.py:
##########
@@ -40,3 +46,101 @@ def test_logical_plan_to_proto(ctx, df) -> None:
execution_plan = ExecutionPlan.from_proto(ctx, execution_plan_bytes)
assert str(original_execution_plan) == str(execution_plan)
+
+
+def test_metrics_tree_walk() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ results = plan.collect_metrics()
+ assert len(results) >= 1
+ found_metrics = False
+ for name, ms in results:
+ assert isinstance(name, str)
+ assert isinstance(ms, MetricsSet)
+ if ms.output_rows is not None and ms.output_rows > 0:
+ found_metrics = True
+ assert found_metrics
+
+
+def test_metric_properties() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ for _, ms in plan.collect_metrics():
+ r = repr(ms)
+ assert isinstance(r, str)
+ for metric in ms.metrics():
+ assert isinstance(metric, Metric)
+ assert isinstance(metric.name, str)
+ assert len(metric.name) > 0
+ assert metric.partition is None or isinstance(metric.partition, int)
+ assert isinstance(metric.labels(), dict)
+ mr = repr(metric)
+ assert isinstance(mr, str)
+ assert len(mr) > 0
+ return
+ pytest.skip("No metrics found")
+
+
+def test_no_metrics_before_execution() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1), (2), (3)")
+ df = ctx.sql("SELECT * FROM t")
+ plan = df.execution_plan()
+ ms = plan.metrics()
+ assert ms is None or ms.output_rows is None or ms.output_rows == 0
+
+
+def test_collect_partitioned_metrics() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+
+ df.collect_partitioned()
+ plan = df.execution_plan()
+
+ found_metrics = False
+ for _, ms in plan.collect_metrics():
+ if ms.output_rows is not None and ms.output_rows > 0:
Review Comment:
Also for the unit tests to follow
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
Review Comment:
I haven't dug in, but is `operator_name` the name of the execution plan?
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
Review Comment:
"Walk the plan tree and collect metrics" probably does not make a lot of sense to someone other than a developer. I think we can make this more user focused.
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
Review Comment:
A bit of an explanation is probably useful here. Again, I don't think we can assume the user understands that there are both individual execution plan metrics as well as aggregate. I *think* that some operators have metrics that cannot be aggregated. In general I suspect we really do need some high level documentation with examples we can point to that makes all of this more concrete.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "ShreyeshArangath (via GitHub)" <gi...@apache.org>.
ShreyeshArangath commented on code in PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#discussion_r2962631250
##########
src/metrics.rs:
##########
@@ -0,0 +1,143 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use datafusion::physical_plan::metrics::{MetricValue, MetricsSet, Metric};
+use pyo3::prelude::*;
+
+#[pyclass(frozen, name = "MetricsSet", module = "datafusion")]
+#[derive(Debug, Clone)]
+pub struct PyMetricsSet {
+ metrics: MetricsSet,
+}
+
+impl PyMetricsSet {
+ pub fn new(metrics: MetricsSet) -> Self {
+ Self { metrics }
+ }
+}
+
+#[pymethods]
+impl PyMetricsSet {
+ /// Returns all individual metrics in this set.
+ fn metrics(&self) -> Vec<PyMetric> {
+ self.metrics
+ .iter()
+ .map(|m| PyMetric::new(Arc::clone(m)))
+ .collect()
+ }
+
+ /// Returns the sum of all `output_rows` metrics, or None if not present.
+ fn output_rows(&self) -> Option<usize> {
+ self.metrics.output_rows()
+ }
+
+ /// Returns the sum of all `elapsed_compute` metrics in nanoseconds, or None if not present.
+ fn elapsed_compute(&self) -> Option<usize> {
+ self.metrics.elapsed_compute()
+ }
+
+ /// Returns the sum of all `spill_count` metrics, or None if not present.
+ fn spill_count(&self) -> Option<usize> {
+ self.metrics.spill_count()
+ }
+
+ /// Returns the sum of all `spilled_bytes` metrics, or None if not present.
+ fn spilled_bytes(&self) -> Option<usize> {
+ self.metrics.spilled_bytes()
+ }
+
+ /// Returns the sum of all `spilled_rows` metrics, or None if not present.
+ fn spilled_rows(&self) -> Option<usize> {
+ self.metrics.spilled_rows()
+ }
+
+ /// Returns the sum of metrics matching the given name.
+ fn sum_by_name(&self, name: &str) -> Option<usize> {
+ self.metrics.sum_by_name(name).map(|v| v.as_usize())
+ }
+
+ fn __repr__(&self) -> String {
+ format!("{}", self.metrics)
+ }
+}
+
+#[pyclass(frozen, name = "Metric", module = "datafusion")]
+#[derive(Debug, Clone)]
+pub struct PyMetric {
+ metric: Arc<Metric>,
+}
+
+impl PyMetric {
+ pub fn new(metric: Arc<Metric>) -> Self {
+ Self { metric }
+ }
+}
+
+#[pymethods]
+impl PyMetric {
+ /// Returns the name of this metric.
+ #[getter]
+ fn name(&self) -> String {
+ self.metric.value().name().to_string()
+ }
+
+ /// Returns the numeric value of this metric, or None for non-numeric types.
+ #[getter]
+ fn value(&self) -> Option<usize> {
Review Comment:
Agreed! Attempted to fixing this, lmk what you think
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "ShreyeshArangath (via GitHub)" <gi...@apache.org>.
ShreyeshArangath commented on PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#issuecomment-4193430482
> How is this coming along? We've got a short list of PRs to merge before we start the release on 53 and I'm wondering if you think this is close to review to see if we should include in this release or hold for the next.
Just trying to fix the last couple build issues, but I think functionally this is close to review, it'd be nice to include it in the next release. Thanks
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "ShreyeshArangath (via GitHub)" <gi...@apache.org>.
ShreyeshArangath commented on code in PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#discussion_r2861995966
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
Review Comment:
+1, I think that would be super helpful. I can extend this to include a new user-facing RST page covering things like what metrics are, when they're available, how the physical plan tree maps to operators, etc.
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
+ """
+
+ def __init__(self, raw: df_internal.MetricsSet) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ def metrics(self) -> list[Metric]:
+ """Return all individual metrics in this set."""
+ return [Metric(m) for m in self._raw.metrics()]
+
+ @property
+ def output_rows(self) -> int | None:
+ """Sum of output_rows across all partitions."""
+ return self._raw.output_rows()
+
+ @property
+ def elapsed_compute(self) -> int | None:
+ """Sum of elapsed_compute across all partitions, in nanoseconds."""
+ return self._raw.elapsed_compute()
+
+ @property
+ def spill_count(self) -> int | None:
+ """Sum of spill_count across all partitions."""
+ return self._raw.spill_count()
+
+ @property
+ def spilled_bytes(self) -> int | None:
+ """Sum of spilled_bytes across all partitions."""
+ return self._raw.spilled_bytes()
+
+ @property
+ def spilled_rows(self) -> int | None:
+ """Sum of spilled_rows across all partitions."""
+ return self._raw.spilled_rows()
+
+ def sum_by_name(self, name: str) -> int | None:
+ """Return the sum of metrics matching the given name."""
+ return self._raw.sum_by_name(name)
+
+ def __repr__(self) -> str:
+ """Return a string representation of the metrics set."""
+ return repr(self._raw)
+
+
+class Metric:
+ """A single execution metric with name, value, partition, and labels."""
+
+ def __init__(self, raw: df_internal.Metric) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ @property
+ def name(self) -> str:
+ """The name of this metric (e.g. ``output_rows``)."""
+ return self._raw.name
+
+ @property
+ def value(self) -> int | None:
+ """The numeric value of this metric, or None for non-numeric types."""
+ return self._raw.value
Review Comment:
yes, i think right now it uses a option<usize>, unrecognised variants return None; we could possibly return richer types by returning `Option<Py<PyAny>>` as you suggested
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
+ """
+
+ def __init__(self, raw: df_internal.MetricsSet) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ def metrics(self) -> list[Metric]:
+ """Return all individual metrics in this set."""
+ return [Metric(m) for m in self._raw.metrics()]
+
+ @property
+ def output_rows(self) -> int | None:
+ """Sum of output_rows across all partitions."""
+ return self._raw.output_rows()
+
+ @property
+ def elapsed_compute(self) -> int | None:
+ """Sum of elapsed_compute across all partitions, in nanoseconds."""
+ return self._raw.elapsed_compute()
+
+ @property
+ def spill_count(self) -> int | None:
+ """Sum of spill_count across all partitions."""
+ return self._raw.spill_count()
+
+ @property
+ def spilled_bytes(self) -> int | None:
+ """Sum of spilled_bytes across all partitions."""
+ return self._raw.spilled_bytes()
+
+ @property
+ def spilled_rows(self) -> int | None:
+ """Sum of spilled_rows across all partitions."""
+ return self._raw.spilled_rows()
+
+ def sum_by_name(self, name: str) -> int | None:
+ """Return the sum of metrics matching the given name."""
Review Comment:
Yes, that's right
##########
python/tests/test_plans.py:
##########
@@ -40,3 +46,101 @@ def test_logical_plan_to_proto(ctx, df) -> None:
execution_plan = ExecutionPlan.from_proto(ctx, execution_plan_bytes)
assert str(original_execution_plan) == str(execution_plan)
+
+
+def test_metrics_tree_walk() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ results = plan.collect_metrics()
+ assert len(results) >= 1
+ found_metrics = False
+ for name, ms in results:
+ assert isinstance(name, str)
+ assert isinstance(ms, MetricsSet)
+ if ms.output_rows is not None and ms.output_rows > 0:
+ found_metrics = True
+ assert found_metrics
+
+
+def test_metric_properties() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ for _, ms in plan.collect_metrics():
+ r = repr(ms)
+ assert isinstance(r, str)
+ for metric in ms.metrics():
+ assert isinstance(metric, Metric)
+ assert isinstance(metric.name, str)
+ assert len(metric.name) > 0
+ assert metric.partition is None or isinstance(metric.partition, int)
+ assert isinstance(metric.labels(), dict)
+ mr = repr(metric)
+ assert isinstance(mr, str)
+ assert len(mr) > 0
+ return
+ pytest.skip("No metrics found")
Review Comment:
Removed all pytest.skip calls
##########
python/tests/test_plans.py:
##########
@@ -40,3 +46,101 @@ def test_logical_plan_to_proto(ctx, df) -> None:
execution_plan = ExecutionPlan.from_proto(ctx, execution_plan_bytes)
assert str(original_execution_plan) == str(execution_plan)
+
+
+def test_metrics_tree_walk() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ results = plan.collect_metrics()
+ assert len(results) >= 1
+ found_metrics = False
+ for name, ms in results:
+ assert isinstance(name, str)
+ assert isinstance(ms, MetricsSet)
+ if ms.output_rows is not None and ms.output_rows > 0:
+ found_metrics = True
+ assert found_metrics
+
+
+def test_metric_properties() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ for _, ms in plan.collect_metrics():
+ r = repr(ms)
+ assert isinstance(r, str)
+ for metric in ms.metrics():
+ assert isinstance(metric, Metric)
+ assert isinstance(metric.name, str)
+ assert len(metric.name) > 0
+ assert metric.partition is None or isinstance(metric.partition, int)
+ assert isinstance(metric.labels(), dict)
+ mr = repr(metric)
+ assert isinstance(mr, str)
+ assert len(mr) > 0
+ return
+ pytest.skip("No metrics found")
+
+
+def test_no_metrics_before_execution() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1), (2), (3)")
+ df = ctx.sql("SELECT * FROM t")
+ plan = df.execution_plan()
+ ms = plan.metrics()
+ assert ms is None or ms.output_rows is None or ms.output_rows == 0
+
+
+def test_collect_partitioned_metrics() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+
+ df.collect_partitioned()
+ plan = df.execution_plan()
+
+ found_metrics = False
+ for _, ms in plan.collect_metrics():
+ if ms.output_rows is not None and ms.output_rows > 0:
Review Comment:
done!
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
+ """
+
+ def __init__(self, raw: df_internal.MetricsSet) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ def metrics(self) -> list[Metric]:
+ """Return all individual metrics in this set."""
+ return [Metric(m) for m in self._raw.metrics()]
+
+ @property
+ def output_rows(self) -> int | None:
+ """Sum of output_rows across all partitions."""
+ return self._raw.output_rows()
+
+ @property
+ def elapsed_compute(self) -> int | None:
+ """Sum of elapsed_compute across all partitions, in nanoseconds."""
+ return self._raw.elapsed_compute()
+
+ @property
+ def spill_count(self) -> int | None:
+ """Sum of spill_count across all partitions."""
Review Comment:
From my understanding it seems to be spill-to-disk events (so no units). Updated the doc
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
Review Comment:
Updated! `operator_name` was meant to be the node.display() str
##########
python/datafusion/plan.py:
##########
@@ -151,3 +153,107 @@ def to_proto(self) -> bytes:
Tables created in memory from record batches are currently not supported.
"""
return self._raw_plan.to_proto()
+
+ def metrics(self) -> MetricsSet | None:
+ """Return metrics for this plan node after execution, or None if unavailable."""
+ raw = self._raw_plan.metrics()
+ if raw is None:
+ return None
+ return MetricsSet(raw)
+
+ def collect_metrics(self) -> list[tuple[str, MetricsSet]]:
+ """Walk the plan tree and collect metrics from all operators.
+
+ Returns a list of (operator_name, MetricsSet) tuples.
+ """
+ result: list[tuple[str, MetricsSet]] = []
+
+ def _walk(node: ExecutionPlan) -> None:
+ ms = node.metrics()
+ if ms is not None:
+ result.append((node.display(), ms))
+ for child in node.children():
+ _walk(child)
+
+ _walk(self)
+ return result
+
+
+class MetricsSet:
+ """A set of metrics for a single execution plan operator.
+
+ Provides both individual metric access and convenience aggregations
+ across partitions.
+ """
+
+ def __init__(self, raw: df_internal.MetricsSet) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ def metrics(self) -> list[Metric]:
+ """Return all individual metrics in this set."""
+ return [Metric(m) for m in self._raw.metrics()]
+
+ @property
+ def output_rows(self) -> int | None:
+ """Sum of output_rows across all partitions."""
+ return self._raw.output_rows()
+
+ @property
+ def elapsed_compute(self) -> int | None:
+ """Sum of elapsed_compute across all partitions, in nanoseconds."""
+ return self._raw.elapsed_compute()
+
+ @property
+ def spill_count(self) -> int | None:
+ """Sum of spill_count across all partitions."""
+ return self._raw.spill_count()
+
+ @property
+ def spilled_bytes(self) -> int | None:
+ """Sum of spilled_bytes across all partitions."""
+ return self._raw.spilled_bytes()
+
+ @property
+ def spilled_rows(self) -> int | None:
+ """Sum of spilled_rows across all partitions."""
+ return self._raw.spilled_rows()
+
+ def sum_by_name(self, name: str) -> int | None:
+ """Return the sum of metrics matching the given name."""
+ return self._raw.sum_by_name(name)
+
+ def __repr__(self) -> str:
+ """Return a string representation of the metrics set."""
+ return repr(self._raw)
+
+
+class Metric:
+ """A single execution metric with name, value, partition, and labels."""
+
+ def __init__(self, raw: df_internal.Metric) -> None:
+ """This constructor should not be called by the end user."""
+ self._raw = raw
+
+ @property
+ def name(self) -> str:
+ """The name of this metric (e.g. ``output_rows``)."""
+ return self._raw.name
+
+ @property
+ def value(self) -> int | None:
+ """The numeric value of this metric, or None for non-numeric types."""
+ return self._raw.value
+
+ @property
+ def partition(self) -> int | None:
+ """The partition this metric applies to, or None if global."""
Review Comment:
this is just a 0-based partition id. `partition 0, partition 1` etc. basically the parallel slots datafusion splits work across. you’d mostly use it to compare the same metric across partitions and spot skew, like when one partition is doing way more work than the others.
updated the docstring to call it a “0-based partition index” and clarified that None just means the metric isn’t tied to any specific partition.
##########
python/tests/test_plans.py:
##########
@@ -40,3 +46,101 @@ def test_logical_plan_to_proto(ctx, df) -> None:
execution_plan = ExecutionPlan.from_proto(ctx, execution_plan_bytes)
assert str(original_execution_plan) == str(execution_plan)
+
+
+def test_metrics_tree_walk() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ results = plan.collect_metrics()
+ assert len(results) >= 1
+ found_metrics = False
+ for name, ms in results:
+ assert isinstance(name, str)
+ assert isinstance(ms, MetricsSet)
+ if ms.output_rows is not None and ms.output_rows > 0:
+ found_metrics = True
+ assert found_metrics
+
+
+def test_metric_properties() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
+ df.collect()
+ plan = df.execution_plan()
+
+ for _, ms in plan.collect_metrics():
+ r = repr(ms)
+ assert isinstance(r, str)
+ for metric in ms.metrics():
+ assert isinstance(metric, Metric)
+ assert isinstance(metric.name, str)
+ assert len(metric.name) > 0
+ assert metric.partition is None or isinstance(metric.partition, int)
+ assert isinstance(metric.labels(), dict)
+ mr = repr(metric)
+ assert isinstance(mr, str)
+ assert len(mr) > 0
+ return
+ pytest.skip("No metrics found")
+
+
+def test_no_metrics_before_execution() -> None:
+ ctx = SessionContext()
+ ctx.sql("CREATE TABLE t AS VALUES (1), (2), (3)")
+ df = ctx.sql("SELECT * FROM t")
+ plan = df.execution_plan()
+ ms = plan.metrics()
+ assert ms is None or ms.output_rows is None or ms.output_rows == 0
Review Comment:
metrics() returns None when the plan node hasn't been executed yet. After execution, it returns a MetricsSet which always contains at least the metrics the operator recorded. I've updated test_no_metrics_before_execution in to assert ms is None directly, rather than the previous looser check, since before execution the root node should have no MetricsSet at all
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "ShreyeshArangath (via GitHub)" <gi...@apache.org>.
ShreyeshArangath commented on PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#issuecomment-3931838295
Thanks for the suggestion! Initially when I designed the change, I did consider moving `collect()` / `execute_*()` onto plan object. The reason I didn’t go that route was more about how observability fits into real usage patterns (from the cases that I have seen).
Today, I think the users naturally treat a dataframe as the primary handle for a query:
```py
df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
batches = df.collect()
```
Requiring metrics to go through ExecutionPlan would effectively change the model to look something like so
```py
df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
plan = df.execution_plan()
batches = plan.collect()
metrics = plan.collect_metrics()
```
I thought that this would require users to restructure pipelines and thread a plan object through call chains purely to have access to metrics. The LoE required to get people to use it seemed high to me.
My goal was to make minimal changes to how users can add support for metrics without changing how they run queries
```py
df = ctx.sql("SELECT * FROM t WHERE column1 > 1")
batches = df.collect()
plan = df.execution_plan()
metrics = plan.collect_metrics()
```
I’m happy to switch to the plan-based approach if we prefer stronger alignment with the upstream API, but I leaned toward this design to make observability easier to adopt without disrupting current usage patterns — lmk what you think
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "ShreyeshArangath (via GitHub)" <gi...@apache.org>.
ShreyeshArangath commented on code in PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#discussion_r3054556163
##########
docs/source/user-guide/dataframe/execution-metrics.rst:
##########
@@ -0,0 +1,212 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+.. or more contributor license agreements. See the NOTICE file
+.. distributed with this work for additional information
+.. regarding copyright ownership. The ASF licenses this file
+.. to you under the Apache License, Version 2.0 (the
+.. "License"); you may not use this file except in compliance
+.. with the License. You may obtain a copy of the License at
+
+.. http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+.. software distributed under the License is distributed on an
+.. "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+.. KIND, either express or implied. See the License for the
+.. specific language governing permissions and limitations
+.. under the License.
+
+.. _execution_metrics:
+
+Execution Metrics
+=================
+
+Overview
+--------
+
+When DataFusion executes a query it compiles the logical plan into a tree of
+*physical plan operators* (e.g. ``FilterExec``, ``ProjectionExec``,
+``HashAggregateExec``). Each operator can record runtime statistics while it
+runs. These statistics are called **execution metrics**.
+
+Typical metrics include:
+
+- **output_rows** – number of rows produced by the operator
+- **elapsed_compute** – total CPU time (nanoseconds) spent inside the operator
+- **spill_count** – number of times the operator spilled data to disk
+- **spilled_bytes** – total bytes written to disk during spills
+- **spilled_rows** – total rows written to disk during spills
+
+Metrics are collected *per-partition*: DataFusion may execute each operator
+in parallel across several partitions. The convenience properties on
+:py:class:`~datafusion.MetricsSet` (e.g. ``output_rows``, ``elapsed_compute``)
+automatically sum the named metric across **all** partitions, giving a single
+aggregate value for the operator as a whole. You can also access the raw
+per-partition :py:class:`~datafusion.Metric` objects via
+:py:meth:`~datafusion.MetricsSet.metrics`.
+
+When Are Metrics Available?
+---------------------------
+
+Metrics are populated only **after** the DataFrame has been executed.
+Execution is triggered by any of the terminal operations:
Review Comment:
updated
##########
crates/core/src/metrics.rs:
##########
@@ -0,0 +1,166 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use datafusion::physical_plan::metrics::{Metric, MetricValue, MetricsSet, Timestamp};
+use pyo3::prelude::*;
+
+#[pyclass(from_py_object, frozen, name = "MetricsSet", module = "datafusion")]
+#[derive(Debug, Clone)]
+pub struct PyMetricsSet {
+ metrics: MetricsSet,
+}
+
+impl PyMetricsSet {
+ pub fn new(metrics: MetricsSet) -> Self {
+ Self { metrics }
+ }
+}
+
+#[pymethods]
+impl PyMetricsSet {
+ fn metrics(&self) -> Vec<PyMetric> {
+ self.metrics
+ .iter()
+ .map(|m| PyMetric::new(Arc::clone(m)))
+ .collect()
+ }
+
+ fn output_rows(&self) -> Option<usize> {
+ self.metrics.output_rows()
+ }
+
+ fn elapsed_compute(&self) -> Option<usize> {
+ self.metrics.elapsed_compute()
+ }
+
+ fn spill_count(&self) -> Option<usize> {
+ self.metrics.spill_count()
+ }
+
+ fn spilled_bytes(&self) -> Option<usize> {
+ self.metrics.spilled_bytes()
+ }
+
+ fn spilled_rows(&self) -> Option<usize> {
+ self.metrics.spilled_rows()
+ }
+
+ fn sum_by_name(&self, name: &str) -> Option<usize> {
+ self.metrics.sum_by_name(name).map(|v| v.as_usize())
+ }
+
+ fn __repr__(&self) -> String {
+ format!("{}", self.metrics)
+ }
+}
+
+#[pyclass(from_py_object, frozen, name = "Metric", module = "datafusion")]
+#[derive(Debug, Clone)]
+pub struct PyMetric {
+ metric: Arc<Metric>,
+}
+
+impl PyMetric {
+ pub fn new(metric: Arc<Metric>) -> Self {
+ Self { metric }
+ }
+
+ fn timestamp_to_pyobject<'py>(
+ py: Python<'py>,
+ ts: &Timestamp,
+ ) -> PyResult<Option<Bound<'py, PyAny>>> {
+ match ts.value() {
+ Some(dt) => {
+ let nanos = dt.timestamp_nanos_opt().ok_or_else(|| {
+ PyErr::new::<pyo3::exceptions::PyOverflowError, _>("timestamp out of range")
+ })?;
+ let datetime_mod = py.import("datetime")?;
+ let datetime_cls = datetime_mod.getattr("datetime")?;
+ let tz_utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
+ let secs = nanos / 1_000_000_000;
+ let micros = (nanos % 1_000_000_000) / 1_000;
+ let result = datetime_cls.call_method1(
+ "fromtimestamp",
+ (secs as f64 + micros as f64 / 1_000_000.0, tz_utc),
+ )?;
+ Ok(Some(result))
+ }
+ None => Ok(None),
+ }
+ }
+}
Review Comment:
nice, updated :)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on code in PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#discussion_r3053458404
##########
crates/core/src/metrics.rs:
##########
@@ -0,0 +1,166 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use datafusion::physical_plan::metrics::{Metric, MetricValue, MetricsSet, Timestamp};
+use pyo3::prelude::*;
+
+#[pyclass(from_py_object, frozen, name = "MetricsSet", module = "datafusion")]
+#[derive(Debug, Clone)]
+pub struct PyMetricsSet {
+ metrics: MetricsSet,
+}
+
+impl PyMetricsSet {
+ pub fn new(metrics: MetricsSet) -> Self {
+ Self { metrics }
+ }
+}
+
+#[pymethods]
+impl PyMetricsSet {
+ fn metrics(&self) -> Vec<PyMetric> {
+ self.metrics
+ .iter()
+ .map(|m| PyMetric::new(Arc::clone(m)))
+ .collect()
+ }
+
+ fn output_rows(&self) -> Option<usize> {
+ self.metrics.output_rows()
+ }
+
+ fn elapsed_compute(&self) -> Option<usize> {
+ self.metrics.elapsed_compute()
+ }
+
+ fn spill_count(&self) -> Option<usize> {
+ self.metrics.spill_count()
+ }
+
+ fn spilled_bytes(&self) -> Option<usize> {
+ self.metrics.spilled_bytes()
+ }
+
+ fn spilled_rows(&self) -> Option<usize> {
+ self.metrics.spilled_rows()
+ }
+
+ fn sum_by_name(&self, name: &str) -> Option<usize> {
+ self.metrics.sum_by_name(name).map(|v| v.as_usize())
+ }
+
+ fn __repr__(&self) -> String {
+ format!("{}", self.metrics)
+ }
+}
+
+#[pyclass(from_py_object, frozen, name = "Metric", module = "datafusion")]
+#[derive(Debug, Clone)]
+pub struct PyMetric {
+ metric: Arc<Metric>,
+}
+
+impl PyMetric {
+ pub fn new(metric: Arc<Metric>) -> Self {
+ Self { metric }
+ }
+
+ fn timestamp_to_pyobject<'py>(
+ py: Python<'py>,
+ ts: &Timestamp,
+ ) -> PyResult<Option<Bound<'py, PyAny>>> {
+ match ts.value() {
+ Some(dt) => {
+ let nanos = dt.timestamp_nanos_opt().ok_or_else(|| {
+ PyErr::new::<pyo3::exceptions::PyOverflowError, _>("timestamp out of range")
+ })?;
+ let datetime_mod = py.import("datetime")?;
+ let datetime_cls = datetime_mod.getattr("datetime")?;
+ let tz_utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
+ let secs = nanos / 1_000_000_000;
+ let micros = (nanos % 1_000_000_000) / 1_000;
+ let result = datetime_cls.call_method1(
+ "fromtimestamp",
+ (secs as f64 + micros as f64 / 1_000_000.0, tz_utc),
+ )?;
+ Ok(Some(result))
+ }
+ None => Ok(None),
+ }
+ }
+}
Review Comment:
My agent suggested this to avoid the floating point precision loss:
```
fn timestamp_to_pyobject<'py>(
py: Python<'py>,
ts: &Timestamp,
) -> PyResult<Option<Bound<'py, PyAny>>> {
match ts.value() {
Some(dt) => {
let datetime_mod = py.import("datetime")?;
let datetime_cls = datetime_mod.getattr("datetime")?;
let tz_utc = datetime_mod.getattr("timezone")?.getattr("utc")?;
let result = datetime_cls.call1((
dt.year(),
dt.month(),
dt.day(),
dt.hour(),
dt.minute(),
dt.second(),
dt.timestamp_subsec_micros(),
tz_utc,
))?;
Ok(Some(result))
}
None => Ok(None),
}
}
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "ShreyeshArangath (via GitHub)" <gi...@apache.org>.
ShreyeshArangath commented on PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#issuecomment-4210167326
> Would you mind if we merge this into main immediately after we release DF53? That would let it soak on main for a while so more of our devs experience the change in case there are some edge cases we haven't thought of.
Yes, that's completely reasonable :) It is a lot of new code so I do expect issues, we can fix them for the next release
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] Add Python bindings for accessing ExecutionMetrics [datafusion-python]
Posted by "timsaucer (via GitHub)" <gi...@apache.org>.
timsaucer commented on PR #1381:
URL: https://github.com/apache/datafusion-python/pull/1381#issuecomment-4144415233
I was taking a look at this to see if we want to include it in DF53, but after rebase I'm getting two problems:
1. metrics are set on `DataSourceExec` even before it's executed so the unit test that would return None is failing. This does expose that we need to have a good explanation / idea of when this does exist or not.
2. `test_union_distinct` is now causing a panic
```
FAILED python/tests/test_dataframe.py::test_union_distinct - pyo3_runtime.PanicException: partition not used yet
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org