query_runtime/metrics/
dynamic_filter.rs

1use query_planning::{
2    correlation::QueryId,
3    dynamic_filter::{DynamicFilter, DynamicFilterConsumerReductionStage, DynamicFilterRoute},
4};
5
6use super::{QueryMetrics, optional_i64, optional_str, optional_u64};
7
8/// Source-read row groups and bytes before or after applying a dynamic filter.
9#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
10pub struct DynamicFilterSourceSnapshot {
11    /// Planned row groups visible to the source consumer.
12    pub row_groups: u64,
13    /// Estimated source-read bytes, when known.
14    pub source_bytes: Option<u64>,
15}
16
17/// Runtime diagnostic row for one dynamic-filter route event.
18#[derive(Debug, Clone, PartialEq, Eq)]
19pub struct DynamicFilterRuntimeMetrics {
20    /// Query-wide diagnostic correlation id, when assigned.
21    pub(crate) query_id: Option<QueryId>,
22    /// Plan-local dynamic-filter route id.
23    pub dynamic_filter_id: String,
24    /// The event is one of `produced`, `skipped`, `applied`,
25    /// `reader_predicate_rows`, `applied_post_decode`, or `post_decode_rows`.
26    pub event: String,
27    /// Producer kind label.
28    pub producer_kind: String,
29    /// Producer plan node id.
30    pub producer_node_id: String,
31    /// Consumer kind label.
32    pub consumer_kind: String,
33    /// Consumer plan node id.
34    pub consumer_node_id: String,
35    /// Consumer lowering label selected during route discovery.
36    pub consumer_lowering: String,
37    /// Stage where the consumer can first reduce work.
38    pub consumer_reduction_stage: String,
39    /// Payload kind, or `none` for skipped events.
40    pub filter_kind: String,
41    /// Exact payload value count, when available.
42    pub payload_value_count: Option<u64>,
43    /// Inclusive integer-like minimum, when the payload carries one.
44    pub payload_min_i64: Option<i64>,
45    /// Inclusive integer-like maximum, when the payload carries one.
46    pub payload_max_i64: Option<i64>,
47    /// Producer output rows observed while creating the payload, when known.
48    pub producer_rows: Option<u64>,
49    /// Stable reason code for skipped events.
50    pub skipped_reason: Option<String>,
51    /// Source row groups before application.
52    pub selected_row_groups_before: Option<u64>,
53    /// Source row groups after application.
54    pub selected_row_groups_after: Option<u64>,
55    /// Source-read bytes before application.
56    pub selected_source_bytes_before: Option<u64>,
57    /// Source-read bytes after application.
58    pub selected_source_bytes_after: Option<u64>,
59    /// Source file index for source-reader row evidence.
60    pub source_file_index: Option<u64>,
61    /// Source chunk index for source-reader row evidence.
62    pub source_chunk_index: Option<u64>,
63    /// Selected source rows before the reader predicate was applied.
64    pub reader_predicate_rows_before: Option<u64>,
65    /// Materialized source rows after the reader predicate was applied.
66    pub reader_predicate_rows_after: Option<u64>,
67    /// Source rows removed by the reader predicate.
68    pub reader_predicate_rows_filtered: u64,
69    /// Materialized rows before a post-decode dynamic-filter stage.
70    pub post_decode_rows_before: Option<u64>,
71    /// Materialized rows after a post-decode dynamic-filter stage.
72    pub post_decode_rows_after: Option<u64>,
73    /// Materialized rows removed by a post-decode dynamic-filter stage.
74    pub post_decode_rows_filtered: u64,
75    /// Row groups removed by this application.
76    pub row_groups_pruned: u64,
77    /// Source-read bytes removed by this application.
78    pub source_bytes_pruned: u64,
79}
80
81impl DynamicFilterRuntimeMetrics {
82    fn produced(route: &DynamicFilterRoute, filter: &DynamicFilter, producer_rows: u64) -> Self {
83        let (payload_min_i64, payload_max_i64) =
84            filter.i64_range().map_or((None, None), |(min, max)| (Some(min), Some(max)));
85        Self {
86            query_id: None,
87            dynamic_filter_id: route.id.to_string(),
88            event: "produced".to_string(),
89            producer_kind: route.producer.kind().to_string(),
90            producer_node_id: route.producer.node_id().to_string(),
91            consumer_kind: route.consumer.kind().to_string(),
92            consumer_node_id: route.consumer.node_id().to_string(),
93            consumer_lowering: route.consumer_lowering.to_string(),
94            consumer_reduction_stage: route.consumer_reduction_stage().to_string(),
95            filter_kind: filter.kind().to_string(),
96            payload_value_count: filter.value_count().and_then(|value| u64::try_from(value).ok()),
97            payload_min_i64,
98            payload_max_i64,
99            producer_rows: Some(producer_rows),
100            skipped_reason: None,
101            selected_row_groups_before: None,
102            selected_row_groups_after: None,
103            selected_source_bytes_before: None,
104            selected_source_bytes_after: None,
105            source_file_index: None,
106            source_chunk_index: None,
107            reader_predicate_rows_before: None,
108            reader_predicate_rows_after: None,
109            reader_predicate_rows_filtered: 0,
110            post_decode_rows_before: None,
111            post_decode_rows_after: None,
112            post_decode_rows_filtered: 0,
113            row_groups_pruned: 0,
114            source_bytes_pruned: 0,
115        }
116    }
117
118    fn skipped(route: &DynamicFilterRoute, reason: impl Into<String>) -> Self {
119        Self {
120            query_id: None,
121            dynamic_filter_id: route.id.to_string(),
122            event: "skipped".to_string(),
123            producer_kind: route.producer.kind().to_string(),
124            producer_node_id: route.producer.node_id().to_string(),
125            consumer_kind: route.consumer.kind().to_string(),
126            consumer_node_id: route.consumer.node_id().to_string(),
127            consumer_lowering: route.consumer_lowering.to_string(),
128            consumer_reduction_stage: route.consumer_reduction_stage().to_string(),
129            filter_kind: "none".to_string(),
130            payload_value_count: None,
131            payload_min_i64: None,
132            payload_max_i64: None,
133            producer_rows: None,
134            skipped_reason: Some(reason.into()),
135            selected_row_groups_before: None,
136            selected_row_groups_after: None,
137            selected_source_bytes_before: None,
138            selected_source_bytes_after: None,
139            source_file_index: None,
140            source_chunk_index: None,
141            reader_predicate_rows_before: None,
142            reader_predicate_rows_after: None,
143            reader_predicate_rows_filtered: 0,
144            post_decode_rows_before: None,
145            post_decode_rows_after: None,
146            post_decode_rows_filtered: 0,
147            row_groups_pruned: 0,
148            source_bytes_pruned: 0,
149        }
150    }
151
152    fn applied(
153        route: &DynamicFilterRoute,
154        filter: &DynamicFilter,
155        before: DynamicFilterSourceSnapshot,
156        after: DynamicFilterSourceSnapshot,
157        row_groups_pruned: u64,
158        source_bytes_pruned: u64,
159    ) -> Self {
160        let (payload_min_i64, payload_max_i64) =
161            filter.i64_range().map_or((None, None), |(min, max)| (Some(min), Some(max)));
162        Self {
163            query_id: None,
164            dynamic_filter_id: route.id.to_string(),
165            event: "applied".to_string(),
166            producer_kind: route.producer.kind().to_string(),
167            producer_node_id: route.producer.node_id().to_string(),
168            consumer_kind: route.consumer.kind().to_string(),
169            consumer_node_id: route.consumer.node_id().to_string(),
170            consumer_lowering: route.consumer_lowering.to_string(),
171            consumer_reduction_stage: route.consumer_reduction_stage().to_string(),
172            filter_kind: filter.kind().to_string(),
173            payload_value_count: filter.value_count().and_then(|value| u64::try_from(value).ok()),
174            payload_min_i64,
175            payload_max_i64,
176            producer_rows: None,
177            skipped_reason: None,
178            selected_row_groups_before: Some(before.row_groups),
179            selected_row_groups_after: Some(after.row_groups),
180            selected_source_bytes_before: before.source_bytes,
181            selected_source_bytes_after: after.source_bytes,
182            source_file_index: None,
183            source_chunk_index: None,
184            reader_predicate_rows_before: None,
185            reader_predicate_rows_after: None,
186            reader_predicate_rows_filtered: 0,
187            post_decode_rows_before: None,
188            post_decode_rows_after: None,
189            post_decode_rows_filtered: 0,
190            row_groups_pruned,
191            source_bytes_pruned,
192        }
193    }
194
195    fn source_reader_predicate_rows(
196        route: &DynamicFilterRoute,
197        filter: &DynamicFilter,
198        source_file_index: Option<usize>,
199        source_chunk_index: Option<usize>,
200        rows_before: Option<u64>,
201        rows_after: Option<u64>,
202    ) -> Self {
203        let (payload_min_i64, payload_max_i64) =
204            filter.i64_range().map_or((None, None), |(min, max)| (Some(min), Some(max)));
205        let rows_filtered = rows_before.zip(rows_after).map_or(0, |(before, after)| before.saturating_sub(after));
206        Self {
207            query_id: None,
208            dynamic_filter_id: route.id.to_string(),
209            event: "reader_predicate_rows".to_string(),
210            producer_kind: route.producer.kind().to_string(),
211            producer_node_id: route.producer.node_id().to_string(),
212            consumer_kind: route.consumer.kind().to_string(),
213            consumer_node_id: route.consumer.node_id().to_string(),
214            consumer_lowering: route.consumer_lowering.as_str().to_string(),
215            consumer_reduction_stage: route.consumer_reduction_stage().to_string(),
216            filter_kind: filter.kind().as_str().to_string(),
217            payload_value_count: filter.value_count().map(|value| value as u64),
218            payload_min_i64,
219            payload_max_i64,
220            producer_rows: None,
221            skipped_reason: None,
222            selected_row_groups_before: None,
223            selected_row_groups_after: None,
224            selected_source_bytes_before: None,
225            selected_source_bytes_after: None,
226            source_file_index: source_file_index.map(|index| index as u64),
227            source_chunk_index: source_chunk_index.map(|index| index as u64),
228            reader_predicate_rows_before: rows_before,
229            reader_predicate_rows_after: rows_after,
230            reader_predicate_rows_filtered: rows_filtered,
231            post_decode_rows_before: None,
232            post_decode_rows_after: None,
233            post_decode_rows_filtered: 0,
234            row_groups_pruned: 0,
235            source_bytes_pruned: 0,
236        }
237    }
238
239    fn applied_post_decode(
240        route: &DynamicFilterRoute,
241        filter: &DynamicFilter,
242        rows_before: Option<u64>,
243        rows_after: Option<u64>,
244    ) -> Self {
245        Self::post_decode_event(route, filter, "applied_post_decode", rows_before, rows_after)
246    }
247
248    fn source_post_decode_stage(
249        route: &DynamicFilterRoute,
250        filter: &DynamicFilter,
251        rows_before: Option<u64>,
252        rows_after: Option<u64>,
253    ) -> Self {
254        Self::post_decode_event_with_stage(
255            route,
256            filter,
257            "applied_post_decode",
258            DynamicFilterConsumerReductionStage::SourcePostDecode,
259            rows_before,
260            rows_after,
261        )
262    }
263
264    fn post_decode_rows(
265        route: &DynamicFilterRoute,
266        filter: &DynamicFilter,
267        rows_before: Option<u64>,
268        rows_after: Option<u64>,
269    ) -> Self {
270        Self::post_decode_event(route, filter, "post_decode_rows", rows_before, rows_after)
271    }
272
273    fn post_decode_event(
274        route: &DynamicFilterRoute,
275        filter: &DynamicFilter,
276        event: &'static str,
277        rows_before: Option<u64>,
278        rows_after: Option<u64>,
279    ) -> Self {
280        Self::post_decode_event_with_stage(
281            route,
282            filter,
283            event,
284            route.consumer_reduction_stage(),
285            rows_before,
286            rows_after,
287        )
288    }
289
290    fn post_decode_event_with_stage(
291        route: &DynamicFilterRoute,
292        filter: &DynamicFilter,
293        event: &'static str,
294        consumer_reduction_stage: DynamicFilterConsumerReductionStage,
295        rows_before: Option<u64>,
296        rows_after: Option<u64>,
297    ) -> Self {
298        let (payload_min_i64, payload_max_i64) =
299            filter.i64_range().map_or((None, None), |(min, max)| (Some(min), Some(max)));
300        let rows_filtered = rows_before.zip(rows_after).map_or(0, |(before, after)| before.saturating_sub(after));
301        Self {
302            query_id: None,
303            dynamic_filter_id: route.id.to_string(),
304            event: event.to_string(),
305            producer_kind: route.producer.kind().to_string(),
306            producer_node_id: route.producer.node_id().to_string(),
307            consumer_kind: route.consumer.kind().to_string(),
308            consumer_node_id: route.consumer.node_id().to_string(),
309            consumer_lowering: route.consumer_lowering.to_string(),
310            consumer_reduction_stage: consumer_reduction_stage.to_string(),
311            filter_kind: filter.kind().to_string(),
312            payload_value_count: filter.value_count().and_then(|value| u64::try_from(value).ok()),
313            payload_min_i64,
314            payload_max_i64,
315            producer_rows: None,
316            skipped_reason: None,
317            selected_row_groups_before: None,
318            selected_row_groups_after: None,
319            selected_source_bytes_before: None,
320            selected_source_bytes_after: None,
321            source_file_index: None,
322            source_chunk_index: None,
323            reader_predicate_rows_before: None,
324            reader_predicate_rows_after: None,
325            reader_predicate_rows_filtered: 0,
326            post_decode_rows_before: rows_before,
327            post_decode_rows_after: rows_after,
328            post_decode_rows_filtered: rows_filtered,
329            row_groups_pruned: 0,
330            source_bytes_pruned: 0,
331        }
332    }
333
334    /// Serializes this row as stable comma-separated key/value pairs.
335    pub fn to_stable_kv_row(&self) -> String {
336        format!(
337            "dynamic_filter_id={},event={},producer_kind={},producer_node_id={},consumer_kind={},consumer_node_id={},consumer_lowering={},consumer_reduction_stage={},filter_kind={},payload_value_count={},payload_min_i64={},payload_max_i64={},producer_rows={},skipped_reason={},selected_row_groups_before={},selected_row_groups_after={},selected_source_bytes_before={},selected_source_bytes_after={},source_file_index={},source_chunk_index={},reader_predicate_rows_before={},reader_predicate_rows_after={},reader_predicate_rows_filtered={},post_decode_rows_before={},post_decode_rows_after={},post_decode_rows_filtered={},row_groups_pruned={},source_bytes_pruned={}",
338            self.dynamic_filter_id,
339            self.event,
340            self.producer_kind,
341            self.producer_node_id,
342            self.consumer_kind,
343            self.consumer_node_id,
344            self.consumer_lowering,
345            self.consumer_reduction_stage,
346            self.filter_kind,
347            optional_u64(self.payload_value_count),
348            optional_i64(self.payload_min_i64),
349            optional_i64(self.payload_max_i64),
350            optional_u64(self.producer_rows),
351            optional_str(self.skipped_reason.as_deref()),
352            optional_u64(self.selected_row_groups_before),
353            optional_u64(self.selected_row_groups_after),
354            optional_u64(self.selected_source_bytes_before),
355            optional_u64(self.selected_source_bytes_after),
356            optional_u64(self.source_file_index),
357            optional_u64(self.source_chunk_index),
358            optional_u64(self.reader_predicate_rows_before),
359            optional_u64(self.reader_predicate_rows_after),
360            self.reader_predicate_rows_filtered,
361            optional_u64(self.post_decode_rows_before),
362            optional_u64(self.post_decode_rows_after),
363            self.post_decode_rows_filtered,
364            self.row_groups_pruned,
365            self.source_bytes_pruned,
366        )
367    }
368}
369
370impl QueryMetrics {
371    fn record_dynamic_filter_runtime_row(&mut self, mut row: DynamicFilterRuntimeMetrics) {
372        if row.query_id.is_none() {
373            row.query_id = self.inherited_query_id();
374        }
375        self.dynamic_filter_runtime_rows.push(row);
376    }
377
378    /// Records a dynamic-filter payload produced at runtime.
379    pub fn record_dynamic_filter_produced(
380        &mut self,
381        route: &DynamicFilterRoute,
382        filter: &DynamicFilter,
383        producer_rows: u64,
384    ) {
385        self.dynamic_filter_produced_count = self.dynamic_filter_produced_count.saturating_add(1);
386        self.record_dynamic_filter_runtime_row(DynamicFilterRuntimeMetrics::produced(route, filter, producer_rows));
387    }
388
389    /// Records a dynamic-filter producer that completed without a usable payload.
390    pub fn record_dynamic_filter_skipped(&mut self, route: &DynamicFilterRoute, reason: impl Into<String>) {
391        self.dynamic_filter_skipped_count = self.dynamic_filter_skipped_count.saturating_add(1);
392        self.record_dynamic_filter_runtime_row(DynamicFilterRuntimeMetrics::skipped(route, reason));
393    }
394
395    /// Records a dynamic-filter payload applied to a local-Parquet source consumer.
396    pub fn record_dynamic_filter_applied(
397        &mut self,
398        route: &DynamicFilterRoute,
399        filter: &DynamicFilter,
400        before: DynamicFilterSourceSnapshot,
401        after: DynamicFilterSourceSnapshot,
402    ) {
403        let row_groups_pruned = before.row_groups.saturating_sub(after.row_groups);
404        let source_bytes_pruned = before.source_bytes.unwrap_or(0).saturating_sub(after.source_bytes.unwrap_or(0));
405        self.dynamic_filter_applied_count = self.dynamic_filter_applied_count.saturating_add(1);
406        self.dynamic_filter_row_groups_pruned = self.dynamic_filter_row_groups_pruned.saturating_add(row_groups_pruned);
407        self.dynamic_filter_source_bytes_pruned =
408            self.dynamic_filter_source_bytes_pruned.saturating_add(source_bytes_pruned);
409        self.record_dynamic_filter_runtime_row(DynamicFilterRuntimeMetrics::applied(
410            route,
411            filter,
412            before,
413            after,
414            row_groups_pruned,
415            source_bytes_pruned,
416        ));
417    }
418
419    /// Records a dynamic-filter payload applied to a non-source consumer.
420    pub fn record_dynamic_filter_applied_post_decode(
421        &mut self,
422        route: &DynamicFilterRoute,
423        filter: &DynamicFilter,
424        rows_before: Option<u64>,
425        rows_after: Option<u64>,
426    ) {
427        self.dynamic_filter_applied_count = self.dynamic_filter_applied_count.saturating_add(1);
428        self.record_dynamic_filter_post_decode_reduction(route.consumer_reduction_stage(), rows_before, rows_after);
429        self.record_dynamic_filter_runtime_row(DynamicFilterRuntimeMetrics::applied_post_decode(
430            route,
431            filter,
432            rows_before,
433            rows_after,
434        ));
435    }
436
437    /// Records an additional post-decode source stage for a route that was
438    /// already counted as applied to its source consumer.
439    pub fn record_dynamic_filter_source_post_decode_stage(
440        &mut self,
441        route: &DynamicFilterRoute,
442        filter: &DynamicFilter,
443        rows_before: Option<u64>,
444        rows_after: Option<u64>,
445    ) {
446        self.record_dynamic_filter_post_decode_reduction(
447            DynamicFilterConsumerReductionStage::SourcePostDecode,
448            rows_before,
449            rows_after,
450        );
451        self.record_dynamic_filter_runtime_row(DynamicFilterRuntimeMetrics::source_post_decode_stage(
452            route,
453            filter,
454            rows_before,
455            rows_after,
456        ));
457    }
458
459    /// Records post-decode row evidence for a route already counted as applied.
460    ///
461    /// Lazy stream consumers install a route once, but the row reduction is
462    /// only known as chunks are consumed.
463    pub fn record_dynamic_filter_post_decode_rows(
464        &mut self,
465        route: &DynamicFilterRoute,
466        filter: &DynamicFilter,
467        rows_before: Option<u64>,
468        rows_after: Option<u64>,
469    ) {
470        self.record_dynamic_filter_post_decode_reduction(route.consumer_reduction_stage(), rows_before, rows_after);
471        self.record_dynamic_filter_runtime_row(DynamicFilterRuntimeMetrics::post_decode_rows(
472            route,
473            filter,
474            rows_before,
475            rows_after,
476        ));
477    }
478
479    /// Records source-reader row evidence for a dynamic-filter reader predicate.
480    pub fn record_dynamic_filter_source_reader_predicate_rows(
481        &mut self,
482        route: &DynamicFilterRoute,
483        filter: &DynamicFilter,
484        source_file_index: usize,
485        source_chunk_index: usize,
486        rows_before: Option<u64>,
487        rows_after: Option<u64>,
488    ) {
489        self.record_dynamic_filter_runtime_row(DynamicFilterRuntimeMetrics::source_reader_predicate_rows(
490            route,
491            filter,
492            Some(source_file_index),
493            Some(source_chunk_index),
494            rows_before,
495            rows_after,
496        ));
497    }
498
499    /// Records source-reader row evidence for a batched dynamic-filter reader predicate.
500    pub fn record_dynamic_filter_source_reader_predicate_batch_rows(
501        &mut self,
502        route: &DynamicFilterRoute,
503        filter: &DynamicFilter,
504        rows_before: Option<u64>,
505        rows_after: Option<u64>,
506    ) {
507        self.record_dynamic_filter_runtime_row(DynamicFilterRuntimeMetrics::source_reader_predicate_rows(
508            route,
509            filter,
510            None,
511            None,
512            rows_before,
513            rows_after,
514        ));
515    }
516
517    /// Records de-duplicated source-reader row reduction for a chunk.
518    pub fn record_dynamic_filter_reader_predicate_source_reduction(
519        &mut self,
520        rows_before: Option<u64>,
521        rows_after: Option<u64>,
522    ) {
523        if let Some(rows_before) = rows_before {
524            self.dynamic_filter_reader_predicate_rows_before =
525                self.dynamic_filter_reader_predicate_rows_before.saturating_add(rows_before);
526        }
527        if let Some(rows_after) = rows_after {
528            self.dynamic_filter_reader_predicate_rows_after =
529                self.dynamic_filter_reader_predicate_rows_after.saturating_add(rows_after);
530        }
531        if let Some(rows_filtered) = rows_before.zip(rows_after).map(|(before, after)| before.saturating_sub(after)) {
532            self.dynamic_filter_reader_predicate_rows_filtered =
533                self.dynamic_filter_reader_predicate_rows_filtered.saturating_add(rows_filtered);
534        }
535    }
536
537    fn record_dynamic_filter_post_decode_reduction(
538        &mut self,
539        stage: DynamicFilterConsumerReductionStage,
540        rows_before: Option<u64>,
541        rows_after: Option<u64>,
542    ) {
543        if let Some(rows_before) = rows_before {
544            self.dynamic_filter_post_decode_rows_before =
545                self.dynamic_filter_post_decode_rows_before.saturating_add(rows_before);
546        }
547        if let Some(rows_after) = rows_after {
548            self.dynamic_filter_post_decode_rows_after =
549                self.dynamic_filter_post_decode_rows_after.saturating_add(rows_after);
550        }
551        if let Some(rows_filtered) = rows_before.zip(rows_after).map(|(before, after)| before.saturating_sub(after)) {
552            self.dynamic_filter_post_decode_rows_filtered =
553                self.dynamic_filter_post_decode_rows_filtered.saturating_add(rows_filtered);
554            match stage {
555                DynamicFilterConsumerReductionStage::SourcePostDecode => {
556                    self.dynamic_filter_source_post_decode_rows_filtered =
557                        self.dynamic_filter_source_post_decode_rows_filtered.saturating_add(rows_filtered);
558                },
559                DynamicFilterConsumerReductionStage::FilterPostDecode => {
560                    self.dynamic_filter_filter_post_decode_rows_filtered =
561                        self.dynamic_filter_filter_post_decode_rows_filtered.saturating_add(rows_filtered);
562                },
563                DynamicFilterConsumerReductionStage::JoinProbePostDecode => {
564                    self.dynamic_filter_join_probe_post_decode_rows_filtered =
565                        self.dynamic_filter_join_probe_post_decode_rows_filtered.saturating_add(rows_filtered);
566                },
567                DynamicFilterConsumerReductionStage::SourceReader => {},
568            }
569        }
570    }
571}
query_runtime/metrics/dynamic_filter.rs

query_runtime/metrics/
dynamic_filter.rs
