delta_funnel/
error.rs

1//! Shared error pattern for DeltaFunnel.
2
3use crate::BatchPipelinePhase;
4use crate::support::{sanitize_text_for_display, sanitize_uri_for_display};
5
6use snafu::Snafu;
7
8/// Phase associated with a Delta scan file read failure.
9#[derive(Debug, Clone, Copy, PartialEq, Eq)]
10pub enum DeltaScanFileReadPhase {
11    /// Parsing the table URI failed.
12    TableUriParsing,
13    /// Converting provider file metadata into kernel file metadata failed.
14    FileMetadataConversion,
15    /// Resolving the Delta add-action path against the table root failed.
16    FilePathResolution,
17    /// Constructing the kernel object-store engine failed.
18    ObjectStoreEngineConstruction,
19    /// Starting a Parquet read failed.
20    ParquetReadSetup,
21    /// Reading a Parquet batch failed.
22    ParquetBatchRead,
23    /// Generating or decoding original row indexes failed.
24    RowIndexGeneration,
25    /// Evaluating a kernel physical predicate failed.
26    PredicateEvaluation,
27    /// Converting kernel engine data into Arrow failed.
28    ArrowConversion,
29    /// Applying a physical-to-logical transform failed.
30    TransformApplication,
31    /// The selected backend cannot read this file task equivalently.
32    UnsupportedReadMode,
33    /// A deletion-vector read was rejected because the requested read mode is not safe.
34    DeletionVectorPredicateRejection,
35    /// Applying a deletion-vector mask failed.
36    DeletionVectorMasking,
37}
38
39impl std::fmt::Display for DeltaScanFileReadPhase {
40    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
41        formatter.write_str(match self {
42            Self::TableUriParsing => "table URI parsing",
43            Self::FileMetadataConversion => "file metadata conversion",
44            Self::FilePathResolution => "file path resolution",
45            Self::ObjectStoreEngineConstruction => "object store engine construction",
46            Self::ParquetReadSetup => "Parquet read setup",
47            Self::ParquetBatchRead => "Parquet batch read",
48            Self::RowIndexGeneration => "row-index generation",
49            Self::PredicateEvaluation => "physical predicate evaluation",
50            Self::ArrowConversion => "Arrow conversion",
51            Self::TransformApplication => "physical-to-logical transform application",
52            Self::UnsupportedReadMode => "unsupported read mode",
53            Self::DeletionVectorPredicateRejection => "deletion-vector predicate read rejection",
54            Self::DeletionVectorMasking => "deletion-vector masking",
55        })
56    }
57}
58
59/// Phase associated with a Delta scan deletion-vector failure.
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61pub enum DeltaScanDeletionVectorPhase {
62    /// Parsing the table URI failed.
63    TableUriParsing,
64    /// Constructing the kernel object-store engine failed.
65    ObjectStoreEngineConstruction,
66    /// Accessing the preserved deletion-vector descriptor failed.
67    DescriptorAccess,
68    /// Reading or decoding the deletion-vector payload failed.
69    PayloadRead,
70    /// The selection vector did not match the physical file row count.
71    SelectionVectorLengthMismatch,
72    /// The selection vector was consumed after it was closed.
73    SelectionVectorExhaustion,
74}
75
76impl std::fmt::Display for DeltaScanDeletionVectorPhase {
77    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
78        formatter.write_str(match self {
79            Self::TableUriParsing => "table URI parsing",
80            Self::ObjectStoreEngineConstruction => "object store engine construction",
81            Self::DescriptorAccess => "deletion-vector descriptor access",
82            Self::PayloadRead => "deletion-vector payload read",
83            Self::SelectionVectorLengthMismatch => "selection-vector length mismatch",
84            Self::SelectionVectorExhaustion => "selection-vector exhaustion",
85        })
86    }
87}
88
89/// Error type used by DeltaFunnel APIs.
90#[derive(Debug, Snafu)]
91#[snafu(visibility(pub(crate)))]
92pub enum DeltaFunnelError {
93    /// Caller configuration is invalid.
94    #[snafu(display("configuration error: {message}"))]
95    Config {
96        /// Sanitized message suitable for logs and Python-facing errors.
97        message: String,
98    },
99
100    /// A Delta source name is not valid for registration.
101    #[snafu(display(
102        "invalid Delta source name `{}`: {reason}",
103        sanitize_source_name_for_display(name)
104    ))]
105    InvalidSourceName {
106        /// Caller-provided source name.
107        name: String,
108        /// Sanitized reason for the validation failure.
109        reason: &'static str,
110    },
111
112    /// Two configured Delta sources use the same registration name.
113    #[snafu(display(
114        "duplicate Delta source name `{}`",
115        sanitize_source_name_for_display(name)
116    ))]
117    DuplicateSourceName {
118        /// Caller-provided duplicate source name.
119        name: String,
120    },
121
122    /// A Delta source URI is not valid for snapshot loading.
123    #[snafu(display("invalid Delta source URI: {reason}"))]
124    InvalidSourceUri {
125        /// Sanitized reason for the validation failure.
126        reason: &'static str,
127    },
128
129    /// A Delta source engine could not be constructed.
130    #[snafu(display("Delta source engine error: {reason}"))]
131    DeltaSourceEngine {
132        /// Sanitized reason for the engine construction failure.
133        reason: &'static str,
134    },
135
136    /// A Delta snapshot could not be loaded.
137    #[snafu(display("Delta snapshot load error: {reason}"))]
138    DeltaSnapshotLoad {
139        /// Sanitized reason for the snapshot load failure.
140        reason: String,
141    },
142
143    /// A Delta source requires an unsupported reader protocol.
144    #[snafu(display(
145        "Delta protocol compatibility error for source `{}` at snapshot version {snapshot_version} ({}): {reason}",
146        sanitize_source_name_for_display(source_name),
147        sanitize_uri_for_display(table_uri)
148    ))]
149    DeltaProtocolCompatibility {
150        /// Caller-provided source name.
151        source_name: String,
152        /// Sanitized or sanitizable Delta table URI context.
153        table_uri: String,
154        /// Resolved Delta snapshot version.
155        snapshot_version: u64,
156        /// Sanitized reason for the compatibility failure.
157        reason: String,
158    },
159
160    /// A Delta source schema could not be exposed to the query engine.
161    #[snafu(display(
162        "Delta source schema error for source `{}` ({}): {}",
163        sanitize_source_name_for_display(source_name),
164        sanitize_uri_for_display(table_uri),
165        sanitize_reason_for_display(reason)
166    ))]
167    DeltaSourceSchema {
168        /// Caller-provided source name.
169        source_name: String,
170        /// Sanitized or sanitizable Delta table URI context.
171        table_uri: String,
172        /// Sanitized reason for the schema failure.
173        reason: String,
174    },
175
176    /// A Delta source could not be registered with DataFusion.
177    #[snafu(display(
178        "DataFusion registration error for source `{}` ({}): {}",
179        sanitize_source_name_for_display(source_name),
180        sanitize_uri_for_display(table_uri),
181        sanitize_reason_for_display(reason)
182    ))]
183    DataFusionRegistration {
184        /// Caller-provided source name.
185        source_name: String,
186        /// Sanitized or sanitizable Delta table URI context.
187        table_uri: String,
188        /// Sanitized reason for the registration failure.
189        reason: String,
190    },
191
192    /// Lazy SQL-derived table planning or alias registration failed.
193    #[snafu(display(
194        "SQL table error during {phase}: {}",
195        sanitize_reason_for_display(message)
196    ))]
197    SqlTable {
198        /// Session SQL table phase that failed.
199        phase: SqlTablePhase,
200        /// Sanitized reason for the failure.
201        message: String,
202    },
203
204    /// A Delta provider scan projection could not be planned.
205    #[snafu(display(
206        "Delta scan projection error for source `{}` ({}): {}",
207        sanitize_source_name_for_display(source_name),
208        sanitize_uri_for_display(table_uri),
209        sanitize_reason_for_display(reason)
210    ))]
211    DeltaScanProjection {
212        /// Caller-provided source name.
213        source_name: String,
214        /// Sanitized or sanitizable Delta table URI context.
215        table_uri: String,
216        /// Sanitized reason for the projection failure.
217        reason: String,
218    },
219
220    /// A pushed Delta provider scan filter could not be planned safely.
221    #[snafu(display(
222        "Delta scan filter error for source `{}` ({}): {}",
223        sanitize_source_name_for_display(source_name),
224        sanitize_uri_for_display(table_uri),
225        sanitize_reason_for_display(reason)
226    ))]
227    DeltaScanFilter {
228        /// Caller-provided source name.
229        source_name: String,
230        /// Sanitized or sanitizable Delta table URI context.
231        table_uri: String,
232        /// Sanitized reason for the filter failure.
233        reason: String,
234    },
235
236    /// A Delta kernel scan could not be constructed.
237    #[snafu(display(
238        "Delta scan construction error for source `{}` ({}): {}",
239        sanitize_source_name_for_display(source_name),
240        sanitize_uri_for_display(table_uri),
241        sanitize_reason_for_display(&source.to_string())
242    ))]
243    DeltaScanConstruction {
244        /// Caller-provided source name.
245        source_name: String,
246        /// Sanitized or sanitizable Delta table URI context.
247        table_uri: String,
248        /// Kernel scan construction failure.
249        #[snafu(source(from(delta_kernel::Error, Box::new)))]
250        source: Box<delta_kernel::Error>,
251    },
252
253    /// Delta scan metadata could not be expanded from kernel scan planning.
254    #[snafu(display(
255        "Delta scan metadata expansion error for source `{}` at snapshot version {snapshot_version} ({}): {}",
256        sanitize_source_name_for_display(source_name),
257        sanitize_uri_for_display(table_uri),
258        sanitize_reason_for_display(&source.to_string())
259    ))]
260    DeltaScanMetadataExpansion {
261        /// Caller-provided source name.
262        source_name: String,
263        /// Sanitized or sanitizable Delta table URI context.
264        table_uri: String,
265        /// Resolved Delta snapshot version.
266        snapshot_version: u64,
267        /// Kernel scan metadata expansion failure.
268        #[snafu(source(from(delta_kernel::Error, Box::new)))]
269        source: Box<delta_kernel::Error>,
270    },
271
272    /// Delta scan metadata could not be converted into provider file tasks.
273    #[snafu(display(
274        "Delta scan file task planning error for source `{}` at snapshot version {snapshot_version} ({}), file `{}`: {}",
275        sanitize_source_name_for_display(source_name),
276        sanitize_uri_for_display(table_uri),
277        sanitize_reason_for_display(path),
278        sanitize_reason_for_display(reason)
279    ))]
280    DeltaScanFileTaskPlanning {
281        /// Caller-provided source name.
282        source_name: String,
283        /// Sanitized or sanitizable Delta table URI context.
284        table_uri: String,
285        /// Resolved Delta snapshot version.
286        snapshot_version: u64,
287        /// Delta add-action path associated with the task planning failure.
288        path: String,
289        /// Sanitized reason for the task planning failure.
290        reason: String,
291    },
292
293    /// Delta scan file tasks could not be grouped into provider scan partitions.
294    #[snafu(display(
295        "Delta scan file task partition planning error for source `{}` at snapshot version {snapshot_version} ({}): {}",
296        sanitize_source_name_for_display(source_name),
297        sanitize_uri_for_display(table_uri),
298        sanitize_reason_for_display(reason)
299    ))]
300    DeltaScanFileTaskPartitionPlanning {
301        /// Caller-provided source name.
302        source_name: String,
303        /// Sanitized or sanitizable Delta table URI context.
304        table_uri: String,
305        /// Resolved Delta snapshot version.
306        snapshot_version: u64,
307        /// Sanitized reason for the partition planning failure.
308        reason: String,
309    },
310
311    /// A Delta scan data file could not be read through the kernel adapter.
312    #[snafu(display(
313        "Delta scan file read error for source `{}` at snapshot version {snapshot_version} ({}), file `{}` during {phase}: {}",
314        sanitize_source_name_for_display(source_name),
315        sanitize_uri_for_display(table_uri),
316        sanitize_reason_for_display(path),
317        sanitize_reason_for_display(&source.to_string())
318    ))]
319    DeltaScanFileRead {
320        /// Caller-provided source name.
321        source_name: String,
322        /// Sanitized or sanitizable Delta table URI context.
323        table_uri: String,
324        /// Resolved Delta snapshot version.
325        snapshot_version: u64,
326        /// Delta add-action path associated with the read failure.
327        path: String,
328        /// Read phase associated with the failure.
329        phase: DeltaScanFileReadPhase,
330        /// Underlying kernel read failure.
331        #[snafu(source(from(delta_kernel::Error, Box::new)))]
332        source: Box<delta_kernel::Error>,
333    },
334
335    /// A Delta scan deletion vector could not be loaded or consumed safely.
336    #[snafu(display(
337        "Delta scan deletion-vector error for source `{}` at snapshot version {snapshot_version} ({}), file `{}` during {phase}: {}",
338        sanitize_source_name_for_display(source_name),
339        sanitize_uri_for_display(table_uri),
340        sanitize_reason_for_display(path),
341        sanitize_reason_for_display(&source.to_string())
342    ))]
343    DeltaScanDeletionVector {
344        /// Caller-provided source name.
345        source_name: String,
346        /// Sanitized or sanitizable Delta table URI context.
347        table_uri: String,
348        /// Resolved Delta snapshot version.
349        snapshot_version: u64,
350        /// Delta add-action path associated with the deletion-vector failure.
351        path: String,
352        /// Deletion-vector phase associated with the failure.
353        phase: DeltaScanDeletionVectorPhase,
354        /// Underlying kernel deletion-vector failure.
355        #[snafu(source(from(delta_kernel::Error, Box::new)))]
356        source: Box<delta_kernel::Error>,
357    },
358
359    /// A required dependency contract is unavailable or incompatible.
360    #[snafu(display("dependency compatibility error: {message}"))]
361    DependencyCompatibility {
362        /// Sanitized message suitable for logs and Python-facing errors.
363        message: String,
364    },
365
366    /// Batch pipeline setup or configuration is invalid.
367    #[snafu(display(
368        "batch pipeline {phase} error for option `{option}`: {}",
369        sanitize_reason_for_display(message)
370    ))]
371    BatchPipeline {
372        /// Batch pipeline phase associated with the failure.
373        phase: BatchPipelinePhase,
374        /// Stable option name associated with the failure.
375        option: &'static str,
376        /// Sanitized message suitable for logs and Python-facing errors.
377        message: String,
378    },
379
380    /// SQL Server target configuration is invalid.
381    #[snafu(display(
382        "MSSQL target configuration error for option `{option}`: {}",
383        sanitize_reason_for_display(message)
384    ))]
385    MssqlTargetConfig {
386        /// Stable option name associated with the failure.
387        option: &'static str,
388        /// Sanitized message suitable for logs and Python-facing errors.
389        message: String,
390    },
391
392    /// A SQL Server target could not resolve a connection.
393    #[snafu(display(
394        "MSSQL target for output `{}` has no effective connection",
395        sanitize_text_for_display(output_name)
396    ))]
397    MissingMssqlConnection {
398        /// Selected output name associated with the target.
399        output_name: String,
400    },
401
402    /// A selected output identity is missing or invalid for SQL Server schema planning.
403    #[snafu(display(
404        "MSSQL schema planning error for output `{}`: {reason}",
405        sanitize_text_for_display(output_name)
406    ))]
407    InvalidMssqlOutputIdentity {
408        /// Selected output name associated with the schema.
409        output_name: String,
410        /// Stable reason for the validation failure.
411        reason: &'static str,
412    },
413
414    /// A selected output has duplicate field names before SQL Server planning.
415    #[snafu(display(
416        "MSSQL schema planning error for output `{}`: duplicate field name `{}` at indexes {first_index} and {duplicate_index}",
417        sanitize_text_for_display(output_name),
418        sanitize_text_for_display(field_name)
419    ))]
420    DuplicateMssqlOutputField {
421        /// Selected output name associated with the schema.
422        output_name: String,
423        /// Duplicate output field name.
424        field_name: String,
425        /// First index where the field name was seen.
426        first_index: usize,
427        /// Duplicate index where the field name was seen again.
428        duplicate_index: usize,
429    },
430
431    /// SQL Server schema planning failed with structured arrow-tiberius diagnostics.
432    #[snafu(display(
433        "MSSQL schema planning error for output `{}`: arrow-tiberius returned {} diagnostic(s)",
434        sanitize_text_for_display(output_name),
435        diagnostics.len()
436    ))]
437    MssqlSchemaPlanning {
438        /// Selected output name associated with the schema.
439        output_name: String,
440        /// Structured diagnostics returned by arrow-tiberius.
441        diagnostics: arrow_tiberius::DiagnosticSet,
442    },
443
444    /// SQL Server schema planning failed before producing diagnostics.
445    #[snafu(display(
446        "MSSQL schema planning error for output `{}`: {}",
447        sanitize_text_for_display(output_name),
448        sanitize_reason_for_display(&source.to_string())
449    ))]
450    MssqlSchemaPlanningFailed {
451        /// Selected output name associated with the schema.
452        output_name: String,
453        /// Underlying arrow-tiberius planning failure.
454        source: arrow_tiberius::Error,
455    },
456
457    /// SQL Server DDL planning failed because a target identifier was invalid.
458    #[snafu(display(
459        "MSSQL DDL planning error for output `{}`: {}",
460        sanitize_text_for_display(output_name),
461        sanitize_reason_for_display(&source.to_string())
462    ))]
463    MssqlDdlTargetIdentifier {
464        /// Selected output name associated with the target.
465        output_name: String,
466        /// Underlying arrow-tiberius identifier validation failure.
467        source: arrow_tiberius::Error,
468    },
469
470    /// SQL Server DDL planning failed for a DeltaFunnel-owned lifecycle reason.
471    #[snafu(display(
472        "MSSQL DDL planning error for output `{}`: {}",
473        sanitize_text_for_display(output_name),
474        sanitize_reason_for_display(message)
475    ))]
476    MssqlDdlPlanning {
477        /// Selected output name associated with the target.
478        output_name: String,
479        /// Sanitized reason for the DDL planning failure.
480        message: String,
481    },
482
483    /// SQL Server lifecycle planning failed for a DeltaFunnel-owned reason.
484    #[snafu(display(
485        "MSSQL lifecycle planning error for output `{}`: {}",
486        sanitize_text_for_display(output_name),
487        sanitize_reason_for_display(message)
488    ))]
489    MssqlLifecyclePlanning {
490        /// Selected output name associated with the target.
491        output_name: String,
492        /// Sanitized reason for the lifecycle planning failure.
493        message: String,
494    },
495
496    /// SQL Server batch writing failed.
497    #[snafu(display(
498        "MSSQL write error: {}",
499        sanitize_reason_for_display(&source.to_string())
500    ))]
501    MssqlWrite {
502        /// Underlying Arrow-to-TDS or Tiberius writer failure.
503        source: arrow_tiberius::Error,
504    },
505
506    /// SQL Server write execution failed during a known phase.
507    #[snafu(display(
508        "MSSQL write error for output `{}` during {}: {}",
509        sanitize_text_for_display(context.output_name()),
510        context.phase(),
511        sanitize_reason_for_display(message)
512    ))]
513    MssqlWritePhase {
514        /// Structured, redacted failure context.
515        context: Box<crate::MssqlWriteFailureContext>,
516        /// Sanitized reason for the write failure.
517        message: String,
518    },
519
520    /// SQL Server output batch schema validation failed.
521    #[snafu(display(
522        "MSSQL write error for output `{}` during {}: {}",
523        sanitize_text_for_display(context.output_name()),
524        context.phase(),
525        sanitize_reason_for_display(&source.to_string())
526    ))]
527    MssqlBatchSchemaValidation {
528        /// Structured, redacted failure context.
529        context: Box<crate::MssqlWriteFailureContext>,
530        /// Underlying arrow-tiberius schema validation failure.
531        source: arrow_tiberius::Error,
532    },
533
534    /// SQL Server multi-output workflow planning failed before output writes.
535    #[snafu(display(
536        "MSSQL workflow planning error: {}",
537        sanitize_reason_for_display(message)
538    ))]
539    MssqlWorkflowPlanning {
540        /// Sanitized reason for the workflow planning failure.
541        message: String,
542    },
543}
544
545/// Phase associated with lazy SQL table construction.
546#[derive(Debug, Clone, Copy, PartialEq, Eq)]
547pub enum SqlTablePhase {
548    /// Caller SQL text failed local validation.
549    ValidateSql,
550    /// DataFusion could not produce a lazy logical table.
551    PlanSql,
552    /// A derived table alias could not be registered in the session catalog.
553    RegisterDerivedAlias,
554}
555
556impl std::fmt::Display for SqlTablePhase {
557    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
558        formatter.write_str(match self {
559            Self::ValidateSql => "SQL validation",
560            Self::PlanSql => "SQL planning",
561            Self::RegisterDerivedAlias => "derived alias registration",
562        })
563    }
564}
565
566fn sanitize_source_name_for_display(name: &str) -> String {
567    sanitize_text_for_display(name)
568}
569
570fn sanitize_reason_for_display(reason: &str) -> String {
571    sanitize_text_for_display(reason)
572}
573
574#[cfg(test)]
575mod tests {
576    use std::error::Error;
577
578    use super::DeltaFunnelError;
579
580    #[test]
581    fn config_error_has_sanitized_display() {
582        let error = DeltaFunnelError::Config {
583            message: "max_concurrent_file_reads_per_scan must be greater than zero".to_owned(),
584        };
585
586        assert_eq!(
587            error.to_string(),
588            "configuration error: max_concurrent_file_reads_per_scan must be greater than zero"
589        );
590    }
591
592    #[test]
593    fn dependency_error_has_sanitized_display() {
594        let error = DeltaFunnelError::DependencyCompatibility {
595            message: "delta_kernel API smoke test failed".to_owned(),
596        };
597
598        assert_eq!(
599            error.to_string(),
600            "dependency compatibility error: delta_kernel API smoke test failed"
601        );
602    }
603
604    #[test]
605    fn batch_pipeline_error_has_sanitized_display() {
606        let error = DeltaFunnelError::BatchPipeline {
607            phase: super::BatchPipelinePhase::Configuration,
608            option: "output_batch_size",
609            message: "must be greater than zero".to_owned(),
610        };
611
612        assert_eq!(
613            error.to_string(),
614            "batch pipeline configuration error for option `output_batch_size`: must be greater than zero"
615        );
616    }
617
618    #[test]
619    fn batch_pipeline_error_display_escapes_control_characters() {
620        let error = DeltaFunnelError::BatchPipeline {
621            phase: super::BatchPipelinePhase::HandoffSetup,
622            option: "consumer_capacity",
623            message: "invalid\nvalue\tprovided".to_owned(),
624        };
625
626        let display = error.to_string();
627
628        assert!(!display.contains('\n'));
629        assert!(!display.contains('\t'));
630        assert!(display.contains(r"invalid\nvalue\tprovided"));
631    }
632
633    #[test]
634    fn mssql_write_error_has_sanitized_display() {
635        let error = DeltaFunnelError::MssqlWrite {
636            source: arrow_tiberius::Error::BackendUnavailable {
637                backend: arrow_tiberius::WriteBackend::DirectRawBulk,
638                reason: "not available\nfor test".to_owned(),
639            },
640        };
641
642        let display = error.to_string();
643
644        assert!(!display.contains('\n'));
645        assert!(display.contains(r"not available\nfor test"));
646    }
647
648    #[test]
649    fn mssql_write_phase_error_has_sanitized_display_and_context() -> Result<(), DeltaFunnelError> {
650        let connection = crate::MssqlConnectionConfig::new(
651            "server=tcp:sql.example.com;database=warehouse;user=admin;password=secret-token",
652        )?
653        .with_display_label("warehouse-primary");
654        let target_config =
655            crate::MssqlTargetConfig::new(crate::MssqlTargetTable::new("dbo", "orders")?);
656        let schema = arrow_schema::Schema::new(vec![arrow_schema::Field::new(
657            "order_id",
658            arrow_schema::DataType::Int64,
659            false,
660        )]);
661        let output_plan = crate::plan_mssql_target_for_output(
662            schema,
663            "orders_output",
664            &target_config,
665            Some(&connection),
666            arrow_tiberius::PlanOptions::default(),
667        )?;
668        let context = crate::MssqlWriteFailureContext::from_output_plan(
669            &output_plan,
670            crate::MssqlWritePhase::WriteBatch,
671            42,
672            3,
673            125,
674            true,
675            crate::MssqlTargetCleanupStatus::NotApplicable,
676        );
677
678        let error = DeltaFunnelError::MssqlWritePhase {
679            context: Box::new(context),
680            message: "batch failed\nwhile writing".to_owned(),
681        };
682
683        let display = error.to_string();
684
685        assert!(display.contains("orders_output"));
686        assert!(display.contains("write batch"));
687        assert!(!display.contains('\n'));
688        assert!(display.contains(r"batch failed\nwhile writing"));
689        assert!(!display.contains("secret-token"));
690        assert!(!display.contains("server=tcp"));
691        let DeltaFunnelError::MssqlWritePhase { context, .. } = error else {
692            return Err(DeltaFunnelError::Config {
693                message: "expected MssqlWritePhase error".to_owned(),
694            });
695        };
696        assert_eq!(context.phase(), crate::MssqlWritePhase::WriteBatch);
697        assert_eq!(context.output_name(), "orders_output");
698        assert_eq!(context.stats().rows_written(), 42);
699        assert!(context.partial_write_possible());
700        Ok(())
701    }
702
703    #[test]
704    fn mssql_batch_schema_validation_error_has_sanitized_display_and_context()
705    -> Result<(), DeltaFunnelError> {
706        let connection = crate::MssqlConnectionConfig::new(
707            "server=tcp:sql.example.com;database=warehouse;user=admin;password=secret-token",
708        )?
709        .with_display_label("warehouse-primary");
710        let target_config =
711            crate::MssqlTargetConfig::new(crate::MssqlTargetTable::new("dbo", "orders")?);
712        let schema = arrow_schema::Schema::new(vec![arrow_schema::Field::new(
713            "order_id",
714            arrow_schema::DataType::Int64,
715            false,
716        )]);
717        let output_plan = crate::plan_mssql_target_for_output(
718            schema,
719            "orders_output",
720            &target_config,
721            Some(&connection),
722            arrow_tiberius::PlanOptions::default(),
723        )?;
724        let context = crate::MssqlWriteFailureContext::from_output_plan(
725            &output_plan,
726            crate::MssqlWritePhase::ValidateBatchSchema,
727            0,
728            0,
729            0,
730            false,
731            crate::MssqlTargetCleanupStatus::NotApplicable,
732        );
733
734        let error = DeltaFunnelError::MssqlBatchSchemaValidation {
735            context: Box::new(context),
736            source: arrow_tiberius::Error::BackendUnavailable {
737                backend: arrow_tiberius::WriteBackend::DirectRawBulk,
738                reason: "schema mismatch\nfor test".to_owned(),
739            },
740        };
741
742        let display = error.to_string();
743
744        assert!(display.contains("orders_output"));
745        assert!(display.contains("validate batch schema"));
746        assert!(!display.contains('\n'));
747        assert!(display.contains(r"schema mismatch\nfor test"));
748        assert!(!display.contains("secret-token"));
749        assert!(!display.contains("server=tcp"));
750        let DeltaFunnelError::MssqlBatchSchemaValidation { context, source } = error else {
751            return Err(DeltaFunnelError::Config {
752                message: "expected MssqlBatchSchemaValidation error".to_owned(),
753            });
754        };
755        assert_eq!(context.phase(), crate::MssqlWritePhase::ValidateBatchSchema);
756        assert_eq!(context.output_name(), "orders_output");
757        assert_eq!(context.stats().rows_written(), 0);
758        assert!(!context.partial_write_possible());
759        assert!(matches!(
760            source,
761            arrow_tiberius::Error::BackendUnavailable { .. }
762        ));
763        Ok(())
764    }
765
766    #[test]
767    fn invalid_source_name_error_has_sanitized_display() {
768        let error = DeltaFunnelError::InvalidSourceName {
769            name: "orders.latest".to_owned(),
770            reason: "source names may contain only ASCII letters, digits, and underscores",
771        };
772
773        assert_eq!(
774            error.to_string(),
775            "invalid Delta source name `orders.latest`: source names may contain only ASCII letters, digits, and underscores"
776        );
777    }
778
779    #[test]
780    fn invalid_source_name_display_escapes_control_characters() {
781        let error = DeltaFunnelError::InvalidSourceName {
782            name: "orders\nlatest\tname".to_owned(),
783            reason: "source names may contain only ASCII letters, digits, and underscores",
784        };
785
786        let display = error.to_string();
787
788        assert!(!display.contains('\n'));
789        assert!(!display.contains('\t'));
790        assert!(display.contains(r"orders\nlatest\tname"));
791    }
792
793    #[test]
794    fn duplicate_source_name_error_has_sanitized_display() {
795        let error = DeltaFunnelError::DuplicateSourceName {
796            name: "Orders".to_owned(),
797        };
798
799        assert_eq!(error.to_string(), "duplicate Delta source name `Orders`");
800    }
801
802    #[test]
803    fn invalid_source_uri_error_has_sanitized_display() {
804        let error = DeltaFunnelError::InvalidSourceUri {
805            reason: "table location could not be parsed or normalized",
806        };
807
808        assert_eq!(
809            error.to_string(),
810            "invalid Delta source URI: table location could not be parsed or normalized"
811        );
812    }
813
814    #[test]
815    fn source_engine_error_has_sanitized_display() {
816        let error = DeltaFunnelError::DeltaSourceEngine {
817            reason: "object store engine could not be constructed",
818        };
819
820        assert_eq!(
821            error.to_string(),
822            "Delta source engine error: object store engine could not be constructed"
823        );
824    }
825
826    #[test]
827    fn snapshot_load_error_has_sanitized_display() {
828        let error = DeltaFunnelError::DeltaSnapshotLoad {
829            reason: "snapshot could not be loaded".to_owned(),
830        };
831
832        assert_eq!(
833            error.to_string(),
834            "Delta snapshot load error: snapshot could not be loaded"
835        );
836    }
837
838    #[test]
839    fn protocol_compatibility_error_has_sanitized_display() {
840        let error = DeltaFunnelError::DeltaProtocolCompatibility {
841            source_name: "orders\nlatest".to_owned(),
842            table_uri: "s3://user:password@example.com/table?token=secret".to_owned(),
843            snapshot_version: 7,
844            reason: "unsupported Delta reader feature `deletionVectors`".to_owned(),
845        };
846
847        let display = error.to_string();
848
849        assert!(display.contains(r"orders\nlatest"));
850        assert!(display.contains("snapshot version 7"));
851        assert!(display.contains("s3://example.com/table"));
852        assert!(display.contains("deletionVectors"));
853        assert!(!display.contains('\n'));
854        assert!(!display.contains("user"));
855        assert!(!display.contains("password"));
856        assert!(!display.contains("token"));
857        assert!(!display.contains("secret"));
858    }
859
860    #[test]
861    fn source_schema_error_has_sanitized_display() {
862        let error = DeltaFunnelError::DeltaSourceSchema {
863            source_name: "orders\nlatest".to_owned(),
864            table_uri: "s3://user:password@example.com/table?token=secret".to_owned(),
865            reason: "field\nname could not be converted".to_owned(),
866        };
867
868        let display = error.to_string();
869
870        assert!(display.contains(r"orders\nlatest"));
871        assert!(display.contains("s3://example.com/table"));
872        assert!(display.contains(r"field\nname"));
873        assert!(!display.contains('\n'));
874        assert!(!display.contains("user"));
875        assert!(!display.contains("password"));
876        assert!(!display.contains("token"));
877        assert!(!display.contains("secret"));
878    }
879
880    #[test]
881    fn datafusion_registration_error_has_sanitized_display() {
882        let error = DeltaFunnelError::DataFusionRegistration {
883            source_name: "orders\nlatest".to_owned(),
884            table_uri: "s3://user:password@example.com/table?token=secret".to_owned(),
885            reason: "table\nalready exists".to_owned(),
886        };
887
888        let display = error.to_string();
889
890        assert!(display.contains(r"orders\nlatest"));
891        assert!(display.contains("s3://example.com/table"));
892        assert!(display.contains(r"table\nalready exists"));
893        assert!(!display.contains('\n'));
894        assert!(!display.contains("user"));
895        assert!(!display.contains("password"));
896        assert!(!display.contains("token"));
897        assert!(!display.contains("secret"));
898    }
899
900    #[test]
901    fn scan_metadata_expansion_error_has_sanitized_display()
902    -> Result<(), Box<dyn std::error::Error>> {
903        let error = DeltaFunnelError::DeltaScanMetadataExpansion {
904            source_name: "orders\nlatest".to_owned(),
905            table_uri: "s3://user:password@example.com/table?token=secret".to_owned(),
906            snapshot_version: 7,
907            source: Box::new(delta_kernel::Error::generic(
908                "scan\nmetadata expansion failed",
909            )),
910        };
911
912        let display = error.to_string();
913
914        assert!(display.contains(r"orders\nlatest"));
915        assert!(display.contains("snapshot version 7"));
916        assert!(display.contains("s3://example.com/table"));
917        assert!(display.contains(r"scan\nmetadata expansion failed"));
918        assert!(!display.contains('\n'));
919        assert!(!display.contains("user"));
920        assert!(!display.contains("password"));
921        assert!(!display.contains("token"));
922        assert!(!display.contains("secret"));
923
924        let source = Error::source(&error)
925            .ok_or("metadata expansion error must preserve its kernel source")?;
926        assert!(
927            source
928                .to_string()
929                .contains("scan\nmetadata expansion failed")
930        );
931
932        Ok(())
933    }
934
935    #[test]
936    fn scan_file_task_planning_error_has_sanitized_display() {
937        let error = DeltaFunnelError::DeltaScanFileTaskPlanning {
938            source_name: "orders\nlatest".to_owned(),
939            table_uri: "s3://user:password@example.com/table?token=secret".to_owned(),
940            snapshot_version: 7,
941            path: "part\n00000.parquet".to_owned(),
942            reason: "kernel\nsize was negative".to_owned(),
943        };
944
945        let display = error.to_string();
946
947        assert!(display.contains(r"orders\nlatest"));
948        assert!(display.contains("snapshot version 7"));
949        assert!(display.contains("s3://example.com/table"));
950        assert!(display.contains(r"part\n00000.parquet"));
951        assert!(display.contains(r"kernel\nsize was negative"));
952        assert!(!display.contains('\n'));
953        assert!(!display.contains("user"));
954        assert!(!display.contains("password"));
955        assert!(!display.contains("token"));
956        assert!(!display.contains("secret"));
957    }
958
959    #[test]
960    fn scan_file_task_partition_planning_error_has_sanitized_display() {
961        let error = DeltaFunnelError::DeltaScanFileTaskPartitionPlanning {
962            source_name: "orders\nlatest".to_owned(),
963            table_uri: "s3://user:password@example.com/table?token=secret".to_owned(),
964            snapshot_version: 7,
965            reason: "target\npartitions was zero".to_owned(),
966        };
967
968        let display = error.to_string();
969
970        assert!(display.contains(r"orders\nlatest"));
971        assert!(display.contains("snapshot version 7"));
972        assert!(display.contains("s3://example.com/table"));
973        assert!(display.contains(r"target\npartitions was zero"));
974        assert!(!display.contains('\n'));
975        assert!(!display.contains("user"));
976        assert!(!display.contains("password"));
977        assert!(!display.contains("token"));
978        assert!(!display.contains("secret"));
979    }
980}
delta_funnel/error.rs

delta_funnel/
error.rs
