Skip to content

Commit d256caa

Browse files
2010YOUY012010YOUY01
authored
feat: Add selectivity metrics to FilterExec (#18406)
## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> Part of #18217 ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. --> In `FilterExec`, selectivity is calculated as `output_rows/input_rows`. This PR supports such metric. I think this metrics provides important application-level insights, and would be commonly used, so it is displayed in the `summary` verbose level. ### Demo in `datafusion-cli` ``` > set datafusion.explain.analyze_level = summary; 0 row(s) fetched. Elapsed 0.000 seconds. > explain analyze select * from generate_series(100) as t1(v1) where v1 <10; +-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | plan_type | plan | +-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | Plan with Metrics | ProjectionExec: expr=[value@0 as v1], metrics=[output_rows=10, elapsed_compute=1.763µs, output_bytes=64.0 KB] | | | CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=10, elapsed_compute=25.833µs, output_bytes=64.0 KB] | | | FilterExec: value@0 < 10, metrics=[output_rows=10, elapsed_compute=34.888µs, output_bytes=128.0 B, selectivity=9.9% (10/101)] | | | RepartitionExec: partitioning=RoundRobinBatch(14), input_partitions=1, metrics=[] | | | LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=0, end=100, batch_size=8192], metrics=[output_rows=101, elapsed_compute=33.167µs, output_bytes=64.0 KB] | | | | +-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ 1 row(s) fetched. Elapsed 0.004 seconds. ``` ## What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR. --> 1. Add a new `MetricValue` for ratio. 2. Tracking selectivity in `FilterExec` with `MetricValue::Ratio` ## Are these changes tested? UT <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 3. Serve as another way to document the expected behavior of the code If tests are not included in your PR, please explain why (for example, are they covered by existing tests)? --> ## Are there any user-facing changes? No <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. --> <!-- If there are any breaking changes to public APIs, please add the `api change` label. -->
1 parent 73038f5 commit d256caa

File tree

5 files changed

+218
-11
lines changed

5 files changed

+218
-11
lines changed

‎datafusion/core/tests/sql/explain_analyze.rs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,11 @@ async fn explain_analyze_baseline_metrics() {
8888
"FilterExec: c13@1 != C2GT5KVyOPZpgKVl110TyZO0NcJ434",
8989
"output_bytes="
9090
);
91+
assert_metrics!(
92+
&formatted,
93+
"FilterExec: c13@1 != C2GT5KVyOPZpgKVl110TyZO0NcJ434",
94+
"selectivity=99% (99/100)"
95+
);
9196
assert_metrics!(
9297
&formatted,
9398
"ProjectionExec: expr=[]",

‎datafusion/physical-plan/src/filter.rs‎

Lines changed: 30 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -32,12 +32,13 @@ use crate::filter_pushdown::{
3232
ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase,
3333
FilterPushdownPropagation, PushedDown, PushedDownPredicate,
3434
};
35+
use crate::metrics::{MetricBuilder, MetricType};
3536
use crate::projection::{
3637
make_with_child, try_embed_projection, update_expr, EmbeddedProjection,
3738
ProjectionExec, ProjectionExpr,
3839
};
3940
use crate::{
40-
metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet},
41+
metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet, RatioMetrics},
4142
DisplayFormatType, ExecutionPlan,
4243
};
4344

@@ -384,12 +385,12 @@ impl ExecutionPlan for FilterExec {
384385
context: Arc<TaskContext>,
385386
) -> Result<SendableRecordBatchStream> {
386387
trace!("Start FilterExec::execute for partition {} of context session_id {} and task_id {:?}", partition, context.session_id(), context.task_id());
387-
let baseline_metrics = BaselineMetrics::new(&self.metrics, partition);
388+
let metrics = FilterExecMetrics::new(&self.metrics, partition);
388389
Ok(Box::pin(FilterExecStream {
389390
schema: self.schema(),
390391
predicate: Arc::clone(&self.predicate),
391392
input: self.input.execute(partition, context)?,
392-
baseline_metrics,
393+
metrics,
393394
projection: self.projection.clone(),
394395
}))
395396
}
@@ -623,11 +624,30 @@ struct FilterExecStream {
623624
/// The input partition to filter.
624625
input: SendableRecordBatchStream,
625626
/// Runtime metrics recording
626-
baseline_metrics: BaselineMetrics,
627+
metrics: FilterExecMetrics,
627628
/// The projection indices of the columns in the input schema
628629
projection: Option<Vec<usize>>,
629630
}
630631

