delta-funnel 0.3.3

use crate::{
    DeltaFunnelError, PhaseTimingReport, ReportReasonCode, observability,
    progress::{ProgressEvent, ProgressOperation, ProgressPhase, ProgressReporter},
    report::sql_server::{
        MssqlDryRunOutputReport, MssqlDryRunSqlIdentityReport, MssqlDryRunWorkflowReport,
    },
    report::{PhaseTimer, usize_to_u64_saturating},
};

use super::{
    DeltaFunnelSession, LazyTable, LazyTableKind, OutputWritePlan, PlannedMssqlOutput, RunMode,
    SourceUsageStatus, sql_server_workflows::ensure_unique_write_all_output_names,
};
use tracing::Instrument;

pub(super) fn stable_sql_identity_hash(sql: &str) -> String {
    const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325;
    const FNV_PRIME: u64 = 0x100000001b3;

    let mut hash = FNV_OFFSET_BASIS;
    for byte in sql.as_bytes() {
        hash ^= u64::from(*byte);
        hash = hash.wrapping_mul(FNV_PRIME);
    }

    format!("{hash:016x}")
}

const QUERY_EXECUTION_PHASE: &str = "query_execution";
const BATCH_SHAPING_PHASE: &str = "batch_shaping";
const SQL_WRITE_PHASE: &str = "sql_write";
const FINALIZE_PHASE: &str = "finalize";
const VALIDATION_PHASE: &str = "validation";
const OUTPUT_PLANNING_PHASE: &str = "output_planning";
const SOURCE_REPORTING_PHASE: &str = "source_reporting";

impl DeltaFunnelSession {
    /// Dry-runs one selected lazy table as an MSSQL output.
    ///
    /// The method reuses the same session output planner as execute mode, then
    /// stops before physical DataFusion planning, row production, SQL Server
    /// lifecycle work, bulk writer construction, or row writes.
    ///
    /// # Errors
    ///
    /// Returns an MSSQL planning error when the request is not in
    /// [`RunMode::DryRun`], or the first error from session output planning.
    pub fn dry_run_to_mssql(
        &self,
        request: &OutputWritePlan,
    ) -> Result<MssqlDryRunOutputReport, DeltaFunnelError> {
        ensure_dry_run_mode(request.target().run_mode())?;
        self.plan_dry_run_output(request)
    }

    pub(crate) fn dry_run_to_mssql_with_reporter(
        &self,
        request: &OutputWritePlan,
        reporter: ProgressReporter,
    ) -> Result<MssqlDryRunOutputReport, DeltaFunnelError> {
        ensure_dry_run_mode(request.target().run_mode())?;

        reporter.emit(&ProgressEvent::started(ProgressOperation::DryRunToMssql));
        reporter.emit(&ProgressEvent::phase_changed(
            ProgressPhase::PlanningOutput,
            Some(request.target().output_name()),
        ));
        let result = self.plan_dry_run_output(request);
        reporter.emit(&if result.is_ok() {
            ProgressEvent::completed()
        } else {
            ProgressEvent::failed()
        });
        result
    }

    /// Dry-runs multiple selected lazy tables as one MSSQL output workflow.
    ///
    /// The method plans each selected output in caller-provided order and stops
    /// before cache materialization, physical DataFusion planning, row
    /// production, SQL Server lifecycle work, bulk writer construction, or row
    /// writes.
    ///
    /// # Errors
    ///
    /// Returns the first duplicate-output, run-mode, or output-planning error.
    pub fn dry_run_all_to_mssql(
        &self,
        requests: &[OutputWritePlan],
    ) -> Result<MssqlDryRunWorkflowReport, DeltaFunnelError> {
        validate_dry_run_all_requests(requests)?;
        self.build_dry_run_all_report(requests, None)
    }

    /// Builds the same report as [`Self::dry_run_all_to_mssql`] while emitting
    /// live progress events.
    ///
    /// Progress events describe the work currently being performed. The
    /// returned report contains the final planned outputs and source summary.
    /// Invalid requests fail before progress starts, and empty requests emit no
    /// progress events.
    fn dry_run_all_to_mssql_with_progress(
        &self,
        requests: &[OutputWritePlan],
        reporter: ProgressReporter,
    ) -> Result<MssqlDryRunWorkflowReport, DeltaFunnelError> {
        validate_dry_run_all_requests(requests)?;
        if requests.is_empty() {
            return self.build_dry_run_all_report(requests, None);
        }

        reporter.emit(&ProgressEvent::started(ProgressOperation::DryRunAllToMssql));
        let result = self.build_dry_run_all_report(requests, Some(&reporter));
        reporter.emit(&if result.is_ok() {
            ProgressEvent::completed()
        } else {
            ProgressEvent::failed()
        });
        result
    }

    /// Runs the multi-output dry run inside one workflow observability boundary.
    ///
    /// Passing a reporter adds live progress events without changing the
    /// workflow tracing span or its start and finish events.
    pub(crate) fn dry_run_all_to_mssql_with_observability(
        &self,
        requests: &[OutputWritePlan],
        reporter: Option<ProgressReporter>,
    ) -> Result<MssqlDryRunWorkflowReport, DeltaFunnelError> {
        let output_count = requests.len();
        let span = observability::workflow_span(RunMode::DryRun, output_count);
        let _guard = span.enter();
        observability::workflow_started(RunMode::DryRun, output_count);
        let result = match reporter {
            Some(reporter) => self.dry_run_all_to_mssql_with_progress(requests, reporter),
            None => self.dry_run_all_to_mssql(requests),
        };
        observability::workflow_finished(RunMode::DryRun, output_count, &result);
        result
    }

