delta_funnel/

error.rs

//! Shared error pattern for DeltaFunnel.

use crate::BatchPipelinePhase;
use crate::support::{sanitize_text_for_display, sanitize_uri_for_display};

use snafu::Snafu;

/// Error type used by DeltaFunnel APIs.
#[derive(Debug, Snafu)]
#[snafu(visibility(pub(crate)))]
pub enum DeltaFunnelError {
    /// Caller configuration is invalid.
    #[snafu(display("configuration error: {message}"))]
    Config {
        /// Sanitized message suitable for logs and Python-facing errors.
        message: String,
    },

    /// A bounded preview failed after its operation lifecycle began.
    #[snafu(display(
        "preview failed during {}: {}",
        context.failed_phase(),
        sanitize_reason_for_display(&source.to_string())
    ))]
    PreviewFailed {
        /// Structured, redacted preview failure context.
        context: Box<crate::PreviewFailureContext>,
        /// Original Delta Funnel failure with its source chain preserved.
        source: Box<DeltaFunnelError>,
    },

    /// A cache-enabled `write_all` call failed after cache selection.
    #[snafu(display(
        "write_all cache failed: {}",
        sanitize_reason_for_display(&source.to_string())
    ))]
    WriteAllCache {
        /// Failed cache attempts and any completed output workflow.
        failure: Box<crate::WriteAllCacheFailure>,
        /// Original primary failure with its source chain preserved.
        source: Box<DeltaFunnelError>,
    },

    /// A Delta source name is not valid for registration.
    #[snafu(display(
        "invalid Delta source name `{}`: {reason}",
        sanitize_source_name_for_display(name)
    ))]
    InvalidSourceName {
        /// Caller-provided source name.
        name: String,
        /// Sanitized reason for the validation failure.
        reason: &'static str,
    },

    /// Two configured Delta sources use the same registration name.
    #[snafu(display(
        "duplicate Delta source name `{}`",
        sanitize_source_name_for_display(name)
    ))]
    DuplicateSourceName {
        /// Caller-provided duplicate source name.
        name: String,
    },

    /// A Delta source URI is not valid for snapshot loading.
    #[snafu(display("invalid Delta source URI: {reason}"))]
    InvalidSourceUri {
        /// Sanitized reason for the validation failure.
        reason: &'static str,
    },

    /// A Delta source engine could not be constructed.
    #[snafu(display("Delta source engine error: {reason}"))]
    DeltaSourceEngine {
        /// Sanitized reason for the engine construction failure.
        reason: &'static str,
    },

    /// A Delta snapshot could not be loaded.
    #[snafu(display("Delta snapshot load error: {reason}"))]
    DeltaSnapshotLoad {
        /// Sanitized reason for the snapshot load failure.
        reason: String,
    },

    /// A Delta source requires an unsupported reader protocol.
    #[snafu(display(
        "Delta protocol compatibility error for source `{}` at snapshot version {snapshot_version} ({}): {reason}",
        sanitize_source_name_for_display(source_name),
        sanitize_uri_for_display(table_uri)
    ))]
    DeltaProtocolCompatibility {
        /// Caller-provided source name.
        source_name: String,
        /// Sanitized or sanitizable Delta table URI context.
        table_uri: String,
        /// Resolved Delta snapshot version.
        snapshot_version: u64,
        /// Sanitized reason for the compatibility failure.
        reason: String,
    },

    /// A Delta source schema could not be exposed to the query engine.
    #[snafu(display(
        "Delta source schema error for source `{}` ({}): {}",
        sanitize_source_name_for_display(source_name),
        sanitize_uri_for_display(table_uri),
        sanitize_reason_for_display(reason)
    ))]
    DeltaSourceSchema {
        /// Caller-provided source name.
        source_name: String,
        /// Sanitized or sanitizable Delta table URI context.
        table_uri: String,
        /// Sanitized reason for the schema failure.
        reason: String,
    },

    /// A Delta source could not be registered with DataFusion.
    #[snafu(display(
        "DataFusion registration error for source `{}` ({}): {}",
        sanitize_source_name_for_display(source_name),
        sanitize_uri_for_display(table_uri),
        sanitize_reason_for_display(reason)
    ))]
    DataFusionRegistration {
        /// Caller-provided source name.
        source_name: String,
        /// Sanitized or sanitizable Delta table URI context.
        table_uri: String,
        /// Sanitized reason for the registration failure.
        reason: String,
    },

    /// Lazy SQL-derived table planning or alias registration failed.
    #[snafu(display(
        "SQL table error during {phase}: {}",
        sanitize_reason_for_display(message)
    ))]
    SqlTable {
        /// Session SQL table phase that failed.
        phase: SqlTablePhase,
        /// Sanitized reason for the failure.
        message: String,
    },

    /// A required dependency contract is unavailable or incompatible.
    #[snafu(display("dependency compatibility error: {message}"))]
    DependencyCompatibility {
        /// Sanitized message suitable for logs and Python-facing errors.
        message: String,
    },

    /// Batch pipeline setup or configuration is invalid.
    #[snafu(display(
        "batch pipeline {phase} error for option `{option}`: {}",
        sanitize_reason_for_display(message)
    ))]
    BatchPipeline {
        /// Batch pipeline phase associated with the failure.
        phase: BatchPipelinePhase,
        /// Stable option name associated with the failure.
        option: &'static str,
        /// Sanitized message suitable for logs and Python-facing errors.
        message: String,
    },

    /// SQL Server target configuration is invalid.
    #[snafu(display(
        "MSSQL target configuration error for option `{option}`: {}",
        sanitize_reason_for_display(message)
    ))]
    MssqlTargetConfig {
        /// Stable option name associated with the failure.
        option: &'static str,
        /// Sanitized message suitable for logs and Python-facing errors.
        message: String,
    },

    /// A SQL Server target could not resolve a connection.
    #[snafu(display(
        "MSSQL target for output `{}` has no effective connection",
        sanitize_text_for_display(output_name)
    ))]
    MissingMssqlConnection {
        /// Selected output name associated with the target.
        output_name: String,
    },

    /// A selected output identity is missing or invalid for SQL Server schema planning.
    #[snafu(display(
        "MSSQL schema planning error for output `{}`: {reason}",
        sanitize_text_for_display(output_name)
    ))]
    InvalidMssqlOutputIdentity {
        /// Selected output name associated with the schema.
        output_name: String,
        /// Stable reason for the validation failure.
        reason: &'static str,
    },

    /// A selected output has duplicate field names before SQL Server planning.
    #[snafu(display(
        "MSSQL schema planning error for output `{}`: duplicate field name `{}` at indexes {first_index} and {duplicate_index}",
        sanitize_text_for_display(output_name),
        sanitize_text_for_display(field_name)
    ))]
    DuplicateMssqlOutputField {
        /// Selected output name associated with the schema.
        output_name: String,
        /// Duplicate output field name.
        field_name: String,
        /// First index where the field name was seen.
        first_index: usize,
        /// Duplicate index where the field name was seen again.
        duplicate_index: usize,
    },

    /// SQL Server schema planning failed with structured arrow-sql-server diagnostics.
    #[snafu(display(
        "MSSQL schema planning error for output `{}`: arrow-sql-server returned {} diagnostic(s)",
        sanitize_text_for_display(output_name),
        diagnostics.len()
    ))]
    MssqlSchemaPlanning {
        /// Selected output name associated with the schema.
        output_name: String,
        /// Structured diagnostics returned by arrow-sql-server.
        diagnostics: arrow_sql_server::DiagnosticSet,
    },

    /// SQL Server schema planning failed before producing diagnostics.
    #[snafu(display(
        "MSSQL schema planning error for output `{}`: {}",
        sanitize_text_for_display(output_name),
        sanitize_reason_for_display(&source.to_string())
    ))]
    MssqlSchemaPlanningFailed {
        /// Selected output name associated with the schema.
        output_name: String,
        /// Underlying arrow-sql-server planning failure.
        source: arrow_sql_server::Error,
    },

    /// SQL Server DDL planning failed because a target identifier was invalid.
    #[snafu(display(
        "MSSQL DDL planning error for output `{}`: {}",
        sanitize_text_for_display(output_name),
        sanitize_reason_for_display(&source.to_string())
    ))]
    MssqlDdlTargetIdentifier {
        /// Selected output name associated with the target.
        output_name: String,
        /// Underlying arrow-sql-server identifier validation failure.
        source: arrow_sql_server::Error,
    },

    /// SQL Server DDL planning failed for a DeltaFunnel-owned lifecycle reason.
    #[snafu(display(
        "MSSQL DDL planning error for output `{}`: {}",
        sanitize_text_for_display(output_name),
        sanitize_reason_for_display(message)
    ))]
    MssqlDdlPlanning {
        /// Selected output name associated with the target.
        output_name: String,
        /// Sanitized reason for the DDL planning failure.
        message: String,
    },

    /// SQL Server lifecycle planning failed for a DeltaFunnel-owned reason.
    #[snafu(display(
        "MSSQL lifecycle planning error for output `{}`: {}",
        sanitize_text_for_display(output_name),
        sanitize_reason_for_display(message)
    ))]
    MssqlLifecyclePlanning {
        /// Selected output name associated with the target.
        output_name: String,
        /// Sanitized reason for the lifecycle planning failure.
        message: String,
    },

    /// SQL Server batch writing failed.
    #[snafu(display(
        "MSSQL write error: {}",
        sanitize_reason_for_display(&source.to_string())
    ))]
    MssqlWrite {
        /// Underlying Arrow-to-TDS or Tiberius writer failure.
        source: arrow_sql_server::Error,
    },

    /// SQL Server write execution failed during a known phase.
    #[snafu(display(
        "MSSQL write error for output `{}` during {}: {}",
        sanitize_text_for_display(context.output_name()),
        context.phase(),
        sanitize_reason_for_display(message)
    ))]
    MssqlWritePhase {
        /// Structured, redacted failure context.
        context: Box<crate::MssqlWriteFailureContext>,
        /// Sanitized reason for the write failure.
        message: String,
    },

    /// SQL Server output query preparation failed during a known phase.
    #[snafu(display(
        "MSSQL write error for output `{}` during {}: {}",
        sanitize_text_for_display(context.output_name()),
        context.phase(),
        sanitize_reason_for_display(&source.to_string())
    ))]
    MssqlQueryPhase {
        /// Structured, redacted failure context.
        context: Box<crate::MssqlWriteFailureContext>,
        /// Original query preparation failure with its source chain preserved.
        source: Box<DeltaFunnelError>,
    },

    /// SQL Server output batch schema validation failed.
    #[snafu(display(
        "MSSQL write error for output `{}` during {}: {}",
        sanitize_text_for_display(context.output_name()),
        context.phase(),
        sanitize_reason_for_display(&source.to_string())
    ))]
    MssqlBatchSchemaValidation {
        /// Structured, redacted failure context.
        context: Box<crate::MssqlWriteFailureContext>,
        /// Underlying arrow-sql-server schema validation failure.
        source: arrow_sql_server::Error,
    },

    /// SQL Server multi-output workflow planning failed before output writes.
    #[snafu(display(
        "MSSQL workflow planning error: {}",
        sanitize_reason_for_display(message)
    ))]
    MssqlWorkflowPlanning {
        /// Sanitized reason for the workflow planning failure.
        message: String,
    },
}

/// Phase associated with lazy SQL table construction.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SqlTablePhase {
    /// Caller SQL text failed local validation.
    ValidateSql,
    /// DataFusion could not produce a lazy logical table.
    PlanSql,
    /// A derived table alias could not be registered in the session catalog.
    RegisterDerivedAlias,
}

impl std::fmt::Display for SqlTablePhase {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(match self {
            Self::ValidateSql => "SQL validation",
            Self::PlanSql => "SQL planning",
            Self::RegisterDerivedAlias => "derived alias registration",
        })
    }
}

fn sanitize_source_name_for_display(name: &str) -> String {
    sanitize_text_for_display(name)
}

fn sanitize_reason_for_display(reason: &str) -> String {
    sanitize_text_for_display(reason)
}

#[cfg(test)]
mod tests {
    use std::error::Error;

    use super::DeltaFunnelError;

    #[test]
    fn config_error_has_sanitized_display() {
        let error = DeltaFunnelError::Config {
            message: "max_concurrent_file_reads_per_scan must be greater than zero".to_owned(),
        };

        assert_eq!(
            error.to_string(),
            "configuration error: max_concurrent_file_reads_per_scan must be greater than zero"
        );
    }

    #[test]
    fn preview_failure_sanitizes_display_and_preserves_source() {
        let context = crate::PreviewFailureContext::new(
            "preview_dataframe_planning".to_owned(),
            vec![crate::PhaseTimingReport::failed(
                "preview_dataframe_planning",
                std::time::Duration::ZERO,
            )],
            None,
        );
        let error = DeltaFunnelError::PreviewFailed {
            context: Box::new(context),
            source: Box::new(DeltaFunnelError::Config {
                message: "planning failed\nfor test".to_owned(),
            }),
        };

        let display = error.to_string();

        assert!(!display.contains('\n'));
        assert!(display.contains(r"planning failed\nfor test"));
        assert!(Error::source(&error).is_some());
    }

    #[test]
    fn dependency_error_has_sanitized_display() {
        let error = DeltaFunnelError::DependencyCompatibility {
            message: "delta_kernel API smoke test failed".to_owned(),
        };

        assert_eq!(
            error.to_string(),
            "dependency compatibility error: delta_kernel API smoke test failed"
        );
    }

    #[test]
    fn batch_pipeline_error_has_sanitized_display() {
        let error = DeltaFunnelError::BatchPipeline {
            phase: super::BatchPipelinePhase::Configuration,
            option: "output_batch_size",
            message: "must be greater than zero".to_owned(),
        };

        assert_eq!(
            error.to_string(),
            "batch pipeline configuration error for option `output_batch_size`: must be greater than zero"
        );
    }

    #[test]
    fn batch_pipeline_error_display_escapes_control_characters() {
        let error = DeltaFunnelError::BatchPipeline {
            phase: super::BatchPipelinePhase::HandoffSetup,
            option: "consumer_capacity",
            message: "invalid\nvalue\tprovided".to_owned(),
        };

        let display = error.to_string();

        assert!(!display.contains('\n'));
        assert!(!display.contains('\t'));
        assert!(display.contains(r"invalid\nvalue\tprovided"));
    }

    #[test]
    fn mssql_write_error_has_sanitized_display() {
        let error = DeltaFunnelError::MssqlWrite {
            source: arrow_sql_server::Error::BackendUnavailable {
                backend: arrow_sql_server::WriteBackend::DirectRawBulk,
                reason: "not available\nfor test".to_owned(),
            },
        };

        let display = error.to_string();

        assert!(!display.contains('\n'));
        assert!(display.contains(r"not available\nfor test"));
    }

