use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fmt::{Display, Formatter};
use utoipa::ToSchema;

/// Delta table write mode.
///
/// Determines how the Delta table connector handles an existing table at the target location.
#[derive(Default, Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)]
pub enum DeltaTableWriteMode {
    /// New updates will be appended to the existing table at the target location.
    ///
    /// If the table doesn't exist, it is created.
    #[default]
    #[serde(rename = "append")]
    Append,

    /// Existing table at the specified location will get truncated.
    ///
    /// The connector truncates the table by outputting delete actions for all
    /// files in the latest snapshot of the table.
    ///
    /// If the table doesn't exist, it is created.
    #[serde(rename = "truncate")]
    Truncate,

    /// If a table exists at the specified location, the operation must fail.
    #[serde(rename = "error_if_exists")]
    ErrorIfExists,
}

/// Delta table output connector configuration.
#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)]
pub struct DeltaTableWriterConfig {
    /// Table URI.
    pub uri: String,

    /// Determines how the Delta table connector handles an existing table at the target location.
    #[serde(default)]
    pub mode: DeltaTableWriteMode,

    /// Checkpoint interval (i.e., the number of commits after which a new checkpoint should be created) for newly created Delta tables.
    ///
    /// The option is only available when creating the Delta table (`mode = append` and there
    /// is no existing table at the target location or `mode = truncate`). It configures the `checkpointInterval`
    /// table property, which determines the number of commits after which a new checkpoint should be created.
    ///
    /// 0 means no checkpoints are created.
    ///
    /// Default: 10.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub checkpoint_interval: Option<u32>,

    /// Log retention duration for newly created Delta tables.
    ///
    /// Configures the `delta.logRetentionDuration` table property, which controls how long the
    /// transaction log history of the table is kept.  Each time a checkpoint is written, Delta Lake
    /// automatically cleans up log entries older than this interval (subject to
    /// `enable_expired_log_cleanup`).
    ///
    /// The option is only available when creating the Delta table (`mode = append` and there is
    /// no existing table at the target location, or `mode = truncate`).
    ///
    /// The value follows the Delta Lake interval syntax: `"interval <N> <unit>"`, where `<unit>`
    /// is one of `nanosecond[s]`, `microsecond[s]`, `millisecond[s]`, `second[s]`, `minute[s]`,
    /// `hour[s]`, `day[s]`, or `week[s]`.  Examples: `"interval 30 days"`, `"interval 6 hours"`.
    ///
    /// Default: `"interval 30 days"` (Delta Lake default).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub log_retention_duration: Option<String>,

    /// Whether to clean up expired log entries when a checkpoint is written.
    ///
    /// Configures the `delta.enableExpiredLogCleanup` table property.  When set to `false`,
    /// transaction log entries are retained indefinitely regardless of `log_retention_duration`.
    ///
    /// The option is only available when creating the Delta table (`mode = append` and there is
    /// no existing table at the target location, or `mode = truncate`).
    ///
    /// Default: `true` (Delta Lake default).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub enable_expired_log_cleanup: Option<bool>,

    /// Maximum number of retries for failed operations.
    ///
    /// The connector performs retries on several levels: individual S3 operations, Delta Lake transaction commits,
    /// and overall operation retries. This setting controls the overall operation retries. When a write to the table
    /// fails, because of an S3 timeout or any other reason that was not resolved by lower-level retries, the connector
    /// will retry the entire operation.
    ///
    /// When not specified, the connector performs infinite retries. When set to 0, the connector doesn't
    /// retry failed operations.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_retries: Option<u32>,

    /// Number of parallel threads used by the connector.
    ///
    /// Increasing this value can improve Delta Lake write throughput
    /// by enabling concurrent writes.
    ///
    /// Default: 1.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[schema(minimum = 1)]
    pub threads: Option<usize>,

    /// Storage options for configuring backend object store.
    ///
    /// For specific options available for different storage backends, see:
    /// * [Azure options](https://docs.rs/object_store/latest/object_store/azure/enum.AzureConfigKey.html)
    /// * [Amazon S3 options](https://docs.rs/object_store/latest/object_store/aws/enum.AmazonS3ConfigKey.html)
    /// * [Google Cloud Storage options](https://docs.rs/object_store/latest/object_store/gcp/enum.GoogleConfigKey.html)
    #[serde(flatten)]
    pub object_store_config: HashMap<String, String>,
}

