delta_funnel/orchestrator/session/
handles.rs

1use std::fmt;
2
3use crate::{
4    ExecutionProfileMode, MssqlTargetConfig, MssqlTargetOutputPlan, PhaseTimingReport,
5    QueryExecutionProfile, ReportReasonCode, ResolvedMssqlTarget,
6    support::sanitize_text_for_display,
7};
8
9pub(crate) const PREVIEW_DATAFRAME_PLANNING_PHASE: &str = "preview_dataframe_planning";
10pub(crate) const PREVIEW_PHYSICAL_PLANNING_PHASE: &str = "preview_physical_planning";
11pub(crate) const PREVIEW_STREAM_SETUP_PHASE: &str = "preview_stream_setup";
12pub(crate) const PREVIEW_EXECUTE_COLLECT_PHASE: &str = "preview_execute_collect";
13pub(crate) const PREVIEW_FORMAT_TEXT_PHASE: &str = "preview_format_text";
14pub(crate) const PREVIEW_FORMAT_HTML_PHASE: &str = "preview_format_html";
15pub(crate) const PREVIEW_TOTAL_PHASE: &str = "preview_total";
16
17pub(crate) const PREVIEW_PHASE_NAMES: [&str; 7] = [
18    PREVIEW_DATAFRAME_PLANNING_PHASE,
19    PREVIEW_PHYSICAL_PLANNING_PHASE,
20    PREVIEW_STREAM_SETUP_PHASE,
21    PREVIEW_EXECUTE_COLLECT_PHASE,
22    PREVIEW_FORMAT_TEXT_PHASE,
23    PREVIEW_FORMAT_HTML_PHASE,
24    PREVIEW_TOTAL_PHASE,
25];
26
27/// Query-load action mode requested by a caller.
28#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
29pub enum RunMode {
30    /// Plan and execute the selected output workflow.
31    #[default]
32    Execute,
33    /// Reuse planning paths without row production or SQL Server write effects.
34    DryRun,
35}
36
37/// Lazy table identity owned by a query-load session.
38#[derive(Debug, Clone, PartialEq, Eq)]
39pub struct LazyTable {
40    id: LazyTableId,
41    kind: LazyTableKind,
42    name: String,
43}
44
45impl LazyTable {
46    /// Creates a placeholder lazy table handle for future registration slices.
47    #[cfg(test)]
48    #[must_use]
49    pub(crate) fn placeholder(id: u64, kind: LazyTableKind) -> Self {
50        Self {
51            id: LazyTableId(id),
52            kind,
53            name: format!("table_{id}"),
54        }
55    }
56
57    pub(super) fn delta_source(id: u64, name: String) -> Self {
58        Self {
59            id: LazyTableId(id),
60            kind: LazyTableKind::DeltaSource,
61            name,
62        }
63    }
64
65    pub(super) fn derived_sql(id: u64) -> Self {
66        Self {
67            id: LazyTableId(id),
68            kind: LazyTableKind::DerivedSql,
69            name: format!("table_{id}"),
70        }
71    }
72
73    pub(super) fn with_name(&self, name: String) -> Self {
74        Self {
75            id: self.id,
76            kind: self.kind,
77            name,
78        }
79    }
80
81    /// Returns the stable session-local table id.
82    #[must_use]
83    pub const fn id(&self) -> u64 {
84        self.id.0
85    }
86
87    /// Returns the lazy table kind.
88    #[must_use]
89    pub const fn kind(&self) -> LazyTableKind {
90        self.kind
91    }
92
93    /// Returns the session-owned table name for this lazy table.
94    #[must_use]
95    pub fn name(&self) -> &str {
96        &self.name
97    }
98}
99
100#[derive(Debug, Clone, Copy, PartialEq, Eq)]
101struct LazyTableId(u64);
102
103/// Kind of lazy table represented by a session handle.
104#[derive(Debug, Clone, Copy, PartialEq, Eq)]
105pub enum LazyTableKind {
106    /// Registered Delta source table.
107    DeltaSource,
108    /// SQL-derived table.
109    DerivedSql,
110}
111
112/// Options for a bounded lazy-table preview.
113#[derive(Debug, Clone, Copy, PartialEq, Eq)]
114pub struct PreviewOptions {
115    limit: usize,
116    execution_profile_mode: ExecutionProfileMode,
117}
118
119impl PreviewOptions {
120    /// Creates preview options with detailed execution profiling disabled.
121    #[must_use]
122    pub const fn new(limit: usize) -> Self {
123        Self {
124            limit,
125            execution_profile_mode: ExecutionProfileMode::Disabled,
126        }
127    }
128
129    /// Selects whether the preview collects a detailed execution profile.
130    #[must_use]
131    pub const fn with_execution_profile_mode(mut self, mode: ExecutionProfileMode) -> Self {
132        self.execution_profile_mode = mode;
133        self
134    }
135
136    /// Returns the requested preview row limit.
137    #[must_use]
138    pub const fn limit(&self) -> usize {
139        self.limit
140    }
141
142    /// Returns the selected execution-profile mode.
143    #[must_use]
144    pub const fn execution_profile_mode(&self) -> ExecutionProfileMode {
145        self.execution_profile_mode
146    }
147}
148
149/// Structured diagnostics retained when a bounded preview fails.
150#[derive(Debug, Clone, PartialEq, Eq)]
151pub struct PreviewFailureContext {
152    failed_phase: String,
153    phase_timings: Vec<PhaseTimingReport>,
154    execution_profile: Option<QueryExecutionProfile>,
155}
156
157impl PreviewFailureContext {
158    pub(crate) fn new(
159        failed_phase: String,
160        phase_timings: Vec<PhaseTimingReport>,
161        execution_profile: Option<QueryExecutionProfile>,
162    ) -> Self {
163        Self {
164            failed_phase,
165            phase_timings,
166            execution_profile,
167        }
168    }
169
170    /// Returns the stable name of the phase that failed.
171    #[must_use]
172    pub fn failed_phase(&self) -> &str {
173        &self.failed_phase
174    }
175
176    /// Returns the complete ordered preview phase timings.
177    #[must_use]
178    pub fn phase_timings(&self) -> &[PhaseTimingReport] {
179        &self.phase_timings
180    }
181
182    /// Returns the terminal execution profile when detailed profiling reached
183    /// a terminal transition, including stream-setup failure.
184    #[must_use]
185    pub const fn execution_profile(&self) -> Option<&QueryExecutionProfile> {
186        self.execution_profile.as_ref()
187    }
188}
189
190/// Rendered bounded preview of a lazy table.
191#[derive(Debug, Clone, PartialEq, Eq)]
192pub struct TablePreview {
193    text: String,
194    html: String,
195    phase_timings: Vec<PhaseTimingReport>,
196    execution_profile: Option<QueryExecutionProfile>,
197}
198
199impl TablePreview {
200    /// Creates a rendered lazy table preview without execution diagnostics.
201    ///
202    /// All seven preview phase timings are marked unavailable with
203    /// [`ReportReasonCode::NotExecuted`], and no execution profile is attached.
204    #[must_use]
205    pub fn new(text: String, html: String) -> Self {
206        Self::from_execution(
207            text,
208            html,
209            PREVIEW_PHASE_NAMES
210                .into_iter()
211                .map(|phase_name| {
212                    PhaseTimingReport::unavailable(phase_name, ReportReasonCode::NotExecuted)
213                })
214                .collect(),
215            None,
216        )
217    }
218
219    pub(crate) fn from_execution(
220        text: String,
221        html: String,
222        phase_timings: Vec<PhaseTimingReport>,
223        execution_profile: Option<QueryExecutionProfile>,
224    ) -> Self {
225        Self {
226            text,
227            html,
228            phase_timings,
229            execution_profile,
230        }
231    }
232
233    /// Returns the plain text table preview.
234    #[must_use]
235    pub fn text(&self) -> &str {
236        &self.text
237    }
238
239    /// Returns the HTML table preview.
240    #[must_use]
241    pub fn html(&self) -> &str {
242        &self.html
243    }
244
245    /// Returns phase timings captured for this preview operation.
246    #[must_use]
247    pub fn phase_timings(&self) -> &[PhaseTimingReport] {
248        &self.phase_timings
249    }
250
251    /// Returns the detailed execution profile when collection was enabled.
252    #[must_use]
253    pub const fn execution_profile(&self) -> Option<&QueryExecutionProfile> {
254        self.execution_profile.as_ref()
255    }
256}
257
258/// MSSQL output target selected from a lazy table.
259#[derive(Clone, PartialEq, Eq)]
260pub struct MssqlOutputTarget {
261    output_name: String,
262    target: MssqlTargetConfig,
263    run_mode: RunMode,
264}
265
266impl MssqlOutputTarget {
267    /// Creates an MSSQL output target request.
268    #[must_use]
269    pub fn new(
270        output_name: impl Into<String>,
271        target: MssqlTargetConfig,
272        run_mode: RunMode,
273    ) -> Self {
274        Self {
275            output_name: output_name.into(),
276            target,
277            run_mode,
278        }
279    }
280
281    /// Returns the selected output name.
282    #[must_use]
283    pub fn output_name(&self) -> &str {
284        &self.output_name
285    }
286
287    /// Returns the SQL Server target config.
288    #[must_use]
289    pub const fn target(&self) -> &MssqlTargetConfig {
290        &self.target
291    }
292
293    /// Returns the requested run mode.
294    #[must_use]
295    pub const fn run_mode(&self) -> RunMode {
296        self.run_mode
297    }
298}
299
300impl fmt::Debug for MssqlOutputTarget {
301    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
302        formatter
303            .debug_struct("MssqlOutputTarget")
304            .field("output_name", &sanitize_text_for_display(&self.output_name))
305            .field("target", &self.target)
306            .field("run_mode", &self.run_mode)
307            .finish()
308    }
309}
310
311/// Planned output write request before schema planning or execution.
312#[derive(Debug, Clone, PartialEq, Eq)]
313pub struct OutputWritePlan {
314    table: LazyTable,
315    target: MssqlOutputTarget,
316}
317
318impl OutputWritePlan {
319    /// Creates an output write request for a lazy table.
320    #[must_use]
321    pub const fn new(table: LazyTable, target: MssqlOutputTarget) -> Self {
322        Self { table, target }
323    }
324
325    /// Returns the selected lazy table.
326    #[must_use]
327    pub const fn table(&self) -> &LazyTable {
328        &self.table
329    }
330
331    /// Returns the selected MSSQL output target.
332    #[must_use]
333    pub const fn target(&self) -> &MssqlOutputTarget {
334        &self.target
335    }
336}
337
338/// Planned MSSQL output request for one selected lazy table.
339#[derive(Debug, Clone, PartialEq, Eq)]
340pub struct PlannedMssqlOutput {
341    request: OutputWritePlan,
342    resolved_target: ResolvedMssqlTarget,
343    output_plan: MssqlTargetOutputPlan,
344    phase_timings: Vec<PhaseTimingReport>,
345}
346
347impl PlannedMssqlOutput {
348    pub(super) fn new(
349        request: OutputWritePlan,
350        resolved_target: ResolvedMssqlTarget,
351        output_plan: MssqlTargetOutputPlan,
352        phase_timings: Vec<PhaseTimingReport>,
353    ) -> Self {
354        Self {
355            request,
356            resolved_target,
357            output_plan,
358            phase_timings,
359        }
360    }
361
362    /// Returns the original lazy-table output request.
363    #[must_use]
364    pub const fn request(&self) -> &OutputWritePlan {
365        &self.request
366    }
367
368    /// Returns the selected lazy table.
369    #[must_use]
370    pub const fn table(&self) -> &LazyTable {
371        self.request.table()
372    }
373
374    /// Returns the selected MSSQL target request.
375    #[must_use]
376    pub const fn target(&self) -> &MssqlOutputTarget {
377        self.request.target()
378    }
379
380    /// Returns the resolved SQL Server target, including the private connection config.
381    #[must_use]
382    pub const fn resolved_target(&self) -> &ResolvedMssqlTarget {
383        &self.resolved_target
384    }
385
386    /// Returns the complete SQL Server target output plan.
387    #[must_use]
388    pub const fn output_plan(&self) -> &MssqlTargetOutputPlan {
389        &self.output_plan
390    }
391
392    /// Returns durable phase timings captured while planning this output.
393    #[must_use]
394    pub fn phase_timings(&self) -> &[PhaseTimingReport] {
395        &self.phase_timings
396    }
397}
398
399#[cfg(test)]
400mod tests {
401    use crate::{
402        DeltaFunnelError, ExecutionProfileMode, LoadMode, MssqlConnectionConfig, MssqlTargetConfig,
403        MssqlTargetTable, PhaseStatus, ReportReasonCode,
404    };
405
406    use super::{
407        LazyTable, LazyTableKind, MssqlOutputTarget, OutputWritePlan, PREVIEW_PHASE_NAMES,
408        PreviewOptions, RunMode, TablePreview,
409    };
410
411    #[test]
412    fn preview_options_default_to_disabled_profiling() {
413        let default = PreviewOptions::new(20);
414        let detailed = default.with_execution_profile_mode(ExecutionProfileMode::Detailed);
415
416        assert_eq!(default.limit(), 20);
417        assert_eq!(
418            default.execution_profile_mode(),
419            ExecutionProfileMode::Disabled
420        );
421        assert_eq!(detailed.limit(), 20);
422        assert_eq!(
423            detailed.execution_profile_mode(),
424            ExecutionProfileMode::Detailed
425        );
426    }
427
428    #[test]
429    fn legacy_table_preview_has_unavailable_timings_and_no_profile() {
430        let preview = TablePreview::new("text".to_owned(), "html".to_owned());
431
432        assert_eq!(preview.text(), "text");
433        assert_eq!(preview.html(), "html");
434        assert_eq!(preview.phase_timings().len(), PREVIEW_PHASE_NAMES.len());
435        for (timing, phase_name) in preview.phase_timings().iter().zip(PREVIEW_PHASE_NAMES) {
436            assert_eq!(timing.phase_name(), phase_name);
437            assert_eq!(
438                timing.status(),
439                PhaseStatus::unavailable(ReportReasonCode::NotExecuted)
440            );
441            assert_eq!(timing.elapsed_micros(), None);
442        }
443        assert_eq!(preview.execution_profile(), None);
444    }
445
446    #[test]
447    fn output_request_shapes_preserve_table_target_and_run_mode() -> Result<(), DeltaFunnelError> {
448        let table = LazyTable::placeholder(7, LazyTableKind::DerivedSql);
449        let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders")?)
450            .with_load_mode(LoadMode::CreateAndLoad)
451            .with_connection(
452                MssqlConnectionConfig::new(
453                    "server=tcp:sql.example.com;database=warehouse;user=admin;password=secret-token",
454                )?
455                .with_display_label("warehouse-primary"),
456            );
457        let target = MssqlOutputTarget::new("orders_output", target_config, RunMode::DryRun);
458        let plan = OutputWritePlan::new(table.clone(), target.clone());
459
460        assert_eq!(table.id(), 7);
461        assert_eq!(table.kind(), LazyTableKind::DerivedSql);
462        assert_eq!(target.output_name(), "orders_output");
463        assert_eq!(target.run_mode(), RunMode::DryRun);
464        assert_eq!(target.target().load_mode(), LoadMode::CreateAndLoad);
465        assert_eq!(plan.table(), &table);
466        assert_eq!(plan.target(), &target);
467
468        let debug = format!("{target:?}");
469        assert!(debug.contains("orders_output"));
470        assert!(!debug.contains("secret-token"));
471        assert!(!debug.contains("password"));
472        assert!(!debug.contains("server=tcp"));
473        Ok(())
474    }
475}
delta_funnel/orchestrator/session/handles.rs

delta_funnel/orchestrator/session/
handles.rs
