delta_funnel/orchestrator/session/
options.rs

1use std::fmt;
2
3use crate::{
4    DeltaFunnelError, DeltaProviderScanExecutionOptions, MssqlConnectionConfig,
5    MssqlSchemaPlanOptions, MssqlWorkflowWriteOptions, MssqlWriteOptions, QueryOptions,
6    ValidationOptions, default_mssql_write_options,
7};
8
9/// Session-wide options for lazy query-load orchestration.
10#[derive(Clone)]
11pub struct SessionOptions {
12    query_options: QueryOptions,
13    provider_scan_options: DeltaProviderScanExecutionOptions,
14    mssql_schema_options: MssqlSchemaPlanOptions,
15    mssql_write_options: MssqlWriteOptions,
16    mssql_workflow_options: MssqlWorkflowWriteOptions,
17    validation_options: ValidationOptions,
18    default_mssql_connection: Option<MssqlConnectionConfig>,
19}
20
21impl Default for SessionOptions {
22    fn default() -> Self {
23        Self {
24            query_options: QueryOptions::default(),
25            provider_scan_options: DeltaProviderScanExecutionOptions::default(),
26            mssql_schema_options: MssqlSchemaPlanOptions::default(),
27            mssql_write_options: default_mssql_write_options(),
28            mssql_workflow_options: MssqlWorkflowWriteOptions::default(),
29            validation_options: ValidationOptions::default(),
30            default_mssql_connection: None,
31        }
32    }
33}
34
35impl SessionOptions {
36    /// Creates default session options.
37    #[must_use]
38    pub fn new() -> Self {
39        Self::default()
40    }
41
42    /// Sets DataFusion query execution options.
43    #[must_use]
44    pub const fn with_query_options(mut self, query_options: QueryOptions) -> Self {
45        self.query_options = query_options;
46        self
47    }
48
49    /// Sets Delta provider scan execution options.
50    #[must_use]
51    pub const fn with_provider_scan_options(
52        mut self,
53        provider_scan_options: DeltaProviderScanExecutionOptions,
54    ) -> Self {
55        self.provider_scan_options = provider_scan_options;
56        self
57    }
58
59    /// Sets SQL Server schema planning options.
60    #[must_use]
61    pub const fn with_mssql_schema_options(
62        mut self,
63        mssql_schema_options: MssqlSchemaPlanOptions,
64    ) -> Self {
65        self.mssql_schema_options = mssql_schema_options;
66        self
67    }
68
69    /// Sets SQL Server write options.
70    #[must_use]
71    pub const fn with_mssql_write_options(
72        mut self,
73        mssql_write_options: MssqlWriteOptions,
74    ) -> Self {
75        self.mssql_write_options = mssql_write_options;
76        self
77    }
78
79    /// Sets SQL Server multi-output workflow options.
80    #[must_use]
81    pub const fn with_mssql_workflow_options(
82        mut self,
83        mssql_workflow_options: MssqlWorkflowWriteOptions,
84    ) -> Self {
85        self.mssql_workflow_options = mssql_workflow_options;
86        self
87    }
88
89    /// Sets locally checkable validation options.
90    #[must_use]
91    pub const fn with_validation_options(mut self, validation_options: ValidationOptions) -> Self {
92        self.validation_options = validation_options;
93        self
94    }
95
96    /// Sets the session-level default SQL Server connection.
97    #[must_use]
98    pub fn with_default_mssql_connection(
99        mut self,
100        default_mssql_connection: MssqlConnectionConfig,
101    ) -> Self {
102        self.default_mssql_connection = Some(default_mssql_connection);
103        self
104    }
105
106    /// Returns DataFusion query execution options.
107    #[must_use]
108    pub const fn query_options(&self) -> QueryOptions {
109        self.query_options
110    }
111
112    /// Returns Delta provider scan execution options.
113    #[must_use]
114    pub const fn provider_scan_options(&self) -> DeltaProviderScanExecutionOptions {
115        self.provider_scan_options
116    }
117
118    /// Returns SQL Server schema planning options.
119    #[must_use]
120    pub const fn mssql_schema_options(&self) -> MssqlSchemaPlanOptions {
121        self.mssql_schema_options
122    }
123
124    /// Returns SQL Server write options.
125    #[must_use]
126    pub const fn mssql_write_options(&self) -> MssqlWriteOptions {
127        self.mssql_write_options
128    }
129
130    /// Returns SQL Server multi-output workflow options.
131    #[must_use]
132    pub const fn mssql_workflow_options(&self) -> MssqlWorkflowWriteOptions {
133        self.mssql_workflow_options
134    }
135
136    /// Returns locally checkable validation options.
137    #[must_use]
138    pub const fn validation_options(&self) -> ValidationOptions {
139        self.validation_options
140    }
141
142    /// Returns the optional session-level default SQL Server connection.
143    #[must_use]
144    pub fn default_mssql_connection(&self) -> Option<&MssqlConnectionConfig> {
145        self.default_mssql_connection.as_ref()
146    }
147
148    /// Validates local options before workflow side effects.
149    ///
150    /// # Errors
151    ///
152    /// Returns the first validation error from query, provider, workflow, or
153    /// local validation options.
154    pub fn validate(&self) -> Result<(), DeltaFunnelError> {
155        self.query_options.validate()?;
156        self.provider_scan_options.validate()?;
157        self.mssql_workflow_options.validate()?;
158        self.validation_options.validate()?;
159        Ok(())
160    }
161}
162
163impl fmt::Debug for SessionOptions {
164    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
165        formatter
166            .debug_struct("SessionOptions")
167            .field("query_options", &self.query_options)
168            .field("provider_scan_options", &self.provider_scan_options)
169            .field("mssql_schema_options", &self.mssql_schema_options)
170            .field("mssql_write_options", &self.mssql_write_options)
171            .field("mssql_workflow_options", &self.mssql_workflow_options)
172            .field("validation_options", &self.validation_options)
173            .field(
174                "default_mssql_connection",
175                &self
176                    .default_mssql_connection
177                    .as_ref()
178                    .map(MssqlConnectionConfig::summary),
179            )
180            .finish()
181    }
182}
183
184#[cfg(test)]
185mod tests {
186    use crate::{
187        BatchPipelinePhase, DeltaFunnelError, DeltaProviderReaderBackend,
188        DeltaProviderScanExecutionOptions, MssqlConnectionConfig, MssqlWorkflowWriteOptions,
189        QueryOptions,
190    };
191
192    use super::SessionOptions;
193    use crate::orchestrator::DeltaFunnelSession;
194
195    #[test]
196    fn default_session_constructs_datafusion_context() -> Result<(), DeltaFunnelError> {
197        let session = DeltaFunnelSession::new(SessionOptions::default())?;
198
199        assert_eq!(session.options().query_options(), QueryOptions::default());
200        assert!(
201            session
202                .options()
203                .validation_options()
204                .require_successful_planning()
205        );
206        assert_eq!(
207            session.context().state().config().target_partitions(),
208            datafusion::prelude::SessionConfig::new().target_partitions()
209        );
210        assert_eq!(session.next_table_id(), 0);
211        Ok(())
212    }
213
214    #[test]
215    fn session_applies_query_options_to_datafusion_context() -> Result<(), DeltaFunnelError> {
216        let session =
217            DeltaFunnelSession::new(SessionOptions::new().with_query_options(QueryOptions {
218                target_partitions: Some(3),
219                output_batch_size: Some(11),
220            }))?;
221
222        assert_eq!(session.context().state().config().target_partitions(), 3);
223        assert_eq!(session.context().state().config().batch_size(), 11);
224        Ok(())
225    }
226
227    #[test]
228    fn query_option_validation_failure_reaches_session_construction() {
229        let error =
230            DeltaFunnelSession::new(SessionOptions::new().with_query_options(QueryOptions {
231                target_partitions: Some(0),
232                output_batch_size: None,
233            }));
234
235        assert!(matches!(
236            error,
237            Err(DeltaFunnelError::BatchPipeline {
238                phase: BatchPipelinePhase::Configuration,
239                option: "target_partitions",
240                ..
241            })
242        ));
243    }
244
245    #[test]
246    fn provider_option_validation_failure_reaches_session_construction() {
247        let error = DeltaFunnelSession::new(SessionOptions::new().with_provider_scan_options(
248            DeltaProviderScanExecutionOptions {
249                reader_backend: DeltaProviderReaderBackend::OfficialKernel,
250                max_concurrent_file_reads_per_scan: Some(1),
251                max_concurrent_file_reads_per_partition: 1,
252                output_buffer_capacity_per_partition: 0,
253                native_async_prefetch_file_count_per_partition: 0,
254            },
255        ));
256
257        assert!(matches!(error, Err(DeltaFunnelError::Config { .. })));
258    }
259
260    #[test]
261    fn workflow_parallelism_validation_failure_reaches_session_construction() {
262        let error = DeltaFunnelSession::new(SessionOptions::new().with_mssql_workflow_options(
263            MssqlWorkflowWriteOptions::new().with_max_parallel_outputs(2),
264        ));
265
266        assert!(matches!(
267            error,
268            Err(DeltaFunnelError::MssqlWorkflowPlanning { message })
269                if message.contains("parallel MSSQL output writers are not supported")
270        ));
271    }
272
273    #[test]
274    fn workflow_zero_parallelism_validation_failure_reaches_session_construction() {
275        let error = DeltaFunnelSession::new(SessionOptions::new().with_mssql_workflow_options(
276            MssqlWorkflowWriteOptions::new().with_max_parallel_outputs(0),
277        ));
278
279        assert!(matches!(
280            error,
281            Err(DeltaFunnelError::MssqlWorkflowPlanning { message })
282                if message.contains("max_parallel_outputs must be at least 1")
283        ));
284    }
285
286    #[test]
287    fn session_debug_redacts_default_mssql_connection() -> Result<(), DeltaFunnelError> {
288        let connection = MssqlConnectionConfig::new(
289            "server=tcp:sql.example.com;database=warehouse;user=admin;password=secret-token",
290        )?
291        .with_display_label("warehouse-primary");
292        let session = DeltaFunnelSession::new(
293            SessionOptions::new().with_default_mssql_connection(connection),
294        )?;
295
296        let debug = format!("{session:?}");
297        assert!(debug.contains("warehouse-primary"));
298        assert!(!debug.contains("secret-token"));
299        assert!(!debug.contains("password"));
300        assert!(!debug.contains("server=tcp"));
301        Ok(())
302    }
303}
delta_funnel/orchestrator/session/options.rs

delta_funnel/orchestrator/session/
options.rs
