query_runtime/metrics/
json.rs

1use std::collections::BTreeMap;
2
3use query_planning::correlation::{PlanNodeId, QueryId, TraceId};
4use serde_json::{Map, Value, json};
5
6use super::{
7    AggregateStrategyDecisionMetrics, BudgetedExecutionEvent, CompiledTaskId, DynamicFilterRuntimeMetrics,
8    GraphAlgorithmRuntimeMetrics, JoinOperatorMetrics, JoinStrategyDecisionMetrics, OperatorMetrics, PlanTaskMapping,
9    QueryMetrics, SourceReadParquetEvidenceMetrics, SourceReadRuntimeMetrics, TaskOperatorMetrics, TaskRuntimeMetrics,
10};
11use crate::observability::graph_observation::GraphMeasuredPeakRow;
12
13/// Identifies the layout of the JSON document [`QueryMetrics::to_json`] emits.
14pub const QUERY_METRICS_JSON_SCHEMA: &str = "query_engine.query_metrics";
15
16impl QueryMetrics {
17    /// Serializes query metrics as a nested, self-describing JSON document.
18    ///
19    /// The `metrics` object is generated by walking [`Self::descriptors`], so
20    /// scalar JSON field names and descriptor names share one catalog.
21    pub fn to_json(&self) -> Value {
22        json!({
23            "schema": QUERY_METRICS_JSON_SCHEMA,
24            "query_id": self.query_id.as_str(),
25            "trace_id": self.trace_id.as_ref().map(TraceId::as_str),
26            "triage": self.triage_summary().to_json(),
27            "completeness": self.completeness_json(),
28            "descriptors": Self::descriptors().iter().map(|descriptor| descriptor.to_json()).collect::<Vec<_>>(),
29            "metrics": self.described_metrics_json(),
30            "rows": {
31                "budgeted_executions": self.budgeted_execution_events.iter().map(BudgetedExecutionEvent::to_json).collect::<Vec<_>>(),
32                "operators": self.operators.iter().map(OperatorMetrics::to_json).collect::<Vec<_>>(),
33                "graph_claim_calibration": self
34                    .graph_claim_calibration_rows
35                    .iter()
36                    .map(GraphMeasuredPeakRow::to_json)
37                    .collect::<Vec<_>>(),
38                "graph_runtime": self.graph_algorithm_runtime_rows.iter().map(GraphAlgorithmRuntimeMetrics::to_json).collect::<Vec<_>>(),
39                "task_operators": self.task_operator_metrics.values().map(TaskOperatorMetrics::to_json).collect::<Vec<_>>(),
40                "task_runtime": self.task_runtime_metrics.values().map(TaskRuntimeMetrics::to_json).collect::<Vec<_>>(),
41                "plan_task_mappings": self.plan_task_mappings().iter().map(PlanTaskMapping::to_json).collect::<Vec<_>>(),
42                "dynamic_filter_runtime": self.dynamic_filter_runtime_rows.iter().map(DynamicFilterRuntimeMetrics::to_json).collect::<Vec<_>>(),
43                "source_read_runtime": self.source_read_runtime_rows.iter().map(SourceReadRuntimeMetrics::to_json).collect::<Vec<_>>(),
44                "join_strategy_decisions": self.join_strategy_decisions.iter().map(JoinStrategyDecisionMetrics::to_json).collect::<Vec<_>>(),
45                "aggregate_strategy_decisions": self.aggregate_strategy_decisions.iter().map(AggregateStrategyDecisionMetrics::to_json).collect::<Vec<_>>(),
46            },
47        })
48    }
49
50    /// Where these rows came from and whether they are all the detail there
51    /// was. A value not projected from an observation owner cannot say, so its
52    /// completeness is reported as unknown.
53    fn completeness_json(&self) -> Value {
54        let Some(projection) = &self.projection else {
55            return json!({ "status": "unknown" });
56        };
57        let completeness = &projection.completeness;
58        json!({
59            "status": completeness.detail_status(),
60            "revision": projection.revision,
61            "sealed": projection.sealed,
62            "capture_mode": completeness.capture_mode.as_str(),
63            "truncated": completeness.truncated,
64            "detail_rows_dropped": completeness.detail_rows_dropped,
65            "totals_saturated": completeness.totals_saturated,
66        })
67    }
68
69    fn described_metrics_json(&self) -> Value {
70        let mut metrics = Map::new();
71        for descriptor in Self::descriptors() {
72            let value = self
73                .metric_value_json(descriptor.name)
74                .unwrap_or_else(|| panic!("QueryMetrics descriptor lacks JSON value: {}", descriptor.name));
75            metrics.insert(
76                descriptor.name.to_string(),
77                json!({
78                    "value": value,
79                    "unit": descriptor.unit.as_str(),
80                    "aggregation": descriptor.aggregation.as_str(),
81                    "direction": descriptor.direction.map(|direction| direction.as_str()),
82                    "healthy": descriptor.healthy,
83                }),
84            );
85        }
86        Value::Object(metrics)
87    }
88
89    fn metric_value_json(&self, name: &str) -> Option<Value> {
90        Some(match name {
91            "input_batches" => json!(self.input_batches),
92            "input_rows" => json!(self.input_rows),
93            "gpu_resident_input_handoffs" => json!(self.boundary.gpu_resident_input_handoffs),
94            "host_to_device_uploads" => json!(self.boundary.host_to_device_uploads),
95            "host_to_device_uploaded_rows" => json!(self.boundary.host_to_device_uploaded_rows),
96            "host_to_device_uploaded_bytes" => json!(self.boundary.host_to_device_uploaded_bytes),
97            "device_to_host_materializations" => json!(self.boundary.device_to_host_materializations),
98            "device_to_host_materialized_rows" => json!(self.boundary.device_to_host_materialized_rows),
99            "device_to_host_materialized_bytes" => json!(self.boundary.device_to_host_materialized_bytes),
100            "device_to_device_copies" => json!(self.boundary.device_to_device_copies),
101            "device_to_device_copied_bytes" => json!(self.boundary.device_to_device_copied_bytes),
102            "max_task_observed_peak_bytes" => optional_u64_json(self.max_task_observed_peak_bytes),
103            "max_task_observed_peak_bytes_observed" => json!(self.max_task_observed_peak_bytes.is_some()),
104            "max_task_retained_bytes" => optional_u64_json(self.max_task_retained_bytes),
105            "max_task_retained_bytes_observed" => json!(self.max_task_retained_bytes.is_some()),
106            "native_graph_claim_calibration_rows" => json!(self.graph_claim_calibration_rows.len()),
107            "native_graph_runtime_rows" => json!(self.graph_algorithm_runtime_rows.len()),
108            "native_task_operator_rows" => json!(self.task_operator_metrics.len()),
109            "native_task_runtime_rows" => json!(self.task_runtime_metrics.len()),
110            "native_plan_task_mapping_rows" => json!(self.plan_task_mappings().len()),
111            "native_dynamic_filter_runtime_rows" => json!(self.dynamic_filter_runtime_rows.len()),
112            "native_source_read_runtime_rows" => json!(self.source_read_runtime_rows.len()),
113            "peak_device_memory_bytes" => optional_u64_json(self.peak_device_memory_bytes),
114            "resident_dataframe_peak_bytes" => optional_u64_json(self.resident_dataframe_peak_bytes),
115            "resident_dataframe_peak_bytes_observed" => json!(self.resident_dataframe_peak_bytes.is_some()),
116            "allocator_high_water_bytes" => optional_u64_json(self.allocator_high_water_bytes),
117            "allocator_high_water_bytes_observed" => json!(self.allocator_high_water_bytes.is_some()),
118            "source_chunks_planned" => json!(self.source_chunks_planned),
119            "source_chunks_attempted" => json!(self.source_chunks_attempted),
120            "source_chunks_committed" => json!(self.source_chunks_committed),
121            "source_chunks_retried" => json!(self.source_chunks_retried),
122            "source_read_prefetches_started" => json!(self.source_read_prefetches_started),
123            "source_read_prefetches_completed" => json!(self.source_read_prefetches_completed),
124            "source_read_prefetches_skipped" => json!(self.source_read_prefetches_skipped),
125            "source_read_prefetches_abandoned" => json!(self.source_read_prefetches_abandoned),
126            "row_groups_planned" => json!(self.row_groups_planned),
127            "row_groups_committed" => json!(self.row_groups_committed),
128            "partial_aggregate_count" => json!(self.partial_aggregate_count),
129            "partial_aggregate_input_rows" => json!(self.partial_aggregate_input_rows),
130            "partial_aggregate_rows" => json!(self.partial_aggregate_rows),
131            "partial_aggregate_retained_bytes" => json!(self.partial_aggregate_retained_bytes),
132            "final_merge_attempts" => json!(self.final_merge_attempts),
133            "final_merge_levels" => json!(self.final_merge_levels),
134            "final_merge_max_fan_in" => json!(self.final_merge_max_fan_in),
135            "compiled_pipeline_attempts" => json!(self.compiled_pipeline_attempts),
136            "compiled_pipeline_tasks_executed" => json!(self.compiled_pipeline_tasks_executed),
137            "compiled_pipeline_host_arrow_sinks" => json!(self.compiled_pipeline_host_arrow_sinks),
138            "parquet_files" => json!(self.parquet_files),
139            "parquet_bytes_known" => json!(self.parquet_bytes_known),
140            "row_groups_selected" => json!(self.row_groups_selected),
141            "read_columns" => json!(self.read_columns),
142            "retry_count" => json!(self.retry_count),
143            "retry_resource_exhausted_count" => json!(self.retry_resource_exhausted_count),
144            "dynamic_filter_produced_count" => json!(self.dynamic_filter_produced_count),
145            "dynamic_filter_skipped_count" => json!(self.dynamic_filter_skipped_count),
146            "dynamic_filter_applied_count" => json!(self.dynamic_filter_applied_count),
147            "dynamic_filter_row_groups_pruned" => json!(self.dynamic_filter_row_groups_pruned),
148            "dynamic_filter_source_bytes_pruned" => json!(self.dynamic_filter_source_bytes_pruned),
149            "dynamic_filter_reader_predicate_rows_before" => {
150                json!(self.dynamic_filter_reader_predicate_rows_before)
151            },
152            "dynamic_filter_reader_predicate_rows_after" => {
153                json!(self.dynamic_filter_reader_predicate_rows_after)
154            },
155            "dynamic_filter_reader_predicate_rows_filtered" => {
156                json!(self.dynamic_filter_reader_predicate_rows_filtered)
157            },
158            "dynamic_filter_post_decode_rows_before" => json!(self.dynamic_filter_post_decode_rows_before),
159            "dynamic_filter_post_decode_rows_after" => json!(self.dynamic_filter_post_decode_rows_after),
160            "dynamic_filter_post_decode_rows_filtered" => json!(self.dynamic_filter_post_decode_rows_filtered),
161            "dynamic_filter_source_post_decode_rows_filtered" => {
162                json!(self.dynamic_filter_source_post_decode_rows_filtered)
163            },
164            "dynamic_filter_filter_post_decode_rows_filtered" => {
165                json!(self.dynamic_filter_filter_post_decode_rows_filtered)
166            },
167            "dynamic_filter_join_probe_post_decode_rows_filtered" => {
168                json!(self.dynamic_filter_join_probe_post_decode_rows_filtered)
169            },
170            "hash_join_build_count" => json!(self.hash_join_build_count),
171            "filtered_join_build_count" => json!(self.filtered_join_build_count),
172            "one_shot_join_count" => json!(self.one_shot_join_count),
173            "hash_join_hash_table_bytes" => json!(self.hash_join_hash_table_bytes),
174            "filtered_join_hash_table_bytes" => json!(self.filtered_join_hash_table_bytes),
175            "filtered_join_probe_state_bytes" => json!(self.filtered_join_probe_state_bytes),
176            "mixed_join_size_data_bytes" => json!(self.mixed_join_size_data_bytes),
177            "one_shot_join_index_bytes" => json!(self.one_shot_join_index_bytes),
178            "one_shot_join_materialization_bytes" => json!(self.one_shot_join_materialization_bytes),
179            "one_shot_join_output_bytes" => json!(self.one_shot_join_output_bytes),
180            "join_gather_map_bytes" => json!(self.join_gather_map_bytes),
181            "native_aggregate_strategy_rows" => json!(self.aggregate_strategy_decisions.len()),
182            "native_join_strategy_rows" => json!(self.join_strategy_decisions.len()),
183            "source_metadata_load_time" => json!(duration_ns(self.source_metadata_load_time)),
184            "source_read_queue_wait_time" => json!(duration_ns(self.source_read_queue_wait_time)),
185            "parquet_read_time" => json!(duration_ns(self.parquet_read_time)),
186            "source_read_time" => json!(duration_ns(self.source_read_time)),
187            "operator_time" => json!(duration_ns(self.operator_time)),
188            "chunk_operator_time" => json!(duration_ns(self.chunk_operator_time)),
189            "final_merge_time" => json!(duration_ns(self.final_merge_time)),
190            "final_sort_time" => json!(duration_ns(self.final_sort_time)),
191            "materialization_time" => json!(duration_ns(self.materialization_time)),
192            _ => return None,
193        })
194    }
195}
196
197impl BudgetedExecutionEvent {
198    /// Serializes this lower-level budget report as named JSON.
199    pub fn to_json(&self) -> Value {
200        json!({
201            "query_id": optional_query_id(self.query_id.as_ref()),
202            "operation": self.operation,
203            "boundary": self.boundary,
204            "budget_bytes": self.budget_bytes,
205            "enforcement": self.enforcement,
206            "status": self.status,
207            "allocator_high_water_bytes": self.allocator_high_water_bytes,
208            "retained_bytes": self.retained_bytes,
209            "output_bytes": self.output_bytes,
210            "output_rows": self.output_rows,
211            "context_facts": self.context_facts,
212        })
213    }
214}
215
216impl OperatorMetrics {
217    /// Serializes this operator metric row as named JSON.
218    pub fn to_json(&self) -> Value {
219        json!({
220            "query_id": optional_query_id(self.query_id.as_ref()),
221            "node_id": self.node_id,
222            "operator_kind": self.operator_kind,
223            "input_rows": self.input_rows,
224            "input_bytes": self.input_bytes,
225            "output_rows": self.output_rows,
226            "output_bytes": self.output_bytes,
227            "join": self.join.as_ref().map(JoinOperatorMetrics::to_json),
228        })
229    }
230}
231
232impl JoinOperatorMetrics {
233    /// Serializes this join-specific operator payload as named JSON.
234    pub fn to_json(&self) -> Value {
235        json!({
236            "query_id": optional_query_id(self.query_id.as_ref()),
237            "join_type": self.join_type,
238            "execution_strategy": self.execution_strategy,
239            "key_count": self.key_count,
240            "residual_predicate": self.residual_predicate,
241            "null_equality": self.null_equality,
242            "build_side": self.build_side,
243            "probe_side": self.probe_side,
244            "build_rows": self.build_rows,
245            "build_bytes": self.build_bytes,
246            "probe_rows": self.probe_rows,
247            "probe_bytes": self.probe_bytes,
248            "hash_table_bytes": self.hash_table_bytes,
249            "filtered_probe_state_bytes": self.filtered_probe_state_bytes,
250            "mixed_size_data_bytes": self.mixed_size_data_bytes,
251            "gather_map_bytes": self.gather_map_bytes,
252            "output_reservation_bytes": self.output_reservation_bytes,
253            "output_rows": self.output_rows,
254            "output_bytes": self.output_bytes,
255        })
256    }
257}
258
259impl GraphAlgorithmRuntimeMetrics {
260    /// Serializes this graph algorithm runtime row as named JSON.
261    pub fn to_json(&self) -> Value {
262        json!({
263            "query_id": optional_query_id(self.query_id.as_ref()),
264            "node_id": self.node_id,
265            "algorithm": self.algorithm,
266            "runtime_outcome": self.runtime_outcome,
267            "input_boundary": self.input_boundary,
268            "output_boundary": self.output_boundary,
269            "construction_policy": self.construction_policy,
270            "source_column": self.source_column,
271            "destination_column": self.destination_column,
272            "weight_column": self.weight_column,
273            "directed": self.directed,
274            "renumber": self.renumber,
275            "algorithm_parameters": self.algorithm_parameters,
276            "algorithm_metadata": self.algorithm_metadata,
277            "input_rows": self.input_rows,
278            "input_columns": self.input_columns,
279            "output_rows": self.output_rows,
280            "output_columns": self.output_columns,
281            "output_device_bytes": self.output_device_bytes,
282            "graph_peak_bytes": self.graph_peak_bytes,
283            "elapsed_ns": self.elapsed_ns,
284            "gpu_resident_input_handoffs": self.gpu_resident_input_handoffs,
285            "host_to_device_uploads": self.host_to_device_uploads,
286            "device_to_host_materializations": self.device_to_host_materializations,
287        })
288    }
289}
290
291impl TaskOperatorMetrics {
292    /// Serializes this per-operator task aggregate row as named JSON.
293    pub fn to_json(&self) -> Value {
294        json!({
295            "query_id": optional_query_id(self.query_id.as_ref()),
296            "operator_kind": self.operator_kind,
297            "task_count": self.task_count,
298            "elapsed_ns": duration_ns(self.elapsed_time),
299            "max_observed_peak_bytes": self.max_observed_peak_bytes,
300            "max_retained_bytes": self.max_retained_bytes,
301        })
302    }
303}
304
305impl TaskRuntimeMetrics {
306    /// Serializes this per-compiled-task runtime row as named JSON.
307    pub fn to_json(&self) -> Value {
308        json!({
309            "query_id": optional_query_id(self.query_id.as_ref()),
310            "compiled_task_id": self.compiled_task_id,
311            "operator_kind": self.operator_kind,
312            "elapsed_ns": duration_ns(self.elapsed_time),
313            "max_observed_peak_bytes": self.max_observed_peak_bytes,
314            "max_retained_bytes": self.max_retained_bytes,
315        })
316    }
317}
318
319impl PlanTaskMapping {
320    /// Serializes this plan-to-task mapping row as named JSON.
321    pub fn to_json(&self) -> Value {
322        json!({
323            "query_id": self.query_id.as_str(),
324            "plan_node_id": optional_plan_node_id(self.plan_node_id.as_ref()),
325            "compiled_task_id": optional_compiled_task_id(self.compiled_task_id.as_ref()),
326            "operator_kind": self.operator_kind,
327            "datafusion_shape": self.datafusion_shape,
328        })
329    }
330}
331
332impl DynamicFilterRuntimeMetrics {
333    /// Serializes this dynamic-filter runtime row as named JSON.
334    pub fn to_json(&self) -> Value {
335        json!({
336            "query_id": optional_query_id(self.query_id.as_ref()),
337            "dynamic_filter_id": self.dynamic_filter_id,
338            "event": self.event,
339            "producer_kind": self.producer_kind,
340            "producer_node_id": self.producer_node_id,
341            "consumer_kind": self.consumer_kind,
342            "consumer_node_id": self.consumer_node_id,
343            "consumer_lowering": self.consumer_lowering,
344            "consumer_reduction_stage": self.consumer_reduction_stage,
345            "filter_kind": self.filter_kind,
346            "payload_value_count": self.payload_value_count,
347            "payload_min_i64": self.payload_min_i64,
348            "payload_max_i64": self.payload_max_i64,
349            "producer_rows": self.producer_rows,
350            "skipped_reason": self.skipped_reason,
351            "selected_row_groups_before": self.selected_row_groups_before,
352            "selected_row_groups_after": self.selected_row_groups_after,
353            "selected_source_bytes_before": self.selected_source_bytes_before,
354            "selected_source_bytes_after": self.selected_source_bytes_after,
355            "source_file_index": self.source_file_index,
356            "source_chunk_index": self.source_chunk_index,
357            "reader_predicate_rows_before": self.reader_predicate_rows_before,
358            "reader_predicate_rows_after": self.reader_predicate_rows_after,
359            "reader_predicate_rows_filtered": self.reader_predicate_rows_filtered,
360            "post_decode_rows_before": self.post_decode_rows_before,
361            "post_decode_rows_after": self.post_decode_rows_after,
362            "post_decode_rows_filtered": self.post_decode_rows_filtered,
363            "row_groups_pruned": self.row_groups_pruned,
364            "source_bytes_pruned": self.source_bytes_pruned,
365        })
366    }
367}
368
369impl SourceReadRuntimeMetrics {
370    /// Serializes this source-read runtime row as named JSON.
371    pub fn to_json(&self) -> Value {
372        json!({
373            "query_id": optional_query_id(self.query_id.as_ref()),
374            "source_node_id": self.source_node_id,
375            "read_kind": self.read_kind,
376            "read_strategy": self.read_strategy,
377            "parquet_read_evidence": self.parquet_read_evidence.to_json(),
378            "batch_index": self.batch_index,
379            "chunk_count": self.chunk_count,
380            "first_chunk_index": self.first_chunk_index,
381            "last_chunk_index": self.last_chunk_index,
382            "first_file_index": self.first_file_index,
383            "last_file_index": self.last_file_index,
384            "file_count": self.file_count,
385            "row_group_count": self.row_group_count,
386            "read_columns_per_file": self.read_columns_per_file,
387            "read_columns_total": self.read_columns_total,
388            "known_file_bytes": self.known_file_bytes,
389            "estimated_source_read_bytes": self.estimated_source_read_bytes,
390            "known_row_group_bytes": self.known_row_group_bytes,
391            "known_projected_compressed_bytes": self.known_projected_compressed_bytes,
392            "known_projected_uncompressed_bytes": self.known_projected_uncompressed_bytes,
393            "known_pure_filter_compressed_bytes": self.known_pure_filter_compressed_bytes,
394            "known_pure_filter_uncompressed_bytes": self.known_pure_filter_uncompressed_bytes,
395            "output_rows": self.output_rows,
396            "reader_predicate": self.reader_predicate,
397            "reader_dynamic_filter_count": self.reader_dynamic_filter_count,
398            "source_batching_reason": self.source_batching_reason,
399            "source_chunk_planning_mode": self.source_chunk_planning_mode,
400            "source_chunk_planning_reason": self.source_chunk_planning_reason,
401            "elapsed_ns": duration_ns(self.elapsed),
402        })
403    }
404}
405
406impl SourceReadParquetEvidenceMetrics {
407    /// Serializes lower-layer Parquet evidence as named JSON.
408    pub fn to_json(&self) -> Value {
409        json!({
410            "input_kind": self.input_kind,
411            "remote_read_mode": self.remote_read_mode,
412            "backend": self.backend,
413            "backend_build_status": self.backend_build_status,
414            "remote_bytes": self.remote_bytes,
415            "remote_range_count": self.remote_range_count,
416            "device_read_support": self.device_read_support,
417            "pruning_input_row_groups": self.pruning_input_row_groups,
418            "row_groups_after_stats_filter": self.row_groups_after_stats_filter,
419            "row_groups_after_bloom_filter": self.row_groups_after_bloom_filter,
420            "bloom_pruning_status": self.bloom_pruning_status,
421        })
422    }
423}
424
425impl JoinStrategyDecisionMetrics {
426    /// Serializes this join-strategy diagnostic row as named JSON.
427    pub fn to_json(&self) -> Value {
428        json!({
429            "query_id": optional_query_id(self.query_id.as_ref()),
430            "node_id": self.node_id,
431            "facts": facts_json(&self.facts),
432        })
433    }
434}
435
436impl AggregateStrategyDecisionMetrics {
437    /// Serializes this aggregate-strategy diagnostic row as named JSON.
438    pub fn to_json(&self) -> Value {
439        json!({
440            "query_id": optional_query_id(self.query_id.as_ref()),
441            "node_id": self.node_id,
442            "facts": facts_json(&self.facts),
443        })
444    }
445}
446
447fn optional_query_id(value: Option<&QueryId>) -> Option<&str> {
448    value.map(QueryId::as_str)
449}
450
451fn optional_plan_node_id(value: Option<&PlanNodeId>) -> Option<&str> {
452    value.map(PlanNodeId::as_str)
453}
454
455fn optional_compiled_task_id(value: Option<&CompiledTaskId>) -> Option<&str> {
456    value.map(CompiledTaskId::as_str)
457}
458
459fn optional_u64_json(value: Option<u64>) -> Value {
460    value.map_or(Value::Null, |value| json!(value))
461}
462
463fn duration_ns(value: std::time::Duration) -> u64 {
464    value.as_nanos().min(u128::from(u64::MAX)) as u64
465}
466
467fn facts_json(facts: &BTreeMap<String, String>) -> Value {
468    json!(facts)
469}
query_runtime/metrics/json.rs

query_runtime/metrics/
json.rs
