timelake 0.7.0

//! Changing a table's settings after creation (`timelake alter`, SPEC §21.9).

use std::collections::HashMap;

use iceberg::spec::TableMetadataBuilder;
use timelake_format::Contract;
use timelake_format::settings::{self, Level};

use crate::commit::publish;
use crate::error::Result;
use crate::lake::metadata_uri;
use crate::table::Table;

impl Table {
    /// Sets table properties, or removes them with `None` (back to the default), in one commit of
    /// new metadata (no new snapshot). Timelake settings may be named in any spelling, with or
    /// without the `timelake.` prefix (`read-dedup`, `timelake.read_dedup`); only those the registry
    /// allows at the table level can change ([`settings::SETTINGS`]). Keys with a dot outside the
    /// `timelake.` prefix are Iceberg properties and pass through.
    ///
    /// ```no_run
    /// # async fn demo(table: timelake::Table) -> timelake::Result<()> {
    /// let table = table.set_properties(&[("read-dedup", Some("false")), ("retention", None)]).await?;
    /// # Ok(()) }
    /// ```
    pub async fn set_properties(&self, changes: &[(&str, Option<&str>)]) -> Result<Table> {
        let mut set = HashMap::new();
        let mut remove = vec![];
        for &(key, value) in changes {
            let key = property_key(key)?;
            match value {
                Some(v) => {
                    set.insert(key, v.to_string());
                }
                None => remove.push(key),
            }
        }
        let set = settings::check_properties(&set, Level::Table)?;
        for key in &remove {
            if let Some(s) =
                settings::find(key).filter(|s| key.starts_with(settings::PREFIX) && !s.allows(Level::Table))
            {
                return Err(settings::not_settable(s, Level::Table).into());
            }
        }
        let mut merged = self.metadata().properties().clone();
        for key in &remove {
            merged.remove(key);
        }
        merged.extend(set.clone());
        let contract = Contract::from_properties(&merged)?;
        self.lake().check_head_format(&contract.head_format)?;
        write_properties(self, &set, &remove).await
    }
}

/// The property a user's key names: a Timelake setting in its canonical spelling (a bare
/// setting name counts), or any other key as it is.
pub(crate) fn property_key(key: &str) -> Result<String> {
    let trimmed = key.trim();
    if trimmed.to_ascii_lowercase().starts_with(settings::PREFIX) || !trimmed.contains('.') {
        return settings::find(trimmed)
            .map(settings::Setting::property)
            .ok_or_else(|| settings::unknown(trimmed).into());
    }
    Ok(trimmed.to_string())
}

/// Commits new table properties without checking them (no new snapshot): for Timelake's own
/// changes, such as `timelake upgrade` recording the format.
pub(crate) async fn write_properties(table: &Table, set: &[(String, String)], remove: &[String]) -> Result<Table> {
    update_properties(table, |_| (set.to_vec(), remove.to_vec())).await
}

/// Like [`write_properties`], with the change computed from the table version it applies to
/// (again after a concurrent commit), such as adding a name to a list.
pub(crate) async fn update_properties(
    table: &Table,
    change: impl Fn(&Table) -> (Vec<(String, String)>, Vec<String>),
) -> Result<Table> {
    let mut base = table.clone();
    loop {
        let (set, remove) = change(&base);
        let previous = Some(metadata_uri(base.location(), base.version()));
        let metadata = TableMetadataBuilder::new_from_metadata(base.metadata().clone(), previous)
            .set_properties(set.into_iter().collect())?
            .remove_properties(&remove)?
            .build()?
            .metadata;
        match publish(&base, metadata, base.known_files().cloned()).await? {
            Some(table) => return Ok(table),
            None => base = base.refresh().await?,
        }
    }
}

#[cfg(test)]
mod tests {
    use crate::property;
    use crate::testing::{faulty_lake, opts, schema};

    #[tokio::test]
    async fn settings_change_after_creation() {
        let (lake, _store) = faulty_lake();
        let t = lake.create_table("iot.r", &schema(), opts()).await.unwrap();
        assert!(t.contract().read_dedup);
        let t = t.set_properties(&[("read_dedup", Some("false")), ("timelake.retention", Some("30d"))]).await.unwrap();
        assert!(!t.contract().read_dedup);
        assert_eq!(t.metadata().properties().get(property::RETENTION).map(String::as_str), Some("30d"));
        let t = t.set_properties(&[("retention", None)]).await.unwrap();
        assert_eq!(t.contract().retention, None);
        assert_eq!(lake.table("iot.r").await.unwrap().version(), t.version());

        for (change, msg) in [
            (("chunk-interval", Some("hour")), "fixed when the table is created"),
            (("time-column", None), "fixed when the table is created"),
            (("format-version", Some("9.0.0")), "set by Timelake itself"),
            (("min-snapshots", Some("many")), "non-negative integer"),
            (("head-format", Some("avro")), "unknown head format"),
            (("no-such-setting", Some("1")), "unknown setting"),
        ] {
            let err = t.set_properties(&[change]).await.unwrap_err().to_string();
            assert!(err.contains(msg), "{change:?}: {err}");
        }
    }
}