delta_funnel/sql_server/planning/
output.rs

1//! SQL Server target output planning.
2
3use arrow_schema::Schema;
4use arrow_tiberius::{PlannedSchema, SchemaMapping};
5
6use crate::DeltaFunnelError;
7
8use super::{
9    LoadMode, MssqlConnectionConfig, MssqlConnectionSource, MssqlConnectionSummary, MssqlDdlPlan,
10    MssqlLifecyclePlan, MssqlSchemaDiagnostic, MssqlSchemaPlan, MssqlSchemaPlanOptions,
11    MssqlTargetConfig, MssqlTargetResolutionContext, MssqlTargetSummary, MssqlTargetTable,
12    ResolvedMssqlTarget, plan_mssql_create_table_ddl, plan_mssql_lifecycle,
13    plan_mssql_output_schema,
14};
15
16/// Complete SQL Server target planning artifact for one selected output.
17#[derive(Debug, Clone, PartialEq, Eq)]
18pub struct MssqlTargetOutputPlan {
19    schema_plan: MssqlSchemaPlan,
20    ddl_plan: MssqlDdlPlan,
21    lifecycle_plan: MssqlLifecyclePlan,
22}
23
24impl MssqlTargetOutputPlan {
25    /// Returns the redacted resolved target summary.
26    #[must_use]
27    pub fn target(&self) -> &MssqlTargetSummary {
28        self.schema_plan.target()
29    }
30
31    /// Returns the selected output name.
32    #[must_use]
33    pub fn output_name(&self) -> &str {
34        self.target().output_name()
35    }
36
37    /// Returns the effective target table.
38    #[must_use]
39    pub fn target_table(&self) -> &MssqlTargetTable {
40        self.target().table()
41    }
42
43    /// Returns the requested target lifecycle mode.
44    #[must_use]
45    pub fn load_mode(&self) -> LoadMode {
46        self.target().load_mode()
47    }
48
49    /// Returns where the effective connection came from.
50    #[must_use]
51    pub fn connection_source(&self) -> MssqlConnectionSource {
52        self.target().connection_source()
53    }
54
55    /// Returns the redacted effective connection summary.
56    #[must_use]
57    pub fn connection(&self) -> &MssqlConnectionSummary {
58        self.target().connection()
59    }
60
61    /// Returns the planned Arrow-to-MSSQL schema mapping.
62    #[must_use]
63    pub fn schema_plan(&self) -> &MssqlSchemaPlan {
64        &self.schema_plan
65    }
66
67    /// Returns the Arrow-to-MSSQL planning options used for this output.
68    #[must_use]
69    pub fn schema_plan_options(&self) -> MssqlSchemaPlanOptions {
70        self.schema_plan.plan_options()
71    }
72
73    /// Returns the profile-bound schema plan passed to arrow-tiberius writers.
74    #[must_use]
75    pub fn planned_schema(&self) -> &PlannedSchema {
76        self.schema_plan.planned_schema()
77    }
78
79    /// Returns planned Arrow-to-MSSQL column mappings in output field order.
80    #[must_use]
81    pub fn schema_mappings(&self) -> &[SchemaMapping] {
82        self.schema_plan.mappings()
83    }
84
85    /// Returns structured schema diagnostic report entries.
86    #[must_use]
87    pub fn schema_diagnostics(&self) -> &[MssqlSchemaDiagnostic] {
88        self.schema_plan.diagnostic_reports()
89    }
90
91    /// Returns the planned SQL Server DDL artifact.
92    #[must_use]
93    pub fn ddl_plan(&self) -> &MssqlDdlPlan {
94        &self.ddl_plan
95    }
96
97    /// Returns concrete planned `CREATE TABLE` SQL when the target name is known at planning time.
98    #[must_use]
99    pub fn create_table_sql(&self) -> Option<&str> {
100        self.ddl_plan.create_table_sql()
101    }
102
103    /// Returns the planned SQL Server lifecycle report.
104    #[must_use]
105    pub fn lifecycle_plan(&self) -> &MssqlLifecyclePlan {
106        &self.lifecycle_plan
107    }
108}
109
110/// Plans one selected output schema for SQL Server writing.
111///
112/// This is the high-level per-output target planning API. It resolves the
113/// target configuration, plans Arrow-to-MSSQL schema mappings, plans optional
114/// create-table DDL, and attaches the lifecycle report. It performs no SQL
115/// Server I/O, DataFusion execution, row reads, or batch handoff polling.
116pub fn plan_mssql_target_for_output(
117    output_schema: impl AsRef<Schema>,
118    output_name: &str,
119    target_config: &MssqlTargetConfig,
120    default_connection: Option<&MssqlConnectionConfig>,
121    options: MssqlSchemaPlanOptions,
122) -> Result<MssqlTargetOutputPlan, DeltaFunnelError> {
123    let target = target_config.resolve(MssqlTargetResolutionContext {
124        output_name: Some(output_name),
125        default_connection,
126    })?;
127
128    plan_mssql_target_for_resolved_output(output_schema, &target, options)
129}
130
131/// Plans one selected output schema for an already resolved SQL Server target.
132///
133/// Use this when an orchestrator needs to retain the resolved target for later
134/// SQL Server connection setup while still reusing the lower-level schema, DDL,
135/// and lifecycle planning path.
136pub fn plan_mssql_target_for_resolved_output(
137    output_schema: impl AsRef<Schema>,
138    target: &ResolvedMssqlTarget,
139    options: MssqlSchemaPlanOptions,
140) -> Result<MssqlTargetOutputPlan, DeltaFunnelError> {
141    let schema_plan = plan_mssql_output_schema(output_schema, target, options)?;
142
143    plan_mssql_target_output(schema_plan)
144}
145
146/// Builds a complete target output plan from an existing schema plan.
147///
148/// This function only composes planning reports. It performs no I/O or
149/// execution, and it returns no partial target output plan on failure.
150pub fn plan_mssql_target_output(
151    schema_plan: MssqlSchemaPlan,
152) -> Result<MssqlTargetOutputPlan, DeltaFunnelError> {
153    let ddl_plan = plan_mssql_create_table_ddl(&schema_plan)?;
154    let lifecycle_plan = plan_mssql_lifecycle(&schema_plan, Some(&ddl_plan))?;
155
156    Ok(MssqlTargetOutputPlan {
157        schema_plan,
158        ddl_plan,
159        lifecycle_plan,
160    })
161}
162
163#[cfg(test)]
164mod tests {
165    use arrow_schema::{DataType, Field, Schema};
166    use arrow_tiberius::{DiagnosticCode, DiagnosticSeverity, PlanOptions};
167
168    use super::*;
169    use crate::{
170        MssqlConnectionConfig, MssqlConnectionSource, MssqlLifecycleExecutionGuardrail,
171        MssqlTargetConfig, MssqlTargetResolutionContext, MssqlTargetTable, MssqlTargetTableState,
172        plan_mssql_output_schema,
173    };
174
175    fn secret_connection() -> Result<MssqlConnectionConfig, DeltaFunnelError> {
176        Ok(MssqlConnectionConfig::new(
177            "server=tcp:sql.example.com;database=warehouse;user=admin;password=secret-token",
178        )?
179        .with_display_label("warehouse-primary"))
180    }
181
182    fn schema_plan(
183        output_name: &str,
184        load_mode: LoadMode,
185        table: MssqlTargetTable,
186    ) -> Result<MssqlSchemaPlan, DeltaFunnelError> {
187        let connection = secret_connection()?;
188        let target = MssqlTargetConfig::new(table)
189            .with_load_mode(load_mode)
190            .resolve(MssqlTargetResolutionContext {
191                output_name: Some(output_name),
192                default_connection: Some(&connection),
193            })?;
194        let schema = Schema::new(vec![
195            Field::new("order_id", DataType::Int64, false),
196            Field::new("status", DataType::Utf8, true),
197        ]);
198
199        plan_mssql_output_schema(&schema, &target, PlanOptions::default())
200    }
201
202    fn orders_schema() -> Schema {
203        Schema::new(vec![
204            Field::new("order_id", DataType::Int64, false),
205            Field::new("status", DataType::Utf8, true),
206        ])
207    }
208
209    fn west_output_schema() -> Schema {
210        Schema::new(vec![
211            Field::new("west_order_id", DataType::Int64, false),
212            Field::new("west_customer", DataType::Utf8, true),
213        ])
214    }
215
216    fn east_output_schema() -> Schema {
217        Schema::new(vec![
218            Field::new("east_order_id", DataType::Int64, false),
219            Field::new("east_total", DataType::Float64, true),
220        ])
221    }
222
223    fn duplicate_field_schema() -> Schema {
224        Schema::new(vec![
225            Field::new("id", DataType::Int64, false),
226            Field::new("id", DataType::Utf8, true),
227        ])
228    }
229
230    fn unsupported_field_schema() -> Schema {
231        Schema::new(vec![Field::new(
232            "items",
233            DataType::new_list(DataType::Int64, true),
234            true,
235        )])
236    }
237
238    #[test]
239    fn high_level_api_plans_append_existing_output() -> Result<(), DeltaFunnelError> {
240        let default_connection = secret_connection()?;
241        let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders")?);
242
243        let output_plan = plan_mssql_target_for_output(
244            orders_schema(),
245            "orders_output",
246            &target_config,
247            Some(&default_connection),
248            PlanOptions::default(),
249        )?;
250
251        assert_eq!(output_plan.target().output_name(), "orders_output");
252        assert_eq!(output_plan.target().table().schema(), Some("dbo"));
253        assert_eq!(output_plan.target().table().table(), "orders");
254        assert_eq!(output_plan.target().load_mode(), LoadMode::AppendExisting);
255        assert_eq!(
256            output_plan.target().connection_source(),
257            MssqlConnectionSource::ContextDefault
258        );
259        assert_eq!(output_plan.ddl_plan().create_table_sql(), None);
260        assert_eq!(
261            output_plan.lifecycle_plan().expected_target_state(),
262            MssqlTargetTableState::Exists
263        );
264        Ok(())
265    }
266
267    #[test]
268    fn high_level_api_plans_create_and_load_output() -> Result<(), DeltaFunnelError> {
269        let default_connection = secret_connection()?;
270        let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders")?)
271            .with_load_mode(LoadMode::CreateAndLoad);
272
273        let output_plan = plan_mssql_target_for_output(
274            orders_schema(),
275            "orders_output",
276            &target_config,
277            Some(&default_connection),
278            PlanOptions::default(),
279        )?;
280
281        assert_eq!(output_plan.target().load_mode(), LoadMode::CreateAndLoad);
282        assert!(
283            output_plan
284                .ddl_plan()
285                .create_table_sql()
286                .unwrap_or_default()
287                .starts_with("CREATE TABLE [dbo].[orders]")
288        );
289        assert_eq!(
290            output_plan.lifecycle_plan().expected_target_state(),
291            MssqlTargetTableState::Absent
292        );
293        Ok(())
294    }
295
296    #[test]
297    fn composed_report_exposes_structured_python_ready_fields() -> Result<(), DeltaFunnelError> {
298        let default_connection = secret_connection()?;
299        let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders")?)
300            .with_load_mode(LoadMode::CreateAndLoad);
301
302        let output_plan = plan_mssql_target_for_output(
303            orders_schema(),
304            "orders_output",
305            &target_config,
306            Some(&default_connection),
307            PlanOptions::default(),
308        )?;
309
310        assert_eq!(output_plan.output_name(), "orders_output");
311        assert_eq!(output_plan.target_table().schema(), Some("dbo"));
312        assert_eq!(output_plan.target_table().table(), "orders");
313        assert_eq!(output_plan.load_mode(), LoadMode::CreateAndLoad);
314        assert_eq!(
315            output_plan.connection_source(),
316            MssqlConnectionSource::ContextDefault
317        );
318        assert_eq!(
319            output_plan.connection().display_label(),
320            Some("warehouse-primary")
321        );
322        assert_eq!(output_plan.schema_mappings()[0].arrow().name(), "order_id");
323        assert_eq!(output_plan.schema_mappings()[1].arrow().name(), "status");
324        assert_eq!(output_plan.schema_plan_options(), PlanOptions::default());
325        assert!(output_plan.schema_diagnostics().is_empty());
326        assert!(
327            output_plan
328                .create_table_sql()
329                .unwrap_or_default()
330                .starts_with("CREATE TABLE [dbo].[orders]")
331        );
332        assert_eq!(
333            output_plan.lifecycle_plan().expected_target_state(),
334            MssqlTargetTableState::Absent
335        );
336        Ok(())
337    }
338
339    #[test]
340    fn output_plan_exposes_schema_plan_options() -> Result<(), DeltaFunnelError> {
341        let default_connection = secret_connection()?;
342        let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders")?);
343        let options = PlanOptions {
344            string_policy: arrow_tiberius::StringPolicy::NVarChar(128),
345            ..PlanOptions::default()
346        };
347
348        let output_plan = plan_mssql_target_for_output(
349            orders_schema(),
350            "orders_output",
351            &target_config,
352            Some(&default_connection),
353            options,
354        )?;
355
356        assert_eq!(output_plan.schema_plan_options(), options);
357        Ok(())
358    }
359
360    #[test]
361    fn high_level_api_returns_structured_missing_connection_error() -> Result<(), DeltaFunnelError>
362    {
363        let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders")?);
364
365        let error = plan_mssql_target_for_output(
366            orders_schema(),
367            "orders_output",
368            &target_config,
369            None,
370            PlanOptions::default(),
371        )
372        .err()
373        .ok_or_else(|| DeltaFunnelError::Config {
374            message: "expected missing connection error".to_owned(),
375        })?;
376
377        assert!(matches!(
378            error,
379            DeltaFunnelError::MissingMssqlConnection { .. }
380        ));
381        Ok(())
382    }
383
384    #[test]
385    fn high_level_api_preserves_duplicate_field_error() -> Result<(), DeltaFunnelError> {
386        let default_connection = secret_connection()?;
387        let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders")?);
388
389        let error = plan_mssql_target_for_output(
390            duplicate_field_schema(),
391            "orders_output",
392            &target_config,
393            Some(&default_connection),
394            PlanOptions::default(),
395        )
396        .err()
397        .ok_or_else(|| DeltaFunnelError::Config {
398            message: "expected duplicate field error".to_owned(),
399        })?;
400
401        assert!(matches!(
402            error,
403            DeltaFunnelError::DuplicateMssqlOutputField {
404                output_name,
405                field_name,
406                first_index: 0,
407                duplicate_index: 1,
408            } if output_name == "orders_output" && field_name == "id"
409        ));
410        Ok(())
411    }
412
413    #[test]
414    fn high_level_api_preserves_structured_schema_diagnostics() -> Result<(), DeltaFunnelError> {
415        let default_connection = secret_connection()?;
416        let target_config = MssqlTargetConfig::new(MssqlTargetTable::new("dbo", "orders")?);
417
418        let error = plan_mssql_target_for_output(
419            unsupported_field_schema(),
420            "orders_output",
421            &target_config,
422            Some(&default_connection),
423            PlanOptions::default(),
424        )
425        .err()
426        .ok_or_else(|| DeltaFunnelError::Config {
427            message: "expected schema planning error".to_owned(),
428        })?;
429
430        let DeltaFunnelError::MssqlSchemaPlanning {
431            output_name,
432            diagnostics,
433        } = error
434        else {
435            return Err(DeltaFunnelError::Config {
436                message: "expected MssqlSchemaPlanning error".to_owned(),
437            });
438        };
439
440        assert_eq!(output_name, "orders_output");
441        assert_eq!(diagnostics.len(), 1);
442        let diagnostic = &diagnostics.all()[0];
443        assert_eq!(diagnostic.severity(), DiagnosticSeverity::Error);
444        assert_eq!(diagnostic.code(), DiagnosticCode::UnsupportedArrowType);
445        let field = diagnostic.field().ok_or_else(|| DeltaFunnelError::Config {
446            message: "expected field diagnostic context".to_owned(),
447        })?;
448        assert_eq!(field.index(), 0);
449        assert_eq!(field.name(), "items");
450        Ok(())
451    }
452
453    #[test]
454    fn high_level_api_plans_outputs_independently() -> Result<(), DeltaFunnelError> {
455        let default_connection = secret_connection()?.with_display_label("context-default");
456        let east_connection = MssqlConnectionConfig::new(
457            "server=tcp:sql.example.com;database=east;user=admin;password=east-secret",
458        )?
459        .with_display_label("east-override");
460        let west_target = MssqlTargetConfig::new(MssqlTargetTable::new("reporting", "west")?);
461        let east_target = MssqlTargetConfig::new(MssqlTargetTable::new("warehouse", "east")?)
462            .with_load_mode(LoadMode::CreateAndLoad)
463            .with_connection(east_connection);
464
465        let west_plan = plan_mssql_target_for_output(
466            west_output_schema(),
467            "west_output",
468            &west_target,
469            Some(&default_connection),
470            PlanOptions::default(),
471        )?;
472        let east_plan = plan_mssql_target_for_output(
473            east_output_schema(),
474            "east_output",
475            &east_target,
476            Some(&default_connection),
477            PlanOptions::default(),
478        )?;
479
480        assert_eq!(west_plan.target().output_name(), "west_output");
481        assert_eq!(west_plan.target().table().schema(), Some("reporting"));
482        assert_eq!(west_plan.target().table().table(), "west");
483        assert_eq!(west_plan.target().load_mode(), LoadMode::AppendExisting);
484        assert_eq!(
485            west_plan.target().connection_source(),
486            MssqlConnectionSource::ContextDefault
487        );
488        assert_eq!(west_plan.ddl_plan().create_table_sql(), None);
489        assert_eq!(
490            west_plan.schema_mappings()[0].arrow().name(),
491            "west_order_id"
492        );
493        assert_eq!(
494            west_plan.schema_mappings()[1].arrow().name(),
495            "west_customer"
496        );
497
498        assert_eq!(east_plan.target().output_name(), "east_output");
499        assert_eq!(east_plan.target().table().schema(), Some("warehouse"));
500        assert_eq!(east_plan.target().table().table(), "east");
501        assert_eq!(east_plan.target().load_mode(), LoadMode::CreateAndLoad);
502        assert_eq!(
503            east_plan.target().connection_source(),
504            MssqlConnectionSource::TargetOverride
505        );
506        assert!(
507            east_plan
508                .create_table_sql()
509                .unwrap_or_default()
510                .starts_with("CREATE TABLE [warehouse].[east]")
511        );
512        assert_eq!(
513            east_plan.schema_mappings()[0].arrow().name(),
514            "east_order_id"
515        );
516        assert_eq!(east_plan.schema_mappings()[1].arrow().name(), "east_total");
517
518        let debug = format!("{west_plan:?}\n{east_plan:?}");
519        assert!(debug.contains("context-default"));
520        assert!(debug.contains("east-override"));
521        assert!(!debug.contains("secret-token"));
522        assert!(!debug.contains("east-secret"));
523        assert!(!debug.contains("password"));
524        Ok(())
525    }
526
527    #[test]
528    fn append_existing_output_plan_includes_lifecycle_report() -> Result<(), DeltaFunnelError> {
529        let schema_plan = schema_plan(
530            "orders_output",
531            LoadMode::AppendExisting,
532            MssqlTargetTable::new("dbo", "orders")?,
533        )?;
534
535        let output_plan = plan_mssql_target_output(schema_plan)?;
536
537        assert_eq!(
538            output_plan.schema_plan().target().output_name(),
539            "orders_output"
540        );
541        assert!(!output_plan.ddl_plan().create_table_sql_present());
542        assert_eq!(output_plan.ddl_plan().create_table_sql(), None);
543        assert_eq!(
544            output_plan.lifecycle_plan().expected_target_state(),
545            MssqlTargetTableState::Exists
546        );
547        assert!(output_plan.lifecycle_plan().executable_in_mvp());
548        Ok(())
549    }
550
551    #[test]
552    fn create_and_load_output_plan_includes_ddl_and_lifecycle_report()
553    -> Result<(), DeltaFunnelError> {
554        let schema_plan = schema_plan(
555            "orders_output",
556            LoadMode::CreateAndLoad,
557            MssqlTargetTable::new("dbo", "orders")?,
558        )?;
559
560        let output_plan = plan_mssql_target_output(schema_plan)?;
561
562        assert!(output_plan.ddl_plan().create_table_sql_present());
563        assert!(
564            output_plan
565                .ddl_plan()
566                .create_table_sql()
567                .unwrap_or_default()
568                .starts_with("CREATE TABLE [dbo].[orders]")
569        );
570        assert_eq!(
571            output_plan.lifecycle_plan().expected_target_state(),
572            MssqlTargetTableState::Absent
573        );
574        assert!(
575            output_plan
576                .lifecycle_plan()
577                .execution_guardrails()
578                .contains(&MssqlLifecycleExecutionGuardrail::CreateTableDdlExecution)
579        );
580        Ok(())
581    }
582
583    #[test]
584    fn replace_output_plan_includes_ddl_and_lifecycle_report() -> Result<(), DeltaFunnelError> {
585        let schema_plan = schema_plan(
586            "orders_output",
587            LoadMode::Replace,
588            MssqlTargetTable::new("dbo", "orders")?,
589        )?;
590
591        let output_plan = plan_mssql_target_output(schema_plan)?;
592
593        assert_eq!(output_plan.load_mode(), LoadMode::Replace);
594        assert!(output_plan.ddl_plan().create_table_sql_present());
595        assert_eq!(output_plan.ddl_plan().create_table_sql(), None);
596        assert_eq!(
597            output_plan.lifecycle_plan().expected_target_state(),
598            MssqlTargetTableState::ExistsOrAbsent
599        );
600        assert!(output_plan.lifecycle_plan().create_table_sql_required());
601        Ok(())
602    }
603
604    #[test]
605    fn output_plan_reports_do_not_expose_connection_secrets() -> Result<(), DeltaFunnelError> {
606        let schema_plan = schema_plan(
607            "orders_output",
608            LoadMode::CreateAndLoad,
609            MssqlTargetTable::new("dbo", "orders")?,
610        )?;
611
612        let output_plan = plan_mssql_target_output(schema_plan)?;
613        let debug = format!("{output_plan:?}");
614
615        assert!(!debug.contains("secret-token"));
616        assert!(!debug.contains("password"));
617        assert!(!debug.contains("server=tcp"));
618        assert!(debug.contains("warehouse-primary"));
619        Ok(())
620    }
621}
delta_funnel/sql_server/planning/output.rs

delta_funnel/sql_server/planning/
output.rs