impl DeltaTableWriterConfig {
    pub fn validate(&self) -> Result<(), String> {
        if self.threads.is_some_and(|t| t == 0) {
            return Err("threads must be greater than 0".to_string());
        }
        if let Some(duration) = &self.log_retention_duration {
            validate_delta_interval(duration)
                .map_err(|e| format!("invalid 'log_retention_duration' value '{duration}': {e}"))?;
        }
        Ok(())
    }
}

/// Validate a Delta Lake interval string (e.g. `"interval 30 days"`).
///
/// Mirrors the grammar accepted by delta-rs (`crates/core/src/table/config.rs::parse_interval`):
/// the keyword `interval` must be lowercase, the number must be a non-negative integer, and the
/// unit must come from delta-rs's closed list.  We replicate the grammar here so misconfigured
/// values fail fast at config-load time with a clear error message rather than being silently
/// dropped by delta-rs at runtime (delta-rs stores the string opaquely at table-create time and
/// only parses it for the log-cleanup hook, where unparseable values fall back to the default).
fn validate_delta_interval(value: &str) -> Result<(), String> {
    let mut tokens = value.split_whitespace();
    if tokens.next() != Some("interval") {
        return Err(
            "expected the value to start with lowercase \"interval\" (e.g. \"interval 30 days\")"
                .to_string(),
        );
    }
    let number_token = tokens
        .next()
        .ok_or_else(|| "expected a number after \"interval\"".to_string())?;
    let unit = tokens
        .next()
        .ok_or_else(|| "expected a unit (e.g. \"days\") after the number".to_string())?;
    if tokens.next().is_some() {
        return Err("unexpected trailing tokens".to_string());
    }
    let number: i64 = number_token
        .parse()
        .map_err(|e| format!("cannot parse '{number_token}' as integer: {e}"))?;
    if number < 0 {
        return Err("interval cannot be negative".to_string());
    }
    match unit {
        "nanosecond" | "nanoseconds" | "microsecond" | "microseconds" | "millisecond"
        | "milliseconds" | "second" | "seconds" | "minute" | "minutes" | "hour" | "hours"
        | "day" | "days" | "week" | "weeks" => Ok(()),
        other => Err(format!(
            "unknown unit '{other}'; expected one of nanosecond[s], microsecond[s], \
             millisecond[s], second[s], minute[s], hour[s], day[s], week[s]"
        )),
    }
}

/// Delta table read mode.
///
/// Three options are available:
///
/// * `snapshot` - read a snapshot of the table and stop.
///
/// * `follow` - continuously ingest changes to the table, starting from a specified version
///   or timestamp.
///
/// * `snapshot_and_follow` - read a snapshot of the table before switching to continuous ingestion
///   mode.
#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)]
pub enum DeltaTableIngestMode {
    /// Read a snapshot of the table and stop.
    #[serde(rename = "snapshot")]
    Snapshot,

    /// Follow the changelog of the table, only ingesting changes (new and deleted rows).
    #[serde(rename = "follow")]
    Follow,

    /// Take a snapshot of the table before switching to the `follow` mode.
    #[serde(rename = "snapshot_and_follow")]
    SnapshotAndFollow,

    /// Change-Data-Capture (CDC) mode.
    ///
    /// The table behaves as an append-only log where every row represents an insert
    /// or delete action.  The order of actions is determined by the `cdc_order_by`
    /// property, and the type of each action is determined by the `cdc_delete_filter`
    /// property.
    ///
    /// In this mode, the connector does not read the initial snapshot of the table
    /// and follows the transaction log starting from the version of the table
    /// specified by the `version` or `datetime` property.
    #[serde(rename = "cdc")]
    Cdc,
}

impl Display for DeltaTableIngestMode {
    fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), std::fmt::Error> {
        match self {
            DeltaTableIngestMode::Snapshot => write!(f, "snapshot"),
            DeltaTableIngestMode::Follow => write!(f, "follow"),
            DeltaTableIngestMode::SnapshotAndFollow => write!(f, "snapshot_and_follow"),
            DeltaTableIngestMode::Cdc => write!(f, "cdc"),
        }
    }
}

fn default_num_parsers() -> u32 {
    4
}