    /// Builds the final report from the planned outputs and their sources.
    ///
    /// When provided, the reporter announces phase changes while the report is
    /// built. It does not change the report contents or planning behavior.
    fn build_dry_run_all_report(
        &self,
        requests: &[OutputWritePlan],
        reporter: Option<&ProgressReporter>,
    ) -> Result<MssqlDryRunWorkflowReport, DeltaFunnelError> {
        let planning_timer = PhaseTimer::start(OUTPUT_PLANNING_PHASE);
        let outputs = self.plan_dry_run_outputs(requests, reporter)?;
        let planning_timing = planning_timer.completed();
        if let Some(reporter) = reporter {
            reporter.emit(&ProgressEvent::phase_changed(
                ProgressPhase::ReportingSources,
                None,
            ));
        }
        let source_timer = PhaseTimer::start(SOURCE_REPORTING_PHASE);
        let sources = self.source_reports_for_dry_run_outputs(&outputs)?;
        let source_timing = source_timer.completed();

        Ok(MssqlDryRunWorkflowReport::new(outputs, sources)
            .with_phase_timings(vec![planning_timing, source_timing]))
    }

    /// Runs [`Self::dry_run_all_to_mssql`] inside a DeltaFunnel workflow tracing span.
    ///
    /// # Errors
    ///
    /// Returns the same error as [`Self::dry_run_all_to_mssql`].
    pub fn dry_run_all_to_mssql_with_tracing(
        &self,
        requests: &[OutputWritePlan],
    ) -> Result<MssqlDryRunWorkflowReport, DeltaFunnelError> {
        self.dry_run_all_to_mssql_with_observability(requests, None)
    }

    /// Dry-runs multiple selected lazy tables and honors source scan-summary options.
    ///
    /// This method is async because
    /// [`crate::DryRunScanSummaryMode::ExhaustScanMetadata`] requires
    /// DataFusion physical planning to expose provider scan metadata. It still
    /// stops before row production, SQL Server lifecycle work, bulk writer
    /// construction, or row writes.
    ///
    /// # Errors
    ///
    /// Returns the first duplicate-output, run-mode, output-planning, or
    /// DataFusion physical-planning error.
    pub async fn dry_run_all_to_mssql_with_scan_summary(
        &self,
        requests: &[OutputWritePlan],
    ) -> Result<MssqlDryRunWorkflowReport, DeltaFunnelError> {
        let planning_timer = PhaseTimer::start(OUTPUT_PLANNING_PHASE);
        let outputs = self.plan_dry_run_all_outputs(requests)?;
        let planning_timing = planning_timer.completed();
        let source_timer = PhaseTimer::start(SOURCE_REPORTING_PHASE);
        let sources = match self
            .options
            .validation_options()
            .dry_run_scan_summary_mode()
        {
            crate::DryRunScanSummaryMode::MetadataOnly => {
                self.source_reports_for_dry_run_outputs(&outputs)?
            }
            crate::DryRunScanSummaryMode::ExhaustScanMetadata => self
                .source_reports_for_dry_run_outputs_with_provider_stats(
                    &outputs,
                    self.provider_read_stats_for_dry_run_outputs(&outputs)
                        .await?,
                )?,
        };
        let source_timing = source_timer.completed();

        Ok(MssqlDryRunWorkflowReport::new(outputs, sources)
            .with_phase_timings(vec![planning_timing, source_timing]))
    }

    /// Runs [`Self::dry_run_all_to_mssql_with_scan_summary`] inside a workflow tracing span.
    ///
    /// # Errors
    ///
    /// Returns the same error as [`Self::dry_run_all_to_mssql_with_scan_summary`].
    pub async fn dry_run_all_to_mssql_with_scan_summary_with_tracing(
        &self,
        requests: &[OutputWritePlan],
    ) -> Result<MssqlDryRunWorkflowReport, DeltaFunnelError> {
        let output_count = requests.len();
        async move {
            observability::workflow_started(RunMode::DryRun, output_count);
            let result = self.dry_run_all_to_mssql_with_scan_summary(requests).await;
            observability::workflow_finished(RunMode::DryRun, output_count, &result);
            result
        }
        .instrument(observability::workflow_span(RunMode::DryRun, output_count))
        .await
    }

    fn plan_dry_run_all_outputs(
        &self,
        requests: &[OutputWritePlan],
    ) -> Result<Vec<MssqlDryRunOutputReport>, DeltaFunnelError> {
        validate_dry_run_all_requests(requests)?;
        self.plan_dry_run_outputs(requests, None)
    }

    /// Plans each requested output in caller-provided order.
    ///
    /// When provided, the reporter announces each output and its position
    /// before that output is planned.
    fn plan_dry_run_outputs(
        &self,
        requests: &[OutputWritePlan],
        reporter: Option<&ProgressReporter>,
    ) -> Result<Vec<MssqlDryRunOutputReport>, DeltaFunnelError> {
        let output_count = usize_to_u64_saturating(requests.len());

        requests
            .iter()
            .enumerate()
            .map(|(output_index, request)| {
                let output_index = usize_to_u64_saturating(output_index.saturating_add(1));
                let output_reporter =
                    reporter.and_then(|reporter| reporter.for_output(output_index, output_count));
                if let Some(output_reporter) = output_reporter {
                    output_reporter.emit(&ProgressEvent::phase_changed(
                        ProgressPhase::PlanningOutput,
                        Some(request.target().output_name()),
                    ));
                }
                self.plan_dry_run_output(request)
            })
            .collect()
    }

    fn plan_dry_run_output(
        &self,
        request: &OutputWritePlan,
    ) -> Result<MssqlDryRunOutputReport, DeltaFunnelError> {
        let target = request.target();
        let target_config = target.target();
        let output_span = observability::output_span(
            target.output_name(),
            target_config.table(),
            target_config.load_mode(),
        );
        let _guard = output_span.enter();
        observability::output_started(
            target.output_name(),
            target_config.table(),
            target_config.load_mode(),
        );
        let result = self.plan_dry_run_output_report(request);
        match &result {
            Ok(_) => observability::output_completed(
                target.output_name(),
                target_config.table(),
                target_config.load_mode(),
            ),
            Err(error) => observability::output_failed(
                target.output_name(),
                target_config.table(),
                target_config.load_mode(),
                &error.to_string(),
            ),
        }

        result
    }

