fix: format Bytes-category custom metrics with byte units

datafusion/datasource-parquet/src/metrics.rs

@@ -227,8 +227,7 @@ impl ParquetFileMetrics {
         let bytes_scanned = builder
             .clone()
             .with_type(MetricType::Summary)
-            .with_category(MetricCategory::Bytes)
-            .counter("bytes_scanned", partition);
+            .bytes_counter("bytes_scanned", partition);
 
         let metadata_load_time = builder
             .clone()
@@ -349,8 +348,7 @@ impl ParquetFileMetrics {
         MetricBuilder::new(metrics)
             .with_new_label("filename", filename.to_string())
             .with_type(MetricType::Summary)
-            .with_category(MetricCategory::Bytes)
-            .counter("bytes_processed", partition)
+            .bytes_counter("bytes_processed", partition)
     }
 
     /// Record pages whose page-index pruning was skipped because the containing

datafusion/datasource-parquet/src/sink.rs

@@ -273,9 +273,8 @@ impl FileSink for ParquetSink {
         // Note: bytes_written is the sum of compressed row group sizes, which
         // may differ slightly from the actual on-disk file size (excludes footer,
         // page indexes, and other Parquet metadata overhead).
-        let bytes_written_counter = MetricBuilder::new(&self.metrics)
-            .with_category(MetricCategory::Bytes)
-            .global_counter("bytes_written");
+        let bytes_written_counter =
+            MetricBuilder::new(&self.metrics).global_bytes_counter("bytes_written");
         let elapsed_compute = MetricBuilder::new(&self.metrics).elapsed_compute(0);
 
         let parquet_opts = &self.parquet_options;

datafusion/ffi/src/tests/mod.rs

@@ -232,6 +232,47 @@ pub(crate) extern "C" fn create_exec_with_statistics() -> FFI_ExecutionPlan {
     FFI_ExecutionPlan::new(plan, None)
 }
 
+/// Registers real Bytes-category [`MetricValue::Count`] and
+/// [`MetricValue::Gauge`] metrics (via the same [`MetricBuilder::bytes_counter`]
+/// and [`MetricBuilder::bytes_gauge`] constructors production code uses for
+/// `bytes_scanned`/`stream_memory_usage`) on the returned plan, so the
+/// consumer-side integration test can exercise the category-aware
+/// byte-formatting `Display` logic through a real cross-library `metrics()`
+/// FFI call rather than only the in-process `FFI_MetricValue` conversion
+/// tests in `physical_expr::metrics`.
+///
+/// This is deliberately exported as its own top-level symbol rather than a
+/// new field on [`ForeignLibraryModule`]: that struct is public and
+/// `#[repr(C)]` with no private/gated constructor, so every field is part of
+/// its exhaustive-construction ABI surface - adding one is exactly what
+/// `cargo-semver-checks`'s `constructible_struct_adds_field` lint flags, even
+/// for a test-only, `integration-tests`-gated struct like this one. A
+/// separate exported symbol, loaded the same way [`load_module`] loads
+/// `datafusion_ffi_get_module`, avoids touching that struct's layout at all.
+///
+/// [`MetricValue::Count`]: datafusion_physical_expr_common::metrics::MetricValue::Count
+/// [`MetricValue::Gauge`]: datafusion_physical_expr_common::metrics::MetricValue::Gauge
+#[unsafe(no_mangle)]
+pub extern "C" fn datafusion_ffi_test_create_exec_with_byte_metrics() -> FFI_ExecutionPlan
+{
+    use datafusion_physical_expr_common::metrics::{
+        ExecutionPlanMetricsSet, MetricBuilder,
+    };
+
+    let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float32, false)]));
+
+    let metrics_set = ExecutionPlanMetricsSet::new();
+    MetricBuilder::new(&metrics_set)
+        .bytes_counter("bytes_scanned", 0)
+        .add(1536);
+    MetricBuilder::new(&metrics_set)
+        .bytes_gauge("stream_memory_usage", 0)
+        .add(2048);
+
+    let plan = Arc::new(EmptyExec::new(schema).with_metrics(metrics_set.clone_inner()));
+    FFI_ExecutionPlan::new(plan, None)
+}
+
 /// Thin wrapper that attaches a fixed [`Statistics`] snapshot to any inner
 /// [`TableProvider`] without changing its scan behaviour.
 #[derive(Debug)]

datafusion/ffi/src/tests/utils.rs

