mz_storage/
sink.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Moving data to external systems
11
12use mz_repr::Timestamp;
13use timely::progress::Antichain;
14
15mod iceberg;
16mod kafka;
17
18/// Formats the error reported when a sink's input has been compacted past the
19/// upper the sink last committed, so the sink can never resume. `remedy` tells
20/// the user how to recover, which differs by sink type.
21fn overcompacted_error(
22    name: &str,
23    as_of: &Antichain<Timestamp>,
24    resume_upper: &Antichain<Timestamp>,
25    remedy: &str,
26) -> String {
27    let fmt = |f: &Antichain<Timestamp>| match f.as_option() {
28        Some(t) => t.to_string(),
29        None => "[]".to_string(),
30    };
31    let gap = match (as_of.as_option(), resume_upper.as_option()) {
32        (Some(as_of), Some(upper)) => {
33            let secs = u64::from(*as_of).saturating_sub(u64::from(*upper)) / 1000;
34            let (h, m, s) = (secs / 3600, secs % 3600 / 60, secs % 60);
35            format!(" (gap {h}h{m:02}m{s:02}s)")
36        }
37        _ => String::new(),
38    };
39    format!(
40        "sink {name} cannot resume: input compacted to {}, past last committed upper {}{gap}; \
41         data in that range is unrecoverable, {remedy}",
42        fmt(as_of),
43        fmt(resume_upper),
44    )
45}
mz_storage/sink.rs

mz_storage/
sink.rs