/// Delta table transaction mode.
///
/// Determines how the connector breaks up its input into transactions.
///
/// * `none` - the connector does not break up its input into transactions.
/// * `snapshot` - ingest the initial snapshot of the table in one or several transactions. If the connector is
///   configured in the `snapshot_and_follow` mode, it will ingest the transaction log of the table without transactions.
/// * `catchup` - batch the initial snapshot and transaction-log catch-up into Feldera transactions (see below).
/// * `always` - all updates generated by the connector are broken up into transactions, both during the initial snapshot
///   and when following the transaction log.
///
/// # How table snapshot is ingested using transactions
///
/// If the connector is configured in the `snapshot` or `snapshot_and_follow` mode, and its
/// `transaction_mode` is set to `snapshot`, `catchup`, or `always`, it will ingest the snapshot of the
/// table in one or several transactions. The exact behavior depends on the value of the `timestamp_column` option.
/// If `timestamp_column` is not set, `catchup` and `always` ingest the snapshot in one Feldera transaction; `snapshot`
/// mode uses the same one-transaction behavior when `timestamp_column` is not set.
///
/// If `timestamp_column` is set, the connector will ingest the snapshot of the table in a series of batches, one for
/// each timestamp range of width equal to the `LATENESS` attribute of the `timestamp_column`. Each range will be
/// ingested in a separate transaction.
///
/// See `timestamp_column` documentation for more details.
///
/// # How transaction log is ingested using transactions
///
/// If the connector is configured in the `follow`, `snapshot_and_follow`, or `cdc` mode, and its
/// `transaction_mode` is set to `catchup`, it ingests the transaction log in batches. When it starts a
/// Feldera transaction, it reads the latest available version of the Delta table (capped by `end_version` if set) and
/// ingests all log entries up to and including that version in a single Feldera transaction before committing. It then
/// repeats for subsequent versions as they appear in the log.
///
/// The `catchup` mode offers the most efficient way for the connector to process the entire contents
/// of the table during backfill and keep up with table changes in steady state.
///
/// If `transaction_mode` is set to `always`, the connector ingests the transaction log in a series of transactions,
/// generating exactly one Feldera transaction for each entry in the table's transaction log. Feldera transaction
/// boundaries then match the transaction boundaries of the Delta Lake table: as long as the table keeps changing, each log
/// entry initiates a new Feldera transaction. Use `always` when that one-to-one alignment is required.
#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema, Default)]
pub enum DeltaTableTransactionMode {
    #[default]
    #[serde(rename = "none")]
    None,
    #[serde(rename = "snapshot")]
    Snapshot,
    #[serde(rename = "always")]
    Always,
    #[serde(rename = "catchup")]
    Catchup,
}

/// Delta table input connector configuration.
#[derive(Debug, Clone, Eq, PartialEq, Deserialize, Serialize, ToSchema)]
pub struct DeltaTableReaderConfig {
    /// Table URI.
    ///
    /// Example: "s3://feldera-fraud-detection-data/demographics_train"
    pub uri: String,

    /// Table read mode.
    pub mode: DeltaTableIngestMode,

    /// Transaction mode.
    ///
    /// Determines how the connector breaks up its input into transactions.
    ///
    /// See [`DeltaTableTransactionMode`]. When omitted in JSON config, defaults to
    /// [`DeltaTableTransactionMode::None`].
    #[serde(default)]
    pub transaction_mode: DeltaTableTransactionMode,

    /// Table column that serves as an event timestamp.
    ///
    /// When this option is specified, and `mode` is one of `snapshot` or `snapshot_and_follow`,
    /// table rows are ingested in the timestamp order, respecting the
    /// [`LATENESS`](https://docs.feldera.com/sql/streaming#lateness-expressions)
    /// property of the column: each ingested row has a timestamp no more than `LATENESS`
    /// time units earlier than the most recent timestamp of any previously ingested row.
    /// The ingestion is performed by partitioning the table into timestamp ranges of width
    /// `LATENESS`. Each range is processed sequentially, in increasing timestamp order.
    ///
    /// # Example
    ///
    /// Consider a table with timestamp column of type `TIMESTAMP` and lateness attribute
    /// `INTERVAL 1 DAY`. Assuming that the oldest timestamp in the table is
    /// `2024-01-01T00:00:00``, the connector will fetch all records with timestamps
    /// from `2024-01-01`, then all records for `2024-01-02`, `2024-01-03`, etc., until all records
    /// in the table have been ingested.
    ///
    /// # Requirements
    ///
    /// * The timestamp column must be of a supported type: integer, `DATE`, or `TIMESTAMP`.
    /// * The timestamp column must be declared with non-zero `LATENESS`.
    /// * For efficient ingest, the table must be optimized for timestamp-based
    ///   queries using partitioning, Z-ordering, or liquid clustering.
    pub timestamp_column: Option<String>,

