buoyant_kernel 0.22.0

//! For now we just use simple functions to deserialize table properties from strings. This allows
//! us to relatively simply implement the functionality described in the protocol and expose
//! 'simple' types to the user in the [`TableProperties`] struct. E.g. we can expose a `bool`
//! directly instead of a `BoolConfig` type that we implement `Deserialize` for.
use std::num::NonZero;
use std::time::Duration;

use tracing::warn;

use super::*;
use crate::expressions::ColumnName;
use crate::table_features::ColumnMappingMode;
use crate::utils::require;

const SECONDS_PER_MINUTE: u64 = 60;
const SECONDS_PER_HOUR: u64 = 60 * SECONDS_PER_MINUTE;
const SECONDS_PER_DAY: u64 = 24 * SECONDS_PER_HOUR;
const SECONDS_PER_WEEK: u64 = 7 * SECONDS_PER_DAY;

impl<K, V, I> From<I> for TableProperties
where
    I: IntoIterator<Item = (K, V)>,
    K: AsRef<str> + Into<String>,
    V: AsRef<str> + Into<String>,
{
    fn from(unparsed: I) -> Self {
        let mut props = TableProperties::default();
        let unparsed = unparsed.into_iter().filter(|(k, v)| {
            // Only keep elements that fail to parse
            try_parse(&mut props, k.as_ref(), v.as_ref()).is_none()
        });
        props.unknown_properties = unparsed.map(|(k, v)| (k.into(), v.into())).collect();
        props
    }
}

// attempt to parse a key-value pair into a `TableProperties` struct. Returns Some(()) if the key
// was successfully parsed, and None otherwise.
fn try_parse(props: &mut TableProperties, k: &str, v: &str) -> Option<()> {
    // Table property key constants are imported via `use super::*` at the top of this file.

    // NOTE!! we do Some(parse(v)?) instead of just parse(v) because we want to return None if the
    // parsing fails. If we simply call 'parse(v)', then we would (incorrectly) return Some(()) and
    // just set the property to None.
    match k {
        APPEND_ONLY => props.append_only = Some(parse_bool(v)?),
        AUTO_COMPACT => props.auto_compact = Some(parse_bool(v)?),
        OPTIMIZE_WRITE => props.optimize_write = Some(parse_bool(v)?),
        CHECKPOINT_INTERVAL => props.checkpoint_interval = Some(parse_positive_int(v)?),
        CHECKPOINT_WRITE_STATS_AS_JSON => {
            props.checkpoint_write_stats_as_json = Some(parse_bool(v)?)
        }
        CHECKPOINT_WRITE_STATS_AS_STRUCT => {
            props.checkpoint_write_stats_as_struct = Some(parse_bool(v)?)
        }
        COLUMN_MAPPING_MODE => props.column_mapping_mode = ColumnMappingMode::try_from(v).ok(),
        COLUMN_MAPPING_MAX_COLUMN_ID => {
            props.column_mapping_max_column_id = Some(parse_non_negative(v)?)
        }
        DATA_SKIPPING_NUM_INDEXED_COLS => {
            props.data_skipping_num_indexed_cols = DataSkippingNumIndexedCols::try_from(v).ok()
        }
        DATA_SKIPPING_STATS_COLUMNS => {
            props.data_skipping_stats_columns = Some(parse_column_names(v)?)
        }
        DELETED_FILE_RETENTION_DURATION => {
            props.deleted_file_retention_duration = Some(parse_interval(v)?)
        }
        ENABLE_CHANGE_DATA_FEED => props.enable_change_data_feed = Some(parse_bool(v)?),
        ENABLE_DELETION_VECTORS => props.enable_deletion_vectors = Some(parse_bool(v)?),
        ENABLE_TYPE_WIDENING => props.enable_type_widening = Some(parse_bool(v)?),
        ENABLE_ICEBERG_COMPAT_V1 => props.enable_iceberg_compat_v1 = Some(parse_bool(v)?),
        ENABLE_ICEBERG_COMPAT_V2 => props.enable_iceberg_compat_v2 = Some(parse_bool(v)?),
        ENABLE_ICEBERG_COMPAT_V3 => props.enable_iceberg_compat_v3 = Some(parse_bool(v)?),
        ISOLATION_LEVEL => props.isolation_level = IsolationLevel::try_from(v).ok(),
        LOG_RETENTION_DURATION => props.log_retention_duration = Some(parse_interval(v)?),
        ENABLE_EXPIRED_LOG_CLEANUP => props.enable_expired_log_cleanup = Some(parse_bool(v)?),
        RANDOMIZE_FILE_PREFIXES => props.randomize_file_prefixes = Some(parse_bool(v)?),
        RANDOM_PREFIX_LENGTH => props.random_prefix_length = Some(parse_positive_int(v)?),
        SET_TRANSACTION_RETENTION_DURATION => {
            props.set_transaction_retention_duration = Some(parse_interval(v)?)
        }
        TARGET_FILE_SIZE => props.target_file_size = Some(parse_positive_int(v)?),
        TUNE_FILE_SIZES_FOR_REWRITES => props.tune_file_sizes_for_rewrites = Some(parse_bool(v)?),
        CHECKPOINT_POLICY => props.checkpoint_policy = CheckpointPolicy::try_from(v).ok(),
        ENABLE_ROW_TRACKING => props.enable_row_tracking = Some(parse_bool(v)?),
        MATERIALIZED_ROW_ID_COLUMN_NAME => {
            props.materialized_row_id_column_name = Some(v.to_string())
        }
        MATERIALIZED_ROW_COMMIT_VERSION_COLUMN_NAME => {
            props.materialized_row_commit_version_column_name = Some(v.to_string())
        }
        ROW_TRACKING_SUSPENDED => props.row_tracking_suspended = Some(parse_bool(v)?),
        PARQUET_FORMAT_VERSION => props.parquet_format_version = Some(v.to_string()),
        PARQUET_COMPRESSION_CODEC => {
            // Preserve unrecognized codecs in `unknown_properties` so connectors can fall back
            // themselves.
            props.parquet_compression_codec = Some(ParquetCompressionCodec::try_from(v).ok()?)
        }
        ENABLE_IN_COMMIT_TIMESTAMPS => props.enable_in_commit_timestamps = Some(parse_bool(v)?),
        IN_COMMIT_TIMESTAMP_ENABLEMENT_VERSION => {
            props.in_commit_timestamp_enablement_version = Some(parse_non_negative(v)?)
        }
        IN_COMMIT_TIMESTAMP_ENABLEMENT_TIMESTAMP => {
            props.in_commit_timestamp_enablement_timestamp = Some(parse_non_negative(v)?)
        }
        _ => return None,
    }
    Some(())
}

