query_runtime/metrics/
source_read.rs

1use std::time::Duration;
2
3use query_planning::correlation::QueryId;
4
5use super::{MetricsProjection, QueryMetrics, optional_u64};
6use crate::observability::QueryMetricsCheckpoint;
7
8/// Runtime evidence for one local-Parquet source read.
9#[derive(Debug, Clone, Default, PartialEq, Eq)]
10pub struct SourceReadRuntimeMetrics {
11    /// Query-wide diagnostic correlation id, when assigned.
12    pub query_id: Option<QueryId>,
13    /// Native source node id.
14    pub source_node_id: String,
15    /// Source read shape, for example `chunk` or `chunk_batch`.
16    pub read_kind: String,
17    /// Parquet read strategy used by the chunk or batch.
18    pub read_strategy: String,
19    /// Lower-layer Parquet read evidence reported by cudf.
20    pub parquet_read_evidence: SourceReadParquetEvidenceMetrics,
21    /// Batch index within the source stream, or the chunk index for singleton reads.
22    pub batch_index: u64,
23    /// Number of source chunks represented by this read.
24    pub chunk_count: u64,
25    /// First source chunk index represented by this read.
26    pub first_chunk_index: u64,
27    /// Last source chunk index represented by this read.
28    pub last_chunk_index: u64,
29    /// First source file index represented by this read.
30    pub first_file_index: u64,
31    /// Last source file index represented by this read.
32    pub last_file_index: u64,
33    /// Number of distinct files represented by this read.
34    pub file_count: u64,
35    /// Selected row groups represented by this read.
36    pub row_group_count: u64,
37    /// Number of physical read columns per represented file.
38    pub read_columns_per_file: u64,
39    /// Sum of physical read-column selections across represented files.
40    pub read_columns_total: u64,
41    /// Sum of known file bytes for represented files, when available.
42    pub known_file_bytes: u64,
43    /// Best available source-read byte estimate for represented chunks.
44    pub estimated_source_read_bytes: u64,
45    /// Sum of known selected row-group bytes, when available.
46    pub known_row_group_bytes: u64,
47    /// Sum of known projected compressed bytes, when available.
48    pub known_projected_compressed_bytes: u64,
49    /// Sum of known projected uncompressed bytes, when available.
50    pub known_projected_uncompressed_bytes: u64,
51    /// Sum of known pure-filter compressed bytes, when available.
52    pub known_pure_filter_compressed_bytes: u64,
53    /// Sum of known pure-filter uncompressed bytes, when available.
54    pub known_pure_filter_uncompressed_bytes: u64,
55    /// Number of output rows materialized by the read after reader predicates.
56    pub output_rows: u64,
57    /// Whether this read carried a source reader predicate.
58    pub reader_predicate: bool,
59    /// Number of dynamic-filter reader predicate routes attached to this read.
60    pub reader_dynamic_filter_count: u64,
61    /// Stable diagnostic reason for the source chunk batching decision.
62    pub source_batching_reason: String,
63    /// Source chunk planning mode requested for this source read.
64    pub source_chunk_planning_mode: String,
65    /// Stable diagnostic reason for the source chunk row-group and byte target.
66    pub source_chunk_planning_reason: String,
67    /// Elapsed time spent in the source read.
68    pub elapsed: Duration,
69}
70
71impl SourceReadRuntimeMetrics {
72    /// Serializes this row as stable comma-separated key/value pairs.
73    pub fn to_stable_kv_row(&self) -> String {
74        format!(
75            "source_node_id={},read_kind={},read_strategy={},batch_index={},chunk_count={},first_chunk_index={},last_chunk_index={},first_file_index={},last_file_index={},file_count={},row_group_count={},read_columns_per_file={},read_columns_total={},known_file_bytes={},estimated_source_read_bytes={},known_row_group_bytes={},known_projected_compressed_bytes={},known_projected_uncompressed_bytes={},known_pure_filter_compressed_bytes={},known_pure_filter_uncompressed_bytes={},output_rows={},reader_predicate={},reader_dynamic_filter_count={},source_batching_reason={},source_chunk_planning_mode={},source_chunk_planning_reason={},elapsed_ns={},parquet_read_input_kind={},remote_read_mode={},parquet_read_backend={},remote_backend_build_status={},remote_read_bytes={},remote_read_range_count={},device_read_support={},parquet_pruning_input_row_groups={},parquet_row_groups_after_stats_filter={},parquet_row_groups_after_bloom_filter={},parquet_bloom_pruning_status={}",
76            self.source_node_id,
77            self.read_kind,
78            self.read_strategy,
79            self.batch_index,
80            self.chunk_count,
81            self.first_chunk_index,
82            self.last_chunk_index,
83            self.first_file_index,
84            self.last_file_index,
85            self.file_count,
86            self.row_group_count,
87            self.read_columns_per_file,
88            self.read_columns_total,
89            self.known_file_bytes,
90            self.estimated_source_read_bytes,
91            self.known_row_group_bytes,
92            self.known_projected_compressed_bytes,
93            self.known_projected_uncompressed_bytes,
94            self.known_pure_filter_compressed_bytes,
95            self.known_pure_filter_uncompressed_bytes,
96            self.output_rows,
97            self.reader_predicate,
98            self.reader_dynamic_filter_count,
99            self.source_batching_reason,
100            self.source_chunk_planning_mode,
101            self.source_chunk_planning_reason,
102            self.elapsed.as_nanos().min(u128::from(u64::MAX)),
103            self.parquet_read_evidence.input_kind,
104            self.parquet_read_evidence.remote_read_mode,
105            self.parquet_read_evidence.backend,
106            self.parquet_read_evidence.backend_build_status,
107            optional_u64(self.parquet_read_evidence.remote_bytes),
108            optional_u64(self.parquet_read_evidence.remote_range_count),
109            self.parquet_read_evidence.device_read_support,
110            optional_u64(self.parquet_read_evidence.pruning_input_row_groups),
111            optional_u64(self.parquet_read_evidence.row_groups_after_stats_filter),
112            optional_u64(self.parquet_read_evidence.row_groups_after_bloom_filter),
113            self.parquet_read_evidence.bloom_pruning_status,
114        )
115    }
116}
117
118/// Lower-layer Parquet read evidence attached to a source-read metric row.
119#[derive(Debug, Clone, PartialEq, Eq)]
120pub struct SourceReadParquetEvidenceMetrics {
121    /// Input kind passed to the Parquet reader.
122    pub input_kind: String,
123    /// Requested remote read mode, or `none` for non-remote reads.
124    pub remote_read_mode: String,
125    /// Parquet reader backend evidence.
126    pub backend: String,
127    /// Build-time remote backend status reported by cudf/libcudf.
128    pub backend_build_status: String,
129    /// Remote bytes read, when reported by the lower layer.
130    pub remote_bytes: Option<u64>,
131    /// Remote range count, when reported by the lower layer.
132    pub remote_range_count: Option<u64>,
133    /// Direct device-read support evidence reported by the lower layer.
134    pub device_read_support: String,
135    /// Row groups participating in this read after any caller row-group selection.
136    pub pruning_input_row_groups: Option<u64>,
137    /// Row groups surviving libcudf statistics pruning, when a predicate was present.
138    pub row_groups_after_stats_filter: Option<u64>,
139    /// Row groups surviving libcudf Bloom-filter pruning, when a predicate was present.
140    pub row_groups_after_bloom_filter: Option<u64>,
141    /// Stable status explaining Bloom-pruning evidence or known ineligibility.
142    pub bloom_pruning_status: String,
143}
144
145impl Default for SourceReadParquetEvidenceMetrics {
146    fn default() -> Self {
147        Self {
148            input_kind: "none".to_string(),
149            remote_read_mode: "none".to_string(),
150            backend: "none".to_string(),
151            backend_build_status: "none".to_string(),
152            remote_bytes: None,
153            remote_range_count: None,
154            device_read_support: "none".to_string(),
155            pruning_input_row_groups: None,
156            row_groups_after_stats_filter: None,
157            row_groups_after_bloom_filter: None,
158            bloom_pruning_status: "not_reported".to_string(),
159        }
160    }
161}
162
163impl QueryMetrics {
164    /// Projects a plan execution's metrics from its owner checkpoint.
165    ///
166    /// Every fact-driven field is replaced by the checkpoint's absolute value,
167    /// so applying the same checkpoint twice leaves the value unchanged. Only
168    /// the identity fields this value was created with are kept;
169    /// the observation owner is the only writer of everything else.
170    pub fn apply_owner_checkpoint(&mut self, checkpoint: &QueryMetricsCheckpoint) {
171        let query_id = self.query_id.clone();
172        let trace_id = self.trace_id.clone();
173        let plan_task_mapping_datafusion_shape = self.plan_task_mapping_datafusion_shape.clone();
174        *self = checkpoint.metrics.clone();
175        self.query_id = query_id;
176        self.trace_id = trace_id;
177        self.plan_task_mapping_datafusion_shape = plan_task_mapping_datafusion_shape;
178        self.apply_source_read_checkpoint(checkpoint);
179        self.stamp_existing_rows_with_query_id();
180    }
181
182    /// Projects a plan execution's source-read metrics from its owner checkpoint.
183    ///
184    /// Every source-read field is replaced by the checkpoint's absolute value,
185    /// so applying the same checkpoint twice leaves the value unchanged. The
186    /// observation owner is the only writer of this family; nothing else may
187    /// set these fields, or this projection would overwrite it.
188    pub(crate) fn apply_source_read_checkpoint(&mut self, checkpoint: &QueryMetricsCheckpoint) {
189        let totals = &checkpoint.summary.source_read;
190        self.input_batches = totals.input_batches;
191        self.input_rows = totals.input_rows;
192        self.parquet_files = totals.parquet_files;
193        self.parquet_bytes_known = totals.parquet_bytes_known;
194        self.row_groups_selected = totals.row_groups_selected;
195        self.read_columns = totals.read_columns;
196        self.source_chunks_planned = totals.source_chunks_planned;
197        self.source_chunks_attempted = totals.source_chunks_attempted;
198        self.source_chunks_committed = totals.source_chunks_committed;
199        self.source_chunks_retried = totals.source_chunks_retried;
200        self.row_groups_planned = totals.row_groups_planned;
201        self.row_groups_committed = totals.row_groups_committed;
202        self.source_read_prefetches_started = totals.source_read_prefetches_started;
203        self.source_read_prefetches_completed = totals.source_read_prefetches_completed;
204        self.source_read_prefetches_skipped = totals.source_read_prefetches_skipped;
205        self.source_read_prefetches_abandoned = totals.source_read_prefetches_abandoned;
206        self.source_metadata_load_time = totals.source_metadata_load_time;
207        self.source_read_queue_wait_time = totals.source_read_queue_wait_time;
208        self.parquet_read_time = totals.parquet_read_time;
209        self.source_read_time = totals.source_read_time;
210
211        let query_id = self.inherited_query_id();
212        self.source_read_runtime_rows = checkpoint
213            .source_read_rows
214            .iter()
215            .cloned()
216            .map(|mut row| {
217                if row.query_id.is_none() {
218                    row.query_id.clone_from(&query_id);
219                }
220                row
221            })
222            .collect();
223        self.projection = Some(MetricsProjection {
224            revision: Some(checkpoint.revision),
225            sealed: checkpoint.sealed,
226            completeness: checkpoint.completeness,
227        });
228    }
229
230    /// Records one local-Parquet source read or source-batch read.
231    pub fn record_source_read_runtime(&mut self, mut metrics: SourceReadRuntimeMetrics) {
232        if metrics.query_id.is_none() {
233            metrics.query_id = self.inherited_query_id();
234        }
235        self.source_read_runtime_rows.push(metrics);
236    }
237
238    /// Records one input batch and its row count.
239    pub fn record_input_batch(&mut self, rows: u64) {
240        self.input_batches += 1;
241        self.input_rows += rows;
242    }
243
244    /// Records one Parquet source file observation.
245    pub fn record_parquet_source(&mut self, known_bytes: Option<u64>, selected_row_groups: u64, read_columns: u64) {
246        self.parquet_files += 1;
247        self.parquet_bytes_known += known_bytes.unwrap_or(0);
248        self.row_groups_selected += selected_row_groups;
249        self.read_columns += read_columns;
250    }
251
252    /// Records that a source chunk was planned.
253    pub fn record_source_chunk_planned(&mut self, row_groups: u64) {
254        self.source_chunks_planned = self.source_chunks_planned.saturating_add(1);
255        self.row_groups_planned = self.row_groups_planned.saturating_add(row_groups);
256    }
257
258    /// Records one source chunk attempt.
259    pub fn record_source_chunk_attempt(&mut self) {
260        self.source_chunks_attempted = self.source_chunks_attempted.saturating_add(1);
261    }
262
263    /// Records one committed source chunk and row-group count.
264    pub fn record_source_chunk_commit(&mut self, row_groups: u64) {
265        self.record_source_chunk_commit_count(1, row_groups);
266    }
267
268    /// Records committed source chunks and row-group count.
269    pub fn record_source_chunk_commit_count(&mut self, chunk_count: u64, row_groups: u64) {
270        self.source_chunks_committed = self.source_chunks_committed.saturating_add(chunk_count);
271        self.row_groups_committed = self.row_groups_committed.saturating_add(row_groups);
272    }
273
274    /// Records one source chunk retry.
275    pub fn record_source_chunk_retry(&mut self) {
276        self.source_chunks_retried = self.source_chunks_retried.saturating_add(1);
277    }
278
279    /// Records a started source-read prefetch.
280    pub fn record_source_read_prefetch_started(&mut self) {
281        self.source_read_prefetches_started = self.source_read_prefetches_started.saturating_add(1);
282    }
283
284    /// Records a completed source-read prefetch.
285    pub fn record_source_read_prefetch_completed(&mut self) {
286        self.source_read_prefetches_completed = self.source_read_prefetches_completed.saturating_add(1);
287    }
288
289    /// Records a skipped source-read prefetch.
290    pub fn record_source_read_prefetch_skipped(&mut self) {
291        self.source_read_prefetches_skipped = self.source_read_prefetches_skipped.saturating_add(1);
292    }
293
294    /// Records an abandoned source-read prefetch.
295    pub fn record_source_read_prefetch_abandoned(&mut self) {
296        self.source_read_prefetches_abandoned = self.source_read_prefetches_abandoned.saturating_add(1);
297    }
298}
299
300#[cfg(test)]
301mod projection_tests {
302    use std::time::Duration;
303
304    use super::{QueryMetrics, SourceReadRuntimeMetrics};
305    use crate::observability::{
306        MetricsCaptureMode, MetricsCompleteness, QueryMetricsCheckpoint, QueryMetricsSummary, SourceReadTotals,
307    };
308
309    fn checkpoint(revision: u64, input_rows: u64, capture_mode: MetricsCaptureMode) -> QueryMetricsCheckpoint {
310        QueryMetricsCheckpoint {
311            revision,
312            sealed: false,
313            summary: QueryMetricsSummary {
314                source_read: SourceReadTotals {
315                    input_rows,
316                    source_read_queue_wait_time: Duration::from_millis(input_rows),
317                    ..SourceReadTotals::default()
318                },
319            },
320            source_read_rows: vec![SourceReadRuntimeMetrics { batch_index: revision, ..Default::default() }],
321            metrics: QueryMetrics::default(),
322            completeness: MetricsCompleteness {
323                capture_mode,
324                detail_rows_dropped: 0,
325                truncated: false,
326                totals_saturated: false,
327            },
328        }
329    }
330
331    #[test]
332    fn projecting_a_checkpoint_again_does_not_add_it_twice() {
333        let checkpoint = checkpoint(4, 10, MetricsCaptureMode::Bounded);
334        let mut metrics = QueryMetrics::new("projection-query");
335
336        metrics.apply_source_read_checkpoint(&checkpoint);
337        let once = metrics.clone();
338        metrics.apply_source_read_checkpoint(&checkpoint);
339
340        assert_eq!(metrics, once, "a repeated projection must be idempotent, not additive");
341        assert_eq!(metrics.input_rows, 10);
342        assert_eq!(metrics.source_read_runtime_rows.len(), 1);
343        assert_eq!(metrics.projection.and_then(|projection| projection.revision), Some(4));
344    }
345
346    #[test]
347    fn a_later_checkpoint_replaces_an_earlier_one() {
348        let mut metrics = QueryMetrics::new("projection-query");
349
350        metrics.apply_source_read_checkpoint(&checkpoint(1, 10, MetricsCaptureMode::Bounded));
351        metrics.apply_source_read_checkpoint(&checkpoint(2, 25, MetricsCaptureMode::Bounded));
352
353        assert_eq!(metrics.input_rows, 25, "a preview then a final projection must not sum");
354        assert_eq!(metrics.source_read_queue_wait_time, Duration::from_millis(25));
355        assert_eq!(metrics.source_read_runtime_rows[0].batch_index, 2);
356    }
357
358    #[test]
359    fn projected_rows_inherit_the_query_id() {
360        let mut metrics = QueryMetrics::new("projection-query");
361
362        metrics.apply_source_read_checkpoint(&checkpoint(1, 1, MetricsCaptureMode::Bounded));
363
364        assert_eq!(
365            metrics.source_read_runtime_rows[0].query_id.as_ref().map(|id| id.as_str()),
366            Some("projection-query")
367        );
368    }
369
370    #[test]
371    fn merging_projections_never_claims_more_than_either_side() {
372        let mut bounded = QueryMetrics::new("merged-query");
373        bounded.apply_source_read_checkpoint(&checkpoint(3, 1, MetricsCaptureMode::Bounded));
374        let mut disabled = QueryMetrics::new("merged-query");
375        disabled.apply_source_read_checkpoint(&checkpoint(7, 1, MetricsCaptureMode::Disabled));
376
377        bounded.merge_from(disabled);
378
379        let projection = bounded.projection.expect("both sides were projected");
380        assert_eq!(projection.revision, None, "a merge of two checkpoints is no single revision");
381        assert_eq!(projection.completeness.capture_mode, MetricsCaptureMode::Disabled);
382        assert!(!projection.completeness.detail_is_complete());
383    }
384
385    #[test]
386    fn json_reports_whether_its_rows_are_the_whole_detail() {
387        let unprojected = QueryMetrics::new("json-query");
388        let mut truncated = QueryMetrics::new("json-query");
389        let mut lossy = checkpoint(9, 1, MetricsCaptureMode::Bounded);
390        lossy.completeness.detail_rows_dropped = 3;
391        lossy.completeness.truncated = true;
392        truncated.apply_source_read_checkpoint(&lossy);
393        let mut not_captured = QueryMetrics::new("json-query");
394        not_captured.apply_source_read_checkpoint(&checkpoint(2, 1, MetricsCaptureMode::Disabled));
395
396        assert_eq!(unprojected.to_json()["completeness"]["status"], "unknown");
397        let completeness = &truncated.to_json()["completeness"];
398        assert_eq!(completeness["status"], "incomplete");
399        assert_eq!(completeness["revision"], 9);
400        assert_eq!(completeness["detail_rows_dropped"], 3);
401        assert_eq!(not_captured.to_json()["completeness"]["status"], "not_captured");
402    }
403
404    #[test]
405    fn an_unprojected_side_does_not_erase_a_projection() {
406        let mut accumulated = QueryMetrics::new("merged-query");
407        let mut stage = QueryMetrics::new("merged-query");
408        stage.apply_source_read_checkpoint(&checkpoint(5, 1, MetricsCaptureMode::Bounded));
409
410        accumulated.merge_from(stage);
411
412        assert_eq!(accumulated.projection.and_then(|projection| projection.revision), Some(5));
413    }
414}
query_runtime/metrics/source_read.rs

query_runtime/metrics/
source_read.rs
