1use std::sync::Arc;
2
3use async_trait::async_trait;
4use datafusion::arrow::datatypes::SchemaRef;
5use tracing::Instrument;
6
7use crate::{
8 DeltaFunnelError, MssqlOutputBatchStream, MssqlTargetOutputPlan, MssqlWriteBackend,
9 MssqlWriteReport, PhaseTimingReport, ReportReasonCode, ResolvedMssqlTarget, ValidationOptions,
10 observability, plan_mssql_target_for_resolved_output, report::PhaseTimer,
11 write_output_batches_to_mssql_with_validation_options,
12};
13
14use super::super::{DeltaFunnelSession, OutputWritePlan, PlannedMssqlOutput, RunMode};
15
16pub(in crate::orchestrator::session) const OUTPUT_SCHEMA_PLANNING_PHASE: &str =
17 "output_schema_planning";
18pub(in crate::orchestrator::session) const SQL_TARGET_PLANNING_PHASE: &str = "sql_target_planning";
19const VALIDATION_PHASE: &str = "validation";
20
21impl DeltaFunnelSession {
22 pub fn plan_mssql_output(
38 &self,
39 request: &OutputWritePlan,
40 ) -> Result<PlannedMssqlOutput, DeltaFunnelError> {
41 let mut phase_timings = self.phase_timings_for_lazy_table(request.table())?;
42
43 let schema_timer = PhaseTimer::start(OUTPUT_SCHEMA_PLANNING_PHASE);
44 let schema = match self.schema_for_lazy_table(request.table()) {
45 Ok(schema) => {
46 phase_timings.push(schema_timer.completed());
47 schema
48 }
49 Err(error) => {
50 phase_timings.push(schema_timer.failed());
51 return Err(error);
52 }
53 };
54
55 let target_timer = PhaseTimer::start(SQL_TARGET_PLANNING_PHASE);
56 let resolved_target =
57 match request
58 .target()
59 .target()
60 .resolve(crate::MssqlTargetResolutionContext {
61 output_name: Some(request.target().output_name()),
62 default_connection: self.options.default_mssql_connection(),
63 }) {
64 Ok(resolved_target) => resolved_target,
65 Err(error) => {
66 phase_timings.push(target_timer.failed());
67 return Err(error);
68 }
69 };
70 let output_plan = match plan_mssql_target_for_resolved_output(
71 schema.as_ref(),
72 &resolved_target,
73 self.options.mssql_schema_options(),
74 ) {
75 Ok(output_plan) => {
76 phase_timings.push(target_timer.completed());
77 output_plan
78 }
79 Err(error) => {
80 phase_timings.push(target_timer.failed());
81 return Err(error);
82 }
83 };
84
85 Ok(PlannedMssqlOutput::new(
86 request.clone(),
87 resolved_target,
88 output_plan,
89 phase_timings,
90 ))
91 }
92
93 pub async fn write_to_mssql(
106 &self,
107 request: &OutputWritePlan,
108 ) -> Result<MssqlWriteReport, DeltaFunnelError> {
109 self.write_to_mssql_with_writer_with_tracing(request, &mut MssqlPublicOneOutputWriter)
110 .await
111 }
112
113 async fn write_to_mssql_with_writer_with_tracing<W>(
114 &self,
115 request: &OutputWritePlan,
116 writer: &mut W,
117 ) -> Result<MssqlWriteReport, DeltaFunnelError>
118 where
119 W: OrchestratorMssqlOutputWriter,
120 {
121 ensure_execute_run_mode(request.target().run_mode())?;
122 let target = request.target();
123 let target_config = target.target();
124 let output_span = observability::output_span(
125 target.output_name(),
126 target_config.table(),
127 target_config.load_mode(),
128 );
129
130 async move {
131 observability::output_started(
132 target.output_name(),
133 target_config.table(),
134 target_config.load_mode(),
135 );
136 let result = self.write_to_mssql_with_writer(request, writer).await;
137 match &result {
138 Ok(report) => observability::output_completed(
139 report.output_name(),
140 report.target_table(),
141 report.load_mode(),
142 ),
143 Err(error) => observability::output_failed(
144 target.output_name(),
145 target_config.table(),
146 target_config.load_mode(),
147 &error.to_string(),
148 ),
149 }
150
151 result
152 }
153 .instrument(output_span)
154 .await
155 }
156
157 pub(crate) async fn write_to_mssql_with_writer<W>(
158 &self,
159 request: &OutputWritePlan,
160 writer: &mut W,
161 ) -> Result<MssqlWriteReport, DeltaFunnelError>
162 where
163 W: OrchestratorMssqlOutputWriter,
164 {
165 ensure_execute_run_mode(request.target().run_mode())?;
166 let planned = self.plan_mssql_output(request)?;
167 let output_schema = Arc::clone(self.schema_for_lazy_table(planned.table())?);
168 let batches = self.batch_stream_for_lazy_table(planned.table()).await?;
169
170 let phase_timings = planned.phase_timings().to_vec();
171 writer
172 .write_output(
173 output_schema,
174 planned.output_plan().clone(),
175 planned.resolved_target().clone(),
176 batches,
177 self.options.mssql_write_backend(),
178 self.options.validation_options(),
179 )
180 .await
181 .map(ensure_validation_phase_timing)
182 .map(|report| report.with_phase_timings(phase_timings))
183 }
184}
185
186fn ensure_validation_phase_timing(report: MssqlWriteReport) -> MssqlWriteReport {
187 if report
188 .phase_timings()
189 .iter()
190 .any(|timing| timing.phase_name() == VALIDATION_PHASE)
191 {
192 return report;
193 }
194
195 report.with_appended_phase_timings(vec![PhaseTimingReport::not_started(
196 VALIDATION_PHASE,
197 ReportReasonCode::NotExecuted,
198 )])
199}
200
201#[async_trait]
202pub(crate) trait OrchestratorMssqlOutputWriter: Send {
203 async fn write_output(
204 &mut self,
205 output_schema: SchemaRef,
206 output_plan: MssqlTargetOutputPlan,
207 resolved_target: ResolvedMssqlTarget,
208 batches: MssqlOutputBatchStream,
209 write_backend: MssqlWriteBackend,
210 validation_options: ValidationOptions,
211 ) -> Result<MssqlWriteReport, DeltaFunnelError>;
212}
213
214struct MssqlPublicOneOutputWriter;
215
216#[async_trait]
217impl OrchestratorMssqlOutputWriter for MssqlPublicOneOutputWriter {
218 async fn write_output(
219 &mut self,
220 output_schema: SchemaRef,
221 output_plan: MssqlTargetOutputPlan,
222 resolved_target: ResolvedMssqlTarget,
223 batches: MssqlOutputBatchStream,
224 write_backend: MssqlWriteBackend,
225 validation_options: ValidationOptions,
226 ) -> Result<MssqlWriteReport, DeltaFunnelError> {
227 write_output_batches_to_mssql_with_validation_options(
228 output_schema.as_ref(),
229 resolved_target,
230 output_plan.schema_plan_options(),
231 batches,
232 write_backend,
233 validation_options,
234 )
235 .await
236 }
237}
238
239fn ensure_execute_run_mode(run_mode: RunMode) -> Result<(), DeltaFunnelError> {
240 match run_mode {
241 RunMode::Execute => Ok(()),
242 RunMode::DryRun => Err(DeltaFunnelError::MssqlWorkflowPlanning {
243 message:
244 "write_to_mssql requires RunMode::Execute; use dry_run_to_mssql for dry-run planning"
245 .to_owned(),
246 }),
247 }
248}
249
250#[cfg(test)]
251mod tests {
252 use async_trait::async_trait;
253 use datafusion::arrow::datatypes::SchemaRef;
254 use futures_util::StreamExt;
255
256 use crate::{
257 DeltaFunnelError, DeltaSourceConfig, LoadMode, MssqlConnectionSource,
258 MssqlOutputBatchStream, MssqlOutputTarget, MssqlTargetCleanupStatus, MssqlTargetConfig,
259 MssqlTargetOutputPlan, MssqlTargetTable, MssqlWriteBackend, MssqlWriteReport,
260 ResolvedMssqlTarget, TargetValidationMode, ValidationOptions,
261 table_formats::RealParquetDeltaTable,
262 };
263
264 use super::super::super::{
265 DeltaFunnelSession, LazyTable, LazyTableKind, OutputWritePlan, RunMode, SessionOptions,
266 test_support::{
267 DeltaLogTable, UNSUPPORTED_SCHEMA_FIELDS_JSON, execute_output_request, output_request,
268 override_connection, secret_connection,
269 },
270 };
271 use super::{
272 OUTPUT_SCHEMA_PLANNING_PHASE, OrchestratorMssqlOutputWriter, SQL_TARGET_PLANNING_PHASE,
273 VALIDATION_PHASE,
274 };
275
276 #[derive(Debug, Clone, PartialEq, Eq)]
277 struct FakeOrchestratorWriteCall {
278 output_name: String,
279 target_table: MssqlTargetTable,
280 connection_source: MssqlConnectionSource,
281 rows: u64,
282 batches: u64,
283 schema_fields: usize,
284 validation_options: ValidationOptions,
285 }
286
287 #[derive(Default)]
288 struct FakeOrchestratorWriter {
289 calls: Vec<FakeOrchestratorWriteCall>,
290 }
291
292 #[async_trait]
293 impl OrchestratorMssqlOutputWriter for FakeOrchestratorWriter {
294 async fn write_output(
295 &mut self,
296 output_schema: SchemaRef,
297 output_plan: MssqlTargetOutputPlan,
298 resolved_target: ResolvedMssqlTarget,
299 mut batches: MssqlOutputBatchStream,
300 _write_backend: MssqlWriteBackend,
301 validation_options: ValidationOptions,
302 ) -> Result<MssqlWriteReport, DeltaFunnelError> {
303 let mut rows = 0_u64;
304 let mut batch_count = 0_u64;
305
306 while let Some(batch) = batches.next().await {
307 let batch = batch?;
308 rows = rows.saturating_add(u64::try_from(batch.num_rows()).map_err(|_| {
309 DeltaFunnelError::Config {
310 message: "fake writer row count overflowed u64".to_owned(),
311 }
312 })?);
313 batch_count = batch_count.saturating_add(1);
314 }
315
316 self.calls.push(FakeOrchestratorWriteCall {
317 output_name: resolved_target.output_name().to_owned(),
318 target_table: resolved_target.table().clone(),
319 connection_source: resolved_target.connection_source(),
320 rows,
321 batches: batch_count,
322 schema_fields: output_schema.fields().len(),
323 validation_options,
324 });
325
326 Ok(MssqlWriteReport::from_output_plan(
327 &output_plan,
328 rows,
329 batch_count,
330 0,
331 false,
332 MssqlTargetCleanupStatus::NotApplicable,
333 ))
334 }
335 }
336
337 #[test]
338 fn plan_mssql_output_uses_source_schema_and_session_connection()
339 -> Result<(), Box<dyn std::error::Error>> {
340 let table = DeltaLogTable::new("orders")?;
341 let mut session = DeltaFunnelSession::new(
342 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
343 )?;
344 let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
345 let request = output_request(
346 source.clone(),
347 "orders_output",
348 "orders_sink",
349 LoadMode::AppendExisting,
350 )?;
351
352 let planned = session.plan_mssql_output(&request)?;
353
354 assert_eq!(planned.request(), &request);
355 assert_eq!(planned.table(), &source);
356 assert_eq!(planned.target().run_mode(), RunMode::DryRun);
357 assert_eq!(planned.output_plan().output_name(), "orders_output");
358 assert_eq!(planned.output_plan().target_table().schema(), Some("dbo"));
359 assert_eq!(planned.output_plan().target_table().table(), "orders_sink");
360 assert_eq!(
361 planned.output_plan().connection_source(),
362 MssqlConnectionSource::ContextDefault
363 );
364 assert_eq!(
365 planned.output_plan().connection().display_label(),
366 Some("warehouse-primary")
367 );
368 assert_eq!(planned.output_plan().schema_mappings().len(), 2);
369 assert_eq!(
370 planned.output_plan().schema_mappings()[0].arrow().name(),
371 "id"
372 );
373 assert_eq!(
374 planned.output_plan().schema_mappings()[1].arrow().name(),
375 "customer_name"
376 );
377 assert_eq!(planned.output_plan().create_table_sql(), None);
378 assert_eq!(
379 planned
380 .phase_timings()
381 .iter()
382 .map(crate::PhaseTimingReport::phase_name)
383 .collect::<Vec<_>>(),
384 vec![OUTPUT_SCHEMA_PLANNING_PHASE, SQL_TARGET_PLANNING_PHASE]
385 );
386 assert!(
387 planned
388 .phase_timings()
389 .iter()
390 .all(|timing| timing.status().is_completed())
391 );
392 assert!(
393 planned
394 .phase_timings()
395 .iter()
396 .all(|timing| timing.elapsed_micros().is_some())
397 );
398 Ok(())
399 }
400
401 #[tokio::test]
402 async fn plan_mssql_output_uses_pending_derived_schema_without_row_execution()
403 -> Result<(), Box<dyn std::error::Error>> {
404 let table = DeltaLogTable::new("orders")?;
405 let mut session = DeltaFunnelSession::new(
406 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
407 )?;
408 session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
409 let derived = session.table_from_sql("select id from orders").await?;
410 let request = output_request(
411 derived.clone(),
412 "derived_orders_output",
413 "derived_orders",
414 LoadMode::CreateAndLoad,
415 )?;
416
417 let planned = session.plan_mssql_output(&request)?;
418
419 assert_eq!(planned.table(), &derived);
420 assert_eq!(planned.output_plan().load_mode(), LoadMode::CreateAndLoad);
421 assert_eq!(planned.output_plan().schema_mappings().len(), 1);
422 assert_eq!(
423 planned.output_plan().schema_mappings()[0].arrow().name(),
424 "id"
425 );
426 let create_table_sql = planned
427 .output_plan()
428 .create_table_sql()
429 .ok_or("expected create table SQL")?;
430 assert!(create_table_sql.contains("[dbo].[derived_orders]"));
431 assert_eq!(
432 planned
433 .phase_timings()
434 .iter()
435 .map(crate::PhaseTimingReport::phase_name)
436 .collect::<Vec<_>>(),
437 vec![
438 "lazy_sql_planning",
439 OUTPUT_SCHEMA_PLANNING_PHASE,
440 SQL_TARGET_PLANNING_PHASE
441 ]
442 );
443 Ok(())
444 }
445
446 #[tokio::test]
447 async fn plan_mssql_output_accepts_registered_derived_alias_handle()
448 -> Result<(), Box<dyn std::error::Error>> {
449 let table = DeltaLogTable::new("orders")?;
450 let mut session = DeltaFunnelSession::new(
451 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
452 )?;
453 session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
454 let derived = session
455 .table_from_sql("select customer_name from orders")
456 .await?;
457 let alias = session.register_alias("customer_names", &derived)?;
458 let request = output_request(
459 alias.clone(),
460 "customer_names_output",
461 "customer_names_sink",
462 LoadMode::AppendExisting,
463 )?;
464
465 let planned = session.plan_mssql_output(&request)?;
466
467 assert_eq!(planned.table(), &alias);
468 assert_eq!(planned.output_plan().schema_mappings().len(), 1);
469 assert_eq!(
470 planned.output_plan().schema_mappings()[0].arrow().name(),
471 "customer_name"
472 );
473 Ok(())
474 }
475
476 #[test]
477 fn plan_mssql_output_connection_override_wins_without_mutating_session_default()
478 -> Result<(), Box<dyn std::error::Error>> {
479 let table = DeltaLogTable::new("orders")?;
480 let mut session = DeltaFunnelSession::new(
481 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
482 )?;
483 let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
484 let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders_sink")?)
485 .with_connection(override_connection()?);
486 let request = OutputWritePlan::new(
487 source,
488 MssqlOutputTarget::new("orders_output", target_config, RunMode::DryRun),
489 );
490
491 let planned = session.plan_mssql_output(&request)?;
492
493 assert_eq!(
494 planned.output_plan().connection_source(),
495 MssqlConnectionSource::TargetOverride
496 );
497 assert_eq!(
498 planned.output_plan().connection().display_label(),
499 Some("warehouse-override")
500 );
501 assert_eq!(
502 session
503 .options()
504 .default_mssql_connection()
505 .ok_or("expected default connection")?
506 .summary()
507 .display_label(),
508 Some("warehouse-primary")
509 );
510 Ok(())
511 }
512
513 #[test]
514 fn plan_mssql_output_missing_effective_connection_fails_before_side_effects()
515 -> Result<(), Box<dyn std::error::Error>> {
516 let table = DeltaLogTable::new("orders")?;
517 let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
518 let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
519 let request = execute_output_request(
520 source,
521 "orders_output",
522 "orders_sink",
523 LoadMode::AppendExisting,
524 )?;
525
526 let error = session.plan_mssql_output(&request);
527
528 assert!(matches!(
529 error,
530 Err(DeltaFunnelError::MissingMssqlConnection { output_name })
531 if output_name == "orders_output"
532 ));
533 Ok(())
534 }
535
536 #[test]
537 fn plan_mssql_output_accepts_replace_before_side_effects()
538 -> Result<(), Box<dyn std::error::Error>> {
539 let table = DeltaLogTable::new("orders")?;
540 let mut session = DeltaFunnelSession::new(
541 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
542 )?;
543 let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
544 let request = output_request(source, "orders_output", "orders_sink", LoadMode::Replace)?;
545
546 let planned = session.plan_mssql_output(&request)?;
547
548 assert_eq!(planned.output_plan().load_mode(), LoadMode::Replace);
549 assert_eq!(planned.output_plan().target_table().table(), "orders_sink");
550 assert!(
551 planned
552 .output_plan()
553 .lifecycle_plan()
554 .create_table_sql_required()
555 );
556 Ok(())
557 }
558
559 #[test]
560 fn plan_mssql_output_rejects_unknown_lazy_table_before_target_planning()
561 -> Result<(), DeltaFunnelError> {
562 let session = DeltaFunnelSession::new(
563 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
564 )?;
565 let request = output_request(
566 LazyTable::placeholder(42, LazyTableKind::DeltaSource),
567 "orders_output",
568 "orders_sink",
569 LoadMode::AppendExisting,
570 )?;
571
572 let error = session.plan_mssql_output(&request);
573
574 assert!(matches!(
575 error,
576 Err(DeltaFunnelError::MssqlWorkflowPlanning { message })
577 if message.contains("not registered in this session")
578 ));
579 Ok(())
580 }
581
582 #[test]
583 fn plan_mssql_output_rejects_invalid_output_name() -> Result<(), Box<dyn std::error::Error>> {
584 let table = DeltaLogTable::new("orders")?;
585 let mut session = DeltaFunnelSession::new(
586 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
587 )?;
588 let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
589 let request = output_request(source, " ", "orders_sink", LoadMode::AppendExisting)?;
590
591 let error = session.plan_mssql_output(&request);
592
593 assert!(matches!(
594 error,
595 Err(DeltaFunnelError::InvalidMssqlOutputIdentity { output_name, .. })
596 if output_name == " "
597 ));
598 Ok(())
599 }
600
601 #[test]
602 fn plan_mssql_output_rejects_invalid_target_identifier_before_side_effects()
603 -> Result<(), Box<dyn std::error::Error>> {
604 let table = DeltaLogTable::new("orders")?;
605 let mut session = DeltaFunnelSession::new(
606 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
607 )?;
608 let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
609 let request = output_request(
610 source,
611 "orders_output",
612 "orders\narchive",
613 LoadMode::CreateAndLoad,
614 )?;
615
616 let error = session.plan_mssql_output(&request);
617
618 assert!(matches!(
619 &error,
620 Err(DeltaFunnelError::MssqlDdlTargetIdentifier { output_name, .. })
621 if output_name == "orders_output"
622 ));
623 let display = error.err().ok_or("expected error")?.to_string();
624 assert!(!display.contains('\n'));
625 assert!(display.contains("control characters"));
626 Ok(())
627 }
628
629 #[test]
630 fn plan_mssql_output_reports_unsupported_source_schema()
631 -> Result<(), Box<dyn std::error::Error>> {
632 let table =
633 DeltaLogTable::new_with_schema("unsupported-schema", UNSUPPORTED_SCHEMA_FIELDS_JSON)?;
634 let mut session = DeltaFunnelSession::new(
635 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
636 )?;
637 let source =
638 session.delta_lake(DeltaSourceConfig::new("unsupported_schema", table.uri()))?;
639 let request = output_request(
640 source,
641 "unsupported_output",
642 "unsupported_sink",
643 LoadMode::AppendExisting,
644 )?;
645
646 let error = session.plan_mssql_output(&request);
647
648 assert!(matches!(
649 error,
650 Err(DeltaFunnelError::MssqlSchemaPlanning {
651 output_name,
652 diagnostics,
653 }) if output_name == "unsupported_output" && !diagnostics.is_empty()
654 ));
655 Ok(())
656 }
657
658 #[test]
659 fn planned_mssql_output_debug_redacts_connection_material()
660 -> Result<(), Box<dyn std::error::Error>> {
661 let table = DeltaLogTable::new("orders")?;
662 let mut session = DeltaFunnelSession::new(
663 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
664 )?;
665 let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
666 let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders_sink")?)
667 .with_connection(override_connection()?);
668 let request = OutputWritePlan::new(
669 source,
670 MssqlOutputTarget::new("orders_output", target_config, RunMode::DryRun),
671 );
672
673 let planned = session.plan_mssql_output(&request)?;
674 let debug = format!("{planned:?}");
675
676 assert!(debug.contains("orders_output"));
677 assert!(debug.contains("warehouse-override"));
678 assert!(!debug.contains("secret-token"));
679 assert!(!debug.contains("override-secret"));
680 assert!(!debug.contains("password"));
681 assert!(!debug.contains("server=tcp"));
682 Ok(())
683 }
684
685 #[tokio::test]
686 async fn write_to_mssql_requires_effective_connection_before_stream_setup()
687 -> Result<(), Box<dyn std::error::Error>> {
688 let table = DeltaLogTable::new("orders")?;
689 let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
690 let source = session.delta_lake(DeltaSourceConfig::new("orders", table.uri()))?;
691 let request = execute_output_request(
692 source,
693 "orders_output",
694 "orders_sink",
695 LoadMode::AppendExisting,
696 )?;
697
698 let error = session.write_to_mssql(&request).await;
699
700 assert!(matches!(
701 error,
702 Err(DeltaFunnelError::MissingMssqlConnection { output_name })
703 if output_name == "orders_output"
704 ));
705 Ok(())
706 }
707
708 #[tokio::test]
709 async fn write_to_mssql_with_writer_hands_query_stream_to_one_output_boundary()
710 -> Result<(), Box<dyn std::error::Error>> {
711 let mut session = DeltaFunnelSession::new(
712 SessionOptions::new()
713 .with_default_mssql_connection(secret_connection()?)
714 .with_validation_options(
715 ValidationOptions::new()
716 .with_target_validation_mode(TargetValidationMode::Require),
717 ),
718 )?;
719 let derived = session
720 .table_from_sql("select 1 as id union all select 2 as id")
721 .await?;
722 let request = execute_output_request(
723 derived,
724 "orders_output",
725 "orders_sink",
726 LoadMode::CreateAndLoad,
727 )?;
728 let mut writer = FakeOrchestratorWriter::default();
729
730 let report = session
731 .write_to_mssql_with_writer(&request, &mut writer)
732 .await?;
733
734 assert_eq!(writer.calls.len(), 1);
735 let call = writer.calls.first().ok_or("expected fake writer call")?;
736 assert_eq!(call.output_name, "orders_output");
737 assert_eq!(call.target_table.schema(), Some("dbo"));
738 assert_eq!(call.target_table.table(), "orders_sink");
739 assert_eq!(
740 call.connection_source,
741 MssqlConnectionSource::ContextDefault
742 );
743 assert_eq!(call.rows, 2);
744 assert!(call.batches >= 1);
745 assert_eq!(call.schema_fields, 1);
746 assert_eq!(
747 call.validation_options.target_validation_mode(),
748 TargetValidationMode::Require
749 );
750 assert_eq!(report.output_name(), "orders_output");
751 assert_eq!(report.stats().rows_written(), 2);
752 assert_eq!(report.stats().batches_written(), call.batches);
753 assert_eq!(report.cleanup(), MssqlTargetCleanupStatus::NotApplicable);
754 assert_eq!(
755 report
756 .phase_timings()
757 .iter()
758 .take(3)
759 .map(crate::PhaseTimingReport::phase_name)
760 .collect::<Vec<_>>(),
761 vec![
762 "lazy_sql_planning",
763 OUTPUT_SCHEMA_PLANNING_PHASE,
764 SQL_TARGET_PLANNING_PHASE
765 ]
766 );
767 let validation_timing = report
768 .phase_timings()
769 .iter()
770 .find(|timing| timing.phase_name() == VALIDATION_PHASE)
771 .ok_or("expected validation phase timing")?;
772 assert_eq!(
773 validation_timing.status(),
774 crate::PhaseStatus::not_started(crate::ReportReasonCode::NotExecuted)
775 );
776 Ok(())
777 }
778
779 #[tokio::test]
780 async fn write_to_mssql_with_writer_executes_real_delta_source_fixture()
781 -> Result<(), Box<dyn std::error::Error>> {
782 let table = RealParquetDeltaTable::new_default("orders")?;
783 let mut session = DeltaFunnelSession::new(
784 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
785 )?;
786 session.delta_lake(DeltaSourceConfig::new(
787 "orders",
788 table.path().to_string_lossy().to_string(),
789 ))?;
790 let selected_orders = session
791 .table_from_sql("select id, customer_name from orders")
792 .await?;
793 let request = execute_output_request(
794 selected_orders,
795 "orders_output",
796 "orders_sink",
797 LoadMode::AppendExisting,
798 )?;
799 let mut writer = FakeOrchestratorWriter::default();
800
801 let report = session
802 .write_to_mssql_with_writer(&request, &mut writer)
803 .await?;
804
805 assert_eq!(writer.calls.len(), 1);
806 let call = writer.calls.first().ok_or("expected fake writer call")?;
807 assert_eq!(call.output_name, "orders_output");
808 assert_eq!(call.rows, u64::try_from(table.rows())?);
809 assert!(call.batches >= 1);
810 assert_eq!(call.schema_fields, 2);
811 assert_eq!(report.stats().rows_written(), u64::try_from(table.rows())?);
812 assert_eq!(report.stats().batches_written(), call.batches);
813 Ok(())
814 }
815
816 #[tokio::test]
817 async fn write_to_mssql_with_writer_executes_multi_source_delta_join_fixture()
818 -> Result<(), Box<dyn std::error::Error>> {
819 let orders = RealParquetDeltaTable::new_default("orders")?;
820 let customers = RealParquetDeltaTable::new_default("customers")?;
821 let mut session = DeltaFunnelSession::new(
822 SessionOptions::new().with_default_mssql_connection(secret_connection()?),
823 )?;
824 session.delta_lake(DeltaSourceConfig::new(
825 "orders",
826 orders.path().to_string_lossy().to_string(),
827 ))?;
828 session.delta_lake(DeltaSourceConfig::new(
829 "customers",
830 customers.path().to_string_lossy().to_string(),
831 ))?;
832 let joined = session
833 .table_from_sql(
834 "select o.id, c.customer_name \
835 from orders o \
836 join customers c on o.id = c.id",
837 )
838 .await?;
839 let request = execute_output_request(
840 joined,
841 "joined_output",
842 "joined_sink",
843 LoadMode::AppendExisting,
844 )?;
845 let mut writer = FakeOrchestratorWriter::default();
846
847 let report = session
848 .write_to_mssql_with_writer(&request, &mut writer)
849 .await?;
850
851 assert_eq!(writer.calls.len(), 1);
852 let call = writer.calls.first().ok_or("expected fake writer call")?;
853 assert_eq!(call.output_name, "joined_output");
854 assert_eq!(call.rows, 3);
855 assert!(call.batches >= 1);
856 assert_eq!(call.schema_fields, 2);
857 assert_eq!(report.stats().rows_written(), 3);
858 assert_eq!(report.stats().batches_written(), call.batches);
859 Ok(())
860 }
861
862 #[tokio::test]
863 async fn write_to_mssql_rejects_dry_run_before_planning_or_writer()
864 -> Result<(), Box<dyn std::error::Error>> {
865 let mut session = DeltaFunnelSession::new(SessionOptions::default())?;
866 let derived = session.table_from_sql("select 1 as id").await?;
867 let request = output_request(
868 derived,
869 "orders_output",
870 "orders_sink",
871 LoadMode::AppendExisting,
872 )?;
873 let mut writer = FakeOrchestratorWriter::default();
874
875 let error = session
876 .write_to_mssql_with_writer(&request, &mut writer)
877 .await;
878
879 assert!(matches!(
880 error,
881 Err(DeltaFunnelError::MssqlWorkflowPlanning { message })
882 if message.contains("RunMode::Execute")
883 && message.contains("dry_run_to_mssql")
884 ));
885 assert!(writer.calls.is_empty());
886 Ok(())
887 }
888}