/// Deserialize a string representing a positive (> 0) integer into an `Option<u64>`. Returns `Some`
/// if successfully parses, and `None` otherwise.
pub(crate) fn parse_positive_int(s: &str) -> Option<NonZero<u64>> {
    // parse as non-negative and verify the result is non-zero
    NonZero::new(parse_non_negative(s)?)
}

/// Deserialize a string representing a non-negative integer into an `Option<u64>`. Returns `Some`
/// if successfully parses, and `None` otherwise.
pub(crate) fn parse_non_negative<T>(s: &str) -> Option<T>
where
    i64: TryInto<T>,
{
    // parse to i64 since java doesn't even allow u64
    let n: i64 = s.parse().ok().filter(|&i| i >= 0)?;
    n.try_into().ok()
}

/// Deserialize a string representing a boolean into an `Option<bool>`. Returns `Some` if
/// successfully parses, and `None` otherwise.
pub(crate) fn parse_bool(s: &str) -> Option<bool> {
    match s {
        "true" => Some(true),
        "false" => Some(false),
        _ => None,
    }
}

/// Deserialize a comma-separated list of column names into an `Option<Vec<ColumnName>>`. Returns
/// `Some` if successfully parses, and `None` otherwise.
pub(crate) fn parse_column_names(s: &str) -> Option<Vec<ColumnName>> {
    ColumnName::parse_column_name_list(s)
        .inspect_err(|e| warn!("column name list failed to parse: {e}"))
        .ok()
}

/// Deserialize an interval string of the form "interval 5 days" into an `Option<Duration>`.
/// Returns `Some` if successfully parses, and `None` otherwise.
pub(crate) fn parse_interval(s: &str) -> Option<Duration> {
    parse_interval_impl(s).ok()
}

#[derive(thiserror::Error, Debug)]
pub enum ParseIntervalError {
    /// The input string is not a valid interval
    #[error("'{0}' is not an interval")]
    NotAnInterval(String),
    /// Couldn't parse the input string as an integer
    #[error("Unable to parse '{0}' as an integer")]
    ParseIntError(String),
    /// Negative intervals aren't supported
    #[error("Interval '{0}' cannot be negative")]
    NegativeInterval(String),
    /// Unsupported interval
    #[error("Unsupported interval '{0}'")]
    UnsupportedInterval(String),
    /// Unknown unit
    #[error("Unknown interval unit '{0}'")]
    UnknownUnit(String),
}

