Skip to main content

algeon_datafusion/native_exec/metrics/
record.rs

1//! Native query-metric recording into DataFusion wrapper metrics.
2
3use std::{
4    collections::{BTreeMap, HashSet},
5    fmt::{self, Write},
6    ops::Deref,
7    sync::{
8        Arc, Mutex,
9        atomic::{AtomicBool, Ordering},
10    },
11};
12
13use arrow::record_batch::RecordBatch;
14use datafusion::{
15    common::utils::memory::get_record_batch_memory_size,
16    physical_plan::metrics::{BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder, MetricType, MetricsSet},
17};
18use query_engine::exec::output::{PlanMetrics, QueryExecutionOutcome, QueryOutput};
19use query_planning::error::QueryError;
20use query_runtime::{
21    metrics::{JoinOperatorMetrics, QueryMetrics},
22    observability::{
23        AttemptId, LakehouseCacheFillEventSnapshot, QueryAttemptReport, RuntimeObservationConfig,
24        graph_observation::{GraphClaimExecutionState, GraphMeasuredPeakRow},
25    },
26};
27
28use super::{
29    SERVER_CAPTURE_LOSS_METRIC_NAME, capture::NativeExecutionMetricsCapture, contract::metric_name,
30    registry::NativeMetricRegistry,
31};
32use crate::{
33    native_exec::{admission::QueryAdmissionFacts, config::AlgeonNativeExecutionContext},
34    native_report::NativeDataFusionExecutionMetrics,
35};
36
37mod mapping;
38
39#[cfg(test)]
40pub(super) use mapping::NATIVE_ALGEON_RECORDING_DESCRIPTORS;
41
42/// A root-native-wrapper target that can receive one attempt projection after
43/// its owner has sealed the immutable report. It does not retain the report,
44/// own attempt lifecycle state, or expose unbounded labels.
45#[derive(Debug, Clone)]
46pub(crate) struct SealedAttemptMetricsTarget {
47    metrics: ExecutionPlanMetricsSet,
48    graph_claim_calibration_captures: Vec<GraphClaimCalibrationCapture>,
49    /// Every native wrapper's live metric set projected at seal, starting with
50    /// the elected root's own set. Only the root receives the attempt row.
51    execution_metrics: Vec<ExecutionPlanMetricsSet>,
52}
53
54impl SealedAttemptMetricsTarget {
55    pub(crate) fn new(
56        metrics: &ExecutionPlanMetricsSet,
57        graph_claim_calibration_capture: GraphClaimCalibrationCapture,
58    ) -> Self {
59        Self {
60            metrics: metrics.clone(),
61            graph_claim_calibration_captures: vec![graph_claim_calibration_capture],
62            execution_metrics: vec![metrics.clone()],
63        }
64    }
65
66    /// Adds the metric sets of the other native wrappers that execute under
67    /// this attempt, such as producers and relation functions below a composed
68    /// root.
69    pub(crate) fn with_execution_metrics(mut self, metrics: impl IntoIterator<Item = ExecutionPlanMetricsSet>) -> Self {
70        self.execution_metrics.extend(metrics);
71        self
72    }
73
74    pub(crate) fn with_graph_claim_calibration_captures(
75        mut self,
76        captures: impl IntoIterator<Item = GraphClaimCalibrationCapture>,
77    ) -> Self {
78        self.graph_claim_calibration_captures.extend(captures);
79        self
80    }
81
82    pub(crate) fn graph_claim_calibration_captures(&self) -> impl Iterator<Item = GraphClaimCalibrationCapture> + '_ {
83        self.graph_claim_calibration_captures.iter().cloned()
84    }
85
86    /// Emit the single fixed-cardinality DataFusion row for this attempt and
87    /// project the attempt's native execution metrics.
88    ///
89    /// The caller elects exactly one root wrapper before execution. All labels
90    /// are derived from the sealed projection; in particular, the attempt id and device
91    /// ordinal deliberately never become metric labels.
92    ///
93    /// Each wrapper's metrics project separately and merge with
94    /// [`NativeDataFusionExecutionMetrics::merge_from`]: cumulative counters
95    /// add and peaks keep the maximum, the rule the engine also applies when
96    /// it merges `QueryMetrics`. Admission sequence, wait, device ordinal, and
97    /// grant come from the sealed report, which records the attempt's one
98    /// admission even when several wrappers observed it.
99    pub(crate) fn record(
100        self,
101        report: &QueryAttemptReport,
102    ) -> (SealedAttemptGraphClaimCalibrationSnapshot, NativeDataFusionExecutionMetrics) {
103        let execution = sealed_execution_metrics(&self.execution_metrics, report);
104        let graph_claim_calibration = SealedAttemptGraphClaimCalibrationSnapshot::from_captures(
105            report.attempt_id,
106            self.graph_claim_calibration_captures,
107        );
108        let labels = report.sealed_projection().labels();
109        MetricBuilder::new(&self.metrics)
110            .with_type(MetricType::Summary)
111            .with_new_label("disposition", labels.disposition)
112            .with_new_label("outcome", labels.outcome)
113            .with_new_label("cleanup_disposition", labels.cleanup_disposition)
114            .with_new_label("cleanup_failure_count_complete", labels.cleanup_failure_count_complete)
115            .with_new_label("capacity_disposition", labels.capacity_disposition)
116            .with_new_label("health_transition", labels.health_transition)
117            .with_new_label("retry_recovery", labels.retry_recovery)
118            .with_new_label("retry_prerequisite", labels.retry_prerequisite)
119            .with_new_label("observation_truncated", labels.observation_truncated)
120            .with_new_label("observation_degraded", labels.observation_degraded)
121            .with_new_label("native_fault_evidence", labels.native_fault_evidence)
122            .with_new_label("resource_evidence", labels.resource_evidence)
123            .global_counter(metric_name::attempt_report_rows())
124            .add(1);
125        (graph_claim_calibration, execution)
126    }
127}
128
129fn sealed_execution_metrics(
130    sets: &[ExecutionPlanMetricsSet],
131    report: &QueryAttemptReport,
132) -> NativeDataFusionExecutionMetrics {
133    let mut seen = HashSet::new();
134    let mut execution: Option<NativeDataFusionExecutionMetrics> = None;
135    for set in sets {
136        // Execution clones and delegating wrappers expose the same metric
137        // handles; each handle counts once.
138        let mut wrapper = MetricsSet::new();
139        for metric in set.clone_inner().iter() {
140            if seen.insert(Arc::as_ptr(metric)) {
141                wrapper.push(Arc::clone(metric));
142            }
143        }
144        if wrapper.iter().next().is_none() {
145            continue;
146        }
147        let projected = NativeDataFusionExecutionMetrics::from_metrics(&wrapper);
148        match execution.as_mut() {
149            Some(execution) => execution.merge_from(projected),
150            None => execution = Some(projected),
151        }
152    }
153    let mut execution = execution.unwrap_or_default();
154    let admission = report.admission.as_ref();
155    execution.admission_sequence = admission.map(|admission| count(admission.admission_sequence));
156    execution.admission_wait_time_ns = admission.map(|admission| count(admission.admission_wait_ns));
157    execution.device_ordinal = report.device_ordinal;
158    execution.query_grant_bytes = admission.map(|admission| count(admission.grant_bytes));
159    execution
160}
161
162/// Maximum graph calibration rows retained by one public sealed-attempt
163/// snapshot. The engine can retain richer diagnostic metrics internally, but
164/// the adapter projection stays bounded for runner handoff.
165pub const SEALED_ATTEMPT_GRAPH_CLAIM_CALIBRATION_MAX_ROWS: usize = 64;
166
167/// Immutable, typed graph claim-versus-peak rows projected by one sealed
168/// attempt.
169///
170/// Rows originate directly from engine [`QueryMetrics`]. The adapter neither
171/// parses DataFusion metric labels nor uses this snapshot for admission,
172/// retries, or device accounting.
173#[derive(Debug, Clone, PartialEq, Eq)]
174pub struct SealedAttemptGraphClaimCalibrationSnapshot {
175    attempt_id: AttemptId,
176    rows: Box<[GraphMeasuredPeakRow]>,
177    truncated: bool,
178    execution: Option<GraphClaimCalibrationExecution>,
179    execution_complete: bool,
180}
181
182/// Typed graph-execution classification sealed with an attempt's calibration
183/// rows. A physical classification requires the engine to record that the
184/// graph executor started; peak qualification remains fail-closed downstream.
185#[derive(Debug, Clone, PartialEq, Eq)]
186pub enum GraphClaimCalibrationExecution {
187    /// The graph algorithm executed locally and can supply physical allocation-peak evidence.
188    Physical,
189    /// The logical request completed without a local graph traversal, for the attached reason.
190    NonPhysical(GraphClaimCalibrationNonPhysicalReason),
191}
192
193/// Closed vocabulary for graph attempts that intentionally did not execute a
194/// graph algorithm locally.
195#[derive(Debug, Clone, PartialEq, Eq)]
196pub enum GraphClaimCalibrationNonPhysicalReason {
197    /// This exact BFS request reused a completed leader result. The leader's
198    /// engine-minted attempt id is retained when the request was attempt-bound
199    /// so qualification can require that leader's physical peak evidence.
200    BfsSingleflightFollower {
201        /// Leader attempt whose traversal supplied this result, or `None` without an attempt binding.
202        leader_attempt_id: Option<AttemptId>,
203    },
204}
205
206impl SealedAttemptGraphClaimCalibrationSnapshot {
207    fn from_captures(attempt_id: AttemptId, captures: impl IntoIterator<Item = GraphClaimCalibrationCapture>) -> Self {
208        let mut rows = Vec::new();
209        let mut truncated = false;
210        let mut execution = None;
211        let mut execution_complete = true;
212        for capture in captures {
213            let captured = capture.seal();
214            truncated |= captured.truncated;
215            if let Some(captured_execution) = captured.execution {
216                // A composed root contributes an otherwise empty capture while
217                // its graph producer owns the classification. An absent
218                // classification is neutral here; only a reported, conflicting
219                // classification can make the merged projection incomplete.
220                execution_complete &= captured.execution_complete;
221                match execution.as_ref() {
222                    // The sealed target may receive the same capture through a
223                    // composed root and its graph producer. Classification is a state.
224                    // Identical observations coalesce.
225                    Some(existing) if existing == &captured_execution => {},
226                    Some(_) => {
227                        execution = Some(captured_execution);
228                        execution_complete = false;
229                    },
230                    None => execution = Some(captured_execution),
231                }
232            }
233            let remaining = SEALED_ATTEMPT_GRAPH_CLAIM_CALIBRATION_MAX_ROWS.saturating_sub(rows.len());
234            if captured.rows.len() > remaining {
235                truncated = true;
236            }
237            rows.extend(captured.rows.into_iter().take(remaining));
238        }
239        Self { attempt_id, rows: rows.into_boxed_slice(), truncated, execution, execution_complete }
240    }
241
242    /// Attempt that owns this immutable diagnostic projection.
243    pub const fn attempt_id(&self) -> AttemptId {
244        self.attempt_id
245    }
246
247    /// Typed engine rows in execution observation order.
248    pub fn rows(&self) -> &[GraphMeasuredPeakRow] {
249        &self.rows
250    }
251
252    /// Whether the adapter omitted rows beyond its fixed public bound.
253    pub const fn truncated(&self) -> bool {
254        self.truncated
255    }
256
257    /// Explicit BFS follower classification or the engine-row-derived graph
258    /// executor state. `None` remains fail-closed when no exact boundary was
259    /// observed.
260    pub fn execution(&self) -> Option<&GraphClaimCalibrationExecution> {
261        self.execution.as_ref()
262    }
263
264    /// Whether the execution owner recorded exactly one non-conflicting
265    /// classification before the attempt sealed.
266    pub const fn execution_complete(&self) -> bool {
267        self.execution_complete && self.execution.is_some()
268    }
269}
270
271/// Per-wrapper typed calibration collector. It is shared by execution clones,
272/// then frozen exactly once by the elected sealed-attempt target.
273#[derive(Debug, Clone, Default)]
274pub(crate) struct GraphClaimCalibrationCapture {
275    state: Arc<Mutex<GraphClaimCalibrationCaptureState>>,
276}
277
278#[derive(Debug, Default)]
279struct GraphClaimCalibrationCaptureState {
280    rows: Vec<GraphMeasuredPeakRow>,
281    truncated: bool,
282    execution: Option<GraphClaimCalibrationExecution>,
283    execution_complete: bool,
284    sealed: bool,
285}
286
287struct CapturedGraphMeasuredPeakRows {
288    rows: Vec<GraphMeasuredPeakRow>,
289    truncated: bool,
290    execution: Option<GraphClaimCalibrationExecution>,
291    execution_complete: bool,
292}
293
294impl GraphClaimCalibrationCapture {
295    /// Records a completed exact-request BFS follower. The follower did not
296    /// execute a graph algorithm; it must point at its physical leader and
297    /// must not borrow or synthesize that leader's peak.
298    #[cfg(feature = "cugraph")]
299    pub(crate) fn record_bfs_singleflight_follower(&self, leader_attempt_id: Option<AttemptId>) {
300        self.record_execution(GraphClaimCalibrationExecution::NonPhysical(
301            GraphClaimCalibrationNonPhysicalReason::BfsSingleflightFollower { leader_attempt_id },
302        ));
303    }
304
305    #[cfg(feature = "cugraph")]
306    fn record_execution(&self, execution: GraphClaimCalibrationExecution) {
307        let mut state = self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
308        if state.sealed {
309            return;
310        }
311        match state.execution.as_ref() {
312            // A wrapper can reach the same physical boundary through nested
313            // execution helpers. Repeating the identical fact does not make
314            // the sealed ownership classification ambiguous.
315            Some(existing) if existing == &execution => {},
316            Some(_) => {
317                state.execution = Some(execution);
318                state.execution_complete = false;
319            },
320            None => {
321                state.execution = Some(execution);
322                state.execution_complete = true;
323            },
324        }
325    }
326
327    pub(crate) fn record(&self, rows: &[GraphMeasuredPeakRow]) {
328        if rows.is_empty() {
329            return;
330        }
331        let mut state = self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
332        if state.sealed {
333            return;
334        }
335        let remaining = SEALED_ATTEMPT_GRAPH_CLAIM_CALIBRATION_MAX_ROWS.saturating_sub(state.rows.len());
336        if rows.len() > remaining {
337            state.truncated = true;
338        }
339        state.rows.extend(rows.iter().take(remaining).cloned());
340    }
341
342    fn seal(&self) -> CapturedGraphMeasuredPeakRows {
343        let mut state = self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
344        state.sealed = true;
345        let (row_execution, row_execution_complete) = graph_claim_execution_from_rows(&state.rows);
346        let (execution, execution_complete) = match state.execution.clone() {
347            // BFS followers have an engine-minted request identity and no local
348            // graph row. Keep that explicit route distinct from the
349            // row-derived graph-executor lifecycle.
350            Some(execution) => (Some(execution), state.execution_complete),
351            None => (row_execution, row_execution_complete),
352        };
353        CapturedGraphMeasuredPeakRows {
354            rows: state.rows.clone(),
355            truncated: state.truncated,
356            execution,
357            execution_complete,
358        }
359    }
360}
361
362/// Derives an attempt classification only from terminal states authored by the
363/// engine's graph-claim row. Pending or inconsistent rows never become a
364/// physical graph execution at the adapter boundary.
365fn graph_claim_execution_from_rows(rows: &[GraphMeasuredPeakRow]) -> (Option<GraphClaimCalibrationExecution>, bool) {
366    let mut execution = None;
367    let mut complete = !rows.is_empty();
368    for row in rows {
369        let row_execution = match row.execution_state {
370            GraphClaimExecutionState::Pending => {
371                complete = false;
372                continue;
373            },
374            GraphClaimExecutionState::AlgorithmStarted => GraphClaimCalibrationExecution::Physical,
375        };
376        match execution.as_ref() {
377            Some(existing) if existing == &row_execution => {},
378            Some(_) => {
379                execution = Some(row_execution);
380                complete = false;
381            },
382            None => execution = Some(row_execution),
383        }
384    }
385    let complete = complete && execution.is_some();
386    (execution, complete)
387}
388
389/// Private translation layer from native QueryMetrics into DataFusion metrics.
390#[derive(Debug, Clone)]
391pub(crate) struct AlgeonNativeMetrics {
392    metrics_set: ExecutionPlanMetricsSet,
393    plan_task_mapping_datafusion_shape: Option<String>,
394    graph_claim_calibration_capture: Option<GraphClaimCalibrationCapture>,
395    pub(crate) baseline: BaselineMetrics,
396    registry: NativeMetricRegistry,
397    row_capture_budget: Arc<Mutex<MetricRowCaptureBudget>>,
398    row_capture_loss: Count,
399    row_capture_loss_reported: Arc<AtomicBool>,
400}
401
402#[derive(Debug)]
403struct MetricRowCaptureBudget {
404    remaining_records: u64,
405    remaining_bytes: u64,
406    truncated: bool,
407    counter_overflow: bool,
408}
409
410impl MetricRowCaptureBudget {
411    fn new(capture: Option<&NativeExecutionMetricsCapture>) -> Self {
412        let limits = capture.map_or_else(
413            || RuntimeObservationConfig::native_execution().limits(),
414            NativeExecutionMetricsCapture::metric_capture_limits,
415        );
416        Self {
417            remaining_records: limits.max_retained_records,
418            remaining_bytes: limits.max_retained_bytes,
419            truncated: false,
420            counter_overflow: false,
421        }
422    }
423
424    fn admit(&mut self, row: &impl fmt::Debug) -> bool {
425        if self.remaining_records == 0 {
426            self.truncated = true;
427            return false;
428        }
429        let mut probe = MetricRowSizeProbe { remaining: self.remaining_bytes, exceeded: false };
430        let _ = write!(probe, "{row:?}");
431        if probe.exceeded {
432            self.truncated = true;
433            return false;
434        }
435        let used = self.remaining_bytes - probe.remaining;
436        self.remaining_records = self.remaining_records.checked_sub(1).unwrap_or_else(|| {
437            self.counter_overflow = true;
438            0
439        });
440        self.remaining_bytes = self.remaining_bytes.checked_sub(used).unwrap_or_else(|| {
441            self.counter_overflow = true;
442            0
443        });
444        true
445    }
446}
447
448struct MetricRowSizeProbe {
449    remaining: u64,
450    exceeded: bool,
451}
452
453impl fmt::Write for MetricRowSizeProbe {
454    fn write_str(&mut self, value: &str) -> fmt::Result {
455        let bytes = u64::try_from(value.len()).unwrap_or(u64::MAX);
456        if bytes > self.remaining {
457            self.exceeded = true;
458            self.remaining = 0;
459            return Err(fmt::Error);
460        }
461        self.remaining -= bytes;
462        Ok(())
463    }
464}
465
466// Recorder methods use descriptor-generated field identifiers directly while the
467// registry retains ownership of all registered DataFusion metric handles.
468impl Deref for AlgeonNativeMetrics {
469    type Target = NativeMetricRegistry;
470
471    fn deref(&self) -> &Self::Target {
472        &self.registry
473    }
474}
475
476impl AlgeonNativeMetrics {
477    #[cfg(test)]
478    pub(crate) fn new(metrics: &ExecutionPlanMetricsSet, partition: usize) -> Self {
479        Self::new_with_capture_and_shape(metrics, partition, None, None)
480    }
481
482    pub(crate) fn new_with_capture(
483        metrics: &ExecutionPlanMetricsSet,
484        partition: usize,
485        capture: Option<NativeExecutionMetricsCapture>,
486    ) -> Self {
487        Self::new_with_capture_and_shape(metrics, partition, capture, None)
488    }
489
490    pub(crate) fn new_with_capture_and_shape(
491        metrics: &ExecutionPlanMetricsSet,
492        partition: usize,
493        capture: Option<NativeExecutionMetricsCapture>,
494        plan_task_mapping_datafusion_shape: Option<&str>,
495    ) -> Self {
496        let row_capture_budget = Arc::new(Mutex::new(MetricRowCaptureBudget::new(capture.as_ref())));
497        let row_capture_loss = MetricBuilder::new(metrics)
498            .with_type(MetricType::Summary)
499            .counter(SERVER_CAPTURE_LOSS_METRIC_NAME, partition);
500        Self {
501            metrics_set: metrics.clone(),
502            plan_task_mapping_datafusion_shape: plan_task_mapping_datafusion_shape.map(ToString::to_string),
503            graph_claim_calibration_capture: None,
504            baseline: BaselineMetrics::new(metrics, partition),
505            registry: NativeMetricRegistry::register(metrics, partition),
506            row_capture_budget,
507            row_capture_loss,
508            row_capture_loss_reported: Arc::new(AtomicBool::new(false)),
509        }
510    }
511
512    /// Attach the elected wrapper's typed graph calibration collector.
513    ///
514    /// This is intentionally separate from DataFusion row metrics: it copies
515    /// engine rows before label rendering and has no admission authority.
516    pub(crate) fn with_graph_claim_calibration_capture(mut self, capture: GraphClaimCalibrationCapture) -> Self {
517        self.graph_claim_calibration_capture = Some(capture);
518        self
519    }
520
521    fn admit_metric_row(&self, row: &impl fmt::Debug) -> bool {
522        let admitted = self.row_capture_budget.lock().unwrap_or_else(std::sync::PoisonError::into_inner).admit(row);
523        if !admitted {
524            self.report_row_capture_loss();
525        }
526        admitted
527    }
528
529    /// Admits a detail row the metrics owner already selected under its
530    /// retention limits. A projected value is recorded as-is; selecting its
531    /// rows a second time would lose rows the owner kept. Only a value built
532    /// outside an owner still goes through this recorder's own budget.
533    fn admit_owner_selected_row(&self, metrics: &QueryMetrics, row: &impl fmt::Debug) -> bool {
534        metrics.projection.is_some() || self.admit_metric_row(row)
535    }
536
537    fn report_row_capture_loss(&self) {
538        if !self.row_capture_loss_reported.swap(true, Ordering::Relaxed) {
539            self.row_capture_loss.add(1);
540        }
541    }
542
543    #[cfg(test)]
544    pub(super) fn row_capture_state(&self) -> (u64, u64, bool, bool) {
545        let budget = self.row_capture_budget.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
546        (budget.remaining_records, budget.remaining_bytes, budget.truncated, budget.counter_overflow)
547    }
548
549    pub(crate) fn record_admission_facts(&self, facts: QueryAdmissionFacts) {
550        self.record_descriptor_admission_facts(facts);
551    }
552
553    pub(crate) fn record_query_metrics(&self, metrics: &QueryMetrics) {
554        if metrics.projection.as_ref().is_some_and(|projection| projection.completeness.truncated) {
555            self.report_row_capture_loss();
556        }
557        if let Some(capture) = &self.graph_claim_calibration_capture {
558            capture.record(metrics.graph_claim_calibration_rows());
559        }
560        self.record_descriptor_query_metrics(metrics);
561    }
562
563    /// Records every evidence surface before returning the native result to the
564    /// adapter's `QueryError -> DataFusionError` conversion boundary.
565    pub(crate) fn finalize_query_outcome(
566        &self,
567        outcome: QueryExecutionOutcome,
568    ) -> std::result::Result<QueryOutput, QueryError> {
569        let (result, plan, _observation) = outcome.into_parts();
570        self.record_query_metrics(&plan.freeze());
571        result
572    }
573
574    /// Finalizes a native outcome whose logical host sink the adapter's
575    /// bounded GPU-to-host cursor runs later. A failed plan records its
576    /// metrics now; a successful one returns them, with `context_lease`, for the
577    /// cursor to hold. Either way the metrics freeze before the lease is released.
578    pub(crate) fn finalize_query_outcome_for_deferred_host(
579        &self,
580        outcome: QueryExecutionOutcome,
581        context_lease: AlgeonNativeExecutionContext,
582    ) -> std::result::Result<(QueryOutput, DeferredHostMetrics), QueryError> {
583        let (result, plan, _observation) = outcome.into_parts();
584        match result {
585            Ok(output) => Ok((output, DeferredHostMetrics::new(self.clone(), plan, context_lease))),
586            Err(error) => {
587                self.record_query_metrics(&plan.freeze());
588                Err(error)
589            },
590        }
591    }
592
593    fn record_lakehouse_cache_fill_event_rows(&self, events: &[LakehouseCacheFillEventSnapshot]) {
594        for event in events {
595            if !self.admit_metric_row(event) {
596                continue;
597            }
598            MetricBuilder::new(&self.metrics_set)
599                .with_type(MetricType::Summary)
600                .with_new_label("outcome", event.outcome.to_string())
601                .with_new_label("bytes", event.bytes.map_or_else(|| "none".to_string(), |value| value.to_string()))
602                .with_new_label("elapsed_ns", event.elapsed.as_nanos().min(u128::from(u64::MAX)).to_string())
603                .with_new_label("cache_path_hash", event.cache_path_hash.clone())
604                .with_new_label("stable_id_hash", event.stable_id_hash.clone())
605                .with_new_label(
606                    "file_index",
607                    event.file_index.map_or_else(|| "none".to_string(), |value| value.to_string()),
608                )
609                .with_new_label(
610                    "chunk_index",
611                    event.chunk_index.map_or_else(|| "none".to_string(), |value| value.to_string()),
612                )
613                .with_new_label("error_code", event.error_code.clone().unwrap_or_else(|| "none".to_string()))
614                .with_new_label("error_kind", event.error_kind.clone().unwrap_or_else(|| "none".to_string()))
615                .with_new_label("error_status", event.error_status.clone().unwrap_or_else(|| "none".to_string()))
616                .with_new_label("error_operation", event.error_operation.clone().unwrap_or_else(|| "none".to_string()))
617                .global_counter(metric_name::native_lakehouse_cache_fill_event_snapshot_rows())
618                .add(1);
619        }
620    }
621
622    fn record_graph_runtime_rows(&self, metrics: &QueryMetrics) {
623        for row in &metrics.graph_algorithm_runtime_rows {
624            if !self.admit_metric_row(row) {
625                continue;
626            }
627            MetricBuilder::new(&self.metrics_set)
628                .with_type(MetricType::Summary)
629                .with_new_label("node_id", row.node_id.clone())
630                .with_new_label("algorithm", row.algorithm.clone())
631                .with_new_label("runtime_outcome", row.runtime_outcome.clone())
632                .with_new_label("input_boundary", row.input_boundary.clone())
633                .with_new_label("output_boundary", row.output_boundary.clone())
634                .with_new_label("construction_policy", row.construction_policy.clone())
635                .with_new_label("source_column", row.source_column.clone())
636                .with_new_label("destination_column", row.destination_column.clone())
637                .with_new_label(
638                    "weight_column",
639                    row.weight_column.as_deref().map_or_else(|| "none".to_string(), ToOwned::to_owned),
640                )
641                .with_new_label("directed", row.directed.to_string())
642                .with_new_label("renumber", row.renumber.to_string())
643                .with_new_label("algorithm_parameters", row.algorithm_parameters.clone())
644                .with_new_label("algorithm_metadata", row.algorithm_metadata.clone())
645                .with_new_label("input_rows", row.input_rows.to_string())
646                .with_new_label("input_columns", row.input_columns.to_string())
647                .with_new_label("output_rows", row.output_rows.to_string())
648                .with_new_label("output_columns", row.output_columns.to_string())
649                .with_new_label(
650                    "output_device_bytes",
651                    row.output_device_bytes.map_or_else(|| "none".to_string(), |value| value.to_string()),
652                )
653                .with_new_label(
654                    "graph_peak_bytes",
655                    row.graph_peak_bytes.map_or_else(|| "none".to_string(), |value| value.to_string()),
656                )
657                .with_new_label("elapsed_ns", row.elapsed_ns.to_string())
658                .with_new_label("gpu_resident_input_handoffs", row.gpu_resident_input_handoffs.to_string())
659                .with_new_label("host_to_device_uploads", row.host_to_device_uploads.to_string())
660                .with_new_label("device_to_host_materializations", row.device_to_host_materializations.to_string())
661                .global_counter(metric_name::native_graph_runtime_rows())
662                .add(1);
663        }
664    }
665
666    fn record_task_operator_rows(&self, metrics: &QueryMetrics) {
667        for row in metrics.task_operator_metrics.values() {
668            if !self.admit_metric_row(row) {
669                continue;
670            }
671            MetricBuilder::new(&self.metrics_set)
672                .with_type(MetricType::Summary)
673                .with_new_label("operator_kind", row.operator_kind.clone())
674                .with_new_label("task_count", row.task_count.to_string())
675                .with_new_label("elapsed_ns", row.elapsed_time.as_nanos().min(u128::from(u64::MAX)).to_string())
676                .with_new_label(
677                    "max_observed_peak_bytes",
678                    row.max_observed_peak_bytes.map_or_else(|| "none".to_string(), |value| value.to_string()),
679                )
680                .with_new_label(
681                    "max_retained_bytes",
682                    row.max_retained_bytes.map_or_else(|| "none".to_string(), |value| value.to_string()),
683                )
684                .global_counter(metric_name::native_task_operator_rows())
685                .add(1);
686        }
687    }
688
689    fn record_task_runtime_rows(&self, metrics: &QueryMetrics) {
690        for row in metrics.task_runtime_metrics.values() {
691            if !self.admit_metric_row(row) {
692                continue;
693            }
694            MetricBuilder::new(&self.metrics_set)
695                .with_type(MetricType::Summary)
696                .with_new_label("compiled_task_id", row.compiled_task_id.clone())
697                .with_new_label("operator_kind", row.operator_kind.clone())
698                .with_new_label("elapsed_ns", row.elapsed_time.as_nanos().min(u128::from(u64::MAX)).to_string())
699                .with_new_label(
700                    "max_observed_peak_bytes",
701                    row.max_observed_peak_bytes.map_or_else(|| "none".to_string(), |value| value.to_string()),
702                )
703                .with_new_label(
704                    "max_retained_bytes",
705                    row.max_retained_bytes.map_or_else(|| "none".to_string(), |value| value.to_string()),
706                )
707                .global_counter(metric_name::native_task_runtime_rows())
708                .add(1);
709        }
710    }
711
712    fn record_plan_task_mapping_rows(&self, metrics: &QueryMetrics) {
713        for mut row in metrics.plan_task_mappings() {
714            if let Some(shape) = &self.plan_task_mapping_datafusion_shape {
715                row.datafusion_shape = Some(shape.clone());
716            }
717            if !self.admit_metric_row(&row) {
718                continue;
719            }
720            MetricBuilder::new(&self.metrics_set)
721                .with_type(MetricType::Summary)
722                .with_new_label("query_id", row.query_id.as_str().to_string())
723                .with_new_label(
724                    "plan_node_id",
725                    row.plan_node_id.as_ref().map_or_else(|| "none".to_string(), |value| value.as_str().to_string()),
726                )
727                .with_new_label(
728                    "compiled_task_id",
729                    row.compiled_task_id
730                        .as_ref()
731                        .map_or_else(|| "none".to_string(), |value| value.as_str().to_string()),
732                )
733                .with_new_label("operator_kind", row.operator_kind.unwrap_or_else(|| "none".to_string()))
734                .with_new_label("datafusion_shape", row.datafusion_shape.unwrap_or_else(|| "none".to_string()))
735                .global_counter(metric_name::native_plan_task_mapping_rows())
736                .add(1);
737        }
738    }
739
740    fn record_dynamic_filter_runtime_rows(&self, metrics: &QueryMetrics) {
741        for row in &metrics.dynamic_filter_runtime_rows {
742            if !self.admit_owner_selected_row(metrics, row) {
743                continue;
744            }
745            MetricBuilder::new(&self.metrics_set)
746                .with_type(MetricType::Summary)
747                .with_new_label("dynamic_filter_id", row.dynamic_filter_id.clone())
748                .with_new_label("event", row.event.clone())
749                .with_new_label("producer_kind", row.producer_kind.clone())
750                .with_new_label("producer_node_id", row.producer_node_id.clone())
751                .with_new_label("consumer_kind", row.consumer_kind.clone())
752                .with_new_label("consumer_node_id", row.consumer_node_id.clone())
753                .with_new_label("consumer_lowering", row.consumer_lowering.clone())
754                .with_new_label("consumer_reduction_stage", row.consumer_reduction_stage.clone())
755                .with_new_label("filter_kind", row.filter_kind.clone())
756                .with_new_label(
757                    "payload_value_count",
758                    row.payload_value_count.map_or_else(|| "none".to_string(), |value| value.to_string()),
759                )
760                .with_new_label(
761                    "payload_min_i64",
762                    row.payload_min_i64.map_or_else(|| "none".to_string(), |value| value.to_string()),
763                )
764                .with_new_label(
765                    "payload_max_i64",
766                    row.payload_max_i64.map_or_else(|| "none".to_string(), |value| value.to_string()),
767                )
768                .with_new_label(
769                    "producer_rows",
770                    row.producer_rows.map_or_else(|| "none".to_string(), |value| value.to_string()),
771                )
772                .with_new_label(
773                    "skipped_reason",
774                    row.skipped_reason.as_deref().map_or_else(|| "none".to_string(), ToOwned::to_owned),
775                )
776                .with_new_label(
777                    "selected_row_groups_before",
778                    row.selected_row_groups_before.map_or_else(|| "none".to_string(), |value| value.to_string()),
779                )
780                .with_new_label(
781                    "selected_row_groups_after",
782                    row.selected_row_groups_after.map_or_else(|| "none".to_string(), |value| value.to_string()),
783                )
784                .with_new_label(
785                    "selected_source_bytes_before",
786                    row.selected_source_bytes_before.map_or_else(|| "none".to_string(), |value| value.to_string()),
787                )
788                .with_new_label(
789                    "selected_source_bytes_after",
790                    row.selected_source_bytes_after.map_or_else(|| "none".to_string(), |value| value.to_string()),
791                )
792                .with_new_label(
793                    "source_file_index",
794                    row.source_file_index.map_or_else(|| "none".to_string(), |value| value.to_string()),
795                )
796                .with_new_label(
797                    "source_chunk_index",
798                    row.source_chunk_index.map_or_else(|| "none".to_string(), |value| value.to_string()),
799                )
800                .with_new_label(
801                    "reader_predicate_rows_before",
802                    row.reader_predicate_rows_before.map_or_else(|| "none".to_string(), |value| value.to_string()),
803                )
804                .with_new_label(
805                    "reader_predicate_rows_after",
806                    row.reader_predicate_rows_after.map_or_else(|| "none".to_string(), |value| value.to_string()),
807                )
808                .with_new_label("reader_predicate_rows_filtered", row.reader_predicate_rows_filtered.to_string())
809                .with_new_label(
810                    "post_decode_rows_before",
811                    row.post_decode_rows_before.map_or_else(|| "none".to_string(), |value| value.to_string()),
812                )
813                .with_new_label(
814                    "post_decode_rows_after",
815                    row.post_decode_rows_after.map_or_else(|| "none".to_string(), |value| value.to_string()),
816                )
817                .with_new_label("post_decode_rows_filtered", row.post_decode_rows_filtered.to_string())
818                .with_new_label("row_groups_pruned", row.row_groups_pruned.to_string())
819                .with_new_label("source_bytes_pruned", row.source_bytes_pruned.to_string())
820                .global_counter(metric_name::native_dynamic_filter_runtime_rows())
821                .add(1);
822        }
823    }
824
825    fn record_source_read_runtime_rows(&self, metrics: &QueryMetrics) {
826        for row in &metrics.source_read_runtime_rows {
827            if !self.admit_owner_selected_row(metrics, row) {
828                continue;
829            }
830            MetricBuilder::new(&self.metrics_set)
831                .with_type(MetricType::Summary)
832                .with_new_label("source_node_id", row.source_node_id.clone())
833                .with_new_label("read_kind", row.read_kind.clone())
834                .with_new_label("read_strategy", row.read_strategy.clone())
835                .with_new_label("parquet_read_input_kind", row.parquet_read_evidence.input_kind.clone())
836                .with_new_label("remote_read_mode", row.parquet_read_evidence.remote_read_mode.clone())
837                .with_new_label("parquet_read_backend", row.parquet_read_evidence.backend.clone())
838                .with_new_label("remote_backend_build_status", row.parquet_read_evidence.backend_build_status.clone())
839                .with_new_label(
840                    "remote_read_bytes",
841                    row.parquet_read_evidence
842                        .remote_bytes
843                        .map_or_else(|| "none".to_string(), |value| value.to_string()),
844                )
845                .with_new_label(
846                    "remote_read_range_count",
847                    row.parquet_read_evidence
848                        .remote_range_count
849                        .map_or_else(|| "none".to_string(), |value| value.to_string()),
850                )
851                .with_new_label("device_read_support", row.parquet_read_evidence.device_read_support.clone())
852                .with_new_label(
853                    "parquet_pruning_input_row_groups",
854                    row.parquet_read_evidence
855                        .pruning_input_row_groups
856                        .map_or_else(|| "none".to_string(), |value| value.to_string()),
857                )
858                .with_new_label(
859                    "parquet_row_groups_after_stats_filter",
860                    row.parquet_read_evidence
861                        .row_groups_after_stats_filter
862                        .map_or_else(|| "none".to_string(), |value| value.to_string()),
863                )
864                .with_new_label(
865                    "parquet_row_groups_after_bloom_filter",
866                    row.parquet_read_evidence
867                        .row_groups_after_bloom_filter
868                        .map_or_else(|| "none".to_string(), |value| value.to_string()),
869                )
870                .with_new_label("parquet_bloom_pruning_status", row.parquet_read_evidence.bloom_pruning_status.clone())
871                .with_new_label("batch_index", row.batch_index.to_string())
872                .with_new_label("chunk_count", row.chunk_count.to_string())
873                .with_new_label("first_chunk_index", row.first_chunk_index.to_string())
874                .with_new_label("last_chunk_index", row.last_chunk_index.to_string())
875                .with_new_label("first_file_index", row.first_file_index.to_string())
876                .with_new_label("last_file_index", row.last_file_index.to_string())
877                .with_new_label("file_count", row.file_count.to_string())
878                .with_new_label("row_group_count", row.row_group_count.to_string())
879                .with_new_label("read_columns_per_file", row.read_columns_per_file.to_string())
880                .with_new_label("read_columns_total", row.read_columns_total.to_string())
881                .with_new_label("known_file_bytes", row.known_file_bytes.to_string())
882                .with_new_label("estimated_source_read_bytes", row.estimated_source_read_bytes.to_string())
883                .with_new_label("known_row_group_bytes", row.known_row_group_bytes.to_string())
884                .with_new_label("known_projected_compressed_bytes", row.known_projected_compressed_bytes.to_string())
885                .with_new_label(
886                    "known_projected_uncompressed_bytes",
887                    row.known_projected_uncompressed_bytes.to_string(),
888                )
889                .with_new_label(
890                    "known_pure_filter_compressed_bytes",
891                    row.known_pure_filter_compressed_bytes.to_string(),
892                )
893                .with_new_label(
894                    "known_pure_filter_uncompressed_bytes",
895                    row.known_pure_filter_uncompressed_bytes.to_string(),
896                )
897                .with_new_label("output_rows", row.output_rows.to_string())
898                .with_new_label("reader_predicate", row.reader_predicate.to_string())
899                .with_new_label("reader_dynamic_filter_count", row.reader_dynamic_filter_count.to_string())
900                .with_new_label("source_batching_reason", row.source_batching_reason.clone())
901                .with_new_label("source_chunk_planning_mode", row.source_chunk_planning_mode.clone())
902                .with_new_label("source_chunk_planning_reason", row.source_chunk_planning_reason.clone())
903                .with_new_label("elapsed_ns", row.elapsed.as_nanos().min(u128::from(u64::MAX)).to_string())
904                .global_counter(metric_name::native_source_read_runtime_rows())
905                .add(1);
906        }
907    }
908
909    fn record_join_strategy_rows(&self, metrics: &QueryMetrics) {
910        let mut decision_facts_by_node: BTreeMap<&str, BTreeMap<String, String>> = BTreeMap::new();
911        for decision in &metrics.join_strategy_decisions {
912            if !self.admit_owner_selected_row(metrics, decision) {
913                continue;
914            }
915            decision_facts_by_node
916                .entry(decision.node_id.as_str())
917                .or_default()
918                .extend(decision.facts.iter().map(|(key, value)| (key.clone(), value.clone())));
919        }
920        for operator in metrics.operators.iter().filter(|operator| operator.operator_kind == "join") {
921            if !self.admit_owner_selected_row(metrics, operator) {
922                continue;
923            }
924            let Some(join) = operator.join.as_ref() else {
925                continue;
926            };
927            let facts = decision_facts_by_node.get(operator.node_id.as_str());
928            let join_strategy = facts
929                .and_then(|facts| facts.get("join_strategy").map(String::as_str))
930                .unwrap_or_else(|| join_strategy_label(join));
931            let runtime = join_runtime_counters(join);
932            let builder = MetricBuilder::new(&self.metrics_set)
933                .with_type(MetricType::Summary)
934                .with_new_label("node_id", operator.node_id.clone())
935                .with_new_label("join_strategy", join_strategy.to_string())
936                .with_new_label("execution_strategy", join.execution_strategy.clone())
937                .with_new_label("join_type", join.join_type.clone())
938                .with_new_label("build_side", join.build_side.clone())
939                .with_new_label("key_count", join.key_count.to_string())
940                .with_new_label("estimated_build_rows", optional_fact(facts, "estimated_build_rows"))
941                .with_new_label(
942                    "observed_build_rows",
943                    facts
944                        .and_then(|facts| facts.get("observed_build_rows").cloned())
945                        .unwrap_or_else(|| join.build_rows.to_string()),
946                )
947                .with_new_label(
948                    "build_side_bytes",
949                    facts
950                        .and_then(|facts| facts.get("build_side_bytes").cloned())
951                        .unwrap_or_else(|| join.build_bytes.to_string()),
952                )
953                .with_new_label("probe_reuse_count", optional_fact(facts, "probe_reuse_count"))
954                .with_new_label("probe_side_bytes", optional_fact(facts, "probe_side_bytes"))
955                .with_new_label("estimated_output_rows", optional_fact(facts, "estimated_output_rows"))
956                .with_new_label("estimated_output_bytes", optional_fact(facts, "estimated_output_bytes"))
957                .with_new_label("build_key_unique", optional_fact(facts, "build_key_unique"))
958                .with_new_label(
959                    "build_key_uniqueness_probe_status",
960                    optional_fact(facts, "build_key_uniqueness_probe_status"),
961                )
962                .with_new_label(
963                    "build_key_uniqueness_probe_error_code",
964                    optional_fact(facts, "build_key_uniqueness_probe_error_code"),
965                )
966                .with_new_label("partition_count", optional_fact(facts, "partition_count"))
967                .with_new_label("partition_target_bytes", optional_fact(facts, "partition_target_bytes"))
968                .with_new_label("build_partition_batch_count", optional_fact(facts, "build_partition_batch_count"))
969                .with_new_label("probe_partition_batch_count", optional_fact(facts, "probe_partition_batch_count"))
970                .with_new_label("build_partitioned_bytes", optional_fact(facts, "build_partitioned_bytes"))
971                .with_new_label("probe_partitioned_bytes", optional_fact(facts, "probe_partitioned_bytes"))
972                .with_new_label(
973                    "reusable_hash_join_rejected_reason",
974                    optional_fact(facts, "reusable_hash_join_rejected_reason"),
975                )
976                .with_new_label(
977                    "partitioned_hash_join_rejected_reason",
978                    optional_fact(facts, "partitioned_hash_join_rejected_reason"),
979                )
980                .with_new_label("one_shot_equi_rejected_reason", optional_fact(facts, "one_shot_equi_rejected_reason"))
981                .with_new_label("partitioned_hash_join_count", runtime.partitioned_hash_join_count.to_string())
982                .with_new_label("hash_join_build_count", runtime.hash_join_build_count.to_string())
983                .with_new_label("hash_join_hash_table_bytes", runtime.hash_join_hash_table_bytes.to_string())
984                .with_new_label("filtered_join_build_count", runtime.filtered_join_build_count.to_string())
985                .with_new_label("filtered_join_hash_table_bytes", runtime.filtered_join_hash_table_bytes.to_string())
986                .with_new_label("filtered_join_probe_state_bytes", runtime.filtered_join_probe_state_bytes.to_string())
987                .with_new_label("mixed_join_size_data_bytes", runtime.mixed_join_size_data_bytes.to_string())
988                .with_new_label("one_shot_join_count", runtime.one_shot_join_count.to_string())
989                .with_new_label("one_shot_join_index_bytes", runtime.one_shot_join_index_bytes.to_string())
990                .with_new_label(
991                    "one_shot_join_materialization_bytes",
992                    runtime.one_shot_join_materialization_bytes.to_string(),
993                )
994                .with_new_label("one_shot_join_output_bytes", runtime.one_shot_join_output_bytes.to_string())
995                .with_new_label("join_gather_map_bytes", join.gather_map_bytes.to_string())
996                .with_new_label(
997                    "join_output_reservation_bytes",
998                    join.output_reservation_bytes.map_or_else(|| "none".to_string(), |bytes| bytes.to_string()),
999                )
1000                .with_new_label("join_output_rows", join.output_rows.to_string())
1001                .with_new_label("join_output_bytes", join.output_bytes.to_string());
1002            builder.global_counter(metric_name::native_join_strategy_rows()).add(1);
1003        }
1004    }
1005
1006    fn record_aggregate_strategy_rows(&self, metrics: &QueryMetrics) {
1007        for decision in &metrics.aggregate_strategy_decisions {
1008            if !self.admit_owner_selected_row(metrics, decision) {
1009                continue;
1010            }
1011            let facts = &decision.facts;
1012            MetricBuilder::new(&self.metrics_set)
1013                .with_type(MetricType::Summary)
1014                .with_new_label("node_id", decision.node_id.clone())
1015                .with_new_label("selected_strategy", aggregate_fact(facts, "selected_strategy"))
1016                .with_new_label("execution_stage", aggregate_fact(facts, "execution_stage"))
1017                .with_new_label("group_key_count", aggregate_fact(facts, "group_key_count"))
1018                .with_new_label("group_key_exprs", aggregate_fact(facts, "group_key_exprs"))
1019                .with_new_label("group_key_dtypes", aggregate_fact(facts, "group_key_dtypes"))
1020                .with_new_label("aggregate_functions", aggregate_fact(facts, "aggregate_functions"))
1021                .with_new_label(
1022                    "floating_statistical_aggregate",
1023                    aggregate_fact(facts, "floating_statistical_aggregate"),
1024                )
1025                .with_new_label(
1026                    "statistical_aggregate_execution_mode",
1027                    aggregate_fact(facts, "statistical_aggregate_execution_mode"),
1028                )
1029                .with_new_label(
1030                    "statistical_aggregate_tolerance_contract",
1031                    aggregate_fact(facts, "statistical_aggregate_tolerance_contract"),
1032                )
1033                .with_new_label(
1034                    "statistical_aggregate_functions",
1035                    aggregate_fact(facts, "statistical_aggregate_functions"),
1036                )
1037                .with_new_label("statistical_aggregate_ddof", aggregate_fact(facts, "statistical_aggregate_ddof"))
1038                .with_new_label("aggregate_filter_count", aggregate_fact(facts, "aggregate_filter_count"))
1039                .with_new_label(
1040                    "filtered_input_materialization_strategy",
1041                    aggregate_fact(facts, "filtered_input_materialization_strategy"),
1042                )
1043                .with_new_label(
1044                    "filtered_input_materialization_estimated_bytes",
1045                    aggregate_fact(facts, "filtered_input_materialization_estimated_bytes"),
1046                )
1047                .with_new_label(
1048                    "filtered_input_materialization_budget_bytes",
1049                    aggregate_fact(facts, "filtered_input_materialization_budget_bytes"),
1050                )
1051                .with_new_label(
1052                    "filtered_input_materialization_threshold_bytes",
1053                    aggregate_fact(facts, "filtered_input_materialization_threshold_bytes"),
1054                )
1055                .with_new_label(
1056                    "filtered_global_rows_per_chunk",
1057                    aggregate_fact(facts, "filtered_global_rows_per_chunk"),
1058                )
1059                .with_new_label(
1060                    "filtered_global_inputs_per_segment",
1061                    aggregate_fact(facts, "filtered_global_inputs_per_segment"),
1062                )
1063                .with_new_label(
1064                    "filtered_global_segment_target_temporary_bytes",
1065                    aggregate_fact(facts, "filtered_global_segment_target_temporary_bytes"),
1066                )
1067                .with_new_label(
1068                    "filtered_global_actual_chunk_count",
1069                    aggregate_fact(facts, "filtered_global_actual_chunk_count"),
1070                )
1071                .with_new_label(
1072                    "filtered_global_actual_segment_count",
1073                    aggregate_fact(facts, "filtered_global_actual_segment_count"),
1074                )
1075                .with_new_label(
1076                    "filtered_global_max_segment_rows",
1077                    aggregate_fact(facts, "filtered_global_max_segment_rows"),
1078                )
1079                .with_new_label(
1080                    "filtered_global_observed_temporary_peak_bytes",
1081                    aggregate_fact(facts, "filtered_global_observed_temporary_peak_bytes"),
1082                )
1083                .with_new_label(
1084                    "filtered_global_observed_retained_peak_bytes",
1085                    aggregate_fact(facts, "filtered_global_observed_retained_peak_bytes"),
1086                )
1087                .with_new_label(
1088                    "filtered_global_state_rows_merged",
1089                    aggregate_fact(facts, "filtered_global_state_rows_merged"),
1090                )
1091                .with_new_label(
1092                    "filtered_grouped_rows_per_chunk",
1093                    aggregate_fact(facts, "filtered_grouped_rows_per_chunk"),
1094                )
1095                .with_new_label(
1096                    "filtered_grouped_inputs_per_segment",
1097                    aggregate_fact(facts, "filtered_grouped_inputs_per_segment"),
1098                )
1099                .with_new_label(
1100                    "filtered_grouped_segment_target_temporary_bytes",
1101                    aggregate_fact(facts, "filtered_grouped_segment_target_temporary_bytes"),
1102                )
1103                .with_new_label(
1104                    "filtered_grouped_actual_chunk_count",
1105                    aggregate_fact(facts, "filtered_grouped_actual_chunk_count"),
1106                )
1107                .with_new_label(
1108                    "filtered_grouped_actual_segment_count",
1109                    aggregate_fact(facts, "filtered_grouped_actual_segment_count"),
1110                )
1111                .with_new_label(
1112                    "filtered_grouped_max_segment_rows",
1113                    aggregate_fact(facts, "filtered_grouped_max_segment_rows"),
1114                )
1115                .with_new_label(
1116                    "filtered_grouped_observed_temporary_peak_bytes",
1117                    aggregate_fact(facts, "filtered_grouped_observed_temporary_peak_bytes"),
1118                )
1119                .with_new_label(
1120                    "filtered_grouped_observed_retained_peak_bytes",
1121                    aggregate_fact(facts, "filtered_grouped_observed_retained_peak_bytes"),
1122                )
1123                .with_new_label(
1124                    "filtered_grouped_state_rows_merged",
1125                    aggregate_fact(facts, "filtered_grouped_state_rows_merged"),
1126                )
1127                .with_new_label(
1128                    "filtered_grouped_high_cardinality_admission",
1129                    aggregate_fact(facts, "filtered_grouped_high_cardinality_admission"),
1130                )
1131                .with_new_label("distinct_aggregate_count", aggregate_fact(facts, "distinct_aggregate_count"))
1132                .with_new_label("input_rows", aggregate_fact(facts, "input_rows"))
1133                .with_new_label("output_group_count", aggregate_fact(facts, "output_group_count"))
1134                .with_new_label("reduction_ratio_ppm", aggregate_fact(facts, "reduction_ratio_ppm"))
1135                .with_new_label("chunk_row_count", aggregate_fact(facts, "chunk_row_count"))
1136                .with_new_label("chunk_count", aggregate_fact(facts, "chunk_count"))
1137                .with_new_label("partial_aggregate_count", aggregate_fact(facts, "partial_aggregate_count"))
1138                .with_new_label("source_adjacent_partial", aggregate_fact(facts, "source_adjacent_partial"))
1139                .with_new_label("source_chunk_row_group_count", aggregate_fact(facts, "source_chunk_row_group_count"))
1140                .with_new_label(
1141                    "source_chunk_reads_all_row_groups",
1142                    aggregate_fact(facts, "source_chunk_reads_all_row_groups"),
1143                )
1144                .with_new_label("source_chunk_known_row_count", aggregate_fact(facts, "source_chunk_known_row_count"))
1145                .with_new_label(
1146                    "source_chunk_estimated_source_read_bytes",
1147                    aggregate_fact(facts, "source_chunk_estimated_source_read_bytes"),
1148                )
1149                .with_new_label(
1150                    "source_chunk_size_estimate_based",
1151                    aggregate_fact(facts, "source_chunk_size_estimate_based"),
1152                )
1153                .with_new_label(
1154                    "source_adjacent_partition_strategy",
1155                    aggregate_fact(facts, "source_adjacent_partition_strategy"),
1156                )
1157                .with_new_label(
1158                    "source_adjacent_partition_count",
1159                    aggregate_fact(facts, "source_adjacent_partition_count"),
1160                )
1161                .with_new_label(
1162                    "source_adjacent_non_empty_partition_count",
1163                    aggregate_fact(facts, "source_adjacent_non_empty_partition_count"),
1164                )
1165                .with_new_label(
1166                    "source_adjacent_partition_key_count",
1167                    aggregate_fact(facts, "source_adjacent_partition_key_count"),
1168                )
1169                .with_new_label(
1170                    "source_adjacent_max_partition_fan_in",
1171                    aggregate_fact(facts, "source_adjacent_max_partition_fan_in"),
1172                )
1173                .with_new_label(
1174                    "partitioned_merge_partition_count",
1175                    aggregate_fact(facts, "partitioned_merge_partition_count"),
1176                )
1177                .with_new_label(
1178                    "partitioned_merge_non_empty_partition_count",
1179                    aggregate_fact(facts, "partitioned_merge_non_empty_partition_count"),
1180                )
1181                .with_new_label(
1182                    "partitioned_merge_max_partition_fan_in",
1183                    aggregate_fact(facts, "partitioned_merge_max_partition_fan_in"),
1184                )
1185                .with_new_label("selection_elapsed_ns", aggregate_fact(facts, "selection_elapsed_ns"))
1186                .with_new_label("encode_elapsed_ns", aggregate_fact(facts, "encode_elapsed_ns"))
1187                .with_new_label("groupby_elapsed_ns", aggregate_fact(facts, "groupby_elapsed_ns"))
1188                .with_new_label("decode_elapsed_ns", aggregate_fact(facts, "decode_elapsed_ns"))
1189                .with_new_label("strategy_operator_elapsed_ns", aggregate_fact(facts, "strategy_operator_elapsed_ns"))
1190                .with_new_label("aggregate_structural_key", aggregate_fact(facts, "aggregate_structural_key"))
1191                .with_new_label(
1192                    "prior_cardinality_evidence_scope",
1193                    aggregate_fact(facts, "prior_cardinality_evidence_scope"),
1194                )
1195                .with_new_label(
1196                    "aggregate_encoded_key_partial_low_cardinality_threshold_ppm",
1197                    aggregate_fact(facts, "aggregate_encoded_key_partial_low_cardinality_threshold_ppm"),
1198                )
1199                .with_new_label(
1200                    "aggregate_encoded_key_partial_min_input_rows",
1201                    aggregate_fact(facts, "aggregate_encoded_key_partial_min_input_rows"),
1202                )
1203                .with_new_label(
1204                    "aggregate_exact_grouped_key_cardinality_probe_max_key_columns",
1205                    aggregate_fact(facts, "aggregate_exact_grouped_key_cardinality_probe_max_key_columns"),
1206                )
1207                .with_new_label(
1208                    "aggregate_exact_single_string_avg_cardinality_probe_min_input_rows",
1209                    aggregate_fact(facts, "aggregate_exact_single_string_avg_cardinality_probe_min_input_rows"),
1210                )
1211                .with_new_label(
1212                    "aggregate_exact_single_key_cardinality_probe_min_input_rows",
1213                    aggregate_fact(facts, "aggregate_exact_single_key_cardinality_probe_min_input_rows"),
1214                )
1215                .with_new_label(
1216                    "aggregate_exact_grouped_key_cardinality_probe_min_input_rows",
1217                    aggregate_fact(facts, "aggregate_exact_grouped_key_cardinality_probe_min_input_rows"),
1218                )
1219                .with_new_label(
1220                    "aggregate_exact_single_string_key_prepared_max_groups",
1221                    aggregate_fact(facts, "aggregate_exact_single_string_key_prepared_max_groups"),
1222                )
1223                .with_new_label(
1224                    "aggregate_exact_three_key_cardinality_probe_max_input_rows",
1225                    aggregate_fact(facts, "aggregate_exact_three_key_cardinality_probe_max_input_rows"),
1226                )
1227                .with_new_label(
1228                    "aggregate_exact_five_key_cardinality_probe_max_input_rows",
1229                    aggregate_fact(facts, "aggregate_exact_five_key_cardinality_probe_max_input_rows"),
1230                )
1231                .with_new_label(
1232                    "aggregate_prepared_sort_groupby_low_cardinality_threshold_ppm",
1233                    aggregate_fact(facts, "aggregate_prepared_sort_groupby_low_cardinality_threshold_ppm"),
1234                )
1235                .with_new_label(
1236                    "aggregate_source_adjacent_prepared_sort_groupby_max_row_groups",
1237                    aggregate_fact(facts, "aggregate_source_adjacent_prepared_sort_groupby_max_row_groups"),
1238                )
1239                .with_new_label(
1240                    "aggregate_source_adjacent_prepared_sort_groupby_max_source_read_bytes",
1241                    aggregate_fact(facts, "aggregate_source_adjacent_prepared_sort_groupby_max_source_read_bytes"),
1242                )
1243                .with_new_label(
1244                    "aggregate_source_adjacent_prepared_sort_groupby_max_fan_in",
1245                    aggregate_fact(facts, "aggregate_source_adjacent_prepared_sort_groupby_max_fan_in"),
1246                )
1247                .with_new_label(
1248                    "encoded_key_partial_cost_model",
1249                    aggregate_fact(facts, "encoded_key_partial_cost_model"),
1250                )
1251                .with_new_label(
1252                    "encoded_key_partial_cost_model_status",
1253                    aggregate_fact(facts, "encoded_key_partial_cost_model_status"),
1254                )
1255                .with_new_label(
1256                    "encoded_key_partial_selection_reason",
1257                    aggregate_fact(facts, "encoded_key_partial_selection_reason"),
1258                )
1259                .with_new_label(
1260                    "encoded_key_partial_reduction_ratio_threshold_ppm",
1261                    aggregate_fact(facts, "encoded_key_partial_reduction_ratio_threshold_ppm"),
1262                )
1263                .with_new_label(
1264                    "encoded_key_partial_prior_reduction_ratio_ppm",
1265                    aggregate_fact(facts, "encoded_key_partial_prior_reduction_ratio_ppm"),
1266                )
1267                .with_new_label(
1268                    "encoded_key_partial_prior_strategy",
1269                    aggregate_fact(facts, "encoded_key_partial_prior_strategy"),
1270                )
1271                .with_new_label(
1272                    "exact_cardinality_probe_status",
1273                    aggregate_fact(facts, "exact_cardinality_probe_status"),
1274                )
1275                .with_new_label("exact_cardinality_probe_key", aggregate_fact(facts, "exact_cardinality_probe_key"))
1276                .with_new_label(
1277                    "exact_cardinality_probe_input_rows",
1278                    aggregate_fact(facts, "exact_cardinality_probe_input_rows"),
1279                )
1280                .with_new_label(
1281                    "exact_cardinality_probe_non_null_distinct_count",
1282                    aggregate_fact(facts, "exact_cardinality_probe_non_null_distinct_count"),
1283                )
1284                .with_new_label(
1285                    "exact_cardinality_probe_null_count",
1286                    aggregate_fact(facts, "exact_cardinality_probe_null_count"),
1287                )
1288                .with_new_label(
1289                    "exact_cardinality_probe_num_groups",
1290                    aggregate_fact(facts, "exact_cardinality_probe_num_groups"),
1291                )
1292                .with_new_label(
1293                    "exact_cardinality_probe_reduction_ratio_ppm",
1294                    aggregate_fact(facts, "exact_cardinality_probe_reduction_ratio_ppm"),
1295                )
1296                .with_new_label(
1297                    "exact_cardinality_probe_elapsed_ns",
1298                    aggregate_fact(facts, "exact_cardinality_probe_elapsed_ns"),
1299                )
1300                .with_new_label(
1301                    "prepared_sort_groupby_rejected_reason",
1302                    aggregate_fact(facts, "prepared_sort_groupby_rejected_reason"),
1303                )
1304                .with_new_label(
1305                    "prepared_sort_groupby_contract_status",
1306                    aggregate_fact(facts, "prepared_sort_groupby_contract_status"),
1307                )
1308                .with_new_label(
1309                    "prepared_sort_groupby_required_contract",
1310                    aggregate_fact(facts, "prepared_sort_groupby_required_contract"),
1311                )
1312                .with_new_label(
1313                    "prepared_sort_groupby_cost_model",
1314                    aggregate_fact(facts, "prepared_sort_groupby_cost_model"),
1315                )
1316                .with_new_label(
1317                    "prepared_sort_groupby_cost_model_status",
1318                    aggregate_fact(facts, "prepared_sort_groupby_cost_model_status"),
1319                )
1320                .with_new_label(
1321                    "prepared_sort_groupby_selection_reason",
1322                    aggregate_fact(facts, "prepared_sort_groupby_selection_reason"),
1323                )
1324                .with_new_label(
1325                    "prepared_sort_groupby_num_groups",
1326                    aggregate_fact(facts, "prepared_sort_groupby_num_groups"),
1327                )
1328                .with_new_label(
1329                    "prepared_sort_groupby_grouped_key_output_bytes",
1330                    aggregate_fact(facts, "prepared_sort_groupby_grouped_key_output_bytes"),
1331                )
1332                .with_new_label(
1333                    "prepared_sort_groupby_max_group_size",
1334                    aggregate_fact(facts, "prepared_sort_groupby_max_group_size"),
1335                )
1336                .with_new_label("rejected_candidate_strategies", aggregate_fact(facts, "rejected_candidate_strategies"))
1337                .global_counter(metric_name::native_aggregate_strategy_rows())
1338                .add(1);
1339            self.native_aggregate_strategy_rows.add(1);
1340        }
1341    }
1342}
1343
1344#[derive(Debug, Clone, Copy, Default)]
1345struct JoinRuntimeCounters {
1346    partitioned_hash_join_count: u64,
1347    hash_join_build_count: u64,
1348    hash_join_hash_table_bytes: u64,
1349    filtered_join_build_count: u64,
1350    filtered_join_hash_table_bytes: u64,
1351    filtered_join_probe_state_bytes: u64,
1352    mixed_join_size_data_bytes: u64,
1353    one_shot_join_count: u64,
1354    one_shot_join_index_bytes: u64,
1355    one_shot_join_materialization_bytes: u64,
1356    one_shot_join_output_bytes: u64,
1357}
1358
1359/// The metrics of a plan whose host sink runs in the adapter's cursor.
1360///
1361/// The cursor records the sink's work into the plan through its recorder.
1362/// The plan's metrics freeze and reach DataFusion once, when this value
1363/// drops: at the end of the output, on a cursor failure, or when the consumer
1364/// stops early.
1365#[derive(Debug)]
1366pub(crate) struct DeferredHostMetrics {
1367    metrics: AlgeonNativeMetrics,
1368    plan: PlanMetrics,
1369    compiled_sink_pending: bool,
1370    // The admitted context of the cursor's steps. A field drops after
1371    // `Drop::drop` freezes the metrics, so the attempt cannot start its
1372    // release before they are recorded.
1373    _context_lease: AlgeonNativeExecutionContext,
1374}
1375
1376impl DeferredHostMetrics {
1377    fn new(metrics: AlgeonNativeMetrics, plan: PlanMetrics, context_lease: AlgeonNativeExecutionContext) -> Self {
1378        let compiled_sink_pending = plan.snapshot().compiled_pipeline_attempts > 0;
1379        Self { metrics, plan, compiled_sink_pending, _context_lease: context_lease }
1380    }
1381
1382    pub(crate) const fn baseline(&self) -> &BaselineMetrics {
1383        &self.metrics.baseline
1384    }
1385
1386    /// Records the host sink a compiled pipeline deferred to the adapter, once,
1387    /// when the cursor has run it: after its first batch, or at the end of an
1388    /// output that had none.
1389    pub(crate) fn record_sink_executed(&mut self) {
1390        if std::mem::take(&mut self.compiled_sink_pending) {
1391            let recorder = self.plan.recorder();
1392            recorder.record_compiled_pipeline_tasks_executed(1);
1393            recorder.record_compiled_pipeline_host_arrow_sink();
1394        }
1395    }
1396
1397    pub(crate) fn record_materialization(&self, batch: &RecordBatch) {
1398        self.plan.recorder().record_device_to_host_materialization(
1399            u64::try_from(batch.num_rows()).unwrap_or(u64::MAX),
1400            u64::try_from(get_record_batch_memory_size(batch)).unwrap_or(u64::MAX),
1401        );
1402    }
1403}
1404
1405impl Drop for DeferredHostMetrics {
1406    fn drop(&mut self) {
1407        self.metrics.record_query_metrics(&self.plan.freeze());
1408    }
1409}
1410
1411fn optional_fact(facts: Option<&BTreeMap<String, String>>, key: &str) -> String {
1412    facts.and_then(|facts| facts.get(key).cloned()).unwrap_or_else(|| "none".to_string())
1413}
1414
1415fn aggregate_fact(facts: &BTreeMap<String, String>, key: &str) -> String {
1416    facts.get(key).cloned().unwrap_or_else(|| "none".to_string())
1417}
1418
1419fn join_strategy_label(join: &JoinOperatorMetrics) -> &'static str {
1420    match join.execution_strategy.as_str() {
1421        "bounded_one_shot_equi_join" => "one_shot_equi",
1422        "bounded_partitioned_hash_equi_join" => "partitioned_hash_equi",
1423        "bounded_partitioned_mixed_keyed_residual_join" => "partitioned_mixed_keyed_residual",
1424        "bounded_reusable_hash_join" | "hash_join" => "reusable_hash_equi",
1425        "bounded_reusable_filtered_join" => "filtered_join",
1426        "bounded_mixed_keyed_residual_join" | "mixed_join" => "mixed_keyed_residual",
1427        "bounded_keyless_conditional_join" | "conditional_join" => "keyless_conditional",
1428        "bounded_projected_cross_join" | "cross_join" => "projected_cross",
1429        _ => "unknown",
1430    }
1431}
1432
1433fn join_runtime_counters(join: &JoinOperatorMetrics) -> JoinRuntimeCounters {
1434    match join_strategy_label(join) {
1435        "reusable_hash_equi" => JoinRuntimeCounters {
1436            hash_join_build_count: u64::from(join.hash_table_bytes > 0),
1437            hash_join_hash_table_bytes: join.hash_table_bytes,
1438            ..Default::default()
1439        },
1440        "filtered_join" => JoinRuntimeCounters {
1441            filtered_join_build_count: u64::from(join.hash_table_bytes > 0),
1442            filtered_join_hash_table_bytes: join.hash_table_bytes,
1443            filtered_join_probe_state_bytes: join.filtered_probe_state_bytes,
1444            ..Default::default()
1445        },
1446        "one_shot_equi" => JoinRuntimeCounters {
1447            one_shot_join_count: 1,
1448            one_shot_join_index_bytes: join.gather_map_bytes,
1449            // Counters sum the bounds that exist; an unbounded output adds none.
1450            one_shot_join_materialization_bytes: join.output_reservation_bytes.unwrap_or_default(),
1451            one_shot_join_output_bytes: join.output_bytes,
1452            ..Default::default()
1453        },
1454        "partitioned_hash_equi" => JoinRuntimeCounters { partitioned_hash_join_count: 1, ..Default::default() },
1455        "partitioned_mixed_keyed_residual" => JoinRuntimeCounters {
1456            partitioned_hash_join_count: 1,
1457            mixed_join_size_data_bytes: join.mixed_size_data_bytes,
1458            ..Default::default()
1459        },
1460        "mixed_keyed_residual" => {
1461            JoinRuntimeCounters { mixed_join_size_data_bytes: join.mixed_size_data_bytes, ..Default::default() }
1462        },
1463        _ => JoinRuntimeCounters::default(),
1464    }
1465}
1466
1467fn count(value: u64) -> usize {
1468    usize::try_from(value).unwrap_or(usize::MAX)
1469}
1470
1471fn add_optional_count(metric: &Count, observed: Option<&Count>, value: Option<u64>) {
1472    if let Some(value) = value {
1473        // `*_observed` distinguishes a real `Some(0)` sample from no observation at all.
1474        if let Some(observed) = observed {
1475            observed.add(1);
1476        }
1477        metric.add(count(value));
1478    }
1479}
1480
1481#[cfg(test)]
1482mod tests {
1483    use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
1484    use query_planning::plan::GraphAlgorithm;
1485    use query_runtime::{
1486        metrics::QueryMetrics,
1487        observability::{
1488            AttemptCapacityDisposition, AttemptCleanupDisposition, AttemptDisposition, AttemptHealthTransition,
1489            AttemptId, AttemptRetryAdvice, RuntimeTerminalOutcome, SEALED_ATTEMPT_PROJECTION_SCHEMA,
1490            SealedAttemptLabels, SealedAttemptProjection,
1491            graph_observation::{GraphClaimExecutionState, GraphMeasuredPeakProvenance, GraphMeasuredPeakRow},
1492        },
1493    };
1494
1495    use super::*;
1496
1497    fn calibration_row(index: usize, algorithm: GraphAlgorithm) -> GraphMeasuredPeakRow {
1498        let mut row = GraphMeasuredPeakRow::pending(None, format!("n{index}"), index, algorithm);
1499        row.record_execution_state_for_test(GraphClaimExecutionState::AlgorithmStarted);
1500        row.record_measured_peak(Some(512 + index as u64), GraphMeasuredPeakProvenance::CacheScopeHighWater);
1501        row
1502    }
1503
1504    #[test]
1505    fn sealed_graph_calibration_snapshot_keeps_typed_rows_and_fixed_bound() {
1506        let metrics_set = ExecutionPlanMetricsSet::new();
1507        let capture = GraphClaimCalibrationCapture::default();
1508        let metrics = AlgeonNativeMetrics::new(&metrics_set, 0).with_graph_claim_calibration_capture(capture.clone());
1509        let mut query_metrics = QueryMetrics::default();
1510        for index in 0..=SEALED_ATTEMPT_GRAPH_CLAIM_CALIBRATION_MAX_ROWS {
1511            query_metrics.graph_claim_calibration_rows.push(calibration_row(index, GraphAlgorithm::PageRank));
1512        }
1513
1514        metrics.record_query_metrics(&query_metrics);
1515        let attempt_id = AttemptId::mint();
1516        let snapshot = SealedAttemptGraphClaimCalibrationSnapshot::from_captures(attempt_id, [capture]);
1517
1518        assert_eq!(snapshot.attempt_id(), attempt_id);
1519        assert!(snapshot.execution_complete());
1520        assert!(matches!(snapshot.execution(), Some(GraphClaimCalibrationExecution::Physical)));
1521        assert_eq!(snapshot.rows().len(), SEALED_ATTEMPT_GRAPH_CLAIM_CALIBRATION_MAX_ROWS);
1522        assert!(snapshot.truncated());
1523        assert_eq!(snapshot.rows()[0].node_index, 0);
1524        assert_eq!(snapshot.rows()[0].algorithm, GraphAlgorithm::PageRank);
1525        assert_eq!(snapshot.rows()[0].measured_peak_provenance, GraphMeasuredPeakProvenance::CacheScopeHighWater);
1526        assert_eq!(
1527            snapshot.rows()[SEALED_ATTEMPT_GRAPH_CLAIM_CALIBRATION_MAX_ROWS - 1].node_index,
1528            SEALED_ATTEMPT_GRAPH_CLAIM_CALIBRATION_MAX_ROWS - 1
1529        );
1530
1531        let mut late_metrics = QueryMetrics::default();
1532        late_metrics.graph_claim_calibration_rows.push(calibration_row(999, GraphAlgorithm::Bfs));
1533        metrics.record_query_metrics(&late_metrics);
1534        assert!(snapshot.rows().iter().all(|row| row.node_index != 999));
1535    }
1536
1537    #[test]
1538    #[cfg(feature = "cugraph")]
1539    fn sealed_graph_calibration_snapshot_retains_a_nonphysical_follower_leader_link() {
1540        let capture = GraphClaimCalibrationCapture::default();
1541        let leader_attempt_id = AttemptId::mint();
1542        capture.record_bfs_singleflight_follower(Some(leader_attempt_id));
1543
1544        let snapshot = SealedAttemptGraphClaimCalibrationSnapshot::from_captures(AttemptId::mint(), [capture]);
1545
1546        assert!(snapshot.execution_complete());
1547        assert!(snapshot.rows().is_empty());
1548        assert_eq!(
1549            snapshot.execution(),
1550            Some(&GraphClaimCalibrationExecution::NonPhysical(
1551                GraphClaimCalibrationNonPhysicalReason::BfsSingleflightFollower {
1552                    leader_attempt_id: Some(leader_attempt_id),
1553                },
1554            ))
1555        );
1556    }
1557
1558    #[test]
1559    fn sealed_graph_calibration_snapshot_uses_the_physical_producer_classification_from_a_composed_root() {
1560        let root_capture = GraphClaimCalibrationCapture::default();
1561        let producer_capture = GraphClaimCalibrationCapture::default();
1562        producer_capture.record(&[calibration_row(1, GraphAlgorithm::PageRank)]);
1563
1564        let snapshot = SealedAttemptGraphClaimCalibrationSnapshot::from_captures(
1565            AttemptId::mint(),
1566            [root_capture, producer_capture],
1567        );
1568
1569        assert!(snapshot.execution_complete());
1570        assert!(matches!(snapshot.execution(), Some(GraphClaimCalibrationExecution::Physical)));
1571    }
1572
1573    #[test]
1574    fn sealed_attempt_metric_labels_derive_from_the_sealed_projection() {
1575        let projection = SealedAttemptProjection {
1576            schema: SEALED_ATTEMPT_PROJECTION_SCHEMA,
1577            attempt_id: AttemptId::mint(),
1578            disposition: AttemptDisposition::Native,
1579            outcome: RuntimeTerminalOutcome::Panicked,
1580            cleanup_disposition: AttemptCleanupDisposition::CompletedWithFailures,
1581            cleanup_failure_count: 7,
1582            cleanup_failure_count_complete: false,
1583            capacity_disposition: AttemptCapacityDisposition::RetainedQuarantine,
1584            health_transition: Some(AttemptHealthTransition::BackendUnhealthy),
1585            retry_advice: AttemptRetryAdvice::indeterminate(),
1586            observation_truncated: true,
1587            observation_degraded: false,
1588            native_fault_evidence_present: true,
1589            resource_evidence_present: false,
1590        };
1591
1592        assert_eq!(
1593            projection.labels(),
1594            SealedAttemptLabels {
1595                disposition: "native",
1596                outcome: "panicked",
1597                cleanup_disposition: "completed_with_failures",
1598                cleanup_failure_count_complete: "false",
1599                capacity_disposition: "retained_quarantine",
1600                health_transition: "backend_unhealthy",
1601                retry_recovery: "indeterminate",
1602                retry_prerequisite: "evidence_unavailable",
1603                observation_truncated: "true",
1604                observation_degraded: "false",
1605                native_fault_evidence: "present",
1606                resource_evidence: "absent",
1607            }
1608        );
1609    }
1610}