    /// Optional row filter.
    ///
    /// When specified, only rows that satisfy the filter condition are read from the delta table.
    /// The condition must be a valid SQL Boolean expression that can be used in
    /// the `where` clause of the `select * from my_table where ...` query.
    pub filter: Option<String>,

    /// Don't read unused columns from the Delta table.
    ///
    /// When set to `true`, this option instructs the connector to avoid reading
    /// columns from the Delta table that are not used in any view definitions.
    /// To be skipped, the columns must be either nullable or have default
    /// values. This can improve ingestion performance, especially for wide
    /// tables.
    ///
    /// Note: The simplest way to exclude unused columns is to omit them from the Feldera SQL table
    /// declaration. The connector never reads columns that aren't declared in the SQL schema.
    /// Additionally, the SQL compiler emits warnings for declared but unused columns—use these as
    /// a guide to optimize your schema.
    #[serde(default)]
    pub skip_unused_columns: bool,

    /// Optional snapshot filter.
    ///
    /// This option is only valid when `mode` is set to `snapshot` or `snapshot_and_follow`.
    ///
    /// When specified, only rows that satisfy the filter condition are included in the
    /// snapshot.  The condition must be a valid SQL Boolean expression that can be used in
    /// the `where` clause of the `select * from snapshot where ...` query.
    ///
    /// Unlike the `filter` option, which applies to all records retrieved from the table, this
    /// filter only applies to rows in the initial snapshot of the table.
    /// For instance, it can be used to specify the range of event times to include in the snapshot,
    /// e.g.: `ts BETWEEN TIMESTAMP '2005-01-01 00:00:00' AND TIMESTAMP '2010-12-31 23:59:59'`.
    ///
    /// This option can be used together with the `filter` option. During the initial snapshot,
    /// only rows that satisfy both `filter` and `snapshot_filter` are retrieved from the Delta table.
    /// When subsequently following changes in the the transaction log (`mode = snapshot_and_follow`),
    /// all rows that meet the `filter` condition are ingested, regardless of `snapshot_filter`.
    pub snapshot_filter: Option<String>,

    /// Optional table version.
    ///
    /// When this option is set, the connector finds and opens the specified version of the table.
    /// In `snapshot` and `snapshot_and_follow` modes, it retrieves the snapshot of this version of
    /// the table.  In `follow`, `snapshot_and_follow`, and `cdc` modes, it follows transaction log records
    /// **after** this version.
    ///
    /// Note: at most one of `version` and `datetime` options can be specified.
    /// When neither of the two options is specified, the latest committed version of the table
    /// is used.
    #[serde(alias = "start_version")]
    pub version: Option<i64>,

    /// Optional timestamp for the snapshot in the ISO-8601/RFC-3339 format, e.g.,
    /// "2024-12-09T16:09:53+00:00".
    ///
    /// When this option is set, the connector finds and opens the version of the table as of the
    /// specified point in time (based on the server time recorded in the transaction log, not the
    /// event time encoded in the data).  In `snapshot` and `snapshot_and_follow` modes, it
    /// retrieves the snapshot of this version of the table.  In `follow`, `snapshot_and_follow`, and
    /// `cdc` modes, it follows transaction log records **after** this version.
    ///
    /// Note: at most one of `version` and `datetime` options can be specified.
    /// When neither of the two options is specified, the latest committed version of the table
    /// is used.
    #[serde(alias = "start_datetime")]
    pub datetime: Option<String>,

    /// Optional final table version.
    ///
    /// Valid only when the connector is configured in `follow`, `snapshot_and_follow`, or `cdc` mode.
    ///
    /// When set, the connector will stop scanning the table’s transaction log after reaching this version or any greater version.
    /// This bound is inclusive: if the specified version appears in the log, it will be processed before signaling end-of-input.
    pub end_version: Option<i64>,

    /// A predicate that determines whether the record represents a deletion.
    ///
    /// This setting is only valid in the `cdc` mode. It specifies a predicate applied to
    /// each row in the Delta table to determine whether the row represents a deletion event.
    /// Its value must be a valid Boolean SQL expression that can be used in a query of the
    /// form `SELECT * from <table> WHERE <cdc_delete_filter>`.
    pub cdc_delete_filter: Option<String>,