/// This is effectively a simpler version of spark's `CalendarInterval` parser. See spark's
/// `stringToInterval`:
/// https://github.com/apache/spark/blob/5a57efdcee9e6569d8de4272bda258788cf349e3/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkIntervalUtils.scala#L134
///
/// Notably we don't support months nor years, nor do we support fractional values, and negative
/// intervals aren't supported.
///
/// For now this is adapted from delta-rs' `parse_interval` function:
/// https://github.com/delta-io/delta-rs/blob/d4f18b3ae9d616e771b5d0e0fa498d0086fd91eb/crates/core/src/table/config.rs#L474
///
/// See issue delta-kernel-rs/#507 for details: https://github.com/delta-io/delta-kernel-rs/issues/507
fn parse_interval_impl(value: &str) -> Result<Duration, ParseIntervalError> {
    let mut it = value.split_whitespace();
    if it.next() != Some("interval") {
        return Err(ParseIntervalError::NotAnInterval(value.to_string()));
    }
    let number = it
        .next()
        .ok_or_else(|| ParseIntervalError::NotAnInterval(value.to_string()))?;
    let number: i64 = number
        .parse()
        .map_err(|_| ParseIntervalError::ParseIntError(number.into()))?;

    // TODO(zach): spark allows negative intervals, but we don't
    require!(
        number >= 0,
        ParseIntervalError::NegativeInterval(value.to_string())
    );

    // convert to u64 since Duration expects it
    let number = number as u64; // non-negative i64 => always safe

    let duration = match it
        .next()
        .ok_or_else(|| ParseIntervalError::NotAnInterval(value.to_string()))?
    {
        "nanosecond" | "nanoseconds" => Duration::from_nanos(number),
        "microsecond" | "microseconds" => Duration::from_micros(number),
        "millisecond" | "milliseconds" => Duration::from_millis(number),
        "second" | "seconds" => Duration::from_secs(number),
        "minute" | "minutes" => Duration::from_secs(number * SECONDS_PER_MINUTE),
        "hour" | "hours" => Duration::from_secs(number * SECONDS_PER_HOUR),
        "day" | "days" => Duration::from_secs(number * SECONDS_PER_DAY),
        "week" | "weeks" => Duration::from_secs(number * SECONDS_PER_WEEK),
        unit @ ("month" | "months") => {
            return Err(ParseIntervalError::UnsupportedInterval(unit.to_string()));
        }
        unit => {
            return Err(ParseIntervalError::UnknownUnit(unit.to_string()));
        }
    };

    Ok(duration)
}

#[cfg(test)]
mod tests {
    use ParquetCompressionCodec::*;

    use super::*;

    #[test]
    fn test_parse_bool() {
        assert!(parse_bool("true").unwrap());
        assert!(!parse_bool("false").unwrap());
        assert_eq!(parse_bool("whatever"), None);
    }

    #[test]
    fn test_parse_int() {
        assert_eq!(parse_positive_int("12").unwrap().get(), 12);
        assert_eq!(parse_positive_int("0"), None);
        assert_eq!(parse_positive_int("-12"), None);
        assert_eq!(parse_non_negative::<u64>("12").unwrap(), 12);
        assert_eq!(parse_non_negative::<u64>("0").unwrap(), 0);
        assert_eq!(parse_non_negative::<u64>("-12"), None);
        assert_eq!(parse_non_negative::<i64>("12").unwrap(), 12);
        assert_eq!(parse_non_negative::<i64>("0").unwrap(), 0);
        assert_eq!(parse_non_negative::<i64>("-12"), None);
    }

    #[test]
    fn test_parse_interval() {
        assert_eq!(
            parse_interval("interval 123 nanosecond").unwrap(),
            Duration::from_nanos(123)
        );

        assert_eq!(
            parse_interval("interval 123 nanoseconds").unwrap(),
            Duration::from_nanos(123)
        );

        assert_eq!(
            parse_interval("interval 123 microsecond").unwrap(),
            Duration::from_micros(123)
        );

        assert_eq!(
            parse_interval("interval 123 microseconds").unwrap(),
            Duration::from_micros(123)
        );

        assert_eq!(
            parse_interval("interval 123 millisecond").unwrap(),
            Duration::from_millis(123)
        );

        assert_eq!(
            parse_interval("interval 123 milliseconds").unwrap(),
            Duration::from_millis(123)
        );

        assert_eq!(
            parse_interval("interval 123 second").unwrap(),
            Duration::from_secs(123)
        );

        assert_eq!(
            parse_interval("interval 123 seconds").unwrap(),
            Duration::from_secs(123)
        );

        assert_eq!(
            parse_interval("interval 123 minute").unwrap(),
            Duration::from_secs(123 * 60)
        );

        assert_eq!(
            parse_interval("interval 123 minutes").unwrap(),
            Duration::from_secs(123 * 60)
        );

        assert_eq!(
            parse_interval("interval 123 hour").unwrap(),
            Duration::from_secs(123 * 3600)
        );

        assert_eq!(
            parse_interval("interval 123 hours").unwrap(),
            Duration::from_secs(123 * 3600)
        );

        assert_eq!(
            parse_interval("interval 123 day").unwrap(),
            Duration::from_secs(123 * 86400)
        );

        assert_eq!(
            parse_interval("interval 123 days").unwrap(),
            Duration::from_secs(123 * 86400)
        );

        assert_eq!(
            parse_interval("interval 123 week").unwrap(),
            Duration::from_secs(123 * 604800)
        );

        assert_eq!(
            parse_interval("interval 123 week").unwrap(),
            Duration::from_secs(123 * 604800)
        );
    }