632+
/// The metrics for `FilterExec`
633+
struct FilterExecMetrics {
634+
// Common metrics for most operators
635+
baseline_metrics: BaselineMetrics,
636+
// Selectivity of the filter, calculated as output_rows / input_rows
637+
selectivity: RatioMetrics,
638+
}
639+
640+
impl FilterExecMetrics {
641+
pub fn new(metrics: &ExecutionPlanMetricsSet, partition: usize) -> Self {
642+
Self {
643+
baseline_metrics: BaselineMetrics::new(metrics, partition),
644+
selectivity: MetricBuilder::new(metrics)
645+
.with_type(MetricType::SUMMARY)
646+
.ratio_metrics("selectivity", partition),
647+
}
648+
}
649+
}
650+
631651
pub fn batch_filter(
632652
batch: &RecordBatch,
633653
predicate: &Arc<dyn PhysicalExpr>,
@@ -679,14 +699,18 @@ impl Stream for FilterExecStream {
679699
loop {
680700
match ready!(self.input.poll_next_unpin(cx)) {
681701
Some(Ok(batch)) => {
682-
let timer = self.baseline_metrics.elapsed_compute().timer();
702+
let timer = self.metrics.baseline_metrics.elapsed_compute().timer();
683703
let filtered_batch = filter_and_project(
684704
&batch,
685705
&self.predicate,
686706
self.projection.as_ref(),
687707
&self.schema,
688708
)?;
689709
timer.done();
710+
711+
self.metrics.selectivity.add_part(filtered_batch.num_rows());
712+
self.metrics.selectivity.add_total(batch.num_rows());
713+
690714
// Skip entirely filtered batches
691715
if filtered_batch.num_rows() == 0 {
692716
continue;
@@ -700,7 +724,7 @@ impl Stream for FilterExecStream {
700724
}
701725
}
702726
}
703-
self.baseline_metrics.record_poll(poll)
727+
self.metrics.baseline_metrics.record_poll(poll)
704728
}
705729

706730
fn size_hint(&self) -> (usize, Option<usize>) {

‎datafusion/physical-plan/src/metrics/builder.rs‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,10 @@
1919
2020
use std::{borrow::Cow, sync::Arc};
2121

22-
use crate::metrics::{value::PruningMetrics, MetricType};
22+
use crate::metrics::{
23+
value::{PruningMetrics, RatioMetrics},
24+
MetricType,
25+
};
2326

2427
use super::{
2528
Count, ExecutionPlanMetricsSet, Gauge, Label, Metric, MetricValue, Time, Timestamp,
@@ -265,4 +268,18 @@ impl<'a> MetricBuilder<'a> {
265268
});
266269
pruning_metrics
267270
}
271+
272+
/// Consumes self and creates a new [`RatioMetrics`]
273+
pub fn ratio_metrics(
274+
self,
275+
name: impl Into<Cow<'static, str>>,
276+
partition: usize,
277+
) -> RatioMetrics {
278+
let ratio_metrics = RatioMetrics::new();
279+
self.with_partition(partition).build(MetricValue::Ratio {
280+
name: name.into(),
281+
ratio_metrics: ratio_metrics.clone(),
282+
});
283+
ratio_metrics
284+
}
268285
}

‎datafusion/physical-plan/src/metrics/mod.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,8 @@ pub use baseline::{BaselineMetrics, RecordOutput, SpillMetrics, SplitMetrics};
3636
pub use builder::MetricBuilder;
3737
pub use custom::CustomMetricValue;
3838
pub use value::{
39-
Count, Gauge, MetricValue, PruningMetrics, ScopedTimerGuard, Time, Timestamp,
39+
Count, Gauge, MetricValue, PruningMetrics, RatioMetrics, ScopedTimerGuard, Time,
40+
Timestamp,
4041
};
4142

4243
/// Something that tracks a value of interest (metric) of a DataFusion
@@ -304,6 +305,7 @@ impl MetricsSet {
304305
MetricValue::StartTimestamp(_) => false,
305306
MetricValue::EndTimestamp(_) => false,
306307
MetricValue::PruningMetrics { name, .. } => name == metric_name,
308+
MetricValue::Ratio { name, .. } => name == metric_name,
307309
MetricValue::Custom { .. } => false,
308310
})
309311
}

‎datafusion/physical-plan/src/metrics/value.rs‎

Lines changed: 162 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -430,6 +430,92 @@ impl PruningMetrics {
430430
}
431431
}
432432

