1use std::fmt;
19
20use crate::{Chunk, FormatError, FormatVersion, Result, parse_duration};
21
22pub const PREFIX: &str = "timelake.";
24
25#[derive(Clone, Copy, Debug, PartialEq, Eq)]
27pub enum Level {
28 Create,
30 Table,
32 Session,
34 Call,
36}
37
38impl Level {
39 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#[derive(Clone, Copy, Debug, PartialEq, Eq)]
58pub enum DefaultValue {
59 Value(&'static str),
61 Unset(&'static str),
63 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#[derive(Debug)]
77pub struct Setting {
78 pub name: &'static str,
80 pub kind: &'static str,
82 pub default: DefaultValue,
84 pub levels: &'static [Level],
86 pub since: &'static str,
88 pub description: &'static str,
90 check: fn(&str) -> Result<()>,
91}
92
93impl Setting {
94 pub fn property(&self) -> String {
96 format!("{PREFIX}{}", self.name)
97 }
98
99 pub fn sql_name(&self) -> String {
101 format!("{PREFIX}{}", self.name.replace('-', "_"))
102 }
103
104 pub fn allows(&self, level: Level) -> bool {
106 self.levels.contains(&level)
107 }
108
109 pub fn check(&self, value: &str) -> Result<()> {
111 (self.check)(value.trim()).map_err(|e| FormatError::new(format!("{}: {e}", self.property())))
112 }
113}
114
115pub 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
123pub 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
146pub fn unknown(name: &str) -> FormatError {
148 FormatError::new(format!("unknown setting {name:?} (`timelake settings` lists them)"))
149}
150
151pub 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
166pub 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: "data-page-rows",
378 kind: "count",
379 default: DefaultValue::Derived("20,000; 8,192 when written from WebAssembly"),
380 levels: &[Level::Table],
381 since: "0.6.0",
382 description: "Rows per Parquet data page in the files written. Smaller pages make compaction from a Worker \
383 use less memory (it holds a page per column of each file it merges: 8,192 rows peaked at 77 MiB \
384 merging 24 TSBS files, 20,000 at 123 MiB) and page pruning finer, at a few percent more bytes \
385 (TSBS: +3.6%) and slower scans below about 8,000 rows.",
386 check: count,
387 },
388 Setting {
389 name: "version-check-interval",
390 kind: "duration",
391 default: DefaultValue::Value("1s"),
392 levels: &[Level::Table, Level::Session],
393 since: "0.6.0",
394 description: "The Timelake reader uses a table version it loaded or checked this long before checking for a \
395 newer one, so queries in that time send no request for it and can miss commits by other \
396 processes. `0s` checks on every query. This process's own commits are seen at once, and \
397 writers always check.",
398 check: duration,
399 },
400 Setting {
401 name: "rollups",
402 kind: "table names",
403 default: DefaultValue::Unset("none"),
404 levels: &[],
405 since: "0.5.0",
406 description: "On a raw table: its rollups, comma-separated. `CREATE` and `DROP MATERIALIZED ROLLUP` keep it.",
407 check: columns,
408 },
409 Setting {
410 name: "rollup-source",
411 kind: "table name",
412 default: DefaultValue::Unset("none"),
413 levels: &[],
414 since: "0.5.0",
415 description: "On a rollup: the raw table it summarises.",
416 check: name,
417 },
418 Setting {
419 name: "rollup-bucket",
420 kind: "duration",
421 default: DefaultValue::Unset("none"),
422 levels: &[],
423 since: "0.5.0",
424 description: "On a rollup: the width of its time buckets.",
425 check: duration,
426 },
427 Setting {
428 name: "rollup-aggregates",
429 kind: "JSON",
430 default: DefaultValue::Unset("none"),
431 levels: &[],
432 since: "0.5.0",
433 description: "On a rollup: its aggregates, each an output column, a function, and the column it reads.",
434 check: name,
435 },
436 Setting {
437 name: "format-version",
438 kind: "semantic version",
439 default: DefaultValue::Derived("the current format for new tables; 0.1.0 if absent"),
440 levels: &[],
441 since: "0.3.0",
442 description: "The table format. Timelake refuses tables of a newer format it can't handle; `timelake upgrade` \
443 brings a table to the current format.",
444 check: format_version,
445 },
446];
447
448fn boolean(s: &str) -> Result<()> {
449 match s.to_ascii_lowercase().as_str() {
450 "true" | "false" => Ok(()),
451 _ => Err(FormatError::new(format!("must be true or false, got {s:?}"))),
452 }
453}
454
455fn count(s: &str) -> Result<()> {
456 s.parse::<u64>().map(drop).map_err(|_| FormatError::new(format!("must be a non-negative integer, got {s:?}")))
457}
458
459fn duration(s: &str) -> Result<()> {
460 parse_duration(s).map(drop)
461}
462
463fn duration_or_off(s: &str) -> Result<()> {
464 if s.eq_ignore_ascii_case("off") { Ok(()) } else { duration(s) }
465}
466
467fn chunk(s: &str) -> Result<()> {
468 s.parse::<Chunk>().map(drop)
469}
470
471fn read_dedup_strategy(s: &str) -> Result<()> {
472 match s.to_ascii_lowercase().as_str() {
473 "merge" | "keys-first" | "keys_first" => Ok(()),
474 _ => Err(FormatError::new(format!("must be merge or keys-first, got {s:?}"))),
475 }
476}
477
478fn format_version(s: &str) -> Result<()> {
479 s.parse::<FormatVersion>().map(drop)
480}
481
482fn name(s: &str) -> Result<()> {
483 if s.is_empty() { Err(FormatError::new("must not be empty")) } else { Ok(()) }
484}
485
486fn column(s: &str) -> Result<()> {
487 name(s)
488}
489
490fn columns(s: &str) -> Result<()> {
491 if s.split(',').any(|c| !c.trim().is_empty()) { Ok(()) } else { Err(FormatError::new("must name a column")) }
492}
493
494#[cfg(test)]
495mod tests {
496 use std::collections::HashMap;
497
498 use super::*;
499 use crate::{Contract, property};
500
501 #[test]
502 fn names_are_unique_kebab_case_and_found_in_both_spellings() {
503 for (i, s) in SETTINGS.iter().enumerate() {
504 assert!(s.name.chars().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-'), "{}", s.name);
505 assert!(SETTINGS[..i].iter().all(|t| t.name != s.name), "{} twice", s.name);
506 for spelling in [s.property(), s.sql_name(), s.name.to_string(), s.sql_name().to_ascii_uppercase()] {
507 assert_eq!(find(&spelling).map(|f| f.name), Some(s.name), "{spelling}");
508 }
509 }
510 assert!(find("timelake.no-such-setting").is_none());
511 }
512
513 #[test]
515 fn every_property_is_registered() {
516 for key in [
517 property::TIME_COLUMN,
518 property::SERIES_KEY,
519 property::CHUNK,
520 property::DEDUP,
521 property::RETENTION,
522 property::TARGET_FILE_SIZE,
523 property::ROW_GROUP_ROWS,
524 property::AUTO_MAINTAIN,
525 property::SNAPSHOT_RETENTION,
526 property::MIN_SNAPSHOTS,
527 property::MAX_SNAPSHOTS,
528 property::COMPACT_MIN_FILES,
529 property::COMPACT_COLD_AFTER,
530 property::INGEST_TIME_COLUMN,
531 property::READ_DEDUP,
532 property::READ_DEDUP_STRATEGY,
533 property::HEAD_FORMAT,
534 property::HEAD_INLINE_BYTES,
535 property::KEEP_SNAPSHOTS_UNTIL,
536 property::FORMAT_VERSION,
537 property::ROLLUPS,
538 property::ROLLUP_ROUTING,
539 property::ROLLUP_SOURCE,
540 property::ROLLUP_BUCKET,
541 property::ROLLUP_AGGREGATES,
542 property::ROLLUP_REFRESH_INTERVAL,
543 property::VERSION_CHECK_INTERVAL,
544 property::DATA_PAGE_ROWS,
545 ] {
546 assert_eq!(find(key).map(Setting::property).as_deref(), Some(key));
547 }
548 assert_eq!(SETTINGS.len(), 28, "a new setting needs a property constant and an entry above");
549 }
550
551 #[test]
553 fn defaults_match_the_contract() {
554 let unset = format!("{:?}", Contract::from_properties(&HashMap::new()).unwrap());
555 for s in SETTINGS {
556 if let DefaultValue::Value(v) = s.default {
557 s.check(v).unwrap();
558 let set = HashMap::from([(s.property(), v.to_string())]);
559 assert_eq!(format!("{:?}", Contract::from_properties(&set).unwrap()), unset, "{}", s.name);
560 }
561 }
562 }
563
564 #[test]
565 fn values_are_checked() {
566 let check = |name: &str, v: &str| find(name).unwrap().check(v);
567 assert!(check("read-dedup", "TRUE").is_ok() && check("read-dedup", "yes").is_err());
568 assert!(check("compact-cold-after", "off").is_ok() && check("compact-cold-after", "soon").is_err());
569 assert!(check("read-dedup-strategy", "keys-first").is_ok() && check("read-dedup-strategy", "hash").is_err());
570 assert!(check("chunk-interval", "1h").is_ok() && check("chunk-interval", "week").is_err());
571 assert!(check("min-snapshots", "-1").is_err() && check("series-key", " , ").is_err());
572 let err = check("read-dedup", "yes").unwrap_err().to_string();
573 assert!(err.starts_with("timelake.read-dedup: must be true or false"), "{err}");
574 }
575
576 #[test]
577 fn properties_are_checked_for_their_level() {
578 let props = |kv: &[(&str, &str)]| -> HashMap<String, String> {
579 kv.iter().map(|(k, v)| ((*k).to_string(), (*v).to_string())).collect()
580 };
581 let ok = check_properties(
582 &props(&[("timelake.read_dedup", " false "), ("write.parquet.compression-codec", "zstd")]),
583 Level::Table,
584 )
585 .unwrap();
586 assert_eq!(
587 ok.into_iter().collect::<HashMap<_, _>>(),
588 props(&[("timelake.read-dedup", "false"), ("write.parquet.compression-codec", "zstd")])
589 );
590 for (kv, msg) in [
591 (("timelake.chunk-interval", "hour"), "fixed when the table is created"),
592 (("timelake.format-version", "0.3.0"), "set by Timelake itself"),
593 (("timelake.no-such", "1"), "unknown setting"),
594 (("timelake.retention", "forever"), "bad duration"),
595 ] {
596 let err = check_properties(&props(&[kv]), Level::Table).unwrap_err().to_string();
597 assert!(err.contains(msg), "{kv:?}: {err}");
598 }
599 let err = check_properties(&props(&[("timelake.min-snapshots", "1")]), Level::Session).unwrap_err();
600 assert!(err.to_string().contains("only at: table"), "{err}");
601 }
602}