    #[test]
    fn test_invalid_parse_interval() {
        assert_eq!(
            parse_interval_impl("whatever").err().unwrap().to_string(),
            "'whatever' is not an interval".to_string()
        );

        assert_eq!(
            parse_interval_impl("interval").err().unwrap().to_string(),
            "'interval' is not an interval".to_string()
        );

        assert_eq!(
            parse_interval_impl("interval 2").err().unwrap().to_string(),
            "'interval 2' is not an interval".to_string()
        );

        assert_eq!(
            parse_interval_impl("interval 2 months")
                .err()
                .unwrap()
                .to_string(),
            "Unsupported interval 'months'".to_string()
        );

        assert_eq!(
            parse_interval_impl("interval 2 years")
                .err()
                .unwrap()
                .to_string(),
            "Unknown interval unit 'years'".to_string()
        );

        assert_eq!(
            parse_interval_impl("interval two years")
                .err()
                .unwrap()
                .to_string(),
            "Unable to parse 'two' as an integer".to_string()
        );

        assert_eq!(
            parse_interval_impl("interval -25 hours")
                .err()
                .unwrap()
                .to_string(),
            "Interval 'interval -25 hours' cannot be negative".to_string()
        );
    }

    #[test]
    fn test_parse_compression_codec_accepts_all_variants_case_insensitively() {
        for (input, expected) in [
            ("uncompressed", Uncompressed),
            ("UNCOMPRESSED", Uncompressed),
            ("UnCoMpReSsEd", Uncompressed),
            ("none", Uncompressed),
            ("NONE", Uncompressed),
            ("nOnE", Uncompressed),
            ("snappy", Snappy),
            ("SNAPPY", Snappy),
            ("sNaPpY", Snappy),
            ("gzip", Gzip),
            ("GZIP", Gzip),
            ("gZiP", Gzip),
            ("lz4", Lz4),
            ("LZ4", Lz4),
            ("lZ4", Lz4),
            ("lz4_raw", Lz4Raw),
            ("LZ4_RAW", Lz4Raw),
            ("Lz4_Raw", Lz4Raw),
            ("zstd", Zstd),
            ("ZSTD", Zstd),
            ("zStD", Zstd),
        ] {
            let props = TableProperties::from([(PARQUET_COMPRESSION_CODEC, input)]);
            assert_eq!(props.parquet_compression_codec, Some(expected), "{input}");
            assert!(props.unknown_properties.is_empty(), "{input}");
        }
    }

    #[test]
    fn test_parse_compression_codec_unknown_falls_through_to_unknown_properties() {
        let props = TableProperties::from([(PARQUET_COMPRESSION_CODEC, "brotli")]);
        assert_eq!(props.parquet_compression_codec, None);
        assert_eq!(
            props.unknown_properties.get(PARQUET_COMPRESSION_CODEC),
            Some(&"brotli".to_string())
        );
    }

    #[test]
    fn test_compression_codec_round_trips_to_canonical_protocol_string() {
        // Each variant must Display-format to its Delta-protocol canonical name (lowercased,
        // snake_case). The `Uncompressed` variant has `none` as a parser-only alias; its
        // canonical output is `uncompressed`.
        for (variant, expected) in [
            (Zstd, "zstd"),
            (Uncompressed, "uncompressed"),
            (Snappy, "snappy"),
            (Gzip, "gzip"),
            (Lz4, "lz4"),
            (Lz4Raw, "lz4_raw"),
        ] {
            assert_eq!(variant.to_string(), expected);
            assert_eq!(<&'static str>::from(variant), expected);
            // And the canonical string parses back to the same variant.
            assert_eq!(
                ParquetCompressionCodec::try_from(expected).unwrap(),
                variant
            );
        }
    }

    #[test]
    fn test_compression_codec_or_default_falls_back_to_zstd_when_absent() {
        // Absent: applies the spec-recommended Zstd fallback.
        let props = TableProperties::default();
        assert_eq!(props.compression_codec_or_default(), Zstd);

        // Present: returns the parsed value as-is.
        let props = TableProperties::from([(PARQUET_COMPRESSION_CODEC, "snappy")]);
        assert_eq!(props.compression_codec_or_default(), Snappy);
    }
}