433+
/// Counters tracking ratio metrics (e.g. matched vs total)
434+
///
435+
/// The counters are thread-safe and shared across clones.
436+
#[derive(Debug, Clone, Default)]
437+
pub struct RatioMetrics {
438+
part: Arc<AtomicUsize>,
439+
total: Arc<AtomicUsize>,
440+
}
441+
442+
impl RatioMetrics {
443+
/// Create a new [`RatioMetrics`]
444+
pub fn new() -> Self {
445+
Self {
446+
part: Arc::new(AtomicUsize::new(0)),
447+
total: Arc::new(AtomicUsize::new(0)),
448+
}
449+
}
450+
451+
/// Add `n` to the numerator (`part`) value
452+
pub fn add_part(&self, n: usize) {
453+
self.part.fetch_add(n, Ordering::Relaxed);
454+
}
455+
456+
/// Add `n` to the denominator (`total`) value
457+
pub fn add_total(&self, n: usize) {
458+
self.total.fetch_add(n, Ordering::Relaxed);
459+
}
460+
461+
/// Merge the value from `other` into `self`
462+
pub fn merge(&self, other: &Self) {
463+
self.add_part(other.part());
464+
self.add_total(other.total());
465+
}
466+
467+
/// Return the numerator (`part`) value
468+
pub fn part(&self) -> usize {
469+
self.part.load(Ordering::Relaxed)
470+
}
471+
472+
/// Return the denominator (`total`) value
473+
pub fn total(&self) -> usize {
474+
self.total.load(Ordering::Relaxed)
475+
}
476+
}
477+
478+
impl PartialEq for RatioMetrics {
479+
fn eq(&self, other: &Self) -> bool {
480+
self.part() == other.part() && self.total() == other.total()
481+
}
482+
}
483+
484+
/// Format a float number with `digits` most significant numbers.
485+
///
486+
/// fmt_significant(12.5) -> "12"
487+
/// fmt_significant(0.0543) -> "0.054"
488+
/// fmt_significant(0.000123) -> "0.00012"
489+
fn fmt_significant(mut x: f64, digits: usize) -> String {
490+
if x == 0.0 {
491+
return "0".to_string();
492+
}
493+
494+
let exp = x.abs().log10().floor(); // exponent of first significant digit
495+
let scale = 10f64.powf(-(exp - (digits as f64 - 1.0)));
496+
x = (x * scale).round() / scale; // round to N significant digits
497+
format!("{x}")
498+
}
499+
500+
impl Display for RatioMetrics {
501+
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
502+
let part = self.part();
503+
let total = self.total();
504+
505+
if total == 0 {
506+
if part == 0 {
507+
write!(f, "N/A (0/0)")
508+
} else {
509+
write!(f, "N/A ({part}/0)")
510+
}
511+
} else {
512+
let percentage = (part as f64 / total as f64) * 100.0;
513+
514+
write!(f, "{}% ({part}/{total})", fmt_significant(percentage, 2))
515+
}
516+
}
517+
}
518+
433519
/// Possible values for a [super::Metric].
434520
///
435521
/// Among other differences, the metric types have different ways to
@@ -499,6 +585,11 @@ pub enum MetricValue {
499585
name: Cow<'static, str>,
500586
pruning_metrics: PruningMetrics,
501587
},
588+
/// Metrics that should be displayed as ratio like (42%)
589+
Ratio {
590+
name: Cow<'static, str>,
591+
ratio_metrics: RatioMetrics,
592+
},
502593
Custom {
503594
/// The provided name of this metric
504595
name: Cow<'static, str>,
@@ -563,6 +654,30 @@ impl PartialEq for MetricValue {
563654
(MetricValue::EndTimestamp(timestamp), MetricValue::EndTimestamp(other)) => {
564655
timestamp == other
565656
}
657+
(
658+
MetricValue::PruningMetrics {
659+
name,
660+
pruning_metrics,
661+
},
662+
MetricValue::PruningMetrics {
663+
name: other_name,
664+
pruning_metrics: other_pruning_metrics,
665+
},
666+
) => {
667+
name == other_name
668+
&& pruning_metrics.pruned() == other_pruning_metrics.pruned()
669+
&& pruning_metrics.matched() == other_pruning_metrics.matched()
670+
}
671+
(
672+
MetricValue::Ratio {
673+
name,
674+
ratio_metrics,
675+
},
676+
MetricValue::Ratio {
677+
name: other_name,
678+
ratio_metrics: other_ratio_metrics,
679+
},
680+
) => name == other_name && ratio_metrics == other_ratio_metrics,
566681
(
567682
MetricValue::Custom { name, value },
568683
MetricValue::Custom {
@@ -593,6 +708,7 @@ impl MetricValue {
593708
Self::StartTimestamp(_) => "start_timestamp",
594709
Self::EndTimestamp(_) => "end_timestamp",
595710
Self::PruningMetrics { name, .. } => name.borrow(),
711+
Self::Ratio { name, .. } => name.borrow(),
596712
Self::Custom { name, .. } => name.borrow(),
597713
}
598714
}
@@ -625,6 +741,8 @@ impl MetricValue {
625741
// like `PruningMetrics`, this function is not supposed to get called.
626742
// Metrics aggregation for them are implemented inside `MetricsSet` directly.
627743
Self::PruningMetrics { .. } => 0,
744+
// Should not be used. See comments in `PruningMetrics` for details.
745+
Self::Ratio { .. } => 0,
628746
Self::Custom { value, .. } => value.as_usize(),
629747
}
630748
}
@@ -658,6 +776,10 @@ impl MetricValue {
658776
name: name.clone(),
659777
pruning_metrics: PruningMetrics::new(),
660778
},
779+
Self::Ratio { name, .. } => Self::Ratio {
780+
name: name.clone(),
781+
ratio_metrics: RatioMetrics::new(),
782+
},
661783
Self::Custom { name, value } => Self::Custom {
662784
name: name.clone(),
663785
value: value.new_empty(),
@@ -723,6 +845,15 @@ impl MetricValue {
723845
pruning_metrics.add_pruned(pruned);
724846
pruning_metrics.add_matched(matched);
725847
}
848+
(
849+
Self::Ratio { ratio_metrics, .. },
850+
Self::Ratio {
851+
ratio_metrics: other_ratio_metrics,
852+
..
853+
},
854+
) => {
855+
ratio_metrics.merge(other_ratio_metrics);
856+
}
726857
(
727858
Self::Custom { value, .. },
728859
Self::Custom {
@@ -770,9 +901,10 @@ impl MetricValue {
770901
Self::Count { .. } => 12,
771902
Self::Gauge { .. } => 13,
772903
Self::Time { .. } => 14,
773-
Self::StartTimestamp(_) => 15, // show timestamps last
774-
Self::EndTimestamp(_) => 16,
775-
Self::Custom { .. } => 17,
904+
Self::Ratio { .. } => 15,
905+
Self::StartTimestamp(_) => 16, // show timestamps last
906+
Self::EndTimestamp(_) => 17,
907+
Self::Custom { .. } => 18,
776908
}
777909
}
778910

@@ -816,6 +948,7 @@ impl Display for MetricValue {
816948
} => {
817949
write!(f, "{pruning_metrics}")
818950
}
951+
Self::Ratio { ratio_metrics, .. } => write!(f, "{ratio_metrics}"),
819952
Self::Custom { name, value } => {
820953
write!(f, "name:{name} {value}")
821954
}
@@ -970,6 +1103,32 @@ mod tests {
9701103
}
9711104
}
9721105

1106+
#[test]
1107+
fn test_display_ratio() {
1108+
let ratio_metrics = RatioMetrics::new();
1109+
let ratio = MetricValue::Ratio {
1110+
name: Cow::Borrowed("ratio_metric"),
1111+
ratio_metrics: ratio_metrics.clone(),
1112+
};
1113+
1114+
assert_eq!("N/A (0/0)", ratio.to_string());
1115+
1116+
ratio_metrics.add_part(10);
1117+
assert_eq!("N/A (10/0)", ratio.to_string());
1118+
1119+
ratio_metrics.add_total(40);
1120+
assert_eq!("25% (10/40)", ratio.to_string());
1121+
1122+
let tiny_ratio_metrics = RatioMetrics::new();
1123+
let tiny_ratio = MetricValue::Ratio {
1124+
name: Cow::Borrowed("tiny_ratio_metric"),
1125+
ratio_metrics: tiny_ratio_metrics.clone(),
1126+
};
1127+
tiny_ratio_metrics.add_part(1);
1128+
tiny_ratio_metrics.add_total(3000);
1129+
assert_eq!("0.033% (1/3000)", tiny_ratio.to_string());
1130+
}
1131+
9731132
#[test]
9741133
fn test_display_timestamp() {
9751134
let timestamp = Timestamp::new();

0 commit comments

Comments
 (0)