    /// An expression that determines the ordering of updates in the Delta table.
    ///
    /// This setting is only valid in the `cdc` mode. It specifies a predicate applied to
    /// each row in the Delta table to determine the order in which updates in the table should
    /// be applied. Its value must be a valid SQL expression that can be used in a query of the
    /// form `SELECT * from <table> ORDER BY <cdc_order_by>`.
    pub cdc_order_by: Option<String>,

    /// The number of parallel parsing tasks the connector uses to process data read from the
    /// table. Increasing this value can enhance performance by allowing more concurrent processing.
    /// Recommended range: 1–10. The default is 4.
    #[serde(default = "default_num_parsers")]
    pub num_parsers: u32,

    /// Maximum number of concurrent object store reads performed by all Delta Lake connectors.
    ///
    /// This setting is used to limit the number of concurrent reads of the object store in a
    /// pipeline with a large number of Delta Lake connectors. When multiple connectors are simultaneously
    /// reading from the object store, this can lead to transport timeouts.
    ///
    /// When enabled, this setting limits the number of concurrent reads across all connectors.
    /// This is a global setting that affects all Delta Lake connectors, and not just the connector
    /// where it is specified. It should therefore be used at most once in a pipeline.  If multiple
    /// connectors specify this setting, they must all use the same value.
    ///
    /// The default value is 6.
    pub max_concurrent_readers: Option<u32>,

    /// Enable verbose logging.
    ///
    /// When enabled, the connector will log detailed information at INFO level.
    ///
    /// Supported values:
    /// * 0 - no verbose logging
    /// * 1 - log all Delta log entries in follow and cdc modes.
    /// * >1 - reserved for future use
    #[serde(default)]
    pub verbose: u32,

    /// Maximum number of retries for failed object store operations.
    ///
    /// Controls how many times the connector retries high-level storage operations,
    /// such as reading a Delta log entry or a Parquet file.
    ///
    /// This is in addition to lower-level retries (e.g., individual S3 operation retries governed
    /// by storage options like `retry_timeout`). If those retries are exhausted
    /// or the failure is otherwise unrecoverable at the storage layer, the
    /// connector retries the entire operation.
    ///
    /// Defaults to unlimited retries. Set to 0 to disable retries.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_retries: Option<u32>,

    /// Storage options for configuring backend object store.
    ///
    /// For specific options available for different storage backends, see:
    /// * [Azure options](https://docs.rs/object_store/latest/object_store/azure/enum.AzureConfigKey.html)
    /// * [Amazon S3 options](https://docs.rs/object_store/latest/object_store/aws/enum.AmazonS3ConfigKey.html)
    /// * [Google Cloud Storage options](https://docs.rs/object_store/latest/object_store/gcp/enum.GoogleConfigKey.html)
    #[serde(flatten)]
    pub object_store_config: HashMap<String, String>,
}

impl DeltaTableReaderConfig {
    pub fn max_retries(&self) -> u32 {
        self.max_retries.unwrap_or(u32::MAX)
    }
}

#[cfg(test)]
#[test]
fn test_delta_reader_config_serde() {
    let config_str = r#"{
            "uri": "protocol:/path/to/somewhere",
            "datetime": "2010-12-31 00:00:00Z",
            "snapshot_filter": "ts BETWEEN '2005-01-01 00:00:00' AND '2010-12-31 23:59:59'",
            "timestamp_column": "ts",
            "customoption1": "val1",
            "customoption2": "val2",
            "mode": "follow",
            "cdc_delete_filter": null,
            "cdc_order_by": null
        }"#;

    let config = serde_json::from_str::<DeltaTableReaderConfig>(config_str).unwrap();

    let serialized_config = serde_json::to_string(&config).unwrap();
    let expected = r#"{"uri":"protocol:/path/to/somewhere","mode":"follow","transaction_mode":"none","timestamp_column":"ts","filter":null,"skip_unused_columns":false,"snapshot_filter":"ts BETWEEN '2005-01-01 00:00:00' AND '2010-12-31 23:59:59'","version":null,"datetime":"2010-12-31 00:00:00Z","end_version":null,"cdc_delete_filter":null,"cdc_order_by":null,"num_parsers":4,"max_concurrent_readers":null,"customoption1":"val1","customoption2":"val2","verbose":0}"#;
    assert_eq!(
        serde_json::from_str::<serde_json::Value>(&serialized_config).unwrap(),
        serde_json::from_str::<serde_json::Value>(expected).unwrap()
    );
}

