From 635e8c7aed492d1f0d2dfa0ac073f4b7ef019a67 Mon Sep 17 00:00:00 2001 From: DevShiba Date: Mon, 10 Aug 2026 08:37:43 -0300 Subject: [PATCH 1/8] fix: format Bytes-category custom metrics with byte units MetricValue::Count/Gauge are the generic variants used by operator-defined metrics with no dedicated enum variant of their own (e.g. bytes_scanned, stream_memory_usage, bytes_written). Unlike the few metrics with dedicated variants (OutputBytes, SpilledBytes, CurrentMemoryUsage, PeakMemoryUsage), MetricValue itself does not carry the metric's category - only the wrapping Metric struct does - so Display for MetricValue's generic Count/Gauge arms had no way to know they held a byte measurement and always fell back to human_readable_count's 1000-based K/M/B/T units. Example from EXPLAIN ANALYZE before this change: output_bytes=7.5 GB bytes_scanned=1.26 B <- actually 1.26 GB, using count units (billion) Move the decision into Display for Metric, which does have both the value and the category, so a Bytes-category Count/Gauge now uses human_readable_size (1024-based KB/MB/GB/TB) like the dedicated byte variants already do. Rows/Timing-category and uncategorized generic metrics are untouched, since human_readable_count was already correct for them. Also fixes the same bug for stream_memory_usage and bytes_written, which share the same root cause but weren't mentioned in the original report. Updated the 14 hardcoded bytes_scanned expected values across two sqllogictest files to match the corrected format. Deliberately did this by hand rather than via `--complete`, which would have also baked in non-deterministic timing values that the tests intentionally wildcard with . Closes #24203 --- .../physical-expr-common/src/metrics/mod.rs | 93 ++++++++++++++++++- .../dynamic_filter_pushdown_config.slt | 2 +- .../parquet_nested_schema_pruning.slt | 26 +++--- 3 files changed, 106 insertions(+), 15 deletions(-) diff --git a/datafusion/physical-expr-common/src/metrics/mod.rs b/datafusion/physical-expr-common/src/metrics/mod.rs index 146c039c75f6a..3f2bc156f0d35 100644 --- a/datafusion/physical-expr-common/src/metrics/mod.rs +++ b/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,29 @@ impl Display for Metric { } // and now the value - write!(f, "={}", self.value) + write!(f, "=")?; + + // MetricValue::Count/Gauge are used for operator-defined metrics with + // no dedicated variant of their own (e.g. `bytes_scanned`), so unlike + // OutputBytes/SpilledBytes/CurrentMemoryUsage/PeakMemoryUsage, their + // Display impl has no way to know they hold a byte measurement - + // MetricValue does not carry the category, only Metric does. Without + // this, a Bytes-category Count/Gauge silently falls back to + // human_readable_count's 1000-based K/M/B/T instead of the correct + // 1024-based KB/MB/GB/TB. + 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) } } @@ -773,6 +796,74 @@ mod tests { assert_eq!(metrics.sum(|_| true), Some(expected_sum)); } + #[test] + fn test_display_generic_count_respects_bytes_category() { + let metrics = ExecutionPlanMetricsSet::new(); + + // A Bytes-category custom counter (like `bytes_scanned`) 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) + .with_category(MetricCategory::Bytes) + .counter("bytes_scanned", 0); + bytes_scanned.add(three_gib); + + // A Rows-category custom counter must keep using human_readable_count, + // since that is already the correct formatter for it. + let output_rows_like = MetricBuilder::new(&metrics) + .with_category(MetricCategory::Rows) + .counter("right_input_rows", 0); + output_rows_like.add(three_gib); + + let rendered: Vec = 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 == "right_input_rows{partition=0}=3.22 B"), + "a Rows-category counter should keep count-formatting, got: {rendered:?}" + ); + } + + #[test] + fn test_display_generic_gauge_respects_bytes_category() { + let metrics = ExecutionPlanMetricsSet::new(); + + // A Bytes-category custom gauge (like `stream_memory_usage`) must + // also render with byte units, not human_readable_count's units. + let three_gib = 3 * 1024 * 1024 * 1024; + let stream_memory_usage = MetricBuilder::new(&metrics) + .with_category(MetricCategory::Bytes) + .gauge("stream_memory_usage", 0); + stream_memory_usage.add(three_gib); + + let rendered: Vec = metrics + .clone_inner() + .iter() + .map(|m| m.to_string()) + .collect(); + + assert!( + rendered + .iter() + .any(|s| s == "stream_memory_usage{partition=0}=3.0 GB"), + "stream_memory_usage should be byte-formatted, got: {rendered:?}" + ); + } + #[test] #[should_panic(expected = "Mismatched metric types. Can not aggregate Count")] fn test_bad_sum() { diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index eec6e5ae179bc..0d4a4d8ce89f7 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -104,7 +104,7 @@ Plan with Metrics 03)----ProjectionExec: expr=[id@0 as id, value@1 as v, value@1 + id@0 as name], metrics=[output_rows=10, ] 04)------FilterExec: value@1 > 3, metrics=[output_rows=10, , selectivity=100% (10/10)] 05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, metrics=[output_rows=10, ] -06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], dynamic_rg_pruning=eligible, pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=210, page_index_load_skipped=1, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio=18.31% (210/1.15 K)] +06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], dynamic_rg_pruning=eligible, pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=210.0 B, page_index_load_skipped=1, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio=18.31% (210/1.15 K)] statement ok set datafusion.explain.analyze_level = dev; diff --git a/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt b/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt index d936a89beb9f7..78b8628e8ba22 100644 --- a/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt +++ b/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt @@ -107,23 +107,23 @@ LOCATION 'test_files/scratch/parquet_nested_schema_pruning/wide.parquet'; query TT explain analyze select events from narrow; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=172] +Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=172.0 B] query TT explain analyze select events from full_schema; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=312] +Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=312.0 B] # Same for the top-level struct column: the clipped read drops `pad`. query TT explain analyze select s from narrow; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=146] +Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=146.0 B] query TT explain analyze select s from full_schema; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=219] +Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=219.0 B] # `get_field` on a schema-narrowed struct becomes `get_field(CAST(s), 'x')`; # the read clips to the cast target (every field the *narrow* schema @@ -135,19 +135,19 @@ Plan with Metrics DataSourceExec: metrics=[output_rows=3, metrics=[output_rows=3, bytes_scanned=146] +Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=146.0 B] # Mixed access -- the whole (narrowed) column and a subfield of it -- still # reads only the narrow schema's leaves. query TT explain analyze select s, s['y'] from narrow; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=146] +Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=146.0 B] query TT explain analyze select s, s['y'] from full_schema; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=219] +Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=219.0 B] # `SELECT *` goes through the same clipped read as an explicit projection. @@ -349,12 +349,12 @@ ORDER BY id, i['group_id']; query TT explain analyze select events from two_level_narrow; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=2, bytes_scanned=381] +Plan with Metrics DataSourceExec: metrics=[output_rows=2, bytes_scanned=381.0 B] query TT explain analyze select events from two_level_full; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=2, bytes_scanned=1.05 K] +Plan with Metrics DataSourceExec: metrics=[output_rows=2, bytes_scanned=1047.0 B] statement ok DROP TABLE two_level_narrow; @@ -459,12 +459,12 @@ SELECT id, s FROM narrow WHERE id >= 2 ORDER BY id; query TT explain analyze select s from narrow where id >= 2; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=2, bytes_scanned=219] +Plan with Metrics DataSourceExec: metrics=[output_rows=2, bytes_scanned=219.0 B] query TT explain analyze select s from full_schema where id >= 2; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=2, bytes_scanned=292] +Plan with Metrics DataSourceExec: metrics=[output_rows=2, bytes_scanned=292.0 B] statement ok set datafusion.execution.parquet.pushdown_filters = false; @@ -540,12 +540,12 @@ NULL NULL query TT explain analyze select CAST(s AS STRUCT) AS q0, CAST(s AS STRUCT) AS q1 from exact; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=219] +Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=219.0 B] query TT explain analyze select s from exact; ---- -Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=219] +Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=219.0 B] statement ok DROP TABLE exact; From 06706680e9d03d86f4edec5374c7c13c7dc28583 Mon Sep 17 00:00:00 2001 From: DevShiba Date: Mon, 10 Aug 2026 14:01:12 -0300 Subject: [PATCH 2/8] fix: register bytes_scanned/bytes_written/stream_memory_usage as dedicated byte-typed metrics Address review feedback: instead of reinterpreting a Bytes-category generic Count/Gauge at Display time, register the three affected metrics as their own dedicated MetricValue variants, matching the existing OutputBytes/SpilledBytes/PeakMemoryUsage precedent. Adds two new MetricValue variants: - BytesCount { name, count }: a named, byte-formatted Count, for bytes_scanned and bytes_written. - BytesGauge { name, gauge }: a named, byte-formatted Gauge, for stream_memory_usage. BytesGauge is deliberately not folded into the existing PeakMemoryUsage variant even though they're structurally identical. PeakMemoryUsage's existing call sites (peak_mem_used, max_mem_used, build_mem_used) are all monotonically-increasing accumulators, so "peak" is accurate for them. stream_memory_usage is `.set()` to a capacity that can shrink as a sliding window prunes old rows - it is not a peak, and mislabeling it as one would be misleading to anyone reading EXPLAIN output or the enum variant itself. Also updates the two places that previously matched on MetricValue exhaustively and would otherwise have silently mis-handled the new variants: - MetricsSet::sum_by_name(), used by bytes_scanned's own tests via sum_by_name("bytes_scanned"). - display.rs's JSON EXPLAIN metric_value_to_json(): without an explicit arm, the new variants would have fallen through to the string-fallback branch, turning bytes_scanned's JSON output from a number into a string like "3.0 GB" - a real behavior change for any JSON EXPLAIN consumer. - The FFI crate's FFI_MetricValue, which documents that new variants must be appended at the end since variant order is part of its stable ABI - done here, with matching conversions and round-trip test coverage in both directions. Display for Metric goes back to the plain `write!(f, "{}", self.value)` it had before the previous commit - MetricValue no longer needs help from Metric's category to know it's holding bytes. Rendered EXPLAIN ANALYZE output is byte-for-byte unchanged (reran the full 502-file sqllogictest suite to confirm), so none of the previously-updated .slt fixtures needed further changes. --- datafusion/datasource-parquet/src/metrics.rs | 3 +- datafusion/datasource-parquet/src/sink.rs | 5 +- datafusion/ffi/src/physical_expr/metrics.rs | 38 ++++++++ .../src/metrics/builder.rs | 47 ++++++++++ .../physical-expr-common/src/metrics/mod.rs | 88 ++++++------------- .../physical-expr-common/src/metrics/value.rs | 80 ++++++++++++++--- datafusion/physical-plan/src/display.rs | 9 +- .../src/joins/stream_join_utils.rs | 5 +- 8 files changed, 194 insertions(+), 81 deletions(-) diff --git a/datafusion/datasource-parquet/src/metrics.rs b/datafusion/datasource-parquet/src/metrics.rs index cbdcb73196b17..4207681d3d2c9 100644 --- a/datafusion/datasource-parquet/src/metrics.rs +++ b/datafusion/datasource-parquet/src/metrics.rs @@ -141,8 +141,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() diff --git a/datafusion/datasource-parquet/src/sink.rs b/datafusion/datasource-parquet/src/sink.rs index 53f6f1e6b4323..b6390b4618398 100644 --- a/datafusion/datasource-parquet/src/sink.rs +++ b/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; diff --git a/datafusion/ffi/src/physical_expr/metrics.rs b/datafusion/ffi/src/physical_expr/metrics.rs index 763cc3f079a01..0d62813a31774 100644 --- a/datafusion/ffi/src/physical_expr/metrics.rs +++ b/datafusion/ffi/src/physical_expr/metrics.rs @@ -177,6 +177,14 @@ pub enum FFI_MetricValue { name: SString, gauge: u64, }, + BytesCount { + name: SString, + count: u64, + }, + BytesGauge { + name: SString, + gauge: u64, + }, } // ----------------------------------------------------------------------------- @@ -436,6 +444,14 @@ impl From<&MetricValue> for FFI_MetricValue { name: SString::from(name.as_ref()), gauge: gauge.value() as u64, }, + MetricValue::BytesCount { name, count } => Self::BytesCount { + name: SString::from(name.as_ref()), + count: count.value() as u64, + }, + MetricValue::BytesGauge { name, gauge } => Self::BytesGauge { + name: SString::from(name.as_ref()), + gauge: gauge.value() as u64, + }, MetricValue::Time { name, time } => Self::Time { name: SString::from(name.as_ref()), time_ns: time.value() as u64, @@ -496,6 +512,14 @@ impl From for MetricValue { name: Cow::Owned(name.into()), gauge: gauge_from_value(gauge), }, + FFI_MetricValue::BytesCount { name, count } => Self::BytesCount { + name: Cow::Owned(name.into()), + count: count_from_value(count), + }, + FFI_MetricValue::BytesGauge { name, gauge } => Self::BytesGauge { + name: Cow::Owned(name.into()), + gauge: gauge_from_value(gauge), + }, FFI_MetricValue::Time { name, time_ns } => Self::Time { name: Cow::Owned(name.into()), time: time_from_nanos(time_ns), @@ -646,6 +670,20 @@ mod tests { gauge: peak_memory, }); + let bytes_count = Count::new(); + bytes_count.add(55); + assert_value_roundtrip(MetricValue::BytesCount { + name: Cow::Borrowed("bytes_scanned"), + count: bytes_count, + }); + + let bytes_gauge = Gauge::new(); + bytes_gauge.add(66); + assert_value_roundtrip(MetricValue::BytesGauge { + name: Cow::Borrowed("stream_memory_usage"), + gauge: bytes_gauge, + }); + let time = Time::new(); time.add_duration(std::time::Duration::from_nanos(33)); assert_value_roundtrip(MetricValue::Time { diff --git a/datafusion/physical-expr-common/src/metrics/builder.rs b/datafusion/physical-expr-common/src/metrics/builder.rs index 7d5a18f535369..189cb6750bf83 100644 --- a/datafusion/physical-expr-common/src/metrics/builder.rs +++ b/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>, + 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>, + partition: usize, + ) -> Gauge { + let gauge = Gauge::new(); + self.with_category(MetricCategory::Bytes) + .with_partition(partition) + .build(MetricValue::BytesGauge { + 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>, + ) -> Count { + let count = Count::new(); + self.with_category(MetricCategory::Bytes) + .build(MetricValue::BytesCount { + 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( diff --git a/datafusion/physical-expr-common/src/metrics/mod.rs b/datafusion/physical-expr-common/src/metrics/mod.rs index 3f2bc156f0d35..8659d8d1b0055 100644 --- a/datafusion/physical-expr-common/src/metrics/mod.rs +++ b/datafusion/physical-expr-common/src/metrics/mod.rs @@ -26,7 +26,6 @@ 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, @@ -128,26 +127,6 @@ impl Display for Metric { // and now the value write!(f, "=")?; - // MetricValue::Count/Gauge are used for operator-defined metrics with - // no dedicated variant of their own (e.g. `bytes_scanned`), so unlike - // OutputBytes/SpilledBytes/CurrentMemoryUsage/PeakMemoryUsage, their - // Display impl has no way to know they hold a byte measurement - - // MetricValue does not carry the category, only Metric does. Without - // this, a Bytes-category Count/Gauge silently falls back to - // human_readable_count's 1000-based K/M/B/T instead of the correct - // 1024-based KB/MB/GB/TB. - 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) } } @@ -322,6 +301,7 @@ impl MetricsSet { pub fn sum_by_name(&self, metric_name: &str) -> Option { self.sum(|m| match m.value() { MetricValue::Count { name, .. } => name == metric_name, + MetricValue::BytesCount { name, .. } => name == metric_name, MetricValue::Time { name, .. } => name == metric_name, MetricValue::OutputRows(_) => false, MetricValue::ElapsedCompute(_) => false, @@ -332,6 +312,7 @@ impl MetricsSet { MetricValue::SpilledRows(_) => false, MetricValue::CurrentMemoryUsage(_) => false, MetricValue::Gauge { name, .. } => name == metric_name, + MetricValue::BytesGauge { name, .. } => name == metric_name, MetricValue::PeakMemoryUsage { name, .. } => name == metric_name, MetricValue::StartTimestamp(_) => false, MetricValue::EndTimestamp(_) => false, @@ -797,26 +778,33 @@ mod tests { } #[test] - fn test_display_generic_count_respects_bytes_category() { + fn test_bytes_counter_and_gauge_use_byte_units() { let metrics = ExecutionPlanMetricsSet::new(); - // A Bytes-category custom counter (like `bytes_scanned`) 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). + // 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) - .with_category(MetricCategory::Bytes) - .counter("bytes_scanned", 0); + let bytes_scanned = + MetricBuilder::new(&metrics).bytes_counter("bytes_scanned", 0); bytes_scanned.add(three_gib); - // A Rows-category custom counter must keep using human_readable_count, - // since that is already the correct formatter for it. - let output_rows_like = MetricBuilder::new(&metrics) - .with_category(MetricCategory::Rows) - .counter("right_input_rows", 0); - output_rows_like.add(three_gib); + let stream_memory_usage = + MetricBuilder::new(&metrics).bytes_gauge("stream_memory_usage", 0); + stream_memory_usage.add(three_gib); + + // A generic (non-byte) Count/Gauge tagged Bytes must NOT be + // reinterpreted based on category - only the dedicated + // BytesCount/BytesGauge variants get byte formatting. This is the + // explicit-typing fix for #24203, replacing an earlier approach that + // reinterpreted any Bytes-category Count/Gauge at Display time. + let generic_bytes_gauge = MetricBuilder::new(&metrics) + .with_category(MetricCategory::Bytes) + .gauge("right_input_rows", 0); + generic_bytes_gauge.add(three_gib); let rendered: Vec = metrics .clone_inner() @@ -833,34 +821,14 @@ mod tests { assert!( rendered .iter() - .any(|s| s == "right_input_rows{partition=0}=3.22 B"), - "a Rows-category counter should keep count-formatting, got: {rendered:?}" + .any(|s| s == "stream_memory_usage{partition=0}=3.0 GB"), + "stream_memory_usage should be byte-formatted, got: {rendered:?}" ); - } - - #[test] - fn test_display_generic_gauge_respects_bytes_category() { - let metrics = ExecutionPlanMetricsSet::new(); - - // A Bytes-category custom gauge (like `stream_memory_usage`) must - // also render with byte units, not human_readable_count's units. - let three_gib = 3 * 1024 * 1024 * 1024; - let stream_memory_usage = MetricBuilder::new(&metrics) - .with_category(MetricCategory::Bytes) - .gauge("stream_memory_usage", 0); - stream_memory_usage.add(three_gib); - - let rendered: Vec = metrics - .clone_inner() - .iter() - .map(|m| m.to_string()) - .collect(); - assert!( rendered .iter() - .any(|s| s == "stream_memory_usage{partition=0}=3.0 GB"), - "stream_memory_usage should be byte-formatted, got: {rendered:?}" + .any(|s| s == "right_input_rows{partition=0}=3.22 B"), + "a generic Gauge must keep count-formatting even when tagged Bytes, got: {rendered:?}" ); } diff --git a/datafusion/physical-expr-common/src/metrics/value.rs b/datafusion/physical-expr-common/src/metrics/value.rs index 232fefcc5f47e..c2c85c17c754c 100644 --- a/datafusion/physical-expr-common/src/metrics/value.rs +++ b/datafusion/physical-expr-common/src/metrics/value.rs @@ -665,6 +665,16 @@ pub enum MetricValue { /// The value of the metric count: Count, }, + /// Operator defined count representing a size in bytes (e.g. + /// `bytes_scanned`, `bytes_written`). Like [`Self::Count`], but always + /// displayed with [`human_readable_size`]'s 1024-based byte units + /// (KB/MB/GB/TB) instead of [`human_readable_count`]'s 1000-based units. + BytesCount { + /// The provided name of this metric + name: Cow<'static, str>, + /// The value of the metric, in bytes + count: Count, + }, /// Operator defined gauge. Gauge { /// The provided name of this metric @@ -672,6 +682,18 @@ pub enum MetricValue { /// The value of the metric gauge: Gauge, }, + /// Operator defined gauge representing a size in bytes (e.g. + /// `stream_memory_usage`) that is not necessarily monotonically + /// increasing. Like [`Self::Gauge`], but always displayed with + /// [`human_readable_size`]'s byte units. Unlike [`Self::PeakMemoryUsage`], + /// this does not imply the value only ever grows - use + /// [`Self::PeakMemoryUsage`] for that. + BytesGauge { + /// The provided name of this metric + name: Cow<'static, str>, + /// The value of the metric, in bytes + gauge: Gauge, + }, /// Operator defined peak memory usage in bytes. PeakMemoryUsage { /// The provided name of this metric @@ -744,6 +766,13 @@ impl PartialEq for MetricValue { name: other_name, count: other_count, }, + ) + | ( + MetricValue::BytesCount { name, count }, + MetricValue::BytesCount { + name: other_name, + count: other_count, + }, ) => name == other_name && count == other_count, ( MetricValue::Gauge { name, gauge }, @@ -752,6 +781,13 @@ impl PartialEq for MetricValue { gauge: other_gauge, }, ) + | ( + MetricValue::BytesGauge { name, gauge }, + MetricValue::BytesGauge { + name: other_name, + gauge: other_gauge, + }, + ) | ( MetricValue::PeakMemoryUsage { name, gauge }, MetricValue::PeakMemoryUsage { @@ -823,10 +859,10 @@ impl MetricValue { Self::SpilledRows(_) => "spilled_rows", Self::CurrentMemoryUsage(_) => "mem_used", Self::ElapsedCompute(_) => "elapsed_compute", - Self::Count { name, .. } => name.borrow(), - Self::Gauge { name, .. } | Self::PeakMemoryUsage { name, .. } => { - name.borrow() - } + Self::Count { name, .. } | Self::BytesCount { name, .. } => name.borrow(), + Self::Gauge { name, .. } + | Self::BytesGauge { name, .. } + | Self::PeakMemoryUsage { name, .. } => name.borrow(), Self::Time { name, .. } => name.borrow(), Self::StartTimestamp(_) => "start_timestamp", Self::EndTimestamp(_) => "end_timestamp", @@ -848,10 +884,10 @@ impl MetricValue { Self::SpilledRows(count) => count.value(), Self::CurrentMemoryUsage(used) => used.value(), Self::ElapsedCompute(time) => time.value(), - Self::Count { count, .. } => count.value(), - Self::Gauge { gauge, .. } | Self::PeakMemoryUsage { gauge, .. } => { - gauge.value() - } + Self::Count { count, .. } | Self::BytesCount { count, .. } => count.value(), + Self::Gauge { gauge, .. } + | Self::BytesGauge { gauge, .. } + | Self::PeakMemoryUsage { gauge, .. } => gauge.value(), Self::Time { time, .. } => time.value(), Self::StartTimestamp(timestamp) => timestamp .value() @@ -889,10 +925,18 @@ impl MetricValue { name: name.clone(), count: Count::new(), }, + Self::BytesCount { name, .. } => Self::BytesCount { + name: name.clone(), + count: Count::new(), + }, Self::Gauge { name, .. } => Self::Gauge { name: name.clone(), gauge: Gauge::new(), }, + Self::BytesGauge { name, .. } => Self::BytesGauge { + name: name.clone(), + gauge: Gauge::new(), + }, Self::PeakMemoryUsage { name, .. } => Self::PeakMemoryUsage { name: name.clone(), gauge: Gauge::new(), @@ -948,6 +992,12 @@ impl MetricValue { Self::Count { count: other_count, .. }, + ) + | ( + Self::BytesCount { count, .. }, + Self::BytesCount { + count: other_count, .. + }, ) => count.add(other_count.value()), (Self::CurrentMemoryUsage(gauge), Self::CurrentMemoryUsage(other_gauge)) | ( @@ -956,6 +1006,12 @@ impl MetricValue { gauge: other_gauge, .. }, ) + | ( + Self::BytesGauge { gauge, .. }, + Self::BytesGauge { + gauge: other_gauge, .. + }, + ) | ( Self::PeakMemoryUsage { gauge, .. }, Self::PeakMemoryUsage { @@ -1058,7 +1114,9 @@ impl MetricValue { _ => 14, }, Self::PeakMemoryUsage { .. } => 13, + Self::BytesCount { .. } => 14, Self::Gauge { .. } => 15, + Self::BytesGauge { .. } => 15, Self::Time { .. } => 16, Self::Ratio { .. } => 17, Self::StartTimestamp(_) => 18, // show timestamps last @@ -1084,7 +1142,9 @@ impl Display for MetricValue { | Self::Count { count, .. } => { write!(f, "{count}") } - Self::SpilledBytes(count) | Self::OutputBytes(count) => { + Self::SpilledBytes(count) + | Self::OutputBytes(count) + | Self::BytesCount { count, .. } => { let readable_count = human_readable_size(count.value()); write!(f, "{readable_count}") } @@ -1093,7 +1153,7 @@ impl Display for MetricValue { let readable_size = human_readable_size(gauge.value()); write!(f, "{readable_size}") } - Self::PeakMemoryUsage { gauge, .. } => { + Self::PeakMemoryUsage { gauge, .. } | Self::BytesGauge { gauge, .. } => { let readable_size = human_readable_size(gauge.value()); write!(f, "{readable_size}") } diff --git a/datafusion/physical-plan/src/display.rs b/datafusion/physical-plan/src/display.rs index 2370e3e6ce6ec..3c0a5abcbe356 100644 --- a/datafusion/physical-plan/src/display.rs +++ b/datafusion/physical-plan/src/display.rs @@ -820,9 +820,12 @@ impl PgJsonExecutionPlanVisitor<'_> { let ms = (t.value() as f64) / 1_000_000.0; 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::Count { count, .. } | MetricValue::BytesCount { count, .. } => { + serde_json::Value::from(count.value()) + } + MetricValue::Gauge { gauge, .. } + | MetricValue::BytesGauge { gauge, .. } + | MetricValue::PeakMemoryUsage { gauge, .. } => { serde_json::Value::from(gauge.value()) } MetricValue::Time { time, .. } => { diff --git a/datafusion/physical-plan/src/joins/stream_join_utils.rs b/datafusion/physical-plan/src/joins/stream_join_utils.rs index 05a56d241102e..424bcf438da4a 100644 --- a/datafusion/physical-plan/src/joins/stream_join_utils.rs +++ b/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, From f9a834a61a6e17bf433d8696fd0f2236abc39e97 Mon Sep 17 00:00:00 2001 From: DevShiba Date: Tue, 18 Aug 2026 11:11:31 -0300 Subject: [PATCH 3/8] fix: recognize BytesCount in assert_bytes_scanned test helper datafusion/core/src/datasource/file_format/parquet.rs's own assert_bytes_scanned test helper pattern-matched MetricValue::Count directly (bypassing MetricsSet::sum_by_name, which was already updated for the new BytesCount variant), so it stopped finding the bytes_scanned metric once its registration moved to MetricValue::BytesCount, failing capture_bytes_scanned_metric in cargo test (amd64), cargo test hash collisions, and cargo test extended_tests - all three were the same helper, just exercised under different feature-flag runs. --- datafusion/core/src/datasource/file_format/parquet.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/datafusion/core/src/datasource/file_format/parquet.rs b/datafusion/core/src/datasource/file_format/parquet.rs index 0f5db4a057d76..8b7925dd6c539 100644 --- a/datafusion/core/src/datasource/file_format/parquet.rs +++ b/datafusion/core/src/datasource/file_format/parquet.rs @@ -1171,7 +1171,7 @@ mod tests { let actual = exec .metrics() .expect("Metrics not recorded") - .sum(|metric| matches!(metric.value(), MetricValue::Count { name, .. } if name == "bytes_scanned")) + .sum(|metric| matches!(metric.value(), MetricValue::Count { name, .. } | MetricValue::BytesCount { name, .. } if name == "bytes_scanned")) .map(|t| t.as_usize()) .expect("bytes_scanned metric not recorded"); From da06bd677309ba252ef67c0d3c7f431c80df31a0 Mon Sep 17 00:00:00 2001 From: DevShiba Date: Tue, 18 Aug 2026 13:02:47 -0300 Subject: [PATCH 4/8] fix: recognize BytesCount in remaining hand-rolled metric matchers Two more test helpers pattern-matched MetricValue::Count directly instead of going through the already-fixed MetricsSet::sum_by_name, each in a different crate: - datafusion/core/tests/parquet/page_pruning.rs's cast_count_metric, which caused without_pushdown_filter to panic on .unwrap() (the actual CI failure). - datafusion/datasource-parquet/src/opener/mod.rs's counter_metric_value, currently only exercised with a plain Count metric so it wasn't failing, but would have silently returned 0 instead of the real value the moment it's used for bytes_scanned. Did a full repo-wide sweep this time, not just the one failing site: grepped every MetricValue::Count and MetricValue::Gauge match plus every "bytes_scanned"/"bytes_written"/"stream_memory_usage" string reference, and manually vetted each one against the new BytesCount/ BytesGauge variants. These two were the only remaining gaps. Before pushing, ran the exact three commands the failing CI jobs use (from .github/workflows/rust.yml and extended.yml) locally, twice: cargo test --profile ci --exclude datafusion-examples \ --exclude ffi_example_table_provider --exclude datafusion-cli \ --workspace --lib --tests --bins \ --features serde,avro,json,backtrace,integration-tests,parquet_encryption,substrait (cd datafusion && cargo test --profile ci \ --exclude datafusion-examples --exclude datafusion-benchmarks \ --exclude datafusion-sqllogictest --exclude datafusion-cli \ --workspace --lib --tests --features=force_hash_collisions,avro) cargo test --profile ci --exclude datafusion-examples \ --exclude datafusion-benchmarks --exclude datafusion-cli \ --workspace --lib --tests --bins \ --features avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption All three: zero failures. Also ran the exact CI clippy script (ci/scripts/rust_clippy.sh, -D warnings) and cargo fmt --check clean, and confirmed the working tree has no stray files after the test runs (the same check CI itself runs). --- datafusion/core/tests/parquet/page_pruning.rs | 4 +++- datafusion/datasource-parquet/src/opener/mod.rs | 3 ++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/datafusion/core/tests/parquet/page_pruning.rs b/datafusion/core/tests/parquet/page_pruning.rs index 372a7a601d492..4966355e906a4 100644 --- a/datafusion/core/tests/parquet/page_pruning.rs +++ b/datafusion/core/tests/parquet/page_pruning.rs @@ -1053,7 +1053,9 @@ async fn test_pages_with_null_values() { fn cast_count_metric(metric: MetricValue) -> Option { match metric { - MetricValue::Count { count, .. } => Some(count.value()), + MetricValue::Count { count, .. } | MetricValue::BytesCount { count, .. } => { + Some(count.value()) + } _ => None, } } diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index 693e9bd2cbf31..d8864856abad1 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -2393,7 +2393,8 @@ mod test { .clone_inner() .sum_by_name(name) .map(|metric| match metric { - MetricValue::Count { count, .. } => count.value(), + MetricValue::Count { count, .. } + | MetricValue::BytesCount { count, .. } => count.value(), _ => 0, }) .unwrap_or(0) From 4e0b9049355ef305814c6bbe0541663f61846019 Mon Sep 17 00:00:00 2001 From: DevShiba Date: Wed, 19 Aug 2026 11:32:18 -0300 Subject: [PATCH 5/8] test: add real cross-library FFI round trip for BytesCount/BytesGauge Addresses kosiew's review on PR #24218: - Add a real cdylib integration test (test_ffi_execution_plan_byte_metrics_cross_library) that registers genuine BytesCount/BytesGauge metrics (via the same MetricBuilder::bytes_counter/bytes_gauge production code uses) on a plan served from a separately loaded copy of the datafusion-ffi cdylib, then calls .metrics() through the real ForeignExecutionPlan vtable to confirm both variants survive an actual FFI_MetricValue round trip across the dylib boundary - not just the in-process From conversions already covered by physical_expr::metrics's roundtrip tests. - New create_exec_with_byte_metrics factory function and ForeignLibraryModule entry, following the existing create_exec_with_statistics pattern. - Also cover global_bytes_counter("bytes_written") in test_bytes_counter_and_gauge_use_byte_units, since ParquetSink uses the global (non-partitioned) builder while the existing coverage only exercised the partitioned bytes_counter path. Verified: the exact three commands the previously-failing CI jobs run (cargo test (amd64), hash collisions, extended_tests), the exact CI clippy script, cargo fmt --check, and the full datafusion-ffi integration-tests suite - all clean. --- datafusion/ffi/src/tests/mod.rs | 32 ++++++++++++++++++ datafusion/ffi/tests/ffi_execution_plan.rs | 33 +++++++++++++++++++ .../physical-expr-common/src/metrics/mod.rs | 11 +++++++ 3 files changed, 76 insertions(+) diff --git a/datafusion/ffi/src/tests/mod.rs b/datafusion/ffi/src/tests/mod.rs index fbc3e83ba49fc..110d14b000a4c 100644 --- a/datafusion/ffi/src/tests/mod.rs +++ b/datafusion/ffi/src/tests/mod.rs @@ -120,6 +120,8 @@ pub struct ForeignLibraryModule { pub create_exec_with_statistics: extern "C" fn() -> FFI_ExecutionPlan, + pub create_exec_with_byte_metrics: extern "C" fn() -> FFI_ExecutionPlan, + pub create_table_with_statistics: extern "C" fn(codec: FFI_LogicalExtensionCodec) -> FFI_TableProvider, @@ -232,6 +234,35 @@ pub(crate) extern "C" fn create_exec_with_statistics() -> FFI_ExecutionPlan { FFI_ExecutionPlan::new(plan, None) } +/// Registers real [`MetricValue::BytesCount`] and [`MetricValue::BytesGauge`] +/// 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 these variants through a real +/// cross-library `metrics()` FFI call rather than only the in-process +/// `FFI_MetricValue` conversion tests in `physical_expr::metrics`. +/// +/// [`MetricValue::BytesCount`]: datafusion_physical_expr_common::metrics::MetricValue::BytesCount +/// [`MetricValue::BytesGauge`]: datafusion_physical_expr_common::metrics::MetricValue::BytesGauge +pub(crate) extern "C" fn 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)] @@ -362,6 +393,7 @@ pub extern "C" fn datafusion_ffi_get_module() -> ForeignLibraryModule { create_exec_with_expressions, create_exec_with_dynamic_expressions, create_exec_with_statistics, + create_exec_with_byte_metrics, create_table_with_statistics, create_physical_optimizer_rule: physical_optimizer::create_physical_optimizer_rule, diff --git a/datafusion/ffi/tests/ffi_execution_plan.rs b/datafusion/ffi/tests/ffi_execution_plan.rs index 4067d7eb49b2a..c087854df3253 100644 --- a/datafusion/ffi/tests/ffi_execution_plan.rs +++ b/datafusion/ffi/tests/ffi_execution_plan.rs @@ -68,6 +68,39 @@ mod tests { Ok(()) } + #[test] + fn test_ffi_execution_plan_byte_metrics_cross_library() -> Result<(), DataFusionError> + { + let module = get_module()?; + let plan = (module.create_exec_with_byte_metrics)(); + let plan: Arc = (&plan).try_into()?; + assert!(plan.is::()); + + // 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 + // MetricValue::BytesCount/BytesGauge 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"); + let rendered: Vec = metrics.iter().map(|m| m.to_string()).collect(); + + assert!( + rendered + .iter() + .any(|s| s == "bytes_scanned{partition=0}=1536.0 B"), + "BytesCount should survive the FFI round trip byte-formatted, got: {rendered:?}" + ); + assert!( + rendered + .iter() + .any(|s| s == "stream_memory_usage{partition=0}=2.0 KB"), + "BytesGauge should survive the FFI round trip byte-formatted, got: {rendered:?}" + ); + + Ok(()) + } + #[test] fn test_ffi_execution_plan_expressions_cross_library() -> Result<(), DataFusionError> { diff --git a/datafusion/physical-expr-common/src/metrics/mod.rs b/datafusion/physical-expr-common/src/metrics/mod.rs index 8659d8d1b0055..f3be2eb4c7ad7 100644 --- a/datafusion/physical-expr-common/src/metrics/mod.rs +++ b/datafusion/physical-expr-common/src/metrics/mod.rs @@ -796,6 +796,13 @@ mod tests { 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 (non-byte) Count/Gauge tagged Bytes must NOT be // reinterpreted based on category - only the dedicated // BytesCount/BytesGauge variants get byte formatting. This is the @@ -824,6 +831,10 @@ mod tests { .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() From 6178dcd5dd0b4f9cc7eca5f1da01c820e46cd47c Mon Sep 17 00:00:00 2001 From: DevShiba Date: Thu, 20 Aug 2026 08:56:45 -0300 Subject: [PATCH 6/8] fix: address kosiew's ABI and discriminant-coverage feedback Two follow-ups from review on PR #24218: 1. The previous commit added create_exec_with_byte_metrics as a new field on ForeignLibraryModule. That struct is public, #[repr(C)], and has no private/gated constructor, so every field is part of its exhaustive-construction ABI surface even though the struct only exists behind the integration-tests feature - exactly what cargo-semver-checks's constructible_struct_adds_field lint flagged. Reworked the new test factory as its own top-level exported symbol (datafusion_ffi_test_create_exec_with_byte_metrics, loaded via a new get_byte_metrics_exec() in tests/utils.rs, mirroring how get_module() already loads datafusion_ffi_get_module) instead of extending the struct. ForeignLibraryModule's fields are now byte-for-byte identical to the pre-fix commit. 2. test_ffi_execution_plan_byte_metrics_cross_library previously only compared the Display-rendered strings, which doesn't prove the BytesCount/BytesGauge discriminants themselves - as opposed to some other variant that happens to render the same text - survived the FFI round trip. Now matches metric.value() against MetricValue::BytesCount{name, count}/BytesGauge{name, gauge} explicitly, asserting both the name and the numeric value, in addition to (not instead of) the existing Display assertions. Verified: full datafusion-ffi --features integration-tests suite, the exact three commands the CI jobs run (cargo test (amd64), hash collisions, extended_tests), the exact CI clippy script, and cargo fmt --check - all clean. Confirmed via git diff against the pre-fix commit that ForeignLibraryModule has zero field additions/removals. --- datafusion/ffi/src/tests/mod.rs | 16 ++++++--- datafusion/ffi/src/tests/utils.rs | 29 ++++++++++++++++ datafusion/ffi/tests/ffi_execution_plan.rs | 40 +++++++++++++++++++--- 3 files changed, 77 insertions(+), 8 deletions(-) diff --git a/datafusion/ffi/src/tests/mod.rs b/datafusion/ffi/src/tests/mod.rs index 110d14b000a4c..3c86f9aefeb7a 100644 --- a/datafusion/ffi/src/tests/mod.rs +++ b/datafusion/ffi/src/tests/mod.rs @@ -120,8 +120,6 @@ pub struct ForeignLibraryModule { pub create_exec_with_statistics: extern "C" fn() -> FFI_ExecutionPlan, - pub create_exec_with_byte_metrics: extern "C" fn() -> FFI_ExecutionPlan, - pub create_table_with_statistics: extern "C" fn(codec: FFI_LogicalExtensionCodec) -> FFI_TableProvider, @@ -242,9 +240,20 @@ pub(crate) extern "C" fn create_exec_with_statistics() -> FFI_ExecutionPlan { /// 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::BytesCount`]: datafusion_physical_expr_common::metrics::MetricValue::BytesCount /// [`MetricValue::BytesGauge`]: datafusion_physical_expr_common::metrics::MetricValue::BytesGauge -pub(crate) extern "C" fn create_exec_with_byte_metrics() -> FFI_ExecutionPlan { +#[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, }; @@ -393,7 +402,6 @@ pub extern "C" fn datafusion_ffi_get_module() -> ForeignLibraryModule { create_exec_with_expressions, create_exec_with_dynamic_expressions, create_exec_with_statistics, - create_exec_with_byte_metrics, create_table_with_statistics, create_physical_optimizer_rule: physical_optimizer::create_physical_optimizer_rule, diff --git a/datafusion/ffi/src/tests/utils.rs b/datafusion/ffi/src/tests/utils.rs index 3834ed82fe83e..2169cf04186d9 100644 --- a/datafusion/ffi/src/tests/utils.rs +++ b/datafusion/ffi/src/tests/utils.rs @@ -91,6 +91,35 @@ pub fn get_module() -> Result { 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 { + 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 diff --git a/datafusion/ffi/tests/ffi_execution_plan.rs b/datafusion/ffi/tests/ffi_execution_plan.rs index c087854df3253..b0318deb72a3c 100644 --- a/datafusion/ffi/tests/ffi_execution_plan.rs +++ b/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::MetricValue; use datafusion_physical_plan::execution_plan::InvariantLevel; use datafusion_physical_plan::{ ChildrenPropertiesMode, ExecutionPlan, ReplaceChildrenOptions, @@ -71,8 +72,7 @@ mod tests { #[test] fn test_ffi_execution_plan_byte_metrics_cross_library() -> Result<(), DataFusionError> { - let module = get_module()?; - let plan = (module.create_exec_with_byte_metrics)(); + let plan = get_byte_metrics_exec()?; let plan: Arc = (&plan).try_into()?; assert!(plan.is::()); @@ -83,8 +83,40 @@ mod tests { // 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"); - let rendered: Vec = metrics.iter().map(|m| m.to_string()).collect(); + // Match the discriminant itself, not just its Display output - this + // proves the BytesCount/BytesGauge variants (not e.g. a generic + // Count/Gauge that happens to render the same string) survived the + // ABI round trip. + let mut found_bytes_count = false; + let mut found_bytes_gauge = false; + for metric in metrics.iter() { + match metric.value() { + MetricValue::BytesCount { name, count } => { + assert_eq!(name.as_ref(), "bytes_scanned"); + assert_eq!(count.value(), 1536); + found_bytes_count = true; + } + MetricValue::BytesGauge { name, gauge } => { + assert_eq!(name.as_ref(), "stream_memory_usage"); + assert_eq!(gauge.value(), 2048); + found_bytes_gauge = true; + } + _ => {} + } + } + assert!( + found_bytes_count, + "expected a BytesCount metric, got: {metrics:?}" + ); + assert!( + found_bytes_gauge, + "expected a BytesGauge metric, got: {metrics:?}" + ); + + // Also confirm the byte-unit Display formatting these variants exist + // for in the first place still applies post-round-trip. + let rendered: Vec = metrics.iter().map(|m| m.to_string()).collect(); assert!( rendered .iter() From ad7201173f7acd064d60317d6c3f6292e47ac13b Mon Sep 17 00:00:00 2001 From: DevShiba Date: Sat, 22 Aug 2026 08:29:26 -0300 Subject: [PATCH 7/8] fix: apply the resolved conflict values that got missed before the merge commit The previous merge commit still had the literal PLACEHOLDER text in place of the two empirically-confirmed values (bytes_processed=1147.0 B and bytes_scanned=75.0 B) - staged before running sed to fill them in, so the merge commit captured the stale index instead of the corrected working tree. Caught by re-running git status/diff against HEAD before pushing, not by CI. --- .../sqllogictest/test_files/dynamic_filter_pushdown_config.slt | 2 +- .../sqllogictest/test_files/parquet_nested_schema_pruning.slt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index 3f4d9968fb3c2..be7cf71768ce8 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -104,7 +104,7 @@ Plan with Metrics 03)----ProjectionExec: expr=[id@0 as id, value@1 as v, value@1 + id@0 as name], metrics=[output_rows=10, ] 04)------FilterExec: value@1 > 3, metrics=[output_rows=10, , selectivity=100% (10/10)] 05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, metrics=[output_rows=10, ] -06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], dynamic_rg_pruning=eligible, pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=PLACEHOLDER, bytes_scanned=210.0 B, page_index_load_skipped=1, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio=18.31% (210/1.15 K)] +06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], dynamic_rg_pruning=eligible, pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=1147.0 B, bytes_scanned=210.0 B, page_index_load_skipped=1, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio=18.31% (210/1.15 K)] statement ok set datafusion.explain.analyze_level = dev; diff --git a/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt b/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt index db6d845ccba96..2f65afa3eab40 100644 --- a/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt +++ b/datafusion/sqllogictest/test_files/parquet_nested_schema_pruning.slt @@ -131,7 +131,7 @@ Plan with Metrics DataSourceExec: metrics=[output_rows=3, metrics=[output_rows=3, bytes_scanned=PLACEHOLDER] +Plan with Metrics DataSourceExec: metrics=[output_rows=3, bytes_scanned=75.0 B] # Mixed access -- the whole (narrowed) column and a subfield of it -- still # reads only the narrow schema's leaves. From 066d7130c09f345172e7f0b9b0ca1ff9efc79e3c Mon Sep 17 00:00:00 2001 From: DevShiba Date: Mon, 24 Aug 2026 08:29:11 -0300 Subject: [PATCH 8/8] fix: revert BytesCount/BytesGauge variants, restore category-aware Display MetricValue is a public, already-released exhaustive enum (55.0.0), so adding BytesCount/BytesGauge as new variants is a SemVer-breaking change (cargo-semver-checks: enum_variant_added) for any downstream crate that exhaustively matches on it. Reverts to the original design: bytes_counter/bytes_gauge/ global_bytes_counter still tag their metric with MetricCategory::Bytes, but construct generic Count/Gauge variants instead of dedicated ones. Display for Metric is restored to check the category and render through human_readable_size when it's Bytes, falling back to the plain Display for MetricValue otherwise. FFI_MetricValue is reverted to its exact pre-existing shape, and the cross-library FFI test now asserts the transported Bytes category, the generic variant/name/value, and the byte-formatted display output separately instead of matching on a dedicated variant discriminant. --- .../src/datasource/file_format/parquet.rs | 2 +- datafusion/core/tests/parquet/page_pruning.rs | 4 +- .../datasource-parquet/src/opener/mod.rs | 3 +- datafusion/ffi/src/physical_expr/metrics.rs | 38 --------- datafusion/ffi/src/tests/mod.rs | 17 ++-- datafusion/ffi/tests/ffi_execution_plan.rs | 47 ++++++----- .../src/metrics/builder.rs | 4 +- .../physical-expr-common/src/metrics/mod.rs | 41 +++++++--- .../physical-expr-common/src/metrics/value.rs | 80 +++---------------- datafusion/physical-plan/src/display.rs | 5 +- 10 files changed, 85 insertions(+), 156 deletions(-) diff --git a/datafusion/core/src/datasource/file_format/parquet.rs b/datafusion/core/src/datasource/file_format/parquet.rs index be5def329d2d6..bbd9d0937ad83 100644 --- a/datafusion/core/src/datasource/file_format/parquet.rs +++ b/datafusion/core/src/datasource/file_format/parquet.rs @@ -1171,7 +1171,7 @@ mod tests { let actual = exec .metrics() .expect("Metrics not recorded") - .sum(|metric| matches!(metric.value(), MetricValue::Count { name, .. } | MetricValue::BytesCount { name, .. } if name == "bytes_scanned")) + .sum(|metric| matches!(metric.value(), MetricValue::Count { name, .. } if name == "bytes_scanned")) .map(|t| t.as_usize()) .expect("bytes_scanned metric not recorded"); diff --git a/datafusion/core/tests/parquet/page_pruning.rs b/datafusion/core/tests/parquet/page_pruning.rs index 4966355e906a4..372a7a601d492 100644 --- a/datafusion/core/tests/parquet/page_pruning.rs +++ b/datafusion/core/tests/parquet/page_pruning.rs @@ -1053,9 +1053,7 @@ async fn test_pages_with_null_values() { fn cast_count_metric(metric: MetricValue) -> Option { match metric { - MetricValue::Count { count, .. } | MetricValue::BytesCount { count, .. } => { - Some(count.value()) - } + MetricValue::Count { count, .. } => Some(count.value()), _ => None, } } diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index a33e2bf2914e1..2e30fcc43038e 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -2450,8 +2450,7 @@ mod test { .clone_inner() .sum_by_name(name) .map(|metric| match metric { - MetricValue::Count { count, .. } - | MetricValue::BytesCount { count, .. } => count.value(), + MetricValue::Count { count, .. } => count.value(), _ => 0, }) .unwrap_or(0) diff --git a/datafusion/ffi/src/physical_expr/metrics.rs b/datafusion/ffi/src/physical_expr/metrics.rs index 0d62813a31774..763cc3f079a01 100644 --- a/datafusion/ffi/src/physical_expr/metrics.rs +++ b/datafusion/ffi/src/physical_expr/metrics.rs @@ -177,14 +177,6 @@ pub enum FFI_MetricValue { name: SString, gauge: u64, }, - BytesCount { - name: SString, - count: u64, - }, - BytesGauge { - name: SString, - gauge: u64, - }, } // ----------------------------------------------------------------------------- @@ -444,14 +436,6 @@ impl From<&MetricValue> for FFI_MetricValue { name: SString::from(name.as_ref()), gauge: gauge.value() as u64, }, - MetricValue::BytesCount { name, count } => Self::BytesCount { - name: SString::from(name.as_ref()), - count: count.value() as u64, - }, - MetricValue::BytesGauge { name, gauge } => Self::BytesGauge { - name: SString::from(name.as_ref()), - gauge: gauge.value() as u64, - }, MetricValue::Time { name, time } => Self::Time { name: SString::from(name.as_ref()), time_ns: time.value() as u64, @@ -512,14 +496,6 @@ impl From for MetricValue { name: Cow::Owned(name.into()), gauge: gauge_from_value(gauge), }, - FFI_MetricValue::BytesCount { name, count } => Self::BytesCount { - name: Cow::Owned(name.into()), - count: count_from_value(count), - }, - FFI_MetricValue::BytesGauge { name, gauge } => Self::BytesGauge { - name: Cow::Owned(name.into()), - gauge: gauge_from_value(gauge), - }, FFI_MetricValue::Time { name, time_ns } => Self::Time { name: Cow::Owned(name.into()), time: time_from_nanos(time_ns), @@ -670,20 +646,6 @@ mod tests { gauge: peak_memory, }); - let bytes_count = Count::new(); - bytes_count.add(55); - assert_value_roundtrip(MetricValue::BytesCount { - name: Cow::Borrowed("bytes_scanned"), - count: bytes_count, - }); - - let bytes_gauge = Gauge::new(); - bytes_gauge.add(66); - assert_value_roundtrip(MetricValue::BytesGauge { - name: Cow::Borrowed("stream_memory_usage"), - gauge: bytes_gauge, - }); - let time = Time::new(); time.add_duration(std::time::Duration::from_nanos(33)); assert_value_roundtrip(MetricValue::Time { diff --git a/datafusion/ffi/src/tests/mod.rs b/datafusion/ffi/src/tests/mod.rs index 3c86f9aefeb7a..7e583b6c5d5bf 100644 --- a/datafusion/ffi/src/tests/mod.rs +++ b/datafusion/ffi/src/tests/mod.rs @@ -232,13 +232,14 @@ pub(crate) extern "C" fn create_exec_with_statistics() -> FFI_ExecutionPlan { FFI_ExecutionPlan::new(plan, None) } -/// Registers real [`MetricValue::BytesCount`] and [`MetricValue::BytesGauge`] -/// metrics (via the same [`MetricBuilder::bytes_counter`] and -/// [`MetricBuilder::bytes_gauge`] constructors production code uses for +/// 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 these variants through a real -/// cross-library `metrics()` FFI call rather than only the in-process -/// `FFI_MetricValue` conversion tests in `physical_expr::metrics`. +/// 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 @@ -249,8 +250,8 @@ pub(crate) extern "C" fn create_exec_with_statistics() -> FFI_ExecutionPlan { /// separate exported symbol, loaded the same way [`load_module`] loads /// `datafusion_ffi_get_module`, avoids touching that struct's layout at all. /// -/// [`MetricValue::BytesCount`]: datafusion_physical_expr_common::metrics::MetricValue::BytesCount -/// [`MetricValue::BytesGauge`]: datafusion_physical_expr_common::metrics::MetricValue::BytesGauge +/// [`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 { diff --git a/datafusion/ffi/tests/ffi_execution_plan.rs b/datafusion/ffi/tests/ffi_execution_plan.rs index b0318deb72a3c..28828695d01af 100644 --- a/datafusion/ffi/tests/ffi_execution_plan.rs +++ b/datafusion/ffi/tests/ffi_execution_plan.rs @@ -27,7 +27,7 @@ mod tests { tests::EmptyExec, }; use datafusion_ffi::tests::utils::{get_byte_metrics_exec, get_module}; - use datafusion_physical_expr_common::metrics::MetricValue; + use datafusion_physical_expr_common::metrics::{MetricCategory, MetricValue}; use datafusion_physical_plan::execution_plan::InvariantLevel; use datafusion_physical_plan::{ ChildrenPropertiesMode, ExecutionPlan, ReplaceChildrenOptions, @@ -78,27 +78,36 @@ mod tests { // 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 - // MetricValue::BytesCount/BytesGauge through FFI_MetricsSet across - // that boundary - not just the in-process From conversions covered - // by physical_expr::metrics's roundtrip tests. + // 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"); - // Match the discriminant itself, not just its Display output - this - // proves the BytesCount/BytesGauge variants (not e.g. a generic - // Count/Gauge that happens to render the same string) survived the - // ABI round trip. + // 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::BytesCount { name, count } => { - assert_eq!(name.as_ref(), "bytes_scanned"); + 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::BytesGauge { name, gauge } => { - assert_eq!(name.as_ref(), "stream_memory_usage"); + 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; } @@ -107,27 +116,27 @@ mod tests { } assert!( found_bytes_count, - "expected a BytesCount metric, got: {metrics:?}" + "expected a Bytes-category bytes_scanned Count metric, got: {metrics:?}" ); assert!( found_bytes_gauge, - "expected a BytesGauge metric, got: {metrics:?}" + "expected a Bytes-category stream_memory_usage Gauge metric, got: {metrics:?}" ); - // Also confirm the byte-unit Display formatting these variants exist - // for in the first place still applies post-round-trip. + // Also confirm the byte-unit Display formatting still applies + // post-round-trip. let rendered: Vec = metrics.iter().map(|m| m.to_string()).collect(); assert!( rendered .iter() .any(|s| s == "bytes_scanned{partition=0}=1536.0 B"), - "BytesCount should survive the FFI round trip byte-formatted, got: {rendered:?}" + "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"), - "BytesGauge should survive the FFI round trip byte-formatted, got: {rendered:?}" + "stream_memory_usage should survive the FFI round trip byte-formatted, got: {rendered:?}" ); Ok(()) diff --git a/datafusion/physical-expr-common/src/metrics/builder.rs b/datafusion/physical-expr-common/src/metrics/builder.rs index 189cb6750bf83..17a9120a08b3e 100644 --- a/datafusion/physical-expr-common/src/metrics/builder.rs +++ b/datafusion/physical-expr-common/src/metrics/builder.rs @@ -273,7 +273,7 @@ impl<'a> MetricBuilder<'a> { let gauge = Gauge::new(); self.with_category(MetricCategory::Bytes) .with_partition(partition) - .build(MetricValue::BytesGauge { + .build(MetricValue::Gauge { name: gauge_name.into(), gauge: gauge.clone(), }); @@ -289,7 +289,7 @@ impl<'a> MetricBuilder<'a> { ) -> Count { let count = Count::new(); self.with_category(MetricCategory::Bytes) - .build(MetricValue::BytesCount { + .build(MetricValue::Count { name: counter_name.into(), count: count.clone(), }); diff --git a/datafusion/physical-expr-common/src/metrics/mod.rs b/datafusion/physical-expr-common/src/metrics/mod.rs index f3be2eb4c7ad7..cf9c73be01395 100644 --- a/datafusion/physical-expr-common/src/metrics/mod.rs +++ b/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, @@ -127,6 +128,18 @@ impl Display for Metric { // and now the 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) } } @@ -301,7 +314,6 @@ impl MetricsSet { pub fn sum_by_name(&self, metric_name: &str) -> Option { self.sum(|m| match m.value() { MetricValue::Count { name, .. } => name == metric_name, - MetricValue::BytesCount { name, .. } => name == metric_name, MetricValue::Time { name, .. } => name == metric_name, MetricValue::OutputRows(_) => false, MetricValue::ElapsedCompute(_) => false, @@ -312,7 +324,6 @@ impl MetricsSet { MetricValue::SpilledRows(_) => false, MetricValue::CurrentMemoryUsage(_) => false, MetricValue::Gauge { name, .. } => name == metric_name, - MetricValue::BytesGauge { name, .. } => name == metric_name, MetricValue::PeakMemoryUsage { name, .. } => name == metric_name, MetricValue::StartTimestamp(_) => false, MetricValue::EndTimestamp(_) => false, @@ -803,16 +814,22 @@ mod tests { MetricBuilder::new(&metrics).global_bytes_counter("bytes_written"); bytes_written.add(three_gib); - // A generic (non-byte) Count/Gauge tagged Bytes must NOT be - // reinterpreted based on category - only the dedicated - // BytesCount/BytesGauge variants get byte formatting. This is the - // explicit-typing fix for #24203, replacing an earlier approach that - // reinterpreted any Bytes-category Count/Gauge at Display time. + // 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_rows", 0); + .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 = metrics .clone_inner() .iter() @@ -835,11 +852,17 @@ mod tests { 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 must keep count-formatting even when tagged Bytes, got: {rendered:?}" + "a generic Gauge with no Bytes category must keep count-formatting, got: {rendered:?}" ); } diff --git a/datafusion/physical-expr-common/src/metrics/value.rs b/datafusion/physical-expr-common/src/metrics/value.rs index edfc4716aba13..37ab5194b2cc4 100644 --- a/datafusion/physical-expr-common/src/metrics/value.rs +++ b/datafusion/physical-expr-common/src/metrics/value.rs @@ -665,16 +665,6 @@ pub enum MetricValue { /// The value of the metric count: Count, }, - /// Operator defined count representing a size in bytes (e.g. - /// `bytes_scanned`, `bytes_written`). Like [`Self::Count`], but always - /// displayed with [`human_readable_size`]'s 1024-based byte units - /// (KB/MB/GB/TB) instead of [`human_readable_count`]'s 1000-based units. - BytesCount { - /// The provided name of this metric - name: Cow<'static, str>, - /// The value of the metric, in bytes - count: Count, - }, /// Operator defined gauge. Gauge { /// The provided name of this metric @@ -682,18 +672,6 @@ pub enum MetricValue { /// The value of the metric gauge: Gauge, }, - /// Operator defined gauge representing a size in bytes (e.g. - /// `stream_memory_usage`) that is not necessarily monotonically - /// increasing. Like [`Self::Gauge`], but always displayed with - /// [`human_readable_size`]'s byte units. Unlike [`Self::PeakMemoryUsage`], - /// this does not imply the value only ever grows - use - /// [`Self::PeakMemoryUsage`] for that. - BytesGauge { - /// The provided name of this metric - name: Cow<'static, str>, - /// The value of the metric, in bytes - gauge: Gauge, - }, /// Operator defined peak memory usage in bytes. PeakMemoryUsage { /// The provided name of this metric @@ -766,13 +744,6 @@ impl PartialEq for MetricValue { name: other_name, count: other_count, }, - ) - | ( - MetricValue::BytesCount { name, count }, - MetricValue::BytesCount { - name: other_name, - count: other_count, - }, ) => name == other_name && count == other_count, ( MetricValue::Gauge { name, gauge }, @@ -781,13 +752,6 @@ impl PartialEq for MetricValue { gauge: other_gauge, }, ) - | ( - MetricValue::BytesGauge { name, gauge }, - MetricValue::BytesGauge { - name: other_name, - gauge: other_gauge, - }, - ) | ( MetricValue::PeakMemoryUsage { name, gauge }, MetricValue::PeakMemoryUsage { @@ -859,10 +823,10 @@ impl MetricValue { Self::SpilledRows(_) => "spilled_rows", Self::CurrentMemoryUsage(_) => "mem_used", Self::ElapsedCompute(_) => "elapsed_compute", - Self::Count { name, .. } | Self::BytesCount { name, .. } => name.borrow(), - Self::Gauge { name, .. } - | Self::BytesGauge { name, .. } - | Self::PeakMemoryUsage { name, .. } => name.borrow(), + Self::Count { name, .. } => name.borrow(), + Self::Gauge { name, .. } | Self::PeakMemoryUsage { name, .. } => { + name.borrow() + } Self::Time { name, .. } => name.borrow(), Self::StartTimestamp(_) => "start_timestamp", Self::EndTimestamp(_) => "end_timestamp", @@ -884,10 +848,10 @@ impl MetricValue { Self::SpilledRows(count) => count.value(), Self::CurrentMemoryUsage(used) => used.value(), Self::ElapsedCompute(time) => time.value(), - Self::Count { count, .. } | Self::BytesCount { count, .. } => count.value(), - Self::Gauge { gauge, .. } - | Self::BytesGauge { gauge, .. } - | Self::PeakMemoryUsage { gauge, .. } => gauge.value(), + Self::Count { count, .. } => count.value(), + Self::Gauge { gauge, .. } | Self::PeakMemoryUsage { gauge, .. } => { + gauge.value() + } Self::Time { time, .. } => time.value(), Self::StartTimestamp(timestamp) => timestamp .value() @@ -925,18 +889,10 @@ impl MetricValue { name: name.clone(), count: Count::new(), }, - Self::BytesCount { name, .. } => Self::BytesCount { - name: name.clone(), - count: Count::new(), - }, Self::Gauge { name, .. } => Self::Gauge { name: name.clone(), gauge: Gauge::new(), }, - Self::BytesGauge { name, .. } => Self::BytesGauge { - name: name.clone(), - gauge: Gauge::new(), - }, Self::PeakMemoryUsage { name, .. } => Self::PeakMemoryUsage { name: name.clone(), gauge: Gauge::new(), @@ -992,12 +948,6 @@ impl MetricValue { Self::Count { count: other_count, .. }, - ) - | ( - Self::BytesCount { count, .. }, - Self::BytesCount { - count: other_count, .. - }, ) => count.add(other_count.value()), (Self::CurrentMemoryUsage(gauge), Self::CurrentMemoryUsage(other_gauge)) | ( @@ -1006,12 +956,6 @@ impl MetricValue { gauge: other_gauge, .. }, ) - | ( - Self::BytesGauge { gauge, .. }, - Self::BytesGauge { - gauge: other_gauge, .. - }, - ) | ( Self::PeakMemoryUsage { gauge, .. }, Self::PeakMemoryUsage { @@ -1114,9 +1058,7 @@ impl MetricValue { _ => 14, }, Self::PeakMemoryUsage { .. } => 13, - Self::BytesCount { .. } => 14, Self::Gauge { .. } => 15, - Self::BytesGauge { .. } => 15, Self::Time { .. } => 16, Self::Ratio { .. } => 17, Self::StartTimestamp(_) => 18, // show timestamps last @@ -1142,9 +1084,7 @@ impl Display for MetricValue { | Self::Count { count, .. } => { write!(f, "{count}") } - Self::SpilledBytes(count) - | Self::OutputBytes(count) - | Self::BytesCount { count, .. } => { + Self::SpilledBytes(count) | Self::OutputBytes(count) => { let readable_count = human_readable_size(count.value()); write!(f, "{readable_count}") } @@ -1153,7 +1093,7 @@ impl Display for MetricValue { let readable_size = human_readable_size(gauge.value()); write!(f, "{readable_size}") } - Self::PeakMemoryUsage { gauge, .. } | Self::BytesGauge { gauge, .. } => { + Self::PeakMemoryUsage { gauge, .. } => { let readable_size = human_readable_size(gauge.value()); write!(f, "{readable_size}") } diff --git a/datafusion/physical-plan/src/display.rs b/datafusion/physical-plan/src/display.rs index 2e1843011acbe..05e731abb389c 100644 --- a/datafusion/physical-plan/src/display.rs +++ b/datafusion/physical-plan/src/display.rs @@ -805,11 +805,8 @@ impl PgJsonExecutionPlanVisitor<'_> { let ms = (t.value() as f64) / 1_000_000.0; serde_json::Value::from(ms) } - MetricValue::Count { count, .. } | MetricValue::BytesCount { count, .. } => { - serde_json::Value::from(count.value()) - } + MetricValue::Count { count, .. } => serde_json::Value::from(count.value()), MetricValue::Gauge { gauge, .. } - | MetricValue::BytesGauge { gauge, .. } | MetricValue::PeakMemoryUsage { gauge, .. } => { serde_json::Value::from(gauge.value()) }