From 8b9844223956a7f6a678d4976963682299be5d4a Mon Sep 17 00:00:00 2001 From: Radu Stoenescu Date: Tue, 29 Sep 2026 16:45:52 +0300 Subject: [PATCH 1/2] fix(datafusion): preserve Delta planning count metrics --- crates/core/src/delta_datafusion/mod.rs | 1 + .../planning_count_metrics.rs | 126 ++++++++++++++++++ .../src/delta_datafusion/table_provider.rs | 10 +- .../table_provider/next/scan/codec.rs | 11 +- 4 files changed, 145 insertions(+), 3 deletions(-) create mode 100644 crates/core/src/delta_datafusion/planning_count_metrics.rs diff --git a/crates/core/src/delta_datafusion/mod.rs b/crates/core/src/delta_datafusion/mod.rs index bf409e0d45..e28e0e88b3 100644 --- a/crates/core/src/delta_datafusion/mod.rs +++ b/crates/core/src/delta_datafusion/mod.rs @@ -105,6 +105,7 @@ pub mod expr; mod file_id; mod find_files; pub mod logical; +mod planning_count_metrics; pub mod physical; pub mod planner; mod session; diff --git a/crates/core/src/delta_datafusion/planning_count_metrics.rs b/crates/core/src/delta_datafusion/planning_count_metrics.rs new file mode 100644 index 0000000000..22b9173b31 --- /dev/null +++ b/crates/core/src/delta_datafusion/planning_count_metrics.rs @@ -0,0 +1,126 @@ +//! Serializable planning count metrics attached to Delta scan plans. + +use std::borrow::Cow; +use std::sync::Arc; + +use datafusion::physical_plan::metrics::{ + Count, ExecutionPlanMetricsSet, Label, Metric, MetricValue, MetricsSet, +}; +use serde::{Deserialize, Serialize}; + +// Consumed by datafusion-distributed's MetricsWrapperExec before task aggregation. +const DFD_AGGREGATION_SCOPE_LABEL: &str = "dfd_aggregation_scope"; +const COORDINATOR_AGGREGATION_SCOPE: &str = "coordinator"; + +#[derive(Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(transparent)] +pub(super) struct PlanningCountMetricsWire(Vec); + +impl PlanningCountMetricsWire { + pub(super) fn is_empty(&self) -> bool { + self.0.is_empty() + } +} + +impl From<&MetricsSet> for PlanningCountMetricsWire { + fn from(metrics: &MetricsSet) -> Self { + Self( + metrics + .iter() + .filter_map(|metric| match metric.value() { + MetricValue::Count { name, count } => Some(PlanningCountMetricWire { + name: name.to_string(), + value: count.value(), + }), + _ => None, + }) + .collect(), + ) + } +} + +impl From for ExecutionPlanMetricsSet { + fn from(wire: PlanningCountMetricsWire) -> Self { + wire.0 + .into_iter() + .map(|metric| { + let count = Count::new(); + count.add(metric.value); + Arc::new(Metric::new_with_labels( + MetricValue::Count { + name: Cow::Owned(metric.name), + count, + }, + None, + vec![Label::new( + DFD_AGGREGATION_SCOPE_LABEL, + COORDINATOR_AGGREGATION_SCOPE, + )], + )) + }) + .collect::() + .into() + } +} + +#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] +struct PlanningCountMetricWire { + name: String, + value: usize, +} + +#[cfg(test)] +mod tests { + use datafusion::physical_plan::metrics::{Label, MetricType}; + + use super::*; + + fn count(value: usize) -> Count { + let count = Count::new(); + count.add(value); + count + } + + #[test] + fn planning_count_metrics_wire_roundtrips_named_counts_only() { + let metrics = vec![ + Arc::new(Metric::new_with_labels( + MetricValue::Count { + name: Cow::Borrowed("files_scanned"), + count: count(4), + }, + Some(7), + vec![Label::new("source", "planning")], + )), + Arc::new(Metric::new( + MetricValue::Count { + name: Cow::Borrowed("files_pruned"), + count: count(2), + }, + None, + )), + Arc::new(Metric::new(MetricValue::OutputRows(count(80)), None)), + ] + .into_iter() + .collect::(); + + let wire = PlanningCountMetricsWire::from(&metrics); + let encoded = serde_json::to_vec(&wire).unwrap(); + let decoded: PlanningCountMetricsWire = serde_json::from_slice(&encoded).unwrap(); + assert_eq!(decoded, wire); + + let decoded = ExecutionPlanMetricsSet::from(decoded).clone_inner(); + assert_eq!(decoded.iter().count(), 2); + assert_eq!(decoded.sum_by_name("files_scanned").unwrap().as_usize(), 4); + assert_eq!(decoded.sum_by_name("files_pruned").unwrap().as_usize(), 2); + assert!(decoded.iter().all(|metric| { + let labels = metric.labels(); + metric.partition().is_none() + && labels.len() == 1 + && labels[0].name() == DFD_AGGREGATION_SCOPE_LABEL + && labels[0].value() == COORDINATOR_AGGREGATION_SCOPE + && metric.metric_type() == MetricType::Dev + && metric.metric_category().is_none() + })); + } +} diff --git a/crates/core/src/delta_datafusion/table_provider.rs b/crates/core/src/delta_datafusion/table_provider.rs index a4972e7b44..01a9080a7f 100644 --- a/crates/core/src/delta_datafusion/table_provider.rs +++ b/crates/core/src/delta_datafusion/table_provider.rs @@ -21,6 +21,7 @@ use serde::{Deserialize, Serialize}; use url::Url; use uuid::Uuid; +use crate::delta_datafusion::planning_count_metrics::PlanningCountMetricsWire; use crate::delta_datafusion::table_provider::next::SnapshotWrapper; use crate::delta_datafusion::{DataFusionMixins as _, FindFilesExprProperties}; use crate::kernel::{Add, EagerSnapshot, Snapshot, Version}; @@ -560,6 +561,8 @@ pub(super) struct DeltaScanWire { pub(crate) table_url: Url, pub(crate) config: DeltaScanConfig, pub(crate) logical_schema: Arc, + #[serde(default, skip_serializing_if = "PlanningCountMetricsWire::is_empty")] + pub(crate) planning_count_metrics: PlanningCountMetricsWire, } impl From<&DeltaScan> for DeltaScanWire { @@ -568,18 +571,21 @@ impl From<&DeltaScan> for DeltaScanWire { table_url: scan.table_url.clone(), config: scan.config.clone(), logical_schema: scan.logical_schema.clone(), + planning_count_metrics: PlanningCountMetricsWire::from(&scan.metrics.clone_inner()), } } } impl DeltaScanWire { pub(super) fn into_delta_scan(self, parquet_scan: Arc) -> DeltaScan { - DeltaScan::new( + let mut scan = DeltaScan::new( &self.table_url, self.config, parquet_scan, self.logical_schema, - ) + ); + scan.metrics = self.planning_count_metrics.into(); + scan } } diff --git a/crates/core/src/delta_datafusion/table_provider/next/scan/codec.rs b/crates/core/src/delta_datafusion/table_provider/next/scan/codec.rs index 45b0f347b5..11184560ee 100644 --- a/crates/core/src/delta_datafusion/table_provider/next/scan/codec.rs +++ b/crates/core/src/delta_datafusion/table_provider/next/scan/codec.rs @@ -12,6 +12,7 @@ use super::{DeltaScanExec, DeltaScanMetaExec, ProjectedScanContract, PublicFileI use crate::DeltaTableConfig; use crate::delta_datafusion::DeltaScanConfig; use crate::delta_datafusion::engine::{to_datafusion_expr, to_delta_expression}; +use crate::delta_datafusion::planning_count_metrics::PlanningCountMetricsWire; use crate::kernel::Snapshot; use crate::kernel::size_limits::SnapshotLoadMetrics; use arrow::datatypes::SchemaRef; @@ -123,6 +124,8 @@ pub(crate) struct DeltaScanExecWire { input_file_id_column: String, file_id_column: Option, public_file_ids: std::collections::HashMap, + #[serde(default, skip_serializing_if = "PlanningCountMetricsWire::is_empty")] + planning_count_metrics: PlanningCountMetricsWire, } #[derive(Debug, Serialize, Deserialize)] @@ -210,6 +213,9 @@ impl TryFrom<&DeltaScanExec> for DeltaScanExecWire { fn try_from(exec: &DeltaScanExec) -> Result { let scan_plan_wire = scan_plan_wire(&exec.scan_plan)?; + let metrics = exec.metrics().ok_or_else(|| { + DataFusionError::Internal("DeltaScanExec did not expose its metrics".to_string()) + })?; let transforms = serialize_transforms(&exec.transforms)?; @@ -233,6 +239,7 @@ impl TryFrom<&DeltaScanExec> for DeltaScanExecWire { input_file_id_column: exec.input_file_id_column.clone(), file_id_column: exec.file_id_column.clone(), public_file_ids, + planning_count_metrics: PlanningCountMetricsWire::from(&metrics), }) } } @@ -487,6 +494,8 @@ impl DeltaScanExecWire { } } + let metrics = self.planning_count_metrics.into(); + let exec = DeltaScanExec::new( Arc::new(scan_plan), execution_plan, @@ -494,7 +503,7 @@ impl DeltaScanExecWire { Arc::new(selection_vectors), Arc::new(public_file_ids), Default::default(), - Default::default(), + metrics, ); Ok(Arc::new(exec)) From 498734df6c3cb381f203715666b2ba39eb34192c Mon Sep 17 00:00:00 2001 From: Radu Stoenescu Date: Wed, 30 Sep 2026 14:38:55 +0300 Subject: [PATCH 2/2] fix(datafusion): harden planning metric serialization --- crates/core/src/delta_datafusion/mod.rs | 2 +- ...ics.rs => planning_count_metrics_codec.rs} | 20 ++++++++++--------- .../src/delta_datafusion/table_provider.rs | 2 +- .../table_provider/next/scan/codec.rs | 6 ++---- 4 files changed, 15 insertions(+), 15 deletions(-) rename crates/core/src/delta_datafusion/{planning_count_metrics.rs => planning_count_metrics_codec.rs} (84%) diff --git a/crates/core/src/delta_datafusion/mod.rs b/crates/core/src/delta_datafusion/mod.rs index e28e0e88b3..c7f864c0c6 100644 --- a/crates/core/src/delta_datafusion/mod.rs +++ b/crates/core/src/delta_datafusion/mod.rs @@ -105,7 +105,7 @@ pub mod expr; mod file_id; mod find_files; pub mod logical; -mod planning_count_metrics; +mod planning_count_metrics_codec; pub mod physical; pub mod planner; mod session; diff --git a/crates/core/src/delta_datafusion/planning_count_metrics.rs b/crates/core/src/delta_datafusion/planning_count_metrics_codec.rs similarity index 84% rename from crates/core/src/delta_datafusion/planning_count_metrics.rs rename to crates/core/src/delta_datafusion/planning_count_metrics_codec.rs index 22b9173b31..3626db738e 100644 --- a/crates/core/src/delta_datafusion/planning_count_metrics.rs +++ b/crates/core/src/delta_datafusion/planning_count_metrics_codec.rs @@ -1,4 +1,9 @@ -//! Serializable planning count metrics attached to Delta scan plans. +//! Codec for planning-time count metrics attached to Delta scan plans. +//! +//! Only [`MetricValue::Count`] metrics are serialized. Runtime metrics are collected by the +//! execution plan on the worker and are not copied from the coordinator. +// TODO: Move this codec into `table_provider::next::scan::codec` after removing the legacy table +// provider. use std::borrow::Cow; use std::sync::Arc; @@ -9,8 +14,8 @@ use datafusion::physical_plan::metrics::{ use serde::{Deserialize, Serialize}; // Consumed by datafusion-distributed's MetricsWrapperExec before task aggregation. -const DFD_AGGREGATION_SCOPE_LABEL: &str = "dfd_aggregation_scope"; -const COORDINATOR_AGGREGATION_SCOPE: &str = "coordinator"; +const DFD_CREATED_AT_LABEL: &str = "created_at"; +const CREATED_AT_PLANNING: &str = "planning"; #[derive(Debug, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(transparent)] @@ -52,10 +57,7 @@ impl From for ExecutionPlanMetricsSet { count, }, None, - vec![Label::new( - DFD_AGGREGATION_SCOPE_LABEL, - COORDINATOR_AGGREGATION_SCOPE, - )], + vec![Label::new(DFD_CREATED_AT_LABEL, CREATED_AT_PLANNING)], )) }) .collect::() @@ -117,8 +119,8 @@ mod tests { let labels = metric.labels(); metric.partition().is_none() && labels.len() == 1 - && labels[0].name() == DFD_AGGREGATION_SCOPE_LABEL - && labels[0].value() == COORDINATOR_AGGREGATION_SCOPE + && labels[0].name() == DFD_CREATED_AT_LABEL + && labels[0].value() == CREATED_AT_PLANNING && metric.metric_type() == MetricType::Dev && metric.metric_category().is_none() })); diff --git a/crates/core/src/delta_datafusion/table_provider.rs b/crates/core/src/delta_datafusion/table_provider.rs index 01a9080a7f..2dddfe8a01 100644 --- a/crates/core/src/delta_datafusion/table_provider.rs +++ b/crates/core/src/delta_datafusion/table_provider.rs @@ -21,7 +21,7 @@ use serde::{Deserialize, Serialize}; use url::Url; use uuid::Uuid; -use crate::delta_datafusion::planning_count_metrics::PlanningCountMetricsWire; +use crate::delta_datafusion::planning_count_metrics_codec::PlanningCountMetricsWire; use crate::delta_datafusion::table_provider::next::SnapshotWrapper; use crate::delta_datafusion::{DataFusionMixins as _, FindFilesExprProperties}; use crate::kernel::{Add, EagerSnapshot, Snapshot, Version}; diff --git a/crates/core/src/delta_datafusion/table_provider/next/scan/codec.rs b/crates/core/src/delta_datafusion/table_provider/next/scan/codec.rs index 11184560ee..556eeecaeb 100644 --- a/crates/core/src/delta_datafusion/table_provider/next/scan/codec.rs +++ b/crates/core/src/delta_datafusion/table_provider/next/scan/codec.rs @@ -12,7 +12,7 @@ use super::{DeltaScanExec, DeltaScanMetaExec, ProjectedScanContract, PublicFileI use crate::DeltaTableConfig; use crate::delta_datafusion::DeltaScanConfig; use crate::delta_datafusion::engine::{to_datafusion_expr, to_delta_expression}; -use crate::delta_datafusion::planning_count_metrics::PlanningCountMetricsWire; +use crate::delta_datafusion::planning_count_metrics_codec::PlanningCountMetricsWire; use crate::kernel::Snapshot; use crate::kernel::size_limits::SnapshotLoadMetrics; use arrow::datatypes::SchemaRef; @@ -213,9 +213,7 @@ impl TryFrom<&DeltaScanExec> for DeltaScanExecWire { fn try_from(exec: &DeltaScanExec) -> Result { let scan_plan_wire = scan_plan_wire(&exec.scan_plan)?; - let metrics = exec.metrics().ok_or_else(|| { - DataFusionError::Internal("DeltaScanExec did not expose its metrics".to_string()) - })?; + let metrics = exec.metrics().unwrap_or_default(); let transforms = serialize_transforms(&exec.transforms)?;