1use std::collections::HashMap;
7use std::fmt;
8use std::str::FromStr;
9use std::time::Duration;
10
11use chrono::Datelike;
12
13use crate::{FORMAT_VERSION, FormatError, FormatVersion, Result, property};
14
15const HOUR_MICROS: i64 = 3_600_000_000;
16const DAY_MICROS: i64 = 24 * HOUR_MICROS;
17const DAY_SECS: u64 = 86_400;
18
19#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
23pub enum Chunk {
24 Hour,
26 Day,
28 Month,
30}
31
32impl Chunk {
33 pub fn as_str(self) -> &'static str {
35 match self {
36 Self::Hour => "hour",
37 Self::Day => "day",
38 Self::Month => "month",
39 }
40 }
41
42 pub fn value(self, micros: i64) -> Result<i32> {
44 let out_of_range = || FormatError::new(format!("timestamp {micros}µs is out of range for {self} partitions"));
45 let t = chrono::DateTime::from_timestamp_micros(micros).ok_or_else(out_of_range)?;
46 let v = match self {
47 Self::Hour => micros.div_euclid(HOUR_MICROS),
48 Self::Day => micros.div_euclid(DAY_MICROS),
49 Self::Month => (i64::from(t.year()) - 1970) * 12 + i64::from(t.month0()),
50 };
51 i32::try_from(v).map_err(|_| out_of_range())
52 }
53
54 pub fn start_micros(self, value: i32) -> Result<i64> {
56 let start = match self {
57 Self::Hour => Some(i64::from(value) * HOUR_MICROS),
58 Self::Day => i64::from(value).checked_mul(DAY_MICROS),
59 Self::Month => {
60 let (year, month0) = (1970 + value.div_euclid(12), value.rem_euclid(12).unsigned_abs());
61 chrono::NaiveDate::from_ymd_opt(year, month0 + 1, 1)
62 .and_then(|d| d.and_hms_opt(0, 0, 0))
63 .map(|t| t.and_utc().timestamp_micros())
64 }
65 };
66 start.ok_or_else(|| self.out_of_range(value))
67 }
68
69 pub fn end_micros(self, value: i32) -> Result<i64> {
71 self.start_micros(value.checked_add(1).ok_or_else(|| self.out_of_range(value))?)
72 }
73
74 pub fn label(self, value: i32) -> String {
77 let Some(t) = self.start_micros(value).ok().and_then(chrono::DateTime::from_timestamp_micros) else {
78 return value.to_string();
79 };
80 let format = match self {
81 Self::Hour => "%Y-%m-%d-%H",
82 Self::Day => "%Y-%m-%d",
83 Self::Month => "%Y-%m",
84 };
85 t.format(format).to_string()
86 }
87
88 fn out_of_range(self, value: i32) -> FormatError {
89 FormatError::new(format!("{self} partition {value} is out of range"))
90 }
91}
92
93impl FromStr for Chunk {
94 type Err = FormatError;
95
96 fn from_str(s: &str) -> Result<Self> {
97 match s.trim().to_ascii_lowercase().as_str() {
98 "hour" | "1h" | "1 hour" => Ok(Self::Hour),
99 "day" | "1d" | "1 day" | "24h" => Ok(Self::Day),
100 "month" | "1 month" => Ok(Self::Month),
101 other => Err(FormatError::new(format!("chunk interval must be hour, day or month, got {other:?}"))),
102 }
103 }
104}
105
106impl fmt::Display for Chunk {
107 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
108 f.write_str(self.as_str())
109 }
110}
111
112#[derive(Clone, Debug, Default)]
115pub struct TableOptions {
116 pub time_column: Option<String>,
118 pub series_key: Vec<String>,
120 pub chunk: Option<Chunk>,
122 pub dedup: Option<bool>,
124 pub retention: Option<Duration>,
126 pub ingest_time: Option<bool>,
128 pub properties: HashMap<String, String>,
130}
131
132#[derive(Clone, Debug)]
134pub struct Contract {
135 pub time_column: Option<String>,
137 pub series_key: Vec<String>,
139 pub chunk: Option<Chunk>,
141 pub dedup: bool,
143 pub ingest_time_column: Option<String>,
145 pub retention: Option<Duration>,
147 pub target_file_size: u64,
149 pub row_group_rows: usize,
151 pub auto_maintain: bool,
153 pub snapshot_retention: Duration,
155 pub min_snapshots: usize,
157 pub max_snapshots: usize,
159 pub compact_min_files: usize,
161 pub compact_cold_after: Option<Duration>,
164 pub read_dedup: bool,
166 pub read_dedup_strategy: String,
168 pub filter_pushdown: String,
170 pub in_list_pruning: bool,
172 pub newest_first: bool,
174 pub time_weighted_mean: String,
176 pub head_format: String,
178 pub head_inline_bytes: u64,
180 pub format_version: FormatVersion,
182 pub keep_snapshots_until: Option<i64>,
184 pub rollups: Vec<String>,
186 pub rollup_routing: bool,
188 pub rollup_refresh_interval: Option<Duration>,
191 pub version_check_interval: Duration,
194 pub data_page_rows: Option<usize>,
196 pub compression_codec: String,
198 pub compression_level: Option<i32>,
200}
201
202impl Contract {
203 const DEFAULT_TARGET_FILE_SIZE: u64 = 256 << 20;
205 const DEFAULT_ROW_GROUP_ROWS: usize = 1 << 20;
206 const DEFAULT_SNAPSHOT_RETENTION: Duration = Duration::from_secs(3600);
207 const DEFAULT_MIN_SNAPSHOTS: usize = 5;
208 const DEFAULT_MAX_SNAPSHOTS: usize = 100;
209 const DEFAULT_COMPACT_MIN_FILES: usize = 8;
210 const DEFAULT_COMPACT_COLD_AFTER: Duration = Duration::from_secs(3600);
211 const DEFAULT_HEAD_FORMAT: &str = "arrow";
212 const DEFAULT_HEAD_INLINE_BYTES: u64 = 1 << 20;
213 const DEFAULT_ROLLUP_REFRESH_INTERVAL: Duration = Duration::from_secs(60);
214 const DEFAULT_VERSION_CHECK_INTERVAL: Duration = Duration::from_secs(1);
215
216 pub fn from_properties(properties: &HashMap<String, String>) -> Result<Self> {
218 let p = Properties(properties);
219 Ok(Self {
220 time_column: p.get(property::TIME_COLUMN).map(str::to_string),
221 series_key: p.list(property::SERIES_KEY),
222 chunk: p.get(property::CHUNK).map(str::parse).transpose()?,
223 dedup: p.flag(property::DEDUP, false)?,
224 ingest_time_column: p.get(property::INGEST_TIME_COLUMN).map(str::to_string),
225 retention: p.duration(property::RETENTION)?,
226 target_file_size: p.number(property::TARGET_FILE_SIZE, Self::DEFAULT_TARGET_FILE_SIZE)?,
227 row_group_rows: p.number(property::ROW_GROUP_ROWS, Self::DEFAULT_ROW_GROUP_ROWS)?,
228 auto_maintain: p.flag(property::AUTO_MAINTAIN, true)?,
229 snapshot_retention: p.duration(property::SNAPSHOT_RETENTION)?.unwrap_or(Self::DEFAULT_SNAPSHOT_RETENTION),
230 min_snapshots: p.number(property::MIN_SNAPSHOTS, Self::DEFAULT_MIN_SNAPSHOTS)?,
231 max_snapshots: p.number(property::MAX_SNAPSHOTS, Self::DEFAULT_MAX_SNAPSHOTS)?,
232 compact_min_files: p.number(property::COMPACT_MIN_FILES, Self::DEFAULT_COMPACT_MIN_FILES)?,
233 compact_cold_after: match p.get(property::COMPACT_COLD_AFTER) {
234 Some(s) if s.eq_ignore_ascii_case("off") => None,
235 Some(s) => Some(parse_duration(s)?),
236 None => Some(Self::DEFAULT_COMPACT_COLD_AFTER),
237 },
238 read_dedup: p.flag(property::READ_DEDUP, p.flag(property::DEDUP, false)?)?,
239 read_dedup_strategy: p.get(property::READ_DEDUP_STRATEGY).unwrap_or("merge").to_ascii_lowercase(),
240 filter_pushdown: p.get(property::FILTER_PUSHDOWN).unwrap_or("auto").to_ascii_lowercase(),
241 in_list_pruning: p.flag(property::IN_LIST_PRUNING, true)?,
242 newest_first: p.flag(property::NEWEST_FIRST, true)?,
243 time_weighted_mean: p.get(property::TIME_WEIGHTED_MEAN).unwrap_or("auto").to_ascii_lowercase(),
244 head_format: p.get(property::HEAD_FORMAT).unwrap_or(Self::DEFAULT_HEAD_FORMAT).to_ascii_lowercase(),
245 head_inline_bytes: p.number(property::HEAD_INLINE_BYTES, Self::DEFAULT_HEAD_INLINE_BYTES)?,
246 format_version: p.get(property::FORMAT_VERSION).map_or(Ok(FormatVersion::UNVERSIONED), str::parse)?,
247 keep_snapshots_until: p
248 .get(property::KEEP_SNAPSHOTS_UNTIL)
249 .map(|_| p.number(property::KEEP_SNAPSHOTS_UNTIL, 0))
250 .transpose()?,
251 rollups: p.list(property::ROLLUPS),
252 rollup_routing: p.flag(property::ROLLUP_ROUTING, true)?,
253 rollup_refresh_interval: match p.get(property::ROLLUP_REFRESH_INTERVAL) {
254 Some(s) if s.eq_ignore_ascii_case("off") => None,
255 Some(s) => Some(parse_duration(s)?),
256 None => Some(Self::DEFAULT_ROLLUP_REFRESH_INTERVAL),
257 },
258 version_check_interval: p
259 .duration(property::VERSION_CHECK_INTERVAL)?
260 .unwrap_or(Self::DEFAULT_VERSION_CHECK_INTERVAL),
261 data_page_rows: p
262 .get(property::DATA_PAGE_ROWS)
263 .map(|_| p.number(property::DATA_PAGE_ROWS, 0))
264 .transpose()?,
265 compression_codec: p.get(property::COMPRESSION_CODEC).unwrap_or("zstd").to_ascii_lowercase(),
267 compression_level: p.get(property::COMPRESSION_LEVEL).and_then(|v| v.parse().ok()),
268 })
269 }
270
271 pub fn check_readable(&self) -> Result<()> {
274 if self.format_version.readable_by(FORMAT_VERSION) {
275 return Ok(());
276 }
277 Err(FormatError::new(format!(
278 "table format {} is newer than this Timelake reads ({FORMAT_VERSION}): upgrade Timelake",
279 self.format_version
280 )))
281 }
282
283 pub fn check_writable(&self) -> Result<()> {
286 if self.format_version.writable_by(FORMAT_VERSION) {
287 return Ok(());
288 }
289 Err(FormatError::new(format!(
290 "table format {} is newer than this Timelake writes ({FORMAT_VERSION}): upgrade Timelake",
291 self.format_version
292 )))
293 }
294
295 pub fn is_time_series(&self) -> bool {
297 self.time_column.is_some()
298 }
299
300 pub fn sort_columns(&self) -> Vec<String> {
302 let mut v = self.series_key.clone();
303 v.extend(self.time_column.clone());
304 v
305 }
306
307 pub fn write_order(&self) -> Vec<String> {
310 let mut v = self.sort_columns();
311 v.extend(self.ingest_time_column.clone());
312 v
313 }
314}
315
316struct Properties<'a>(&'a HashMap<String, String>);
318
319impl Properties<'_> {
320 fn get(&self, key: &str) -> Option<&str> {
321 self.0.get(key).map(|s| s.trim()).filter(|s| !s.is_empty())
322 }
323
324 fn number<T: FromStr>(&self, key: &str, default: T) -> Result<T> {
325 self.get(key).map_or(Ok(default), |s| {
326 s.parse().map_err(|_| FormatError::new(format!("{key} must be a non-negative integer, got {s:?}")))
327 })
328 }
329
330 fn flag(&self, key: &str, default: bool) -> Result<bool> {
331 self.get(key).map_or(Ok(default), |s| {
332 s.to_ascii_lowercase()
333 .parse()
334 .map_err(|_| FormatError::new(format!("{key} must be true or false, got {s:?}")))
335 })
336 }
337
338 fn duration(&self, key: &str) -> Result<Option<Duration>> {
339 self.get(key).map(parse_duration).transpose()
340 }
341
342 fn list(&self, key: &str) -> Vec<String> {
344 self.get(key)
345 .map(|s| s.split(',').map(str::trim).filter(|c| !c.is_empty()).map(str::to_string).collect())
346 .unwrap_or_default()
347 }
348}
349
350pub fn parse_duration(s: &str) -> Result<Duration> {
352 let s = s.trim();
353 let split = s.find(|c: char| !c.is_ascii_digit()).unwrap_or(s.len());
354 let (n, unit) = s.split_at(split);
355 let n: u64 = n.parse().map_err(|_| FormatError::new(format!("bad duration {s:?}")))?;
356 let unit_secs = match unit.trim().to_ascii_lowercase().as_str() {
357 "s" | "sec" | "secs" | "second" | "seconds" => 1,
358 "m" | "min" | "mins" | "minute" | "minutes" => 60,
359 "h" | "hr" | "hrs" | "hour" | "hours" => 3600,
360 "d" | "day" | "days" => DAY_SECS,
361 "w" | "week" | "weeks" => 7 * DAY_SECS,
362 "y" | "year" | "years" => 365 * DAY_SECS,
363 _ => return Err(FormatError::new(format!("bad duration unit in {s:?} (use s, m, h, d, w, y)"))),
364 };
365 let secs = n.checked_mul(unit_secs).ok_or_else(|| FormatError::new(format!("duration {s:?} is too long")))?;
366 Ok(Duration::from_secs(secs))
367}
368
369pub fn format_duration(d: Duration) -> String {
372 let secs = d.as_secs();
373 match [(DAY_SECS, 'd'), (3600, 'h'), (60, 'm')]
374 .into_iter()
375 .find(|(unit, _)| secs != 0 && secs.is_multiple_of(*unit))
376 {
377 Some((unit, suffix)) => format!("{}{suffix}", secs / unit),
378 None => format!("{secs}s"),
379 }
380}
381
382#[cfg(test)]
383mod tests {
384 use super::*;
385 use crate::property::*;
386
387 const T: i64 = 1_767_225_600_000_000; #[test]
390 fn chunk_values_round_trip() {
391 for c in [Chunk::Hour, Chunk::Day, Chunk::Month] {
392 let v = c.value(T + 1).unwrap();
393 assert_eq!(c.start_micros(v).unwrap(), T, "{c:?}");
394 assert!(c.end_micros(v).unwrap() > T);
395 }
396 assert_eq!(Chunk::Day.label(Chunk::Day.value(T).unwrap()), "2026-01-01");
397 assert_eq!(Chunk::Hour.label(Chunk::Hour.value(T).unwrap()), "2026-01-01-00");
398 assert_eq!(Chunk::Month.label(Chunk::Month.value(T).unwrap()), "2026-01");
399 assert_eq!(Chunk::Month.value(T).unwrap(), (2026 - 1970) * 12);
400 assert_eq!(Chunk::Day.value(-1).unwrap(), -1);
402 assert_eq!(Chunk::Month.value(-1).unwrap(), -1);
403 assert_eq!(Chunk::Month.start_micros(-1).unwrap(), -31 * DAY_MICROS);
404 }
405
406 #[test]
407 fn out_of_range_times_are_errors() {
408 for c in [Chunk::Hour, Chunk::Day, Chunk::Month] {
409 assert!(c.value(i64::MAX).is_err(), "{c:?}");
410 assert!(c.value(i64::MIN).is_err(), "{c:?}");
411 }
412 assert!(Chunk::Day.start_micros(i32::MAX).is_err());
413 assert!(Chunk::Month.start_micros(i32::MAX).is_err());
414 assert!(Chunk::Hour.end_micros(i32::MAX).is_err());
415 assert_eq!(Chunk::Month.label(i32::MAX), i32::MAX.to_string());
416 }
417
418 #[test]
419 fn chunk_parses_and_displays() {
420 for (s, c) in [("hour", Chunk::Hour), (" 1 Day ", Chunk::Day), ("24h", Chunk::Day), ("month", Chunk::Month)] {
421 assert_eq!(s.parse::<Chunk>().unwrap(), c, "{s}");
422 }
423 assert!("week".parse::<Chunk>().is_err());
424 for c in [Chunk::Hour, Chunk::Day, Chunk::Month] {
425 assert_eq!(c.to_string().parse::<Chunk>().unwrap(), c);
426 }
427 }
428
429 #[test]
430 fn durations() {
431 assert_eq!(parse_duration("90d").unwrap().as_secs(), 90 * 86_400);
432 assert_eq!(parse_duration("1h").unwrap().as_secs(), 3600);
433 assert_eq!(parse_duration(" 0s ").unwrap(), Duration::ZERO);
434 assert!(parse_duration("1 parsec").is_err());
435 assert!(parse_duration("d").is_err());
436 assert!(parse_duration("-1d").is_err());
437 }
438
439 #[test]
440 fn durations_accept_singular_and_plural_units() {
441 assert_eq!(parse_duration("1 day").unwrap().as_secs(), 86_400);
442 assert_eq!(parse_duration("1 hour").unwrap().as_secs(), 3600);
443 assert_eq!(parse_duration("1 week").unwrap().as_secs(), 7 * 86_400);
444 assert_eq!(parse_duration("2 Weeks").unwrap().as_secs(), 14 * 86_400);
445 assert_eq!(parse_duration("1 minute").unwrap().as_secs(), 60);
446 assert_eq!(parse_duration("1 year").unwrap().as_secs(), 365 * 86_400);
447 }
448
449 #[test]
450 fn duration_overflow_is_an_error() {
451 assert!(parse_duration("99999999999999999y").is_err());
452 assert!(parse_duration("99999999999999999999s").is_err());
453 }
454
455 #[test]
456 fn durations_format_in_the_largest_exact_unit() {
457 for (secs, s) in [(90 * 86_400, "90d"), (36 * 3600, "36h"), (90, "90s"), (120, "2m"), (0, "0s")] {
458 assert_eq!(format_duration(Duration::from_secs(secs)), s);
459 assert_eq!(parse_duration(s).unwrap().as_secs(), secs);
460 }
461 }
462
463 fn props(kv: &[(&str, &str)]) -> HashMap<String, String> {
464 kv.iter().map(|(k, v)| ((*k).to_string(), (*v).to_string())).collect()
465 }
466
467 #[test]
468 fn contract_defaults() {
469 let c = Contract::from_properties(&HashMap::new()).unwrap();
470 assert!(!c.is_time_series() && !c.dedup && c.auto_maintain);
471 assert_eq!(c.retention, None);
472 assert_eq!(c.snapshot_retention, Duration::from_secs(3600));
473 assert_eq!((c.min_snapshots, c.max_snapshots, c.compact_min_files), (5, 100, 8));
474 assert_eq!(
475 (c.head_format.as_str(), c.head_inline_bytes, c.format_version),
476 ("arrow", 1 << 20, FormatVersion::UNVERSIONED)
477 );
478 assert_eq!((c.compact_cold_after, c.read_dedup), (Some(Duration::from_hours(1)), false));
479 let dedup = Contract::from_properties(&props(&[(property::DEDUP, "true")])).unwrap();
480 assert!(dedup.read_dedup, "read dedup follows dedup");
481 let off =
482 props(&[(property::DEDUP, "true"), (property::READ_DEDUP, "false"), (property::COMPACT_COLD_AFTER, "off")]);
483 let off = Contract::from_properties(&off).unwrap();
484 assert_eq!((off.compact_cold_after, off.read_dedup), (None, false));
485 let cold = Contract::from_properties(&props(&[(property::COMPACT_COLD_AFTER, "6h")])).unwrap();
486 assert_eq!(cold.compact_cold_after, Some(Duration::from_hours(6)));
487 assert_eq!(c.sort_columns(), [] as [String; 0]);
488 }
489
490 #[test]
491 fn contract_parses_properties() {
492 let c = Contract::from_properties(&props(&[
493 (TIME_COLUMN, "time"),
494 (SERIES_KEY, "site, device,"),
495 (CHUNK, "hour"),
496 (DEDUP, "True"),
497 (AUTO_MAINTAIN, "FALSE"),
498 (RETENTION, "1 day"),
499 (MIN_SNAPSHOTS, "3"),
500 (TARGET_FILE_SIZE, " 1024 "),
501 (HEAD_FORMAT, "JSON"),
502 (HEAD_INLINE_BYTES, "0"),
503 ]))
504 .unwrap();
505 assert_eq!((c.head_format.as_str(), c.head_inline_bytes), ("json", 0));
506 assert_eq!(c.sort_columns(), ["site", "device", "time"]);
507 assert_eq!(c.chunk, Some(Chunk::Hour));
508 assert!(c.dedup && !c.auto_maintain);
509 assert_eq!(c.retention, Some(Duration::from_hours(24)));
510 assert_eq!((c.min_snapshots, c.target_file_size), (3, 1024));
511 }
512
513 #[test]
514 fn contract_rejects_bad_values() {
515 for kv in [(DEDUP, "yes"), (AUTO_MAINTAIN, "0"), (CHUNK, "week"), (RETENTION, "forever"), (MAX_SNAPSHOTS, "-1")]
516 {
517 assert!(Contract::from_properties(&props(&[kv])).is_err(), "{kv:?}");
518 }
519 }
520}