@@ -91,6 +91,35 @@ pub fn get_module() -> Result<ForeignLibraryModule> {
     load_module(&find_library()?)
 }
 
+/// Load [`crate::execution_plan::FFI_ExecutionPlan`] from a fresh call to
+/// `datafusion_ffi_test_create_exec_with_byte_metrics`, exported as its own
+/// top-level symbol rather than a [`ForeignLibraryModule`] field precisely so
+/// that adding this test factory never touches that struct's `#[repr(C)]`
+/// layout - see the doc comment on the exported function for the rationale.
+pub fn get_byte_metrics_exec() -> Result<crate::execution_plan::FFI_ExecutionPlan> {
+    let lib_path = find_library()?;
+
+    let lib = unsafe {
+        libloading::Library::new(&lib_path)
+            .map_err(|e| DataFusionError::External(Box::new(e)))?
+    };
+
+    let create_exec: libloading::Symbol<
+        extern "C" fn() -> crate::execution_plan::FFI_ExecutionPlan,
+    > = unsafe {
+        lib.get(b"datafusion_ffi_test_create_exec_with_byte_metrics")
+            .map_err(|e| DataFusionError::External(Box::new(e)))?
+    };
+
+    let plan = create_exec();
+
+    // Leak the library to keep it loaded for the duration of the test
+    #[expect(clippy::mem_forget)]
+    std::mem::forget(lib);
+
+    Ok(plan)
+}
+
 /// Load an independent copy of the integration-test cdylib.
 ///
 /// Copying to a unique path makes the dynamic loader create a separate image

datafusion/ffi/tests/ffi_execution_plan.rs

@@ -26,7 +26,8 @@ mod tests {
         ExecutionPlanPrivateData, FFI_ExecutionPlan, ForeignExecutionPlan,
         tests::EmptyExec,
     };
