address some more comments

Author: ShreyeshArangath

Cargo.lock

@@ -35,6 +35,7 @@ tokio = { version = "1.50" }
 pyo3 = { version = "0.28" }
 pyo3-async-runtimes = { version = "0.28" }
 pyo3-log = "0.13.3"
+chrono = { version = "0.4", default-features = false }
 arrow = { version = "58" }
 arrow-array = { version = "58" }
 arrow-schema = { version = "58" }

Cargo.toml

@@ -47,6 +47,7 @@ pyo3 = { workspace = true, features = [
 ] }
 pyo3-async-runtimes = { workspace = true, features = ["tokio-runtime"] }
 pyo3-log = { workspace = true }
+chrono = { workspace = true }
 arrow = { workspace = true, features = ["pyarrow"] }
 arrow-select = { workspace = true }
 datafusion = { workspace = true, features = ["avro", "unicode_expressions"] }

crates/core/Cargo.toml

@@ -18,6 +18,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use chrono::{Datelike, Timelike};
 use datafusion::physical_plan::metrics::{Metric, MetricValue, MetricsSet, Timestamp};
 use pyo3::prelude::*;
 
@@ -88,18 +89,19 @@ impl PyMetric {
     ) -> 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),
-                )?;
+                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),
@@ -138,6 +140,7 @@ impl PyMetric {
         }
     }
 
+    #[getter]
     fn value_as_datetime<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyAny>>> {
         match self.metric.value() {
             MetricValue::StartTimestamp(ts) | MetricValue::EndTimestamp(ts) => {

crates/core/src/metrics.rs

@@ -47,8 +47,12 @@ per-partition :py:class:`~datafusion.Metric` objects via
 When Are Metrics Available?
 ---------------------------
 
-Metrics are populated only **after** the DataFrame has been executed.
-Execution is triggered by any of the terminal operations:
+Some operators (for example ``DataSourceExec``) eagerly create a
+:py:class:`~datafusion.MetricsSet` when the physical plan is built, so
+:py:meth:`~datafusion.ExecutionPlan.metrics` may return a set even before any
+rows have been processed.  However, metric **values** such as ``output_rows``
+are only meaningful **after** the DataFrame has been executed via one of the
+terminal operations:
 
 - :py:meth:`~datafusion.DataFrame.collect`
 - :py:meth:`~datafusion.DataFrame.collect_partitioned`
@@ -57,8 +61,7 @@ Execution is triggered by any of the terminal operations:
 - :py:meth:`~datafusion.DataFrame.execute_stream_partitioned`
   (metrics are available once all partition streams have been fully consumed)
 
-Calling :py:meth:`~datafusion.ExecutionPlan.collect_metrics` before execution
-returns an empty list or entries whose values are ``None`` / ``0``.
+Before execution, metric values will be ``0`` or ``None``.
 
 .. note::
 

docs/source/user-guide/dataframe/execution-metrics.rst

@@ -157,7 +157,7 @@ def to_proto(self) -> bytes:
         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.
+        """Return metrics for this plan node, or None if this plan has no MetricsSet.
 
         Some operators (e.g. DataSourceExec) eagerly initialize a MetricsSet
         when the plan is created, so this may return a set even before
@@ -304,7 +304,7 @@ def value_as_datetime(self) -> datetime.datetime | None:
         Returns ``None`` for all non-timestamp metrics and for timestamp
         metrics whose value has not been set (e.g. before execution).
         """
-        return self._raw.value_as_datetime()
+        return self._raw.value_as_datetime
 
     @property
     def partition(self) -> int | None: