delta_kernel 0.29.0

//! Definitions of errors that the delta kernel can encounter

use std::backtrace::{Backtrace, BacktraceStatus};
use std::convert::Infallible;
use std::num::ParseIntError;
use std::str::Utf8Error;

#[cfg(feature = "default-engine-base")]
use crate::arrow::error::ArrowError;
#[cfg(feature = "default-engine-base")]
use crate::object_store;
use crate::schema::{DataType, StructType};
use crate::table_properties::ParseIntervalError;
use crate::Version;

/// Details of a failed conversion from a scalar into a Rust value.
///
/// Conversion code adds path elements as an error unwinds, producing a path from the outermost
/// value to the value that failed without carrying mutable path state through successful parsing.
#[derive(Debug)]
pub struct ScalarConversionError {
    expected: String,
    actual: String,
    // Stored innermost-first because parent context is appended as conversion errors unwind.
    path: Vec<String>,
}

impl ScalarConversionError {
    pub(crate) fn new(expected: impl Into<String>, actual: impl Into<String>) -> Self {
        Self {
            expected: expected.into(),
            actual: actual.into(),
            path: Vec::new(),
        }
    }

    fn add_path_context(mut self, element: impl Into<String>) -> Self {
        self.path.push(element.into());
        self
    }

    fn path_string(&self) -> String {
        let mut path = String::new();
        for element in self.path.iter().rev() {
            if !path.is_empty() && !element.starts_with('[') {
                path.push('.');
            }
            path.push_str(element);
        }
        path
    }
}

impl std::fmt::Display for ScalarConversionError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut target = self.path_string();
        if target.is_empty() {
            target.push_str("scalar");
        }
        write!(
            f,
            "Cannot convert {target}: expected {}, found {}",
            self.expected, self.actual
        )
    }
}

impl std::error::Error for ScalarConversionError {}

/// Adds an outer path element to a scalar conversion error as nested conversion unwinds.
///
/// Other error variants are returned unchanged: a field's `TryFrom<Scalar>` implementation may
/// report a failure unrelated to scalar shape, and this helper must not reclassify it.
pub(crate) fn add_scalar_path_context(error: Error, element: impl Into<String>) -> Error {
    match error {
        Error::ScalarConversion(error) => Error::ScalarConversion(error.add_path_context(element)),
        other => other,
    }
}

/// A [`std::result::Result`] that has the kernel [`Error`] as the error variant
pub type DeltaResult<T, E = Error> = std::result::Result<T, E>;

/// A boxed, `Send` iterator of [`DeltaResult<T>`] items.
///
/// Convenience alias for the common pattern of returning a streaming, fallible iterator from
/// kernel APIs.
pub type DeltaResultIterator<'a, T> = Box<dyn Iterator<Item = DeltaResult<T>> + Send + 'a>;

/// `'static` counterpart to [`DeltaResultIterator`] for cases where the iterator does not
/// reference borrowed data.
pub type DeltaResultIteratorStatic<T> = DeltaResultIterator<'static, T>;

/// An error validating connector-provided state for snapshot construction.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum SnapshotHintError {
    /// A hint was combined with a log tail.
    #[error("Invalid snapshot hint: A snapshot hint cannot be combined with a log tail")]
    LogTail,
    /// A hint was combined with incremental CRC replay.
    #[error(
        "Invalid snapshot hint: A snapshot hint cannot be combined with incremental CRC replay"
    )]
    IncrementalReplay,
    /// The builder requested a version different from the hint's version.
    #[error(
        "Invalid snapshot hint: Requested version {requested} does not match snapshot hint version {hint}"
    )]
    VersionMismatch {
        /// The version requested from the snapshot builder.
        requested: Version,
        /// The version described by the snapshot hint.
        hint: Version,
    },
    /// The maximum catalog version differs from the hint when no time-travel version was
    /// requested.
    #[error(
        "Invalid snapshot hint: Max catalog version {max_catalog_version} does not match snapshot \
         hint version {hint}"
    )]
    MaxCatalogVersionMismatch {
        /// The maximum version ratified by the catalog.
        max_catalog_version: Version,
        /// The version described by the snapshot hint.
        hint: Version,
    },
    /// A hint marked latest conflicts with a later catalog-ratified version.
    #[error(
        "Invalid snapshot hint: version {hint} is marked latest but max catalog version is {max_catalog_version}"
    )]
    LatestVersionConflict {
        /// The version described by the snapshot hint.
        hint: Version,
        /// The latest version ratified by the catalog.
        max_catalog_version: Version,
    },
    /// The supplied log files contain log compaction files, which snapshot hints do not support.
    #[error("Invalid snapshot hint: log compaction files are not supported")]
    LogCompaction,
    /// The supplied log files cannot form a valid log segment.
    #[error("Invalid snapshot hint: supplied log files do not form a valid log segment")]
    LogSegment {
        /// The log-segment construction error.
        #[source]
        source: Box<Error>,
    },
    /// The hint includes a published version after its snapshot version.
    #[error("Invalid snapshot hint: max_published_version exceeds snapshot hint version {hint}")]
    MaxPublishedVersion {
        /// The version described by the snapshot hint.
        hint: Version,
    },
    /// The hint has neither a complete checkpoint nor commit version zero.
    #[error("Invalid snapshot hint: snapshot history does not start at version 0")]
    MissingHistoryAnchor,
    /// The supplied CRC describes a different table version.
    #[error(
        "Invalid snapshot hint: CRC version {crc} does not match snapshot hint version {hint}"
    )]
    CrcVersion {
        /// The version described by the CRC.
        crc: Version,
        /// The version described by the snapshot hint.
        hint: Version,
    },
    /// The supplied CRC protocol differs from the hint protocol.
    #[error("Invalid snapshot hint: CRC protocol does not match snapshot hint protocol")]
    CrcProtocol,
    /// The supplied CRC metadata differs from the hint metadata.
    #[error("Invalid snapshot hint: CRC metadata does not match snapshot hint metadata")]
    CrcMetadata,
    /// A connector reported invalid snapshot-hint state, optionally with an underlying error.
    #[error("Invalid snapshot hint: {message}")]
    Connector {
        /// A description of the invalid connector state.
        message: String,
        /// The underlying validation error, if available.
        #[source]
        source: Option<Box<Error>>,
    },
}

impl From<SnapshotHintError> for Error {
    fn from(error: SnapshotHintError) -> Self {
        Box::new(error).into()
    }
}

/// All the types of errors that the kernel can run into
#[non_exhaustive]
#[derive(thiserror::Error, Debug)]
pub enum Error {
    /// This is an error that includes a backtrace. To have a particular type of error include such
    /// backtrace (when RUST_BACKTRACE=1), annotate the error with `#[error(transparent)]` and then
    /// add the error type and enum variant to the `from_with_backtrace!` macro invocation
    /// below. See IOError for an example.
    #[error("{source}\n{backtrace}")]
    Backtraced {
        source: Box<Self>,
        backtrace: Box<Backtrace>,
    },

    /// An error performing operations on arrow data
    #[cfg(feature = "default-engine-base")]
    #[error(transparent)]
    Arrow(ArrowError),

    #[error("Error writing checkpoint: {0}")]
    CheckpointWrite(String),

    /// User tried to convert engine data to the wrong type
    #[error("Invalid engine data type. Could not convert to {0}")]
    EngineDataType(String),

    /// Could not extract the specified type
    #[error("Error extracting type {0}: {1}")]
    Extract(&'static str, &'static str),

    /// A scalar could not be converted into the requested Rust value.
    #[error(transparent)]
    ScalarConversion(#[from] ScalarConversionError),

    /// A generic error with a message
    #[error("Generic delta kernel error: {0}")]
    Generic(String),

    /// A generic error wrapping another error
    #[error("Generic error: {source}")]
    GenericError {
        /// Source error
        source: Box<dyn std::error::Error + Send + Sync + 'static>,
    },

    /// An error involving the maximum catalog-ratified version when building a snapshot.
    #[error("Max catalog version error: {0}")]
    MaxCatalogVersion(String),

    /// The supplied log tail contains adjacent versions that are not contiguous.
    #[error("Log tail versions {first_version} and {second_version} are not contiguous")]
    LogTailVersionsNotContiguous {
        /// Earlier version in the invalid adjacent pair.
        first_version: Version,
        /// Later version in the invalid adjacent pair.
        second_version: Version,
    },

    /// Some kind of [`std::io::Error`]
    #[error(transparent)]
    IOError(std::io::Error),

    /// An internal error that means kernel found an unexpected situation, which is likely a bug
    #[error("Internal error {0}. This is a kernel bug, please report.")]
    InternalError(String),

    /// An error enountered while working with parquet data
    #[cfg(feature = "default-engine-base")]
    #[error("Arrow error: {0}")]
    Parquet(#[from] crate::parquet::errors::ParquetError),

    /// An error interacting with the object_store crate
    // We don't use [#from] object_store::Error here as our From impl transforms
    // object_store::Error::NotFound into Self::FileNotFound
    #[cfg(feature = "default-engine-base")]
    #[error("Error interacting with object store: {0}")]
    ObjectStore(object_store::Error),

    /// An error working with paths from the object_store crate
    #[cfg(feature = "default-engine-base")]
    #[error("Object store path error: {0}")]
    ObjectStorePath(#[from] object_store::path::Error),

    #[cfg(feature = "default-engine-base")]
    #[error("Reqwest Error: {0}")]
    Reqwest(#[from] reqwest::Error),

    /// A specified file could not be found
    #[error("File not found: {0}")]
    FileNotFound(String),

    /// A column was requested, but not found
    #[error("{0}")]
    MissingColumn(String),

    /// The connector-provided partition values are invalid (missing/extra/duplicate keys,
    /// or a value type does not match the schema column type).
    #[error("Invalid partition values: {0}")]
    InvalidPartitionValues(String),

    /// A column was specified with a specific type, but it is not of that type
    #[error("Expected column type: {0}")]
    UnexpectedColumnType(String),

    /// Data was expected, but not found
    #[error("Expected is missing: {0}")]
    MissingData(String),

    /// No table versions were found for the requested log operation.
    #[error("No table version found.")]
    EmptyLog,

    /// One or more table versions required by a log operation are unavailable.
    ///
    /// The payload is the lowest version that the operation requires but cannot obtain.
    #[error("Table version {0} is missing or unavailable for this log operation.")]
    MissingVersion(Version),

    /// A table version required by an operation has not been published to the Delta log.
    ///
    /// The payload is the first unpublished version.
    #[error("Table version {0} has not been published to the Delta log.")]
    UnpublishedVersion(Version),

    /// An error occurred while working with deletion vectors
    #[error("Deletion Vector error: {0}")]
    DeletionVector(String),

    /// A selection vector is larger than data length
    #[error("Selection vector is larger than data length: {0}")]
    InvalidSelectionVector(String),

    /// Transaction state is invalid for the requested operation
    #[error("Invalid transaction state: {0}")]
    InvalidTransactionState(String),

    /// A specified URL was invalid
    #[error("Invalid url: {0}")]
    InvalidUrl(#[from] url::ParseError),

    /// serde encountered malformed json
    #[error(transparent)]
    MalformedJson(serde_json::Error),

    /// There was no metadata action in the delta log
    #[error("No table metadata found in delta log.")]
    MissingMetadata,

    /// There was no protocol action in the delta log
    #[error("No protocol found in delta log.")]
    MissingProtocol,

    /// Invalid protocol action was read from the log
    #[error("Invalid protocol action in the delta log: {0}")]
    InvalidProtocol(String),

    /// Neither metadata nor protocol could be found in the delta log
    #[error("No table metadata or protocol found in delta log.")]
    MissingMetadataAndProtocol,

    /// A string failed to parse as the specified data type
    #[error("Failed to parse value '{0}' as '{1}'")]
    ParseError(String, DataType),

    /// A tokio executor failed to join a task
    #[error("Join failure: {0}")]
    JoinFailure(String),

    /// Could not convert to string from utf-8
    #[error("Could not convert to string from utf-8: {0}")]
    Utf8Error(#[from] Utf8Error),

    /// Could not parse an integer
    #[error("Could not parse int: {0}")]
    ParseIntError(#[from] ParseIntError),

    #[error("Invalid column mapping mode: {0}")]
    InvalidColumnMappingMode(String),

    /// Asked for a table at an invalid location
    #[error("Invalid table location: {0}.")]
    InvalidTableLocation(String),

    /// Precision or scale not compliant with delta specification
    #[error("Invalid decimal: {0}")]
    InvalidDecimal(String),

    /// Invalid CRS or other parameter for a Geometry / Geography type
    #[error("Invalid geo parameters: {0}")]
    InvalidGeoParams(String),

    /// Inconsistent data passed to struct scalar
    #[error("Invalid struct data: {0}")]
    InvalidStructData(String),

    /// Expressions did not parse or evaluate correctly
    #[error("Invalid expression evaluation: {0}")]
    InvalidExpressionEvaluation(String),

    /// Unable to parse the name of a log path
    #[error("Invalid log path: {0}")]
    InvalidLogPath(String),

    /// The assembled log segment is inconsistent with its declared file kinds, ordering, or
    /// version bounds. Malformed checkpoint file sets use [`Error::InvalidCheckpoint`].
    #[error("Invalid log segment: {0}")]
    InvalidLogSegment(String),

    /// Snapshot-hint validation failed. Log-segment errors caused by supplied hint state,
    /// including invalid paths and checkpoints, are wrapped in `SnapshotHintError::LogSegment`.
    /// Failures outside hint validation retain their existing categories.
    #[error(transparent)]
    SnapshotHint(#[from] Box<SnapshotHintError>),

    /// The file already exists at the path, prohibiting a non-overwrite write
    #[error("File already exists: {0}")]
    FileAlreadyExists(String),

    /// Some functionality is currently unsupported
    #[error("Unsupported: {0}")]
    Unsupported(String),

    /// Cannot write a version checksum (CRC) file for this snapshot
    #[error("Checksum write unsupported: {0}")]
    ChecksumWriteUnsupported(String),

    /// Parsing error when attempting to deserialize an interval
    #[error(transparent)]
    ParseIntervalError(#[from] ParseIntervalError),

    #[error("Change data feed is unsupported for the table at version {0}")]
    ChangeDataFeedUnsupported(Version),

    /// Row tracking (`delta.enableRowTracking`) must be enabled for the entire version range of a
    /// row-tracking change feed, but it is not enabled at the given version.
    #[error(
        "Row tracking (delta.enableRowTracking) must be enabled for the entire row-tracking change \
         feed range, but it is not enabled at version {0}"
    )]
    RowTrackingChangeFeedUnsupported(Version),

    #[error("Change data feed encountered incompatible schema. Expected {0}, got {1}")]
    ChangeDataFeedIncompatibleSchema(String, String),

    /// Invalid checkpoint files
    #[error("Invalid Checkpoint: {0}")]
    InvalidCheckpoint(String),

    /// Schema mismatch has occurred or invalid/not-kernel-supported schema used somewhere
    #[error("Schema error: {0}")]
    Schema(String),

    /// Validation error for file statistics (e.g., missing required clustering column stats)
    #[error("Stats validation error: {0}")]
    StatsValidation(String),

    /// Error during log history operations (timestamp queries, version lookups)
    #[error(transparent)]
    LogHistory(#[from] Box<crate::history_manager::error::LogHistoryError>),

    #[cfg(feature = "declarative-plans")]
    #[error("Declarative plan execution yielded the incorrect type: expected PlanResult::{expected}, got PlanResult::{actual}")]
    PlanResultTypeMismatch {
        expected: &'static str,
        actual: &'static str,
    },

    /// The operation was cancelled via a [`CancellationToken`](crate::CancellationToken).
    ///
    /// Surfaced by cancellation-aware operations when a cancellation is detected. See the
    /// [Engine operation cancellation contract](crate::cancellation#engine-operation-contract).
    #[error("Operation cancelled")]
    Cancelled,
}

// Convenience constructors for Error types that take a String argument
impl Error {
    pub(crate) fn scalar_conversion(
        expected: impl Into<String>,
        actual: impl Into<String>,
    ) -> Self {
        ScalarConversionError::new(expected, actual).into()
    }

    pub(crate) fn checkpoint_write(msg: impl ToString) -> Self {
        Self::CheckpointWrite(msg.to_string())
    }

    pub fn generic_err(source: impl Into<Box<dyn std::error::Error + Send + Sync>>) -> Self {
        Self::GenericError {
            source: source.into(),
        }
    }
    pub fn generic(msg: impl ToString) -> Self {
        Self::Generic(msg.to_string())
    }
    pub fn file_not_found(path: impl ToString) -> Self {
        Self::FileNotFound(path.to_string())
    }
    pub fn missing_column(name: impl ToString) -> Self {
        Self::MissingColumn(name.to_string()).with_backtrace()
    }
    pub fn unexpected_column_type(name: impl ToString) -> Self {
        Self::UnexpectedColumnType(name.to_string())
    }
    pub fn invalid_partition_values(msg: impl ToString) -> Self {
        Self::InvalidPartitionValues(msg.to_string())
    }
    pub fn missing_data(name: impl ToString) -> Self {
        Self::MissingData(name.to_string())
    }
    pub fn deletion_vector(msg: impl ToString) -> Self {
        Self::DeletionVector(msg.to_string())
    }
    pub fn engine_data_type(msg: impl ToString) -> Self {
        Self::EngineDataType(msg.to_string())
    }
    pub fn join_failure(msg: impl ToString) -> Self {
        Self::JoinFailure(msg.to_string())
    }
    pub fn invalid_table_location(location: impl ToString) -> Self {
        Self::InvalidTableLocation(location.to_string())
    }
    pub fn invalid_column_mapping_mode(mode: impl ToString) -> Self {
        Self::InvalidColumnMappingMode(mode.to_string())
    }
    pub fn invalid_decimal(msg: impl ToString) -> Self {
        Self::InvalidDecimal(msg.to_string())
    }
    #[cfg(feature = "geo-type-in-dev")]
    pub fn invalid_geo_params(msg: impl ToString) -> Self {
        Self::InvalidGeoParams(msg.to_string())
    }
    pub fn invalid_struct_data(msg: impl ToString) -> Self {
        Self::InvalidStructData(msg.to_string())
    }
    pub fn invalid_expression(msg: impl ToString) -> Self {
        Self::InvalidExpressionEvaluation(msg.to_string())
    }
    pub(crate) fn invalid_log_path(msg: impl ToString) -> Self {
        Self::InvalidLogPath(msg.to_string())
    }

    pub(crate) fn invalid_log_segment(msg: impl ToString) -> Self {
        Self::InvalidLogSegment(msg.to_string())
    }

    pub fn internal_error(msg: impl ToString) -> Self {
        Self::InternalError(msg.to_string()).with_backtrace()
    }

    pub fn invalid_protocol(msg: impl ToString) -> Self {
        Self::InvalidProtocol(msg.to_string())
    }

    pub fn invalid_transaction_state(msg: impl ToString) -> Self {
        Self::InvalidTransactionState(msg.to_string())
    }

    pub fn unsupported(msg: impl ToString) -> Self {
        Self::Unsupported(msg.to_string())
    }
    pub fn change_data_feed_unsupported(version: impl Into<Version>) -> Self {
        Self::ChangeDataFeedUnsupported(version.into())
    }
    /// Creates an [`Error::RowTrackingChangeFeedUnsupported`] for the given version, used when row
    /// tracking is not enabled at some point in a row-tracking change feed's version range.
    pub(crate) fn row_tracking_change_feed_unsupported(version: impl Into<Version>) -> Self {
        Self::RowTrackingChangeFeedUnsupported(version.into())
    }
    pub(crate) fn change_data_feed_incompatible_schema(
        expected: &StructType,
        actual: &StructType,
    ) -> Self {
        Self::ChangeDataFeedIncompatibleSchema(expected.to_string(), actual.to_string())
    }

    /// Creates an incompatible-schema error that identifies the version of `actual`.
    pub(crate) fn change_data_feed_incompatible_schema_at_version(
        expected: &StructType,
        actual: &StructType,
        version: Version,
    ) -> Self {
        Self::ChangeDataFeedIncompatibleSchema(
            expected.to_string(),
            format!("schema at version {version}: {actual}"),
        )
    }

    pub fn invalid_checkpoint(msg: impl ToString) -> Self {
        Self::InvalidCheckpoint(msg.to_string())
    }

    pub fn schema(msg: impl ToString) -> Self {
        Self::Schema(msg.to_string())
    }

    pub fn stats_validation(msg: impl ToString) -> Self {
        Self::StatsValidation(msg.to_string())
    }

    #[cfg(feature = "declarative-plans")]
    pub fn plan_result_type_mismatch(expected: &'static str, actual: &'static str) -> Self {
        Self::PlanResultTypeMismatch { expected, actual }
    }

    // Capture a backtrace when the error is constructed.
    #[must_use]
    pub fn with_backtrace(self) -> Self {
        let backtrace = Backtrace::capture();
        match backtrace.status() {
            BacktraceStatus::Captured => Self::Backtraced {
                source: Box::new(self),
                backtrace: Box::new(backtrace),
            },
            _ => self,
        }
    }
}

macro_rules! from_with_backtrace(
    ( $(($error_type: ty, $error_variant: ident)), * ) => {
        $(
            impl From<$error_type> for Error {
                fn from(value: $error_type) -> Self {
                    Self::$error_variant(value).with_backtrace()
                }
            }
        )*
    };
);

from_with_backtrace!(
    (serde_json::Error, MalformedJson),
    (std::io::Error, IOError)
);

#[cfg(feature = "default-engine-base")]
impl From<ArrowError> for Error {
    fn from(value: ArrowError) -> Self {
        Self::Arrow(value).with_backtrace()
    }
}

#[cfg(feature = "default-engine-base")]
impl From<object_store::Error> for Error {
    fn from(value: object_store::Error) -> Self {
        match value {
            object_store::Error::NotFound { path, .. } => Self::file_not_found(path),
            err => Self::ObjectStore(err),
        }
    }
}

/// This impl is needed so the `?` operator can auto-convert `Result<T, Infallible>` to
/// `DeltaResult<T>`. For example, `TryFrom` impls for infallible conversions use `Infallible` as
/// their error type, and this allows those results to be propagated with `?` in functions
/// returning `DeltaResult`. The match is unreachable since `Infallible` has no variants.
impl From<Infallible> for Error {
    fn from(value: Infallible) -> Self {
        match value {}
    }
}