-    use datafusion_ffi::tests::utils::get_module;
+    use datafusion_ffi::tests::utils::{get_byte_metrics_exec, get_module};
+    use datafusion_physical_expr_common::metrics::{MetricCategory, MetricValue};
     use datafusion_physical_plan::execution_plan::InvariantLevel;
     use datafusion_physical_plan::{
         ChildrenPropertiesMode, ExecutionPlan, ReplaceChildrenOptions,
@@ -68,6 +69,79 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_ffi_execution_plan_byte_metrics_cross_library() -> Result<(), DataFusionError>
+    {
+        let plan = get_byte_metrics_exec()?;
+        let plan: Arc<dyn ExecutionPlan> = (&plan).try_into()?;
+        assert!(plan.is::<ForeignExecutionPlan>());
+
+        // metrics() crosses the FFI boundary for real here: `plan` is a
+        // ForeignExecutionPlan backed by a separately loaded copy of this
+        // same cdylib, so this call marshals a MetricsSet containing a
+        // Bytes-category MetricValue::Count/Gauge through FFI_MetricsSet
+        // across that boundary - not just the in-process From conversions
+        // covered by physical_expr::metrics's roundtrip tests.
+        let metrics = plan.metrics().expect("plan should report metrics");
+
+        // Assert the transported Bytes category, the generic variant/name/
+        // value, and (below) the byte-formatted display output - MetricValue
+        // is a public, already-released exhaustive enum, so this crosses the
+        // FFI boundary as a generic Count/Gauge tagged MetricCategory::Bytes
+        // rather than as a dedicated variant.
+        let mut found_bytes_count = false;
+        let mut found_bytes_gauge = false;
+        for metric in metrics.iter() {
+            let is_bytes = metric.metric_category() == Some(MetricCategory::Bytes);
+            match metric.value() {
+                MetricValue::Count { name, count }
+                    if name.as_ref() == "bytes_scanned" =>
+                {
+                    assert!(is_bytes, "bytes_scanned should be tagged Bytes category");
+                    assert_eq!(count.value(), 1536);
+                    found_bytes_count = true;
+                }
+                MetricValue::Gauge { name, gauge }
+                    if name.as_ref() == "stream_memory_usage" =>
+                {
+                    assert!(
+                        is_bytes,
+                        "stream_memory_usage should be tagged Bytes category"
+                    );
+                    assert_eq!(gauge.value(), 2048);
+                    found_bytes_gauge = true;
+                }
+                _ => {}
+            }
+        }
+        assert!(
+            found_bytes_count,
+            "expected a Bytes-category bytes_scanned Count metric, got: {metrics:?}"
+        );
+        assert!(
+            found_bytes_gauge,
+            "expected a Bytes-category stream_memory_usage Gauge metric, got: {metrics:?}"
+        );
+
+        // Also confirm the byte-unit Display formatting still applies
+        // post-round-trip.
+        let rendered: Vec<String> = metrics.iter().map(|m| m.to_string()).collect();
+        assert!(
+            rendered
+                .iter()
+                .any(|s| s == "bytes_scanned{partition=0}=1536.0 B"),
+            "bytes_scanned should survive the FFI round trip byte-formatted, got: {rendered:?}"
+        );
+        assert!(
+            rendered
+                .iter()
+                .any(|s| s == "stream_memory_usage{partition=0}=2.0 KB"),
+            "stream_memory_usage should survive the FFI round trip byte-formatted, got: {rendered:?}"
+        );
+
+        Ok(())
+    }
+
     #[test]
     fn test_ffi_execution_plan_expressions_cross_library() -> Result<(), DataFusionError>
     {

datafusion/physical-expr-common/src/metrics/builder.rs

@@ -249,6 +249,53 @@ impl<'a> MetricBuilder<'a> {
         gauge
     }
 
+    /// Consumes self and creates a new [`Count`] for recording some
+    /// arbitrary byte-measured metric of an operator (e.g. `bytes_scanned`),
+    /// always displayed with byte units regardless of [`MetricCategory`].
+    pub fn bytes_counter(
+        self,
+        counter_name: impl Into<Cow<'static, str>>,
+        partition: usize,
+    ) -> Count {
+        self.with_partition(partition)
+            .global_bytes_counter(counter_name)
+    }
+
+    /// Consumes self and creates a new [`Gauge`] for reporting some
+    /// arbitrary byte-measured metric of an operator (e.g.
+    /// `stream_memory_usage`), always displayed with byte units regardless
+    /// of [`MetricCategory`].
+    pub fn bytes_gauge(
+        self,
+        gauge_name: impl Into<Cow<'static, str>>,
+        partition: usize,
+    ) -> Gauge {
+        let gauge = Gauge::new();
+        self.with_category(MetricCategory::Bytes)
+            .with_partition(partition)
+            .build(MetricValue::Gauge {
+                name: gauge_name.into(),
+                gauge: gauge.clone(),
+            });
+        gauge
+    }
+
+    /// Consumes self and creates a new [`Count`] for recording a
+    /// byte-measured metric of an overall operator (not per partition),
+    /// always displayed with byte units regardless of [`MetricCategory`].
+    pub fn global_bytes_counter(
+        self,
+        counter_name: impl Into<Cow<'static, str>>,
+    ) -> Count {
+        let count = Count::new();
+        self.with_category(MetricCategory::Bytes)
+            .build(MetricValue::Count {
+                name: counter_name.into(),
+                count: count.clone(),
+            });
+        count
+    }
+
     /// Consumes self and creates a new [`Gauge`] for recording peak memory
     /// usage in bytes.
     pub fn peak_memory_usage(

datafusion/physical-expr-common/src/metrics/mod.rs

@@ -26,6 +26,7 @@ mod value;
 
 use datafusion_common::HashMap;
 pub use datafusion_common::format::{MetricCategory, MetricType};
+use datafusion_common::human_readable_size;
 use parking_lot::Mutex;
 use std::{
     borrow::Cow,
@@ -125,7 +126,21 @@ impl Display for Metric {
         }
 
         // and now the value
-        write!(f, "={}", self.value)
+        write!(f, "=")?;
+
+        if self.metric_category == Some(MetricCategory::Bytes) {
+            match &self.value {
+                MetricValue::Count { count, .. } => {
+                    return write!(f, "{}", human_readable_size(count.value()));
+                }
+                MetricValue::Gauge { gauge, .. } => {
+                    return write!(f, "{}", human_readable_size(gauge.value()));
+                }
+                _ => {}
+            }
+        }
+
+        write!(f, "{}", self.value)
     }
 }
 
@@ -775,6 +790,84 @@ mod tests {
         assert_eq!(metrics.sum(|_| true), Some(expected_sum));
     }
 
+    #[test]
+    fn test_bytes_counter_and_gauge_use_byte_units() {
+        let metrics = ExecutionPlanMetricsSet::new();
+
+        // A dedicated byte counter/gauge (like `bytes_scanned` or
+        // `stream_memory_usage`) must render with human_readable_size's
+        // 1024-based units (KB/MB/GB), not human_readable_count's 1000-based
+        // units (K/M/B) - see #24203. 3 GiB, chosen to clear
+        // human_readable_size's >= 2x-tier threshold for GB (below that it
+        // falls back to a large MB value).
+        let three_gib = 3 * 1024 * 1024 * 1024;
+        let bytes_scanned =
+            MetricBuilder::new(&metrics).bytes_counter("bytes_scanned", 0);
+        bytes_scanned.add(three_gib);
+
+        let stream_memory_usage =
+            MetricBuilder::new(&metrics).bytes_gauge("stream_memory_usage", 0);
+        stream_memory_usage.add(three_gib);
+
+        // ParquetSink uses the global (non-partitioned) builder for
+        // `bytes_written`, distinct from `bytes_scanned`'s partitioned
+        // `bytes_counter` above - cover that path too.
+        let bytes_written =
+            MetricBuilder::new(&metrics).global_bytes_counter("bytes_written");
+        bytes_written.add(three_gib);
+
+        // A generic Count/Gauge explicitly tagged Bytes must ALSO be
+        // byte-formatted at Display time - see #24203. `MetricValue` is a
+        // public, already-released exhaustive enum, so dedicated
+        // BytesCount/BytesGauge variants would be a SemVer break; instead
+        // `Display for Metric` reinterprets any Bytes-category Count/Gauge.
+        let generic_bytes_gauge = MetricBuilder::new(&metrics)
+            .with_category(MetricCategory::Bytes)
+            .gauge("right_input_bytes", 0);
+        generic_bytes_gauge.add(three_gib);
+
+        // A generic Count/Gauge with NO Bytes category must keep
+        // count-formatting (human_readable_count), not byte-formatting.
+        let generic_rows_gauge =
+            MetricBuilder::new(&metrics).gauge("right_input_rows", 0);
+        generic_rows_gauge.add(three_gib);
+
+        let rendered: Vec<String> = metrics
+            .clone_inner()
+            .iter()
+            .map(|m| m.to_string())
+            .collect();
+
+        assert!(
+            rendered
+                .iter()
+                .any(|s| s == "bytes_scanned{partition=0}=3.0 GB"),
+            "bytes_scanned should be byte-formatted, got: {rendered:?}"
+        );
+        assert!(
+            rendered
+                .iter()
+                .any(|s| s == "stream_memory_usage{partition=0}=3.0 GB"),
+            "stream_memory_usage should be byte-formatted, got: {rendered:?}"
+        );
+        assert!(
+            rendered.iter().any(|s| s == "bytes_written=3.0 GB"),
+            "bytes_written (global_bytes_counter, no partition) should be byte-formatted, got: {rendered:?}"
+        );
+        assert!(
+            rendered
+                .iter()
+                .any(|s| s == "right_input_bytes{partition=0}=3.0 GB"),
+            "a generic Gauge tagged Bytes should be byte-formatted, got: {rendered:?}"
+        );
+        assert!(
+            rendered
+                .iter()
+                .any(|s| s == "right_input_rows{partition=0}=3.22 B"),
+            "a generic Gauge with no Bytes category must keep count-formatting, got: {rendered:?}"
+        );
+    }
+
     #[test]
     fn test_sum_by_name_custom_metric() {
         #[derive(Debug)]

datafusion/physical-plan/src/display.rs

@@ -813,8 +813,8 @@ impl PgJsonExecutionPlanVisitor<'_> {
                 serde_json::Value::from(ms)
             }
             MetricValue::Count { count, .. } => serde_json::Value::from(count.value()),
-            MetricValue::Gauge { gauge, .. } => serde_json::Value::from(gauge.value()),
-            MetricValue::PeakMemoryUsage { gauge, .. } => {
+            MetricValue::Gauge { gauge, .. }
+            | MetricValue::PeakMemoryUsage { gauge, .. } => {
                 serde_json::Value::from(gauge.value())
             }
             MetricValue::Time { time, .. } => {

datafusion/physical-plan/src/joins/stream_join_utils.rs

@@ -727,9 +727,8 @@ impl StreamJoinMetrics {
             input_rows,
         };
 
-        let stream_memory_usage = MetricBuilder::new(metrics)
-            .with_category(MetricCategory::Bytes)
-            .gauge("stream_memory_usage", partition);
+        let stream_memory_usage =
+            MetricBuilder::new(metrics).bytes_gauge("stream_memory_usage", partition);
 
         Self {
             left,