    #[test]
    fn mssql_write_phase_error_has_sanitized_display_and_context() -> Result<(), DeltaFunnelError> {
        let connection = crate::MssqlConnectionConfig::new(
            "server=tcp:sql.example.com;database=warehouse;user=admin;password=secret-token",
        )?
        .with_display_label("warehouse-primary");
        let target_config =
            crate::MssqlTargetConfig::new(crate::MssqlTargetTable::new("dbo", "orders")?);
        let schema = arrow_schema::Schema::new(vec![arrow_schema::Field::new(
            "order_id",
            arrow_schema::DataType::Int64,
            false,
        )]);
        let output_plan = crate::plan_mssql_target_for_output(
            schema,
            "orders_output",
            &target_config,
            Some(&connection),
            arrow_sql_server::PlanOptions::default(),
        )?;
        let context = crate::MssqlWriteFailureContext::from_output_plan(
            &output_plan,
            crate::MssqlWritePhase::WriteBatch,
            42,
            3,
            125,
            true,
            crate::MssqlTargetCleanupStatus::NotApplicable,
        );

        let error = DeltaFunnelError::MssqlWritePhase {
            context: Box::new(context),
            message: "batch failed\nwhile writing".to_owned(),
        };

        let display = error.to_string();

        assert!(display.contains("orders_output"));
        assert!(display.contains("write batch"));
        assert!(!display.contains('\n'));
        assert!(display.contains(r"batch failed\nwhile writing"));
        assert!(!display.contains("secret-token"));
        assert!(!display.contains("server=tcp"));
        let DeltaFunnelError::MssqlWritePhase { context, .. } = error else {
            return Err(DeltaFunnelError::Config {
                message: "expected MssqlWritePhase error".to_owned(),
            });
        };
        assert_eq!(context.phase(), crate::MssqlWritePhase::WriteBatch);
        assert_eq!(context.output_name(), "orders_output");
        assert_eq!(context.stats().rows_written(), 42);
        assert!(context.partial_write_possible());
        Ok(())
    }

