delta_funnel/sql_server/execution/
workflow.rs

1//! Sequential multi-output SQL Server write orchestration.
2//!
3//! This module keeps the multi-output workflow layer separate from the
4//! one-output sink. The MVP runs outputs sequentially, stops on the first
5//! failure, and marks later outputs as skipped without invoking their lazy batch
6//! stream factories.
7
8use std::{fmt, future::Future, pin::Pin};
9
10use arrow_schema::SchemaRef;
11use async_trait::async_trait;
12use datafusion::arrow::record_batch::RecordBatch;
13use futures_util::Stream;
14use tracing::Instrument;
15
16use crate::{
17    DeltaFunnelError, PhaseTimingReport, ReportReasonCode, RowCount, ValidationOptions,
18    ValidationStatus, observability, report::PhaseTimer, support::sanitize_text_for_display,
19};
20
21use super::{
22    LoadMode, MssqlBatchShapingReport, MssqlConnectionSource, MssqlConnectionSummary,
23    MssqlSchemaPlanOptions, MssqlTargetSummary, MssqlTargetTable, MssqlWriteFailureContext,
24    MssqlWriteOptions, MssqlWriteReport, ResolvedMssqlTarget, default_mssql_write_options,
25    write_output_batches_to_mssql_with_validation_options,
26};
27
28const OUTPUT_STREAM_SETUP_PHASE: &str = "output_stream_setup";
29const SQL_WRITE_PHASE: &str = "sql_write";
30const VALIDATION_PHASE: &str = "validation";
31
32/// Lazy stream produced only when a SQL Server output is attempted.
33pub type MssqlOutputBatchStream =
34    Pin<Box<dyn Stream<Item = Result<RecordBatch, DeltaFunnelError>> + Send>>;
35
36/// Fallible future that constructs a direct batch stream for one attempted output.
37pub type MssqlOutputBatchStreamFuture =
38    Pin<Box<dyn Future<Output = Result<MssqlOutputBatchStream, DeltaFunnelError>> + Send>>;
39
40/// Async factory that constructs a direct batch stream for one attempted output.
41pub type MssqlOutputBatchStreamFactory = Box<dyn FnOnce() -> MssqlOutputBatchStreamFuture + Send>;
42
43/// One deferred SQL Server output write job.
44///
45/// The job owns an already resolved SQL Server target plus a lazy batch stream
46/// factory. The workflow awaits the factory only after the output becomes the
47/// next attempted output. Skipped jobs keep their stream factories uncalled, so
48/// skipped outputs do not start source reads, DataFusion execution, stream
49/// setup, SQL connections, lifecycle preparation, writer initialization, or
50/// batch polling through this API.
51pub struct MssqlOutputWriteJob {
52    output_schema: SchemaRef,
53    resolved_target: ResolvedMssqlTarget,
54    schema_options: MssqlSchemaPlanOptions,
55    batches: MssqlOutputBatchStreamFactory,
56    write_options: MssqlWriteOptions,
57    validation_options: ValidationOptions,
58    phase_timings: Vec<PhaseTimingReport>,
59}
60
61impl MssqlOutputWriteJob {
62    /// Creates a deferred SQL Server output write job.
63    pub fn new<F, Fut, S>(
64        output_schema: SchemaRef,
65        resolved_target: ResolvedMssqlTarget,
66        schema_options: MssqlSchemaPlanOptions,
67        batches: F,
68        write_options: MssqlWriteOptions,
69        validation_options: ValidationOptions,
70    ) -> Self
71    where
72        F: FnOnce() -> Fut + Send + 'static,
73        Fut: Future<Output = Result<S, DeltaFunnelError>> + Send + 'static,
74        S: Stream<Item = Result<RecordBatch, DeltaFunnelError>> + Send + 'static,
75    {
76        Self {
77            output_schema,
78            resolved_target,
79            schema_options,
80            batches: Box::new(move || {
81                Box::pin(async move {
82                    let stream = batches().await?;
83                    Ok(Box::pin(stream) as MssqlOutputBatchStream)
84                })
85            }),
86            write_options,
87            validation_options,
88            phase_timings: Vec::new(),
89        }
90    }
91
92    /// Adds phase timings that completed before this deferred write job runs.
93    #[must_use]
94    pub fn with_phase_timings(mut self, phase_timings: Vec<PhaseTimingReport>) -> Self {
95        self.phase_timings = phase_timings;
96        self
97    }
98
99    /// Creates a deferred SQL Server output write job using default write options.
100    pub fn with_default_write_options<F, Fut, S>(
101        output_schema: SchemaRef,
102        resolved_target: ResolvedMssqlTarget,
103        schema_options: MssqlSchemaPlanOptions,
104        batches: F,
105    ) -> Self
106    where
107        F: FnOnce() -> Fut + Send + 'static,
108        Fut: Future<Output = Result<S, DeltaFunnelError>> + Send + 'static,
109        S: Stream<Item = Result<RecordBatch, DeltaFunnelError>> + Send + 'static,
110    {
111        Self::new(
112            output_schema,
113            resolved_target,
114            schema_options,
115            batches,
116            default_mssql_write_options(),
117            ValidationOptions::default(),
118        )
119    }
120
121    /// Returns the selected output name.
122    #[must_use]
123    pub fn output_name(&self) -> &str {
124        self.resolved_target.output_name()
125    }
126
127    /// Returns a redacted target summary for reports.
128    #[must_use]
129    pub fn target_summary(&self) -> MssqlTargetSummary {
130        self.resolved_target.summary()
131    }
132
133    /// Returns phase timings that completed before this deferred write job runs.
134    #[must_use]
135    pub fn phase_timings(&self) -> &[PhaseTimingReport] {
136        &self.phase_timings
137    }
138
139    fn into_parts(
140        self,
141    ) -> (
142        SchemaRef,
143        ResolvedMssqlTarget,
144        MssqlSchemaPlanOptions,
145        MssqlOutputBatchStreamFactory,
146        MssqlWriteOptions,
147        ValidationOptions,
148        Vec<PhaseTimingReport>,
149    ) {
150        (
151            self.output_schema,
152            self.resolved_target,
153            self.schema_options,
154            self.batches,
155            self.write_options,
156            self.validation_options,
157            self.phase_timings,
158        )
159    }
160}
161
162/// SQL Server multi-output workflow options.
163#[derive(Debug, Clone, Copy, PartialEq, Eq)]
164pub struct MssqlWorkflowWriteOptions {
165    max_parallel_outputs: usize,
166}
167
168impl Default for MssqlWorkflowWriteOptions {
169    fn default() -> Self {
170        Self {
171            max_parallel_outputs: 1,
172        }
173    }
174}
175
176impl MssqlWorkflowWriteOptions {
177    /// Creates default sequential workflow options.
178    #[must_use]
179    pub const fn new() -> Self {
180        Self {
181            max_parallel_outputs: 1,
182        }
183    }
184
185    /// Sets the requested maximum number of parallel output writers.
186    ///
187    /// The current MVP supports only `1`. Values greater than `1` are rejected
188    /// explicitly so callers do not mistake the workflow for a parallel writer
189    /// pool or cross-output transaction boundary.
190    #[must_use]
191    pub const fn with_max_parallel_outputs(mut self, max_parallel_outputs: usize) -> Self {
192        self.max_parallel_outputs = max_parallel_outputs;
193        self
194    }
195
196    /// Returns the requested maximum number of parallel output writers.
197    #[must_use]
198    pub const fn max_parallel_outputs(&self) -> usize {
199        self.max_parallel_outputs
200    }
201
202    /// Validates workflow write options before any output write side effects.
203    ///
204    /// # Errors
205    ///
206    /// Returns [`DeltaFunnelError::MssqlWorkflowPlanning`] when no output
207    /// writer is allowed or when parallel output writers are requested. The
208    /// current MVP is intentionally single-writer so callers cannot mistake
209    /// this workflow for a parallel writer pool or cross-output transaction.
210    pub fn validate(&self) -> Result<(), DeltaFunnelError> {
211        match self.max_parallel_outputs() {
212            1 => Ok(()),
213            0 => Err(DeltaFunnelError::MssqlWorkflowPlanning {
214                message: "max_parallel_outputs must be at least 1".to_owned(),
215            }),
216            max_parallel_outputs => Err(DeltaFunnelError::MssqlWorkflowPlanning {
217                message: format!(
218                    "parallel MSSQL output writers are not supported; requested {max_parallel_outputs}"
219                ),
220            }),
221        }
222    }
223}
224
225/// Structured report for a multi-output SQL Server write workflow.
226#[derive(Debug, Clone, PartialEq, Eq)]
227pub struct MssqlWorkflowWriteReport {
228    outputs: Vec<MssqlOutputWriteStatus>,
229}
230
231impl MssqlWorkflowWriteReport {
232    fn new(outputs: Vec<MssqlOutputWriteStatus>) -> Self {
233        Self { outputs }
234    }
235
236    /// Returns the number of selected outputs represented by this report.
237    #[must_use]
238    pub fn len(&self) -> usize {
239        self.outputs.len()
240    }
241
242    /// Returns whether this report contains no selected outputs.
243    #[must_use]
244    pub fn is_empty(&self) -> bool {
245        self.outputs.is_empty()
246    }
247
248    /// Returns per-output statuses in caller-provided order.
249    #[must_use]
250    pub fn outputs(&self) -> &[MssqlOutputWriteStatus] {
251        &self.outputs
252    }
253
254    /// Returns whether every selected output completed successfully.
255    #[must_use]
256    pub fn all_succeeded(&self) -> bool {
257        self.outputs
258            .iter()
259            .all(MssqlOutputWriteStatus::is_succeeded)
260    }
261
262    /// Returns the number of outputs that completed successfully.
263    #[must_use]
264    pub fn succeeded_count(&self) -> usize {
265        self.outputs
266            .iter()
267            .filter(|status| status.is_succeeded())
268            .count()
269    }
270
271    /// Returns the number of outputs that failed.
272    #[must_use]
273    pub fn failed_count(&self) -> usize {
274        self.outputs
275            .iter()
276            .filter(|status| status.is_failed())
277            .count()
278    }
279
280    /// Returns the number of outputs skipped after a previous output failed.
281    #[must_use]
282    pub fn skipped_count(&self) -> usize {
283        self.outputs
284            .iter()
285            .filter(|status| status.is_skipped())
286            .count()
287    }
288}
289
290impl fmt::Display for MssqlWorkflowWriteReport {
291    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
292        let succeeded = self.succeeded_count();
293        let failed = self.failed_count();
294        let skipped = self.skipped_count();
295
296        write!(
297            formatter,
298            "MSSQL workflow write report: {succeeded} succeeded, {failed} failed, {skipped} skipped"
299        )
300    }
301}
302
303/// Final write status for one selected SQL Server output.
304#[derive(Debug, Clone, PartialEq, Eq)]
305pub enum MssqlOutputWriteStatus {
306    /// The output completed successfully.
307    Succeeded(MssqlWriteReport),
308    /// The output was the first attempted output to fail.
309    Failed(MssqlWriteFailureReport),
310    /// The output was not attempted because an earlier output failed.
311    Skipped(MssqlWriteSkippedReport),
312}
313
314impl MssqlOutputWriteStatus {
315    /// Returns the selected output name for this status.
316    #[must_use]
317    pub fn output_name(&self) -> &str {
318        match self {
319            Self::Succeeded(report) => report.output_name(),
320            Self::Failed(report) => report.output_name(),
321            Self::Skipped(report) => report.output_name(),
322        }
323    }
324
325    /// Returns the effective SQL Server target table for this status.
326    #[must_use]
327    pub fn target_table(&self) -> &MssqlTargetTable {
328        match self {
329            Self::Succeeded(report) => report.target_table(),
330            Self::Failed(report) => report.target().table(),
331            Self::Skipped(report) => report.target().table(),
332        }
333    }
334
335    /// Returns the requested lifecycle mode for this output.
336    #[must_use]
337    pub fn load_mode(&self) -> LoadMode {
338        match self {
339            Self::Succeeded(report) => report.load_mode(),
340            Self::Failed(report) => report.target().load_mode(),
341            Self::Skipped(report) => report.target().load_mode(),
342        }
343    }
344
345    /// Returns where the effective connection came from.
346    #[must_use]
347    pub fn connection_source(&self) -> MssqlConnectionSource {
348        match self {
349            Self::Succeeded(report) => report.connection_source(),
350            Self::Failed(report) => report.target().connection_source(),
351            Self::Skipped(report) => report.target().connection_source(),
352        }
353    }
354
355    /// Returns the redacted effective connection summary.
356    #[must_use]
357    pub fn connection(&self) -> &MssqlConnectionSummary {
358        match self {
359            Self::Succeeded(report) => report.connection(),
360            Self::Failed(report) => report.target().connection(),
361            Self::Skipped(report) => report.target().connection(),
362        }
363    }
364
365    /// Returns query output row evidence for this output.
366    #[must_use]
367    pub fn output_row_count(&self) -> RowCount {
368        match self {
369            Self::Succeeded(report) => report.output_row_count(),
370            Self::Failed(report) => report.output_row_count(),
371            Self::Skipped(report) => report.output_row_count(),
372        }
373    }
374
375    /// Returns target-side row count evidence for this output.
376    #[must_use]
377    pub fn target_row_count(&self) -> RowCount {
378        match self {
379            Self::Succeeded(report) => report.target_row_count(),
380            Self::Failed(report) => report.target_row_count(),
381            Self::Skipped(report) => report.target_row_count(),
382        }
383    }
384
385    /// Returns target-side validation status for this output.
386    #[must_use]
387    pub fn validation_status(&self) -> ValidationStatus {
388        match self {
389            Self::Succeeded(report) => report.validation_status(),
390            Self::Failed(report) => report.validation_status(),
391            Self::Skipped(report) => report.validation_status(),
392        }
393    }
394
395    /// Returns batch-shaping counters for this output.
396    #[must_use]
397    pub fn batch_shaping(&self) -> MssqlBatchShapingReport {
398        match self {
399            Self::Succeeded(report) => report.batch_shaping(),
400            Self::Failed(report) => report.batch_shaping(),
401            Self::Skipped(report) => report.batch_shaping(),
402        }
403    }
404
405    /// Returns workflow phase timing reports for this output when available.
406    #[must_use]
407    pub fn phase_timings(&self) -> &[PhaseTimingReport] {
408        match self {
409            Self::Succeeded(report) => report.phase_timings(),
410            Self::Failed(report) => report.phase_timings(),
411            Self::Skipped(report) => report.phase_timings(),
412        }
413    }
414
415    /// Returns whether this output succeeded.
416    #[must_use]
417    pub const fn is_succeeded(&self) -> bool {
418        matches!(self, Self::Succeeded(_))
419    }
420
421    /// Returns whether this output failed.
422    #[must_use]
423    pub const fn is_failed(&self) -> bool {
424        matches!(self, Self::Failed(_))
425    }
426
427    /// Returns whether this output was skipped before any work was attempted.
428    #[must_use]
429    pub const fn is_skipped(&self) -> bool {
430        matches!(self, Self::Skipped(_))
431    }
432}
433
434/// Structured report for the first failed SQL Server output.
435#[derive(Debug, Clone, PartialEq, Eq)]
436pub struct MssqlWriteFailureReport {
437    target: MssqlTargetSummary,
438    error: String,
439    context: Option<Box<MssqlWriteFailureContext>>,
440    output_row_count: RowCount,
441    target_row_count: RowCount,
442    validation_status: ValidationStatus,
443    batch_shaping: MssqlBatchShapingReport,
444    phase_timings: Vec<PhaseTimingReport>,
445}
446
447impl MssqlWriteFailureReport {
448    fn from_error(
449        target: MssqlTargetSummary,
450        error: DeltaFunnelError,
451        phase_timings: Vec<PhaseTimingReport>,
452    ) -> Self {
453        let context = failure_context(&error).cloned().map(Box::new);
454        let phase_timings = merged_failure_phase_timings(phase_timings, context.as_deref());
455        let output_row_count = context.as_deref().map_or(
456            RowCount::unavailable(),
457            MssqlWriteFailureContext::output_row_count,
458        );
459        let target_row_count = context.as_deref().map_or(
460            RowCount::unavailable(),
461            MssqlWriteFailureContext::target_row_count,
462        );
463        let validation_status = context.as_deref().map_or(
464            ValidationStatus::skipped(ReportReasonCode::FailureBeforeValidation),
465            MssqlWriteFailureContext::validation_status,
466        );
467        let batch_shaping = context.as_deref().map_or_else(
468            || MssqlBatchShapingReport::not_started(ReportReasonCode::NotExecuted),
469            MssqlWriteFailureContext::batch_shaping,
470        );
471        Self {
472            target,
473            error: sanitize_text_for_display(&error.to_string()),
474            context,
475            output_row_count,
476            target_row_count,
477            validation_status,
478            batch_shaping,
479            phase_timings,
480        }
481    }
482
483    /// Returns the redacted target summary for the failed output.
484    #[must_use]
485    pub const fn target(&self) -> &MssqlTargetSummary {
486        &self.target
487    }
488
489    /// Returns the selected output name.
490    #[must_use]
491    pub fn output_name(&self) -> &str {
492        self.target.output_name()
493    }
494
495    /// Returns the sanitized error message for the failed output.
496    #[must_use]
497    pub fn error(&self) -> &str {
498        &self.error
499    }
500
501    /// Returns phase-aware write failure context when the one-output sink
502    /// provided it.
503    #[must_use]
504    pub fn context(&self) -> Option<&MssqlWriteFailureContext> {
505        self.context.as_deref()
506    }
507
508    /// Returns query output row evidence known at failure time.
509    #[must_use]
510    pub const fn output_row_count(&self) -> RowCount {
511        self.output_row_count
512    }
513
514    /// Returns target-side row count evidence known at failure time.
515    #[must_use]
516    pub const fn target_row_count(&self) -> RowCount {
517        self.target_row_count
518    }
519
520    /// Returns target-side validation status known at failure time.
521    #[must_use]
522    pub const fn validation_status(&self) -> ValidationStatus {
523        self.validation_status
524    }
525
526    /// Returns batch-shaping counters known at failure time.
527    #[must_use]
528    pub const fn batch_shaping(&self) -> MssqlBatchShapingReport {
529        self.batch_shaping
530    }
531
532    /// Returns workflow phase timing reports for this failed output.
533    #[must_use]
534    pub fn phase_timings(&self) -> &[PhaseTimingReport] {
535        &self.phase_timings
536    }
537}
538
539fn merged_failure_phase_timings(
540    mut phase_timings: Vec<PhaseTimingReport>,
541    context: Option<&MssqlWriteFailureContext>,
542) -> Vec<PhaseTimingReport> {
543    let Some(context) = context else {
544        return phase_timings;
545    };
546
547    for timing in context.phase_timings() {
548        if let Some(existing) = phase_timings
549            .iter_mut()
550            .find(|existing| existing.phase_name() == timing.phase_name())
551        {
552            *existing = timing.clone();
553        } else {
554            phase_timings.push(timing.clone());
555        }
556    }
557
558    phase_timings
559}
560
561/// Structured report for a skipped SQL Server output.
562#[derive(Debug, Clone, PartialEq, Eq)]
563pub struct MssqlWriteSkippedReport {
564    target: MssqlTargetSummary,
565    reason: MssqlWriteSkippedReason,
566    output_row_count: RowCount,
567    target_row_count: RowCount,
568    validation_status: ValidationStatus,
569    batch_shaping: MssqlBatchShapingReport,
570    phase_timings: Vec<PhaseTimingReport>,
571}
572
573impl MssqlWriteSkippedReport {
574    fn previous_output_failed(
575        target: MssqlTargetSummary,
576        failed_output_name: String,
577        phase_timings: Vec<PhaseTimingReport>,
578    ) -> Self {
579        Self {
580            target,
581            reason: MssqlWriteSkippedReason::PreviousOutputFailed { failed_output_name },
582            output_row_count: RowCount::unavailable(),
583            target_row_count: RowCount::unavailable(),
584            validation_status: ValidationStatus::skipped(ReportReasonCode::PriorFailure),
585            batch_shaping: MssqlBatchShapingReport::skipped(ReportReasonCode::PriorFailure),
586            phase_timings: skipped_after_prior_failure_phase_timings(phase_timings),
587        }
588    }
589
590    /// Returns the redacted target summary for the skipped output.
591    #[must_use]
592    pub const fn target(&self) -> &MssqlTargetSummary {
593        &self.target
594    }
595
596    /// Returns the selected output name.
597    #[must_use]
598    pub fn output_name(&self) -> &str {
599        self.target.output_name()
600    }
601
602    /// Returns why this output was skipped before any work was attempted.
603    #[must_use]
604    pub const fn reason(&self) -> &MssqlWriteSkippedReason {
605        &self.reason
606    }
607
608    /// Returns query output row evidence for this skipped output.
609    #[must_use]
610    pub const fn output_row_count(&self) -> RowCount {
611        self.output_row_count
612    }
613
614    /// Returns target-side row count evidence for this skipped output.
615    #[must_use]
616    pub const fn target_row_count(&self) -> RowCount {
617        self.target_row_count
618    }
619
620    /// Returns target-side validation status for this skipped output.
621    #[must_use]
622    pub const fn validation_status(&self) -> ValidationStatus {
623        self.validation_status
624    }
625
626    /// Returns batch-shaping counters for this skipped output.
627    #[must_use]
628    pub const fn batch_shaping(&self) -> MssqlBatchShapingReport {
629        self.batch_shaping
630    }
631
632    /// Returns workflow phase timing reports for this skipped output.
633    #[must_use]
634    pub fn phase_timings(&self) -> &[PhaseTimingReport] {
635        &self.phase_timings
636    }
637}
638
639/// Reason one selected SQL Server output was skipped.
640#[derive(Debug, Clone, PartialEq, Eq)]
641pub enum MssqlWriteSkippedReason {
642    /// A previous output failed and the MVP stops on first failure.
643    PreviousOutputFailed {
644        /// Output name of the first failed output.
645        failed_output_name: String,
646    },
647}
648
649/// Writes multiple SQL Server outputs sequentially.
650///
651/// This workflow calls the public one-output sink once per attempted output,
652/// in caller-provided order. It stops on the first failed output and marks all
653/// later outputs as skipped without invoking their lazy batch stream factories.
654/// The report is per-output and does not imply all-or-nothing transaction
655/// behavior across outputs, target tables, or SQL Server connections.
656pub async fn write_mssql_outputs_to_mssql(
657    jobs: impl IntoIterator<Item = MssqlOutputWriteJob>,
658    options: MssqlWorkflowWriteOptions,
659) -> Result<MssqlWorkflowWriteReport, DeltaFunnelError> {
660    write_mssql_outputs_with_writer(jobs, options, MssqlPublicOneOutputWriter).await
661}
662
663#[async_trait]
664pub(crate) trait MssqlWorkflowOutputWriter: Send {
665    async fn write_output(
666        &mut self,
667        output_schema: SchemaRef,
668        resolved_target: ResolvedMssqlTarget,
669        schema_options: MssqlSchemaPlanOptions,
670        batches: MssqlOutputBatchStream,
671        write_options: MssqlWriteOptions,
672        validation_options: ValidationOptions,
673    ) -> Result<MssqlWriteReport, DeltaFunnelError>;
674}
675
676struct MssqlPublicOneOutputWriter;
677
678#[async_trait]
679impl MssqlWorkflowOutputWriter for MssqlPublicOneOutputWriter {
680    async fn write_output(
681        &mut self,
682        output_schema: SchemaRef,
683        resolved_target: ResolvedMssqlTarget,
684        schema_options: MssqlSchemaPlanOptions,
685        batches: MssqlOutputBatchStream,
686        write_options: MssqlWriteOptions,
687        validation_options: ValidationOptions,
688    ) -> Result<MssqlWriteReport, DeltaFunnelError> {
689        write_output_batches_to_mssql_with_validation_options(
690            output_schema.as_ref(),
691            resolved_target,
692            schema_options,
693            batches,
694            write_options,
695            validation_options,
696        )
697        .await
698    }
699}
700
701pub(crate) async fn write_mssql_outputs_with_writer<W>(
702    jobs: impl IntoIterator<Item = MssqlOutputWriteJob>,
703    options: MssqlWorkflowWriteOptions,
704    mut writer: W,
705) -> Result<MssqlWorkflowWriteReport, DeltaFunnelError>
706where
707    W: MssqlWorkflowOutputWriter,
708{
709    ensure_sequential_options(options)?;
710
711    let mut statuses = Vec::new();
712    let mut failed_output_name = None::<String>;
713
714    for job in jobs {
715        if let Some(failed_output_name) = failed_output_name.as_ref() {
716            statuses.push(skipped_output_status_with_tracing(
717                job.target_summary(),
718                failed_output_name.clone(),
719                job.phase_timings().to_vec(),
720            ));
721            continue;
722        }
723
724        let status = write_mssql_output_job_with_tracing(job, &mut writer).await;
725        if let MssqlOutputWriteStatus::Failed(failure) = &status {
726            failed_output_name = Some(failure.output_name().to_owned());
727        }
728        statuses.push(status);
729    }
730
731    Ok(MssqlWorkflowWriteReport::new(statuses))
732}
733
734fn skipped_output_status_with_tracing(
735    target: MssqlTargetSummary,
736    failed_output_name: String,
737    planned_phase_timings: Vec<PhaseTimingReport>,
738) -> MssqlOutputWriteStatus {
739    let output_span =
740        observability::output_span(target.output_name(), target.table(), target.load_mode());
741    output_span.in_scope(|| {
742        let skipped = skipped_output_status(target, failed_output_name, planned_phase_timings);
743        observability::output_skipped(
744            skipped.output_name(),
745            skipped.target().table(),
746            skipped.target().load_mode(),
747            "prior_failure",
748        );
749        MssqlOutputWriteStatus::Skipped(skipped)
750    })
751}
752
753fn skipped_output_status(
754    target: MssqlTargetSummary,
755    failed_output_name: String,
756    planned_phase_timings: Vec<PhaseTimingReport>,
757) -> MssqlWriteSkippedReport {
758    MssqlWriteSkippedReport::previous_output_failed(
759        target,
760        failed_output_name,
761        planned_phase_timings,
762    )
763}
764
765async fn write_mssql_output_job_with_tracing<W>(
766    job: MssqlOutputWriteJob,
767    writer: &mut W,
768) -> MssqlOutputWriteStatus
769where
770    W: MssqlWorkflowOutputWriter,
771{
772    let target = job.target_summary();
773    let output_span =
774        observability::output_span(target.output_name(), target.table(), target.load_mode());
775
776    async move {
777        observability::output_started(target.output_name(), target.table(), target.load_mode());
778        let status = write_mssql_output_job(job, writer).await;
779        match &status {
780            MssqlOutputWriteStatus::Succeeded(report) => {
781                observability::output_completed(
782                    report.output_name(),
783                    report.target_table(),
784                    report.load_mode(),
785                );
786            }
787            MssqlOutputWriteStatus::Failed(failure) => {
788                observability::output_failed(
789                    failure.output_name(),
790                    failure.target().table(),
791                    failure.target().load_mode(),
792                    failure.error(),
793                );
794            }
795            MssqlOutputWriteStatus::Skipped(_) => {}
796        }
797        status
798    }
799    .instrument(output_span)
800    .await
801}
802
803async fn write_mssql_output_job<W>(
804    job: MssqlOutputWriteJob,
805    writer: &mut W,
806) -> MssqlOutputWriteStatus
807where
808    W: MssqlWorkflowOutputWriter,
809{
810    let target = job.target_summary();
811    let (
812        output_schema,
813        resolved_target,
814        schema_options,
815        batches,
816        write_options,
817        validation_options,
818        planned_phase_timings,
819    ) = job.into_parts();
820    let stream_setup_timer = PhaseTimer::start(OUTPUT_STREAM_SETUP_PHASE);
821    let (batches, stream_setup_timing) = match batches().await {
822        Ok(batches) => (batches, stream_setup_timer.completed()),
823        Err(error) => {
824            let failure = MssqlWriteFailureReport::from_error(
825                target,
826                error,
827                stream_setup_failure_phase_timings(
828                    planned_phase_timings,
829                    stream_setup_timer.failed(),
830                ),
831            );
832            return MssqlOutputWriteStatus::Failed(failure);
833        }
834    };
835
836    let write_timer = PhaseTimer::start(SQL_WRITE_PHASE);
837    match writer
838        .write_output(
839            output_schema,
840            resolved_target,
841            schema_options,
842            batches,
843            write_options,
844            validation_options,
845        )
846        .await
847    {
848        Ok(report) => {
849            let report = report.with_phase_timings(output_write_phase_timings(
850                planned_phase_timings,
851                stream_setup_timing,
852                write_timer.completed(),
853            ));
854            MssqlOutputWriteStatus::Succeeded(report)
855        }
856        Err(error) => {
857            let failure = MssqlWriteFailureReport::from_error(
858                target,
859                error,
860                output_write_failure_phase_timings(
861                    planned_phase_timings,
862                    stream_setup_timing,
863                    write_timer.failed(),
864                ),
865            );
866            MssqlOutputWriteStatus::Failed(failure)
867        }
868    }
869}
870
871fn ensure_sequential_options(options: MssqlWorkflowWriteOptions) -> Result<(), DeltaFunnelError> {
872    options.validate()
873}
874
875fn failure_context(error: &DeltaFunnelError) -> Option<&MssqlWriteFailureContext> {
876    match error {
877        DeltaFunnelError::MssqlWritePhase { context, .. }
878        | DeltaFunnelError::MssqlBatchSchemaValidation { context, .. } => Some(context.as_ref()),
879        _ => None,
880    }
881}
882
883fn output_write_phase_timings(
884    mut phase_timings: Vec<PhaseTimingReport>,
885    stream_setup_timing: PhaseTimingReport,
886    write_timing: PhaseTimingReport,
887) -> Vec<PhaseTimingReport> {
888    phase_timings.extend([stream_setup_timing, write_timing]);
889    phase_timings
890}
891
892fn output_write_failure_phase_timings(
893    mut phase_timings: Vec<PhaseTimingReport>,
894    stream_setup_timing: PhaseTimingReport,
895    write_timing: PhaseTimingReport,
896) -> Vec<PhaseTimingReport> {
897    phase_timings.extend([
898        stream_setup_timing,
899        write_timing,
900        PhaseTimingReport::not_started(VALIDATION_PHASE, ReportReasonCode::FailureBeforeValidation),
901    ]);
902    phase_timings
903}
904
905fn stream_setup_failure_phase_timings(
906    mut phase_timings: Vec<PhaseTimingReport>,
907    stream_setup_timing: PhaseTimingReport,
908) -> Vec<PhaseTimingReport> {
909    phase_timings.extend([
910        stream_setup_timing,
911        PhaseTimingReport::not_started(SQL_WRITE_PHASE, ReportReasonCode::NotExecuted),
912        PhaseTimingReport::not_started(VALIDATION_PHASE, ReportReasonCode::FailureBeforeValidation),
913    ]);
914    phase_timings
915}
916
917fn skipped_after_prior_failure_phase_timings(
918    mut phase_timings: Vec<PhaseTimingReport>,
919) -> Vec<PhaseTimingReport> {
920    phase_timings.extend([
921        PhaseTimingReport::skipped(OUTPUT_STREAM_SETUP_PHASE, ReportReasonCode::PriorFailure),
922        PhaseTimingReport::skipped(SQL_WRITE_PHASE, ReportReasonCode::PriorFailure),
923        PhaseTimingReport::skipped(VALIDATION_PHASE, ReportReasonCode::PriorFailure),
924    ]);
925    phase_timings
926}
927
928#[cfg(test)]
929mod tests {
930    use std::collections::VecDeque;
931    use std::sync::{Arc, Mutex, MutexGuard};
932    use std::time::Duration;
933
934    use arrow_schema::{DataType, Field, Schema};
935    use async_trait::async_trait;
936    use futures_util::{StreamExt, stream};
937
938    use super::*;
939    use crate::{
940        LoadMode, MssqlConnectionConfig, MssqlTargetCleanupStatus, MssqlTargetConfig,
941        MssqlTargetOutputPlan, MssqlTargetResolutionContext, MssqlTargetTable, MssqlWritePhase,
942        PhaseStatus, PhaseTimingReport, ValidationStatus, plan_mssql_target_for_output,
943        report::sql_server::MssqlWriteReportMetrics,
944    };
945
946    const PLANNED_PHASE: &str = "planned_phase";
947
948    #[derive(Default)]
949    struct FakeWorkflowWriter {
950        outcomes: VecDeque<Result<MssqlWriteReport, DeltaFunnelError>>,
951        attempted_outputs: Arc<Mutex<Vec<String>>>,
952    }
953
954    #[derive(Default)]
955    struct StreamPollingWorkflowWriter {
956        attempted_outputs: Arc<Mutex<Vec<String>>>,
957    }
958
959    impl FakeWorkflowWriter {
960        fn new(outcomes: Vec<Result<MssqlWriteReport, DeltaFunnelError>>) -> Self {
961            Self {
962                outcomes: outcomes.into(),
963                attempted_outputs: Arc::new(Mutex::new(Vec::new())),
964            }
965        }
966
967        fn attempted_outputs(&self) -> Arc<Mutex<Vec<String>>> {
968            Arc::clone(&self.attempted_outputs)
969        }
970    }
971
972    impl StreamPollingWorkflowWriter {
973        fn attempted_outputs(&self) -> Arc<Mutex<Vec<String>>> {
974            Arc::clone(&self.attempted_outputs)
975        }
976    }
977
978    #[async_trait]
979    impl MssqlWorkflowOutputWriter for FakeWorkflowWriter {
980        async fn write_output(
981            &mut self,
982            _output_schema: SchemaRef,
983            resolved_target: ResolvedMssqlTarget,
984            _schema_options: MssqlSchemaPlanOptions,
985            _batches: MssqlOutputBatchStream,
986            _write_options: MssqlWriteOptions,
987            _validation_options: ValidationOptions,
988        ) -> Result<MssqlWriteReport, DeltaFunnelError> {
989            self.attempted_outputs
990                .lock()
991                .map_err(|_| test_error("attempted output lock poisoned"))?
992                .push(resolved_target.output_name().to_owned());
993
994            self.outcomes
995                .pop_front()
996                .ok_or_else(|| test_error("missing fake writer outcome"))?
997        }
998    }
999
1000    #[async_trait]
1001    impl MssqlWorkflowOutputWriter for StreamPollingWorkflowWriter {
1002        async fn write_output(
1003            &mut self,
1004            _output_schema: SchemaRef,
1005            resolved_target: ResolvedMssqlTarget,
1006            _schema_options: MssqlSchemaPlanOptions,
1007            mut batches: MssqlOutputBatchStream,
1008            _write_options: MssqlWriteOptions,
1009            _validation_options: ValidationOptions,
1010        ) -> Result<MssqlWriteReport, DeltaFunnelError> {
1011            self.attempted_outputs
1012                .lock()
1013                .map_err(|_| test_error("attempted output lock poisoned"))?
1014                .push(resolved_target.output_name().to_owned());
1015
1016            match batches.next().await {
1017                Some(Ok(_batch)) => Err(test_error("expected stream polling error")),
1018                Some(Err(error)) => Err(error),
1019                None => Err(test_error("expected at least one stream item")),
1020            }
1021        }
1022    }
1023
1024    #[tokio::test]
1025    async fn empty_workflow_report_has_zero_counts() -> Result<(), DeltaFunnelError> {
1026        let writer = FakeWorkflowWriter::default();
1027
1028        let report = write_mssql_outputs_with_writer(
1029            Vec::new(),
1030            MssqlWorkflowWriteOptions::default(),
1031            writer,
1032        )
1033        .await?;
1034
1035        assert!(report.is_empty());
1036        assert_eq!(report.len(), 0);
1037        assert_eq!(report.outputs(), []);
1038        assert!(report.all_succeeded());
1039        assert_eq!(report.succeeded_count(), 0);
1040        assert_eq!(report.failed_count(), 0);
1041        assert_eq!(report.skipped_count(), 0);
1042        assert_eq!(
1043            report.to_string(),
1044            "MSSQL workflow write report: 0 succeeded, 0 failed, 0 skipped"
1045        );
1046
1047        Ok(())
1048    }
1049
1050    #[tokio::test]
1051    async fn two_successful_outputs_produce_two_success_statuses() -> Result<(), DeltaFunnelError> {
1052        let first = output_plan("first", LoadMode::AppendExisting)?;
1053        let second = output_plan("second", LoadMode::AppendExisting)?;
1054        let first_report =
1055            write_report(&first, 2, 1, false, MssqlTargetCleanupStatus::NotApplicable);
1056        let second_report = write_report(
1057            &second,
1058            3,
1059            2,
1060            false,
1061            MssqlTargetCleanupStatus::NotApplicable,
1062        );
1063        let writer = FakeWorkflowWriter::new(vec![Ok(first_report), Ok(second_report)]);
1064        let attempted = writer.attempted_outputs();
1065
1066        let report = write_mssql_outputs_with_writer(
1067            vec![job(first)?, job(second)?],
1068            MssqlWorkflowWriteOptions::default(),
1069            writer,
1070        )
1071        .await?;
1072
1073        assert_eq!(report.outputs().len(), 2);
1074        assert!(report.all_succeeded());
1075        assert_status_output(report.outputs(), 0, "first")?;
1076        assert_status_output(report.outputs(), 1, "second")?;
1077        assert_eq!(report.outputs()[0].output_row_count(), RowCount::exact(2));
1078        assert_batch_shaping(
1079            report.outputs()[0].batch_shaping(),
1080            PhaseStatus::completed(),
1081            1,
1082            2,
1083            1,
1084            2,
1085        );
1086        assert_eq!(report.outputs()[1].output_row_count(), RowCount::exact(3));
1087        assert_batch_shaping(
1088            report.outputs()[1].batch_shaping(),
1089            PhaseStatus::completed(),
1090            2,
1091            3,
1092            2,
1093            3,
1094        );
1095        assert_phase_timing(
1096            &report.outputs()[0],
1097            PLANNED_PHASE,
1098            PhaseStatus::completed(),
1099        )?;
1100        assert_phase_timing(
1101            &report.outputs()[0],
1102            OUTPUT_STREAM_SETUP_PHASE,
1103            PhaseStatus::completed(),
1104        )?;
1105        assert_phase_timing(
1106            &report.outputs()[0],
1107            SQL_WRITE_PHASE,
1108            PhaseStatus::completed(),
1109        )?;
1110        assert_eq!(
1111            locked(&attempted)?.as_slice(),
1112            ["first".to_owned(), "second".to_owned()]
1113        );
1114
1115        Ok(())
1116    }
1117
1118    #[tokio::test]
1119    async fn first_success_remains_successful_when_second_output_fails()
1120    -> Result<(), DeltaFunnelError> {
1121        let first = output_plan("first", LoadMode::AppendExisting)?;
1122        let second = output_plan("second", LoadMode::AppendExisting)?;
1123        let first_report =
1124            write_report(&first, 2, 1, false, MssqlTargetCleanupStatus::NotApplicable);
1125        let failure_context = MssqlWriteFailureContext::from_output_plan(
1126            &second,
1127            MssqlWritePhase::WriteBatch,
1128            1,
1129            1,
1130            0,
1131            true,
1132            MssqlTargetCleanupStatus::NotApplicable,
1133        )
1134        .with_phase_timings(vec![
1135            PhaseTimingReport::completed("prepare_target_lifecycle", Duration::from_micros(10)),
1136            PhaseTimingReport::failed("write_batch", Duration::from_micros(20)),
1137            PhaseTimingReport::not_started(
1138                VALIDATION_PHASE,
1139                ReportReasonCode::FailureBeforeValidation,
1140            ),
1141        ]);
1142        let failure = phase_error_with_context(failure_context, "write failed");
1143        let writer = FakeWorkflowWriter::new(vec![Ok(first_report), Err(failure)]);
1144
1145        let report = write_mssql_outputs_with_writer(
1146            vec![job(first)?, job(second)?],
1147            MssqlWorkflowWriteOptions::default(),
1148            writer,
1149        )
1150        .await?;
1151
1152        let [first_status, second_status] = report.outputs() else {
1153            return Err(test_error("expected two output statuses"));
1154        };
1155        assert!(matches!(first_status, MssqlOutputWriteStatus::Succeeded(_)));
1156        let MssqlOutputWriteStatus::Failed(failure) = second_status else {
1157            return Err(test_error("expected second output to fail"));
1158        };
1159        assert_eq!(failure.output_name(), "second");
1160        let context = failure
1161            .context()
1162            .ok_or_else(|| test_error("expected write failure context"))?;
1163        assert_eq!(context.phase(), MssqlWritePhase::WriteBatch);
1164        assert!(context.partial_write_possible());
1165        assert_eq!(context.stats().rows_written(), 1);
1166        assert_eq!(context.stats().batches_written(), 1);
1167        assert_phase_timing(
1168            second_status,
1169            OUTPUT_STREAM_SETUP_PHASE,
1170            PhaseStatus::completed(),
1171        )?;
1172        assert_phase_timing(second_status, SQL_WRITE_PHASE, PhaseStatus::failed())?;
1173        assert_phase_timing(second_status, "write_batch", PhaseStatus::failed())?;
1174        assert_phase_timing(
1175            second_status,
1176            VALIDATION_PHASE,
1177            PhaseStatus::not_started(ReportReasonCode::FailureBeforeValidation),
1178        )?;
1179        assert_eq!(
1180            second_status
1181                .phase_timings()
1182                .iter()
1183                .filter(|timing| timing.phase_name() == VALIDATION_PHASE)
1184                .count(),
1185            1
1186        );
1187
1188        Ok(())
1189    }
1190
1191    #[tokio::test]
1192    async fn batch_schema_validation_failure_preserves_failure_context()
1193    -> Result<(), DeltaFunnelError> {
1194        let output = output_plan("schema_failure", LoadMode::AppendExisting)?;
1195        let context = MssqlWriteFailureContext::from_output_plan(
1196            &output,
1197            MssqlWritePhase::ValidateBatchSchema,
1198            0,
1199            0,
1200            0,
1201            false,
1202            MssqlTargetCleanupStatus::NotApplicable,
1203        );
1204        let failure = DeltaFunnelError::MssqlBatchSchemaValidation {
1205            context: Box::new(context),
1206            source: arrow_tiberius::Error::BackendUnavailable {
1207                backend: arrow_tiberius::WriteBackend::DirectRawBulk,
1208                reason: "schema mismatch".to_owned(),
1209            },
1210        };
1211        let writer = FakeWorkflowWriter::new(vec![Err(failure)]);
1212
1213        let report = write_mssql_outputs_with_writer(
1214            vec![job(output)?],
1215            MssqlWorkflowWriteOptions::default(),
1216            writer,
1217        )
1218        .await?;
1219
1220        let [MssqlOutputWriteStatus::Failed(failure)] = report.outputs() else {
1221            return Err(test_error("expected failed output status"));
1222        };
1223        let context = failure
1224            .context()
1225            .ok_or_else(|| test_error("expected schema validation context"))?;
1226        assert_eq!(failure.output_name(), "schema_failure");
1227        assert_eq!(context.phase(), MssqlWritePhase::ValidateBatchSchema);
1228        assert!(!context.partial_write_possible());
1229        assert_eq!(context.stats().rows_written(), 0);
1230
1231        Ok(())
1232    }
1233
1234    #[tokio::test]
1235    async fn first_failure_marks_later_outputs_skipped_without_attempting_them()
1236    -> Result<(), DeltaFunnelError> {
1237        let first = output_plan("first", LoadMode::AppendExisting)?;
1238        let second = output_plan("second", LoadMode::AppendExisting)?;
1239        let third = output_plan("third", LoadMode::AppendExisting)?;
1240        let failure = phase_error(
1241            &first,
1242            MssqlWritePhase::Connect,
1243            0,
1244            0,
1245            false,
1246            MssqlTargetCleanupStatus::NotApplicable,
1247            "connect failed",
1248        );
1249        let writer = FakeWorkflowWriter::new(vec![Err(failure)]);
1250        let attempted = writer.attempted_outputs();
1251
1252        let report = write_mssql_outputs_with_writer(
1253            vec![job(first)?, job(second)?, job(third)?],
1254            MssqlWorkflowWriteOptions::default(),
1255            writer,
1256        )
1257        .await?;
1258
1259        assert_eq!(locked(&attempted)?.as_slice(), ["first".to_owned()]);
1260        let [failed, skipped_second, skipped_third] = report.outputs() else {
1261            return Err(test_error("expected three output statuses"));
1262        };
1263        assert_eq!(report.len(), 3);
1264        assert_eq!(report.succeeded_count(), 0);
1265        assert_eq!(report.failed_count(), 1);
1266        assert_eq!(report.skipped_count(), 2);
1267        assert!(matches!(failed, MssqlOutputWriteStatus::Failed(_)));
1268        assert_eq!(failed.output_row_count(), RowCount::partial(0));
1269        assert_batch_shaping(failed.batch_shaping(), PhaseStatus::failed(), 0, 0, 0, 0);
1270        assert_phase_timing(failed, OUTPUT_STREAM_SETUP_PHASE, PhaseStatus::completed())?;
1271        assert_phase_timing(failed, SQL_WRITE_PHASE, PhaseStatus::failed())?;
1272        assert_skipped_after(skipped_second, "second", "first")?;
1273        assert_skipped_after(skipped_third, "third", "first")?;
1274
1275        Ok(())
1276    }
1277
1278    #[tokio::test]
1279    async fn output_status_accessors_cover_success_failure_and_skipped_variants()
1280    -> Result<(), DeltaFunnelError> {
1281        let first = output_plan("first", LoadMode::AppendExisting)?;
1282        let second = output_plan("second", LoadMode::CreateAndLoad)?;
1283        let third = output_plan("third", LoadMode::AppendExisting)?;
1284        let first_report =
1285            write_report(&first, 2, 1, false, MssqlTargetCleanupStatus::NotApplicable);
1286        let failure = phase_error(
1287            &second,
1288            MssqlWritePhase::InitializeWriter,
1289            0,
1290            0,
1291            false,
1292            MssqlTargetCleanupStatus::NotAttempted,
1293            "writer init failed",
1294        );
1295        let writer = FakeWorkflowWriter::new(vec![Ok(first_report), Err(failure)]);
1296
1297        let report = write_mssql_outputs_with_writer(
1298            vec![job(first)?, job(second)?, job(third)?],
1299            MssqlWorkflowWriteOptions::default(),
1300            writer,
1301        )
1302        .await?;
1303
1304        let [success, failed, skipped] = report.outputs() else {
1305            return Err(test_error("expected three output statuses"));
1306        };
1307        assert!(success.is_succeeded());
1308        assert_eq!(success.output_name(), "first");
1309        assert_eq!(success.target_table().table(), "first_orders");
1310        assert_eq!(success.load_mode(), LoadMode::AppendExisting);
1311        assert_eq!(
1312            success.connection().display_label(),
1313            Some("test connection")
1314        );
1315        assert_eq!(success.target_row_count(), RowCount::unavailable());
1316        assert_eq!(
1317            success.validation_status(),
1318            ValidationStatus::skipped(ReportReasonCode::NotExecuted)
1319        );
1320
1321        assert!(failed.is_failed());
1322        assert_eq!(failed.output_name(), "second");
1323        assert_eq!(failed.target_table().table(), "second_orders");
1324        assert_eq!(failed.load_mode(), LoadMode::CreateAndLoad);
1325        assert_eq!(failed.connection().display_label(), Some("test connection"));
1326        assert_eq!(failed.target_row_count(), RowCount::unavailable());
1327        assert_eq!(
1328            failed.validation_status(),
1329            ValidationStatus::skipped(ReportReasonCode::NotExecuted)
1330        );
1331
1332        assert!(skipped.is_skipped());
1333        assert_eq!(skipped.output_name(), "third");
1334        assert_eq!(skipped.target_table().table(), "third_orders");
1335        assert_eq!(skipped.load_mode(), LoadMode::AppendExisting);
1336        assert_eq!(
1337            skipped.connection().display_label(),
1338            Some("test connection")
1339        );
1340        assert_eq!(skipped.target_row_count(), RowCount::unavailable());
1341        assert_eq!(
1342            skipped.validation_status(),
1343            ValidationStatus::skipped(ReportReasonCode::PriorFailure)
1344        );
1345
1346        Ok(())
1347    }
1348
1349    #[tokio::test]
1350    async fn failed_output_status_exposes_validation_evidence() -> Result<(), DeltaFunnelError> {
1351        let output = output_plan("validation_failed", LoadMode::CreateAndLoad)?;
1352        let metrics = MssqlWriteReportMetrics::new(
1353            RowCount::exact(3),
1354            MssqlBatchShapingReport::completed(1, 3, 1, 3),
1355            3,
1356            1,
1357            0,
1358            false,
1359            MssqlTargetCleanupStatus::Succeeded,
1360        )
1361        .with_target_validation(RowCount::exact(4), ValidationStatus::failed())
1362        .with_phase_timings(vec![PhaseTimingReport::failed(
1363            VALIDATION_PHASE,
1364            Duration::from_micros(5),
1365        )]);
1366        let failure = phase_error_with_context(
1367            MssqlWriteFailureContext::from_output_plan_with_metrics(
1368                &output,
1369                MssqlWritePhase::Validation,
1370                metrics,
1371            ),
1372            "target row count did not match exact output rows",
1373        );
1374        let writer = FakeWorkflowWriter::new(vec![Err(failure)]);
1375
1376        let report = write_mssql_outputs_with_writer(
1377            vec![job(output)?],
1378            MssqlWorkflowWriteOptions::default(),
1379            writer,
1380        )
1381        .await?;
1382
1383        let [status] = report.outputs() else {
1384            return Err(test_error("expected one output status"));
1385        };
1386        let MssqlOutputWriteStatus::Failed(failure) = status else {
1387            return Err(test_error("expected failed output status"));
1388        };
1389        assert_eq!(failure.target_row_count(), RowCount::exact(4));
1390        assert_eq!(failure.validation_status(), ValidationStatus::failed());
1391        assert_eq!(status.target_row_count(), RowCount::exact(4));
1392        assert_eq!(status.validation_status(), ValidationStatus::failed());
1393        assert_phase_timing(status, VALIDATION_PHASE, PhaseStatus::failed())?;
1394        Ok(())
1395    }
1396
1397    #[tokio::test]
1398    async fn skipped_output_stream_factories_are_not_invoked() -> Result<(), DeltaFunnelError> {
1399        let first = output_plan("first", LoadMode::AppendExisting)?;
1400        let second = output_plan("second", LoadMode::AppendExisting)?;
1401        let factory_calls = Arc::new(Mutex::new(Vec::new()));
1402        let failure = phase_error(
1403            &first,
1404            MssqlWritePhase::Connect,
1405            0,
1406            0,
1407            false,
1408            MssqlTargetCleanupStatus::NotApplicable,
1409            "connect failed",
1410        );
1411        let writer = FakeWorkflowWriter::new(vec![Err(failure)]);
1412
1413        let report = write_mssql_outputs_with_writer(
1414            vec![
1415                counted_job(first, Arc::clone(&factory_calls))?,
1416                counted_job(second, Arc::clone(&factory_calls))?,
1417            ],
1418            MssqlWorkflowWriteOptions::default(),
1419            writer,
1420        )
1421        .await?;
1422
1423        assert_eq!(locked(&factory_calls)?.as_slice(), ["first".to_owned()]);
1424        assert_eq!(report.outputs().len(), 2);
1425        assert!(report.outputs()[1].is_skipped());
1426
1427        Ok(())
1428    }
1429
1430    #[tokio::test]
1431    async fn skipped_outputs_do_not_reach_one_output_writer() -> Result<(), DeltaFunnelError> {
1432        let first = output_plan("first", LoadMode::AppendExisting)?;
1433        let second = output_plan("second", LoadMode::AppendExisting)?;
1434        let failure = phase_error(
1435            &first,
1436            MssqlWritePhase::PrepareTargetLifecycle,
1437            0,
1438            0,
1439            false,
1440            MssqlTargetCleanupStatus::NotApplicable,
1441            "prepare failed",
1442        );
1443        let writer = FakeWorkflowWriter::new(vec![Err(failure)]);
1444        let attempted = writer.attempted_outputs();
1445
1446        let report = write_mssql_outputs_with_writer(
1447            vec![job(first)?, job(second)?],
1448            MssqlWorkflowWriteOptions::default(),
1449            writer,
1450        )
1451        .await?;
1452
1453        assert_eq!(locked(&attempted)?.as_slice(), ["first".to_owned()]);
1454        assert!(report.outputs()[1].is_skipped());
1455
1456        Ok(())
1457    }
1458
1459    #[tokio::test]
1460    async fn stream_factory_setup_failure_fails_output_before_writer_and_skips_later_factories()
1461    -> Result<(), DeltaFunnelError> {
1462        let first = output_plan("first", LoadMode::AppendExisting)?;
1463        let second = output_plan("second", LoadMode::AppendExisting)?;
1464        let third = output_plan("third", LoadMode::AppendExisting)?;
1465        let first_report =
1466            write_report(&first, 1, 1, false, MssqlTargetCleanupStatus::NotApplicable);
1467        let factory_calls = Arc::new(Mutex::new(Vec::new()));
1468        let writer = FakeWorkflowWriter::new(vec![Ok(first_report)]);
1469        let attempted = writer.attempted_outputs();
1470
1471        let report = write_mssql_outputs_with_writer(
1472            vec![
1473                counted_job(first, Arc::clone(&factory_calls))?,
1474                failing_factory_job(
1475                    second,
1476                    Arc::clone(&factory_calls),
1477                    "stream setup failed before SQL writer",
1478                )?,
1479                counted_job(third, Arc::clone(&factory_calls))?,
1480            ],
1481            MssqlWorkflowWriteOptions::default(),
1482            writer,
1483        )
1484        .await?;
1485
1486        assert_eq!(
1487            locked(&factory_calls)?.as_slice(),
1488            ["first".to_owned(), "second".to_owned()]
1489        );
1490        assert_eq!(locked(&attempted)?.as_slice(), ["first".to_owned()]);
1491        let [first_status, second_status, third_status] = report.outputs() else {
1492            return Err(test_error("expected three output statuses"));
1493        };
1494        assert!(first_status.is_succeeded());
1495        let MssqlOutputWriteStatus::Failed(failure) = second_status else {
1496            return Err(test_error("expected second output to fail"));
1497        };
1498        assert_eq!(failure.output_name(), "second");
1499        assert!(failure.context().is_none());
1500        assert_eq!(failure.output_row_count(), RowCount::unavailable());
1501        assert_batch_shaping(
1502            failure.batch_shaping(),
1503            PhaseStatus::not_started(ReportReasonCode::NotExecuted),
1504            0,
1505            0,
1506            0,
1507            0,
1508        );
1509        assert_phase_timing(second_status, PLANNED_PHASE, PhaseStatus::completed())?;
1510        assert_phase_timing(
1511            second_status,
1512            OUTPUT_STREAM_SETUP_PHASE,
1513            PhaseStatus::failed(),
1514        )?;
1515        assert_phase_timing(
1516            second_status,
1517            SQL_WRITE_PHASE,
1518            PhaseStatus::not_started(ReportReasonCode::NotExecuted),
1519        )?;
1520        assert_phase_timing(
1521            second_status,
1522            VALIDATION_PHASE,
1523            PhaseStatus::not_started(ReportReasonCode::FailureBeforeValidation),
1524        )?;
1525        assert!(
1526            failure
1527                .error()
1528                .contains("stream setup failed before SQL writer")
1529        );
1530        assert_skipped_after(third_status, "third", "second")?;
1531
1532        Ok(())
1533    }
1534
1535    #[tokio::test]
1536    async fn stream_polling_failure_after_setup_reaches_writer_boundary()
1537    -> Result<(), DeltaFunnelError> {
1538        let output = output_plan("poll_failure", LoadMode::AppendExisting)?;
1539        let factory_calls = Arc::new(Mutex::new(Vec::new()));
1540        let writer = StreamPollingWorkflowWriter::default();
1541        let attempted = writer.attempted_outputs();
1542
1543        let report = write_mssql_outputs_with_writer(
1544            vec![polling_error_job(output, Arc::clone(&factory_calls))?],
1545            MssqlWorkflowWriteOptions::default(),
1546            writer,
1547        )
1548        .await?;
1549
1550        assert_eq!(
1551            locked(&factory_calls)?.as_slice(),
1552            ["poll_failure".to_owned()]
1553        );
1554        assert_eq!(locked(&attempted)?.as_slice(), ["poll_failure".to_owned()]);
1555        let [MssqlOutputWriteStatus::Failed(failure)] = report.outputs() else {
1556            return Err(test_error("expected failed output status"));
1557        };
1558        assert_eq!(failure.output_name(), "poll_failure");
1559        assert!(failure.error().contains("stream failed during polling"));
1560        assert!(failure.context().is_none());
1561        assert_eq!(failure.output_row_count(), RowCount::unavailable());
1562        assert_batch_shaping(
1563            failure.batch_shaping(),
1564            PhaseStatus::not_started(ReportReasonCode::NotExecuted),
1565            0,
1566            0,
1567            0,
1568            0,
1569        );
1570        assert_phase_timing(
1571            &report.outputs()[0],
1572            OUTPUT_STREAM_SETUP_PHASE,
1573            PhaseStatus::completed(),
1574        )?;
1575        assert_phase_timing(&report.outputs()[0], SQL_WRITE_PHASE, PhaseStatus::failed())?;
1576
1577        Ok(())
1578    }
1579
1580    #[tokio::test]
1581    async fn failed_create_and_load_cleanup_status_is_preserved() -> Result<(), DeltaFunnelError> {
1582        let output = output_plan("created", LoadMode::CreateAndLoad)?;
1583        let failure = phase_error(
1584            &output,
1585            MssqlWritePhase::Finalize,
1586            2,
1587            1,
1588            false,
1589            MssqlTargetCleanupStatus::Succeeded,
1590            "finalize failed",
1591        );
1592        let writer = FakeWorkflowWriter::new(vec![Err(failure)]);
1593
1594        let report = write_mssql_outputs_with_writer(
1595            vec![job(output)?],
1596            MssqlWorkflowWriteOptions::default(),
1597            writer,
1598        )
1599        .await?;
1600
1601        let [MssqlOutputWriteStatus::Failed(failure)] = report.outputs() else {
1602            return Err(test_error("expected failed output status"));
1603        };
1604        let context = failure
1605            .context()
1606            .ok_or_else(|| test_error("expected write failure context"))?;
1607        assert_eq!(context.cleanup(), MssqlTargetCleanupStatus::Succeeded);
1608        assert_eq!(context.stats().rows_written(), 2);
1609        assert_eq!(context.stats().batches_written(), 1);
1610
1611        Ok(())
1612    }
1613
1614    #[tokio::test]
1615    async fn parallel_writer_configuration_is_rejected() -> Result<(), DeltaFunnelError> {
1616        let output = output_plan("first", LoadMode::AppendExisting)?;
1617        let writer = FakeWorkflowWriter::new(vec![Ok(write_report(
1618            &output,
1619            1,
1620            1,
1621            false,
1622            MssqlTargetCleanupStatus::NotApplicable,
1623        ))]);
1624
1625        let error = write_mssql_outputs_with_writer(
1626            vec![job(output)?],
1627            MssqlWorkflowWriteOptions::new().with_max_parallel_outputs(2),
1628            writer,
1629        )
1630        .await;
1631        let Err(error) = error else {
1632            return Err(test_error("parallel writer config should be rejected"));
1633        };
1634
1635        assert!(error.to_string().contains("parallel MSSQL output writers"));
1636
1637        Ok(())
1638    }
1639
1640    #[tokio::test]
1641    async fn zero_parallel_writer_configuration_is_rejected_before_attempting_outputs()
1642    -> Result<(), DeltaFunnelError> {
1643        let output = output_plan("first", LoadMode::AppendExisting)?;
1644        let writer = FakeWorkflowWriter::new(vec![Ok(write_report(
1645            &output,
1646            1,
1647            1,
1648            false,
1649            MssqlTargetCleanupStatus::NotApplicable,
1650        ))]);
1651        let attempted = writer.attempted_outputs();
1652
1653        let error = write_mssql_outputs_with_writer(
1654            vec![job(output)?],
1655            MssqlWorkflowWriteOptions::new().with_max_parallel_outputs(0),
1656            writer,
1657        )
1658        .await;
1659        let Err(error) = error else {
1660            return Err(test_error("zero writer config should be rejected"));
1661        };
1662
1663        assert!(error.to_string().contains("must be at least 1"));
1664        assert!(locked(&attempted)?.is_empty());
1665
1666        Ok(())
1667    }
1668
1669    #[tokio::test]
1670    async fn workflow_report_debug_and_display_redact_connection_credentials()
1671    -> Result<(), DeltaFunnelError> {
1672        let output = output_plan("first", LoadMode::AppendExisting)?;
1673        let report = write_report(
1674            &output,
1675            1,
1676            1,
1677            false,
1678            MssqlTargetCleanupStatus::NotApplicable,
1679        );
1680        let writer = FakeWorkflowWriter::new(vec![Ok(report)]);
1681
1682        let report = write_mssql_outputs_with_writer(
1683            vec![job(output)?],
1684            MssqlWorkflowWriteOptions::default(),
1685            writer,
1686        )
1687        .await?;
1688
1689        let debug = format!("{report:?}");
1690        let display = report.to_string();
1691        assert!(!debug.contains("secret"));
1692        assert!(!display.contains("secret"));
1693        assert!(display.contains("1 succeeded"));
1694        assert!(!display.to_lowercase().contains("transaction"));
1695
1696        Ok(())
1697    }
1698
1699    fn job(output_plan: MssqlTargetOutputPlan) -> Result<MssqlOutputWriteJob, DeltaFunnelError> {
1700        counted_job(output_plan, Arc::new(Mutex::new(Vec::new())))
1701    }
1702
1703    fn counted_job(
1704        output_plan: MssqlTargetOutputPlan,
1705        factory_calls: Arc<Mutex<Vec<String>>>,
1706    ) -> Result<MssqlOutputWriteJob, DeltaFunnelError> {
1707        let output_name = output_plan.output_name().to_owned();
1708        Ok(MssqlOutputWriteJob::with_default_write_options(
1709            output_schema(),
1710            resolved_target(output_plan)?,
1711            MssqlSchemaPlanOptions::default(),
1712            move || {
1713                if let Ok(mut calls) = factory_calls.lock() {
1714                    calls.push(output_name);
1715                }
1716                async { Ok(stream::empty()) }
1717            },
1718        )
1719        .with_phase_timings(planned_phase_timings()))
1720    }
1721
1722    fn failing_factory_job(
1723        output_plan: MssqlTargetOutputPlan,
1724        factory_calls: Arc<Mutex<Vec<String>>>,
1725        message: &'static str,
1726    ) -> Result<MssqlOutputWriteJob, DeltaFunnelError> {
1727        let output_name = output_plan.output_name().to_owned();
1728        Ok(MssqlOutputWriteJob::with_default_write_options(
1729            output_schema(),
1730            resolved_target(output_plan)?,
1731            MssqlSchemaPlanOptions::default(),
1732            move || {
1733                if let Ok(mut calls) = factory_calls.lock() {
1734                    calls.push(output_name);
1735                }
1736                async move {
1737                    Err::<stream::Empty<Result<RecordBatch, DeltaFunnelError>>, DeltaFunnelError>(
1738                        test_error(message),
1739                    )
1740                }
1741            },
1742        )
1743        .with_phase_timings(planned_phase_timings()))
1744    }
1745
1746    fn polling_error_job(
1747        output_plan: MssqlTargetOutputPlan,
1748        factory_calls: Arc<Mutex<Vec<String>>>,
1749    ) -> Result<MssqlOutputWriteJob, DeltaFunnelError> {
1750        let output_name = output_plan.output_name().to_owned();
1751        Ok(MssqlOutputWriteJob::with_default_write_options(
1752            output_schema(),
1753            resolved_target(output_plan)?,
1754            MssqlSchemaPlanOptions::default(),
1755            move || {
1756                if let Ok(mut calls) = factory_calls.lock() {
1757                    calls.push(output_name);
1758                }
1759                async {
1760                    Ok(stream::iter(vec![Err(
1761                        DeltaFunnelError::MssqlWorkflowPlanning {
1762                            message: "stream failed during polling".to_owned(),
1763                        },
1764                    )]))
1765                }
1766            },
1767        )
1768        .with_phase_timings(planned_phase_timings()))
1769    }
1770
1771    fn planned_phase_timings() -> Vec<PhaseTimingReport> {
1772        vec![PhaseTimingReport::completed(
1773            PLANNED_PHASE,
1774            Duration::from_micros(1),
1775        )]
1776    }
1777
1778    fn resolved_target(
1779        output_plan: MssqlTargetOutputPlan,
1780    ) -> Result<ResolvedMssqlTarget, DeltaFunnelError> {
1781        let connection = secret_connection()?;
1782
1783        MssqlTargetConfig::new(output_plan.target_table().clone())
1784            .with_load_mode(output_plan.load_mode())
1785            .resolve(MssqlTargetResolutionContext {
1786                output_name: Some(output_plan.output_name()),
1787                default_connection: Some(&connection),
1788            })
1789    }
1790
1791    fn output_plan(
1792        output_name: &str,
1793        load_mode: LoadMode,
1794    ) -> Result<MssqlTargetOutputPlan, DeltaFunnelError> {
1795        let connection = secret_connection()?;
1796        let target = MssqlTargetConfig::new(MssqlTargetTable::new(
1797            "dbo",
1798            format!("{output_name}_orders"),
1799        )?)
1800        .with_load_mode(load_mode);
1801        plan_mssql_target_for_output(
1802            output_schema(),
1803            output_name,
1804            &target,
1805            Some(&connection),
1806            MssqlSchemaPlanOptions::default(),
1807        )
1808    }
1809
1810    fn output_schema() -> SchemaRef {
1811        Arc::new(Schema::new(vec![Field::new(
1812            "order_id",
1813            DataType::Int64,
1814            false,
1815        )]))
1816    }
1817
1818    fn secret_connection() -> Result<MssqlConnectionConfig, DeltaFunnelError> {
1819        Ok(MssqlConnectionConfig::new(
1820            "server=tcp:example.invalid,1433;user id=sa;password=secret",
1821        )?
1822        .with_display_label("test connection"))
1823    }
1824
1825    fn write_report(
1826        output_plan: &MssqlTargetOutputPlan,
1827        rows_written: u64,
1828        batches_written: u64,
1829        partial_write_possible: bool,
1830        cleanup: MssqlTargetCleanupStatus,
1831    ) -> MssqlWriteReport {
1832        MssqlWriteReport::from_output_plan(
1833            output_plan,
1834            rows_written,
1835            batches_written,
1836            0,
1837            partial_write_possible,
1838            cleanup,
1839        )
1840    }
1841
1842    fn phase_error(
1843        output_plan: &MssqlTargetOutputPlan,
1844        phase: MssqlWritePhase,
1845        rows_written: u64,
1846        batches_written: u64,
1847        partial_write_possible: bool,
1848        cleanup: MssqlTargetCleanupStatus,
1849        message: &str,
1850    ) -> DeltaFunnelError {
1851        phase_error_with_context(
1852            MssqlWriteFailureContext::from_output_plan(
1853                output_plan,
1854                phase,
1855                rows_written,
1856                batches_written,
1857                0,
1858                partial_write_possible,
1859                cleanup,
1860            ),
1861            message,
1862        )
1863    }
1864
1865    fn phase_error_with_context(
1866        context: MssqlWriteFailureContext,
1867        message: &str,
1868    ) -> DeltaFunnelError {
1869        DeltaFunnelError::MssqlWritePhase {
1870            context: Box::new(context),
1871            message: message.to_owned(),
1872        }
1873    }
1874
1875    fn assert_status_output(
1876        statuses: &[MssqlOutputWriteStatus],
1877        index: usize,
1878        expected_output_name: &str,
1879    ) -> Result<(), DeltaFunnelError> {
1880        match statuses.get(index) {
1881            Some(MssqlOutputWriteStatus::Succeeded(report)) => {
1882                assert_eq!(report.output_name(), expected_output_name);
1883                Ok(())
1884            }
1885            Some(other) => Err(test_error(format!(
1886                "expected success at index {index}, got {other:?}"
1887            ))),
1888            None => Err(test_error(format!("missing status at index {index}"))),
1889        }
1890    }
1891
1892    fn assert_skipped_after(
1893        status: &MssqlOutputWriteStatus,
1894        expected_output_name: &str,
1895        expected_failed_output_name: &str,
1896    ) -> Result<(), DeltaFunnelError> {
1897        let MssqlOutputWriteStatus::Skipped(skipped) = status else {
1898            return Err(test_error(format!(
1899                "expected skipped status, got {status:?}"
1900            )));
1901        };
1902        assert_eq!(skipped.output_name(), expected_output_name);
1903        assert_eq!(
1904            skipped.reason(),
1905            &MssqlWriteSkippedReason::PreviousOutputFailed {
1906                failed_output_name: expected_failed_output_name.to_owned()
1907            }
1908        );
1909        assert_eq!(status.output_row_count(), RowCount::unavailable());
1910        assert_batch_shaping(
1911            status.batch_shaping(),
1912            PhaseStatus::skipped(ReportReasonCode::PriorFailure),
1913            0,
1914            0,
1915            0,
1916            0,
1917        );
1918        assert_phase_timing(status, PLANNED_PHASE, PhaseStatus::completed())?;
1919        assert_phase_timing(
1920            status,
1921            OUTPUT_STREAM_SETUP_PHASE,
1922            PhaseStatus::skipped(ReportReasonCode::PriorFailure),
1923        )?;
1924        assert_phase_timing(
1925            status,
1926            SQL_WRITE_PHASE,
1927            PhaseStatus::skipped(ReportReasonCode::PriorFailure),
1928        )?;
1929        assert_phase_timing(
1930            status,
1931            VALIDATION_PHASE,
1932            PhaseStatus::skipped(ReportReasonCode::PriorFailure),
1933        )?;
1934        Ok(())
1935    }
1936
1937    fn assert_phase_timing(
1938        status: &MssqlOutputWriteStatus,
1939        phase_name: &str,
1940        expected_status: PhaseStatus,
1941    ) -> Result<(), DeltaFunnelError> {
1942        let timing = status
1943            .phase_timings()
1944            .iter()
1945            .find(|timing| timing.phase_name() == phase_name)
1946            .ok_or_else(|| test_error(format!("missing phase timing {phase_name}")))?;
1947
1948        assert_eq!(timing.status(), expected_status);
1949        if expected_status.is_completed() || expected_status.is_failed() {
1950            assert!(timing.elapsed_micros().is_some());
1951        } else {
1952            assert_eq!(timing.elapsed_micros(), None);
1953        }
1954        Ok(())
1955    }
1956
1957    fn assert_batch_shaping(
1958        report: MssqlBatchShapingReport,
1959        expected_status: PhaseStatus,
1960        expected_input_batches: u64,
1961        expected_input_rows: u64,
1962        expected_output_batches: u64,
1963        expected_output_rows: u64,
1964    ) {
1965        assert_eq!(report.status(), expected_status);
1966        assert_eq!(report.input_batches(), expected_input_batches);
1967        assert_eq!(report.input_rows(), expected_input_rows);
1968        assert_eq!(report.output_batches(), expected_output_batches);
1969        assert_eq!(report.output_rows(), expected_output_rows);
1970    }
1971
1972    fn locked<T>(mutex: &Mutex<T>) -> Result<MutexGuard<'_, T>, DeltaFunnelError> {
1973        mutex.lock().map_err(|_| test_error("mutex lock poisoned"))
1974    }
1975
1976    fn test_error(message: impl Into<String>) -> DeltaFunnelError {
1977        DeltaFunnelError::MssqlWorkflowPlanning {
1978            message: message.into(),
1979        }
1980    }
1981}
delta_funnel/sql_server/execution/workflow.rs

delta_funnel/sql_server/execution/
workflow.rs
