query_planning/source/
diagnostics.rs

1use std::path::Path;
2
3use super::{LocalParquetFile, ParquetSourceLocation, PartitionColumn, SourceColumnRef};
4use crate::util::fnv1a64;
5
6// Borrowed column-layout snapshot used to build source diagnostics.
7#[derive(Debug, Clone, Copy)]
8pub(super) struct LocalParquetDiagnosticColumns<'a> {
9    pub(super) read_columns: &'a [String],
10    pub(super) read_column_file_indices: &'a [usize],
11    pub(super) projected_output_columns: &'a [String],
12    pub(super) projected_output_column_source_refs: &'a [SourceColumnRef],
13    pub(super) pure_filter_columns: &'a [String],
14    pub(super) pure_filter_column_source_refs: &'a [SourceColumnRef],
15    pub(super) partition_columns: &'a [PartitionColumn],
16    pub(super) missing_columns: &'a [super::MissingColumn],
17    pub(super) scan_limit: Option<usize>,
18}
19
20/// Row-group selection evidence for a local Parquet source.
21///
22/// `None` on the parent diagnostics means the source does not have enough
23/// row-group metadata to distinguish unknown selection from an all-row-group
24/// scan. Byte counts are optional because some frontends only know row counts
25/// or file sizes at lowering time.
26#[derive(Debug, Clone, PartialEq, Eq)]
27pub struct SelectedRowGroupStats {
28    /// Number of row groups selected for scanning.
29    pub selected_row_groups: usize,
30    /// Total row groups represented by the source metadata.
31    pub total_row_groups: usize,
32    /// Selected uncompressed source-read bytes, when known.
33    pub(crate) selected_uncompressed_bytes: Option<u64>,
34    /// Total uncompressed source-read bytes, when known.
35    pub(crate) total_uncompressed_bytes: Option<u64>,
36    /// Selected compressed source-read bytes, when known.
37    pub selected_compressed_bytes: Option<u64>,
38    /// Total compressed source-read bytes, when known.
39    pub total_compressed_bytes: Option<u64>,
40    /// Selected source rows, when known.
41    pub selected_rows: Option<u64>,
42    /// Total source rows, when known.
43    pub total_rows: Option<u64>,
44}
45
46impl SelectedRowGroupStats {
47    /// Best available selected byte evidence for source-read sizing.
48    pub fn selected_source_bytes(&self) -> Option<u64> {
49        self.selected_uncompressed_bytes.or(self.selected_compressed_bytes)
50    }
51
52    /// Best available total byte evidence for source-read sizing.
53    pub fn total_source_bytes(&self) -> Option<u64> {
54        self.total_uncompressed_bytes.or(self.total_compressed_bytes)
55    }
56}
57
58/// Frontend-visible source capability and lowering diagnostics.
59///
60/// Diagnostics explain why a source can or cannot use the native direct device
61/// path. They are intentionally structural so adapters and tests can assert on
62/// stable fields without parsing displayed plans. Only local Parquet sources
63/// produce these diagnostics.
64#[derive(Debug, Clone, PartialEq, Eq)]
65pub struct SourceDiagnostics {
66    /// Comma-separated high-level Parquet read strategies represented by this source.
67    pub source_read_strategy: String,
68    /// Comma-separated credential modes represented by this source.
69    pub credential_mode: String,
70    /// Comma-separated local byte-cache statuses represented by this source.
71    pub lakehouse_cache_status: String,
72    /// Comma-separated foreground lakehouse cache-read modes represented by this source.
73    pub lakehouse_cache_read_mode: String,
74    /// Comma-separated complete-object lakehouse cache fill modes represented by this source.
75    pub lakehouse_cache_fill_mode: String,
76    /// True when the source is proven to use a local path/cache reader that
77    /// can satisfy the direct-device source-read boundary.
78    pub direct_device_read: bool,
79    /// Number of files represented after any frontend coalescing.
80    pub file_count: usize,
81    /// Number of source file entries that carried byte ranges.
82    pub ranged_file_count: usize,
83    /// Total known bytes across represented files, or `None` when incomplete.
84    pub known_bytes: Option<u64>,
85    /// Total selected row groups, or `None` when row-group selection is unknown.
86    pub selected_row_groups: Option<usize>,
87    /// Selected-versus-total row-group, byte, and row evidence.
88    pub selected_row_group_stats: Option<SelectedRowGroupStats>,
89    /// How pushed-down source predicates were represented.
90    pub predicate_disposition: PredicateDisposition,
91    /// How byte-range file entries were represented.
92    pub byte_range_disposition: ByteRangeDisposition,
93    /// How partition columns were represented.
94    pub partition_column_disposition: PartitionColumnDisposition,
95    /// Whether metadata-only zero-column count was required and represented.
96    pub zero_column_count_disposition: ZeroColumnCountDisposition,
97    /// Whether a source-level scan limit is absent or represented.
98    pub scan_limit_disposition: ScanLimitDisposition,
99    /// How logical source columns map to physical or synthetic source columns.
100    pub schema_mapping_disposition: SchemaMappingDisposition,
101    /// Number of physical Parquet columns read by the native source.
102    pub read_column_count: usize,
103    /// Number of logical source columns expected to remain after
104    /// source-adjacent native filtering/projection.
105    pub projected_output_column_count: usize,
106    /// Number of emitted source columns used only for source filtering.
107    pub pure_filter_column_count: usize,
108    /// Stable feature codes explaining why direct device read was rejected.
109    pub rejected_features: Vec<RejectedSourceFeature>,
110}
111
112impl SourceDiagnostics {
113    /// Sets the read strategy, credential mode, and byte-cache labels to the
114    /// ones a source over the resolved `files` reports, and direct device read
115    /// to whether those files prove it and no feature was rejected.
116    pub fn set_resolved_read_labels(&mut self, files: &[LocalParquetFile]) {
117        self.source_read_strategy = summarize_source_read_strategy(files);
118        self.credential_mode = summarize_credential_mode(files);
119        self.lakehouse_cache_status = summarize_lakehouse_cache_status(files);
120        self.lakehouse_cache_read_mode = summarize_lakehouse_cache_read_mode(files);
121        self.lakehouse_cache_fill_mode = summarize_lakehouse_cache_fill_mode(files);
122        self.direct_device_read = self.rejected_features.is_empty() && files_have_proven_direct_device_read(files);
123    }
124
125    pub(super) fn for_local_parquet_files(
126        files: &[LocalParquetFile],
127        columns: LocalParquetDiagnosticColumns<'_>,
128    ) -> Self {
129        let known_bytes = files
130            .iter()
131            .map(|file| file.file_size_bytes)
132            .try_fold(0u64, |total, bytes| bytes.map(|bytes| total.saturating_add(bytes)));
133        let selected_row_groups = files
134            .iter()
135            .any(|file| file.row_groups.is_some())
136            .then(|| files.iter().map(LocalParquetFile::selected_row_group_count).sum());
137        let selected_row_group_stats = selected_row_group_stats_for_files(files);
138        let zero_column_count_disposition = if columns.read_columns.is_empty()
139            && columns.missing_columns.is_empty()
140            && columns.partition_columns.is_empty()
141        {
142            ZeroColumnCountDisposition::Rejected
143        } else {
144            ZeroColumnCountDisposition::NotRequired
145        };
146        let mut rejected_features = Vec::new();
147        if zero_column_count_disposition == ZeroColumnCountDisposition::Rejected {
148            rejected_features.push(RejectedSourceFeature::ZeroColumnCount);
149        }
150        let schema_refs_are_valid = columns.read_columns.len() == columns.read_column_file_indices.len()
151            && columns.projected_output_columns.len() == columns.projected_output_column_source_refs.len()
152            && columns.pure_filter_columns.len() == columns.pure_filter_column_source_refs.len();
153        let schema_mapping_disposition = if schema_refs_are_valid {
154            if columns
155                .projected_output_column_source_refs
156                .iter()
157                .chain(columns.pure_filter_column_source_refs.iter())
158                .any(|source_ref| matches!(source_ref, SourceColumnRef::MissingNull(_)))
159            {
160                SchemaMappingDisposition::MissingNullableColumns
161            } else {
162                SchemaMappingDisposition::DirectByName
163            }
164        } else {
165            rejected_features.push(RejectedSourceFeature::SchemaMapping);
166            SchemaMappingDisposition::Rejected
167        };
168        let partition_column_disposition = if columns.partition_columns.is_empty() {
169            PartitionColumnDisposition::Absent
170        } else {
171            PartitionColumnDisposition::Represented
172        };
173        let scan_limit_disposition =
174            if columns.scan_limit.is_some() { ScanLimitDisposition::Represented } else { ScanLimitDisposition::Absent };
175        Self {
176            source_read_strategy: summarize_source_read_strategy(files),
177            credential_mode: summarize_credential_mode(files),
178            lakehouse_cache_status: summarize_lakehouse_cache_status(files),
179            lakehouse_cache_read_mode: summarize_lakehouse_cache_read_mode(files),
180            lakehouse_cache_fill_mode: summarize_lakehouse_cache_fill_mode(files),
181            direct_device_read: rejected_features.is_empty() && files_have_proven_direct_device_read(files),
182            file_count: files.len(),
183            ranged_file_count: 0,
184            known_bytes,
185            selected_row_groups,
186            selected_row_group_stats,
187            predicate_disposition: PredicateDisposition::Absent,
188            byte_range_disposition: ByteRangeDisposition::FullFiles,
189            partition_column_disposition,
190            zero_column_count_disposition,
191            scan_limit_disposition,
192            schema_mapping_disposition,
193            read_column_count: columns.read_columns.len(),
194            projected_output_column_count: columns.projected_output_columns.len(),
195            pure_filter_column_count: columns.pure_filter_columns.len(),
196            rejected_features,
197        }
198    }
199}
200
201pub fn redacted_path_diagnostic(path: &Path) -> String {
202    let path = path.display().to_string();
203    redacted_location_diagnostic(path.as_str()).unwrap_or(path)
204}
205
206pub fn parquet_source_path_fact(location: &ParquetSourceLocation, path: &Path) -> String {
207    if location.is_local_path() { redacted_path_diagnostic(path) } else { location.diagnostic_location() }
208}
209
210pub(crate) fn parquet_source_stable_id_fact(location: &ParquetSourceLocation, stable_id: &str) -> String {
211    location.diagnostic_stable_id(stable_id)
212}
213
214pub fn parquet_file_path_fact(file: &LocalParquetFile) -> String {
215    parquet_source_path_fact(&file.location, &file.path)
216}
217
218pub fn parquet_file_stable_id_fact(file: &LocalParquetFile) -> String {
219    parquet_source_stable_id_fact(&file.location, file.stable_id.as_str())
220}
221
222pub fn local_parquet_file_row_group_metadata_count(file: &LocalParquetFile) -> Option<usize> {
223    file.row_group_bytes
224        .as_ref()
225        .or(file.projected_row_group_uncompressed_bytes.as_ref())
226        .or(file.projected_row_group_compressed_bytes.as_ref())
227        .or(file.pure_filter_row_group_uncompressed_bytes.as_ref())
228        .or(file.pure_filter_row_group_compressed_bytes.as_ref())
229        .or(file.row_group_rows.as_ref())
230        .map(Vec::len)
231}
232
233pub(crate) fn redacted_stable_id_diagnostic(stable_id: &str) -> String {
234    redacted_location_diagnostic(stable_id).unwrap_or_else(|| stable_id.to_string())
235}
236
237pub fn redacted_object_store_path_diagnostic(path: &str) -> String {
238    format!("object_store_path:fnv64:{:016x}", fnv1a64(path.as_bytes()))
239}
240
241pub(super) fn redacted_debug_value(kind: &str, value: &str) -> String {
242    format!("{kind}:fnv64:{:016x}:len:{}", fnv1a64(value.as_bytes()), value.len())
243}
244
245pub(super) fn redacted_debug_option(kind: &str, value: Option<&str>) -> Option<String> {
246    value.map(|value| redacted_debug_value(kind, value))
247}
248
249fn redacted_location_diagnostic(location: &str) -> Option<String> {
250    let (scheme, _) = location.split_once("://")?;
251    let scheme = scheme.to_ascii_lowercase();
252    Some(object_store_diagnostic_id(scheme.as_str(), location))
253}
254
255pub(super) fn object_store_diagnostic_id(scheme: &str, identity: &str) -> String {
256    format!("object_store:{scheme}:fnv64:{:016x}", fnv1a64(identity.as_bytes()))
257}
258
259fn summarize_source_read_strategy(files: &[LocalParquetFile]) -> String {
260    summarize_file_values(files, |file| file.resolved_source.read_strategy().as_str())
261}
262
263pub(super) fn files_have_proven_direct_device_read(files: &[LocalParquetFile]) -> bool {
264    !files.is_empty()
265        && files.iter().all(|file| {
266            matches!(
267                file.resolved_source.read_strategy(),
268                super::ParquetReadStrategy::LocalKvikio | super::ParquetReadStrategy::LocalCacheKvikio
269            )
270        })
271}
272
273fn summarize_credential_mode(files: &[LocalParquetFile]) -> String {
274    summarize_file_values(files, |file| file.resolved_source.credential_mode().as_str())
275}
276
277fn summarize_lakehouse_cache_status(files: &[LocalParquetFile]) -> String {
278    summarize_file_values(files, |file| file.resolved_source.cache_status.as_str())
279}
280
281fn summarize_lakehouse_cache_read_mode(files: &[LocalParquetFile]) -> String {
282    summarize_file_values(files, |file| file.resolved_source.lakehouse_cache_read_mode.as_str())
283}
284
285fn summarize_lakehouse_cache_fill_mode(files: &[LocalParquetFile]) -> String {
286    summarize_file_values(files, |file| file.resolved_source.lakehouse_cache_fill_mode.as_str())
287}
288
289fn summarize_file_values(files: &[LocalParquetFile], value: impl Fn(&LocalParquetFile) -> &'static str) -> String {
290    let mut values = files.iter().map(value).collect::<Vec<_>>();
291    values.sort_unstable();
292    values.dedup();
293    values.join(",")
294}
295
296fn selected_row_group_stats_for_files(files: &[LocalParquetFile]) -> Option<SelectedRowGroupStats> {
297    let mut aggregate = SelectedRowGroupStatsAccumulator::default();
298    for file in files {
299        let file_stats = selected_row_group_stats_for_file(file)?;
300        aggregate.add(file_stats)?;
301    }
302    aggregate.finish()
303}
304
305fn selected_row_group_stats_for_file(file: &LocalParquetFile) -> Option<SelectedRowGroupStats> {
306    let total_row_groups = local_parquet_file_row_group_metadata_count(file)?;
307    if total_row_groups == 0 {
308        return None;
309    }
310    let selected_indices = file.row_groups.clone().unwrap_or_else(|| (0..total_row_groups).collect::<Vec<_>>());
311    if selected_indices.iter().any(|index| *index >= total_row_groups) {
312        return None;
313    }
314
315    let total_compressed_bytes = source_read_bytes(
316        file.projected_row_group_compressed_bytes.as_ref(),
317        file.pure_filter_row_group_compressed_bytes.as_ref(),
318        file.row_group_bytes.as_ref(),
319        0..total_row_groups,
320    )
321    .or(file.file_size_bytes);
322    let selected_compressed_bytes = source_read_bytes(
323        file.projected_row_group_compressed_bytes.as_ref(),
324        file.pure_filter_row_group_compressed_bytes.as_ref(),
325        file.row_group_bytes.as_ref(),
326        selected_indices.iter().copied(),
327    )
328    .or_else(|| proportional_file_bytes(file.file_size_bytes, selected_indices.len(), total_row_groups));
329    let total_uncompressed_bytes = source_read_bytes(
330        file.projected_row_group_uncompressed_bytes.as_ref(),
331        file.pure_filter_row_group_uncompressed_bytes.as_ref(),
332        file.row_group_bytes.as_ref(),
333        0..total_row_groups,
334    );
335    let selected_uncompressed_bytes = source_read_bytes(
336        file.projected_row_group_uncompressed_bytes.as_ref(),
337        file.pure_filter_row_group_uncompressed_bytes.as_ref(),
338        file.row_group_bytes.as_ref(),
339        selected_indices.iter().copied(),
340    );
341    let total_rows = row_group_sum(file.row_group_rows.as_ref(), 0..total_row_groups);
342    let selected_rows = row_group_sum(file.row_group_rows.as_ref(), selected_indices.iter().copied());
343
344    Some(SelectedRowGroupStats {
345        selected_row_groups: selected_indices.len(),
346        total_row_groups,
347        selected_uncompressed_bytes,
348        total_uncompressed_bytes,
349        selected_compressed_bytes,
350        total_compressed_bytes,
351        selected_rows,
352        total_rows,
353    })
354}
355
356fn source_read_bytes(
357    projected: Option<&Vec<u64>>,
358    pure_filter: Option<&Vec<u64>>,
359    fallback: Option<&Vec<u64>>,
360    indexes: impl IntoIterator<Item = usize> + Clone,
361) -> Option<u64> {
362    match (projected, pure_filter) {
363        (Some(projected), Some(pure_filter)) => {
364            checked_row_group_sum(projected, indexes.clone())?.checked_add(checked_row_group_sum(pure_filter, indexes)?)
365        },
366        (Some(projected), None) => checked_row_group_sum(projected, indexes),
367        (None, Some(pure_filter)) => checked_row_group_sum(pure_filter, indexes),
368        (None, None) => row_group_sum(fallback, indexes),
369    }
370}
371
372fn row_group_sum(values: Option<&Vec<u64>>, indexes: impl IntoIterator<Item = usize>) -> Option<u64> {
373    checked_row_group_sum(values?, indexes)
374}
375
376fn checked_row_group_sum(values: &[u64], indexes: impl IntoIterator<Item = usize>) -> Option<u64> {
377    indexes.into_iter().try_fold(0_u64, |total, index| total.checked_add(*values.get(index)?))
378}
379
380fn proportional_file_bytes(
381    file_size_bytes: Option<u64>,
382    selected_row_groups: usize,
383    total_row_groups: usize,
384) -> Option<u64> {
385    if total_row_groups == 0 {
386        return None;
387    }
388    let file_size_bytes = file_size_bytes?;
389    let numerator = u128::from(file_size_bytes).checked_mul(selected_row_groups as u128)?;
390    u64::try_from(numerator / total_row_groups as u128).ok()
391}
392
393#[derive(Debug)]
394struct SelectedRowGroupStatsAccumulator {
395    selected_row_groups: usize,
396    total_row_groups: usize,
397    selected_uncompressed_bytes: Option<u64>,
398    total_uncompressed_bytes: Option<u64>,
399    selected_compressed_bytes: Option<u64>,
400    total_compressed_bytes: Option<u64>,
401    selected_rows: Option<u64>,
402    total_rows: Option<u64>,
403}
404
405impl Default for SelectedRowGroupStatsAccumulator {
406    fn default() -> Self {
407        Self {
408            selected_row_groups: 0,
409            total_row_groups: 0,
410            selected_uncompressed_bytes: Some(0),
411            total_uncompressed_bytes: Some(0),
412            selected_compressed_bytes: Some(0),
413            total_compressed_bytes: Some(0),
414            selected_rows: Some(0),
415            total_rows: Some(0),
416        }
417    }
418}
419
420impl SelectedRowGroupStatsAccumulator {
421    fn add(&mut self, stats: SelectedRowGroupStats) -> Option<()> {
422        self.selected_row_groups = self.selected_row_groups.checked_add(stats.selected_row_groups)?;
423        self.total_row_groups = self.total_row_groups.checked_add(stats.total_row_groups)?;
424        self.selected_uncompressed_bytes =
425            add_optional_u64(self.selected_uncompressed_bytes, stats.selected_uncompressed_bytes)?;
426        self.total_uncompressed_bytes =
427            add_optional_u64(self.total_uncompressed_bytes, stats.total_uncompressed_bytes)?;
428        self.selected_compressed_bytes =
429            add_optional_u64(self.selected_compressed_bytes, stats.selected_compressed_bytes)?;
430        self.total_compressed_bytes = add_optional_u64(self.total_compressed_bytes, stats.total_compressed_bytes)?;
431        self.selected_rows = add_optional_u64(self.selected_rows, stats.selected_rows)?;
432        self.total_rows = add_optional_u64(self.total_rows, stats.total_rows)?;
433        Some(())
434    }
435
436    fn finish(self) -> Option<SelectedRowGroupStats> {
437        (self.total_row_groups > 0).then_some(SelectedRowGroupStats {
438            selected_row_groups: self.selected_row_groups,
439            total_row_groups: self.total_row_groups,
440            selected_uncompressed_bytes: self.selected_uncompressed_bytes,
441            total_uncompressed_bytes: self.total_uncompressed_bytes,
442            selected_compressed_bytes: self.selected_compressed_bytes,
443            total_compressed_bytes: self.total_compressed_bytes,
444            selected_rows: self.selected_rows,
445            total_rows: self.total_rows,
446        })
447    }
448}
449
450fn add_optional_u64(left: Option<u64>, right: Option<u64>) -> Option<Option<u64>> {
451    match (left, right) {
452        (Some(left), Some(right)) => Some(Some(left.checked_add(right)?)),
453        _ => Some(None),
454    }
455}
456
457/// Disposition of a source-level scan limit.
458#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
459pub enum ScanLimitDisposition {
460    /// No scan limit was attached to the source.
461    Absent,
462    /// Scan limit is represented by native source execution.
463    Represented,
464    /// Scan limit was present but could not be represented safely.
465    Rejected,
466}
467
468impl ScanLimitDisposition {
469    /// Stable lower-case code used by reports.
470    pub const fn as_str(self) -> &'static str {
471        match self {
472            Self::Absent => "absent",
473            Self::Represented => "represented",
474            Self::Rejected => "rejected",
475        }
476    }
477}
478
479/// Disposition of a source-level predicate.
480#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
481pub enum PredicateDisposition {
482    /// No predicate was attached to the source.
483    Absent,
484    /// Predicate is represented only for file or row-group pruning.
485    PruningOnly,
486    /// Predicate is represented as a reader-level Parquet predicate.
487    ReaderPushdown,
488    /// Predicate remains represented by native post-scan filtering.
489    NativePostScanOnly,
490    /// Predicate exists but cannot be represented by the native source.
491    Rejected,
492}
493
494impl PredicateDisposition {
495    /// Stable lower-case code used by reports.
496    pub const fn as_str(self) -> &'static str {
497        match self {
498            Self::Absent => "absent",
499            Self::PruningOnly => "pruning_only",
500            Self::ReaderPushdown => "reader_pushdown",
501            Self::NativePostScanOnly => "native_post_scan_only",
502            Self::Rejected => "rejected",
503        }
504    }
505}
506
507/// Disposition of byte-range file entries.
508///
509/// This describes the planning-time representation of file entries. The
510/// lower-level reads performed by the selected Parquet backend are not described
511/// here.
512#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
513pub enum ByteRangeDisposition {
514    /// Source entries describe whole files.
515    FullFiles,
516    /// Byte ranges were safely coalesced back to whole-file reads.
517    CoalescedFullFiles,
518    /// Byte-ranged source entries are represented by explicit row-group
519    /// selections on whole-file local reads.
520    RowGroupSelection,
521    /// Byte ranges remain and are not supported by the native source.
522    Rejected,
523}
524
525impl ByteRangeDisposition {
526    /// Stable lower-case code used by reports.
527    pub const fn as_str(self) -> &'static str {
528        match self {
529            Self::FullFiles => "full_files",
530            Self::CoalescedFullFiles => "coalesced_full_files",
531            Self::RowGroupSelection => "row_group_selection",
532            Self::Rejected => "rejected",
533        }
534    }
535}
536
537/// Disposition of partition columns used by the native source.
538#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
539pub enum PartitionColumnDisposition {
540    /// No partition columns are required.
541    Absent,
542    /// Required partition columns or values could not be represented.
543    Rejected,
544    /// Required partition columns are represented as synthetic source columns.
545    Represented,
546}
547
548impl PartitionColumnDisposition {
549    /// Stable lower-case code used by reports.
550    pub const fn as_str(self) -> &'static str {
551        match self {
552            Self::Absent => "absent",
553            Self::Rejected => "rejected",
554            Self::Represented => "represented",
555        }
556    }
557}
558
559/// Disposition of metadata-only row-count sources.
560#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
561pub enum ZeroColumnCountDisposition {
562    /// The plan reads at least one source column, so zero-column count is unused.
563    NotRequired,
564    /// The plan required zero-column count, but it was not proven safe.
565    Rejected,
566    /// The source represents row counts without reading physical columns.
567    Represented,
568}
569
570impl ZeroColumnCountDisposition {
571    /// Stable lower-case code used by reports.
572    pub const fn as_str(self) -> &'static str {
573        match self {
574            Self::NotRequired => "not_required",
575            Self::Rejected => "rejected",
576            Self::Represented => "represented",
577        }
578    }
579}
580
581/// Disposition of logical-to-source column mapping.
582#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
583pub enum SchemaMappingDisposition {
584    /// Logical columns map directly by name to file or partition columns.
585    DirectByName,
586    /// Nullable logical columns absent from the physical file are represented
587    /// as typed null columns.
588    MissingNullableColumns,
589    /// Required column mapping could not be represented.
590    Rejected,
591}
592
593impl SchemaMappingDisposition {
594    /// Stable lower-case code used by reports.
595    pub const fn as_str(self) -> &'static str {
596        match self {
597            Self::DirectByName => "direct_by_name",
598            Self::MissingNullableColumns => "missing_nullable_columns",
599            Self::Rejected => "rejected",
600        }
601    }
602}
603
604/// Stable rejection feature codes for native source diagnostics.
605#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
606pub enum RejectedSourceFeature {
607    /// Source is not backed by a local filesystem object store.
608    NonLocalObjectStore,
609    /// Required partition columns could not be represented.
610    PartitionColumns,
611    /// DataFusion scan expression adapter is present.
612    ExprAdapter,
613    /// A custom Parquet file reader factory is present.
614    ParquetFileReaderFactory,
615    /// DataFusion scan limit is present.
616    ScanLimit,
617    /// Source requires preserved input order.
618    PreserveOrder,
619    /// Source declares output ordering.
620    OutputOrdering,
621    /// Pushed-down Parquet predicate cannot be represented safely.
622    PushedDownParquetPredicate,
623    /// Source has no files.
624    EmptySource,
625    /// Required partition values are missing or unsupported.
626    PartitionValues,
627    /// Byte-range file entries are not supported as represented.
628    ByteRange,
629    /// Per-file ordering metadata is present.
630    FileOrdering,
631    /// Row-level selection is present.
632    RowSelection,
633    /// Zero-column count was required but not represented.
634    ZeroColumnCount,
635    /// Logical-to-source column mapping is invalid.
636    SchemaMapping,
637    /// A remote object's footer was not available to planning.
638    RemoteFooterUnavailable,
639}
640
641impl RejectedSourceFeature {
642    /// Stable lower-case code used by reports.
643    pub const fn as_str(self) -> &'static str {
644        match self {
645            Self::NonLocalObjectStore => "non_local_object_store",
646            Self::PartitionColumns => "partition_columns",
647            Self::ExprAdapter => "expr_adapter",
648            Self::ParquetFileReaderFactory => "parquet_file_reader_factory",
649            Self::ScanLimit => "scan_limit",
650            Self::PreserveOrder => "preserve_order",
651            Self::OutputOrdering => "output_ordering",
652            Self::PushedDownParquetPredicate => "pushed_down_parquet_predicate",
653            Self::EmptySource => "empty_source",
654            Self::PartitionValues => "partition_values",
655            Self::ByteRange => "byte_range",
656            Self::FileOrdering => "file_ordering",
657            Self::RowSelection => "row_selection",
658            Self::ZeroColumnCount => "zero_column_count",
659            Self::SchemaMapping => "schema_mapping",
660            Self::RemoteFooterUnavailable => "remote_footer_unavailable",
661        }
662    }
663}
664
665#[cfg(test)]
666mod tests {
667    use std::path::PathBuf;
668
669    use super::*;
670
671    #[test]
672    fn parquet_source_path_fact_redacts_uri_shaped_local_paths() {
673        let path = PathBuf::from("s3://secret-bucket/private/key.parquet?X-Amz-Signature=abc");
674        let fact = parquet_source_path_fact(&ParquetSourceLocation::local_path(&path), &path);
675
676        assert!(fact.starts_with("object_store:s3:fnv64:"));
677        assert!(!fact.contains("secret-bucket"));
678        assert!(!fact.contains("X-Amz-Signature"));
679    }
680
681    #[test]
682    fn parquet_source_path_fact_uses_object_store_diagnostic_location() {
683        let uri = "s3://secret-bucket/private/key.parquet?X-Amz-Signature=abc";
684        let location = ParquetSourceLocation::object_store_uri(uri).expect("valid object-store URI");
685        let fact = parquet_source_path_fact(&location, PathBuf::from(uri).as_path());
686
687        assert!(fact.starts_with("object_store:s3:fnv64:"));
688        assert!(!fact.contains("secret-bucket"));
689        assert!(!fact.contains("X-Amz-Signature"));
690    }
691
692    #[test]
693    fn local_parquet_file_row_group_metadata_count_uses_first_available_vector() {
694        let file = LocalParquetFile::new("lineitem.parquet")
695            .with_projected_row_group_bytes(vec![10, 20, 30], vec![100, 200, 300])
696            .with_row_group_rows(vec![1, 2, 3]);
697
698        assert_eq!(local_parquet_file_row_group_metadata_count(&file), Some(3));
699    }
700}
query_planning/source/diagnostics.rs

query_planning/source/
diagnostics.rs