    #[test]
    fn mssql_batch_schema_validation_error_has_sanitized_display_and_context()
    -> Result<(), DeltaFunnelError> {
        let connection = crate::MssqlConnectionConfig::new(
            "server=tcp:sql.example.com;database=warehouse;user=admin;password=secret-token",
        )?
        .with_display_label("warehouse-primary");
        let target_config =
            crate::MssqlTargetConfig::new(crate::MssqlTargetTable::new("dbo", "orders")?);
        let schema = arrow_schema::Schema::new(vec![arrow_schema::Field::new(
            "order_id",
            arrow_schema::DataType::Int64,
            false,
        )]);
        let output_plan = crate::plan_mssql_target_for_output(
            schema,
            "orders_output",
            &target_config,
            Some(&connection),
            arrow_sql_server::PlanOptions::default(),
        )?;
        let context = crate::MssqlWriteFailureContext::from_output_plan(
            &output_plan,
            crate::MssqlWritePhase::ValidateBatchSchema,
            0,
            0,
            0,
            false,
            crate::MssqlTargetCleanupStatus::NotApplicable,
        );

        let error = DeltaFunnelError::MssqlBatchSchemaValidation {
            context: Box::new(context),
            source: arrow_sql_server::Error::BackendUnavailable {
                backend: arrow_sql_server::WriteBackend::DirectRawBulk,
                reason: "schema mismatch\nfor test".to_owned(),
            },
        };

        let display = error.to_string();

        assert!(display.contains("orders_output"));
        assert!(display.contains("validate batch schema"));
        assert!(!display.contains('\n'));
        assert!(display.contains(r"schema mismatch\nfor test"));
        assert!(!display.contains("secret-token"));
        assert!(!display.contains("server=tcp"));
        let DeltaFunnelError::MssqlBatchSchemaValidation { context, source } = error else {
            return Err(DeltaFunnelError::Config {
                message: "expected MssqlBatchSchemaValidation error".to_owned(),
            });
        };
        assert_eq!(context.phase(), crate::MssqlWritePhase::ValidateBatchSchema);
        assert_eq!(context.output_name(), "orders_output");
        assert_eq!(context.stats().rows_written(), 0);
        assert!(!context.partial_write_possible());
        assert!(matches!(
            source,
            arrow_sql_server::Error::BackendUnavailable { .. }
        ));
        Ok(())
    }

    #[test]
    fn invalid_source_name_error_has_sanitized_display() {
        let error = DeltaFunnelError::InvalidSourceName {
            name: "orders.latest".to_owned(),
            reason: "source names may contain only ASCII letters, digits, and underscores",
        };

        assert_eq!(
            error.to_string(),
            "invalid Delta source name `orders.latest`: source names may contain only ASCII letters, digits, and underscores"
        );
    }