    fn plan_dry_run_output_report(
        &self,
        request: &OutputWritePlan,
    ) -> Result<MssqlDryRunOutputReport, DeltaFunnelError> {
        let planned = self.plan_mssql_output(request)?;

        let phase_timings = dry_run_output_phase_timings(planned.phase_timings().to_vec());
        self.dry_run_output_report_for_plan(planned, phase_timings)
    }

    fn dry_run_output_report_for_plan(
        &self,
        planned_output: PlannedMssqlOutput,
        phase_timings: Vec<PhaseTimingReport>,
    ) -> Result<MssqlDryRunOutputReport, DeltaFunnelError> {
        let sql_identity = self.sql_identity_for_lazy_table(planned_output.table());
        let (source_usage_status, used_source_names) =
            self.source_usage_for_lazy_table(planned_output.table())?;
        Ok(MssqlDryRunOutputReport::new(
            planned_output,
            sql_identity,
            source_usage_status,
            used_source_names,
            phase_timings,
        ))
    }

    fn source_usage_for_lazy_table(
        &self,
        table: &LazyTable,
    ) -> Result<(SourceUsageStatus, Vec<String>), DeltaFunnelError> {
        let Some(source_ids) = self.known_source_dependencies_for_table(table)? else {
            return Ok((SourceUsageStatus::Unknown, Vec::new()));
        };

        let used_source_names = self
            .sources
            .iter()
            .filter(|source| source_ids.contains(&source.table().id()))
            .map(|source| source.name().to_owned())
            .collect::<Vec<_>>();
        let source_usage_status = if used_source_names.is_empty() {
            SourceUsageStatus::NotUsed
        } else {
            SourceUsageStatus::Used
        };

        Ok((source_usage_status, used_source_names))
    }

    fn sql_identity_for_lazy_table(&self, table: &LazyTable) -> MssqlDryRunSqlIdentityReport {
        if table.kind() != LazyTableKind::DerivedSql {
            return MssqlDryRunSqlIdentityReport::absent();
        }

        match self.sql_text_for_derived_table(table) {
            Ok(sql_text) => {
                MssqlDryRunSqlIdentityReport::present(stable_sql_identity_hash(sql_text))
            }
            Err(_) => {
                MssqlDryRunSqlIdentityReport::unavailable(ReportReasonCode::CapabilityUnavailable)
            }
        }
    }
}

fn dry_run_output_phase_timings(
    mut planning_timings: Vec<PhaseTimingReport>,
) -> Vec<PhaseTimingReport> {
    planning_timings.extend([
        PhaseTimingReport::skipped(QUERY_EXECUTION_PHASE, ReportReasonCode::DryRun),
        PhaseTimingReport::skipped(BATCH_SHAPING_PHASE, ReportReasonCode::DryRun),
        PhaseTimingReport::skipped(SQL_WRITE_PHASE, ReportReasonCode::DryRun),
        PhaseTimingReport::skipped(FINALIZE_PHASE, ReportReasonCode::DryRun),
        PhaseTimingReport::skipped(VALIDATION_PHASE, ReportReasonCode::DryRun),
    ]);
    planning_timings
}

fn ensure_dry_run_mode(run_mode: RunMode) -> Result<(), DeltaFunnelError> {
    match run_mode {
        RunMode::DryRun => Ok(()),
        RunMode::Execute => Err(DeltaFunnelError::MssqlWorkflowPlanning {
            message: "dry_run_to_mssql requires RunMode::DryRun; use write_to_mssql for execution"
                .to_owned(),
        }),
    }
}

fn ensure_write_all_dry_run_mode(run_mode: RunMode) -> Result<(), DeltaFunnelError> {
    match run_mode {
        RunMode::DryRun => Ok(()),
        RunMode::Execute => Err(DeltaFunnelError::MssqlWorkflowPlanning {
            message: "dry_run_all_to_mssql requires RunMode::DryRun; use write_all for execution"
                .to_owned(),
        }),
    }
}

