Skip to main content

datui_lib/widgets/
data_quality.rs

1use crate::analysis::analysis_modal::{
2    AnalysisFocus, AnalysisTool, DetailScroll, EvidenceRead, SetupRow,
3};
4use crate::analysis::data_quality::{
5    ColumnQualityProfile, CopyPlan, DataQualityPlan, DataQualityResults, IntervalClock,
6    IntervalFact, NoCopy, ObservationKind, QualityComparison, QualityCompute, QualityGrain,
7    QualityMetric, QualityPage, QualityPrecision, QualityScope, SegmentCount,
8    TemporalLatencyProfile, TemporalRole, interval_label, window_cadence,
9};
10use crate::analysis::quality_report::{
11    CHECKS_SHOWN, Check, Coverage, Evidence, EvidenceRows, FindingOrder, FindingsView, Outcome,
12    QualityReport, Severity, advice, coverage, describe, verdict,
13};
14use crate::analysis::quality_trends::{
15    GapCheck, GapKind, Gaps, TrendBar, TrendMeasure, TrendRow, TrendView, trend_view,
16};
17use crate::config::Theme;
18use crate::glyphs;
19use crate::numfmt;
20use crate::render::context::RenderContext;
21use crate::render::layout::dialog_in;
22use crate::table::DataTableState;
23use crate::widgets::ui::{FormValue, Picker, Surface};
24use polars::prelude::{DataType, Schema};
25use ratatui::buffer::Buffer;
26use ratatui::layout::{Alignment, Constraint, Direction, Layout, Rect};
27use ratatui::style::{Modifier, Style};
28use ratatui::text::{Line, Span};
29use ratatui::widgets::{Cell, Paragraph, Row, StatefulWidget, Table, TableState, Widget, Wrap};
30
31/// What Setup says about the draft beside its rows, all of it known without a read.
32#[derive(Debug, Clone, Default)]
33pub struct SetupView<'a> {
34    /// The columns a time role can be given: date and time columns, then text.
35    pub time_candidates: &'a [String],
36    /// A run would start from the rows the last run read.
37    pub reuses_sample: bool,
38    /// A random sample reads seeded runs of one file instead of streaming it.
39    pub reads_blocks: bool,
40    /// Where a run gets exact segment totals: see [`SegmentCount`].
41    pub segment_count: SegmentCount,
42    /// The rows this draft's sample reads were read earlier and released to the
43    /// memory budget.
44    pub released: bool,
45    /// The session cache holds this draft's report.
46    pub cached: bool,
47    /// The report on screen was measured with exactly this draft.
48    pub unchanged: bool,
49    /// The draft differs from the report on screen only in the windows it expects or
50    /// what its segments are compared with: Run works both out from the report.
51    pub relabel_only: bool,
52    /// Setup holds edits Esc would discard.
53    pub edited: bool,
54    /// Why Enter did not run.
55    pub note: Option<&'a str>,
56    /// A cancelled run still going, while the screen should say so.
57    pub cancelling: Option<Cancelling>,
58    /// The rows runs kept for reuse, which `d` releases.
59    pub kept: Option<KeptRows>,
60    /// How a full scan of a remote source reads it: see [`CopyPlan`].
61    pub copy: CopyPlan,
62    /// This dataset's local copy was released, so a Run fetches it again.
63    pub copy_released: bool,
64}
65
66/// The rows Data Quality's sampled runs kept this session, for later runs to reuse.
67#[derive(Debug, Clone, Copy, PartialEq, Eq)]
68pub struct KeptRows {
69    /// Samples kept: one per scope, method, size and seed a run read.
70    pub samples: usize,
71    pub rows: usize,
72    /// Near enough to budget by: [`crate::analysis::data_quality::QualitySample::estimated_bytes`].
73    pub bytes: usize,
74    /// Bytes on disk in a full scan's local copy of a remote source; 0 for none.
75    pub copy_bytes: u64,
76}
77
78impl KeptRows {
79    /// `100,000 rows kept · 12.4 MiB`, or `2 samples, 200,000 rows kept · …`, with
80    /// `, local copy · 17.3 MiB` after it when a full scan's copy is kept.
81    pub fn label(&self) -> String {
82        let middot = glyphs::get().middot;
83        let copy = format!(
84            "local copy {middot} {}",
85            crate::numfmt::bytes(self.copy_bytes)
86        );
87        if self.samples == 0 {
88            return copy;
89        }
90        let rows = format!(
91            "{} {} kept {middot} {}",
92            numfmt::group_chrome(self.rows),
93            if self.rows == 1 { "row" } else { "rows" },
94            crate::numfmt::bytes(self.bytes as u64)
95        );
96        let rows = if self.samples > 1 {
97            format!("{} samples, {rows}", numfmt::group_chrome(self.samples))
98        } else {
99            rows
100        };
101        if self.copy_bytes > 0 {
102            format!("{rows}, {copy}")
103        } else {
104            rows
105        }
106    }
107}
108
109/// A cancelled run that has not exited yet.
110#[derive(Debug, Clone, Copy)]
111pub struct Cancelling {
112    /// When the cancel came.
113    pub since: std::time::Instant,
114    /// The cancel came during a read of the source nothing can stop, which runs to
115    /// its end; otherwise the run is stopping and has outlasted its batch.
116    pub read_runs_out: bool,
117}
118
119impl Cancelling {
120    /// What is still going, after "Cancelling: " or "Run waits: ".
121    fn what(self) -> &'static str {
122        if self.read_runs_out {
123            "source read finishing"
124        } else {
125            "run stopping"
126        }
127    }
128}
129
130pub struct DataQualityWidgetConfig<'a> {
131    /// The clean entry's checks table shows every check, not the first few.
132    pub checks_expanded: bool,
133    pub state: &'a DataTableState,
134    pub plan: &'a DataQualityPlan,
135    /// The plan the result on screen was measured with; the header says this one
136    /// even while Setup edits another.
137    pub measured: &'a DataQualityPlan,
138    pub results: Option<&'a DataQualityResults>,
139    pub from_cache: bool,
140    pub metric: QualityMetric,
141    pub segment_index: usize,
142    /// The interval a detail shows.
143    pub interval_index: usize,
144    /// The Trends line a bar detail shows.
145    pub trend_line: usize,
146    /// The Expected editor, while it is open.
147    pub expected_form: Option<&'a crate::analysis::analysis_modal::ExpectedForm>,
148    pub segments_by_change: bool,
149    pub page: QualityPage,
150    pub setup: SetupView<'a>,
151    pub plan_field: usize,
152    pub show_access: bool,
153    pub observation_detail: bool,
154    /// How Overview narrows and orders the findings.
155    pub findings: &'a FindingsView,
156    /// The rows the report measured are still in memory: a finding's open from them.
157    pub rows_kept: bool,
158    /// A read of a finding's rows, waiting for Enter.
159    pub evidence_read: Option<&'a EvidenceRead>,
160    pub focus: AnalysisFocus,
161    pub theme: &'a Theme,
162    /// The dialogs' Surfaces and the table's number formatting.
163    pub ctx: &'a RenderContext,
164    /// One column's declared intent, being edited over the Column intent list.
165    pub intent_form: Option<&'a crate::analysis::intent_modal::IntentForm>,
166    /// The dialog writing the report on screen to a file.
167    pub export_form: Option<&'a crate::analysis::quality_export::ExportForm>,
168}
169
170/// The tool list's width: the width every analysis tool gives it, and none on a
171/// terminal too narrow to hold it beside the result (a picker pops up instead).
172fn data_quality_sidebar_width(width: u16) -> u16 {
173    if width >= 76 {
174        crate::widgets::analysis::sidebar_width(width)
175    } else {
176        0
177    }
178}
179
180/// Where the result goes: under the header and the page tabs, left of the tool list.
181/// The Sample form fills it before the first run.
182pub(crate) fn main_pane(area: Rect) -> Rect {
183    Rect {
184        y: area.y + 2,
185        height: area.height.saturating_sub(2),
186        width: area
187            .width
188            .saturating_sub(data_quality_sidebar_width(area.width)),
189        ..area
190    }
191}
192
193pub fn render(
194    config: DataQualityWidgetConfig<'_>,
195    table_state: &mut TableState,
196    sidebar_state: &mut TableState,
197    detail_scroll: &mut DetailScroll,
198    area: Rect,
199    buf: &mut Buffer,
200) {
201    let sidebar_width = data_quality_sidebar_width(area.width);
202    let vertical = Layout::default()
203        .direction(Direction::Vertical)
204        .constraints([
205            Constraint::Length(1),
206            Constraint::Length(1),
207            Constraint::Fill(1),
208        ])
209        .split(area);
210
211    render_header(&config, vertical[0], buf);
212    render_tabs(&config, vertical[1], buf);
213
214    let body = if sidebar_width > 0 {
215        let horizontal = Layout::default()
216            .direction(Direction::Horizontal)
217            .constraints([Constraint::Fill(1), Constraint::Length(sidebar_width)])
218            .split(vertical[2]);
219        render_sidebar(&config, sidebar_state, horizontal[1], buf);
220        horizontal[0]
221    } else {
222        vertical[2]
223    };
224
225    match config.page {
226        QualityPage::Setup => render_setup(&config, body, buf),
227        QualityPage::TimeRoles => render_time_roles(&config, table_state, body, buf),
228        QualityPage::IntervalPairs => render_interval_pairs(&config, table_state, body, buf),
229        QualityPage::Intent => {
230            crate::widgets::quality_intent::render_list(&config, table_state, body, buf)
231        }
232        QualityPage::Overview => render_overview(&config, table_state, body, buf),
233        QualityPage::Columns => render_columns(&config, table_state, body, buf),
234        QualityPage::Segments => render_segments(&config, table_state, body, buf),
235        QualityPage::SegmentDetail => {
236            render_segment_detail(&config, table_state, config.segment_index, body, buf)
237        }
238        QualityPage::Trends => render_trends(&config, table_state, body, buf),
239        QualityPage::TrendDetail => render_trend_detail(&config, table_state, body, buf),
240        QualityPage::Gaps => render_gaps(&config, table_state, body, buf),
241        QualityPage::ExpectedWindows => render_expected_windows(&config, body, buf),
242        QualityPage::Intervals => render_intervals(&config, table_state, body, buf),
243        QualityPage::IntervalDetail => render_interval_detail(&config, table_state, body, buf),
244        QualityPage::Detail => render_detail(&config, table_state, body, buf),
245    }
246
247    // An overlay owns the keys: what it covers takes no clicks.
248    if config.intent_form.is_some()
249        || config.export_form.is_some()
250        || config.show_access
251        || config.observation_detail
252    {
253        crate::app::pointer::record(area, crate::app::pointer::Hit::Modal);
254    }
255    if let Some(form) = config.intent_form {
256        crate::widgets::quality_intent::render_form(form, &config, area, buf);
257    } else if let Some(form) = config.export_form {
258        crate::widgets::quality_export::render(form, &config, area, buf);
259    } else if config.show_access {
260        render_access_plan(&config, area, buf);
261    } else if config.observation_detail {
262        render_finding_detail(&config, table_state, detail_scroll, area, buf);
263    } else if sidebar_width == 0 && config.focus == AnalysisFocus::Sidebar {
264        crate::app::pointer::record(area, crate::app::pointer::Hit::Modal);
265        render_narrow_tool_picker(&config, sidebar_state, area, buf);
266    }
267    // A staged read of rows sits over whatever asked for it: a finding or a count.
268    if let Some(read) = config.evidence_read {
269        crate::app::pointer::record(area, crate::app::pointer::Hit::Modal);
270        render_evidence_read(&config, read, area, buf);
271    }
272}
273
274/// One line, as every analysis tool heads its result: the tool, then what the
275/// numbers were measured on.
276fn render_header(config: &DataQualityWidgetConfig<'_>, area: Rect, buf: &mut Buffer) {
277    let middot = glyphs::get().middot;
278    let mut text = "Data Quality".to_string();
279    if let Some(results) = config.results.filter(|_| !config.page.is_setup()) {
280        let plan = config.measured;
281        text.push_str(&format!(" {middot} {}", measured_on(plan, results)));
282        if plan.grain != QualityGrain::Dataset {
283            text.push_str(&format!(" {middot} {}", plan.grain.label()));
284        }
285        if plan.comparison != QualityComparison::None {
286            text.push_str(&format!(
287                " {middot} compared with {}",
288                plan.comparison_label()
289            ));
290        }
291    }
292    let mut spans = vec![Span::raw(text)];
293    if config.from_cache && !config.page.is_setup() {
294        spans.push(Span::styled(
295            "  [session cache]",
296            Style::default().fg(config.theme.dimmed()),
297        ));
298    }
299    // State, not a message: it holds until the worker exits, on every page.
300    if let Some(cancelling) = config.setup.cancelling {
301        spans.push(Span::styled(
302            format!(
303                "  Cancelling: {} {} {}",
304                cancelling.what(),
305                glyphs::get().middot,
306                crate::numfmt::clock(cancelling.since.elapsed())
307            ),
308            Style::default().fg(config.theme.warning()),
309        ));
310    }
311    Paragraph::new(Line::from(spans))
312        .style(crate::widgets::analysis::header_style(
313            config.theme.controls_bg(),
314            config.theme.table_header(),
315        ))
316        .render(area, buf);
317}
318
319/// The rows a result was measured on, in the words the other tools' headers use.
320fn measured_on(plan: &DataQualityPlan, results: &DataQualityResults) -> String {
321    let sample = plan.sample();
322    let scope = if plan.scope == QualityScope::CurrentView {
323        String::new()
324    } else {
325        format!(" {} {}", glyphs::get().middot, plan.scope.label())
326    };
327    match (results.precision, results.total_rows) {
328        (QualityPrecision::Metadata, _) => format!("file metadata only, no values read{scope}"),
329        (QualityPrecision::Exact, _) => sample.outcome(results.evaluated_rows, None, None),
330        (QualityPrecision::Sampled, Some(total)) => {
331            sample.outcome(total, Some(results.evaluated_rows), results.per_value)
332        }
333        (QualityPrecision::Sampled, None) => format!(
334            "sample of {} rows{scope}",
335            numfmt::group_chrome(results.evaluated_rows)
336        ),
337    }
338}
339
340/// The pages, with the one shown carrying the accent. `←→` walk them. Setup is not
341/// one of them: it names itself alone, since the report's pages do not apply to it.
342fn render_tabs(config: &DataQualityWidgetConfig<'_>, area: Rect, buf: &mut Buffer) {
343    if config.page.is_setup() {
344        Paragraph::new(Line::styled(
345            "Setup",
346            Style::default()
347                .fg(config.theme.accent())
348                .add_modifier(Modifier::BOLD),
349        ))
350        .render(area, buf);
351        return;
352    }
353    let shown = config.page.tab();
354    let dimmed = Style::default().fg(config.theme.dimmed());
355    let accent = Style::default()
356        .fg(config.theme.accent())
357        .add_modifier(Modifier::BOLD);
358    // Each title padded a space each side, a space between: a click on one presses
359    // its number, as typed.
360    let mut spans = Vec::new();
361    let mut clicks = Vec::new();
362    for (i, page) in QualityPage::TABS.iter().enumerate() {
363        if i > 0 {
364            spans.push(Span::styled(" ", dimmed));
365        }
366        spans.push(Span::styled(" ", dimmed));
367        clicks.push((
368            spans.len(),
369            crate::app::pointer::Hit::Key(crossterm::event::KeyEvent::new(
370                crossterm::event::KeyCode::Char(char::from(b'1' + i as u8)),
371                crossterm::event::KeyModifiers::NONE,
372            )),
373        ));
374        spans.push(Span::styled(
375            page.title(),
376            if *page == shown { accent } else { dimmed },
377        ));
378        spans.push(Span::styled(" ", dimmed));
379    }
380    let line = Line::from(spans);
381    crate::app::pointer::record_spans(area, &line, clicks);
382    Paragraph::new(line).style(dimmed).render(area, buf);
383}
384/// One line of Setup, top to bottom.
385enum SetupLine {
386    Rule(&'static str, Option<String>),
387    Row(SetupRow),
388    Note(String, bool),
389}
390
391/// Data Quality Setup: every run setting in four sections, and what the run will read.
392/// Reads nothing (only schema, on-screen rows and kept runs); edits wait for Enter.
393fn render_setup(config: &DataQualityWidgetConfig<'_>, area: Rect, buf: &mut Buffer) {
394    let plan = config.plan;
395    let view = &config.setup;
396    let ctx = config.ctx;
397    // A short terminal gives up the top margin before it gives up a row.
398    let top = u16::from(area.height >= 20);
399    let area = Rect {
400        x: area.x + 1,
401        y: area.y + top,
402        width: area.width.saturating_sub(2),
403        height: area.height.saturating_sub(top),
404    };
405    if area.height < 3 || area.width < 20 {
406        return;
407    }
408    let width = area.width as usize;
409    let schema = config.state.quality_schema(&plan.scope);
410
411    let mut lines = vec![
412        SetupLine::Rule("Rows & sample", None),
413        SetupLine::Row(SetupRow::Sample),
414    ];
415    let texts = config.state.quality_text_columns(&plan.scope).len();
416    let times = view.time_candidates.len().saturating_sub(texts);
417    let counts = [(times, "date or time"), (texts, "text")]
418        .into_iter()
419        .filter(|(count, _)| *count > 0)
420        .map(|(count, kind)| format!("{} {kind}", numfmt::group_chrome(count)))
421        .collect::<Vec<_>>()
422        .join(&format!(" {} ", glyphs::get().middot));
423    lines.push(SetupLine::Rule(
424        "Columns",
425        (!counts.is_empty()).then_some(counts),
426    ));
427    lines.push(SetupLine::Row(SetupRow::TextAsTime));
428    lines.push(SetupLine::Row(SetupRow::TimeRoles));
429    lines.push(SetupLine::Row(SetupRow::Intervals));
430    lines.push(SetupLine::Row(SetupRow::Intent));
431    for (note, warn) in column_notes(plan, schema) {
432        for line in crate::widgets::info::wrap_to(&note, width.saturating_sub(2)) {
433            lines.push(SetupLine::Note(line, warn));
434        }
435    }
436    lines.push(SetupLine::Rule("Study", None));
437    for row in [
438        SetupRow::Grain,
439        SetupRow::Expected,
440        SetupRow::Compare,
441        SetupRow::Values,
442        SetupRow::Latency,
443        SetupRow::WindowBy,
444    ] {
445        lines.push(SetupLine::Row(row));
446    }
447    lines.push(SetupLine::Rule("Read", view.kept.map(|kept| kept.label())));
448    let read_start = lines.len();
449    for note in read_lines(config) {
450        for line in crate::widgets::info::wrap_to(&note, width.saturating_sub(2)) {
451            lines.push(SetupLine::Note(line, false));
452        }
453    }
454
455    // The status line keeps the bottom row: why Run waits, or that it would
456    // replace the report on screen.
457    let status = setup_status(config);
458    let body_height = area.height.saturating_sub(1) as usize;
459    // Scrolled only as far as the focused row needs; the read summary, last, is
460    // what gives way, and says how much of it is off screen.
461    let focused = lines
462        .iter()
463        .position(
464            |line| matches!(line, SetupLine::Row(row) if *row == SetupRow::at(config.plan_field)),
465        )
466        .unwrap_or(0);
467    let (offset, room) = if lines.len() <= body_height {
468        (0, body_height)
469    } else {
470        // One line counts what is cut, unless scrolling reaches the end anyway.
471        let room = body_height.saturating_sub(1);
472        let offset = (focused + 1).saturating_sub(room);
473        if offset + body_height >= lines.len() {
474            (lines.len() - body_height, body_height)
475        } else {
476            (offset, room)
477        }
478    };
479    let end = (offset + room).min(lines.len());
480    let cut = lines.len() - end;
481    let dimmed = Style::default().fg(ctx.dimmed);
482    let warning = Style::default().fg(ctx.warning);
483    if cut > 0 {
484        Paragraph::new(Line::styled(
485            format!(
486                "  {} {cut} more{}",
487                glyphs::get().ellipsis,
488                if end >= read_start { " (p)" } else { "" }
489            ),
490            dimmed,
491        ))
492        .render(
493            Rect {
494                y: area.y + room as u16,
495                height: 1,
496                ..area
497            },
498            buf,
499        );
500    }
501    for (index, line) in lines.iter().enumerate().take(end).skip(offset) {
502        let row_area = Rect {
503            y: area.y + (index - offset) as u16,
504            height: 1,
505            ..area
506        };
507        match line {
508            SetupLine::Rule(title, chip) => crate::widgets::ui::SectionRule {
509                title,
510                chip: chip.as_deref(),
511            }
512            .render(row_area, buf, ctx),
513            SetupLine::Row(row) => {
514                let (value, placeholder) = setup_value(config, *row);
515                let room = (row_area.width as usize).saturating_sub(1 + SETUP_LABEL_WIDTH as usize);
516                let value = glyphs::fit(&value, room);
517                crate::widgets::ui::FormRow {
518                    label: row.label(),
519                    value: if placeholder {
520                        FormValue::Placeholder(&value)
521                    } else {
522                        FormValue::Choice(&value)
523                    },
524                    focused: config.focus == AnalysisFocus::Main
525                        && SetupRow::at(config.plan_field) == *row,
526                    label_width: SETUP_LABEL_WIDTH,
527                }
528                .render(row_area, buf, ctx);
529            }
530            SetupLine::Note(text, warn) => {
531                Paragraph::new(Line::styled(
532                    format!("  {text}"),
533                    if *warn { warning } else { dimmed },
534                ))
535                .render(row_area, buf);
536            }
537        }
538    }
539    if let Some((text, warn)) = status {
540        Paragraph::new(Line::styled(text, if warn { warning } else { dimmed })).render(
541            Rect {
542                y: area.y + area.height - 1,
543                height: 1,
544                ..area
545            },
546            buf,
547        );
548    }
549}
550
551/// Where Setup's values start, past the rail gutter and the longest label.
552const SETUP_LABEL_WIDTH: u16 = 15;
553
554/// A Setup row's value, and whether it is a placeholder: a choice not made, or one
555/// that has nothing to apply to yet.
556fn setup_value(config: &DataQualityWidgetConfig<'_>, row: SetupRow) -> (String, bool) {
557    let plan = config.plan;
558    match row {
559        SetupRow::Sample => {
560            // The view's rows, when counted: a sample never promises more.
561            let sample = plan.sample();
562            let known = (sample.scope == crate::analysis::data_quality::QualityScope::CurrentView)
563                .then(|| config.state.num_rows_if_valid())
564                .flatten();
565            (sample.summary_within(known), false)
566        }
567        SetupRow::TextAsTime if plan.time_formats.is_empty() => {
568            ("none: every text column is text".to_string(), true)
569        }
570        // The columns here, each one's format on its own line below.
571        SetupRow::TextAsTime => (
572            plan.time_formats
573                .iter()
574                .map(|format| format.column.as_str())
575                .collect::<Vec<_>>()
576                .join(", "),
577            false,
578        ),
579        SetupRow::TimeRoles if config.setup.time_candidates.is_empty() => {
580            ("none: no date, time or text columns".to_string(), true)
581        }
582        SetupRow::TimeRoles if plan.temporal_roles.is_empty() => ("none".to_string(), true),
583        SetupRow::TimeRoles => (
584            plan.temporal_roles
585                .iter()
586                .map(|role| format!("{} = {}", role.role.label(), role.column))
587                .collect::<Vec<_>>()
588                .join(", "),
589            false,
590        ),
591        SetupRow::Intervals if plan.candidate_pairs().is_empty() => {
592            ("needs two time roles".to_string(), true)
593        }
594        SetupRow::Intervals if plan.interval_pairs().is_empty() => ("none".to_string(), true),
595        SetupRow::Intervals => (
596            plan.interval_pairs()
597                .into_iter()
598                .map(interval_label)
599                .collect::<Vec<_>>()
600                .join(", "),
601            false,
602        ),
603        SetupRow::Intent if plan.intent.is_empty() => ("none".to_string(), true),
604        SetupRow::Intent => (plan.intent.summary(), false),
605        SetupRow::Grain => (plan.grain.label(), false),
606        SetupRow::Expected => match (&plan.grain, plan.expected.as_ref()) {
607            (QualityGrain::TimeWindows { every, .. }, Some(expected)) => (
608                format!(
609                    "{}, {}",
610                    expected.cadence_label(every),
611                    expected.range_label()
612                ),
613                false,
614            ),
615            (QualityGrain::TimeWindows { .. }, None) => {
616                ("none: no window is a gap".to_string(), true)
617            }
618            _ => ("needs a time-window grain".to_string(), true),
619        },
620        SetupRow::Compare => (
621            match (plan.comparison, plan.baseline_segment.as_deref()) {
622                (QualityComparison::Baseline, Some(segment)) => format!("baseline {segment}"),
623                (comparison, _) => comparison.choice_label().to_string(),
624            },
625            false,
626        ),
627        SetupRow::Values if plan.compute == QualityCompute::Metadata => {
628            ("file metadata only".to_string(), false)
629        }
630        SetupRow::Values => ("read".to_string(), false),
631        SetupRow::Latency if plan.interval_pairs().is_empty() => {
632            ("needs an interval to measure".to_string(), true)
633        }
634        SetupRow::Latency => (
635            crate::analysis::analysis_modal::threshold_label(plan.latency_threshold_seconds),
636            plan.latency_threshold_seconds.is_none(),
637        ),
638        SetupRow::WindowBy => match (&plan.grain, plan.interval_clock) {
639            (QualityGrain::TimeWindows { .. }, _) if plan.interval_pairs().is_empty() => {
640                ("needs an interval".to_string(), true)
641            }
642            (QualityGrain::TimeWindows { column, .. }, IntervalClock::Grain) => {
643                (format!("the grain's column, {column}"), false)
644            }
645            (QualityGrain::TimeWindows { .. }, clock) => (clock.label().to_string(), false),
646            _ => ("needs a time-window grain".to_string(), true),
647        },
648    }
649}
650
651/// What the columns section says under its rows: text that needs a format before
652/// it can be read as time, and which intervals the roles measure, or why none.
653fn column_notes(plan: &DataQualityPlan, schema: &Schema) -> Vec<(String, bool)> {
654    let mut notes = plan
655        .time_formats
656        .iter()
657        .map(|format| {
658            (
659                format!("{} read as {}", format.column, format.label()),
660                false,
661            )
662        })
663        .collect::<Vec<_>>();
664    let mut unread = Vec::new();
665    let grain = match &plan.grain {
666        QualityGrain::TimeWindows { column, .. } => Some(column.as_str()),
667        _ => None,
668    };
669    for column in plan
670        .temporal_roles
671        .iter()
672        .map(|role| role.column.as_str())
673        .chain(grain)
674    {
675        if !plan.reads_as_time(column, schema) && !unread.contains(&column) {
676            unread.push(column);
677        }
678    }
679    for column in unread {
680        notes.push((
681            format!(
682                "{column}: text, no format {} set Text as time",
683                glyphs::get().middot
684            ),
685            true,
686        ));
687    }
688    // A role that is in no interval measures nothing; say which, and where to fix it.
689    let unpaired = plan.unpaired_roles();
690    if plan.temporal_roles.len() == 1 {
691        notes.push((
692            crate::glyphs::dotted("1 role, no interval · add one under Time roles"),
693            true,
694        ));
695    } else if !unpaired.is_empty() {
696        notes.push((
697            format!(
698                "In no interval: {} {} set Intervals",
699                unpaired
700                    .iter()
701                    .map(|role| role.label())
702                    .collect::<Vec<_>>()
703                    .join(", "),
704                glyphs::get().middot
705            ),
706            true,
707        ));
708    }
709    // An interval from a time with no zone to an instant reads the first as UTC.
710    let mut naive = Vec::new();
711    for (start, end) in plan.interval_pairs() {
712        let (Some(start), Some(end)) = (plan.role_column(start), plan.role_column(end)) else {
713            continue;
714        };
715        for (column, other) in [(start, end), (end, start)] {
716            if plan.zoned(column, schema) == Some(false)
717                && plan.zoned(other, schema) == Some(true)
718                && !naive.contains(&column)
719            {
720                naive.push(column);
721            }
722        }
723    }
724    if !naive.is_empty() {
725        notes.push((
726            format!("No time zone, read as UTC: {}", naive.join(", ")),
727            false,
728        ));
729    }
730    // Intent on a column the scope does not have measures nothing.
731    let absent = plan
732        .intent
733        .declared_columns()
734        .into_iter()
735        .filter(|column| schema.get(column).is_none())
736        .collect::<Vec<_>>();
737    if !absent.is_empty() {
738        notes.push((
739            format!("Not in scope, not checked: {}", absent.join(", ")),
740            true,
741        ));
742    }
743    notes
744}
745
746/// What Run will read, in the order it happens, before it happens.
747fn read_lines(config: &DataQualityWidgetConfig<'_>) -> Vec<String> {
748    let plan = config.plan;
749    let view = &config.setup;
750    let state = config.state;
751    let mut lines = Vec::new();
752    if view.unchanged {
753        lines.push(crate::glyphs::dotted(
754            "Report on screen: this setup · no read",
755        ));
756        return lines;
757    }
758    if view.relabel_only {
759        let compare = config.measured.compares_differently(plan);
760        let expected = config.measured.expected != plan.expected;
761        lines.push(glyphs::dotted(match (compare, expected) {
762            (true, true) => "Changed: Compare, Expected · no read",
763            (true, false) => "Changed: Compare · no read",
764            _ => "Changed: Expected · no read",
765        }));
766        return lines;
767    }
768    if view.cached {
769        lines.push(crate::glyphs::dotted("Session cache: this setup · no read"));
770        return lines;
771    }
772    let scope_rows = planned_scope_rows(state, plan);
773    let n = numfmt::group_chrome(plan.dataset_rows);
774    match plan.compute {
775        QualityCompute::Metadata => {
776            lines.push("File metadata only: no values read".to_string());
777            return lines;
778        }
779        QualityCompute::Full => {
780            lines.extend(copy_lines(view, full_passes(config)));
781            // Each column an interval is windowed by is a grouping of its own.
782            let clocks = crate::analysis::data_quality::interval_passes(
783                plan,
784                state.quality_schema(&plan.scope),
785            );
786            if clocks > 1 {
787                lines.push(format!(
788                    "Window by {}: {clocks} of those passes, 1 per column",
789                    plan.interval_clock.label()
790                ));
791            }
792        }
793        QualityCompute::Sample if view.reuses_sample => {
794            lines.push(crate::glyphs::dotted(
795                "Rows: from an earlier run · no source read",
796            ));
797        }
798        QualityCompute::Sample => lines.push(match &plan.method {
799            crate::analysis::sampling::SampleMethod::FirstRows => {
800                format!("First {n} rows of the scope")
801            }
802            crate::analysis::sampling::SampleMethod::PerPartition { column } => format!(
803                "1 pass over every eligible row {} {n} kept per {column}",
804                glyphs::get().middot
805            ),
806            _ if scope_rows.is_some_and(|rows| rows <= plan.dataset_rows) => {
807                crate::glyphs::dotted("Every row · scope no larger than the sample")
808            }
809            _ if view.reads_blocks => crate::glyphs::dotted(&format!(
810                "Seeded runs of the file · about {n} rows, no full pass"
811            )),
812            _ => crate::glyphs::dotted(&format!(
813                "1 streaming pass over every eligible row · seeded {n} kept"
814            )),
815        }),
816    }
817    if plan.compute == QualityCompute::Sample && !view.reuses_sample && view.released {
818        lines.push(crate::glyphs::dotted(
819            "Released since last read · read again",
820        ));
821    }
822    let exact = scope_rows.is_some_and(|rows| rows <= plan.dataset_rows);
823    if plan.compute == QualityCompute::Sample && !exact {
824        let column = match &plan.grain {
825            QualityGrain::Partition(column) | QualityGrain::TimeWindows { column, .. } => {
826                column.as_str()
827            }
828            _ => "",
829        };
830        match &view.segment_count {
831            SegmentCount::CountPass => lines.push(format!(
832                "+1 count of {column} {} exact segment totals, kept",
833                glyphs::get().middot
834            )),
835            SegmentCount::InSamplePass => lines.push(format!(
836                "Segment totals: exact, counted by {column} in that pass"
837            )),
838            SegmentCount::Retained => lines.push(crate::glyphs::dotted(
839                "Segment totals: from an earlier count · no read",
840            )),
841            SegmentCount::RolledUp(finer) => lines.push(format!(
842                "Segment totals: summed from earlier {} counts",
843                window_cadence(finer)
844            )),
845            SegmentCount::TooMany => lines.push(format!(
846                "Too many segments {} to count {} choose a coarser grain",
847                plan.grain.label(),
848                glyphs::get().middot
849            )),
850            SegmentCount::NotNeeded | SegmentCount::PerValue => {}
851        }
852    }
853    if plan.compute == QualityCompute::Sample {
854        lines.push(crate::glyphs::dotted(
855            "Measured in memory · no further reads",
856        ));
857    }
858    lines.extend(intent_read_lines(plan, exact));
859    // Gaps come from the segment counts the run already takes, never a read of
860    // their own.
861    if plan.expected_windows().is_some() && plan.compute != QualityCompute::Metadata {
862        lines.push(crate::glyphs::dotted(
863            "Expected windows: from the segment counts · no read",
864        ));
865    }
866    let rows = planned_rows(state, plan)
867        .map(numfmt::group_chrome)
868        .unwrap_or_else(|| "unknown".to_string());
869    let sampled = plan.compute == QualityCompute::Sample;
870    let read = if sampled && view.reuses_sample && view.segment_count.reads() && !exact {
871        "the grain's column, for the count".to_string()
872    } else if sampled && view.reuses_sample {
873        "none".to_string()
874    } else if let Some(transfer) = copy_transfer(view.copy, state) {
875        transfer
876    } else if state.is_remote_source() {
877        "unknown".to_string()
878    } else if sampled
879        && view.reads_blocks
880        && !exact
881        && matches!(plan.method, crate::analysis::sampling::SampleMethod::Spread)
882    {
883        // A ceiling of the whole file would say more than the runs read.
884        "the row groups the runs fall in".to_string()
885    } else {
886        planned_read_label(state, plan)
887    };
888    lines.push(format!(
889        "Rows evaluated {rows} {} {} {read}",
890        glyphs::get().middot,
891        if state.is_remote_source() {
892            "remote transfer"
893        } else {
894            "local read"
895        }
896    ));
897    lines
898}
899
900/// A full scan's first lines in Read: one pass per check, over the source, over a
901/// local copy fetched first, or over one fetched earlier, and why no copy where none.
902fn copy_lines(view: &SetupView<'_>, passes: usize) -> Vec<String> {
903    let bytes = crate::numfmt::bytes;
904    let over_source = crate::glyphs::dotted(&format!(
905        "Every eligible row · up to {passes} passes over the source, 1 per check"
906    ));
907    match view.copy {
908        CopyPlan::NotApplicable => vec![crate::glyphs::dotted(&format!(
909            "Every eligible row · up to {passes} passes over the scope, 1 per check"
910        ))],
911        CopyPlan::Fetch {
912            bytes: size,
913            objects,
914        } => {
915            let mut lines = vec![
916                crate::glyphs::dotted(&format!(
917                    "1 fetch of {} ({size}) to a local copy · up to {passes} passes over it",
918                    objects_label(objects),
919                    size = bytes(size)
920                )),
921                crate::glyphs::dotted("Local copy kept for later full scans · d releases"),
922            ];
923            if view.copy_released {
924                lines.push(crate::glyphs::dotted(
925                    "Released since last copy · fetched again",
926                ));
927            }
928            lines
929        }
930        CopyPlan::Kept { bytes: size, .. } => vec![crate::glyphs::dotted(&format!(
931            "Every eligible row · up to {passes} passes over the local copy ({}) · no source read",
932            bytes(size)
933        ))],
934        CopyPlan::Passes(why) => vec![
935            over_source,
936            match why {
937                NoCopy::Off => "No local copy: quality_local_copy = 0".to_string(),
938                NoCopy::SizeUnknown => "No local copy: object sizes unknown".to_string(),
939                NoCopy::Unusable => "No local copy: copy did not read as the source".to_string(),
940                NoCopy::PartOfTheSource => {
941                    "No local copy: scope reads part of the source".to_string()
942                }
943                NoCopy::TooLarge { bytes: size, limit } => format!(
944                    "No local copy: {} over the {} limit",
945                    bytes(size),
946                    bytes(limit)
947                ),
948                NoCopy::NoRoom {
949                    bytes: size,
950                    free: Some(free),
951                } => format!(
952                    "No local copy: {} needed, {} free on disk",
953                    bytes(size),
954                    bytes(free)
955                ),
956                NoCopy::NoRoom { free: None, .. } => {
957                    "No local copy: free disk space unknown".to_string()
958                }
959            },
960        ],
961    }
962}
963
964fn objects_label(objects: usize) -> String {
965    if objects == 1 {
966        "1 object".to_string()
967    } else {
968        format!("{} objects", numfmt::group_chrome(objects))
969    }
970}
971
972/// What a full scan takes from a remote source when it reads a local copy: the
973/// objects once, or nothing. `None` when its passes read the source.
974fn copy_transfer(copy: CopyPlan, state: &DataTableState) -> Option<String> {
975    let conflicts = if state.quality_conflict_reads() > 0 {
976        ", plus the conflict reads"
977    } else {
978        ""
979    };
980    match copy {
981        CopyPlan::Fetch { bytes, .. } => Some(format!(
982            "{}, one request per object{conflicts}",
983            crate::numfmt::bytes(bytes)
984        )),
985        CopyPlan::Kept { .. } if conflicts.is_empty() => Some("none".to_string()),
986        CopyPlan::Kept { .. } => Some("only the conflict reads".to_string()),
987        _ => None,
988    }
989}
990
991/// What declared intent costs, said before Run: no extra rows, but a key on a sample
992/// covers only sampled rows, and on a full scan is a grouping of its own.
993fn intent_read_lines(plan: &DataQualityPlan, every_row: bool) -> Vec<String> {
994    if plan.intent.is_empty() {
995        return Vec::new();
996    }
997    let key = !plan.intent.key.is_empty();
998    match plan.compute {
999        QualityCompute::Metadata => vec!["Column intent: not checked, needs values".to_string()],
1000        QualityCompute::Full if key => {
1001            vec![crate::glyphs::dotted(
1002                "Column intent: in the profile pass · key adds 1 pass",
1003            )]
1004        }
1005        QualityCompute::Full => {
1006            vec![crate::glyphs::dotted(
1007                "Column intent: in the profile pass · no extra pass",
1008            )]
1009        }
1010        QualityCompute::Sample => {
1011            let mut lines = vec![crate::glyphs::dotted(
1012                "Column intent: on the rows read · no extra read",
1013            )];
1014            if key && !every_row {
1015                lines.push(crate::glyphs::dotted(&format!(
1016                    "Key: repeats among the {} sampled rows only · Every row checks all",
1017                    numfmt::group_chrome(plan.dataset_rows)
1018                )));
1019            }
1020            lines
1021        }
1022    }
1023}
1024
1025/// Collects a full run makes over its scope: one per check, plus a count first when the
1026/// scope's size is unknown.
1027fn full_passes(config: &DataQualityWidgetConfig<'_>) -> usize {
1028    let plan = config.plan;
1029    let state = config.state;
1030    let texts = state
1031        .quality_schema(&plan.scope)
1032        .iter()
1033        .filter(|(_, dtype)| matches!(dtype, DataType::String | DataType::Categorical(..)))
1034        .count();
1035    // Profile, most common values, duplicates, and at most one for shared nulls.
1036    let mut passes = 4 + texts;
1037    if !matches!(plan.grain, QualityGrain::Dataset) {
1038        passes += 1;
1039    }
1040    passes +=
1041        crate::analysis::data_quality::interval_passes(plan, state.quality_schema(&plan.scope));
1042    // The declared key is a grouping of its columns: one pass of its own.
1043    passes += usize::from(!plan.intent.key.is_empty());
1044    if planned_scope_rows(state, plan).is_none() {
1045        passes += 1;
1046    }
1047    passes + state.quality_conflict_reads()
1048}
1049
1050/// Setup's bottom line: a cancelled read still running, why Enter did not run, or
1051/// that the draft differs from the report it would replace.
1052fn setup_status(config: &DataQualityWidgetConfig<'_>) -> Option<(String, bool)> {
1053    let view = &config.setup;
1054    if let Some(cancelling) = view.cancelling {
1055        // Said as a reason once Enter has been refused for it, short enough to keep
1056        // its clock beside the tool list at 80 columns.
1057        return Some((
1058            format!(
1059                "{}{} {} {}",
1060                if view.note.is_some() {
1061                    "Run waits: "
1062                } else {
1063                    "Cancelling: "
1064                },
1065                cancelling.what(),
1066                glyphs::get().middot,
1067                crate::numfmt::clock(cancelling.since.elapsed())
1068            ),
1069            true,
1070        ));
1071    }
1072    if let Some(note) = view.note {
1073        return Some((note.to_string(), true));
1074    }
1075    if view.edited {
1076        return Some((
1077            crate::glyphs::dotted("Edited · Enter runs, Esc discards"),
1078            false,
1079        ));
1080    }
1081    None
1082}
1083
1084fn render_overview(
1085    config: &DataQualityWidgetConfig<'_>,
1086    table_state: &mut TableState,
1087    area: Rect,
1088    buf: &mut Buffer,
1089) {
1090    let Some(results) = config.results else {
1091        render_run_prompt(area, config.theme, buf);
1092        return;
1093    };
1094    let report = results.report();
1095    let all_checks = results.checks();
1096    let notes = config.state.notes();
1097    let notes_height = if notes.is_empty() {
1098        0
1099    } else {
1100        (notes.len() as u16).min(3) + 2
1101    };
1102    // Coverage sits under the verdict on every report, clean or not: four lines
1103    // where there is room, two on a short terminal.
1104    let coverage = coverage_lines(
1105        &coverage(results, all_checks, config.measured),
1106        area.width.saturating_sub(2),
1107        if area.height >= 16 { 4 } else { 2 },
1108        config.theme,
1109    );
1110    let sections = Layout::default()
1111        .direction(Direction::Vertical)
1112        .constraints([
1113            Constraint::Length(2 + coverage.len() as u16),
1114            Constraint::Length(notes_height),
1115            Constraint::Fill(1),
1116        ])
1117        .horizontal_margin(1)
1118        .vertical_margin(1)
1119        .split(area);
1120    render_verdict(config, report, sections[0], buf);
1121    Paragraph::new(coverage).render(
1122        Rect {
1123            y: sections[0].y + 1,
1124            height: sections[0].height.saturating_sub(1),
1125            ..sections[0]
1126        },
1127        buf,
1128    );
1129    if !notes.is_empty() {
1130        let mut lines = vec![rule_line(
1131            "Dataset notes",
1132            Some(&numfmt::group_chrome(notes.len())),
1133            sections[1].width,
1134            config.theme,
1135        )];
1136        lines.extend(notes.iter().take(3).map(|note| {
1137            Line::raw(format!(
1138                "{} {} {}",
1139                note.summary,
1140                glyphs::get().dash,
1141                note.scope
1142            ))
1143        }));
1144        Paragraph::new(lines).render(sections[1], buf);
1145    }
1146
1147    let mut list = sections[2];
1148    if report.findings.is_empty() {
1149        let message = if report.metadata_only {
1150            "No values read · e Setup"
1151        } else if report.no_rows {
1152            "No rows in scope · e Setup"
1153        } else {
1154            "No columns to check"
1155        };
1156        Paragraph::new(glyphs::dotted(message))
1157            .wrap(Wrap { trim: true })
1158            .style(Style::default().fg(config.theme.dimmed()))
1159            .render(list, buf);
1160        return;
1161    }
1162    let shown = config.findings.shown(report);
1163    // Narrowed or reordered, the list says so above itself, and what it holds.
1164    let facets = config.findings.describe();
1165    if !facets.is_empty() && list.height > 1 {
1166        let total = report
1167            .findings
1168            .iter()
1169            .filter(|finding| finding.kind.is_some())
1170            .count();
1171        let listed = shown
1172            .iter()
1173            .filter(|index| report.findings[**index].kind.is_some())
1174            .count();
1175        let middot = glyphs::get().middot;
1176        let mut text = if config.findings.narrowed() {
1177            format!("{listed} of {total} findings")
1178        } else {
1179            format!(
1180                "{total} {}",
1181                if total == 1 { "finding" } else { "findings" }
1182            )
1183        };
1184        for facet in &facets {
1185            text.push_str(&format!(" {middot} {facet}"));
1186        }
1187        Paragraph::new(Line::styled(
1188            glyphs::fit(&text, list.width as usize),
1189            Style::default().fg(config.theme.dimmed()),
1190        ))
1191        .render(Rect { height: 1, ..list }, buf);
1192        list.y += 1;
1193        list.height -= 1;
1194    }
1195    normalize_selection(table_state, shown.len());
1196    if shown.is_empty() {
1197        Paragraph::new(glyphs::dotted("No findings match · Esc shows all"))
1198            .wrap(Wrap { trim: true })
1199            .style(Style::default().fg(config.theme.dimmed()))
1200            .render(list, buf);
1201        return;
1202    }
1203    if report.problems == 0 && report.notes == 0 && !report.metadata_only && !report.no_rows {
1204        // Nothing to fix: what was checked is the answer, so it is on the page
1205        // rather than behind the clean entry.
1206        let parts = Layout::default()
1207            .direction(Direction::Vertical)
1208            .constraints([Constraint::Length(3), Constraint::Fill(1)])
1209            .split(list);
1210        render_findings(config, report, table_state, parts[0], buf);
1211        Paragraph::new(check_lines(all_checks, parts[1].width, None, config.theme))
1212            .render(parts[1], buf);
1213        return;
1214    }
1215    render_findings(config, report, table_state, list, buf);
1216}
1217
1218/// The answer before the evidence: a mark and the counts. What they were measured on
1219/// is the strip above.
1220fn render_verdict(
1221    config: &DataQualityWidgetConfig<'_>,
1222    report: &QualityReport,
1223    area: Rect,
1224    buf: &mut Buffer,
1225) {
1226    let g = glyphs::get();
1227    let theme = config.theme;
1228    let (mark, tone) = if report.problems > 0 {
1229        (g.warning, theme.warning())
1230    } else if report.metadata_only || report.no_rows {
1231        (g.middot, theme.dimmed())
1232    } else {
1233        (g.check, theme.success())
1234    };
1235    Paragraph::new(Line::from(vec![
1236        Span::styled(format!("{mark} "), Style::default().fg(tone)),
1237        Span::styled(
1238            verdict(report),
1239            Style::default().fg(tone).add_modifier(Modifier::BOLD),
1240        ),
1241    ]))
1242    .render(area, buf);
1243}
1244
1245/// The label gutter of the coverage lines, as wide as its longest label and a gap.
1246const COVERAGE_LABEL: usize = 8;
1247
1248/// The coverage under the verdict (checks, rows, limits), each a labeled line wrapping
1249/// between facts. Over `max_lines`, rows go first (the header states them); a cut
1250/// section counts what it omitted.
1251fn coverage_lines(
1252    coverage: &Coverage,
1253    width: u16,
1254    max_lines: usize,
1255    theme: &Theme,
1256) -> Vec<Line<'static>> {
1257    let mut sections = [
1258        ("Checks", coverage.checks()),
1259        ("Rows", coverage.rows.clone()),
1260        ("Limits", coverage.limits()),
1261    ]
1262    .into_iter()
1263    .filter(|(_, facts)| !facts.is_empty())
1264    .collect::<Vec<_>>();
1265    if sections.len() > max_lines {
1266        sections.retain(|(label, _)| *label != "Rows");
1267    }
1268    sections.truncate(max_lines);
1269    let text_width = (width as usize).saturating_sub(COVERAGE_LABEL).max(1);
1270    let dimmed = Style::default().fg(theme.dimmed());
1271    let plain = Style::default().fg(theme.text_primary());
1272    let mut lines = Vec::new();
1273    for (index, (label, facts)) in sections.iter().enumerate() {
1274        // Every later section keeps one line; this one may take the rest.
1275        let later = sections.len() - index - 1;
1276        let room = max_lines.saturating_sub(lines.len() + later).max(1);
1277        for (row, text) in pack_facts(facts, text_width, room).into_iter().enumerate() {
1278            let gutter = if row == 0 {
1279                format!("{label:<COVERAGE_LABEL$}")
1280            } else {
1281                " ".repeat(COVERAGE_LABEL)
1282            };
1283            lines.push(Line::from(vec![
1284                Span::styled(gutter, dimmed),
1285                Span::styled(text, plain),
1286            ]));
1287        }
1288    }
1289    lines
1290}
1291
1292/// `facts` joined by middots into at most `room` lines of `width`, breaking between
1293/// facts. What does not fit is counted at the end of the last line: "+2 more".
1294fn pack_facts(facts: &[String], width: usize, room: usize) -> Vec<String> {
1295    let sep = format!(" {} ", glyphs::get().middot);
1296    let join = |line: &[String]| line.join(&sep);
1297    let mut lines: Vec<Vec<String>> = vec![Vec::new()];
1298    let mut placed = 0;
1299    for fact in facts {
1300        let fact = glyphs::fit(fact, width);
1301        let current = lines.last_mut().expect("one line at least");
1302        let mut joined = current.clone();
1303        joined.push(fact.clone());
1304        if current.is_empty() || glyphs::display_width(&join(&joined)) <= width {
1305            *current = joined;
1306        } else if lines.len() < room {
1307            lines.push(vec![fact]);
1308        } else {
1309            break;
1310        }
1311        placed += 1;
1312    }
1313    let left = facts.len() - placed;
1314    let last = lines.last_mut().expect("one line at least");
1315    if left > 0 {
1316        // Make room for the count by giving up facts from the end of the line.
1317        let mut dropped = left;
1318        loop {
1319            let more = format!("+{dropped} more");
1320            let mut line = last.clone();
1321            line.push(more);
1322            if glyphs::display_width(&join(&line)) <= width || last.is_empty() {
1323                *last = line;
1324                break;
1325            }
1326            // A line's only fact is cut short rather than given up, so a line
1327            // never says only how much it left out.
1328            if let [only] = last.as_mut_slice() {
1329                let room =
1330                    width.saturating_sub(glyphs::display_width(&format!("{sep}+{dropped} more")));
1331                if room >= 8 {
1332                    *only = glyphs::fit(only, room);
1333                    continue;
1334                }
1335            }
1336            last.pop();
1337            dropped += 1;
1338        }
1339    }
1340    lines.iter().map(|line| join(line)).collect()
1341}
1342
1343/// A title on a rule with a flat count chip, as `SectionRule` draws it, from the
1344/// theme this widget is handed.
1345pub(crate) fn rule_line(
1346    title: &str,
1347    chip: Option<&str>,
1348    width: u16,
1349    theme: &Theme,
1350) -> Line<'static> {
1351    let mut spans = vec![
1352        Span::styled(
1353            title.to_string(),
1354            Style::default()
1355                .fg(theme.accent())
1356                .add_modifier(Modifier::BOLD),
1357        ),
1358        Span::raw(" "),
1359    ];
1360    let mut used = glyphs::display_width(title) + 1;
1361    if let Some(chip) = chip {
1362        let chip = format!(" {chip} ");
1363        used += glyphs::display_width(&chip) + 1;
1364        spans.push(Span::styled(
1365            chip,
1366            Style::default()
1367                .bg(theme.controls_bg())
1368                .fg(theme.text_primary()),
1369        ));
1370        spans.push(Span::raw(" "));
1371    }
1372    spans.push(Span::styled(
1373        glyphs::get()
1374            .rule_h
1375            .repeat((width as usize).saturating_sub(used)),
1376        Style::default().fg(theme.table_column_separator()),
1377    ));
1378    Line::from(spans)
1379}
1380
1381fn severity_mark(severity: Severity, theme: &Theme) -> Span<'static> {
1382    let g = glyphs::get();
1383    match severity {
1384        Severity::Problem => Span::styled(g.warning, Style::default().fg(theme.warning())),
1385        Severity::Note => Span::styled(g.middot, Style::default().fg(theme.text_primary())),
1386        Severity::Clean => Span::styled(g.check, Style::default().fg(theme.success())),
1387    }
1388}
1389
1390/// Findings under one rule per severity. The selection lives in finding space, so
1391/// the keys never land on a rule; the scroll offset lives in line space.
1392fn render_findings(
1393    config: &DataQualityWidgetConfig<'_>,
1394    report: &QualityReport,
1395    table_state: &mut TableState,
1396    area: Rect,
1397    buf: &mut Buffer,
1398) {
1399    enum Item {
1400        Rule(Severity, usize),
1401        Gap,
1402        Finding(usize),
1403    }
1404    // Positions in the list as shown, which is what the selection counts; each
1405    // severity's rule counts what is shown under it.
1406    let shown = config.findings.shown(report);
1407    let mut items = Vec::new();
1408    let mut current = None;
1409    for (position, index) in shown.iter().enumerate() {
1410        let finding = &report.findings[*index];
1411        if current != Some(finding.severity) {
1412            if current.is_some() {
1413                items.push(Item::Gap);
1414            }
1415            let count = shown
1416                .iter()
1417                .filter(|other| report.findings[**other].severity == finding.severity)
1418                .count();
1419            let count = match finding.severity {
1420                // The clean entry is one row naming many columns; count the columns.
1421                Severity::Clean => report.clean_columns,
1422                _ => count,
1423            };
1424            items.push(Item::Rule(finding.severity, count));
1425            current = Some(finding.severity);
1426        }
1427        items.push(Item::Finding(position));
1428    }
1429
1430    let height = area.height as usize;
1431    if height == 0 {
1432        return;
1433    }
1434    let selected = table_state.selected().unwrap_or(0);
1435    let selected_line = items
1436        .iter()
1437        .position(|item| matches!(item, Item::Finding(index) if *index == selected))
1438        .unwrap_or(0);
1439    let mut offset = table_state.offset().min(items.len().saturating_sub(1));
1440    if selected_line < offset {
1441        // Bring the section's rule along when the selection is its first row.
1442        offset = if selected_line > 0 && matches!(items[selected_line - 1], Item::Rule(..)) {
1443            selected_line - 1
1444        } else {
1445            selected_line
1446        };
1447    } else if selected_line >= offset + height {
1448        offset = selected_line + 1 - height;
1449    }
1450    *table_state.offset_mut() = offset;
1451
1452    // Mark, title, columns, summary. The title column fits the longest title and a
1453    // gap; the columns take a share of what is left, so the summary keeps the rest.
1454    let title_width = report
1455        .findings
1456        .iter()
1457        .map(|finding| glyphs::display_width(finding.title))
1458        .max()
1459        .unwrap_or(0)
1460        + 2;
1461    let width = area.width as usize;
1462    let lead = 4; // rail + space + mark + space
1463    let rest = width.saturating_sub(lead + title_width);
1464    let columns_width = (rest * 2 / 5).clamp(10.min(rest), 32);
1465    // Ordered by a number, the number is on every row, right-aligned, so the order
1466    // can be read; the summary gives up the room.
1467    let ordered = config.findings.order != FindingOrder::Ranked;
1468    let numbers = |finding: &crate::analysis::quality_report::Finding| {
1469        if finding.kind.is_none() {
1470            return (String::new(), String::new());
1471        }
1472        (
1473            numfmt::group_chrome(finding.affected_rows),
1474            crate::numfmt::percent_of(finding.affected_rows, finding.evaluated_rows),
1475        )
1476    };
1477    let widest = |pick: fn((String, String)) -> String| {
1478        shown
1479            .iter()
1480            .map(|index| glyphs::display_width(&pick(numbers(&report.findings[*index]))))
1481            .max()
1482            .unwrap_or(0)
1483    };
1484    let (count_width, rate_width) = (widest(|(count, _)| count), widest(|(_, rate)| rate));
1485    // The number the list is ordered by leads, each in its own aligned column.
1486    let affected = |finding: &crate::analysis::quality_report::Finding| {
1487        let (count, rate) = numbers(finding);
1488        match config.findings.order {
1489            FindingOrder::Rate => format!("{rate:>rate_width$}  {count:>count_width$}"),
1490            _ => format!("{count:>count_width$}  {rate:>rate_width$}"),
1491        }
1492    };
1493    let affected_width = if ordered {
1494        count_width + rate_width + 3
1495    } else {
1496        0
1497    };
1498    let summary_width = rest.saturating_sub(columns_width + 1 + affected_width);
1499    // Too narrow to say anything, the summary gives its room to the numbers.
1500    let (summary_width, affected_width) = if ordered && summary_width < 12 {
1501        (
1502            0,
1503            rest.saturating_sub(columns_width + 1).max(affected_width),
1504        )
1505    } else {
1506        (summary_width, affected_width)
1507    };
1508    let g = glyphs::get();
1509    let theme = config.theme;
1510    let focused = config.focus == AnalysisFocus::Main;
1511    for (row, item) in items.iter().skip(offset).take(height).enumerate() {
1512        let line_area = Rect {
1513            y: area.y + row as u16,
1514            height: 1,
1515            ..area
1516        };
1517        let line = match item {
1518            Item::Gap => continue,
1519            Item::Rule(severity, count) => rule_line(
1520                severity.heading(),
1521                Some(&numfmt::group_chrome(*count)),
1522                area.width,
1523                theme,
1524            ),
1525            Item::Finding(position) => {
1526                let finding = &report.findings[shown[*position]];
1527                let is_selected = *position == selected;
1528                let columns = glyphs::fit(&finding.columns_label(columns_width), columns_width);
1529                let summary = glyphs::fit(&finding.summary, summary_width);
1530                let mut spans = vec![
1531                    Span::styled(
1532                        if is_selected { g.rail } else { " " },
1533                        Style::default().fg(theme.accent()),
1534                    ),
1535                    Span::raw(" "),
1536                    severity_mark(finding.severity, theme),
1537                    Span::raw(" "),
1538                    Span::styled(
1539                        format!("{:<title_width$}", finding.title),
1540                        Style::default().fg(theme.text_primary()),
1541                    ),
1542                    Span::styled(
1543                        format!("{columns:<columns_width$} "),
1544                        Style::default().fg(theme.text_primary()),
1545                    ),
1546                    Span::styled(
1547                        if ordered {
1548                            format!("{summary:<summary_width$}")
1549                        } else {
1550                            summary
1551                        },
1552                        Style::default().fg(theme.dimmed()),
1553                    ),
1554                ];
1555                if ordered {
1556                    spans.push(Span::styled(
1557                        format!("{:>affected_width$}", affected(finding)),
1558                        Style::default().fg(theme.text_primary()),
1559                    ));
1560                }
1561                let line = Line::from(spans);
1562                if is_selected {
1563                    let style = if focused {
1564                        theme.highlight_style()
1565                    } else {
1566                        Style::default()
1567                    };
1568                    buf.set_style(line_area, style);
1569                }
1570                line
1571            }
1572        };
1573        line.render(line_area, buf);
1574    }
1575}
1576
1577/// The checks the run made, one per line: mark, name, reach, outcome, and what each
1578/// looks for when there is room. `limit` keeps the first few; the rest are counted.
1579fn check_lines(
1580    checks: &[Check],
1581    width: u16,
1582    limit: Option<usize>,
1583    theme: &Theme,
1584) -> Vec<Line<'static>> {
1585    let width = width as usize;
1586    let dimmed = Style::default().fg(theme.dimmed());
1587    let g = glyphs::get();
1588    let shown = limit.unwrap_or(checks.len()).min(checks.len());
1589    let outcome = |check: &Check| match &check.outcome {
1590        Outcome::Passed => "passed".to_string(),
1591        Outcome::Found { detail, .. } => format!("{detail} flagged"),
1592        Outcome::Skipped(reason) => format!("skipped: {reason}"),
1593        Outcome::Unavailable(reason) => format!("unavailable: {reason}"),
1594    };
1595    // Each column as wide as its widest entry across checks, so showing all moves nothing;
1596    // what a check looks for takes the rest, wrapping under itself.
1597    let column = |text: &dyn Fn(&Check) -> String| {
1598        checks
1599            .iter()
1600            .map(|check| glyphs::display_width(&text(check)))
1601            .max()
1602            .unwrap_or(0)
1603            + 2
1604    };
1605    let name_width = column(&|check| check.name.to_string());
1606    let reach_width = column(&|check| check.applies_to.clone());
1607    let outcome_width = column(&outcome);
1608    let lead = 2 + name_width + reach_width + outcome_width;
1609    // Too narrow for a readable fourth column: it goes on the lines below instead.
1610    let (looks_indent, looks_width) = if width >= lead + 24 {
1611        (lead, width - lead)
1612    } else {
1613        (2, width.saturating_sub(2).max(1))
1614    };
1615    let mut lines = vec![rule_line(
1616        "Checks",
1617        Some(&numfmt::group_chrome(checks.len())),
1618        width as u16,
1619        theme,
1620    )];
1621    for check in &checks[..shown] {
1622        let (mark, style) = match &check.outcome {
1623            Outcome::Passed => (
1624                Span::styled(g.check, Style::default().fg(theme.success())),
1625                Style::default().fg(theme.text_primary()),
1626            ),
1627            Outcome::Found { tier, .. } => (
1628                severity_mark(*tier, theme),
1629                Style::default().fg(theme.text_primary()),
1630            ),
1631            Outcome::Skipped(_) | Outcome::Unavailable(_) => (Span::styled(g.dash, dimmed), dimmed),
1632        };
1633        let beside = looks_indent == lead;
1634        let mut spans = vec![
1635            mark,
1636            Span::raw(" "),
1637            Span::styled(format!("{:<name_width$}", check.name), style),
1638            Span::styled(format!("{:<reach_width$}", check.applies_to), dimmed),
1639            // Padded only when something follows it on the line: trailing spaces
1640            // past the frame would wrap into a blank line.
1641            Span::styled(
1642                if beside {
1643                    format!("{:<outcome_width$}", outcome(check))
1644                } else {
1645                    outcome(check)
1646                },
1647                style,
1648            ),
1649        ];
1650        let mut looks = crate::widgets::info::wrap_to(check.looks_for, looks_width).into_iter();
1651        if beside {
1652            spans.extend(looks.next().map(|text| Span::styled(text, dimmed)));
1653        }
1654        lines.push(Line::from(spans));
1655        lines.extend(
1656            looks.map(|text| Line::styled(format!("{}{text}", " ".repeat(looks_indent)), dimmed)),
1657        );
1658    }
1659    if shown < checks.len() {
1660        lines.push(Line::styled(
1661            format!(
1662                "  {} more {}",
1663                checks.len() - shown,
1664                if checks.len() - shown == 1 {
1665                    "check"
1666                } else {
1667                    "checks"
1668                }
1669            ),
1670            dimmed,
1671        ));
1672    }
1673    lines
1674}
1675
1676/// A segment label as drawn: `∅` (no value) uses the null glyph (ASCII under `LANG=C`).
1677/// The label itself is unchanged, since evidence finds rows by it.
1678fn segment_text(label: &str) -> String {
1679    label.replace('∅', glyphs::get().null)
1680}
1681
1682/// The finding itself: what it is in one sentence with its numbers, why it matters,
1683/// what to check, and the evidence, in a frame on top of the list.
1684fn render_finding_detail(
1685    config: &DataQualityWidgetConfig<'_>,
1686    table_state: &TableState,
1687    scroll: &mut DetailScroll,
1688    area: Rect,
1689    buf: &mut Buffer,
1690) {
1691    let Some(results) = config.results else {
1692        return;
1693    };
1694    let report = results.report();
1695    let Some(finding) = table_state
1696        .selected()
1697        .and_then(|position| config.findings.selected(report, position))
1698    else {
1699        return;
1700    };
1701    let theme = config.theme;
1702    let dimmed = Style::default().fg(theme.dimmed());
1703    let (headline, evidence) = describe(finding, results);
1704    // A reading surface: cap the measure on a wide terminal. The checks table on
1705    // the clean entry is a table, and may use more of the width.
1706    let width = area
1707        .width
1708        .saturating_sub(4)
1709        .min(if finding.kind.is_none() { 118 } else { 84 });
1710    let inner = width.saturating_sub(4).max(1);
1711    // The name and the columns on the frame; the facts as a list under it.
1712    let title = if finding.kind.is_none() {
1713        finding.title.to_string()
1714    } else {
1715        let name = format!("{} {} ", finding.title, glyphs::get().middot);
1716        let room = (width as usize).saturating_sub(glyphs::display_width(&name) + 4);
1717        format!("{name}{}", finding.columns_label(room))
1718    };
1719    let bullet = |text: String| {
1720        Line::from(vec![
1721            Span::styled(format!("{} ", glyphs::get().middot), dimmed),
1722            Span::raw(text),
1723        ])
1724    };
1725    let mut lines = Vec::new();
1726    // The frame cut the list short: the whole of it, unless the evidence lists
1727    // every column with its rate already.
1728    if finding.kind.is_some()
1729        && finding.columns.len() > 1
1730        && !finding.lists_columns()
1731        && finding.columns_label(usize::MAX) != finding.columns_label(inner as usize)
1732    {
1733        lines.push(bullet(finding.columns.join(", ")));
1734    }
1735    lines.push(bullet(headline));
1736    lines.extend(
1737        evidence
1738            .into_iter()
1739            .map(|line| Line::raw(format!("  {line}"))),
1740    );
1741    lines.extend(advice(finding).into_iter().map(bullet));
1742    if finding.kind.is_none() {
1743        lines.push(Line::raw(""));
1744        lines.extend(check_lines(
1745            results.checks(),
1746            inner,
1747            (!config.checks_expanded).then_some(CHECKS_SHOWN),
1748            theme,
1749        ));
1750    }
1751    if finding.kind.is_some() {
1752        lines.push(Line::raw(""));
1753        lines.push(Line::styled(
1754            evidence_line(config, finding, results),
1755            dimmed,
1756        ));
1757    }
1758    // Grow with the text up to the screen, then scroll inside the frame; the last
1759    // row counts what is below.
1760    let rows = lines
1761        .iter()
1762        .map(|line| crate::render::home_view::wrapped_rows(line, inner as usize))
1763        .sum::<usize>()
1764        .min(u16::MAX as usize) as u16;
1765    let height = (rows + 2).min(area.height.saturating_sub(2));
1766    let room = height.saturating_sub(2);
1767    scroll.max = rows.saturating_sub(room);
1768    scroll.offset = scroll.offset.min(scroll.max);
1769    // Short of the end, the last row counts what is below instead of showing text.
1770    let shown = if scroll.offset < scroll.max {
1771        room.saturating_sub(1)
1772    } else {
1773        room
1774    };
1775    let below = rows.saturating_sub(scroll.offset + shown);
1776    let popup = dialog_in(area, width, height);
1777    let content = Surface::new(&title)
1778        .border_style(Style::default().fg(config.ctx.modal_border_active))
1779        .render(popup, buf, config.ctx);
1780    Paragraph::new(lines)
1781        .wrap(Wrap { trim: false })
1782        .scroll((scroll.offset, 0))
1783        .render(
1784            Rect {
1785                height: shown.min(content.height),
1786                ..content
1787            },
1788            buf,
1789        );
1790    if below > 0 && content.height > shown {
1791        Paragraph::new(Line::styled(
1792            format!("{} {below} more", glyphs::get().ellipsis),
1793            dimmed,
1794        ))
1795        .alignment(Alignment::Right)
1796        .render(
1797            Rect {
1798                y: content.y + shown,
1799                height: 1,
1800                ..content
1801            },
1802            buf,
1803        );
1804    }
1805}
1806
1807/// What Enter does with a finding's rows, in one sentence: shows the ones the run
1808/// kept, reads ones it did not keep once asked, or says why there are none.
1809fn evidence_line(
1810    config: &DataQualityWidgetConfig<'_>,
1811    finding: &crate::analysis::quality_report::Finding,
1812    results: &DataQualityResults,
1813) -> String {
1814    let rows = match finding.evidence(results) {
1815        Ok(rows) => rows,
1816        Err(reason) => return reason,
1817    };
1818    if let EvidenceRows::Files(QualityScope::SourceFiles(files)) = &rows {
1819        return format!(
1820            "Enter: rows of the {} named {} (asks first)",
1821            files.len(),
1822            if files.len() == 1 { "file" } else { "files" }
1823        );
1824    }
1825    let sampled = results.precision == QualityPrecision::Sampled;
1826    let noun = |count: usize| {
1827        format!(
1828            "{} {}{}",
1829            numfmt::group_chrome(count),
1830            if sampled { "sampled " } else { "" },
1831            if count == 1 { "row" } else { "rows" }
1832        )
1833    };
1834    let spec = finding.kind.map(ObservationKind::spec);
1835    let what = match (spec, finding.evidence_count(results)) {
1836        // The measurement counts rows beyond one per value; the rows that share a
1837        // value are always more.
1838        (Some(spec), _) if spec.evidence == Evidence::SharingValue => format!(
1839            "every {}row that shares a repeated value",
1840            if sampled { "sampled " } else { "" }
1841        ),
1842        (spec, Some(count)) => format!(
1843            "the {}{}",
1844            noun(count),
1845            spec.map_or("", |spec| spec.rows_are)
1846        ),
1847        // Grouped columns: a row missing in any of them; the table counts them.
1848        (_, None) => format!(
1849            "the {}rows with any of them",
1850            if sampled { "sampled " } else { "" }
1851        ),
1852    };
1853    if config.rows_kept {
1854        let together = if matches!(rows, EvidenceRows::Duplicates) {
1855            ", copies together"
1856        } else {
1857            ""
1858        };
1859        format!("Enter: {what}{together}")
1860    } else if sampled {
1861        format!("Enter: {what} (sample released, asks first)")
1862    } else if config.measured.compute == QualityCompute::Full {
1863        format!("Enter: {what} (full scan keeps none, asks first)")
1864    } else {
1865        format!("Enter: {what} (rows released, asks first)")
1866    }
1867}
1868
1869/// A read of a finding's rows, before it reads: what, why, how much, from where.
1870fn render_evidence_read(
1871    config: &DataQualityWidgetConfig<'_>,
1872    read: &EvidenceRead,
1873    area: Rect,
1874    buf: &mut Buffer,
1875) {
1876    let rows = read
1877        .summary
1878        .iter()
1879        .map(|(label, value)| FieldRow {
1880            mark: None,
1881            label: label.to_string(),
1882            value: value.clone(),
1883        })
1884        .collect::<Vec<_>>();
1885    let width = 72.min(area.width.saturating_sub(2));
1886    let label_width = rows
1887        .iter()
1888        .map(|row| glyphs::display_width(&row.label))
1889        .max()
1890        .unwrap_or(0)
1891        + 2;
1892    let lines = field_lines(&rows, label_width, width.saturating_sub(4) as usize, false);
1893    let popup = dialog_in(area, width, lines.len() as u16 + 2);
1894    let content = Surface::new("Read Rows")
1895        .border_style(Style::default().fg(config.ctx.modal_border_active))
1896        .render(popup, buf, config.ctx);
1897    render_counted(lines, content, config.theme, buf);
1898}
1899
1900fn render_time_roles(
1901    config: &DataQualityWidgetConfig<'_>,
1902    table_state: &mut TableState,
1903    area: Rect,
1904    buf: &mut Buffer,
1905) {
1906    let theme = config.theme;
1907    let dimmed = Style::default().fg(theme.dimmed());
1908    let accent = Style::default().fg(theme.accent());
1909    let columns = config.setup.time_candidates;
1910    let roles = TemporalRole::ALL.len() as u16;
1911    let sections = Layout::default()
1912        .direction(Direction::Vertical)
1913        .constraints([
1914            Constraint::Length(2),
1915            Constraint::Length(roles + 1),
1916            Constraint::Length(1),
1917            Constraint::Length(2),
1918            Constraint::Fill(1),
1919        ])
1920        .margin(1)
1921        .split(area);
1922    Paragraph::new(rule_line("Time roles", None, sections[0].width, theme))
1923        .render(sections[0], buf);
1924
1925    let focused_role = TemporalRole::ALL[config.plan_field.min(TemporalRole::ALL.len() - 1)];
1926    let assigned = |role: TemporalRole| {
1927        config
1928            .plan
1929            .temporal_roles
1930            .iter()
1931            .find(|assignment| assignment.role == role)
1932            .map(|assignment| assignment.column.clone())
1933    };
1934    let rows = TemporalRole::ALL.iter().map(|role| {
1935        Row::new(vec![
1936            Cell::from(role.label()),
1937            match assigned(*role) {
1938                Some(column) => Cell::from(column),
1939                None => Cell::from(Span::styled("unassigned", dimmed)),
1940            },
1941        ])
1942    });
1943    table_state.select(Some(config.plan_field.min(TemporalRole::ALL.len() - 1)));
1944    let table = Table::new(rows, [Constraint::Length(20), Constraint::Fill(1)])
1945        .header(Row::new(["Role", "Column"]).style(dimmed))
1946        .row_highlight_style(theme.highlight_style())
1947        .highlight_symbol(glyphs::get().selector);
1948    StatefulWidget::render(table, sections[1], buf, table_state);
1949
1950    // The candidates, with a few of their values from the rows already on screen,
1951    // so choosing which column is "received" is choosing among things seen.
1952    Paragraph::new(rule_line(
1953        "Date, time and text columns",
1954        Some(&numfmt::group_chrome(columns.len())),
1955        sections[3].width,
1956        theme,
1957    ))
1958    .render(sections[3], buf);
1959    if columns.is_empty() {
1960        Paragraph::new(Span::styled("None in this data", dimmed)).render(sections[4], buf);
1961        return;
1962    }
1963    let name_width = columns
1964        .iter()
1965        .map(|column| glyphs::display_width(column))
1966        .max()
1967        .unwrap_or(0)
1968        + 2;
1969    let types = columns
1970        .iter()
1971        .map(|column| {
1972            // Text says how it is read, so a role on it is not mistaken for a cast.
1973            match (
1974                config.plan.time_format(column),
1975                config.state.quality_schema(&config.plan.scope).get(column),
1976            ) {
1977                (Some(format), _) => format!("text as {}", format.kind.label()),
1978                (None, Some(DataType::String | DataType::Categorical(..))) => {
1979                    "text, no format".to_string()
1980                }
1981                (None, dtype) => dtype.map(|dtype| dtype.to_string()).unwrap_or_default(),
1982            }
1983        })
1984        .collect::<Vec<_>>();
1985    let type_width = types
1986        .iter()
1987        .map(|dtype| glyphs::display_width(dtype))
1988        .max()
1989        .unwrap_or(0)
1990        + 2;
1991    let chosen = assigned(focused_role);
1992    let values_width = (sections[4].width as usize).saturating_sub(2 + name_width + type_width);
1993    let lines = columns
1994        .iter()
1995        .zip(&types)
1996        .map(|(column, dtype)| {
1997            let is_chosen = chosen.as_deref() == Some(column.as_str());
1998            let values = config.state.buffered_values(column, 3);
1999            let values = if values.is_empty() {
2000                "not in the rows on screen".to_string()
2001            } else {
2002                values.join("   ")
2003            };
2004            Line::from(vec![
2005                Span::styled(if is_chosen { glyphs::get().rail } else { " " }, accent),
2006                Span::raw(" "),
2007                Span::styled(
2008                    format!("{column:<name_width$}"),
2009                    if is_chosen {
2010                        accent
2011                    } else {
2012                        Style::default().fg(theme.text_primary())
2013                    },
2014                ),
2015                Span::styled(format!("{dtype:<type_width$}"), dimmed),
2016                Span::raw(glyphs::fit(&values, values_width)),
2017            ])
2018        })
2019        .collect::<Vec<_>>();
2020    Paragraph::new(lines).render(sections[4], buf);
2021}
2022
2023fn render_columns(
2024    config: &DataQualityWidgetConfig<'_>,
2025    table_state: &mut TableState,
2026    area: Rect,
2027    buf: &mut Buffer,
2028) {
2029    let Some(results) = config.results else {
2030        render_run_prompt(area, config.theme, buf);
2031        return;
2032    };
2033    let report = results.report();
2034    let sections = Layout::default()
2035        .direction(Direction::Vertical)
2036        .constraints([Constraint::Length(2), Constraint::Fill(1)])
2037        .margin(1)
2038        .split(area);
2039    let title = rule_line(
2040        "Columns",
2041        Some(&numfmt::group_chrome(results.columns.len())),
2042        sections[0].width,
2043        config.theme,
2044    );
2045    Paragraph::new(title).render(sections[0], buf);
2046    let layout = if sections[1].width >= 120 {
2047        2
2048    } else if sections[1].width >= 72 {
2049        1
2050    } else {
2051        0
2052    };
2053    let dimmed = Style::default().fg(config.theme.dimmed());
2054    let rows = results.columns.iter().enumerate().map(|(index, profile)| {
2055        let severity = report
2056            .column_status
2057            .get(index)
2058            .copied()
2059            .unwrap_or(Severity::Clean);
2060        // With no values read, a column with no finding is unknown, not clean.
2061        let unknown = severity == Severity::Clean && (report.no_rows || report.metadata_only);
2062        let mark = if unknown {
2063            Cell::from("")
2064        } else {
2065            Cell::from(Line::from(severity_mark(severity, config.theme)))
2066        };
2067        let findings = report
2068            .column_findings
2069            .get(index)
2070            .filter(|titles| !titles.is_empty())
2071            .map(|titles| Cell::from(titles.join(", ")))
2072            .unwrap_or_else(|| Cell::from(Span::styled("none", dimmed)));
2073        let most_common = profile
2074            .dominant_value
2075            .as_ref()
2076            .zip(profile.dominant_count)
2077            .map(|(value, count)| {
2078                format!(
2079                    "{value} ({})",
2080                    crate::numfmt::percent_of(count, profile.non_null_rows())
2081                )
2082            })
2083            .unwrap_or_else(|| "-".to_string());
2084        let missing = if profile.null_count == 0 {
2085            Cell::from(Span::styled("0", dimmed))
2086        } else {
2087            Cell::from(format!(
2088                "{} ({})",
2089                numfmt::group_chrome(profile.null_count),
2090                crate::numfmt::percent_of(profile.null_count, profile.evaluated_rows)
2091            ))
2092        };
2093        let distinct = count_label(profile.distinct_count);
2094        Row::new(match layout {
2095            2 => vec![
2096                mark,
2097                Cell::from(profile.name.clone()),
2098                Cell::from(profile.dtype.to_string()),
2099                missing,
2100                Cell::from(distinct),
2101                Cell::from(most_common),
2102                findings,
2103            ],
2104            1 => vec![
2105                mark,
2106                Cell::from(profile.name.clone()),
2107                Cell::from(profile.dtype.to_string()),
2108                missing,
2109                findings,
2110            ],
2111            _ => vec![mark, Cell::from(profile.name.clone()), findings],
2112        })
2113    });
2114    normalize_selection(table_state, results.columns.len());
2115    let (headers, widths) = match layout {
2116        2 => (
2117            vec![
2118                "",
2119                "Column",
2120                "Type",
2121                "Missing",
2122                "Distinct",
2123                "Most common",
2124                "Findings",
2125            ],
2126            vec![
2127                Constraint::Length(1),
2128                Constraint::Length(22),
2129                Constraint::Length(10),
2130                Constraint::Length(16),
2131                Constraint::Length(10),
2132                Constraint::Length(22),
2133                Constraint::Fill(1),
2134            ],
2135        ),
2136        1 => (
2137            vec!["", "Column", "Type", "Missing", "Findings"],
2138            vec![
2139                Constraint::Length(1),
2140                Constraint::Length(20),
2141                Constraint::Length(8),
2142                Constraint::Length(16),
2143                Constraint::Fill(1),
2144            ],
2145        ),
2146        _ => (
2147            vec!["", "Column", "Findings"],
2148            vec![
2149                Constraint::Length(1),
2150                Constraint::Length(16),
2151                Constraint::Fill(1),
2152            ],
2153        ),
2154    };
2155    let table = Table::new(rows, widths)
2156        .header(Row::new(headers).style(dimmed))
2157        .row_highlight_style(config.theme.highlight_style())
2158        .highlight_symbol(glyphs::get().selector);
2159    StatefulWidget::render(table, sections[1], buf, table_state);
2160}
2161
2162fn render_segments(
2163    config: &DataQualityWidgetConfig<'_>,
2164    table_state: &mut TableState,
2165    area: Rect,
2166    buf: &mut Buffer,
2167) {
2168    let Some(results) = config.results else {
2169        render_run_prompt(area, config.theme, buf);
2170        return;
2171    };
2172    let theme = config.theme;
2173    let dimmed = Style::default().fg(theme.dimmed());
2174    let sections = Layout::default()
2175        .direction(Direction::Vertical)
2176        .constraints([Constraint::Length(2), Constraint::Fill(1)])
2177        .margin(1)
2178        .split(area);
2179    if config.plan.grain == QualityGrain::Dataset {
2180        Paragraph::new(rule_line("Segments", Some("1"), sections[0].width, theme))
2181            .render(sections[0], buf);
2182        Paragraph::new(
2183            "The rows are one segment. Set grain in Setup (e) to split them by file, \
2184             partition, row chunk or time window, and compare the parts.",
2185        )
2186        .wrap(Wrap { trim: true })
2187        .style(Style::default().fg(theme.text_primary()))
2188        .render(sections[1], buf);
2189        return;
2190    }
2191    Paragraph::new(rule_line(
2192        if config.segments_by_change {
2193            "Segments, largest change first"
2194        } else {
2195            "Segments"
2196        },
2197        Some(&numfmt::group_chrome(results.segments.len())),
2198        sections[0].width,
2199        theme,
2200    ))
2201    .render(sections[0], buf);
2202    // A thin sample per segment can only name large moves; say how thin.
2203    if results.precision == QualityPrecision::Sampled && !results.segments.is_empty() {
2204        let typical = {
2205            let mut rows = results
2206                .segments
2207                .iter()
2208                .map(|segment| segment.evaluated_rows)
2209                .collect::<Vec<_>>();
2210            rows.sort_unstable();
2211            rows[rows.len() / 2]
2212        };
2213        Paragraph::new(Line::styled(
2214            format!(
2215                "About {} sampled rows a segment: a change is named only when it is past \
2216                 sampling noise. Trends pools segments to show smaller ones",
2217                numfmt::group_chrome(typical)
2218            ),
2219            dimmed,
2220        ))
2221        .render(
2222            Rect {
2223                y: sections[0].y + 1,
2224                height: 1,
2225                ..sections[0]
2226            },
2227            buf,
2228        );
2229    }
2230
2231    // Per segment without choosing anything: rows, emptiness, and its largest move against
2232    // its comparison. Enter shows the rest.
2233    let rows_label =
2234        |segment: &crate::analysis::data_quality::SegmentQualityProfile| match segment.total_rows {
2235            Some(total) if total != segment.evaluated_rows => format!(
2236                "{} of {}",
2237                numfmt::group_chrome(segment.evaluated_rows),
2238                numfmt::group_chrome(total)
2239            ),
2240            _ => numfmt::group_chrome(segment.evaluated_rows),
2241        };
2242    let label_width = results
2243        .segments
2244        .iter()
2245        .map(|segment| glyphs::display_width(&segment_text(&segment.label)))
2246        .max()
2247        .unwrap_or(0)
2248        .clamp(7, 40) as u16
2249        + 2;
2250    let rows_width = results
2251        .segments
2252        .iter()
2253        .map(|segment| glyphs::display_width(&rows_label(segment)))
2254        .max()
2255        .unwrap_or(0)
2256        .max(4) as u16
2257        + 2;
2258    let compared = config.plan.comparison != QualityComparison::None;
2259    let order = crate::analysis::data_quality::segment_order(results, config.segments_by_change);
2260    let rows = order
2261        .iter()
2262        .map(|&index| &results.segments[index])
2263        .map(|segment| {
2264            let mut cells = vec![
2265                Cell::from(segment_text(&segment.label)),
2266                Cell::from(rows_label(segment)),
2267                Cell::from(crate::numfmt::percent(segment.null_rate)),
2268            ];
2269            if compared {
2270                cells.push(Cell::from(
2271                    segment.largest_change.clone().unwrap_or_default(),
2272                ));
2273            }
2274            Row::new(cells)
2275        });
2276    let mut headers = vec!["Segment", "Rows", "Null cells"];
2277    let mut widths = vec![
2278        Constraint::Length(label_width),
2279        Constraint::Length(rows_width),
2280        Constraint::Length(12),
2281    ];
2282    if compared {
2283        headers.push("Largest change");
2284        widths.push(Constraint::Fill(1));
2285    }
2286    normalize_selection(table_state, results.segments.len());
2287    let table = Table::new(rows, widths)
2288        .header(Row::new(headers).style(dimmed))
2289        .row_highlight_style(theme.highlight_style())
2290        .highlight_symbol(glyphs::get().selector);
2291    StatefulWidget::render(table, sections[1], buf, table_state);
2292}
2293
2294/// One segment's columns with every measure beside its comparison, largest move first:
2295/// "what changed here" without choosing a column or measure.
2296fn render_segment_detail(
2297    config: &DataQualityWidgetConfig<'_>,
2298    table_state: &mut TableState,
2299    segment_index: usize,
2300    area: Rect,
2301    buf: &mut Buffer,
2302) {
2303    let Some(results) = config.results else {
2304        render_run_prompt(area, config.theme, buf);
2305        return;
2306    };
2307    let Some(segment) = results.segments.get(segment_index) else {
2308        return;
2309    };
2310    let theme = config.theme;
2311    let dimmed = Style::default().fg(theme.dimmed());
2312    let changes = crate::analysis::data_quality::segment_changes(results, segment_index);
2313    let sections = Layout::default()
2314        .direction(Direction::Vertical)
2315        .constraints([Constraint::Length(2), Constraint::Fill(1)])
2316        .margin(1)
2317        .split(area);
2318    let other = segment.compared_with.as_deref();
2319    let title = match other {
2320        Some(other) => format!(
2321            "{} vs {}",
2322            segment_text(&segment.label),
2323            segment_text(other)
2324        ),
2325        None => segment_text(&segment.label),
2326    };
2327    Paragraph::new(rule_line(
2328        &title,
2329        Some(&numfmt::group_chrome(changes.len())),
2330        sections[0].width,
2331        theme,
2332    ))
2333    .render(sections[0], buf);
2334    if changes.is_empty() {
2335        Paragraph::new(Span::styled("Nothing measured is above zero here", dimmed))
2336            .render(sections[1], buf);
2337        return;
2338    }
2339    let name_width = changes
2340        .iter()
2341        .map(|change| glyphs::display_width(&change.column))
2342        .max()
2343        .unwrap_or(0)
2344        .clamp(6, 32) as u16
2345        + 2;
2346    let value_width = |label: &str| (glyphs::display_width(label).clamp(8, 24) + 2) as u16;
2347    let rows = changes.iter().map(|change| {
2348        let style = if change.clear || other.is_none() {
2349            Style::default()
2350        } else {
2351            dimmed
2352        };
2353        let mut cells = vec![
2354            Cell::from(Span::styled(change.column.clone(), style)),
2355            Cell::from(Span::styled(change.metric.short_label(), dimmed)),
2356        ];
2357        if other.is_some() {
2358            cells.push(Cell::from(
2359                change
2360                    .before
2361                    .map(crate::numfmt::percent)
2362                    .unwrap_or_else(|| "-".to_string()),
2363            ));
2364        }
2365        cells.push(Cell::from(crate::numfmt::percent(change.now)));
2366        if other.is_some() {
2367            cells.push(match change.change() {
2368                Some(points) => Cell::from(Span::styled(format!("{points:+.1} pp"), style)),
2369                None => Cell::from(""),
2370            });
2371        }
2372        Row::new(cells)
2373    });
2374    let mut headers = vec!["Column".to_string(), "Measure".to_string()];
2375    let mut widths = vec![Constraint::Length(name_width), Constraint::Length(10)];
2376    if let Some(other) = other {
2377        headers.push(segment_text(other));
2378        widths.push(Constraint::Length(value_width(&segment_text(other))));
2379    }
2380    headers.push(segment_text(&segment.label));
2381    widths.push(Constraint::Length(value_width(&segment_text(
2382        &segment.label,
2383    ))));
2384    if other.is_some() {
2385        headers.push("Change".to_string());
2386        widths.push(Constraint::Fill(1));
2387    }
2388    normalize_selection(table_state, changes.len());
2389    let table = Table::new(rows, widths)
2390        .header(Row::new(headers).style(dimmed))
2391        .row_highlight_style(theme.highlight_style())
2392        .highlight_symbol(glyphs::get().selector);
2393    StatefulWidget::render(table, sections[1], buf, table_state);
2394}
2395
2396fn render_trends(
2397    config: &DataQualityWidgetConfig<'_>,
2398    table_state: &mut TableState,
2399    area: Rect,
2400    buf: &mut Buffer,
2401) {
2402    let Some(results) = config.results else {
2403        render_run_prompt(area, config.theme, buf);
2404        return;
2405    };
2406    let [area] = Layout::default()
2407        .direction(Direction::Vertical)
2408        .constraints([Constraint::Fill(1)])
2409        .margin(1)
2410        .areas(area);
2411    if crate::analysis::data_quality::shows_trend(config.plan, results) {
2412        render_trend_table(config, results, table_state, area, buf);
2413        return;
2414    }
2415    let [title, body] = Layout::default()
2416        .direction(Direction::Vertical)
2417        .constraints([Constraint::Length(2), Constraint::Fill(1)])
2418        .areas(area);
2419    Paragraph::new(rule_line(
2420        "Across segments",
2421        None,
2422        title.width,
2423        config.theme,
2424    ))
2425    .render(title, buf);
2426    let mut text = vec![Line::raw(
2427        "Set grain in Setup (e) to a partition column, to days, weeks or months of a \
2428         date, or to chunks of rows, to follow each column from one to the next.",
2429    )];
2430    // One window found is no trend, but the windows expected around it still are.
2431    if let Some(gaps) = crate::analysis::quality_trends::expected_gaps(config.plan, results) {
2432        text.extend([Line::raw(""), Line::raw(gaps_summary(config.plan, &gaps))]);
2433    }
2434    Paragraph::new(text)
2435        .wrap(Wrap { trim: true })
2436        .style(Style::default().fg(config.theme.text_primary()))
2437        .render(body, buf);
2438}
2439
2440/// `count` of `of`, with its rate when there is one: `2 of 5 (40.0%)`.
2441fn share(count: usize, of: usize) -> String {
2442    if of == 0 || count == 0 {
2443        format!(
2444            "{} of {}",
2445            numfmt::group_chrome(count),
2446            numfmt::group_chrome(of)
2447        )
2448    } else {
2449        format!(
2450            "{} of {} ({})",
2451            numfmt::group_chrome(count),
2452            numfmt::group_chrome(of),
2453            crate::numfmt::percent(count as f64 / of as f64)
2454        )
2455    }
2456}
2457
2458/// One row per interval per segment: which, where, and the key counts; Enter opens
2459/// every count, so nothing is lost to a narrow terminal.
2460fn render_intervals(
2461    config: &DataQualityWidgetConfig<'_>,
2462    table_state: &mut TableState,
2463    area: Rect,
2464    buf: &mut Buffer,
2465) {
2466    let Some(results) = config.results else {
2467        render_run_prompt(area, config.theme, buf);
2468        return;
2469    };
2470    let theme = config.theme;
2471    let plan = config.plan;
2472    let dimmed = Style::default().fg(theme.dimmed());
2473    let [title, note, body] = Layout::default()
2474        .direction(Direction::Vertical)
2475        .constraints([
2476            Constraint::Length(1),
2477            Constraint::Length(1),
2478            Constraint::Fill(1),
2479        ])
2480        .margin(1)
2481        .areas(area);
2482    let profiles = &results.temporal;
2483    let chip = (!profiles.is_empty()).then(|| numfmt::group_chrome(profiles.len()));
2484    Paragraph::new(rule_line(
2485        "Time between dates",
2486        chip.as_deref(),
2487        title.width,
2488        theme,
2489    ))
2490    .render(title, buf);
2491    if profiles.is_empty() {
2492        let schema = config.state.quality_schema(&plan.scope);
2493        let message = if config.setup.time_candidates.is_empty() {
2494            "No date, time or text columns"
2495        } else if plan.compute == QualityCompute::Metadata && !plan.interval_pairs().is_empty() {
2496            "File metadata only · no values read · e Setup"
2497        } else if plan.interval_pairs().iter().any(|(start, end)| {
2498            [start, end].into_iter().any(|role| {
2499                plan.role_column(*role)
2500                    .is_some_and(|column| !plan.reads_as_time(column, schema))
2501            })
2502        }) {
2503            "Interval column is text with no format · Text as time in Setup (e)"
2504        } else if !plan.interval_pairs().is_empty() {
2505            "No rows to measure"
2506        } else if !plan.candidate_pairs().is_empty() {
2507            "Time roles make no interval · Intervals in Setup (e)"
2508        } else {
2509            "No time roles · Time roles in Setup (e)"
2510        };
2511        Paragraph::new(glyphs::dotted(message))
2512            .wrap(Wrap { trim: true })
2513            .style(Style::default().fg(theme.text_primary()))
2514            .render(
2515                Rect {
2516                    height: body.height + 1,
2517                    ..note
2518                },
2519                buf,
2520            );
2521        return;
2522    }
2523    // What the last column counts, and out of what, said once above it.
2524    let threshold = plan.latency_threshold_seconds;
2525    let over = match threshold {
2526        Some(_) => format!(
2527            "Over: duration > {}, of rows with both ends",
2528            crate::analysis::analysis_modal::threshold_label(threshold)
2529        ),
2530        None => "Negative: end before start, of rows with both ends".to_string(),
2531    };
2532    Paragraph::new(Line::styled(
2533        glyphs::fit(&over, note.width as usize),
2534        dimmed,
2535    ))
2536    .render(note, buf);
2537
2538    let segmented = profiles
2539        .iter()
2540        .any(|profile| profile.segment != profiles[0].segment);
2541    let segment_width = profiles
2542        .iter()
2543        .map(|profile| glyphs::display_width(&segment_text(&profile.segment)))
2544        .max()
2545        .unwrap_or(0)
2546        .clamp(8, 22) as u16;
2547    let last = |profile: &TemporalLatencyProfile, counted: bool| {
2548        let count = match profile.above_threshold_count {
2549            Some(count) if threshold.is_some() => count,
2550            _ => profile.negative_count,
2551        };
2552        let rate = crate::numfmt::percent_of(count, profile.paired_rows);
2553        if counted && profile.paired_rows > 0 {
2554            format!("{} ({rate})", numfmt::group_chrome(count))
2555        } else {
2556            rate
2557        }
2558    };
2559    let last_header = if threshold.is_some() {
2560        "Over"
2561    } else {
2562        "Negative"
2563    };
2564    let wide = body.width >= 92;
2565    let medium = !wide && body.width >= 44;
2566    let mut headers = vec!["Interval"];
2567    let mut fixed = vec![];
2568    if segmented && (wide || medium) {
2569        headers.push("Segment");
2570        fixed.push(segment_width);
2571    }
2572    if wide {
2573        headers.extend(["Both ends", "Missing s/e", "p50", "p95"]);
2574        fixed.extend([9, 12, 7, 7]);
2575    } else {
2576        headers.push("p50");
2577        fixed.push(7);
2578    }
2579    headers.push(last_header);
2580    fixed.push(if wide { 15 } else { 9 });
2581    // What the interval's name has left: a long one ends in an ellipsis rather
2582    // than losing its last letters unmarked, and the detail says it in full.
2583    let interval_width = (body.width as usize).saturating_sub(
2584        glyphs::display_width(glyphs::get().selector)
2585            + fixed.iter().map(|width| *width as usize + 1).sum::<usize>(),
2586    );
2587    let rows = profiles.iter().map(|profile| {
2588        let mut cells = vec![glyphs::fit(&profile.label(), interval_width)];
2589        if segmented && (wide || medium) {
2590            cells.push(glyphs::fit(
2591                &segment_text(&profile.segment),
2592                segment_width as usize,
2593            ));
2594        }
2595        if wide {
2596            cells.extend([
2597                numfmt::group_chrome(profile.paired_rows),
2598                format!(
2599                    "{} / {}",
2600                    numfmt::group_chrome(profile.missing_start),
2601                    numfmt::group_chrome(profile.missing_end)
2602                ),
2603                crate::numfmt::duration_or_dash(profile.p50_seconds),
2604                crate::numfmt::duration_or_dash(profile.p95_seconds),
2605            ]);
2606        } else {
2607            cells.push(crate::numfmt::duration_or_dash(profile.p50_seconds));
2608        }
2609        cells.push(last(profile, wide));
2610        Row::new(cells)
2611    });
2612    let widths = std::iter::once(Constraint::Fill(1))
2613        .chain(fixed.into_iter().map(Constraint::Length))
2614        .collect::<Vec<_>>();
2615    normalize_selection(table_state, profiles.len());
2616    let table = Table::new(rows, widths)
2617        .header(Row::new(headers).style(dimmed))
2618        .row_highlight_style(theme.highlight_style())
2619        .highlight_symbol(glyphs::get().selector);
2620    StatefulWidget::render(table, body, buf, table_state);
2621}
2622
2623/// One interval in one segment: endpoints and every count with its denominator.
2624/// Missing ends, unparsed text and negative durations stay separate. Counts with rows
2625/// take the cursor; Enter opens them. Reads nothing.
2626fn render_interval_detail(
2627    config: &DataQualityWidgetConfig<'_>,
2628    table_state: &mut TableState,
2629    area: Rect,
2630    buf: &mut Buffer,
2631) {
2632    let Some(results) = config.results else {
2633        render_run_prompt(area, config.theme, buf);
2634        return;
2635    };
2636    let Some(profile) = results.temporal.get(config.interval_index) else {
2637        return;
2638    };
2639    let theme = config.theme;
2640    let plan = config.plan;
2641    let [title, body] = Layout::default()
2642        .direction(Direction::Vertical)
2643        .constraints([Constraint::Length(2), Constraint::Fill(1)])
2644        .margin(1)
2645        .areas(area);
2646    Paragraph::new(rule_line(
2647        &profile.label(),
2648        Some(&segment_text(&profile.segment)),
2649        title.width,
2650        theme,
2651    ))
2652    .render(title, buf);
2653
2654    let endpoint = |role: TemporalRole, column: &str| match plan.time_format(column) {
2655        Some(format) if format.zoned() => format!("{}: {column}, text with offset", role.label()),
2656        Some(format) => format!(
2657            "{}: {column}, text as {}",
2658            role.label(),
2659            format.kind.label()
2660        ),
2661        None => format!("{}: {column}", role.label()),
2662    };
2663    let grain = plan.interval_grain(&profile.start_column, &profile.end_column);
2664    let facts = IntervalFact::ALL
2665        .into_iter()
2666        .filter_map(|fact| profile.count(fact, plan).map(|count| (fact, count)))
2667        .collect::<Vec<_>>();
2668    let selected = facts
2669        .get(
2670            table_state
2671                .selected()
2672                .unwrap_or(0)
2673                .min(facts.len().saturating_sub(1)),
2674        )
2675        .map(|(fact, _)| *fact);
2676    let plain = |label: &str, value: String| {
2677        (
2678            FieldRow {
2679                mark: None,
2680                label: label.to_string(),
2681                value,
2682            },
2683            None,
2684        )
2685    };
2686    let fact_row = |fact: IntervalFact| {
2687        let (count, of) = profile.count(fact, plan)?;
2688        let value = match fact {
2689            IntervalFact::Negative | IntervalFact::Zero | IntervalFact::OverThreshold => {
2690                format!("{} with both ends", share(count, of))
2691            }
2692            _ => share(count, of),
2693        };
2694        let mark = (selected == Some(fact))
2695            .then(|| Span::styled(glyphs::get().rail, Style::default().fg(theme.accent())));
2696        Some((
2697            FieldRow {
2698                mark,
2699                label: fact.label(profile),
2700                value,
2701            },
2702            Some(fact),
2703        ))
2704    };
2705    let mut rows = vec![
2706        plain("Start", endpoint(profile.start_role, &profile.start_column)),
2707        plain("End", endpoint(profile.end_role, &profile.end_column)),
2708    ];
2709    if grain != QualityGrain::Dataset {
2710        rows.push(plain(
2711            "Segment",
2712            format!("{}, {}", segment_text(&profile.segment), grain.label()),
2713        ));
2714    }
2715    rows.push(plain("Rows", numfmt::group_chrome(profile.evaluated_rows)));
2716    rows.push(plain(
2717        "Both ends",
2718        share(profile.paired_rows, profile.evaluated_rows),
2719    ));
2720    rows.extend(
2721        [
2722            IntervalFact::MissingStart,
2723            IntervalFact::MissingEnd,
2724            IntervalFact::UnparsedStart,
2725            IntervalFact::UnparsedEnd,
2726            IntervalFact::Negative,
2727            IntervalFact::Zero,
2728        ]
2729        .into_iter()
2730        .filter_map(fact_row),
2731    );
2732    let pair = |left: Option<i64>, right: Option<i64>| {
2733        format!(
2734            "{}, {}",
2735            crate::numfmt::duration_or_dash(left),
2736            crate::numfmt::duration_or_dash(right)
2737        )
2738    };
2739    rows.push(plain(
2740        "p50, p90",
2741        pair(profile.p50_seconds, profile.p90_seconds),
2742    ));
2743    rows.push(plain(
2744        "p95, p99",
2745        pair(profile.p95_seconds, profile.p99_seconds),
2746    ));
2747    rows.push(plain(
2748        "Maximum",
2749        crate::numfmt::duration_or_dash(profile.max_seconds),
2750    ));
2751    rows.push(plain(
2752        "Threshold",
2753        match profile.threshold_seconds {
2754            Some(_) => format!(
2755                "duration > {}, strictly",
2756                crate::analysis::analysis_modal::threshold_label(profile.threshold_seconds)
2757            ),
2758            None => "none: set Latency over in Setup".to_string(),
2759        },
2760    ));
2761    rows.extend(fact_row(IntervalFact::OverThreshold));
2762
2763    let label_width = rows
2764        .iter()
2765        .map(|(row, _)| glyphs::display_width(&row.label))
2766        .max()
2767        .unwrap_or(0)
2768        + 2;
2769    let width = (body.width as usize).min(DETAIL_MEASURE);
2770    // Each row's lines, and where the selected count's start, so a short terminal
2771    // scrolls to keep it in view.
2772    let mut lines = Vec::new();
2773    let mut focus = 0;
2774    // Enter opens nothing in a row chunk or a file, so say why under the segment
2775    // rather than leave it to be found out.
2776    let closed = (!profile.segment_opens(plan)).then(|| {
2777        let what = match grain {
2778            QualityGrain::File => "a file",
2779            _ => "a row chunk",
2780        };
2781        Line::styled(
2782            glyphs::fit(
2783                &format!("  Rows do not open: {what} is not a filter value"),
2784                width,
2785            ),
2786            Style::default().fg(theme.dimmed()),
2787        )
2788    });
2789    for (row, fact) in &rows {
2790        if fact.is_some() && *fact == selected {
2791            focus = lines.len();
2792        }
2793        let mut row_lines = field_lines(std::slice::from_ref(row), label_width, width, true);
2794        if fact.is_some() && *fact == selected {
2795            for line in &mut row_lines {
2796                if let Some(label) = line.spans.get_mut(2) {
2797                    label.style = Style::default().fg(theme.accent());
2798                }
2799            }
2800        }
2801        lines.extend(row_lines);
2802        if row.label == "Segment" {
2803            lines.extend(closed.clone());
2804        }
2805    }
2806    let height = body.height as usize;
2807    let offset = if lines.len() > height {
2808        // Room for the count of what is below, and for the focused row; never
2809        // past the last line, which would leave the bottom blank.
2810        (focus + 2)
2811            .saturating_sub(height.saturating_sub(1))
2812            .min(lines.len() - height)
2813    } else {
2814        0
2815    };
2816    render_counted(lines.split_off(offset), body, theme, buf);
2817}
2818
2819/// Which starts and ends are measured, chosen from every pair the assigned roles
2820/// make, the suggested ones first. Space turns the pair under the cursor on or off.
2821fn render_interval_pairs(
2822    config: &DataQualityWidgetConfig<'_>,
2823    table_state: &mut TableState,
2824    area: Rect,
2825    buf: &mut Buffer,
2826) {
2827    let theme = config.theme;
2828    let plan = config.plan;
2829    let dimmed = Style::default().fg(theme.dimmed());
2830    let [title, body] = Layout::default()
2831        .direction(Direction::Vertical)
2832        .constraints([Constraint::Length(2), Constraint::Fill(1)])
2833        .margin(1)
2834        .areas(area);
2835    let candidates = plan.candidate_pairs();
2836    let chosen = plan.interval_pairs();
2837    Paragraph::new(rule_line(
2838        "Intervals",
2839        Some(&format!(
2840            "{} of {}",
2841            numfmt::group_chrome(chosen.len()),
2842            numfmt::group_chrome(candidates.len())
2843        )),
2844        title.width,
2845        theme,
2846    ))
2847    .render(title, buf);
2848    if candidates.is_empty() {
2849        Paragraph::new(Span::styled(
2850            glyphs::dotted("Needs 2 time roles · Time roles in Setup (e)"),
2851            dimmed,
2852        ))
2853        .wrap(Wrap { trim: true })
2854        .render(body, buf);
2855        return;
2856    }
2857    let g = glyphs::get();
2858    let label_width = candidates
2859        .iter()
2860        .map(|pair| glyphs::display_width(&interval_label(*pair)))
2861        .max()
2862        .unwrap_or(0) as u16
2863        + 2;
2864    // The selector, the box, the label and the gaps between them come first.
2865    let columns_width = (body.width as usize)
2866        .saturating_sub(glyphs::display_width(g.selector) + 1 + label_width as usize + 2);
2867    let rows = candidates.iter().map(|pair| {
2868        let on = chosen.contains(pair);
2869        let columns = format!(
2870            "{} to {}",
2871            plan.role_column(pair.0).unwrap_or_default(),
2872            plan.role_column(pair.1).unwrap_or_default()
2873        );
2874        Row::new(vec![
2875            Cell::from(if on { g.checkbox_on } else { g.checkbox_off }),
2876            Cell::from(interval_label(*pair)),
2877            Cell::from(Span::styled(glyphs::fit(&columns, columns_width), dimmed)),
2878        ])
2879    });
2880    table_state.select(Some(
2881        config.plan_field.min(candidates.len().saturating_sub(1)),
2882    ));
2883    let table = Table::new(
2884        rows,
2885        [
2886            Constraint::Length(glyphs::display_width(g.checkbox_on).max(1) as u16),
2887            Constraint::Length(label_width),
2888            Constraint::Fill(1),
2889        ],
2890    )
2891    .row_highlight_style(theme.highlight_style())
2892    .highlight_symbol(g.selector);
2893    StatefulWidget::render(table, body, buf, table_state);
2894}
2895
2896/// The width of the Trends table's range column.
2897const TREND_RANGE: u16 = 20;
2898
2899/// The Trends name column width and how many bars fit in `width`; shared with a bar's
2900/// detail so both lay out the same.
2901fn trend_layout(results: &DataQualityResults, width: u16) -> (u16, usize) {
2902    let name_width = results
2903        .columns
2904        .iter()
2905        .map(|profile| glyphs::display_width(&profile.name))
2906        .max()
2907        .unwrap_or(0)
2908        .clamp(12, 28) as u16
2909        + 2;
2910    let bars = width.saturating_sub(trend_lead(name_width)).max(8) as usize;
2911    (name_width, bars)
2912}
2913
2914/// Where a Trends line's first bar sits: past the selector, the name, the range
2915/// and the space after each.
2916fn trend_lead(name_width: u16) -> u16 {
2917    glyphs::display_width(glyphs::get().selector) as u16 + name_width + TREND_RANGE + 2
2918}
2919
2920/// What a segment of `grain` is called, many of them.
2921fn trend_unit(grain: &QualityGrain) -> &'static str {
2922    match grain {
2923        QualityGrain::TimeWindows { every, .. } => match every.as_str() {
2924            "1h" => "hours",
2925            "1d" => "days",
2926            "1w" => "weeks",
2927            _ => "months",
2928        },
2929        QualityGrain::Partition(_) => "partitions",
2930        QualityGrain::File => "files",
2931        _ => "chunks",
2932    }
2933}
2934
2935/// One of `unit`, or many.
2936fn counted_unit(count: usize, unit: &str) -> String {
2937    let unit = if count == 1 {
2938        unit.strip_suffix('s').unwrap_or(unit)
2939    } else {
2940        unit
2941    };
2942    format!("{} {unit}", numfmt::group_chrome(count))
2943}
2944
2945/// A bar's mark in `line`: its level, the unsampled mark where the sample drew no row
2946/// (never a short or blank bar), or blank where the measure does not apply. The exact
2947/// rows line has a level for every bar.
2948fn bar_mark(line: &TrendRow, bar: &TrendBar, index: usize, g: &glyphs::Glyphs) -> &'static str {
2949    if bar.evaluated == 0 && line.measure != TrendMeasure::Rows {
2950        return g.unsampled;
2951    }
2952    match line.bars.get(index).copied().flatten() {
2953        Some(value) if line.high > 0.0 => {
2954            g.mini_bars[((value / line.high) * 7.0).round().clamp(0.0, 7.0) as usize]
2955        }
2956        Some(_) => g.mini_bars[0],
2957        None => " ",
2958    }
2959}
2960
2961/// A line's lowest and highest bar, as the Range column says them.
2962fn trend_range(line: &TrendRow) -> String {
2963    if line.rows() {
2964        format!(
2965            "{} to {}",
2966            numfmt::group_chrome(line.low.round() as usize),
2967            numfmt::group_chrome(line.high.round() as usize)
2968        )
2969    } else if (line.high - line.low).abs() < 1e-9 {
2970        crate::numfmt::percent(line.high)
2971    } else {
2972        format!(
2973            "{} to {}",
2974            crate::numfmt::percent(line.low),
2975            crate::numfmt::percent(line.high)
2976        )
2977    }
2978}
2979
2980/// The coarser grain `w` stages, in words: "weekly".
2981fn coarser_label(grain: &QualityGrain) -> String {
2982    match grain {
2983        QualityGrain::TimeWindows { every, .. } => window_cadence(every).to_string(),
2984        other => other.label(),
2985    }
2986}
2987
2988/// What the Trends page says under its title: the span, how much of it the sample
2989/// reached, and the expected windows with no rows. Each fact a line.
2990fn trend_notes(
2991    config: &DataQualityWidgetConfig<'_>,
2992    results: &DataQualityResults,
2993    view: &TrendView<'_>,
2994) -> Vec<String> {
2995    let plan = config.plan;
2996    let unit = trend_unit(&plan.grain);
2997    let middot = glyphs::get().middot;
2998    let first = view.slots.first().map(|slot| slot.label).unwrap_or("");
2999    let last = view.slots.last().map(|slot| slot.label).unwrap_or("");
3000    let mut facts = vec![format!(
3001        "{} to {}, {}",
3002        segment_text(first),
3003        segment_text(last),
3004        counted_unit(view.slots.len(), unit)
3005    )];
3006    if view.per_bar > 1 {
3007        facts.push(format!("each bar {}", counted_unit(view.per_bar, unit)));
3008    }
3009    if view.sampled {
3010        facts.push("a bar's rate pools its sampled rows".to_string());
3011    }
3012    let mut notes = vec![facts.join(&format!(" {middot} "))];
3013    let (unsampled, thin) = view.coverage();
3014    if view.sampled && unsampled + thin > 0 {
3015        let mut reach = Vec::new();
3016        if unsampled > 0 {
3017            reach.push(format!(
3018                "{} of {} not sampled ({})",
3019                numfmt::group_chrome(unsampled),
3020                counted_unit(view.slots.len(), unit),
3021                glyphs::get().unsampled
3022            ));
3023        }
3024        if thin > 0 {
3025            reach.push(format!(
3026                "{} under {} sampled rows",
3027                numfmt::group_chrome(thin),
3028                crate::analysis::quality_report::THIN_SEGMENT_ROWS
3029            ));
3030        }
3031        let mut note = reach.join(", ");
3032        if let Some(coarser) = plan.coarser_grain() {
3033            note.push_str(&format!("; w stages {}", coarser_label(&coarser)));
3034        }
3035        notes.push(note);
3036    }
3037    if let Some(gaps) = crate::analysis::quality_trends::expected_gaps(plan, results) {
3038        notes.push(gaps_summary(plan, &gaps));
3039    }
3040    notes
3041}
3042
3043/// The expected windows in one line, for the Trends page.
3044fn gaps_summary(plan: &DataQualityPlan, gaps: &Gaps) -> String {
3045    let every = match &plan.grain {
3046        QualityGrain::TimeWindows { every, .. } => every.as_str(),
3047        _ => "",
3048    };
3049    let cadence = plan
3050        .expected
3051        .as_ref()
3052        .map(|expected| expected.cadence_label(every))
3053        .unwrap_or_default();
3054    let unit = trend_unit(&plan.grain);
3055    match gaps {
3056        Gaps::NoValues => format!("Expected {cadence}: file metadata only, no windows counted"),
3057        Gaps::NoWindows => {
3058            format!("Expected {cadence}: no window found, no range set")
3059        }
3060        Gaps::TooMany { windows } => format!(
3061            "Expected {cadence}: {} in range, over {} {dot} narrow in Setup (e)",
3062            counted_unit(*windows, unit),
3063            numfmt::group_chrome(crate::analysis::quality_trends::MAX_EXPECTED_WINDOWS),
3064            dot = glyphs::get().middot,
3065        ),
3066        // From typed past the last window found, with Before blank.
3067        Gaps::Checked(check) if check.expected == 0 => {
3068            format!("Expected {cadence}: no window in range")
3069        }
3070        Gaps::Checked(check) if check.gaps() == 0 => format!(
3071            "Expected {cadence}: all {} have rows",
3072            counted_unit(check.expected, unit)
3073        ),
3074        Gaps::Checked(check) => format!(
3075            "Expected {cadence}, {}: {} {} g lists",
3076            counted_unit(check.expected, unit),
3077            gap_counts(check),
3078            glyphs::get().middot,
3079        ),
3080    }
3081}
3082
3083/// "5 empty, 2 not sampled": each kind of gap there is, with its count.
3084fn gap_counts(check: &GapCheck) -> String {
3085    [
3086        (check.empty, GapKind::Empty),
3087        (check.unsampled, GapKind::Unsampled),
3088        (check.out_of_scope, GapKind::OutOfScope),
3089    ]
3090    .into_iter()
3091    .filter(|(count, _)| *count > 0)
3092    .map(|(count, kind)| format!("{} {}", numfmt::group_chrome(count), kind.label()))
3093    .collect::<Vec<_>>()
3094    .join(", ")
3095}
3096
3097/// Each column's measure across segments as a line of bars, segment rows first. The
3098/// whole range fits: a bar pools consecutive segments. Missed segments get their own
3099/// mark; notes count them and empty expected windows.
3100fn render_trend_table(
3101    config: &DataQualityWidgetConfig<'_>,
3102    results: &DataQualityResults,
3103    table_state: &mut TableState,
3104    area: Rect,
3105    buf: &mut Buffer,
3106) {
3107    let theme = config.theme;
3108    let dimmed = Style::default().fg(theme.dimmed());
3109    let (name_width, bars) = trend_layout(results, area.width);
3110    let view = trend_view(results, config.metric, bars);
3111    let notes = trend_notes(config, results, &view)
3112        .iter()
3113        .flat_map(|note| crate::widgets::info::wrap_to(note, area.width as usize))
3114        .collect::<Vec<_>>();
3115    // A short terminal keeps the table: the notes give way first.
3116    let room = (area.height as usize).saturating_sub(5).max(1);
3117    let notes = notes.into_iter().take(room).collect::<Vec<_>>();
3118    let [title, note, table_area] = Layout::default()
3119        .direction(Direction::Vertical)
3120        .constraints([
3121            Constraint::Length(1),
3122            Constraint::Length(notes.len() as u16 + 1),
3123            Constraint::Fill(1),
3124        ])
3125        .areas(area);
3126    Paragraph::new(rule_line(
3127        &format!("{} over time", config.metric.label()),
3128        Some(&numfmt::group_chrome(
3129            view.lines
3130                .iter()
3131                .filter(|row| !row.rows())
3132                .map(|row| row.names.len())
3133                .sum::<usize>(),
3134        )),
3135        title.width,
3136        theme,
3137    ))
3138    .render(title, buf);
3139    Paragraph::new(
3140        notes
3141            .iter()
3142            .map(|text| Line::styled(glyphs::fit(text, note.width as usize), dimmed))
3143            .collect::<Vec<_>>(),
3144    )
3145    .render(note, buf);
3146    let g = glyphs::get();
3147    let accent = Style::default().fg(theme.accent());
3148    let lines = view.lines.iter().map(|row| {
3149        let spark = view
3150            .bars
3151            .iter()
3152            .enumerate()
3153            .map(|(index, bar)| {
3154                let mark = bar_mark(row, bar, index, g);
3155                Span::styled(mark, if mark == g.unsampled { dimmed } else { accent })
3156            })
3157            .collect::<Vec<_>>();
3158        let name =
3159            crate::analysis::quality_report::columns_label(&row.names, name_width as usize - 2);
3160        Row::new(vec![
3161            Cell::from(if row.rows() {
3162                Span::styled(name, dimmed)
3163            } else {
3164                Span::raw(name)
3165            }),
3166            Cell::from(Span::styled(trend_range(row), dimmed)),
3167            Cell::from(Line::from(spark)),
3168        ])
3169    });
3170    normalize_selection(table_state, view.lines.len());
3171    let table = Table::new(
3172        lines,
3173        [
3174            Constraint::Length(name_width),
3175            Constraint::Length(TREND_RANGE),
3176            Constraint::Fill(1),
3177        ],
3178    )
3179    .header(Row::new(["Column", "Range", "Trend"]).style(dimmed))
3180    .row_highlight_style(theme.highlight_style())
3181    .highlight_symbol(g.selector);
3182    StatefulWidget::render(table, table_area, buf, table_state);
3183}
3184
3185/// One bar of a Trends line, drawn under the line with a pointer: its span, segments
3186/// pooled and read, value and denominator, sample confidence, and comparison. Reads
3187/// nothing; ↑↓ walk the bars.
3188fn render_trend_detail(
3189    config: &DataQualityWidgetConfig<'_>,
3190    table_state: &mut TableState,
3191    area: Rect,
3192    buf: &mut Buffer,
3193) {
3194    let Some(results) = config.results else {
3195        render_run_prompt(area, config.theme, buf);
3196        return;
3197    };
3198    let theme = config.theme;
3199    let g = glyphs::get();
3200    let dimmed = Style::default().fg(theme.dimmed());
3201    let accent = Style::default().fg(theme.accent());
3202    let [area] = Layout::default()
3203        .direction(Direction::Vertical)
3204        .constraints([Constraint::Fill(1)])
3205        .margin(1)
3206        .areas(area);
3207    let (name_width, bars) = trend_layout(results, area.width);
3208    let view = trend_view(results, config.metric, bars);
3209    let (Some(line), false) = (view.lines.get(config.trend_line), view.bars.is_empty()) else {
3210        Paragraph::new(glyphs::dotted("No bars · Esc Trends"))
3211            .style(Style::default().fg(theme.text_primary()))
3212            .render(area, buf);
3213        return;
3214    };
3215    // Held to the last bar: the keys only know an upper bound, the width decides.
3216    let index = table_state.selected().unwrap_or(0).min(view.bars.len() - 1);
3217    table_state.select(Some(index));
3218    let [title, spark, pointer, _, body] = Layout::default()
3219        .direction(Direction::Vertical)
3220        .constraints([
3221            Constraint::Length(1),
3222            Constraint::Length(1),
3223            Constraint::Length(1),
3224            Constraint::Length(1),
3225            Constraint::Fill(1),
3226        ])
3227        .areas(area);
3228    let measure = match line.measure {
3229        TrendMeasure::Rows => "Rows per segment".to_string(),
3230        TrendMeasure::SampledRows => "Sampled rows per segment".to_string(),
3231        TrendMeasure::Column(_) => config.metric.label().to_string(),
3232    };
3233    Paragraph::new(rule_line(
3234        &measure,
3235        Some(&format!(
3236            "bar {} of {}",
3237            numfmt::group_chrome(index + 1),
3238            numfmt::group_chrome(view.bars.len())
3239        )),
3240        title.width,
3241        theme,
3242    ))
3243    .render(title, buf);
3244
3245    // The line as Trends drew it, its selected bar pointed at from below.
3246    let name = crate::analysis::quality_report::columns_label(&line.names, name_width as usize - 2);
3247    let lead = |text: String, width: u16| {
3248        format!(
3249            "{text}{}",
3250            " ".repeat((width as usize).saturating_sub(glyphs::display_width(&text)))
3251        )
3252    };
3253    let mut spans = vec![
3254        Span::raw(g.selector_blank),
3255        Span::raw(lead(name, name_width + 1)),
3256        Span::styled(lead(trend_range(line), TREND_RANGE + 1), dimmed),
3257    ];
3258    spans.extend(view.bars.iter().enumerate().map(|(at, candidate)| {
3259        let mark = bar_mark(line, candidate, at, g);
3260        let style = if mark == g.unsampled { dimmed } else { accent };
3261        Span::styled(
3262            mark,
3263            if at == index {
3264                style.add_modifier(Modifier::BOLD)
3265            } else {
3266                style
3267            },
3268        )
3269    }));
3270    Paragraph::new(Line::from(spans)).render(spark, buf);
3271    Paragraph::new(Line::from(vec![
3272        Span::raw(" ".repeat(trend_lead(name_width) as usize + index)),
3273        Span::styled(g.pointer, accent),
3274    ]))
3275    .render(pointer, buf);
3276
3277    let rows = trend_bar_fields(config, results, &view, line, index);
3278    let label_width = rows
3279        .iter()
3280        .map(|row| glyphs::display_width(&row.label))
3281        .max()
3282        .unwrap_or(0)
3283        + 2;
3284    let width = (body.width as usize).min(DETAIL_MEASURE);
3285    render_counted(
3286        field_lines(&rows, label_width, width, false),
3287        body,
3288        theme,
3289        buf,
3290    );
3291}
3292
3293/// A bar's facts as label and value rows.
3294fn trend_bar_fields(
3295    config: &DataQualityWidgetConfig<'_>,
3296    results: &DataQualityResults,
3297    view: &TrendView<'_>,
3298    line: &TrendRow,
3299    index: usize,
3300) -> Vec<FieldRow> {
3301    let plan = config.plan;
3302    let bar = &view.bars[index];
3303    let unit = trend_unit(&plan.grain);
3304    let row = |label: &str, value: String| FieldRow {
3305        mark: None,
3306        label: label.to_string(),
3307        value,
3308    };
3309    let mut rows = vec![row(
3310        "Span",
3311        segment_text(&crate::analysis::quality_trends::bar_span(
3312            view,
3313            bar,
3314            &plan.grain,
3315        )),
3316    )];
3317    // How much of the bar the run reached, before any number taken from it.
3318    let mut reach = vec![counted_unit(bar.segments(), unit)];
3319    if view.sampled {
3320        if bar.unsampled > 0 {
3321            reach.push(format!(
3322                "{} not sampled",
3323                numfmt::group_chrome(bar.unsampled)
3324            ));
3325        }
3326        if bar.thin > 0 {
3327            reach.push(format!(
3328                "{} under {} sampled rows",
3329                numfmt::group_chrome(bar.thin),
3330                crate::analysis::quality_report::THIN_SEGMENT_ROWS
3331            ));
3332        }
3333        if bar.unsampled + bar.thin == 0 {
3334            reach.push("each sampled".to_string());
3335        }
3336    }
3337    rows.push(row("Segments", reach.join(", ")));
3338    let every_row = !view.sampled || bar.eligible == Some(bar.evaluated);
3339    rows.push(row(
3340        "Rows",
3341        match (view.sampled, bar.eligible) {
3342            (false, Some(eligible)) => {
3343                format!("{}, every row read", numfmt::group_chrome(eligible))
3344            }
3345            (false, None) => format!("{}, every row read", numfmt::group_chrome(bar.evaluated)),
3346            (true, Some(eligible)) => format!(
3347                "{} sampled of {} ({})",
3348                numfmt::group_chrome(bar.evaluated),
3349                numfmt::group_chrome(eligible),
3350                crate::numfmt::percent_of(bar.evaluated, eligible)
3351            ),
3352            (true, None) => format!(
3353                "{} sampled, total not counted",
3354                numfmt::group_chrome(bar.evaluated)
3355            ),
3356        },
3357    ));
3358    let compared = crate::analysis::quality_trends::compared_bar(view, index, plan);
3359    let against = if plan.comparison == QualityComparison::Baseline {
3360        "Baseline bar"
3361    } else {
3362        "Previous bar"
3363    };
3364    if line.rows() {
3365        // Rows per segment: the mean, and the smallest and largest segment.
3366        let counts = view.slots[bar.slots.clone()]
3367            .iter()
3368            .map(|slot| match line.measure {
3369                TrendMeasure::Rows => slot.total.unwrap_or(0),
3370                _ => slot.evaluated,
3371            })
3372            .collect::<Vec<_>>();
3373        let (low, high) = (
3374            counts.iter().min().copied().unwrap_or(0),
3375            counts.iter().max().copied().unwrap_or(0),
3376        );
3377        let mean = line.bars[index].unwrap_or(0.0);
3378        rows.push(row(
3379            "Per segment",
3380            if low == high {
3381                numfmt::group_chrome(low)
3382            } else {
3383                format!(
3384                    "{} on average, {} to {}",
3385                    numfmt::group_chrome(mean.round() as usize),
3386                    numfmt::group_chrome(low),
3387                    numfmt::group_chrome(high)
3388                )
3389            },
3390        ));
3391        if let Some(other) = compared {
3392            let before = line.bars[other].unwrap_or(0.0);
3393            let moved = if before > 0.0 {
3394                format!(" ({:+.1}%)", (mean - before) / before * 100.0)
3395            } else {
3396                String::new()
3397            };
3398            rows.push(row(
3399                against,
3400                format!(
3401                    "{} to {} per segment{moved}",
3402                    numfmt::group_chrome(before.round() as usize),
3403                    numfmt::group_chrome(mean.round() as usize)
3404                ),
3405            ));
3406        }
3407        return rows;
3408    }
3409    let (count, of) = line.parts[index];
3410    let noun = match config.metric {
3411        QualityMetric::DistinctShare
3412        | QualityMetric::IntegerParseShare
3413        | QualityMetric::DecimalParseShare => "values",
3414        _ => "rows",
3415    };
3416    rows.push(row(
3417        config.metric.label(),
3418        if of > 0.0 {
3419            format!(
3420                "{} of {} {noun} ({})",
3421                numfmt::group_chrome(count.round() as usize),
3422                numfmt::group_chrome(of.round() as usize),
3423                crate::numfmt::percent(count / of)
3424            )
3425        } else if bar.evaluated == 0 {
3426            "none: no row sampled".to_string()
3427        } else {
3428            format!("none: no {noun} to measure")
3429        },
3430    ));
3431    // How sure a sample is of the rate: never a point, and never for a distinct
3432    // share, which does not stand for the whole.
3433    let interval = if of <= 0.0 {
3434        None
3435    } else if every_row {
3436        Some("none needed: every row read".to_string())
3437    } else if config.metric == QualityMetric::DistinctShare {
3438        Some("none: a distinct share does not stand for the whole".to_string())
3439    } else {
3440        crate::analysis::quality_trends::wilson_interval(count, of).map(|(low, high)| {
3441            let mut text = format!(
3442                "{} to {}",
3443                crate::numfmt::percent(low),
3444                crate::numfmt::percent(high)
3445            );
3446            if (of as usize) < crate::analysis::quality_report::THIN_SEGMENT_ROWS {
3447                text.push_str(&format!(
3448                    ", from under {} {noun}",
3449                    crate::analysis::quality_report::THIN_SEGMENT_ROWS
3450                ));
3451            }
3452            text
3453        })
3454    };
3455    if let Some(interval) = interval {
3456        rows.push(row("95% interval", interval));
3457    }
3458    if let Some(other) = compared {
3459        let other_bar = &view.bars[other];
3460        let exact = results.precision == QualityPrecision::Exact
3461            || (every_row && other_bar.eligible == Some(other_bar.evaluated));
3462        rows.push(row(
3463            against,
3464            match crate::analysis::quality_trends::bar_change(line, index, other, exact) {
3465                Some(change) => format!(
3466                    "{} to {}, {:+.1} points: {}",
3467                    crate::numfmt::percent(change.before),
3468                    crate::numfmt::percent(change.now),
3469                    change.points(),
3470                    // Segments never judge a distinct share: it falls as a segment
3471                    // grows, so bars of different sizes differ by it whatever the data.
3472                    if config.metric == QualityMetric::DistinctShare {
3473                        "not judged for a distinct share"
3474                    } else if change.clear {
3475                        "a clear change"
3476                    } else if change.points().abs()
3477                        < crate::analysis::data_quality::MATERIAL_CHANGE_PP
3478                    {
3479                        "under a point"
3480                    } else {
3481                        "within sampling noise"
3482                    }
3483                ),
3484                None => "nothing to compare: one side has no rate".to_string(),
3485            },
3486        ));
3487    }
3488    rows
3489}
3490
3491/// The expected windows with no rows, a line per run of them: why, which windows,
3492/// and the rows a sample missed. Bounded: the runs past the cap are counted.
3493fn render_gaps(
3494    config: &DataQualityWidgetConfig<'_>,
3495    table_state: &mut TableState,
3496    area: Rect,
3497    buf: &mut Buffer,
3498) {
3499    let Some(results) = config.results else {
3500        render_run_prompt(area, config.theme, buf);
3501        return;
3502    };
3503    let theme = config.theme;
3504    let plan = config.plan;
3505    let dimmed = Style::default().fg(theme.dimmed());
3506    let [area] = Layout::default()
3507        .direction(Direction::Vertical)
3508        .constraints([Constraint::Fill(1)])
3509        .margin(1)
3510        .areas(area);
3511    let Some(gaps) = crate::analysis::quality_trends::expected_gaps(plan, results) else {
3512        Paragraph::new(
3513            "No expected windows: set Expected in Setup (e) to say which windows rows \
3514             belong in.",
3515        )
3516        .wrap(Wrap { trim: true })
3517        .style(Style::default().fg(theme.text_primary()))
3518        .render(area, buf);
3519        return;
3520    };
3521    let check = match &gaps {
3522        Gaps::Checked(check) if check.expected > 0 => check,
3523        _ => {
3524            Paragraph::new(gaps_summary(plan, &gaps))
3525                .wrap(Wrap { trim: true })
3526                .style(Style::default().fg(theme.text_primary()))
3527                .render(area, buf);
3528            return;
3529        }
3530    };
3531    let unit = trend_unit(&plan.grain);
3532    let every = check.every.as_str();
3533    let sampled = matches!(results.precision, QualityPrecision::Sampled);
3534    let mut notes = vec![format!(
3535        "{}, {}: {} of {} {}",
3536        check.column,
3537        crate::analysis::quality_trends::calendar_span(
3538            check.from,
3539            crate::analysis::quality_trends::floor_window(
3540                check.before - chrono::Duration::microseconds(1),
3541                every
3542            ),
3543            every
3544        ),
3545        numfmt::group_chrome(check.with_rows),
3546        counted_unit(check.expected, unit),
3547        if sampled { "sampled" } else { "with rows" }
3548    )];
3549    if check.weekend > 0 {
3550        notes.push(format!(
3551            "{} on weekends, not expected",
3552            counted_unit(check.weekend, unit)
3553        ));
3554    }
3555    // What each kind means, for the kinds listed: empty only on an exact count.
3556    let scope = plan.scope.label();
3557    for (count, text) in [
3558        (
3559            check.empty,
3560            format!("Empty: no rows in {scope}, by exact count"),
3561        ),
3562        (
3563            check.unsampled,
3564            match missed_rows(check) {
3565                Some(rows) if check.counted => format!(
3566                    "Not sampled: {} rows there, none drawn",
3567                    numfmt::group_chrome(rows)
3568                ),
3569                None if check.counted => "Not sampled: rows there, none drawn".to_string(),
3570                _ => "Not sampled: none drawn, rows not counted".to_string(),
3571            },
3572        ),
3573        (
3574            check.out_of_scope,
3575            "Out of scope: outside the time range the scope reads".to_string(),
3576        ),
3577    ] {
3578        if count > 0 {
3579            notes.push(text);
3580        }
3581    }
3582    let notes = notes
3583        .iter()
3584        .flat_map(|note| crate::widgets::info::wrap_to(note, area.width as usize))
3585        .take((area.height as usize).saturating_sub(5).max(1))
3586        .collect::<Vec<_>>();
3587    let [title, note, body] = Layout::default()
3588        .direction(Direction::Vertical)
3589        .constraints([
3590            Constraint::Length(1),
3591            Constraint::Length(notes.len() as u16 + 1),
3592            Constraint::Fill(1),
3593        ])
3594        .areas(area);
3595    Paragraph::new(rule_line(
3596        &format!(
3597            "Expected {}",
3598            plan.expected
3599                .as_ref()
3600                .map(|expected| expected.cadence_label(every))
3601                .unwrap_or_default()
3602        ),
3603        Some(&format!(
3604            "{} {}",
3605            numfmt::group_chrome(check.gaps()),
3606            if check.gaps() == 1 { "gap" } else { "gaps" }
3607        )),
3608        title.width,
3609        theme,
3610    ))
3611    .render(title, buf);
3612    Paragraph::new(
3613        notes
3614            .iter()
3615            .map(|text| Line::styled(glyphs::fit(text, note.width as usize), dimmed))
3616            .collect::<Vec<_>>(),
3617    )
3618    .render(note, buf);
3619    if check.runs.is_empty() {
3620        Paragraph::new("Every expected window has rows")
3621            .style(Style::default().fg(theme.text_primary()))
3622            .render(body, buf);
3623        return;
3624    }
3625    // The windows first, which is what a gap is; the rows a sample missed where
3626    // there is room for them.
3627    let span_width = if every == "1h" { 35 } else { 24 };
3628    let wide = body.width as usize >= span_width + 2 + 12 + 10 + 14 + 4;
3629    let rows = check
3630        .runs
3631        .iter()
3632        .map(|run| {
3633            let mut cells = vec![
3634                Cell::from(crate::analysis::quality_trends::calendar_span(
3635                    run.first, run.last, every,
3636                )),
3637                Cell::from(run.kind.label()),
3638                Cell::from(Span::styled(counted_unit(run.windows, unit), dimmed)),
3639            ];
3640            if wide {
3641                cells.push(Cell::from(Span::styled(
3642                    run.rows
3643                        .map(|rows| format!("{} rows", numfmt::group_chrome(rows)))
3644                        .unwrap_or_default(),
3645                    dimmed,
3646                )));
3647            }
3648            Row::new(cells)
3649        })
3650        .chain((check.more_runs > 0).then(|| {
3651            Row::new(vec![Cell::from(Span::styled(
3652                format!(
3653                    "{} {} more",
3654                    glyphs::get().ellipsis,
3655                    numfmt::group_chrome(check.more_runs)
3656                ),
3657                dimmed,
3658            ))])
3659        }))
3660        .collect::<Vec<_>>();
3661    normalize_selection(table_state, check.runs.len());
3662    let mut widths = vec![
3663        Constraint::Length(span_width as u16),
3664        Constraint::Length(12),
3665        Constraint::Length(10),
3666    ];
3667    let mut headers = vec!["Windows", "Gap", "Length"];
3668    if wide {
3669        widths.push(Constraint::Length(14));
3670        headers.push("Rows");
3671    }
3672    let table = Table::new(rows, widths)
3673        .header(Row::new(headers).style(dimmed))
3674        .row_highlight_style(theme.highlight_style())
3675        .highlight_symbol(glyphs::get().selector);
3676    StatefulWidget::render(table, body, buf, table_state);
3677}
3678
3679/// The rows in the windows a sample missed, when every one was counted.
3680fn missed_rows(check: &GapCheck) -> Option<usize> {
3681    check
3682        .runs
3683        .iter()
3684        .filter(|run| run.kind == GapKind::Unsampled)
3685        .map(|run| run.rows)
3686        .sum::<Option<usize>>()
3687        .filter(|_| check.more_runs == 0)
3688}
3689
3690/// Setup's Expected editor: which windows rows are expected in, and from and before
3691/// when. Nothing here reads; Enter writes it into the draft.
3692fn render_expected_windows(config: &DataQualityWidgetConfig<'_>, area: Rect, buf: &mut Buffer) {
3693    let ctx = config.ctx;
3694    let theme = config.theme;
3695    let Some(form) = config.expected_form else {
3696        return;
3697    };
3698    let [area] = Layout::default()
3699        .direction(Direction::Vertical)
3700        .constraints([Constraint::Fill(1)])
3701        .margin(1)
3702        .areas(area);
3703    let every = match &config.plan.grain {
3704        QualityGrain::TimeWindows { every, .. } => every.as_str(),
3705        _ => "1d",
3706    };
3707    let notes = [
3708        format!("Grain: {}", config.plan.grain.label()),
3709        "From, Before: a date or a UTC timestamp, such as 2024-01-01 or \
3710         2024-01-01T09:00:00Z. Blank: the first or last window found."
3711            .to_string(),
3712        "A window in range with no rows is listed under Trends (g); a weekend only \
3713         when every day is expected."
3714            .to_string(),
3715    ];
3716    let width = (area.width as usize).min(DETAIL_MEASURE);
3717    let mut y = area.y;
3718    put_line(
3719        rule_line("Expected Windows", None, area.width, theme),
3720        area,
3721        &mut y,
3722        buf,
3723    );
3724    put_line(Line::raw(""), area, &mut y, buf);
3725    let cadence = form.cadence_label(every);
3726    for (field, label) in crate::analysis::analysis_modal::EXPECTED_ROWS
3727        .iter()
3728        .enumerate()
3729    {
3730        let row_area = Rect {
3731            y,
3732            height: 1,
3733            ..area
3734        };
3735        if y < area.y + area.height {
3736            crate::widgets::ui::FormRow {
3737                label,
3738                value: match field {
3739                    0 => FormValue::Choice(&cadence),
3740                    1 => FormValue::Input(&form.from),
3741                    _ => FormValue::Input(&form.before),
3742                },
3743                focused: form.field == field,
3744                label_width: 10,
3745            }
3746            .render(row_area, buf, ctx);
3747            crate::app::pointer::record_field::<crate::analysis::analysis_modal::ExpectedForm>(
3748                row_area, field,
3749            );
3750        }
3751        y += 1;
3752    }
3753    put_line(Line::raw(""), area, &mut y, buf);
3754    let dimmed = Style::default().fg(theme.dimmed());
3755    for note in &notes {
3756        for text in crate::widgets::info::wrap_to(note, width.saturating_sub(2)) {
3757            put_line(Line::styled(format!("  {text}"), dimmed), area, &mut y, buf);
3758        }
3759    }
3760    if let Some(error) = &form.error {
3761        put_line(Line::raw(""), area, &mut y, buf);
3762        put_line(
3763            Line::styled(
3764                glyphs::fit(error, width),
3765                Style::default().fg(theme.warning()),
3766            ),
3767            area,
3768            &mut y,
3769            buf,
3770        );
3771    }
3772}
3773
3774/// `line` on row `y` of `area` when it is inside it, and the next row after.
3775fn put_line(line: Line<'static>, area: Rect, y: &mut u16, buf: &mut Buffer) {
3776    if *y < area.y + area.height {
3777        Paragraph::new(line).render(
3778            Rect {
3779                y: *y,
3780                height: 1,
3781                ..area
3782            },
3783            buf,
3784        );
3785    }
3786    *y += 1;
3787}
3788
3789/// One column's findings, then its measurements, as aligned label and value rows
3790/// on the page, the way Setup lays out its rows.
3791fn render_detail(
3792    config: &DataQualityWidgetConfig<'_>,
3793    table_state: &mut TableState,
3794    area: Rect,
3795    buf: &mut Buffer,
3796) {
3797    let Some(results) = config.results else {
3798        render_run_prompt(area, config.theme, buf);
3799        return;
3800    };
3801    let index = table_state.selected().unwrap_or(0);
3802    let Some(profile) = results.columns.get(index) else {
3803        Paragraph::new("No column selected")
3804            .alignment(Alignment::Center)
3805            .render(area, buf);
3806        return;
3807    };
3808    let theme = config.theme;
3809    let [title, body] = Layout::default()
3810        .direction(Direction::Vertical)
3811        .constraints([Constraint::Length(2), Constraint::Fill(1)])
3812        .margin(1)
3813        .areas(area);
3814    Paragraph::new(rule_line(&profile.name, None, title.width, theme)).render(title, buf);
3815
3816    // The column's findings first, in the report's own words; the measurements
3817    // under them are the evidence.
3818    let report = results.report();
3819    let mut findings = report
3820        .findings
3821        .iter()
3822        .filter(|finding| finding.kind.is_some() && finding.columns.contains(&profile.name))
3823        .map(|finding| FieldRow {
3824            mark: Some(severity_mark(finding.severity, theme)),
3825            label: finding.title.to_string(),
3826            value: finding.summary.clone(),
3827        })
3828        .collect::<Vec<_>>();
3829    if findings.is_empty() {
3830        findings.push(FieldRow {
3831            mark: Some(severity_mark(Severity::Clean, theme)),
3832            label: "No findings".to_string(),
3833            value: String::new(),
3834        });
3835    }
3836    let measurements = detail_measurements(config.ctx, results, profile);
3837    let label_width = findings
3838        .iter()
3839        .chain(&measurements)
3840        .map(|row| glyphs::display_width(&row.label))
3841        .max()
3842        .unwrap_or(0)
3843        + 2;
3844    // A reading surface: values wrap at a comfortable measure on a wide terminal.
3845    let width = (body.width as usize).min(DETAIL_MEASURE);
3846    let mut lines = field_lines(&findings, label_width, width, true);
3847    lines.push(Line::raw(""));
3848    lines.extend(field_lines(&measurements, label_width, width, true));
3849    render_counted(lines, body, theme, buf);
3850}
3851
3852/// The widest a reading surface's text runs, however wide the terminal.
3853const DETAIL_MEASURE: usize = 100;
3854
3855/// The most of one value the Detail page shows: a range's end, the most common
3856/// value, a spelling.
3857const END_WIDTH: usize = 32;
3858
3859/// A label and its value on one row, with an optional mark in the lead.
3860struct FieldRow {
3861    mark: Option<Span<'static>>,
3862    label: String,
3863    value: String,
3864}
3865
3866/// Rows in two aligned columns: label padded to `label_width`, value wrapped under
3867/// itself (multi-line values keep their lines). With `marks`, a two-column lead holds
3868/// each row's mark.
3869fn field_lines(
3870    rows: &[FieldRow],
3871    label_width: usize,
3872    width: usize,
3873    marks: bool,
3874) -> Vec<Line<'static>> {
3875    let lead = if marks { 2 } else { 0 };
3876    let value_width = width.saturating_sub(lead + label_width).max(8);
3877    let mut lines = Vec::new();
3878    for row in rows {
3879        let mut values = row
3880            .value
3881            .lines()
3882            // A value that fits is kept as written: wrapping splits on whitespace
3883            // and would drop the leading, trailing or doubled spaces a value holds.
3884            .flat_map(|line| {
3885                if glyphs::display_width(line) <= value_width {
3886                    vec![line.to_string()]
3887                } else {
3888                    crate::widgets::info::wrap_to(line, value_width)
3889                }
3890            })
3891            .collect::<Vec<_>>()
3892            .into_iter();
3893        let mut spans = Vec::new();
3894        if marks {
3895            spans.push(row.mark.clone().unwrap_or_else(|| Span::raw(" ")));
3896            spans.push(Span::raw(" "));
3897        }
3898        spans.push(Span::raw(format!(
3899            "{}{}",
3900            row.label,
3901            " ".repeat(label_width.saturating_sub(glyphs::display_width(&row.label)))
3902        )));
3903        spans.extend(
3904            values
3905                .next()
3906                .map(|value| Span::raw(glyphs::fit(&value, value_width))),
3907        );
3908        lines.push(Line::from(spans));
3909        lines.extend(values.map(|value| {
3910            Line::raw(format!(
3911                "{}{}",
3912                " ".repeat(lead + label_width),
3913                glyphs::fit(&value, value_width)
3914            ))
3915        }));
3916    }
3917    lines
3918}
3919
3920/// `lines` in `area`; when they run past it, the last row counts what is below
3921/// rather than showing half of it.
3922fn render_counted(mut lines: Vec<Line<'static>>, area: Rect, theme: &Theme, buf: &mut Buffer) {
3923    let height = area.height as usize;
3924    if lines.len() > height && height > 0 {
3925        let below = lines.len() - (height - 1);
3926        lines.truncate(height - 1);
3927        lines.push(Line::styled(
3928            format!("  {} {below} more", glyphs::get().ellipsis),
3929            Style::default().fg(theme.dimmed()),
3930        ));
3931    }
3932    Paragraph::new(lines).render(area, buf);
3933}
3934
3935/// What was measured on a column, table-formatted. Measures not applying to its type
3936/// are omitted (not dashed); header facts are not repeated.
3937fn detail_measurements(
3938    ctx: &RenderContext,
3939    results: &DataQualityResults,
3940    profile: &ColumnQualityProfile,
3941) -> Vec<FieldRow> {
3942    // Long text is cut, so both ends of a range and a count after a value stay
3943    // in view.
3944    let value = |text: &str| glyphs::fit(&table_value(ctx, profile, text), END_WIDTH);
3945    let row = |label: &str, value: String| FieldRow {
3946        mark: None,
3947        label: label.to_string(),
3948        value,
3949    };
3950    let mut rows = vec![
3951        row("Type", profile.dtype.to_string()),
3952        row(
3953            "Missing",
3954            if profile.null_count == 0 {
3955                "0".to_string()
3956            } else {
3957                format!(
3958                    "{} ({})",
3959                    numfmt::group_chrome(profile.null_count),
3960                    crate::numfmt::percent_of(profile.null_count, profile.evaluated_rows)
3961                )
3962            },
3963        ),
3964    ];
3965    for (label, measured) in [
3966        ("Distinct", profile.distinct_count),
3967        ("Empty text", profile.empty_count),
3968        ("Blank text", profile.whitespace_count),
3969        ("NaN", profile.nan_count),
3970    ] {
3971        if let Some(measured) = measured {
3972            rows.push(row(label, numfmt::group_chrome(measured)));
3973        }
3974    }
3975    if profile.positive_infinity_count.is_some() || profile.negative_infinity_count.is_some() {
3976        rows.push(row(
3977            "Infinite",
3978            format!(
3979                "{} positive, {} negative",
3980                count_label(profile.positive_infinity_count),
3981                count_label(profile.negative_infinity_count)
3982            ),
3983        ));
3984    }
3985    if profile.min.is_some() || profile.max.is_some() {
3986        rows.push(row(
3987            "Range",
3988            format!(
3989                "{} to {}",
3990                profile.min.as_deref().map(value).unwrap_or_default(),
3991                profile.max.as_deref().map(value).unwrap_or_default()
3992            ),
3993        ));
3994    }
3995    if let Some((dominant, count)) = profile.dominant_value.as_ref().zip(profile.dominant_count) {
3996        rows.push(row(
3997            "Most common",
3998            format!(
3999                "{}, {} {}",
4000                value(dominant),
4001                numfmt::group_chrome(count),
4002                if count == 1 { "row" } else { "rows" }
4003            ),
4004        ));
4005    }
4006    if profile.min_length.is_some() || profile.max_length.is_some() {
4007        rows.push(row(
4008            if matches!(profile.dtype, polars::prelude::DataType::List(_)) {
4009                "List length"
4010            } else {
4011                "Text length"
4012            },
4013            format!(
4014                "{} to {}",
4015                count_label(profile.min_length),
4016                count_label(profile.max_length)
4017            ),
4018        ));
4019    }
4020    // Only the parsers that accepted something: four zeros say less than "none".
4021    let parses = [
4022        ("integer", profile.integer_parse_count),
4023        ("decimal", profile.decimal_parse_count),
4024        ("date", profile.date_parse_count),
4025        ("datetime", profile.datetime_parse_count),
4026    ];
4027    if parses.iter().any(|(_, count)| count.is_some()) {
4028        let found = parses
4029            .iter()
4030            .filter_map(|(parser, count)| {
4031                count
4032                    .filter(|count| *count > 0)
4033                    .map(|count| format!("{parser} {}", numfmt::group_chrome(count)))
4034            })
4035            .collect::<Vec<_>>();
4036        rows.push(row(
4037            "Parses as",
4038            if found.is_empty() {
4039                "none".to_string()
4040            } else {
4041                found.join(", ")
4042            },
4043        ));
4044    }
4045    for group in results
4046        .category_variants
4047        .iter()
4048        .filter(|group| group.column == profile.name)
4049        .take(3)
4050    {
4051        // Spellings differ by case and by spaces the table does not show, so they are
4052        // quoted as the finding quotes them: "West " and "West" read apart.
4053        rows.push(row(
4054            "Spellings",
4055            group
4056                .variants
4057                .iter()
4058                .map(|(variant, count)| {
4059                    format!(
4060                        "{} ({})",
4061                        crate::analysis::quality_report::quoted(variant, END_WIDTH),
4062                        numfmt::group_chrome(*count)
4063                    )
4064                })
4065                .collect::<Vec<_>>()
4066                .join("\n"),
4067        ));
4068    }
4069    rows
4070}
4071
4072/// A value the profile holds as text, formatted as the table formats its column:
4073/// grouped, or to fixed places, when the number format says so; as-is otherwise.
4074fn table_value(ctx: &RenderContext, profile: &ColumnQualityProfile, text: &str) -> String {
4075    use polars::prelude::{AnyValue, DataType};
4076    let formatter = ctx
4077        .number_format
4078        .formatter_for(&profile.name, &profile.dtype);
4079    if formatter.is_passthrough() {
4080        return text.to_string();
4081    }
4082    let parsed = match profile.dtype {
4083        DataType::Float32 | DataType::Float64 => text.parse::<f64>().ok().map(AnyValue::Float64),
4084        DataType::UInt8 | DataType::UInt16 | DataType::UInt32 | DataType::UInt64 => {
4085            text.parse::<u64>().ok().map(AnyValue::UInt64)
4086        }
4087        _ => text.parse::<i64>().ok().map(AnyValue::Int64),
4088    };
4089    parsed
4090        .map(|parsed| {
4091            numfmt::format_any_value(&formatter, &parsed, &mut String::new()).into_owned()
4092        })
4093        .unwrap_or_else(|| text.to_string())
4094}
4095
4096fn count_label(value: Option<usize>) -> String {
4097    value
4098        .map(numfmt::group_chrome)
4099        .unwrap_or_else(|| "-".to_string())
4100}
4101
4102fn render_sidebar(
4103    config: &DataQualityWidgetConfig<'_>,
4104    sidebar_state: &mut TableState,
4105    area: Rect,
4106    buf: &mut Buffer,
4107) {
4108    crate::widgets::analysis::render_sidebar(
4109        area,
4110        buf,
4111        sidebar_state,
4112        Some(AnalysisTool::DataQuality),
4113        config.focus,
4114        config.theme,
4115    );
4116}
4117
4118/// The tool list, on a terminal too narrow to keep it beside the result.
4119fn render_narrow_tool_picker(
4120    config: &DataQualityWidgetConfig<'_>,
4121    sidebar_state: &mut TableState,
4122    area: Rect,
4123    buf: &mut Buffer,
4124) {
4125    let tools = vec![
4126        "Describe",
4127        "Distribution Analysis",
4128        "Correlation Matrix",
4129        "Data Quality",
4130    ];
4131    let popup = dialog_in(area, 28, tools.len() as u16 + 2);
4132    let content = Surface::new("Analysis Tools").render(popup, buf, config.ctx);
4133    Picker::new(tools, sidebar_state.selected(), true)
4134        .on_click(crate::widgets::ui::Clicks::Tool)
4135        .render(content, buf, config.ctx);
4136}
4137
4138/// What a run of the plan will read and write, before it runs.
4139fn render_access_plan(config: &DataQualityWidgetConfig<'_>, area: Rect, buf: &mut Buffer) {
4140    let state = config.state;
4141    let plan = config.plan;
4142    let remote = state.is_remote_source();
4143    let copy = config.setup.copy;
4144    let row = |label: &str, value: String| FieldRow {
4145        mark: None,
4146        label: label.to_string(),
4147        value,
4148    };
4149    let source_files = if plan.scope.uses_source() {
4150        Some(state.quality_source_file_count())
4151            .filter(|count| *count > 0)
4152            .or_else(|| state.source_file_count())
4153    } else {
4154        state.source_file_count()
4155    };
4156    let rows = [
4157        row(
4158            "Source",
4159            if remote { "remote" } else { "local" }.to_string(),
4160        ),
4161        row("Scope", plan.scope.label()),
4162        row("Grain", plan.grain.label()),
4163        row("Sample", compute_label(plan)),
4164        row(
4165            "Rows evaluated",
4166            planned_rows(state, plan)
4167                .map(numfmt::group_chrome)
4168                .unwrap_or_else(|| "unknown".to_string()),
4169        ),
4170        row(
4171            "Value reads",
4172            match copy {
4173                CopyPlan::Fetch { bytes, .. } => crate::glyphs::dotted(&format!(
4174                    "{}, fetched once · passes read the copy",
4175                    crate::numfmt::bytes(bytes)
4176                )),
4177                CopyPlan::Kept { bytes, .. } => {
4178                    format!(
4179                        "none: every pass reads the local copy ({})",
4180                        crate::numfmt::bytes(bytes)
4181                    )
4182                }
4183                _ if remote => "unknown".to_string(),
4184                _ => planned_read_label(state, plan),
4185            },
4186        ),
4187        row(
4188            "Requests",
4189            match copy {
4190                CopyPlan::Fetch { objects, .. } => format!("{objects}, one per object"),
4191                CopyPlan::Kept { .. } => "none".to_string(),
4192                _ if remote => "unknown".to_string(),
4193                _ => "none".to_string(),
4194            },
4195        ),
4196        row(
4197            "Known source files",
4198            source_files
4199                .map(numfmt::group_chrome)
4200                .unwrap_or_else(|| "unknown".to_string()),
4201        ),
4202        row(
4203            "Conflict values",
4204            match state.quality_conflict_reads() {
4205                0 => "none: no file holds a column in an unreadable type".to_string(),
4206                reads if plan.compute == QualityCompute::Full => {
4207                    format!("{reads} extra one-column file reads")
4208                }
4209                reads => crate::glyphs::dotted(&format!(
4210                    "not read · full scan adds {reads} one-column file reads"
4211                )),
4212            },
4213        ),
4214        row("Remote writes", "none".to_string()),
4215        row(
4216            "Local file writes",
4217            match copy {
4218                CopyPlan::Fetch { bytes, .. } => crate::glyphs::dotted(&format!(
4219                    "{} copy in the cache directory · kept until d, reopen or exit",
4220                    crate::numfmt::bytes(bytes)
4221                )),
4222                _ => "none".to_string(),
4223            },
4224        ),
4225        row("Passes", passes_label(config)),
4226        row("Column intent", {
4227            let every_row =
4228                planned_scope_rows(state, plan).is_some_and(|rows| rows <= plan.dataset_rows);
4229            let lines = intent_read_lines(plan, every_row);
4230            if lines.is_empty() {
4231                "none declared".to_string()
4232            } else {
4233                // The row names it; the lines say what it costs.
4234                lines
4235                    .iter()
4236                    .map(|line| line.strip_prefix("Column intent: ").unwrap_or(line))
4237                    .collect::<Vec<_>>()
4238                    .join(&format!(" {} ", glyphs::get().middot))
4239            }
4240        }),
4241        row(
4242            "Estimate basis",
4243            match plan.compute {
4244                QualityCompute::Metadata => "file footers read when the data opened".to_string(),
4245                QualityCompute::Sample => {
4246                    "rows: the sample; bytes: a ceiling, the whole scope".to_string()
4247                }
4248                QualityCompute::Full if matches!(copy, CopyPlan::Fetch { .. }) => {
4249                    "rows: every eligible row; bytes: the object sizes listed when it opened"
4250                        .to_string()
4251                }
4252                QualityCompute::Full => {
4253                    "rows: every eligible row; bytes: unknown, per source and its caches"
4254                        .to_string()
4255                }
4256            },
4257        ),
4258    ];
4259    let width = 72.min(area.width.saturating_sub(2));
4260    let label_width = rows
4261        .iter()
4262        .map(|row| glyphs::display_width(&row.label))
4263        .max()
4264        .unwrap_or(0)
4265        + 2;
4266    // The frame and its gutters take four columns of the width.
4267    let lines = field_lines(&rows, label_width, width.saturating_sub(4) as usize, false);
4268    let popup = dialog_in(area, width, lines.len() as u16 + 2);
4269    let content = Surface::new("Access Plan").render(popup, buf, config.ctx);
4270    render_counted(lines, content, config.theme, buf);
4271}
4272
4273/// The reads a run makes over its scope, counted from the plan: honest about the
4274/// passes even where their bytes are unknown.
4275fn passes_label(config: &DataQualityWidgetConfig<'_>) -> String {
4276    let plan = config.plan;
4277    let view = &config.setup;
4278    match plan.compute {
4279        _ if view.unchanged || view.cached => "none: the report is already here".to_string(),
4280        QualityCompute::Metadata => "none".to_string(),
4281        QualityCompute::Full => match view.copy {
4282            CopyPlan::Fetch { .. } => format!(
4283                "one fetch, then up to {} over the copy, one per check",
4284                full_passes(config)
4285            ),
4286            CopyPlan::Kept { .. } => {
4287                format!(
4288                    "up to {} over the local copy, one per check",
4289                    full_passes(config)
4290                )
4291            }
4292            _ => format!("up to {}, one per check", full_passes(config)),
4293        },
4294        QualityCompute::Sample => {
4295            let sample = if view.reuses_sample {
4296                "none for the sample: rows already read"
4297            } else {
4298                "one sampling pass"
4299            };
4300            match view.segment_count {
4301                SegmentCount::CountPass => {
4302                    format!("{sample}, then one count of the grain's column")
4303                }
4304                SegmentCount::InSamplePass => format!("{sample}, counting the grain's rows"),
4305                _ => sample.to_string(),
4306            }
4307        }
4308    }
4309}
4310
4311fn render_run_prompt(area: Rect, theme: &Theme, buf: &mut Buffer) {
4312    Paragraph::new(glyphs::dotted("No report · e Setup, then Enter"))
4313        .alignment(Alignment::Center)
4314        .style(Style::default().fg(theme.text_primary()))
4315        .render(area, buf);
4316}
4317
4318fn normalize_selection(state: &mut TableState, len: usize) {
4319    if len == 0 {
4320        state.select(None);
4321    } else if state.selected().is_none_or(|selected| selected >= len) {
4322        state.select(Some(0));
4323    }
4324}
4325
4326fn planned_rows(state: &DataTableState, plan: &DataQualityPlan) -> Option<usize> {
4327    if plan.compute == QualityCompute::Metadata {
4328        return Some(0);
4329    }
4330    let total = match &plan.scope {
4331        QualityScope::CurrentView => state.num_rows_if_valid()?,
4332        QualityScope::FirstRows(rows) => state.num_rows_if_valid()?.min(*rows),
4333        QualityScope::ViewRows { start, end } => {
4334            let rows = state.num_rows_if_valid()?;
4335            rows.min(*end).saturating_sub(start.saturating_sub(1))
4336        }
4337        _ => return None,
4338    };
4339    match plan.compute {
4340        QualityCompute::Metadata => unreachable!(),
4341        QualityCompute::Sample => Some(total.min(dataset_rows(plan))),
4342        QualityCompute::Full => Some(total),
4343    }
4344}
4345
4346fn planned_read_bytes(state: &DataTableState, plan: &DataQualityPlan) -> Option<usize> {
4347    if plan.scope.uses_source() {
4348        return None;
4349    }
4350    let rows = match plan.compute {
4351        QualityCompute::Metadata => 0,
4352        // The head reads what it keeps; otherwise a ceiling: a spread sample is a few dozen
4353        // runs of one file or one stream, and per-partition streams the scope.
4354        QualityCompute::Sample
4355            if plan.method == crate::analysis::sampling::SampleMethod::FirstRows =>
4356        {
4357            planned_scope_rows(state, plan)?.min(dataset_rows(plan))
4358        }
4359        QualityCompute::Sample => planned_scope_rows(state, plan)?,
4360        QualityCompute::Full => return None,
4361    };
4362    Some(rows.saturating_mul(state.estimated_row_bytes()))
4363}
4364
4365fn planned_scope_rows(state: &DataTableState, plan: &DataQualityPlan) -> Option<usize> {
4366    let rows = state.num_rows_if_valid()?;
4367    Some(match &plan.scope {
4368        QualityScope::FirstRows(limit) => rows.min(*limit),
4369        QualityScope::ViewRows { start, end } => {
4370            rows.min(*end).saturating_sub(start.saturating_sub(1))
4371        }
4372        QualityScope::CurrentView => rows,
4373        _ => return None,
4374    })
4375}
4376
4377/// The planned read in words: a sample of a scope larger than itself reads at most
4378/// the scope, and how much less depends on the source.
4379pub(crate) fn planned_read_label(state: &DataTableState, plan: &DataQualityPlan) -> String {
4380    let bytes = planned_read_bytes(state, plan);
4381    let sampled = plan.compute == QualityCompute::Sample
4382        && plan.method != crate::analysis::sampling::SampleMethod::FirstRows
4383        && planned_scope_rows(state, plan).is_some_and(|rows| rows > dataset_rows(plan));
4384    match bytes {
4385        Some(bytes) if sampled => format!("up to {}", crate::numfmt::bytes(bytes as u64)),
4386        Some(bytes) => format!("about {}", crate::numfmt::bytes(bytes as u64)),
4387        None => "unknown".to_string(),
4388    }
4389}
4390
4391/// What reading every row of `plan`'s scope again costs, as far as it is known: a
4392/// ceiling from the rows and their width, and unknown for a remote or source scope.
4393pub(crate) fn scope_read_label(state: &DataTableState, plan: &DataQualityPlan) -> String {
4394    if state.is_remote_source() {
4395        return "unknown".to_string();
4396    }
4397    match planned_scope_rows(state, plan) {
4398        Some(rows) => format!(
4399            "up to {}",
4400            crate::numfmt::bytes(rows.saturating_mul(state.estimated_row_bytes()) as u64)
4401        ),
4402        None => "unknown".to_string(),
4403    }
4404}
4405
4406/// Rows a dataset-grain run keeps: the shared sample's size.
4407fn dataset_rows(plan: &DataQualityPlan) -> usize {
4408    plan.dataset_rows
4409}
4410
4411pub(crate) fn compute_label(plan: &DataQualityPlan) -> String {
4412    match plan.compute {
4413        QualityCompute::Metadata => "metadata only".to_string(),
4414        QualityCompute::Sample => format!(
4415            "{} rows {} / seed {}",
4416            numfmt::group_chrome(dataset_rows(plan)),
4417            plan.method.label().to_lowercase(),
4418            plan.sample_seed
4419        ),
4420        QualityCompute::Full => "full scan".to_string(),
4421    }
4422}
4423
4424#[cfg(test)]
4425mod tests;
4426
4427/// Intervals on screen: the report's list and one interval's detail, and what
4428/// Setup says about roles and zones before a run.
4429#[cfg(test)]
4430mod interval_tests;
4431
4432#[cfg(test)]
4433mod trend_tests;