    #[test]
    fn invalid_source_name_display_escapes_control_characters() {
        let error = DeltaFunnelError::InvalidSourceName {
            name: "orders\nlatest\tname".to_owned(),
            reason: "source names may contain only ASCII letters, digits, and underscores",
        };

        let display = error.to_string();

        assert!(!display.contains('\n'));
        assert!(!display.contains('\t'));
        assert!(display.contains(r"orders\nlatest\tname"));
    }

    #[test]
    fn duplicate_source_name_error_has_sanitized_display() {
        let error = DeltaFunnelError::DuplicateSourceName {
            name: "Orders".to_owned(),
        };

        assert_eq!(error.to_string(), "duplicate Delta source name `Orders`");
    }

    #[test]
    fn invalid_source_uri_error_has_sanitized_display() {
        let error = DeltaFunnelError::InvalidSourceUri {
            reason: "table location could not be parsed or normalized",
        };

        assert_eq!(
            error.to_string(),
            "invalid Delta source URI: table location could not be parsed or normalized"
        );
    }

    #[test]
    fn source_engine_error_has_sanitized_display() {
        let error = DeltaFunnelError::DeltaSourceEngine {
            reason: "object store engine could not be constructed",
        };

        assert_eq!(
            error.to_string(),
            "Delta source engine error: object store engine could not be constructed"
        );
    }

    #[test]
    fn snapshot_load_error_has_sanitized_display() {
        let error = DeltaFunnelError::DeltaSnapshotLoad {
            reason: "snapshot could not be loaded".to_owned(),
        };

        assert_eq!(
            error.to_string(),
            "Delta snapshot load error: snapshot could not be loaded"
        );
    }

    #[test]
    fn protocol_compatibility_error_has_sanitized_display() {
        let error = DeltaFunnelError::DeltaProtocolCompatibility {
            source_name: "orders\nlatest".to_owned(),
            table_uri: "s3://user:password@example.com/table?token=secret".to_owned(),
            snapshot_version: 7,
            reason: "unsupported Delta reader feature `deletionVectors`".to_owned(),
        };

        let display = error.to_string();

        assert!(display.contains(r"orders\nlatest"));
        assert!(display.contains("snapshot version 7"));
        assert!(display.contains("s3://example.com/table"));
        assert!(display.contains("deletionVectors"));
        assert!(!display.contains('\n'));
        assert!(!display.contains("user"));
        assert!(!display.contains("password"));
        assert!(!display.contains("token"));
        assert!(!display.contains("secret"));
    }

    #[test]
    fn source_schema_error_has_sanitized_display() {
        let error = DeltaFunnelError::DeltaSourceSchema {
            source_name: "orders\nlatest".to_owned(),
            table_uri: "s3://user:password@example.com/table?token=secret".to_owned(),
            reason: "field\nname could not be converted".to_owned(),
        };

        let display = error.to_string();

        assert!(display.contains(r"orders\nlatest"));
        assert!(display.contains("s3://example.com/table"));
        assert!(display.contains(r"field\nname"));
        assert!(!display.contains('\n'));
        assert!(!display.contains("user"));
        assert!(!display.contains("password"));
        assert!(!display.contains("token"));
        assert!(!display.contains("secret"));
    }

    #[test]
    fn datafusion_registration_error_has_sanitized_display() {
        let error = DeltaFunnelError::DataFusionRegistration {
            source_name: "orders\nlatest".to_owned(),
            table_uri: "s3://user:password@example.com/table?token=secret".to_owned(),
            reason: "table\nalready exists".to_owned(),
        };

        let display = error.to_string();

        assert!(display.contains(r"orders\nlatest"));
        assert!(display.contains("s3://example.com/table"));
        assert!(display.contains(r"table\nalready exists"));
        assert!(!display.contains('\n'));
        assert!(!display.contains("user"));
        assert!(!display.contains("password"));
        assert!(!display.contains("token"));
        assert!(!display.contains("secret"));
    }
}