fn validate_dry_run_all_requests(requests: &[OutputWritePlan]) -> Result<(), DeltaFunnelError> {
    ensure_unique_write_all_output_names(requests)?;
    for request in requests {
        ensure_write_all_dry_run_mode(request.target().run_mode())?;
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use std::{
        panic::{AssertUnwindSafe, catch_unwind, resume_unwind},
        sync::{
            Arc, Barrier, Mutex,
            atomic::{AtomicBool, AtomicUsize, Ordering},
        },
        thread,
    };

    use crate::{
        DeltaFunnelError, DeltaSourceConfig, LoadMode, MssqlOutputTarget, MssqlTargetConfig,
        MssqlTargetTable, OutputStatus, ReportReasonCode, ValidationOptions, ValidationStatus,
        WorkflowStatus,
        progress::{ProgressEventKind, ProgressOperation, ProgressPhase, ProgressReporter},
    };

    use super::super::{
        DeltaFunnelSession, LazyTableKind, OutputWritePlan, RunMode, SessionOptions,
        SourceUsageStatus,
        sql_server_workflows::{OUTPUT_SCHEMA_PLANNING_PHASE, SQL_TARGET_PLANNING_PHASE},
        test_support::{
            DeltaLogTable, execute_output_request, output_request, override_connection,
            scan_counting_marker_region_provider, secret_connection,
        },
    };
    use super::{
        BATCH_SHAPING_PHASE, FINALIZE_PHASE, OUTPUT_PLANNING_PHASE, QUERY_EXECUTION_PHASE,
        SOURCE_REPORTING_PHASE, SQL_WRITE_PHASE, VALIDATION_PHASE, stable_sql_identity_hash,
    };
    use crate::MssqlDryRunSqlIdentityState;

    const LAZY_SQL_PLANNING_PHASE: &str = "lazy_sql_planning";

    type RecordedProgress = (
        ProgressEventKind,
        Option<ProgressOperation>,
        Option<ProgressPhase>,
    );

    fn recording_reporter() -> (ProgressReporter, Arc<Mutex<Vec<RecordedProgress>>>) {
        let events = Arc::new(Mutex::new(Vec::new()));
        let callback_events = Arc::clone(&events);
        let reporter = ProgressReporter::new(move |event| {
            let mut events = match callback_events.lock() {
                Ok(events) => events,
                Err(poisoned) => poisoned.into_inner(),
            };
            events.push((event.kind(), event.operation(), event.phase()));
        });
        (reporter, events)
    }

    fn progress_events(events: &Mutex<Vec<RecordedProgress>>) -> Vec<RecordedProgress> {
        match events.lock() {
            Ok(events) => events.clone(),
            Err(poisoned) => poisoned.into_inner().clone(),
        }
    }

    #[test]
    fn sql_identity_status_and_hash_are_stable() {
        assert_eq!(MssqlDryRunSqlIdentityState::Present.as_str(), "present");
        assert_eq!(MssqlDryRunSqlIdentityState::Absent.to_string(), "absent");
        assert_eq!(
            MssqlDryRunSqlIdentityState::Unavailable.as_str(),
            "unavailable"
        );
        assert_eq!(
            stable_sql_identity_hash("select marker where region = 'west'"),
            "cbd6889e027b0f88"
        );
    }

    fn phase_timing<'a>(
        report: &'a crate::MssqlDryRunOutputReport,
        phase_name: &str,
    ) -> Result<&'a crate::PhaseTimingReport, Box<dyn std::error::Error>> {
        report
            .phase_timings()
            .iter()
            .find(|timing| timing.phase_name() == phase_name)
            .ok_or_else(|| format!("missing phase timing `{phase_name}`").into())
    }

    fn workflow_phase_timing<'a>(
        report: &'a crate::MssqlDryRunWorkflowReport,
        phase_name: &str,
    ) -> Result<&'a crate::PhaseTimingReport, Box<dyn std::error::Error>> {
        report
            .phase_timings()
            .iter()
            .find(|timing| timing.phase_name() == phase_name)
            .ok_or_else(|| format!("missing workflow phase timing `{phase_name}`").into())
    }

    #[tokio::test]
    async fn dry_run_to_mssql_plans_output_without_row_or_writer_side_effects()
    -> Result<(), Box<dyn std::error::Error>> {
        let mut session = DeltaFunnelSession::new(
            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
        )?;
        let (source_provider, source_scans) = scan_counting_marker_region_provider("shared")?;
        session
            .context()
            .register_table("orders_source", source_provider)?;
        let output = session
            .table_from_sql("select marker from orders_source where region = 'west'")
            .await?;
        let request = output_request(
            output,
            "west_output",
            "west_orders",
            LoadMode::CreateAndLoad,
        )?;
        let (reporter, events) = recording_reporter();

        let report = session.dry_run_to_mssql_with_reporter(&request, reporter)?;

        assert_eq!(report.output_name(), "west_output");
        assert_eq!(report.run_mode(), RunMode::DryRun);
        assert_eq!(report.status(), OutputStatus::dry_run_planned());
        assert_eq!(
            report.validation_status(),
            ValidationStatus::skipped(ReportReasonCode::DryRun)
        );
        assert_eq!(report.target_table().schema(), Some("dbo"));
        assert_eq!(report.target_table().table(), "west_orders");
        assert_eq!(report.load_mode(), LoadMode::CreateAndLoad);
        assert_eq!(report.target_schema_plan().mappings().len(), 1);
        assert!(report.target_ddl_plan().create_table_sql_present());
        assert!(report.target_lifecycle_plan().create_table_sql_required());
        assert_eq!(
            report.target_lifecycle_plan().expected_target_state(),
            crate::MssqlTargetTableState::Absent
        );
        assert_eq!(
            report.planned_output().output_plan().target_table().table(),
            "west_orders"
        );
        assert_eq!(
            report
                .planned_output()
                .output_plan()
                .schema_mappings()
                .len(),
            1
        );
        assert_eq!(report.output_schema().len(), 1);
        assert_eq!(report.output_schema()[0].index(), 0);
        assert_eq!(report.output_schema()[0].name(), "marker");
        assert_eq!(report.output_schema()[0].arrow_type(), "Utf8");
        assert!(!report.output_schema()[0].nullable());
        assert_eq!(report.source_usage_status(), SourceUsageStatus::NotUsed);
        assert!(report.used_source_names().is_empty());
        assert_eq!(report.output_row_count(), crate::RowCount::unavailable());
        assert_eq!(
            report.output_row_count_reason(),
            Some(ReportReasonCode::NotExecuted)
        );
        assert_eq!(
            report.sql_identity().state(),
            MssqlDryRunSqlIdentityState::Present
        );
        assert_eq!(report.sql_identity().hash(), Some("a65390dacb7eb6f1"));
        assert_eq!(report.sql_identity().reason(), None);
        let debug = format!("{report:?}");
        assert!(debug.contains("a65390dacb7eb6f1"));
        assert!(!debug.contains("select marker"));
        assert!(!debug.contains("region = 'west'"));
        assert!(!report.sql_server_contacted());
        assert!(!report.row_production_started());
        assert!(!report.table_lifecycle_started());
        assert!(!report.bulk_writer_started());
        assert_eq!(report.phase_timings().len(), 8);
        for phase_name in [
            LAZY_SQL_PLANNING_PHASE,
            OUTPUT_SCHEMA_PLANNING_PHASE,
            SQL_TARGET_PLANNING_PHASE,
        ] {
            let planning = phase_timing(&report, phase_name)?;
            assert!(planning.status().is_completed());
            assert!(planning.elapsed_micros().is_some());
        }
        for phase_name in [
            QUERY_EXECUTION_PHASE,
            BATCH_SHAPING_PHASE,
            SQL_WRITE_PHASE,
            FINALIZE_PHASE,
            VALIDATION_PHASE,
        ] {
            let timing = phase_timing(&report, phase_name)?;
            assert_eq!(
                timing.status(),
                crate::PhaseStatus::skipped(ReportReasonCode::DryRun)
            );
            assert_eq!(timing.elapsed_micros(), None);
        }
        assert_eq!(source_scans.load(Ordering::SeqCst), 0);
        assert_eq!(
            progress_events(&events),
            vec![
                (
                    ProgressEventKind::Started,
                    Some(ProgressOperation::DryRunToMssql),
                    None,
                ),
                (
                    ProgressEventKind::PhaseChanged,
                    None,
                    Some(ProgressPhase::PlanningOutput),
                ),
                (ProgressEventKind::Completed, None, None),
            ]
        );
        Ok(())
    }

    #[tokio::test]
    async fn dry_run_all_to_mssql_reports_all_outputs_without_row_side_effects()
    -> Result<(), Box<dyn std::error::Error>> {
        let mut session = DeltaFunnelSession::new(
            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
        )?;
        let (source_provider, source_scans) = scan_counting_marker_region_provider("shared")?;
        session
            .context()
            .register_table("orders_source", source_provider)?;
        let west = session
            .table_from_sql("select marker from orders_source where region = 'west'")
            .await?;
        let east = session
            .table_from_sql("select marker from orders_source where region = 'east'")
            .await?;
        let west_table_id = west.id();
        let west_table_name = west.name().to_owned();
        let west = output_request(west, "west_output", "west_orders", LoadMode::CreateAndLoad)?;
        let east = output_request(east, "east_output", "east_orders", LoadMode::AppendExisting)?;
        let events = Arc::new(Mutex::new(Vec::new()));
        let callback_events = Arc::clone(&events);
        let reporter = ProgressReporter::new(move |event| {
            let mut events = match callback_events.lock() {
                Ok(events) => events,
                Err(poisoned) => poisoned.into_inner(),
            };
            events.push((
                event.kind(),
                event.operation(),
                event.phase(),
                event.output_name().map(str::to_owned),
                event.output_index(),
                event.output_count(),
            ));
        });

        let report = session.dry_run_all_to_mssql_with_progress(&[west, east], reporter)?;

        assert_eq!(report.run_mode(), RunMode::DryRun);
        assert_eq!(report.status(), WorkflowStatus::success());
        assert_eq!(report.len(), 2);
        assert!(!report.is_empty());
        assert_eq!(report.outputs()[0].output_name(), "west_output");
        assert_eq!(report.outputs()[1].output_name(), "east_output");
        assert_eq!(report.outputs()[0].table_id(), west_table_id);
        assert_eq!(report.outputs()[0].table_kind(), LazyTableKind::DerivedSql);
        assert_eq!(report.outputs()[0].table_name(), west_table_name);
        assert_eq!(
            report.outputs()[0].status(),
            OutputStatus::dry_run_planned()
        );
        assert_eq!(
            report.outputs()[0].validation_status(),
            ValidationStatus::skipped(ReportReasonCode::DryRun)
        );
        assert_eq!(
            report.outputs()[0].output_row_count(),
            crate::RowCount::unavailable()
        );
        assert_eq!(
            report.outputs()[0].output_row_count_reason(),
            Some(ReportReasonCode::NotExecuted)
        );
        assert!(report.sources().is_empty());
        assert_eq!(report.outputs()[0].target_table().table(), "west_orders");
        assert_eq!(report.outputs()[0].load_mode(), LoadMode::CreateAndLoad);
        assert_eq!(
            report.outputs()[0]
                .target_lifecycle_plan()
                .expected_target_state(),
            crate::MssqlTargetTableState::Absent
        );
        assert_eq!(report.outputs()[1].target_table().table(), "east_orders");
        assert_eq!(report.outputs()[1].load_mode(), LoadMode::AppendExisting);
        assert_eq!(
            report.outputs()[1]
                .target_lifecycle_plan()
                .expected_target_state(),
            crate::MssqlTargetTableState::Exists
        );
        assert!(!report.sql_server_contacted());
        assert!(!report.row_production_started());
        assert!(!report.table_lifecycle_started());
        assert!(!report.bulk_writer_started());
        assert_eq!(report.phase_timings().len(), 2);
        for phase_name in [OUTPUT_PLANNING_PHASE, SOURCE_REPORTING_PHASE] {
            let timing = workflow_phase_timing(&report, phase_name)?;
            assert!(timing.status().is_completed());
            assert!(timing.elapsed_micros().is_some());
        }
        assert_eq!(source_scans.load(Ordering::SeqCst), 0);
        let events = match events.lock() {
            Ok(events) => events,
            Err(poisoned) => poisoned.into_inner(),
        };
        assert_eq!(
            events.as_slice(),
            [
                (
                    ProgressEventKind::Started,
                    Some(ProgressOperation::DryRunAllToMssql),
                    None,
                    None,
                    None,
                    None,
                ),
                (
                    ProgressEventKind::PhaseChanged,
                    None,
                    Some(ProgressPhase::PlanningOutput),
                    Some("west_output".to_owned()),
                    Some(1),
                    Some(2),
                ),
                (
                    ProgressEventKind::PhaseChanged,
                    None,
                    Some(ProgressPhase::PlanningOutput),
                    Some("east_output".to_owned()),
                    Some(2),
                    Some(2),
                ),
                (
                    ProgressEventKind::PhaseChanged,
                    None,
                    Some(ProgressPhase::ReportingSources),
                    None,
                    None,
                    None,
                ),
                (ProgressEventKind::Completed, None, None, None, None, None,),
            ]
        );
        Ok(())
    }

    #[test]
    fn empty_dry_run_all_emits_no_progress() -> Result<(), Box<dyn std::error::Error>> {
        let session = DeltaFunnelSession::new(SessionOptions::default())?;
        let (reporter, events) = recording_reporter();

        let report = session.dry_run_all_to_mssql_with_progress(&[], reporter)?;

        assert!(report.is_empty());
        assert!(progress_events(&events).is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn dry_run_all_to_mssql_includes_registered_delta_source_reports()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new("orders")?;
        let mut session = DeltaFunnelSession::new(
            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
        )?;
        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
        let request = output_request(
            source,
            "orders_output",
            "orders_sink",
            LoadMode::CreateAndLoad,
        )?;

        let report = session.dry_run_all_to_mssql(&[request])?;

        assert_eq!(report.outputs().len(), 1);
        assert!(!report.query_used_source_scan_metadata_exhausted());
        assert_eq!(
            report.outputs()[0].sql_identity().state(),
            MssqlDryRunSqlIdentityState::Absent
        );
        assert_eq!(report.outputs()[0].sql_identity().hash(), None);
        assert_eq!(report.outputs()[0].sql_identity().reason(), None);
        assert_eq!(
            report.outputs()[0].source_usage_status(),
            SourceUsageStatus::Used
        );
        assert_eq!(
            report.outputs()[0].used_source_names(),
            &["orders".to_owned()]
        );
        assert_eq!(report.sources().len(), 1);
        let source = &report.sources()[0];
        assert_eq!(source.source_name(), "orders");
        assert_eq!(source.snapshot_version(), 1);
        assert_eq!(source.protocol().source_name, "orders");
        assert_eq!(source.file_count(), crate::FileCount::unavailable());
        assert_eq!(
            source.file_count_reason(),
            Some(crate::ReportReasonCode::CostAvoidance)
        );
        assert!(!source.scan_metadata_exhausted());
        assert_eq!(source.usage_status(), SourceUsageStatus::Used);
        assert_eq!(source.used_by_output_names(), &["orders_output".to_owned()]);
        assert!(!report.row_production_started());
        Ok(())
    }

    #[tokio::test]
    async fn dry_run_all_to_mssql_scan_summary_exhausts_provider_metadata_without_rows()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new("orders")?;
        let mut session = DeltaFunnelSession::new(
            SessionOptions::new()
                .with_default_mssql_connection(secret_connection()?)
                .with_validation_options(ValidationOptions::new().with_dry_run_scan_summary_mode(
                    crate::DryRunScanSummaryMode::ExhaustScanMetadata,
                )),
        )?;
        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
        let request = output_request(
            source,
            "orders_output",
            "orders_sink",
            LoadMode::CreateAndLoad,
        )?;

        let report = session
            .dry_run_all_to_mssql_with_scan_summary(&[request])
            .await?;

        assert_eq!(report.outputs().len(), 1);
        assert!(!report.outputs()[0].row_production_started());
        assert_eq!(report.sources().len(), 1);
        let source = &report.sources()[0];
        assert_eq!(source.source_name(), "orders");
        assert_eq!(source.usage_status(), SourceUsageStatus::Used);
        assert_eq!(source.used_by_output_names(), &["orders_output".to_owned()]);
        assert_eq!(source.provider_stats_reason(), None);
        let stats = source
            .provider_read_stats()
            .ok_or("expected provider stats from dry-run scan summary")?;
        assert_eq!(stats.source_name, "orders");
        assert_eq!(stats.files_started, 0);
        assert_eq!(stats.files_completed, 0);
        assert_eq!(stats.batches_produced, 0);
        assert_eq!(stats.rows_produced, 0);
        match stats.scan_metadata_exhausted {
            Some(true) => {
                assert_eq!(
                    source.file_count(),
                    crate::FileCount::exact(stats.files_planned)
                );
                assert_eq!(source.file_count_reason(), None);
            }
            Some(false) => {
                assert_eq!(
                    source.file_count(),
                    crate::FileCount::estimated(stats.files_planned)
                );
                assert_eq!(source.file_count_reason(), None);
            }
            None => {
                assert_eq!(source.file_count(), crate::FileCount::unavailable());
                assert_eq!(
                    source.file_count_reason(),
                    Some(crate::ReportReasonCode::CapabilityUnavailable)
                );
            }
        }
        assert_eq!(
            report.query_used_source_scan_metadata_exhausted(),
            source.scan_metadata_exhausted()
        );
        Ok(())
    }

    #[tokio::test]
    async fn dry_run_all_to_mssql_reports_multi_source_usage_when_lineage_is_known()
    -> Result<(), Box<dyn std::error::Error>> {
        let orders_table = DeltaLogTable::new("orders")?;
        let customers_table = DeltaLogTable::new("customers")?;
        let inventory_table = DeltaLogTable::new("inventory")?;
        let mut session = DeltaFunnelSession::new(
            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
        )?;
        session.delta_lake(DeltaSourceConfig::new("orders", orders_table.uri()))?;
        session.delta_lake(DeltaSourceConfig::new("customers", customers_table.uri()))?;
        session.delta_lake(DeltaSourceConfig::new("inventory", inventory_table.uri()))?;
        let joined = session
            .table_from_sql(
                "select orders.id from orders inner join customers on orders.id = customers.id",
            )
            .await?;
        let request = output_request(
            joined,
            "joined_output",
            "joined_sink",
            LoadMode::CreateAndLoad,
        )?;

        let report = session.dry_run_all_to_mssql(&[request])?;

        assert!(!report.query_used_source_scan_metadata_exhausted());
        assert_eq!(
            report.outputs()[0].source_usage_status(),
            SourceUsageStatus::Used
        );
        assert_eq!(
            report.outputs()[0].used_source_names(),
            &["orders".to_owned(), "customers".to_owned()]
        );
        assert_eq!(report.sources().len(), 3);
        for source in report.sources() {
            match source.source_name() {
                "orders" | "customers" => {
                    assert_eq!(source.usage_status(), SourceUsageStatus::Used);
                    assert_eq!(source.used_by_output_names(), &["joined_output".to_owned()]);
                }
                "inventory" => {
                    assert_eq!(source.usage_status(), SourceUsageStatus::NotUsed);
                    assert!(source.used_by_output_names().is_empty());
                }
                name => return Err(format!("unexpected source report: {name}").into()),
            }
        }
        Ok(())
    }

    #[tokio::test]
    async fn dry_run_all_to_mssql_rejects_execute_request_before_row_side_effects()
    -> Result<(), Box<dyn std::error::Error>> {
        let mut session = DeltaFunnelSession::new(
            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
        )?;
        let (source_provider, source_scans) = scan_counting_marker_region_provider("shared")?;
        session
            .context()
            .register_table("orders_source", source_provider)?;
        let output = session
            .table_from_sql("select marker from orders_source")
            .await?;
        let request = execute_output_request(
            output,
            "orders_output",
            "orders_sink",
            LoadMode::AppendExisting,
        )?;

        let (reporter, events) = recording_reporter();

        let error = session.dry_run_all_to_mssql_with_progress(&[request], reporter);

        assert!(matches!(
            error,
            Err(DeltaFunnelError::MssqlWorkflowPlanning { message })
                if message.contains("dry_run_all_to_mssql requires RunMode::DryRun")
        ));
        assert_eq!(source_scans.load(Ordering::SeqCst), 0);
        assert!(progress_events(&events).is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn dry_run_all_to_mssql_rejects_missing_connection_before_row_side_effects()
    -> Result<(), Box<dyn std::error::Error>> {
        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
        let (source_provider, source_scans) = scan_counting_marker_region_provider("shared")?;
        session
            .context()
            .register_table("orders_source", source_provider)?;
        let output = session
            .table_from_sql("select marker from orders_source")
            .await?;
        let request = output_request(
            output,
            "orders_output",
            "orders_sink",
            LoadMode::AppendExisting,
        )?;

        let (reporter, events) = recording_reporter();

        let error = session.dry_run_all_to_mssql_with_progress(&[request], reporter);

        assert!(matches!(
            error,
            Err(DeltaFunnelError::MissingMssqlConnection { output_name })
                if output_name == "orders_output"
        ));
        assert_eq!(source_scans.load(Ordering::SeqCst), 0);
        assert_eq!(
            progress_events(&events),
            vec![
                (
                    ProgressEventKind::Started,
                    Some(ProgressOperation::DryRunAllToMssql),
                    None,
                ),
                (
                    ProgressEventKind::PhaseChanged,
                    None,
                    Some(ProgressPhase::PlanningOutput),
                ),
                (ProgressEventKind::Failed, None, None),
            ]
        );
        Ok(())
    }

    #[tokio::test]
    async fn dry_run_all_to_mssql_rejects_duplicate_output_names()
    -> Result<(), Box<dyn std::error::Error>> {
        let mut session = DeltaFunnelSession::new(
            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
        )?;
        let west = session.table_from_sql("select 1 as id").await?;
        let east = session.table_from_sql("select 2 as id").await?;
        let west = output_request(
            west,
            "orders_output",
            "west_orders",
            LoadMode::AppendExisting,
        )?;
        let east = output_request(
            east,
            "orders_output",
            "east_orders",
            LoadMode::AppendExisting,
        )?;

        let (reporter, events) = recording_reporter();

        let error = session.dry_run_all_to_mssql_with_progress(&[west, east], reporter);

        assert!(matches!(
            error,
            Err(DeltaFunnelError::MssqlWorkflowPlanning { message })
                if message.contains("write_all output names must be unique")
                    && message.contains("orders_output")
        ));
        assert!(progress_events(&events).is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn dry_run_to_mssql_rejects_execute_request_before_planning()
    -> Result<(), Box<dyn std::error::Error>> {
        let mut session = DeltaFunnelSession::new(
            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
        )?;
        let output = session.table_from_sql("select 1 as id").await?;
        let request = execute_output_request(
            output,
            "orders_output",
            "orders_sink",
            LoadMode::AppendExisting,
        )?;
        let (reporter, events) = recording_reporter();

        let error = session.dry_run_to_mssql_with_reporter(&request, reporter);

        assert!(matches!(
            error,
            Err(DeltaFunnelError::MssqlWorkflowPlanning { message })
                if message.contains("dry_run_to_mssql requires RunMode::DryRun")
        ));
        assert!(progress_events(&events).is_empty());
        Ok(())
    }

    #[test]
    fn dry_run_to_mssql_rejects_missing_connection_before_side_effects()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new("orders")?;
        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
        let request = output_request(
            source,
            "orders_output",
            "orders_sink",
            LoadMode::AppendExisting,
        )?;
        let (reporter, events) = recording_reporter();

        let error = session.dry_run_to_mssql_with_reporter(&request, reporter);

        assert!(matches!(
            error,
            Err(DeltaFunnelError::MissingMssqlConnection { output_name })
                if output_name == "orders_output"
        ));
        assert_eq!(
            progress_events(&events),
            vec![
                (
                    ProgressEventKind::Started,
                    Some(ProgressOperation::DryRunToMssql),
                    None,
                ),
                (
                    ProgressEventKind::PhaseChanged,
                    None,
                    Some(ProgressPhase::PlanningOutput),
                ),
                (ProgressEventKind::Failed, None, None),
            ]
        );
        Ok(())
    }

    #[test]
    fn dry_run_to_mssql_accepts_replace_before_side_effects()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new("orders")?;
        let mut session = DeltaFunnelSession::new(
            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
        )?;
        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
        let request = output_request(source, "orders_output", "orders_sink", LoadMode::Replace)?;

        let report = session.dry_run_to_mssql(&request)?;

        assert_eq!(report.output_name(), "orders_output");
        assert_eq!(report.load_mode(), LoadMode::Replace);
        assert_eq!(report.target_table().table(), "orders_sink");
        assert!(report.target_ddl_plan().create_table_sql_present());
        assert_eq!(report.target_ddl_plan().create_table_sql(), None);
        assert!(report.target_lifecycle_plan().create_table_sql_required());
        assert_eq!(
            report.target_lifecycle_plan().expected_target_state(),
            crate::MssqlTargetTableState::ExistsOrAbsent
        );
        assert!(!report.sql_server_contacted());
        assert!(!report.row_production_started());
        assert!(!report.table_lifecycle_started());
        assert!(!report.bulk_writer_started());
        Ok(())
    }

    #[test]
    fn host_callback_panic_after_started_escapes_without_an_unwind_delivery()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new("orders")?;
        let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
        let request = output_request(
            source,
            "orders_output",
            "orders_sink",
            LoadMode::AppendExisting,
        )?;
        let deliveries = Arc::new(AtomicUsize::new(0));
        let callback_deliveries = Arc::clone(&deliveries);
        let reporter = ProgressReporter::new(move |_| {
            if callback_deliveries.fetch_add(1, Ordering::SeqCst) == 1 {
                resume_unwind(Box::new("host callback panic"));
            }
        });

        let panic = catch_unwind(AssertUnwindSafe(|| {
            let _ = session.dry_run_to_mssql_with_reporter(&request, reporter);
        }));

        assert!(panic.is_err());
        assert_eq!(deliveries.load(Ordering::SeqCst), 2);
        Ok(())
    }

    #[test]
    fn interleaved_actions_keep_callbacks_serial_ordered_and_isolated()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new("interleaved-progress")?;
        let mut session = DeltaFunnelSession::new(
            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
        )?;
        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
        let west_request = output_request(
            source.clone(),
            "west_output",
            "west_sink",
            LoadMode::AppendExisting,
        )?;
        let east_request =
            output_request(source, "east_output", "east_sink", LoadMode::AppendExisting)?;
        let session = Arc::new(session);

        type ActionEvents = Arc<Mutex<Vec<(ProgressEventKind, Option<String>)>>>;
        let barrier = Arc::new(Barrier::new(2));
        let make_reporter = |events: ActionEvents, active: Arc<AtomicBool>| {
            let barrier = Arc::clone(&barrier);
            ProgressReporter::new(move |event| {
                assert!(!active.swap(true, Ordering::SeqCst));
                let mut events = match events.lock() {
                    Ok(events) => events,
                    Err(poisoned) => poisoned.into_inner(),
                };
                events.push((event.kind(), event.output_name().map(str::to_owned)));
                drop(events);
                if event.kind() == ProgressEventKind::Started {
                    barrier.wait();
                }
                assert!(active.swap(false, Ordering::SeqCst));
            })
        };
        let west_events = Arc::new(Mutex::new(Vec::new()));
        let east_events = Arc::new(Mutex::new(Vec::new()));
        let west_active = Arc::new(AtomicBool::new(false));
        let east_active = Arc::new(AtomicBool::new(false));
        let west_reporter = make_reporter(Arc::clone(&west_events), Arc::clone(&west_active));
        let east_reporter = make_reporter(Arc::clone(&east_events), Arc::clone(&east_active));
        let west_session = Arc::clone(&session);
        let east_session = Arc::clone(&session);

        let west_action = thread::spawn(move || {
            west_session.dry_run_to_mssql_with_reporter(&west_request, west_reporter)
        });
        let east_action = thread::spawn(move || {
            east_session.dry_run_to_mssql_with_reporter(&east_request, east_reporter)
        });

        west_action
            .join()
            .map_err(|_| std::io::Error::other("west progress action panicked"))??;
        east_action
            .join()
            .map_err(|_| std::io::Error::other("east progress action panicked"))??;

        assert!(!west_active.load(Ordering::SeqCst));
        assert!(!east_active.load(Ordering::SeqCst));
        let expected = |output_name: &str| {
            vec![
                (ProgressEventKind::Started, None),
                (
                    ProgressEventKind::PhaseChanged,
                    Some(output_name.to_owned()),
                ),
                (ProgressEventKind::Completed, None),
            ]
        };
        assert_eq!(
            west_events
                .lock()
                .map_err(|_| std::io::Error::other("west progress event mutex was poisoned"))?
                .as_slice(),
            expected("west_output")
        );
        assert_eq!(
            east_events
                .lock()
                .map_err(|_| std::io::Error::other("east progress event mutex was poisoned"))?
                .as_slice(),
            expected("east_output")
        );
        Ok(())
    }

    #[test]
    fn dry_run_to_mssql_report_debug_redacts_connection_material()
    -> Result<(), Box<dyn std::error::Error>> {
        let table = DeltaLogTable::new("orders")?;
        let mut session = DeltaFunnelSession::new(
            SessionOptions::new().with_default_mssql_connection(secret_connection()?),
        )?;
        let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
        let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders_sink")?)
            .with_connection(override_connection()?);
        let request = OutputWritePlan::new(
            source,
            MssqlOutputTarget::new("orders_output", target_config, RunMode::DryRun),
        );

        let report = session.dry_run_to_mssql(&request)?;
        let debug = format!("{report:?}");

        assert!(debug.contains("orders_output"));
        assert!(debug.contains("warehouse-override"));
        assert!(!debug.contains("secret-token"));
        assert!(!debug.contains("override-secret"));
        assert!(!debug.contains("password"));
        assert!(!debug.contains("server=tcp"));
        Ok(())
    }
}