delta_funnel/orchestrator/session/
dry_run_report.rs

1use crate::{
2    DeltaFunnelError, PhaseTimingReport, ReportReasonCode, observability,
3    report::PhaseTimer,
4    report::sql_server::{
5        MssqlDryRunOutputReport, MssqlDryRunSqlIdentityReport, MssqlDryRunWorkflowReport,
6    },
7};
8
9use super::{
10    DeltaFunnelSession, LazyTable, LazyTableKind, OutputWritePlan, PlannedMssqlOutput, RunMode,
11    SourceUsageStatus, sql_server_workflows::ensure_unique_write_all_output_names,
12};
13use tracing::Instrument;
14
15pub(super) fn stable_sql_identity_hash(sql: &str) -> String {
16    const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325;
17    const FNV_PRIME: u64 = 0x100000001b3;
18
19    let mut hash = FNV_OFFSET_BASIS;
20    for byte in sql.as_bytes() {
21        hash ^= u64::from(*byte);
22        hash = hash.wrapping_mul(FNV_PRIME);
23    }
24
25    format!("{hash:016x}")
26}
27
28const QUERY_EXECUTION_PHASE: &str = "query_execution";
29const BATCH_SHAPING_PHASE: &str = "batch_shaping";
30const SQL_WRITE_PHASE: &str = "sql_write";
31const FINALIZE_PHASE: &str = "finalize";
32const VALIDATION_PHASE: &str = "validation";
33const OUTPUT_PLANNING_PHASE: &str = "output_planning";
34const SOURCE_REPORTING_PHASE: &str = "source_reporting";
35
36impl DeltaFunnelSession {
37    /// Dry-runs one selected lazy table as an MSSQL output.
38    ///
39    /// The method reuses the same session output planner as execute mode, then
40    /// stops before physical DataFusion planning, row production, SQL Server
41    /// lifecycle work, bulk writer construction, or row writes.
42    ///
43    /// # Errors
44    ///
45    /// Returns an MSSQL planning error when the request is not in
46    /// [`RunMode::DryRun`], or the first error from session output planning.
47    pub fn dry_run_to_mssql(
48        &self,
49        request: &OutputWritePlan,
50    ) -> Result<MssqlDryRunOutputReport, DeltaFunnelError> {
51        ensure_dry_run_mode(request.target().run_mode())?;
52
53        self.plan_dry_run_output(request)
54    }
55
56    /// Dry-runs multiple selected lazy tables as one MSSQL output workflow.
57    ///
58    /// The method plans each selected output in caller-provided order and stops
59    /// before cache materialization, physical DataFusion planning, row
60    /// production, SQL Server lifecycle work, bulk writer construction, or row
61    /// writes.
62    ///
63    /// # Errors
64    ///
65    /// Returns the first duplicate-output, run-mode, or output-planning error.
66    pub fn dry_run_all_to_mssql(
67        &self,
68        requests: &[OutputWritePlan],
69    ) -> Result<MssqlDryRunWorkflowReport, DeltaFunnelError> {
70        let planning_timer = PhaseTimer::start(OUTPUT_PLANNING_PHASE);
71        let outputs = self.plan_dry_run_all_outputs(requests)?;
72        let planning_timing = planning_timer.completed();
73        let source_timer = PhaseTimer::start(SOURCE_REPORTING_PHASE);
74        let sources = self.source_reports_for_dry_run_outputs(&outputs)?;
75        let source_timing = source_timer.completed();
76
77        Ok(MssqlDryRunWorkflowReport::new(outputs, sources)
78            .with_phase_timings(vec![planning_timing, source_timing]))
79    }
80
81    /// Runs [`Self::dry_run_all_to_mssql`] inside a DeltaFunnel workflow tracing span.
82    ///
83    /// # Errors
84    ///
85    /// Returns the same error as [`Self::dry_run_all_to_mssql`].
86    pub fn dry_run_all_to_mssql_with_tracing(
87        &self,
88        requests: &[OutputWritePlan],
89    ) -> Result<MssqlDryRunWorkflowReport, DeltaFunnelError> {
90        let output_count = requests.len();
91        let span = observability::workflow_span(RunMode::DryRun, output_count);
92        let _guard = span.enter();
93        observability::workflow_started(RunMode::DryRun, output_count);
94        let result = self.dry_run_all_to_mssql(requests);
95        observability::workflow_finished(RunMode::DryRun, output_count, &result);
96        result
97    }
98
99    /// Dry-runs multiple selected lazy tables and honors source scan-summary options.
100    ///
101    /// This method is async because
102    /// [`crate::DryRunScanSummaryMode::ExhaustScanMetadata`] requires
103    /// DataFusion physical planning to expose provider scan metadata. It still
104    /// stops before row production, SQL Server lifecycle work, bulk writer
105    /// construction, or row writes.
106    ///
107    /// # Errors
108    ///
109    /// Returns the first duplicate-output, run-mode, output-planning, or
110    /// DataFusion physical-planning error.
111    pub async fn dry_run_all_to_mssql_with_scan_summary(
112        &self,
113        requests: &[OutputWritePlan],
114    ) -> Result<MssqlDryRunWorkflowReport, DeltaFunnelError> {
115        let planning_timer = PhaseTimer::start(OUTPUT_PLANNING_PHASE);
116        let outputs = self.plan_dry_run_all_outputs(requests)?;
117        let planning_timing = planning_timer.completed();
118        let source_timer = PhaseTimer::start(SOURCE_REPORTING_PHASE);
119        let sources = match self
120            .options
121            .validation_options()
122            .dry_run_scan_summary_mode()
123        {
124            crate::DryRunScanSummaryMode::MetadataOnly => {
125                self.source_reports_for_dry_run_outputs(&outputs)?
126            }
127            crate::DryRunScanSummaryMode::ExhaustScanMetadata => self
128                .source_reports_for_dry_run_outputs_with_provider_stats(
129                    &outputs,
130                    self.provider_read_stats_for_dry_run_outputs(&outputs)
131                        .await?,
132                )?,
133        };
134        let source_timing = source_timer.completed();
135
136        Ok(MssqlDryRunWorkflowReport::new(outputs, sources)
137            .with_phase_timings(vec![planning_timing, source_timing]))
138    }
139
140    /// Runs [`Self::dry_run_all_to_mssql_with_scan_summary`] inside a workflow tracing span.
141    ///
142    /// # Errors
143    ///
144    /// Returns the same error as [`Self::dry_run_all_to_mssql_with_scan_summary`].
145    pub async fn dry_run_all_to_mssql_with_scan_summary_with_tracing(
146        &self,
147        requests: &[OutputWritePlan],
148    ) -> Result<MssqlDryRunWorkflowReport, DeltaFunnelError> {
149        let output_count = requests.len();
150        async move {
151            observability::workflow_started(RunMode::DryRun, output_count);
152            let result = self.dry_run_all_to_mssql_with_scan_summary(requests).await;
153            observability::workflow_finished(RunMode::DryRun, output_count, &result);
154            result
155        }
156        .instrument(observability::workflow_span(RunMode::DryRun, output_count))
157        .await
158    }
159
160    fn plan_dry_run_all_outputs(
161        &self,
162        requests: &[OutputWritePlan],
163    ) -> Result<Vec<MssqlDryRunOutputReport>, DeltaFunnelError> {
164        ensure_unique_write_all_output_names(requests)?;
165
166        requests
167            .iter()
168            .map(|request| {
169                ensure_write_all_dry_run_mode(request.target().run_mode())?;
170
171                self.plan_dry_run_output(request)
172            })
173            .collect()
174    }
175
176    fn plan_dry_run_output(
177        &self,
178        request: &OutputWritePlan,
179    ) -> Result<MssqlDryRunOutputReport, DeltaFunnelError> {
180        let target = request.target();
181        let target_config = target.target();
182        let output_span = observability::output_span(
183            target.output_name(),
184            target_config.table(),
185            target_config.load_mode(),
186        );
187        let _guard = output_span.enter();
188        observability::output_started(
189            target.output_name(),
190            target_config.table(),
191            target_config.load_mode(),
192        );
193        let result = self.plan_dry_run_output_report(request);
194        match &result {
195            Ok(_) => observability::output_completed(
196                target.output_name(),
197                target_config.table(),
198                target_config.load_mode(),
199            ),
200            Err(error) => observability::output_failed(
201                target.output_name(),
202                target_config.table(),
203                target_config.load_mode(),
204                &error.to_string(),
205            ),
206        }
207
208        result
209    }
210
211    fn plan_dry_run_output_report(
212        &self,
213        request: &OutputWritePlan,
214    ) -> Result<MssqlDryRunOutputReport, DeltaFunnelError> {
215        let planned = self.plan_mssql_output(request)?;
216
217        let phase_timings = dry_run_output_phase_timings(planned.phase_timings().to_vec());
218        self.dry_run_output_report_for_plan(planned, phase_timings)
219    }
220
221    fn dry_run_output_report_for_plan(
222        &self,
223        planned_output: PlannedMssqlOutput,
224        phase_timings: Vec<PhaseTimingReport>,
225    ) -> Result<MssqlDryRunOutputReport, DeltaFunnelError> {
226        let sql_identity = self.sql_identity_for_lazy_table(planned_output.table());
227        let (source_usage_status, used_source_names) =
228            self.source_usage_for_lazy_table(planned_output.table())?;
229        Ok(MssqlDryRunOutputReport::new(
230            planned_output,
231            sql_identity,
232            source_usage_status,
233            used_source_names,
234            phase_timings,
235        ))
236    }
237
238    fn source_usage_for_lazy_table(
239        &self,
240        table: &LazyTable,
241    ) -> Result<(SourceUsageStatus, Vec<String>), DeltaFunnelError> {
242        let Some(source_ids) = self.known_source_dependencies_for_table(table)? else {
243            return Ok((SourceUsageStatus::Unknown, Vec::new()));
244        };
245
246        let used_source_names = self
247            .sources
248            .iter()
249            .filter(|source| source_ids.contains(&source.table().id()))
250            .map(|source| source.name().to_owned())
251            .collect::<Vec<_>>();
252        let source_usage_status = if used_source_names.is_empty() {
253            SourceUsageStatus::NotUsed
254        } else {
255            SourceUsageStatus::Used
256        };
257
258        Ok((source_usage_status, used_source_names))
259    }
260
261    fn sql_identity_for_lazy_table(&self, table: &LazyTable) -> MssqlDryRunSqlIdentityReport {
262        if table.kind() != LazyTableKind::DerivedSql {
263            return MssqlDryRunSqlIdentityReport::absent();
264        }
265
266        match self.sql_text_for_derived_table(table) {
267            Ok(sql_text) => {
268                MssqlDryRunSqlIdentityReport::present(stable_sql_identity_hash(sql_text))
269            }
270            Err(_) => {
271                MssqlDryRunSqlIdentityReport::unavailable(ReportReasonCode::CapabilityUnavailable)
272            }
273        }
274    }
275}
276
277fn dry_run_output_phase_timings(
278    mut planning_timings: Vec<PhaseTimingReport>,
279) -> Vec<PhaseTimingReport> {
280    planning_timings.extend([
281        PhaseTimingReport::skipped(QUERY_EXECUTION_PHASE, ReportReasonCode::DryRun),
282        PhaseTimingReport::skipped(BATCH_SHAPING_PHASE, ReportReasonCode::DryRun),
283        PhaseTimingReport::skipped(SQL_WRITE_PHASE, ReportReasonCode::DryRun),
284        PhaseTimingReport::skipped(FINALIZE_PHASE, ReportReasonCode::DryRun),
285        PhaseTimingReport::skipped(VALIDATION_PHASE, ReportReasonCode::DryRun),
286    ]);
287    planning_timings
288}
289
290fn ensure_dry_run_mode(run_mode: RunMode) -> Result<(), DeltaFunnelError> {
291    match run_mode {
292        RunMode::DryRun => Ok(()),
293        RunMode::Execute => Err(DeltaFunnelError::MssqlWorkflowPlanning {
294            message: "dry_run_to_mssql requires RunMode::DryRun; use write_to_mssql for execution"
295                .to_owned(),
296        }),
297    }
298}
299
300fn ensure_write_all_dry_run_mode(run_mode: RunMode) -> Result<(), DeltaFunnelError> {
301    match run_mode {
302        RunMode::DryRun => Ok(()),
303        RunMode::Execute => Err(DeltaFunnelError::MssqlWorkflowPlanning {
304            message: "dry_run_all_to_mssql requires RunMode::DryRun; use write_all for execution"
305                .to_owned(),
306        }),
307    }
308}
309
310#[cfg(test)]
311mod tests {
312    use std::sync::atomic::Ordering;
313
314    use crate::{
315        DeltaFunnelError, DeltaSourceConfig, LoadMode, MssqlOutputTarget, MssqlTargetConfig,
316        MssqlTargetTable, OutputStatus, ReportReasonCode, ValidationOptions, ValidationStatus,
317        WorkflowStatus,
318    };
319
320    use super::super::{
321        DeltaFunnelSession, LazyTableKind, OutputWritePlan, RunMode, SessionOptions,
322        SourceUsageStatus,
323        sql_server_workflows::{OUTPUT_SCHEMA_PLANNING_PHASE, SQL_TARGET_PLANNING_PHASE},
324        test_support::{
325            DeltaLogTable, execute_output_request, output_request, override_connection,
326            scan_counting_marker_region_provider, secret_connection,
327        },
328    };
329    use super::{
330        BATCH_SHAPING_PHASE, FINALIZE_PHASE, OUTPUT_PLANNING_PHASE, QUERY_EXECUTION_PHASE,
331        SOURCE_REPORTING_PHASE, SQL_WRITE_PHASE, VALIDATION_PHASE, stable_sql_identity_hash,
332    };
333    use crate::MssqlDryRunSqlIdentityState;
334
335    const LAZY_SQL_PLANNING_PHASE: &str = "lazy_sql_planning";
336
337    #[test]
338    fn sql_identity_status_and_hash_are_stable() {
339        assert_eq!(MssqlDryRunSqlIdentityState::Present.as_str(), "present");
340        assert_eq!(MssqlDryRunSqlIdentityState::Absent.to_string(), "absent");
341        assert_eq!(
342            MssqlDryRunSqlIdentityState::Unavailable.as_str(),
343            "unavailable"
344        );
345        assert_eq!(
346            stable_sql_identity_hash("select marker where region = 'west'"),
347            "cbd6889e027b0f88"
348        );
349    }
350
351    fn phase_timing<'a>(
352        report: &'a crate::MssqlDryRunOutputReport,
353        phase_name: &str,
354    ) -> Result<&'a crate::PhaseTimingReport, Box<dyn std::error::Error>> {
355        report
356            .phase_timings()
357            .iter()
358            .find(|timing| timing.phase_name() == phase_name)
359            .ok_or_else(|| format!("missing phase timing `{phase_name}`").into())
360    }
361
362    fn workflow_phase_timing<'a>(
363        report: &'a crate::MssqlDryRunWorkflowReport,
364        phase_name: &str,
365    ) -> Result<&'a crate::PhaseTimingReport, Box<dyn std::error::Error>> {
366        report
367            .phase_timings()
368            .iter()
369            .find(|timing| timing.phase_name() == phase_name)
370            .ok_or_else(|| format!("missing workflow phase timing `{phase_name}`").into())
371    }
372
373    #[tokio::test]
374    async fn dry_run_to_mssql_plans_output_without_row_or_writer_side_effects()
375    -> Result<(), Box<dyn std::error::Error>> {
376        let mut session = DeltaFunnelSession::new(
377            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
378        )?;
379        let (source_provider, source_scans) = scan_counting_marker_region_provider("shared")?;
380        session
381            .context()
382            .register_table("orders_source", source_provider)?;
383        let output = session
384            .table_from_sql("select marker from orders_source where region = 'west'")
385            .await?;
386        let request = output_request(
387            output,
388            "west_output",
389            "west_orders",
390            LoadMode::CreateAndLoad,
391        )?;
392
393        let report = session.dry_run_to_mssql(&request)?;
394
395        assert_eq!(report.output_name(), "west_output");
396        assert_eq!(report.run_mode(), RunMode::DryRun);
397        assert_eq!(report.status(), OutputStatus::dry_run_planned());
398        assert_eq!(
399            report.validation_status(),
400            ValidationStatus::skipped(ReportReasonCode::DryRun)
401        );
402        assert_eq!(report.target_table().schema(), Some("dbo"));
403        assert_eq!(report.target_table().table(), "west_orders");
404        assert_eq!(report.load_mode(), LoadMode::CreateAndLoad);
405        assert_eq!(report.target_schema_plan().mappings().len(), 1);
406        assert!(report.target_ddl_plan().create_table_sql_present());
407        assert!(report.target_lifecycle_plan().create_table_sql_required());
408        assert_eq!(
409            report.target_lifecycle_plan().expected_target_state(),
410            crate::MssqlTargetTableState::Absent
411        );
412        assert_eq!(
413            report.planned_output().output_plan().target_table().table(),
414            "west_orders"
415        );
416        assert_eq!(
417            report
418                .planned_output()
419                .output_plan()
420                .schema_mappings()
421                .len(),
422            1
423        );
424        assert_eq!(report.output_schema().len(), 1);
425        assert_eq!(report.output_schema()[0].index(), 0);
426        assert_eq!(report.output_schema()[0].name(), "marker");
427        assert_eq!(report.output_schema()[0].arrow_type(), "Utf8");
428        assert!(!report.output_schema()[0].nullable());
429        assert_eq!(report.source_usage_status(), SourceUsageStatus::NotUsed);
430        assert!(report.used_source_names().is_empty());
431        assert_eq!(report.output_row_count(), crate::RowCount::unavailable());
432        assert_eq!(
433            report.output_row_count_reason(),
434            Some(ReportReasonCode::NotExecuted)
435        );
436        assert_eq!(
437            report.sql_identity().state(),
438            MssqlDryRunSqlIdentityState::Present
439        );
440        assert_eq!(report.sql_identity().hash(), Some("a65390dacb7eb6f1"));
441        assert_eq!(report.sql_identity().reason(), None);
442        let debug = format!("{report:?}");
443        assert!(debug.contains("a65390dacb7eb6f1"));
444        assert!(!debug.contains("select marker"));
445        assert!(!debug.contains("region = 'west'"));
446        assert!(!report.sql_server_contacted());
447        assert!(!report.row_production_started());
448        assert!(!report.table_lifecycle_started());
449        assert!(!report.bulk_writer_started());
450        assert_eq!(report.phase_timings().len(), 8);
451        for phase_name in [
452            LAZY_SQL_PLANNING_PHASE,
453            OUTPUT_SCHEMA_PLANNING_PHASE,
454            SQL_TARGET_PLANNING_PHASE,
455        ] {
456            let planning = phase_timing(&report, phase_name)?;
457            assert!(planning.status().is_completed());
458            assert!(planning.elapsed_micros().is_some());
459        }
460        for phase_name in [
461            QUERY_EXECUTION_PHASE,
462            BATCH_SHAPING_PHASE,
463            SQL_WRITE_PHASE,
464            FINALIZE_PHASE,
465            VALIDATION_PHASE,
466        ] {
467            let timing = phase_timing(&report, phase_name)?;
468            assert_eq!(
469                timing.status(),
470                crate::PhaseStatus::skipped(ReportReasonCode::DryRun)
471            );
472            assert_eq!(timing.elapsed_micros(), None);
473        }
474        assert_eq!(source_scans.load(Ordering::SeqCst), 0);
475        Ok(())
476    }
477
478    #[tokio::test]
479    async fn dry_run_all_to_mssql_reports_all_outputs_without_row_side_effects()
480    -> Result<(), Box<dyn std::error::Error>> {
481        let mut session = DeltaFunnelSession::new(
482            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
483        )?;
484        let (source_provider, source_scans) = scan_counting_marker_region_provider("shared")?;
485        session
486            .context()
487            .register_table("orders_source", source_provider)?;
488        let west = session
489            .table_from_sql("select marker from orders_source where region = 'west'")
490            .await?;
491        let east = session
492            .table_from_sql("select marker from orders_source where region = 'east'")
493            .await?;
494        let west_table_id = west.id();
495        let west_table_name = west.name().to_owned();
496        let west = output_request(west, "west_output", "west_orders", LoadMode::CreateAndLoad)?;
497        let east = output_request(east, "east_output", "east_orders", LoadMode::AppendExisting)?;
498
499        let report = session.dry_run_all_to_mssql(&[west, east])?;
500
501        assert_eq!(report.run_mode(), RunMode::DryRun);
502        assert_eq!(report.status(), WorkflowStatus::success());
503        assert_eq!(report.len(), 2);
504        assert!(!report.is_empty());
505        assert_eq!(report.outputs()[0].output_name(), "west_output");
506        assert_eq!(report.outputs()[1].output_name(), "east_output");
507        assert_eq!(report.outputs()[0].table_id(), west_table_id);
508        assert_eq!(report.outputs()[0].table_kind(), LazyTableKind::DerivedSql);
509        assert_eq!(report.outputs()[0].table_name(), west_table_name);
510        assert_eq!(
511            report.outputs()[0].status(),
512            OutputStatus::dry_run_planned()
513        );
514        assert_eq!(
515            report.outputs()[0].validation_status(),
516            ValidationStatus::skipped(ReportReasonCode::DryRun)
517        );
518        assert_eq!(
519            report.outputs()[0].output_row_count(),
520            crate::RowCount::unavailable()
521        );
522        assert_eq!(
523            report.outputs()[0].output_row_count_reason(),
524            Some(ReportReasonCode::NotExecuted)
525        );
526        assert!(report.sources().is_empty());
527        assert_eq!(report.outputs()[0].target_table().table(), "west_orders");
528        assert_eq!(report.outputs()[0].load_mode(), LoadMode::CreateAndLoad);
529        assert_eq!(
530            report.outputs()[0]
531                .target_lifecycle_plan()
532                .expected_target_state(),
533            crate::MssqlTargetTableState::Absent
534        );
535        assert_eq!(report.outputs()[1].target_table().table(), "east_orders");
536        assert_eq!(report.outputs()[1].load_mode(), LoadMode::AppendExisting);
537        assert_eq!(
538            report.outputs()[1]
539                .target_lifecycle_plan()
540                .expected_target_state(),
541            crate::MssqlTargetTableState::Exists
542        );
543        assert!(!report.sql_server_contacted());
544        assert!(!report.row_production_started());
545        assert!(!report.table_lifecycle_started());
546        assert!(!report.bulk_writer_started());
547        assert_eq!(report.phase_timings().len(), 2);
548        for phase_name in [OUTPUT_PLANNING_PHASE, SOURCE_REPORTING_PHASE] {
549            let timing = workflow_phase_timing(&report, phase_name)?;
550            assert!(timing.status().is_completed());
551            assert!(timing.elapsed_micros().is_some());
552        }
553        assert_eq!(source_scans.load(Ordering::SeqCst), 0);
554        Ok(())
555    }
556
557    #[tokio::test]
558    async fn dry_run_all_to_mssql_includes_registered_delta_source_reports()
559    -> Result<(), Box<dyn std::error::Error>> {
560        let table = DeltaLogTable::new("orders")?;
561        let mut session = DeltaFunnelSession::new(
562            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
563        )?;
564        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
565        let request = output_request(
566            source,
567            "orders_output",
568            "orders_sink",
569            LoadMode::CreateAndLoad,
570        )?;
571
572        let report = session.dry_run_all_to_mssql(&[request])?;
573
574        assert_eq!(report.outputs().len(), 1);
575        assert!(!report.query_used_source_scan_metadata_exhausted());
576        assert_eq!(
577            report.outputs()[0].sql_identity().state(),
578            MssqlDryRunSqlIdentityState::Absent
579        );
580        assert_eq!(report.outputs()[0].sql_identity().hash(), None);
581        assert_eq!(report.outputs()[0].sql_identity().reason(), None);
582        assert_eq!(
583            report.outputs()[0].source_usage_status(),
584            SourceUsageStatus::Used
585        );
586        assert_eq!(
587            report.outputs()[0].used_source_names(),
588            &["orders".to_owned()]
589        );
590        assert_eq!(report.sources().len(), 1);
591        let source = &report.sources()[0];
592        assert_eq!(source.source_name(), "orders");
593        assert_eq!(source.snapshot_version(), 1);
594        assert_eq!(source.protocol().source_name, "orders");
595        assert_eq!(source.file_count(), crate::FileCount::unavailable());
596        assert_eq!(
597            source.file_count_reason(),
598            Some(crate::ReportReasonCode::CostAvoidance)
599        );
600        assert!(!source.scan_metadata_exhausted());
601        assert_eq!(source.usage_status(), SourceUsageStatus::Used);
602        assert_eq!(source.used_by_output_names(), &["orders_output".to_owned()]);
603        assert!(!report.row_production_started());
604        Ok(())
605    }
606
607    #[tokio::test]
608    async fn dry_run_all_to_mssql_scan_summary_exhausts_provider_metadata_without_rows()
609    -> Result<(), Box<dyn std::error::Error>> {
610        let table = DeltaLogTable::new("orders")?;
611        let mut session = DeltaFunnelSession::new(
612            SessionOptions::new()
613                .with_default_mssql_connection(secret_connection()?)
614                .with_validation_options(ValidationOptions::new().with_dry_run_scan_summary_mode(
615                    crate::DryRunScanSummaryMode::ExhaustScanMetadata,
616                )),
617        )?;
618        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
619        let request = output_request(
620            source,
621            "orders_output",
622            "orders_sink",
623            LoadMode::CreateAndLoad,
624        )?;
625
626        let report = session
627            .dry_run_all_to_mssql_with_scan_summary(&[request])
628            .await?;
629
630        assert_eq!(report.outputs().len(), 1);
631        assert!(!report.outputs()[0].row_production_started());
632        assert_eq!(report.sources().len(), 1);
633        let source = &report.sources()[0];
634        assert_eq!(source.source_name(), "orders");
635        assert_eq!(source.usage_status(), SourceUsageStatus::Used);
636        assert_eq!(source.used_by_output_names(), &["orders_output".to_owned()]);
637        assert_eq!(source.provider_stats_reason(), None);
638        let stats = source
639            .provider_read_stats()
640            .ok_or("expected provider stats from dry-run scan summary")?;
641        assert_eq!(stats.source_name, "orders");
642        assert_eq!(stats.files_started, 0);
643        assert_eq!(stats.files_completed, 0);
644        assert_eq!(stats.batches_produced, 0);
645        assert_eq!(stats.rows_produced, 0);
646        match stats.scan_metadata_exhausted {
647            Some(true) => {
648                assert_eq!(
649                    source.file_count(),
650                    crate::FileCount::exact(stats.files_planned)
651                );
652                assert_eq!(source.file_count_reason(), None);
653            }
654            Some(false) => {
655                assert_eq!(
656                    source.file_count(),
657                    crate::FileCount::estimated(stats.files_planned)
658                );
659                assert_eq!(source.file_count_reason(), None);
660            }
661            None => {
662                assert_eq!(source.file_count(), crate::FileCount::unavailable());
663                assert_eq!(
664                    source.file_count_reason(),
665                    Some(crate::ReportReasonCode::CapabilityUnavailable)
666                );
667            }
668        }
669        assert_eq!(
670            report.query_used_source_scan_metadata_exhausted(),
671            source.scan_metadata_exhausted()
672        );
673        Ok(())
674    }
675
676    #[tokio::test]
677    async fn dry_run_all_to_mssql_reports_multi_source_usage_when_lineage_is_known()
678    -> Result<(), Box<dyn std::error::Error>> {
679        let orders_table = DeltaLogTable::new("orders")?;
680        let customers_table = DeltaLogTable::new("customers")?;
681        let inventory_table = DeltaLogTable::new("inventory")?;
682        let mut session = DeltaFunnelSession::new(
683            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
684        )?;
685        session.delta_lake(DeltaSourceConfig::new("orders", orders_table.uri()))?;
686        session.delta_lake(DeltaSourceConfig::new("customers", customers_table.uri()))?;
687        session.delta_lake(DeltaSourceConfig::new("inventory", inventory_table.uri()))?;
688        let joined = session
689            .table_from_sql(
690                "select orders.id from orders inner join customers on orders.id = customers.id",
691            )
692            .await?;
693        let request = output_request(
694            joined,
695            "joined_output",
696            "joined_sink",
697            LoadMode::CreateAndLoad,
698        )?;
699
700        let report = session.dry_run_all_to_mssql(&[request])?;
701
702        assert!(!report.query_used_source_scan_metadata_exhausted());
703        assert_eq!(
704            report.outputs()[0].source_usage_status(),
705            SourceUsageStatus::Used
706        );
707        assert_eq!(
708            report.outputs()[0].used_source_names(),
709            &["orders".to_owned(), "customers".to_owned()]
710        );
711        assert_eq!(report.sources().len(), 3);
712        for source in report.sources() {
713            match source.source_name() {
714                "orders" | "customers" => {
715                    assert_eq!(source.usage_status(), SourceUsageStatus::Used);
716                    assert_eq!(source.used_by_output_names(), &["joined_output".to_owned()]);
717                }
718                "inventory" => {
719                    assert_eq!(source.usage_status(), SourceUsageStatus::NotUsed);
720                    assert!(source.used_by_output_names().is_empty());
721                }
722                name => return Err(format!("unexpected source report: {name}").into()),
723            }
724        }
725        Ok(())
726    }
727
728    #[tokio::test]
729    async fn dry_run_all_to_mssql_rejects_execute_request_before_row_side_effects()
730    -> Result<(), Box<dyn std::error::Error>> {
731        let mut session = DeltaFunnelSession::new(
732            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
733        )?;
734        let (source_provider, source_scans) = scan_counting_marker_region_provider("shared")?;
735        session
736            .context()
737            .register_table("orders_source", source_provider)?;
738        let output = session
739            .table_from_sql("select marker from orders_source")
740            .await?;
741        let request = execute_output_request(
742            output,
743            "orders_output",
744            "orders_sink",
745            LoadMode::AppendExisting,
746        )?;
747
748        let error = session.dry_run_all_to_mssql(&[request]);
749
750        assert!(matches!(
751            error,
752            Err(DeltaFunnelError::MssqlWorkflowPlanning { message })
753                if message.contains("dry_run_all_to_mssql requires RunMode::DryRun")
754        ));
755        assert_eq!(source_scans.load(Ordering::SeqCst), 0);
756        Ok(())
757    }
758
759    #[tokio::test]
760    async fn dry_run_all_to_mssql_rejects_missing_connection_before_row_side_effects()
761    -> Result<(), Box<dyn std::error::Error>> {
762        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
763        let (source_provider, source_scans) = scan_counting_marker_region_provider("shared")?;
764        session
765            .context()
766            .register_table("orders_source", source_provider)?;
767        let output = session
768            .table_from_sql("select marker from orders_source")
769            .await?;
770        let request = output_request(
771            output,
772            "orders_output",
773            "orders_sink",
774            LoadMode::AppendExisting,
775        )?;
776
777        let error = session.dry_run_all_to_mssql(&[request]);
778
779        assert!(matches!(
780            error,
781            Err(DeltaFunnelError::MissingMssqlConnection { output_name })
782                if output_name == "orders_output"
783        ));
784        assert_eq!(source_scans.load(Ordering::SeqCst), 0);
785        Ok(())
786    }
787
788    #[tokio::test]
789    async fn dry_run_all_to_mssql_rejects_duplicate_output_names()
790    -> Result<(), Box<dyn std::error::Error>> {
791        let mut session = DeltaFunnelSession::new(
792            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
793        )?;
794        let west = session.table_from_sql("select 1 as id").await?;
795        let east = session.table_from_sql("select 2 as id").await?;
796        let west = output_request(
797            west,
798            "orders_output",
799            "west_orders",
800            LoadMode::AppendExisting,
801        )?;
802        let east = output_request(
803            east,
804            "orders_output",
805            "east_orders",
806            LoadMode::AppendExisting,
807        )?;
808
809        let error = session.dry_run_all_to_mssql(&[west, east]);
810
811        assert!(matches!(
812            error,
813            Err(DeltaFunnelError::MssqlWorkflowPlanning { message })
814                if message.contains("write_all output names must be unique")
815                    && message.contains("orders_output")
816        ));
817        Ok(())
818    }
819
820    #[tokio::test]
821    async fn dry_run_to_mssql_rejects_execute_request_before_planning()
822    -> Result<(), Box<dyn std::error::Error>> {
823        let mut session = DeltaFunnelSession::new(
824            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
825        )?;
826        let output = session.table_from_sql("select 1 as id").await?;
827        let request = execute_output_request(
828            output,
829            "orders_output",
830            "orders_sink",
831            LoadMode::AppendExisting,
832        )?;
833
834        let error = session.dry_run_to_mssql(&request);
835
836        assert!(matches!(
837            error,
838            Err(DeltaFunnelError::MssqlWorkflowPlanning { message })
839                if message.contains("dry_run_to_mssql requires RunMode::DryRun")
840        ));
841        Ok(())
842    }
843
844    #[test]
845    fn dry_run_to_mssql_rejects_missing_connection_before_side_effects()
846    -> Result<(), Box<dyn std::error::Error>> {
847        let table = DeltaLogTable::new("orders")?;
848        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
849        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
850        let request = output_request(
851            source,
852            "orders_output",
853            "orders_sink",
854            LoadMode::AppendExisting,
855        )?;
856
857        let error = session.dry_run_to_mssql(&request);
858
859        assert!(matches!(
860            error,
861            Err(DeltaFunnelError::MissingMssqlConnection { output_name })
862                if output_name == "orders_output"
863        ));
864        Ok(())
865    }
866
867    #[test]
868    fn dry_run_to_mssql_accepts_replace_before_side_effects()
869    -> Result<(), Box<dyn std::error::Error>> {
870        let table = DeltaLogTable::new("orders")?;
871        let mut session = DeltaFunnelSession::new(
872            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
873        )?;
874        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
875        let request = output_request(source, "orders_output", "orders_sink", LoadMode::Replace)?;
876
877        let report = session.dry_run_to_mssql(&request)?;
878
879        assert_eq!(report.output_name(), "orders_output");
880        assert_eq!(report.load_mode(), LoadMode::Replace);
881        assert_eq!(report.target_table().table(), "orders_sink");
882        assert!(report.target_ddl_plan().create_table_sql_present());
883        assert_eq!(report.target_ddl_plan().create_table_sql(), None);
884        assert!(report.target_lifecycle_plan().create_table_sql_required());
885        assert_eq!(
886            report.target_lifecycle_plan().expected_target_state(),
887            crate::MssqlTargetTableState::ExistsOrAbsent
888        );
889        assert!(!report.sql_server_contacted());
890        assert!(!report.row_production_started());
891        assert!(!report.table_lifecycle_started());
892        assert!(!report.bulk_writer_started());
893        Ok(())
894    }
895
896    #[test]
897    fn dry_run_to_mssql_report_debug_redacts_connection_material()
898    -> Result<(), Box<dyn std::error::Error>> {
899        let table = DeltaLogTable::new("orders")?;
900        let mut session = DeltaFunnelSession::new(
901            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
902        )?;
903        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
904        let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders_sink")?)
905            .with_connection(override_connection()?);
906        let request = OutputWritePlan::new(
907            source,
908            MssqlOutputTarget::new("orders_output", target_config, RunMode::DryRun),
909        );
910
911        let report = session.dry_run_to_mssql(&request)?;
912        let debug = format!("{report:?}");
913
914        assert!(debug.contains("orders_output"));
915        assert!(debug.contains("warehouse-override"));
916        assert!(!debug.contains("secret-token"));
917        assert!(!debug.contains("override-secret"));
918        assert!(!debug.contains("password"));
919        assert!(!debug.contains("server=tcp"));
920        Ok(())
921    }
922}
delta_funnel/orchestrator/session/dry_run_report.rs

delta_funnel/orchestrator/session/
dry_run_report.rs