#[cfg(test)]
#[test]
fn test_delta_transaction_mode_catchup_serde() {
    let config: DeltaTableReaderConfig = serde_json::from_str(
        r#"{"uri":"s3://bucket/t","mode":"follow","transaction_mode":"catchup"}"#,
    )
    .unwrap();
    assert_eq!(config.transaction_mode, DeltaTableTransactionMode::Catchup);
}

#[cfg(test)]
#[test]
fn test_delta_table_ingest_mode_display() {
    assert_eq!(DeltaTableIngestMode::Snapshot.to_string(), "snapshot");
    assert_eq!(DeltaTableIngestMode::Follow.to_string(), "follow");
    assert_eq!(
        DeltaTableIngestMode::SnapshotAndFollow.to_string(),
        "snapshot_and_follow"
    );
    assert_eq!(DeltaTableIngestMode::Cdc.to_string(), "cdc");
}

#[cfg(test)]
mod log_retention_tests {
    use super::*;

    fn make_config(log_retention: Option<&str>) -> DeltaTableWriterConfig {
        DeltaTableWriterConfig {
            uri: "memory://".to_string(),
            mode: DeltaTableWriteMode::default(),
            checkpoint_interval: None,
            log_retention_duration: log_retention.map(str::to_string),
            enable_expired_log_cleanup: None,
            max_retries: None,
            threads: None,
            object_store_config: HashMap::new(),
        }
    }

    #[test]
    fn validate_delta_interval_accepts_known_units() {
        for unit in [
            "nanosecond",
            "nanoseconds",
            "microsecond",
            "microseconds",
            "millisecond",
            "milliseconds",
            "second",
            "seconds",
            "minute",
            "minutes",
            "hour",
            "hours",
            "day",
            "days",
            "week",
            "weeks",
        ] {
            let s = format!("interval 5 {unit}");
            assert!(validate_delta_interval(&s).is_ok(), "failed for {s}");
        }
    }

    #[test]
    fn validate_delta_interval_accepts_zero() {
        // Zero is a valid duration; users who want to disable cleanup do so via
        // `enable_expired_log_cleanup = false`.
        assert!(validate_delta_interval("interval 0 seconds").is_ok());
        assert!(validate_delta_interval("interval 0 days").is_ok());
    }

    #[test]
    fn validate_delta_interval_rejects_malformed() {
        for bad in [
            "",
            "30 days",
            "interval",
            "interval 30",
            "interval days",
            "interval -5 days",
            "interval foo days",
            "interval 30 fortnights",
            "interval 5 dayss",
            "interval 1 hours extra",
            // delta-rs's parser is lowercase-only, so we reject anything else
            // to fail fast rather than letting it silently fall back at cleanup time.
            "INTERVAL 30 days",
            "Interval 30 days",
            "interval 30 DAYS",
        ] {
            assert!(
                validate_delta_interval(bad).is_err(),
                "expected error for {bad:?}"
            );
        }
    }

    #[test]
    fn validate_propagates_interval_errors() {
        let cfg = make_config(Some("not an interval"));
        let err = cfg.validate().unwrap_err();
        assert!(err.contains("log_retention_duration"), "unexpected: {err}");
    }

    #[test]
    fn validate_accepts_well_formed_interval() {
        let cfg = make_config(Some("interval 30 days"));
        assert!(cfg.validate().is_ok());
    }

    #[test]
    fn validate_accepts_unset_interval() {
        let cfg = make_config(None);
        assert!(cfg.validate().is_ok());
    }
}

impl DeltaTableReaderConfig {
    /// `true` if the configuration requires taking an initial snapshot of the table.
    pub fn snapshot(&self) -> bool {
        matches!(
            &self.mode,
            DeltaTableIngestMode::Snapshot | DeltaTableIngestMode::SnapshotAndFollow
        )
    }

    /// `true` if the configuration requires following the transaction log of the table
    /// (possibly after taking an initial snapshot).
    pub fn follow(&self) -> bool {
        matches!(
            &self.mode,
            DeltaTableIngestMode::SnapshotAndFollow
                | DeltaTableIngestMode::Follow
                | DeltaTableIngestMode::Cdc
        )
    }

    pub fn is_cdc(&self) -> bool {
        matches!(&self.mode, DeltaTableIngestMode::Cdc)
    }
}
