timelake_format/
settings.rs

1//! The registry of Timelake's settings (SPEC §21.9): every switch is declared once here, with its
2//! type, default, the levels it may be set at, and the version that added it. Table properties,
3//! SQL `SET`, the CLI's `timelake settings` and the docs all come from [`SETTINGS`].
4//!
5//! A setting has one name in kebab-case (`read-dedup`). As a table property it is
6//! `timelake.read-dedup`; in SQL, where names are identifiers, `timelake.read_dedup`. Both
7//! spellings find the same setting.
8//!
9//! ```
10//! use timelake_format::settings::{self, Level};
11//!
12//! let s = settings::find("timelake.read_dedup").unwrap();
13//! assert_eq!(s.property(), "timelake.read-dedup");
14//! assert!(s.allows(Level::Session));
15//! assert!(s.check("maybe").is_err());
16//! ```
17
18use std::fmt;
19
20use crate::{Chunk, FormatError, FormatVersion, Result, parse_duration};
21
22/// The prefix of Timelake's table properties and session settings.
23pub const PREFIX: &str = "timelake.";
24
25/// Where a setting can be set. The more specific level wins: a call over a session over a table.
26#[derive(Clone, Copy, Debug, PartialEq, Eq)]
27pub enum Level {
28    /// Only when the table is created, through the table options (`time_column`, `chunk`, …).
29    Create,
30    /// As a table property, at creation or later (`timelake alter`, `Table::set_properties`).
31    Table,
32    /// For one SQL session: `SET timelake.<name> = …`, or `--set` on the CLI.
33    Session,
34    /// For one call, through its options (`AppendOptions`, `MaintainOptions`).
35    Call,
36}
37
38impl Level {
39    /// The level's name, as the docs and `timelake settings` show it.
40    pub fn as_str(self) -> &'static str {
41        match self {
42            Self::Create => "create",
43            Self::Table => "table",
44            Self::Session => "session",
45            Self::Call => "call",
46        }
47    }
48}
49
50impl fmt::Display for Level {
51    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
52        f.write_str(self.as_str())
53    }
54}
55
56/// A setting's value when nothing sets it.
57#[derive(Clone, Copy, Debug, PartialEq, Eq)]
58pub enum DefaultValue {
59    /// This value.
60    Value(&'static str),
61    /// Unset, with what that means (`forever`, `none`).
62    Unset(&'static str),
63    /// Follows from other settings, as described.
64    Derived(&'static str),
65}
66
67impl fmt::Display for DefaultValue {
68    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
69        match self {
70            Self::Value(s) | Self::Unset(s) | Self::Derived(s) => f.write_str(s),
71        }
72    }
73}
74
75/// One setting.
76#[derive(Debug)]
77pub struct Setting {
78    /// The name in kebab-case, without the `timelake.` prefix.
79    pub name: &'static str,
80    /// The kind of value, as the docs show it (`boolean`, `duration`, …).
81    pub kind: &'static str,
82    /// The value when nothing sets it.
83    pub default: DefaultValue,
84    /// Where it can be set. Empty for properties Timelake manages itself.
85    pub levels: &'static [Level],
86    /// The Timelake version that added it.
87    pub since: &'static str,
88    /// What it does, for the docs.
89    pub description: &'static str,
90    check: fn(&str) -> Result<()>,
91}
92
93impl Setting {
94    /// The table property: `timelake.read-dedup`.
95    pub fn property(&self) -> String {
96        format!("{PREFIX}{}", self.name)
97    }
98
99    /// The name in SQL: `timelake.read_dedup`.
100    pub fn sql_name(&self) -> String {
101        format!("{PREFIX}{}", self.name.replace('-', "_"))
102    }
103
104    /// Whether the setting can be set at `level`.
105    pub fn allows(&self, level: Level) -> bool {
106        self.levels.contains(&level)
107    }
108
109    /// Errors when `value` isn't a valid value of this setting.
110    pub fn check(&self, value: &str) -> Result<()> {
111        (self.check)(value.trim()).map_err(|e| FormatError::new(format!("{}: {e}", self.property())))
112    }
113}
114
115/// Finds a setting by name: `read-dedup`, `read_dedup`, `timelake.read-dedup` or
116/// `timelake.read_dedup`, in any case.
117pub fn find(name: &str) -> Option<&'static Setting> {
118    let name = name.trim().to_ascii_lowercase().replace('_', "-");
119    let name = name.strip_prefix(PREFIX).unwrap_or(&name);
120    SETTINGS.iter().find(|s| s.name == name)
121}
122
123/// Checks Timelake's properties among `properties` for `level` ([`Level::Table`] when creating
124/// or altering a table) and returns them with every Timelake key in its canonical spelling. Keys
125/// outside the `timelake.` prefix (Iceberg's own) pass through.
126pub fn check_properties<'a>(
127    properties: impl IntoIterator<Item = (&'a String, &'a String)>,
128    level: Level,
129) -> Result<Vec<(String, String)>> {
130    let mut out = vec![];
131    for (key, value) in properties {
132        if !key.trim().to_ascii_lowercase().starts_with(PREFIX) {
133            out.push((key.clone(), value.clone()));
134            continue;
135        }
136        let setting = find(key).ok_or_else(|| unknown(key))?;
137        if !setting.allows(level) {
138            return Err(not_settable(setting, level));
139        }
140        setting.check(value)?;
141        out.push((setting.property(), value.trim().to_string()));
142    }
143    Ok(out)
144}
145
146/// The error for a name that isn't a setting.
147pub fn unknown(name: &str) -> FormatError {
148    FormatError::new(format!("unknown setting {name:?} (`timelake settings` lists them)"))
149}
150
151/// The error for a setting set where it can't be: says where it can be.
152pub fn not_settable(setting: &Setting, level: Level) -> FormatError {
153    let name = setting.property();
154    let msg = if setting.levels.is_empty() {
155        format!("{name} is set by Timelake itself, not by hand")
156    } else if setting.levels == [Level::Create] {
157        let options = "time_column, series_key, chunk, dedup, ingest_time";
158        format!("{name} is fixed when the table is created, by its options ({options})")
159    } else {
160        let levels: Vec<&str> = setting.levels.iter().map(|l| l.as_str()).collect();
161        format!("{name} can't be set at the {level} level, only at: {}", levels.join(", "))
162    };
163    FormatError::new(msg)
164}
165
166/// Every setting, in the order the docs list them.
167pub static SETTINGS: &[Setting] = &[
168    Setting {
169        name: "time-column",
170        kind: "column",
171        default: DefaultValue::Unset("none"),
172        levels: &[Level::Create],
173        since: "0.1.0",
174        description: "The time column. Its presence makes the table a time-series table.",
175        check: column,
176    },
177    Setting {
178        name: "series-key",
179        kind: "columns",
180        default: DefaultValue::Unset("none"),
181        levels: &[Level::Create],
182        since: "0.1.0",
183        description: "Comma-separated columns that identify a series.",
184        check: columns,
185    },
186    Setting {
187        name: "chunk-interval",
188        kind: "hour, day, or month",
189        default: DefaultValue::Derived("day for time-series tables"),
190        levels: &[Level::Create],
191        since: "0.1.0",
192        description: "Partition size: how much time one partition covers.",
193        check: chunk,
194    },
195    Setting {
196        name: "dedup",
197        kind: "boolean",
198        default: DefaultValue::Derived("true with a series key, else false"),
199        levels: &[Level::Create],
200        since: "0.1.0",
201        description: "Rows with the same (series, time) count once; the latest write wins. Applied within an append, \
202                      by the Timelake reader, and by compaction. Fixed at creation: files written with it off may \
203                      hold copies, which the reader would then miss.",
204        check: boolean,
205    },
206    Setting {
207        name: "ingest-time-column",
208        kind: "column",
209        default: DefaultValue::Derived("_ingest_ts for time-series tables"),
210        levels: &[Level::Create],
211        since: "0.5.0",
212        description: "The column holding each row's insertion time (`ingest_time=False` or `--no-ingest-time` leaves \
213                      it out). Each append fills it with the time the append started, unless the input brings its \
214                      own values. Among rows with the same (series, time), the latest insertion wins, then the \
215                      later commit.",
216        check: column,
217    },
218    Setting {
219        name: "retention",
220        kind: "duration",
221        default: DefaultValue::Unset("forever"),
222        levels: &[Level::Create, Level::Table],
223        since: "0.1.0",
224        description: "Maintenance drops partitions that end before now minus this duration, e.g. 90d.",
225        check: duration,
226    },
227    Setting {
228        name: "auto-maintain",
229        kind: "boolean",
230        default: DefaultValue::Value("true"),
231        levels: &[Level::Table, Level::Call],
232        since: "0.1.0",
233        description: "Run size-tiered compaction, retention, and snapshot expiry inline after appends. One append \
234                      can override it (`maintain=` in Python, `--no-maintain` on the CLI).",
235        check: boolean,
236    },
237    Setting {
238        name: "compact-min-files",
239        kind: "count",
240        default: DefaultValue::Value("8"),
241        levels: &[Level::Table],
242        since: "0.1.0",
243        description: "Inline compaction merges a size tier of a partition once it has this many files.",
244        check: count,
245    },
246    Setting {
247        name: "compact-cold-after",
248        kind: "duration or off",
249        default: DefaultValue::Value("1h"),
250        levels: &[Level::Table, Level::Call],
251        since: "0.5.0",
252        description: "Inline maintenance rewrites a partition whole (sorted, duplicates removed) once the table holds \
253                      data newer than the partition's end by this much and the partition has two small files, \
254                      whether the append touched it or not. `off` leaves it to the size tiers.",
255        check: duration_or_off,
256    },
257    Setting {
258        name: "read-dedup",
259        kind: "boolean",
260        default: DefaultValue::Derived("as dedup"),
261        levels: &[Level::Table, Level::Session],
262        since: "0.5.0",
263        description: "The Timelake reader returns only the latest write of each (series, time), before compaction \
264                      has removed the older copies. Has no effect on tables with dedup off.",
265        check: boolean,
266    },
267    Setting {
268        name: "read-dedup-strategy",
269        kind: "merge or keys-first",
270        default: DefaultValue::Value("merge"),
271        levels: &[Level::Table, Level::Session],
272        since: "0.5.0",
273        description: "How the reader removes copies: `merge` merges the files that may share keys in key order; \
274                      `keys-first` lists the older copies from the key columns, then reads the files like any scan \
275                      and leaves them out (slower on the industrial IoT benchmark).",
276        check: read_dedup_strategy,
277    },
278    Setting {
279        name: "target-file-size-bytes",
280        kind: "bytes",
281        default: DefaultValue::Value("268435456"),
282        levels: &[Level::Table],
283        since: "0.1.0",
284        description: "Size at which compaction output and writes roll to a new file (256 MiB).",
285        check: count,
286    },
287    Setting {
288        name: "row-group-rows",
289        kind: "count",
290        default: DefaultValue::Value("1048576"),
291        levels: &[Level::Table],
292        since: "0.1.0",
293        description: "Parquet row group size. Page indexes keep single-series reads selective inside large row \
294                      groups.",
295        check: count,
296    },
297    Setting {
298        name: "snapshot-retention",
299        kind: "duration",
300        default: DefaultValue::Value("1h"),
301        levels: &[Level::Table],
302        since: "0.1.0",
303        description: "Expire snapshots older than this, keeping at least `min-snapshots` and at most \
304                      `max-snapshots`.",
305        check: duration,
306    },
307    Setting {
308        name: "min-snapshots",
309        kind: "count",
310        default: DefaultValue::Value("5"),
311        levels: &[Level::Table],
312        since: "0.1.0",
313        description: "Snapshots always kept, whatever their age.",
314        check: count,
315    },
316    Setting {
317        name: "max-snapshots",
318        kind: "count",
319        default: DefaultValue::Value("100"),
320        levels: &[Level::Table],
321        since: "0.1.0",
322        description: "Snapshots kept at most: older ones expire even if younger than `snapshot-retention`.",
323        check: count,
324    },
325    Setting {
326        name: "head-format",
327        kind: "codec name or none",
328        default: DefaultValue::Value("arrow"),
329        levels: &[Level::Table],
330        since: "0.3.0",
331        description: "Format of the head object (`metadata/timelake-head`): `arrow`, `json`, a registered codec, or \
332                      `none` for no head.",
333        check: name,
334    },
335    Setting {
336        name: "head-inline-bytes",
337        kind: "bytes",
338        default: DefaultValue::Value("1048576"),
339        levels: &[Level::Table],
340        since: "0.3.0",
341        description: "Room in the head for the last-value index, then the Parquet footers of the newest files \
342                      (1 MiB). What doesn't fit is read separately.",
343        check: count,
344    },
345    Setting {
346        name: "keep-snapshots-until",
347        kind: "milliseconds since the epoch",
348        default: DefaultValue::Unset("none"),
349        levels: &[Level::Table],
350        since: "0.3.0",
351        description: "No snapshot expires before this time. `timelake upgrade` sets it so the upgrade can be rolled \
352                      back.",
353        check: count,
354    },
355    Setting {
356        name: "rollup-routing",
357        kind: "boolean",
358        default: DefaultValue::Value("true"),
359        levels: &[Level::Table, Level::Session],
360        since: "0.5.0",
361        description: "The Timelake reader answers a `GROUP BY time_bucket(…)` query on this table from a rollup \
362                      that matches it. Buckets changed since the rollup's last refresh, and partial buckets at the \
363                      edges of the time range, still come from this table, so the answer is the same either way.",
364        check: boolean,
365    },
366    Setting {
367        name: "rollup-refresh-interval",
368        kind: "duration or off",
369        default: DefaultValue::Value("1m"),
370        levels: &[Level::Table],
371        since: "0.5.0",
372        description: "On a rollup: inline maintenance after an append to its raw table refreshes it once its last \
373                      refresh is this old. `off` leaves it to `timelake maintain` and `REFRESH MATERIALIZED ROLLUP`.",
374        check: duration_or_off,
375    },
376    Setting {
377        name: "rollups",
378        kind: "table names",
379        default: DefaultValue::Unset("none"),
380        levels: &[],
381        since: "0.5.0",
382        description: "On a raw table: its rollups, comma-separated. `CREATE` and `DROP MATERIALIZED ROLLUP` keep it.",
383        check: columns,
384    },
385    Setting {
386        name: "rollup-source",
387        kind: "table name",
388        default: DefaultValue::Unset("none"),
389        levels: &[],
390        since: "0.5.0",
391        description: "On a rollup: the raw table it summarises.",
392        check: name,
393    },
394    Setting {
395        name: "rollup-bucket",
396        kind: "duration",
397        default: DefaultValue::Unset("none"),
398        levels: &[],
399        since: "0.5.0",
400        description: "On a rollup: the width of its time buckets.",
401        check: duration,
402    },
403    Setting {
404        name: "rollup-aggregates",
405        kind: "JSON",
406        default: DefaultValue::Unset("none"),
407        levels: &[],
408        since: "0.5.0",
409        description: "On a rollup: its aggregates, each an output column, a function, and the column it reads.",
410        check: name,
411    },
412    Setting {
413        name: "format-version",
414        kind: "semantic version",
415        default: DefaultValue::Derived("the current format for new tables; 0.1.0 if absent"),
416        levels: &[],
417        since: "0.3.0",
418        description: "The table format. Timelake refuses tables of a newer format it can't handle; `timelake upgrade` \
419                      brings a table to the current format.",
420        check: format_version,
421    },
422];
423
424fn boolean(s: &str) -> Result<()> {
425    match s.to_ascii_lowercase().as_str() {
426        "true" | "false" => Ok(()),
427        _ => Err(FormatError::new(format!("must be true or false, got {s:?}"))),
428    }
429}
430
431fn count(s: &str) -> Result<()> {
432    s.parse::<u64>().map(drop).map_err(|_| FormatError::new(format!("must be a non-negative integer, got {s:?}")))
433}
434
435fn duration(s: &str) -> Result<()> {
436    parse_duration(s).map(drop)
437}
438
439fn duration_or_off(s: &str) -> Result<()> {
440    if s.eq_ignore_ascii_case("off") { Ok(()) } else { duration(s) }
441}
442
443fn chunk(s: &str) -> Result<()> {
444    s.parse::<Chunk>().map(drop)
445}
446
447fn read_dedup_strategy(s: &str) -> Result<()> {
448    match s.to_ascii_lowercase().as_str() {
449        "merge" | "keys-first" | "keys_first" => Ok(()),
450        _ => Err(FormatError::new(format!("must be merge or keys-first, got {s:?}"))),
451    }
452}
453
454fn format_version(s: &str) -> Result<()> {
455    s.parse::<FormatVersion>().map(drop)
456}
457
458fn name(s: &str) -> Result<()> {
459    if s.is_empty() { Err(FormatError::new("must not be empty")) } else { Ok(()) }
460}
461
462fn column(s: &str) -> Result<()> {
463    name(s)
464}
465
466fn columns(s: &str) -> Result<()> {
467    if s.split(',').any(|c| !c.trim().is_empty()) { Ok(()) } else { Err(FormatError::new("must name a column")) }
468}
469
470#[cfg(test)]
471mod tests {
472    use std::collections::HashMap;
473
474    use super::*;
475    use crate::{Contract, property};
476
477    #[test]
478    fn names_are_unique_kebab_case_and_found_in_both_spellings() {
479        for (i, s) in SETTINGS.iter().enumerate() {
480            assert!(s.name.chars().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-'), "{}", s.name);
481            assert!(SETTINGS[..i].iter().all(|t| t.name != s.name), "{} twice", s.name);
482            for spelling in [s.property(), s.sql_name(), s.name.to_string(), s.sql_name().to_ascii_uppercase()] {
483                assert_eq!(find(&spelling).map(|f| f.name), Some(s.name), "{spelling}");
484            }
485        }
486        assert!(find("timelake.no-such-setting").is_none());
487    }
488
489    /// Every property the contract reads is a setting.
490    #[test]
491    fn every_property_is_registered() {
492        for key in [
493            property::TIME_COLUMN,
494            property::SERIES_KEY,
495            property::CHUNK,
496            property::DEDUP,
497            property::RETENTION,
498            property::TARGET_FILE_SIZE,
499            property::ROW_GROUP_ROWS,
500            property::AUTO_MAINTAIN,
501            property::SNAPSHOT_RETENTION,
502            property::MIN_SNAPSHOTS,
503            property::MAX_SNAPSHOTS,
504            property::COMPACT_MIN_FILES,
505            property::COMPACT_COLD_AFTER,
506            property::INGEST_TIME_COLUMN,
507            property::READ_DEDUP,
508            property::READ_DEDUP_STRATEGY,
509            property::HEAD_FORMAT,
510            property::HEAD_INLINE_BYTES,
511            property::KEEP_SNAPSHOTS_UNTIL,
512            property::FORMAT_VERSION,
513            property::ROLLUPS,
514            property::ROLLUP_ROUTING,
515            property::ROLLUP_SOURCE,
516            property::ROLLUP_BUCKET,
517            property::ROLLUP_AGGREGATES,
518            property::ROLLUP_REFRESH_INTERVAL,
519        ] {
520            assert_eq!(find(key).map(Setting::property).as_deref(), Some(key));
521        }
522        assert_eq!(SETTINGS.len(), 26, "a new setting needs a property constant and an entry above");
523    }
524
525    /// A stated default is what the contract uses when the property is unset.
526    #[test]
527    fn defaults_match_the_contract() {
528        let unset = format!("{:?}", Contract::from_properties(&HashMap::new()).unwrap());
529        for s in SETTINGS {
530            if let DefaultValue::Value(v) = s.default {
531                s.check(v).unwrap();
532                let set = HashMap::from([(s.property(), v.to_string())]);
533                assert_eq!(format!("{:?}", Contract::from_properties(&set).unwrap()), unset, "{}", s.name);
534            }
535        }
536    }
537
538    #[test]
539    fn values_are_checked() {
540        let check = |name: &str, v: &str| find(name).unwrap().check(v);
541        assert!(check("read-dedup", "TRUE").is_ok() && check("read-dedup", "yes").is_err());
542        assert!(check("compact-cold-after", "off").is_ok() && check("compact-cold-after", "soon").is_err());
543        assert!(check("read-dedup-strategy", "keys-first").is_ok() && check("read-dedup-strategy", "hash").is_err());
544        assert!(check("chunk-interval", "1h").is_ok() && check("chunk-interval", "week").is_err());
545        assert!(check("min-snapshots", "-1").is_err() && check("series-key", " , ").is_err());
546        let err = check("read-dedup", "yes").unwrap_err().to_string();
547        assert!(err.starts_with("timelake.read-dedup: must be true or false"), "{err}");
548    }
549
550    #[test]
551    fn properties_are_checked_for_their_level() {
552        let props = |kv: &[(&str, &str)]| -> HashMap<String, String> {
553            kv.iter().map(|(k, v)| ((*k).to_string(), (*v).to_string())).collect()
554        };
555        let ok = check_properties(
556            &props(&[("timelake.read_dedup", " false "), ("write.parquet.compression-codec", "zstd")]),
557            Level::Table,
558        )
559        .unwrap();
560        assert_eq!(
561            ok.into_iter().collect::<HashMap<_, _>>(),
562            props(&[("timelake.read-dedup", "false"), ("write.parquet.compression-codec", "zstd")])
563        );
564        for (kv, msg) in [
565            (("timelake.chunk-interval", "hour"), "fixed when the table is created"),
566            (("timelake.format-version", "0.3.0"), "set by Timelake itself"),
567            (("timelake.no-such", "1"), "unknown setting"),
568            (("timelake.retention", "forever"), "bad duration"),
569        ] {
570            let err = check_properties(&props(&[kv]), Level::Table).unwrap_err().to_string();
571            assert!(err.contains(msg), "{kv:?}: {err}");
572        }
573        let err = check_properties(&props(&[("timelake.min-snapshots", "1")]), Level::Session).unwrap_err();
574        assert!(err.to_string().contains("only at: table"), "{err}");
575    }
576}
timelake_format/settings.rs

timelake_format/
settings.rs
