query_runtime/metrics/
mod.rs

1//! Query execution metrics and stable diagnostic event rows.
2//!
3//! Byte counters are device bytes unless the field name explicitly says host,
4//! Arrow, or Parquet compressed bytes.
5
6use std::{collections::BTreeMap, time::Duration};
7
8use crate::observability::graph_observation::GraphMeasuredPeakRow;
9
10mod aggregate;
11mod backend_descriptor;
12mod boundary;
13mod correlation;
14mod descriptor;
15mod dynamic_filter;
16mod graph;
17mod join;
18mod json;
19mod memory;
20mod operator;
21mod source_read;
22mod task;
23
24use correlation::inherited_query_id;
25pub(crate) use query_planning::correlation::CompiledTaskId;
26use query_planning::correlation::{QueryId, TraceId};
27
28pub(crate) use self::{
29    aggregate::AggregateStrategyDecisionMetrics,
30    correlation::PlanTaskMapping,
31    dynamic_filter::DynamicFilterRuntimeMetrics,
32    task::{TaskOperatorMetrics, TaskRuntimeMetrics},
33};
34pub use self::{
35    backend_descriptor::{
36        BACKEND_ATTEMPT_DURATION_BUCKETS_NS, BACKEND_METRIC_DESCRIPTORS, BackendDurationProjectionError, BackendMetric,
37        BackendMetricDescriptor, BackendMetricKind, BackendMetricLabel, BackendMetricUnit, BackendMetricUpdatePoint,
38        backend_metric_descriptors, project_backend_duration_histogram,
39    },
40    boundary::BoundaryMetrics,
41    descriptor::{MetricAggregation, MetricDescriptor, MetricDirection, MetricUnit},
42    dynamic_filter::DynamicFilterSourceSnapshot,
43    graph::GraphAlgorithmRuntimeMetrics,
44    join::{JoinOperatorMetrics, JoinStrategyDecisionMetrics},
45    json::QUERY_METRICS_JSON_SCHEMA,
46    operator::OperatorMetrics,
47    source_read::{SourceReadParquetEvidenceMetrics, SourceReadRuntimeMetrics},
48    task::BudgetedExecutionEvent,
49};
50use crate::observability::{LakehouseCacheFillEventSnapshot, MetricsCaptureMode, MetricsCompleteness};
51
52/// Query-scoped metrics accumulated during native execution.
53#[derive(Debug, Clone, Default, PartialEq, Eq)]
54pub struct QueryMetrics {
55    /// Query-wide diagnostic correlation id.
56    pub query_id: QueryId,
57    /// Optional parent trace id.
58    pub trace_id: Option<TraceId>,
59    /// Data-boundary transfer counters.
60    pub boundary: BoundaryMetrics,
61    /// Operator-level metric rows in observation order.
62    pub operators: Vec<OperatorMetrics>,
63    /// Graph algorithm runtime rows in observation order.
64    pub graph_algorithm_runtime_rows: Vec<GraphAlgorithmRuntimeMetrics>,
65    /// Fixed-shape measured graph-execution peak rows in observation order.
66    /// These are diagnostic projections and never form admission evidence.
67    pub graph_claim_calibration_rows: Vec<GraphMeasuredPeakRow>,
68    /// Join-strategy diagnostics captured during planning/execution.
69    pub join_strategy_decisions: Vec<JoinStrategyDecisionMetrics>,
70    /// Aggregate-strategy diagnostics captured during planning/execution.
71    pub aggregate_strategy_decisions: Vec<AggregateStrategyDecisionMetrics>,
72    /// Dynamic-filter runtime event rows in observation order.
73    pub dynamic_filter_runtime_rows: Vec<DynamicFilterRuntimeMetrics>,
74    /// Source-read runtime event rows in observation order.
75    pub source_read_runtime_rows: Vec<SourceReadRuntimeMetrics>,
76    /// Background lakehouse cache-fill completions this plan execution
77    /// carries: those its owner retained since the previous plan froze.
78    pub lakehouse_cache_fill_events: Vec<LakehouseCacheFillEventSnapshot>,
79    /// Number of input batches observed by native execution.
80    pub input_batches: u64,
81    /// Number of input rows observed by native execution.
82    pub input_rows: u64,
83    /// Backward-compatible alias for `resident_dataframe_peak_bytes`.
84    ///
85    /// This is not an allocator high-water mark.
86    pub peak_device_memory_bytes: Option<u64>,
87    /// Peak bytes retained by resident cudf DataFrames.
88    pub resident_dataframe_peak_bytes: Option<u64>,
89    /// Allocation high-water bytes of the attempt's allocation domain when the
90    /// plan execution ended. Absent without an admitted attempt.
91    pub allocator_high_water_bytes: Option<u64>,
92    /// Largest lower-level budgeted cuDF peak reported by a task-like operation.
93    pub max_task_observed_peak_bytes: Option<u64>,
94    /// Largest retained byte count reported by a task-like lower-level cuDF operation.
95    pub max_task_retained_bytes: Option<u64>,
96    #[doc(hidden)]
97    pub task_kind_by_compiled_task_id: BTreeMap<String, String>,
98    /// Metrics aggregated by task operator kind.
99    pub task_operator_metrics: BTreeMap<String, TaskOperatorMetrics>,
100    /// Metrics keyed by compiled pipeline task id.
101    pub task_runtime_metrics: BTreeMap<String, TaskRuntimeMetrics>,
102    /// Query-local mapping rows between native plan nodes and compiled tasks.
103    pub plan_task_mapping_rows: Vec<PlanTaskMapping>,
104    /// Optional frontend physical-plan shape used to annotate plan/task mapping rows.
105    pub plan_task_mapping_datafusion_shape: Option<String>,
106    /// Successful lower-level cuDF calls that returned an execution-budget report.
107    ///
108    /// Token-routed operations without a report-returning cuDF API may still attach budget failure
109    /// facts on errors, but they do not synthesize success events here.
110    pub budgeted_execution_events: Vec<BudgetedExecutionEvent>,
111    /// Source chunks produced by native source planning.
112    pub source_chunks_planned: u64,
113    /// Includes both first attempts and retries.
114    pub source_chunks_attempted: u64,
115    /// Source chunks whose native read committed successfully.
116    pub source_chunks_committed: u64,
117    /// Additional source chunk attempts after recoverable resource failures.
118    pub source_chunks_retried: u64,
119    /// Source prefetch jobs submitted to workers.
120    pub source_read_prefetches_started: u64,
121    /// Source prefetch jobs that returned completed read results.
122    pub source_read_prefetches_completed: u64,
123    /// Source prefetch opportunities skipped before a job was started.
124    pub source_read_prefetches_skipped: u64,
125    /// Started source prefetches abandoned before their result was consumed.
126    pub source_read_prefetches_abandoned: u64,
127    /// Row groups planned across source chunks.
128    pub row_groups_planned: u64,
129    /// Row groups committed across source chunks.
130    pub row_groups_committed: u64,
131    /// Number of partial aggregate tasks.
132    pub partial_aggregate_count: u64,
133    /// Input rows consumed by partial aggregates.
134    pub partial_aggregate_input_rows: u64,
135    /// Rows produced by partial aggregates.
136    pub partial_aggregate_rows: u64,
137    /// Peak retained bytes from partial aggregate outputs.
138    pub partial_aggregate_retained_bytes: u64,
139    /// Native final aggregate merge attempts, including bounded retries.
140    pub final_merge_attempts: u64,
141    /// Aggregate merge levels executed over retained partial results.
142    pub final_merge_levels: u64,
143    /// Largest fan-in observed across final-merge levels.
144    pub final_merge_max_fan_in: u64,
145    /// Bounded native execution retries recorded by their owning operators.
146    pub retry_count: u64,
147    /// Retries caused by lower-level resource exhaustion.
148    pub retry_resource_exhausted_count: u64,
149    /// Number of reusable cuDF hash-join artifacts built during this query.
150    pub hash_join_build_count: u64,
151    /// Number of reusable cuDF filtered-join artifacts built during this query.
152    pub filtered_join_build_count: u64,
153    /// Number of one-shot keyed equality joins executed during this query.
154    pub one_shot_join_count: u64,
155    /// Unique device bytes retained by reusable cuDF hash-join build artifacts.
156    ///
157    /// The value is recorded when an artifact is constructed. Reusing an artifact
158    /// does not record it again.
159    pub hash_join_hash_table_bytes: u64,
160    /// Unique device bytes retained by reusable cuDF filtered-join build artifacts.
161    ///
162    /// The value is recorded when an artifact is constructed. Reusing an artifact
163    /// does not record it again.
164    pub filtered_join_hash_table_bytes: u64,
165    /// Unique device bytes retained by filtered-join probe states.
166    pub filtered_join_probe_state_bytes: u64,
167    /// Unique device bytes retained by mixed inner/left join size-data states.
168    pub mixed_join_size_data_bytes: u64,
169    /// Device bytes retained by one-shot join index maps while materializing output.
170    pub one_shot_join_index_bytes: u64,
171    /// Reservation bytes requested for one-shot join output materialization.
172    pub one_shot_join_materialization_bytes: u64,
173    /// Observed device bytes of one-shot join output frames.
174    pub one_shot_join_output_bytes: u64,
175    /// Number of Parquet files observed by source reads.
176    pub parquet_files: u64,
177    /// Sum of known Parquet file bytes.
178    pub parquet_bytes_known: u64,
179    /// Row groups selected by source reads.
180    pub row_groups_selected: u64,
181    /// Physical read columns selected by source reads.
182    pub read_columns: u64,
183    /// Time spent loading source metadata.
184    pub source_metadata_load_time: Duration,
185    /// Time spent waiting in the source-read queue.
186    pub source_read_queue_wait_time: Duration,
187    /// Time spent inside Parquet read calls.
188    pub parquet_read_time: Duration,
189    /// Time spent in source-read orchestration.
190    pub source_read_time: Duration,
191    /// Time spent in non-source operators.
192    pub operator_time: Duration,
193    /// Time spent in per-chunk operators.
194    pub chunk_operator_time: Duration,
195    /// Time spent in final merge operators.
196    pub final_merge_time: Duration,
197    /// Time spent in final sort operators.
198    pub final_sort_time: Duration,
199    /// Time spent materializing final outputs.
200    pub materialization_time: Duration,
201    /// Bounded queries whose capability-admitted graph was accepted by the
202    /// compiled-pipeline executor before execution started.
203    pub compiled_pipeline_attempts: u64,
204    /// Compiled-pipeline tasks completed by the scheduler.
205    pub compiled_pipeline_tasks_executed: u64,
206    /// Host Arrow sink boundaries materialized by the compiled-pipeline executor.
207    pub compiled_pipeline_host_arrow_sinks: u64,
208    /// Dynamic-filter payloads successfully produced.
209    pub dynamic_filter_produced_count: u64,
210    /// Dynamic-filter routes whose producers completed without a usable payload.
211    pub dynamic_filter_skipped_count: u64,
212    /// Dynamic-filter payloads applied to a runtime consumer.
213    pub dynamic_filter_applied_count: u64,
214    /// Row groups removed from source reads by dynamic filters.
215    pub dynamic_filter_row_groups_pruned: u64,
216    /// Estimated source-read bytes removed by dynamic filters.
217    pub dynamic_filter_source_bytes_pruned: u64,
218    /// Source rows selected before dynamic reader predicates were applied.
219    ///
220    /// Counted once per source chunk even when several dynamic-filter routes
221    /// merge into one reader predicate; query-level totals do not double count.
222    pub dynamic_filter_reader_predicate_rows_before: u64,
223    /// Source rows materialized after dynamic reader predicates were applied.
224    ///
225    /// Counted once per source chunk, matching
226    /// [`Self::dynamic_filter_reader_predicate_rows_before`].
227    pub dynamic_filter_reader_predicate_rows_after: u64,
228    /// Source rows removed by dynamic reader predicates.
229    ///
230    /// Counted once per source chunk and only when both before/after row counts
231    /// are known.
232    pub dynamic_filter_reader_predicate_rows_filtered: u64,
233    /// Materialized rows before post-decode dynamic filters were applied.
234    ///
235    /// Counted once per post-decode filter application or stream chunk.
236    pub dynamic_filter_post_decode_rows_before: u64,
237    /// Materialized rows after post-decode dynamic filters were applied.
238    ///
239    /// Counted once per post-decode filter application or stream chunk.
240    pub dynamic_filter_post_decode_rows_after: u64,
241    /// Materialized rows removed by post-decode dynamic filters.
242    ///
243    /// Counted once per post-decode filter application or stream chunk and
244    /// only when both before/after row counts are known.
245    pub dynamic_filter_post_decode_rows_filtered: u64,
246    /// Materialized rows removed by source-adjacent post-decode dynamic filters.
247    pub dynamic_filter_source_post_decode_rows_filtered: u64,
248    /// Materialized rows removed by filter-operator post-decode dynamic filters.
249    pub dynamic_filter_filter_post_decode_rows_filtered: u64,
250    /// Materialized rows removed by hash-join-probe post-decode dynamic filters.
251    pub dynamic_filter_join_probe_post_decode_rows_filtered: u64,
252    /// Where this value's owner-projected metrics came from and what they can
253    /// claim. `None` means no observation owner has projected into it.
254    pub projection: Option<MetricsProjection>,
255    /// Join gather-map bytes across every recorded operator. Kept as a total
256    /// because operator rows are retained under a bound and may be incomplete.
257    pub join_gather_map_bytes: u64,
258}
259
260/// Provenance and completeness of metrics projected from an observation owner.
261///
262/// A `QueryMetrics` is a detached value: editing it never writes back to the
263/// runtime. This records which checkpoint it was projected from, so a consumer
264/// can tell a complete report from a truncated or uncaptured one.
265#[derive(Debug, Clone, Copy, PartialEq, Eq)]
266pub struct MetricsProjection {
267    /// Revision of the one checkpoint this value was projected from, or `None`
268    /// once projections from several checkpoints were merged into it.
269    pub(crate) revision: Option<u64>,
270    /// Whether that checkpoint's owner had already sealed.
271    pub(crate) sealed: bool,
272    /// Capture and retention completeness inherited from the source checkpoint.
273    pub completeness: MetricsCompleteness,
274}
275
276impl MetricsProjection {
277    /// Combines two projections so the result never claims more than either:
278    /// detail is disabled if either side was, and losses and saturation add up.
279    fn merged(self, other: Self) -> Self {
280        let capture_mode = match (self.completeness.capture_mode, other.completeness.capture_mode) {
281            (MetricsCaptureMode::Bounded, MetricsCaptureMode::Bounded) => MetricsCaptureMode::Bounded,
282            _ => MetricsCaptureMode::Disabled,
283        };
284        let detail_rows_dropped =
285            self.completeness.detail_rows_dropped.saturating_add(other.completeness.detail_rows_dropped);
286        Self {
287            revision: None,
288            sealed: self.sealed && other.sealed,
289            completeness: MetricsCompleteness {
290                capture_mode,
291                detail_rows_dropped,
292                truncated: self.completeness.truncated || other.completeness.truncated,
293                totals_saturated: self.completeness.totals_saturated || other.completeness.totals_saturated,
294            },
295        }
296    }
297}
298
299impl QueryMetrics {
300    /// Creates empty metrics for a query id minted or supplied by the caller.
301    pub fn new(query_id: impl Into<QueryId>) -> Self {
302        Self { query_id: query_id.into(), ..Default::default() }
303    }
304
305    /// Returns the query-wide diagnostic correlation id.
306    pub fn query_id(&self) -> &QueryId {
307        &self.query_id
308    }
309
310    /// Assigns a query id to this metrics value and any already-collected runtime rows.
311    pub fn set_query_id(&mut self, query_id: impl Into<QueryId>) {
312        self.query_id = query_id.into();
313        self.stamp_existing_rows_with_query_id();
314    }
315
316    /// Consumes this metrics value, assigns the query id as `set_query_id` does, and returns it.
317    #[must_use]
318    pub fn with_query_id(mut self, query_id: impl Into<QueryId>) -> Self {
319        self.set_query_id(query_id);
320        self
321    }
322
323    /// Sets the frontend shape on this value and on each plan/task mapping row that has
324    /// none. An empty shape is ignored.
325    #[must_use]
326    pub fn with_plan_task_mapping_datafusion_shape(mut self, shape: impl Into<String>) -> Self {
327        let shape = shape.into();
328        if !shape.is_empty() {
329            self.plan_task_mapping_datafusion_shape = Some(shape.clone());
330            for row in &mut self.plan_task_mapping_rows {
331                if row.datafusion_shape.is_none() {
332                    row.datafusion_shape = Some(shape.clone());
333                }
334            }
335        }
336        self
337    }
338
339    pub(super) fn inherited_query_id(&self) -> Option<QueryId> {
340        inherited_query_id(&self.query_id)
341    }
342
343    pub(crate) fn stamp_existing_rows_with_query_id(&mut self) {
344        let Some(query_id) = self.inherited_query_id() else {
345            return;
346        };
347        for row in &mut self.budgeted_execution_events {
348            row.query_id = Some(query_id.clone());
349        }
350        for row in &mut self.dynamic_filter_runtime_rows {
351            row.query_id = Some(query_id.clone());
352        }
353        for row in &mut self.source_read_runtime_rows {
354            row.query_id = Some(query_id.clone());
355        }
356        for row in &mut self.join_strategy_decisions {
357            row.query_id = Some(query_id.clone());
358        }
359        for row in &mut self.aggregate_strategy_decisions {
360            row.query_id = Some(query_id.clone());
361        }
362        for row in self.task_operator_metrics.values_mut() {
363            row.query_id = Some(query_id.clone());
364        }
365        for row in self.task_runtime_metrics.values_mut() {
366            row.query_id = Some(query_id.clone());
367        }
368        for row in &mut self.plan_task_mapping_rows {
369            row.query_id = query_id.clone();
370        }
371        for row in &mut self.operators {
372            row.query_id = Some(query_id.clone());
373            if let Some(join) = &mut row.join {
374                join.query_id = Some(query_id.clone());
375            }
376        }
377        for row in &mut self.graph_algorithm_runtime_rows {
378            row.query_id = Some(query_id.clone());
379        }
380        for row in &mut self.graph_claim_calibration_rows {
381            row.query_id = Some(query_id.clone());
382        }
383    }
384
385    /// Merges another query metrics value into this one.
386    pub fn merge_from(&mut self, mut other: QueryMetrics) {
387        if !self.query_id.is_assigned() && other.query_id.is_assigned() {
388            self.query_id = other.query_id.clone();
389            self.stamp_existing_rows_with_query_id();
390        }
391        if self.query_id.is_assigned() && !other.query_id.is_assigned() {
392            other.set_query_id(self.query_id.clone());
393        }
394        if self.trace_id.is_none() {
395            self.trace_id = other.trace_id.clone();
396        }
397        self.boundary.merge_from(other.boundary);
398        self.operators.extend(other.operators);
399        self.graph_algorithm_runtime_rows.extend(other.graph_algorithm_runtime_rows);
400        self.graph_claim_calibration_rows.extend(other.graph_claim_calibration_rows);
401        self.join_strategy_decisions.extend(other.join_strategy_decisions);
402        self.aggregate_strategy_decisions.extend(other.aggregate_strategy_decisions);
403        self.dynamic_filter_runtime_rows.extend(other.dynamic_filter_runtime_rows);
404        self.source_read_runtime_rows.extend(other.source_read_runtime_rows);
405        self.lakehouse_cache_fill_events.extend(other.lakehouse_cache_fill_events);
406        self.input_batches = self.input_batches.saturating_add(other.input_batches);
407        self.input_rows = self.input_rows.saturating_add(other.input_rows);
408        self.peak_device_memory_bytes = self.peak_device_memory_bytes.max(other.peak_device_memory_bytes);
409        self.resident_dataframe_peak_bytes =
410            self.resident_dataframe_peak_bytes.max(other.resident_dataframe_peak_bytes);
411        self.allocator_high_water_bytes = self.allocator_high_water_bytes.max(other.allocator_high_water_bytes);
412        self.max_task_observed_peak_bytes = self.max_task_observed_peak_bytes.max(other.max_task_observed_peak_bytes);
413        self.max_task_retained_bytes = self.max_task_retained_bytes.max(other.max_task_retained_bytes);
414        self.task_kind_by_compiled_task_id.extend(other.task_kind_by_compiled_task_id);
415        self.plan_task_mapping_rows.extend(other.plan_task_mapping_rows);
416        if self.plan_task_mapping_datafusion_shape.is_none() {
417            self.plan_task_mapping_datafusion_shape = other.plan_task_mapping_datafusion_shape;
418        }
419        for (operator_kind, metrics) in other.task_operator_metrics {
420            self.task_operator_metrics
421                .entry(operator_kind.clone())
422                .or_insert_with(|| TaskOperatorMetrics { operator_kind, ..Default::default() })
423                .merge_from(metrics);
424        }
425        for (compiled_task_id, metrics) in other.task_runtime_metrics {
426            self.task_runtime_metrics
427                .entry(compiled_task_id.clone())
428                .or_insert_with(|| TaskRuntimeMetrics { compiled_task_id, ..Default::default() })
429                .merge_from(metrics);
430        }
431        self.budgeted_execution_events.extend(other.budgeted_execution_events);
432        self.source_chunks_planned = self.source_chunks_planned.saturating_add(other.source_chunks_planned);
433        self.source_chunks_attempted = self.source_chunks_attempted.saturating_add(other.source_chunks_attempted);
434        self.source_chunks_committed = self.source_chunks_committed.saturating_add(other.source_chunks_committed);
435        self.source_chunks_retried = self.source_chunks_retried.saturating_add(other.source_chunks_retried);
436        self.source_read_prefetches_started =
437            self.source_read_prefetches_started.saturating_add(other.source_read_prefetches_started);
438        self.source_read_prefetches_completed =
439            self.source_read_prefetches_completed.saturating_add(other.source_read_prefetches_completed);
440        self.source_read_prefetches_skipped =
441            self.source_read_prefetches_skipped.saturating_add(other.source_read_prefetches_skipped);
442        self.source_read_prefetches_abandoned =
443            self.source_read_prefetches_abandoned.saturating_add(other.source_read_prefetches_abandoned);
444        self.row_groups_planned = self.row_groups_planned.saturating_add(other.row_groups_planned);
445        self.row_groups_committed = self.row_groups_committed.saturating_add(other.row_groups_committed);
446        self.partial_aggregate_count = self.partial_aggregate_count.saturating_add(other.partial_aggregate_count);
447        self.partial_aggregate_input_rows =
448            self.partial_aggregate_input_rows.saturating_add(other.partial_aggregate_input_rows);
449        self.partial_aggregate_rows = self.partial_aggregate_rows.saturating_add(other.partial_aggregate_rows);
450        self.partial_aggregate_retained_bytes =
451            self.partial_aggregate_retained_bytes.max(other.partial_aggregate_retained_bytes);
452        self.final_merge_attempts = self.final_merge_attempts.saturating_add(other.final_merge_attempts);
453        self.final_merge_levels = self.final_merge_levels.saturating_add(other.final_merge_levels);
454        self.final_merge_max_fan_in = self.final_merge_max_fan_in.max(other.final_merge_max_fan_in);
455        self.retry_count = self.retry_count.saturating_add(other.retry_count);
456        self.retry_resource_exhausted_count =
457            self.retry_resource_exhausted_count.saturating_add(other.retry_resource_exhausted_count);
458        self.hash_join_build_count = self.hash_join_build_count.saturating_add(other.hash_join_build_count);
459        self.filtered_join_build_count = self.filtered_join_build_count.saturating_add(other.filtered_join_build_count);
460        self.one_shot_join_count = self.one_shot_join_count.saturating_add(other.one_shot_join_count);
461        self.hash_join_hash_table_bytes =
462            self.hash_join_hash_table_bytes.saturating_add(other.hash_join_hash_table_bytes);
463        self.filtered_join_hash_table_bytes =
464            self.filtered_join_hash_table_bytes.saturating_add(other.filtered_join_hash_table_bytes);
465        self.filtered_join_probe_state_bytes =
466            self.filtered_join_probe_state_bytes.saturating_add(other.filtered_join_probe_state_bytes);
467        self.mixed_join_size_data_bytes =
468            self.mixed_join_size_data_bytes.saturating_add(other.mixed_join_size_data_bytes);
469        self.one_shot_join_index_bytes = self.one_shot_join_index_bytes.saturating_add(other.one_shot_join_index_bytes);
470        self.one_shot_join_materialization_bytes =
471            self.one_shot_join_materialization_bytes.saturating_add(other.one_shot_join_materialization_bytes);
472        self.one_shot_join_output_bytes =
473            self.one_shot_join_output_bytes.saturating_add(other.one_shot_join_output_bytes);
474        self.parquet_files = self.parquet_files.saturating_add(other.parquet_files);
475        self.parquet_bytes_known = self.parquet_bytes_known.saturating_add(other.parquet_bytes_known);
476        self.row_groups_selected = self.row_groups_selected.saturating_add(other.row_groups_selected);
477        self.read_columns = self.read_columns.saturating_add(other.read_columns);
478        self.source_metadata_load_time += other.source_metadata_load_time;
479        self.source_read_queue_wait_time += other.source_read_queue_wait_time;
480        self.parquet_read_time += other.parquet_read_time;
481        self.source_read_time += other.source_read_time;
482        self.operator_time += other.operator_time;
483        self.chunk_operator_time += other.chunk_operator_time;
484        self.final_merge_time += other.final_merge_time;
485        self.final_sort_time += other.final_sort_time;
486        self.materialization_time += other.materialization_time;
487        self.compiled_pipeline_attempts =
488            self.compiled_pipeline_attempts.saturating_add(other.compiled_pipeline_attempts);
489        self.compiled_pipeline_tasks_executed =
490            self.compiled_pipeline_tasks_executed.saturating_add(other.compiled_pipeline_tasks_executed);
491        self.compiled_pipeline_host_arrow_sinks =
492            self.compiled_pipeline_host_arrow_sinks.saturating_add(other.compiled_pipeline_host_arrow_sinks);
493        self.dynamic_filter_produced_count =
494            self.dynamic_filter_produced_count.saturating_add(other.dynamic_filter_produced_count);
495        self.dynamic_filter_skipped_count =
496            self.dynamic_filter_skipped_count.saturating_add(other.dynamic_filter_skipped_count);
497        self.dynamic_filter_applied_count =
498            self.dynamic_filter_applied_count.saturating_add(other.dynamic_filter_applied_count);
499        self.dynamic_filter_row_groups_pruned =
500            self.dynamic_filter_row_groups_pruned.saturating_add(other.dynamic_filter_row_groups_pruned);
501        self.dynamic_filter_source_bytes_pruned =
502            self.dynamic_filter_source_bytes_pruned.saturating_add(other.dynamic_filter_source_bytes_pruned);
503        self.dynamic_filter_reader_predicate_rows_before = self
504            .dynamic_filter_reader_predicate_rows_before
505            .saturating_add(other.dynamic_filter_reader_predicate_rows_before);
506        self.dynamic_filter_reader_predicate_rows_after = self
507            .dynamic_filter_reader_predicate_rows_after
508            .saturating_add(other.dynamic_filter_reader_predicate_rows_after);
509        self.dynamic_filter_reader_predicate_rows_filtered = self
510            .dynamic_filter_reader_predicate_rows_filtered
511            .saturating_add(other.dynamic_filter_reader_predicate_rows_filtered);
512        self.dynamic_filter_post_decode_rows_before =
513            self.dynamic_filter_post_decode_rows_before.saturating_add(other.dynamic_filter_post_decode_rows_before);
514        self.dynamic_filter_post_decode_rows_after =
515            self.dynamic_filter_post_decode_rows_after.saturating_add(other.dynamic_filter_post_decode_rows_after);
516        self.dynamic_filter_post_decode_rows_filtered = self
517            .dynamic_filter_post_decode_rows_filtered
518            .saturating_add(other.dynamic_filter_post_decode_rows_filtered);
519        self.dynamic_filter_source_post_decode_rows_filtered = self
520            .dynamic_filter_source_post_decode_rows_filtered
521            .saturating_add(other.dynamic_filter_source_post_decode_rows_filtered);
522        self.dynamic_filter_filter_post_decode_rows_filtered = self
523            .dynamic_filter_filter_post_decode_rows_filtered
524            .saturating_add(other.dynamic_filter_filter_post_decode_rows_filtered);
525        self.dynamic_filter_join_probe_post_decode_rows_filtered = self
526            .dynamic_filter_join_probe_post_decode_rows_filtered
527            .saturating_add(other.dynamic_filter_join_probe_post_decode_rows_filtered);
528        self.join_gather_map_bytes = self.join_gather_map_bytes.saturating_add(other.join_gather_map_bytes);
529        // A side with no projection contributed no owner-projected metrics, so
530        // it is the identity here. An absent projection does not erase the other.
531        self.projection = match (self.projection, other.projection) {
532            (Some(left), Some(right)) => Some(left.merged(right)),
533            (left, right) => left.or(right),
534        };
535        self.stamp_existing_rows_with_query_id();
536    }
537}
538fn optional_str(value: Option<&str>) -> &str {
539    value.unwrap_or("none")
540}
541
542fn optional_u64(value: Option<u64>) -> String {
543    value.map_or_else(|| "none".to_string(), |value| value.to_string())
544}
545
546fn optional_i64(value: Option<i64>) -> String {
547    value.map_or_else(|| "none".to_string(), |value| value.to_string())
548}
query_runtime/metrics/mod.rs

query_runtime/metrics/
mod.rs
