use time::Duration;
use crate::arrow::datatypes::{DataType, Field};
use crate::error::{Error, Result};
pub use metering::ids::EicType;
pub const CHECK_VALUES_KEY: &str = "meterstore.check_values";
#[must_use]
pub fn coded_column(name: &str, allowed: &[&str], nullable: bool) -> Field {
debug_assert!(
allowed.iter().all(|c| !c.contains(',')),
"coded_column values must not contain a comma"
);
Field::new(name, DataType::Utf8, nullable).with_metadata(std::collections::HashMap::from([(
CHECK_VALUES_KEY.to_string(),
allowed.join(","),
)]))
}
pub const VALUE_CHECK_KEY: &str = "meterstore.value_check";
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum ValueCheck {
Eic(Option<EicType>),
Malo,
Melo,
Bdew,
}
impl ValueCheck {
pub const ALL: [Self; 1 + EicType::ALL.len() + 3] = {
let types = EicType::ALL;
let mut out = [Self::Eic(None); 1 + EicType::ALL.len() + 3];
let mut i = 0;
while i < types.len() {
out[1 + i] = Self::Eic(Some(types[i]));
i += 1;
}
out[1 + types.len()] = Self::Malo;
out[2 + types.len()] = Self::Melo;
out[3 + types.len()] = Self::Bdew;
out
};
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::Eic(None) => "EIC",
Self::Eic(Some(EicType::Party)) => "EIC:X",
Self::Eic(Some(EicType::Area)) => "EIC:Y",
Self::Eic(Some(EicType::MeasurementPoint)) => "EIC:Z",
Self::Eic(Some(EicType::ResourceObject)) => "EIC:W",
Self::Eic(Some(EicType::TieLine)) => "EIC:T",
Self::Eic(Some(EicType::Location)) => "EIC:V",
Self::Eic(Some(EicType::Substation)) => "EIC:A",
Self::Malo => "MALO",
Self::Melo => "MELO",
Self::Bdew => "BDEW",
}
}
#[must_use]
pub fn from_code(code: &str) -> Option<Self> {
Self::ALL.into_iter().find(|c| c.as_str() == code)
}
#[must_use]
pub fn known_codes() -> Vec<&'static str> {
Self::ALL.into_iter().map(Self::as_str).collect()
}
#[must_use]
pub const fn shape_pattern(self) -> &'static str {
match self {
Self::Eic(None) => "^[0-9A-Z-]{2}[A-Z][0-9A-Z-]{12}[0-9A-Z]$",
Self::Eic(Some(EicType::Party)) => "^[0-9A-Z-]{2}X[0-9A-Z-]{12}[0-9A-Z]$",
Self::Eic(Some(EicType::Area)) => "^[0-9A-Z-]{2}Y[0-9A-Z-]{12}[0-9A-Z]$",
Self::Eic(Some(EicType::MeasurementPoint)) => "^[0-9A-Z-]{2}Z[0-9A-Z-]{12}[0-9A-Z]$",
Self::Eic(Some(EicType::ResourceObject)) => "^[0-9A-Z-]{2}W[0-9A-Z-]{12}[0-9A-Z]$",
Self::Eic(Some(EicType::TieLine)) => "^[0-9A-Z-]{2}T[0-9A-Z-]{12}[0-9A-Z]$",
Self::Eic(Some(EicType::Location)) => "^[0-9A-Z-]{2}V[0-9A-Z-]{12}[0-9A-Z]$",
Self::Eic(Some(EicType::Substation)) => "^[0-9A-Z-]{2}A[0-9A-Z-]{12}[0-9A-Z]$",
Self::Malo => "^[1-9][0-9]{10}$",
Self::Melo => "^[A-Z]{2}[0-9]{6}[0-9A-Z]{25}$",
Self::Bdew => "^[0-9]{13}$",
}
}
pub fn canonicalise(self, column: &str, text: &str) -> Result<String> {
let refuse = |reason: String| {
Error::encode(
column,
format!("{text:?} is not {}: {reason} {}", self.noun(), self.why()),
)
};
let parsed = |r: std::result::Result<String, metering::ParseError>| {
r.map_err(|e| refuse(format!("{e}.")))
};
match self {
Self::Eic(want) => {
let eic: metering::ids::Eic = text
.parse()
.map_err(|e: metering::ParseError| refuse(format!("{e}.")))?;
if let Some(want) = want {
let found = eic.object_type();
if found != Some(want) {
return Err(refuse(format!(
"this column declares object type {} ({}), and the code carries \
{} at position 3.",
want.as_str(),
want.name(),
found.map_or_else(
|| format!(
"{:?}, which is not a type letter this build lists",
eic.as_str().chars().nth(2).unwrap_or('?')
),
|f| format!("{} ({})", f.as_str(), f.name()),
),
)));
}
}
Ok(eic.into())
}
Self::Malo => parsed(text.parse::<metering::ids::MaloId>().map(String::from)),
Self::Melo => parsed(text.parse::<metering::ids::MeloId>().map(String::from)),
Self::Bdew => parsed(text.parse::<metering::ids::BdewCode>().map(String::from)),
}
}
#[must_use]
pub const fn noun(self) -> &'static str {
match self {
Self::Eic(None) => "an EIC",
Self::Eic(Some(EicType::Party)) => "an X (party) EIC",
Self::Eic(Some(EicType::Area)) => "a Y (area) EIC",
Self::Eic(Some(EicType::MeasurementPoint)) => "a Z (measurement point) EIC",
Self::Eic(Some(EicType::ResourceObject)) => "a W (resource object) EIC",
Self::Eic(Some(EicType::TieLine)) => "a T (tie-line) EIC",
Self::Eic(Some(EicType::Location)) => "a V (location) EIC",
Self::Eic(Some(EicType::Substation)) => "an A (substation) EIC",
Self::Malo => "a MaLo-ID",
Self::Melo => "a MeLo-ID",
Self::Bdew => "a Marktpartner-ID",
}
}
#[must_use]
pub const fn why(self) -> &'static str {
match self {
Self::Eic(_) => {
"The check character is part of the code, so a transposition is detectable \
here — and only here, while the delivery that carried it is still in hand."
}
Self::Malo => {
"The check digit is part of the identifier, so a transposition is detectable \
here — and only here, while the delivery that carried it is still in hand."
}
Self::Melo => {
"There is no check digit, so the structure is the whole of the rule — and a \
value that fails it would have been stored as a Messlokation nothing else \
names."
}
Self::Bdew => {
"The thirteenth digit is deliberately not checked, because BDEW's \
Bildungsvorschrift exempts GS1-issued GLNs — so thirteen digits is the \
whole of the rule that can be enforced, and a value failing it is not a \
Marktpartner-ID at all."
}
}
}
}
impl std::fmt::Display for ValueCheck {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
impl std::str::FromStr for ValueCheck {
type Err = Error;
fn from_str(s: &str) -> Result<Self> {
Self::from_code(s).ok_or_else(|| {
Error::config(format!(
"unknown value check {s:?}; this build knows {:?}",
Self::known_codes()
))
})
}
}
#[must_use]
pub fn checked_column(name: &str, check: ValueCheck, nullable: bool) -> Field {
Field::new(name, DataType::Utf8, nullable).with_metadata(std::collections::HashMap::from([(
VALUE_CHECK_KEY.to_string(),
check.as_str().to_string(),
)]))
}
#[must_use]
pub fn declared_value_check(field: &Field) -> Option<&str> {
field.metadata().get(VALUE_CHECK_KEY).map(String::as_str)
}
#[derive(
Debug, Clone, Copy, PartialEq, Eq, Hash, Default, serde::Serialize, serde::Deserialize,
)]
#[serde(rename_all = "lowercase")]
pub enum TimeModel {
#[default]
Interval,
Point,
}
impl TimeModel {
pub const fn has_interval_end(self) -> bool {
matches!(self, Self::Interval)
}
pub const fn as_str(self) -> &'static str {
match self {
Self::Interval => "INTERVAL",
Self::Point => "POINT",
}
}
}
impl std::fmt::Display for TimeModel {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.as_str())
}
}
pub mod defaults {
use time::Duration;
pub const ARCHIVAL_STEP: Duration = Duration::DAY;
pub const SETTLEMENT_LAG: Duration = Duration::weeks(1);
pub const PARTITION_HEADROOM: Duration = Duration::weeks(2);
pub const TARGET_FILE_SIZE: usize = 512 * 1024 * 1024;
pub const SCAN_CHUNK_ROWS: usize = 50_000;
pub const SNAPSHOT_RETENTION: Duration = Duration::days(3_653);
pub const MIN_SNAPSHOTS_TO_KEEP: usize = 20;
}
#[derive(Debug, Clone)]
pub struct TableConfig {
name: String,
time_model: TimeModel,
archival_step: Duration,
settlement_lag: Duration,
partition_headroom: Duration,
scan_chunk_rows: usize,
snapshot_retention: Duration,
min_snapshots_to_keep: usize,
melo_in_merge_key: Option<bool>,
identity_columns: Vec<Field>,
attribute_columns: Vec<Field>,
subject_column: Option<String>,
}
impl TableConfig {
pub fn new(name: impl Into<String>) -> Self {
Self {
name: name.into(),
time_model: TimeModel::Interval,
archival_step: defaults::ARCHIVAL_STEP,
settlement_lag: defaults::SETTLEMENT_LAG,
partition_headroom: defaults::PARTITION_HEADROOM,
scan_chunk_rows: defaults::SCAN_CHUNK_ROWS,
snapshot_retention: defaults::SNAPSHOT_RETENTION,
min_snapshots_to_keep: defaults::MIN_SNAPSHOTS_TO_KEEP,
melo_in_merge_key: None,
identity_columns: Vec::new(),
attribute_columns: Vec::new(),
subject_column: None,
}
}
pub fn identify_by_melo(mut self, yes: bool) -> Self {
self.melo_in_merge_key = Some(yes);
self
}
pub fn time_model(mut self, model: TimeModel) -> Self {
self.time_model = model;
self
}
pub fn archival_step(mut self, step: Duration) -> Self {
self.archival_step = step;
self
}
pub fn settlement_lag(mut self, lag: Duration) -> Self {
self.settlement_lag = lag;
self
}
pub fn partition_headroom(mut self, headroom: Duration) -> Self {
self.partition_headroom = headroom;
self
}
pub fn scan_chunk_rows(mut self, rows: usize) -> Self {
self.scan_chunk_rows = rows;
self
}
pub fn snapshot_retention(mut self, retention: Duration) -> Self {
self.snapshot_retention = retention;
self
}
pub fn min_snapshots_to_keep(mut self, count: usize) -> Self {
self.min_snapshots_to_keep = count;
self
}
pub fn identity_column(mut self, field: Field) -> Self {
self.identity_columns.push(field);
self
}
pub fn attribute_column(mut self, field: Field) -> Self {
self.attribute_columns.push(field);
self
}
pub fn subject_column(mut self, name: impl Into<String>) -> Self {
let name = name.into();
if !self.attribute_columns.iter().any(|f| *f.name() == name) {
self.attribute_columns
.push(Field::new(&name, DataType::Utf8, true));
}
self.subject_column = Some(name);
self
}
pub fn build(self) -> Result<ValidatedTableConfig> {
if !is_plain_identifier(&self.name) {
return Err(Error::config(format!(
"table name {:?} must be a plain identifier — a letter or underscore \
followed by letters, digits or underscores. The name is written into \
DDL, partition relation names and SQL as an identifier, which cannot \
be parameterised",
self.name
)));
}
const MAX_TABLE_NAME: usize = 63 - PARTITION_SUFFIX_LEN - LONGEST_CONSTRAINT_SUFFIX;
if self.name.len() > MAX_TABLE_NAME {
return Err(Error::config(format!(
"table name {:?} is {} characters; at most {MAX_TABLE_NAME} fit. \
PostgreSQL truncates an identifier at 63 bytes and this crate derives \
longer ones from it — a partition relation adds {PARTITION_SUFFIX_LEN} \
characters and its integrity constraints another \
{LONGEST_CONSTRAINT_SUFFIX}",
self.name,
self.name.len(),
)));
}
if self.archival_step <= Duration::ZERO {
return Err(Error::config("archival_step must be positive"));
}
if self.archival_step < Duration::MINUTE {
return Err(Error::config(format!(
"archival_step is {}, and a partition relation is named to the minute \
(<table>_YYYY_MM_DD_HHMM) — two consecutive windows would name one \
relation. One minute is the floor",
fmt_duration(self.archival_step),
)));
}
if self.settlement_lag < Duration::ZERO {
return Err(Error::config("settlement_lag must not be negative"));
}
if self.settlement_lag < self.archival_step {
return Err(Error::config(format!(
"settlement_lag ({}) must be at least one archival_step ({}), \
or corrections can arrive for an already-archived window",
fmt_duration(self.settlement_lag),
fmt_duration(self.archival_step),
)));
}
if self.partition_headroom < self.archival_step {
return Err(Error::config(
"partition_headroom must cover at least one archival_step, \
or inserts will fail before a new partition exists",
));
}
if self.scan_chunk_rows == 0 {
return Err(Error::config(
"scan_chunk_rows must be positive: a zero-row chunk would page forever",
));
}
if self.snapshot_retention <= Duration::ZERO {
return Err(Error::config("snapshot_retention must be positive"));
}
if self.min_snapshots_to_keep == 0 {
return Err(Error::config(
"min_snapshots_to_keep must be at least 1: expiring every snapshot \
would leave the table unreadable",
));
}
for f in self.identity_columns.iter().chain(&self.attribute_columns) {
if !matches!(f.data_type(), DataType::Utf8) {
return Err(Error::config(format!(
"extra column {:?} is {:?}; only Utf8 is supported — every \
attribute deployments have wanted (tenant, Bilanzkreis, grid area) \
is a string, and supporting more needs a bind arm per type",
f.name(),
f.data_type()
)));
}
}
for f in &self.identity_columns {
if f.is_nullable() {
return Err(Error::config(format!(
"identity column {:?} must be non-nullable: a null cannot identify a reading",
f.name()
)));
}
}
if let Some(subject) = &self.subject_column
&& self.identity_columns.iter().any(|f| f.name() == subject)
{
return Err(Error::config(format!(
"{subject:?} is declared both as an identity column and as the subject column. A pseudonymous reference must not join the merge key: a correction carrying a re-derived reference would get a different key and silently fail to supersede the value it corrects. Erasure does not need it in the key — it destroys the mapping, which leaves every row unattributable wherever the column sits"
)));
}
let mut seen = std::collections::HashSet::new();
for f in self.identity_columns.iter().chain(&self.attribute_columns) {
if !is_plain_identifier(f.name()) {
return Err(Error::config(format!(
"column name {:?} must be a plain identifier — a letter or underscore \
followed by letters, digits or underscores. Declared names are written \
into DDL and SQL as identifiers, which cannot be parameterised",
f.name()
)));
}
if !seen.insert(f.name().clone()) {
return Err(Error::config(format!("duplicate column {:?}", f.name())));
}
if crate::encode::schema::storage_schema(&[])
.field_with_name(f.name())
.is_ok()
{
return Err(Error::config(format!(
"extra column {:?} collides with a core column",
f.name()
)));
}
}
Ok(ValidatedTableConfig(self))
}
}
#[derive(Debug, Clone)]
pub struct ValidatedTableConfig(TableConfig);
impl ValidatedTableConfig {
pub fn name(&self) -> &str {
&self.0.name
}
pub fn time_model(&self) -> TimeModel {
self.0.time_model
}
pub fn archival_step(&self) -> Duration {
self.0.archival_step
}
pub fn settlement_lag(&self) -> Duration {
self.0.settlement_lag
}
pub fn partition_headroom(&self) -> Duration {
self.0.partition_headroom
}
pub fn scan_chunk_rows(&self) -> usize {
self.0.scan_chunk_rows
}
pub fn scan_spec(&self) -> crate::tiering::store::ScanSpec {
crate::tiering::store::ScanSpec::new(
self.merge_key(),
self.extra_columns()
.iter()
.map(|f| f.name().clone())
.collect(),
)
.with_chunk_rows(self.scan_chunk_rows())
}
pub fn snapshot_retention(&self) -> Duration {
self.0.snapshot_retention
}
pub fn min_snapshots_to_keep(&self) -> usize {
self.0.min_snapshots_to_keep
}
pub fn identity_columns(&self) -> &[Field] {
&self.0.identity_columns
}
pub fn identity_column_names(&self) -> Vec<String> {
self.0
.identity_columns
.iter()
.map(|f| f.name().clone())
.collect()
}
pub fn attribute_columns(&self) -> &[Field] {
&self.0.attribute_columns
}
pub fn subject_column(&self) -> Option<&str> {
self.0.subject_column.as_deref()
}
pub fn extra_columns(&self) -> Vec<Field> {
self.0
.identity_columns
.iter()
.chain(&self.0.attribute_columns)
.cloned()
.collect()
}
pub fn melo_in_merge_key(&self) -> bool {
self.0
.melo_in_merge_key
.unwrap_or(matches!(self.0.time_model, TimeModel::Point))
}
pub fn merge_key(&self) -> Vec<String> {
let mut key = vec![crate::encode::schema::col::MALO_ID.to_string()];
if self.melo_in_merge_key() {
key.push(crate::encode::schema::col::MELO_ID.to_string());
}
key.extend(
crate::encode::schema::MERGE_KEY
.iter()
.filter(|c| **c != crate::encode::schema::col::MALO_ID)
.map(|s| (*s).to_string()),
);
key.extend(self.0.identity_columns.iter().map(|f| f.name().clone()));
key
}
pub fn discriminator_columns(&self) -> Vec<String> {
let core = crate::encode::schema::MERGE_KEY;
self.merge_key()
.into_iter()
.filter(|c| !core.contains(&c.as_str()))
.collect()
}
pub fn expected_hot_partitions(&self) -> i64 {
let span = self.0.settlement_lag + self.0.partition_headroom;
(span.whole_seconds() / self.0.archival_step.whole_seconds()).max(1)
}
}
const PARTITION_SUFFIX_LEN: usize = "_2026_07_20_0000".len();
const LONGEST_CONSTRAINT_SUFFIX: usize = "_one_operator".len();
fn is_plain_identifier(name: &str) -> bool {
let mut chars = name.chars();
chars
.next()
.is_some_and(|c| c.is_ascii_alphabetic() || c == '_')
&& chars.all(|c| c.is_ascii_alphanumeric() || c == '_')
}
fn fmt_duration(d: Duration) -> String {
let secs = d.whole_seconds();
match secs {
s if s % 86_400 == 0 => format!("{}d", s / 86_400),
s if s % 3_600 == 0 => format!("{}h", s / 3_600),
s if s % 60 == 0 => format!("{}m", s / 60),
s => format!("{s}s"),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::arrow::datatypes::DataType;
fn base() -> TableConfig {
TableConfig::new("readings")
}
#[test]
fn coded_column_carries_its_vocabulary_in_metadata() {
let f = coded_column("source", &["MSCONS", "DIRECT_PUSH"], true);
assert_eq!(f.name(), "source");
assert_eq!(f.data_type(), &DataType::Utf8);
assert!(f.is_nullable());
assert_eq!(
f.metadata().get(CHECK_VALUES_KEY).map(String::as_str),
Some("MSCONS,DIRECT_PUSH")
);
}
#[test]
fn a_checked_column_declares_the_identifier_scheme_its_values_must_parse_as() {
for scheme in ValueCheck::ALL {
let f = checked_column("c", scheme, true);
assert_eq!(f.data_type(), &DataType::Utf8);
assert!(f.is_nullable());
assert_eq!(declared_value_check(&f), Some(scheme.as_str()));
assert!(f.metadata().get(CHECK_VALUES_KEY).is_none());
}
assert_eq!(declared_value_check(&coded_column("s", &["A"], true)), None);
assert_eq!(
declared_value_check(&Field::new("plain", DataType::Utf8, true)),
None
);
}
#[test]
fn a_checked_column_may_be_an_identity_column() {
let c = base()
.identity_column(checked_column(
"bilanzkreis",
ValueCheck::Eic(Some(EicType::Party)),
false,
))
.build()
.unwrap();
assert!(c.merge_key().contains(&"bilanzkreis".to_string()));
assert_eq!(
declared_value_check(&c.identity_columns()[0]),
Some("EIC:X")
);
}
#[test]
fn a_value_checks_code_round_trips_and_is_the_only_spelling() {
for scheme in ValueCheck::ALL {
assert_eq!(ValueCheck::from_code(scheme.as_str()), Some(scheme));
assert_eq!(scheme.to_string(), scheme.as_str());
assert_eq!(
scheme.as_str().parse::<ValueCheck>().expect("known"),
scheme
);
}
assert_eq!(
ValueCheck::known_codes(),
vec![
"EIC", "EIC:X", "EIC:Y", "EIC:Z", "EIC:W", "EIC:T", "EIC:V", "EIC:A", "MALO",
"MELO", "BDEW",
]
);
for ty in EicType::ALL {
assert_eq!(
ValueCheck::from_code(&format!("EIC:{}", ty.as_str())),
Some(ValueCheck::Eic(Some(ty))),
"{ty}"
);
}
for unknown in ["IBAN", "eic", "EIC:", "EIC:Q", "EIC:XX", "EIC:x", "MALO:X"] {
assert_eq!(ValueCheck::from_code(unknown), None, "{unknown}");
}
let err = "EIC:Q"
.parse::<ValueCheck>()
.expect_err("unknown")
.to_string();
assert!(err.contains("EIC:Q"), "{err}");
assert!(err.contains("EIC:X"), "{err}");
}
#[test]
fn canonicalising_is_the_domain_types_own_spelling() {
let ok = |c: ValueCheck, text: &str| c.canonicalise("declared", text).unwrap();
assert_eq!(
ok(ValueCheck::Eic(None), " 11xbk0000000001a "),
"11XBK0000000001A"
);
assert_eq!(
ok(ValueCheck::Melo, " de00056266802ao6g56m11sn51g21m24s "),
"DE00056266802AO6G56M11SN51G21M24S"
);
assert_eq!(ok(ValueCheck::Malo, " 41373559241 "), "41373559241");
assert_eq!(ok(ValueCheck::Bdew, " 9900987654321 "), "9900987654321");
let bad = |c: ValueCheck, text: &str| {
c.canonicalise("declared", text)
.expect_err(text)
.to_string()
};
assert!(bad(ValueCheck::Eic(None), "11XBK0000000001B").contains("declared"));
assert!(bad(ValueCheck::Malo, "41373559214").contains("check digit"));
assert_eq!(ok(ValueCheck::Bdew, "9900987654320"), "9900987654320");
}
#[test]
fn an_eic_of_the_wrong_object_type_is_refused_and_the_message_says_which() {
let bilanzkreis = ValueCheck::Eic(Some(EicType::Party));
assert_eq!(
bilanzkreis
.canonicalise("bilanzkreis", "11XBK0000000001A")
.unwrap(),
"11XBK0000000001A"
);
let err = bilanzkreis
.canonicalise("bilanzkreis", "11YN000000000016")
.expect_err("an area code is not a party code")
.to_string();
assert!(err.contains("bilanzkreis"), "{err}");
assert!(err.contains("X (Party)"), "{err}");
assert!(err.contains("Y (Area or Domain)"), "{err}");
let malformed = bilanzkreis
.canonicalise("bilanzkreis", "11XBK0000000001B")
.expect_err("the check character is wrong")
.to_string();
assert!(!malformed.contains("object type"), "{malformed}");
for code in ["11XBK0000000001A", "11YN000000000016"] {
assert!(
ValueCheck::Eic(None).canonicalise("c", code).is_ok(),
"{code}"
);
}
}
#[test]
fn naming_an_already_declared_column_as_the_subject_column_adopts_it() {
let c = base()
.attribute_column(coded_column("subject_ref", &["A", "B"], true))
.subject_column("subject_ref")
.build()
.unwrap();
assert_eq!(c.subject_column(), Some("subject_ref"));
let declared: Vec<_> = c
.attribute_columns()
.iter()
.filter(|f| f.name() == "subject_ref")
.collect();
assert_eq!(declared.len(), 1);
assert_eq!(
declared[0]
.metadata()
.get(CHECK_VALUES_KEY)
.map(String::as_str),
Some("A,B"),
);
}
#[test]
fn a_subject_column_may_not_be_an_identity_column() {
let err = base()
.identity_column(Field::new("subject_ref", DataType::Utf8, false))
.subject_column("subject_ref")
.build()
.unwrap_err()
.to_string();
assert!(err.contains("merge key"), "{err}");
}
#[test]
fn a_plain_attribute_column_declares_no_vocabulary() {
let f = Field::new("bilanzkreis", DataType::Utf8, true);
assert!(f.metadata().get(CHECK_VALUES_KEY).is_none());
}
#[test]
fn defaults_validate() {
let c = base().build().unwrap();
assert_eq!(c.name(), "readings");
assert_eq!(c.archival_step(), Duration::DAY);
}
#[test]
fn settlement_lag_must_cover_at_least_one_window() {
let err = base()
.archival_step(Duration::days(7))
.settlement_lag(Duration::DAY)
.build()
.unwrap_err();
assert!(err.to_string().contains("settlement_lag"));
}
#[test]
fn zero_settlement_lag_is_rejected_when_a_window_is_positive() {
assert!(base().settlement_lag(Duration::ZERO).build().is_err());
}
#[test]
fn headroom_must_cover_a_partition() {
let err = base()
.partition_headroom(Duration::hours(1))
.build()
.unwrap_err();
assert!(err.to_string().contains("partition_headroom"));
}
#[test]
fn non_positive_steps_are_rejected() {
assert!(base().archival_step(Duration::ZERO).build().is_err());
assert!(base().archival_step(-Duration::DAY).build().is_err());
}
#[test]
fn a_sub_minute_step_is_refused_because_partitions_are_named_to_the_minute() {
let err = base()
.archival_step(Duration::seconds(30))
.settlement_lag(Duration::minutes(5))
.partition_headroom(Duration::minutes(5))
.build()
.unwrap_err()
.to_string();
assert!(err.contains("minute"), "{err}");
assert!(
base()
.archival_step(Duration::MINUTE)
.settlement_lag(Duration::minutes(5))
.partition_headroom(Duration::minutes(5))
.build()
.is_ok()
);
}
#[test]
fn a_table_name_must_be_a_plain_identifier() {
assert!(TableConfig::new("").build().is_err());
assert!(
TableConfig::new("readings\"; DROP TABLE x --")
.build()
.is_err()
);
assert!(TableConfig::new("has space").build().is_err());
assert!(TableConfig::new("1_leading_digit").build().is_err());
assert!(TableConfig::new("Messwerte_Ä").build().is_err());
assert!(TableConfig::new("readings").build().is_ok());
assert!(TableConfig::new("readings_versions").build().is_ok());
assert!(TableConfig::new("_private2").build().is_ok());
}
#[test]
fn the_scan_spec_carries_everything_a_chunked_scan_needs() {
let c = base()
.identity_column(Field::new("tenant", DataType::Utf8, false))
.attribute_column(Field::new("bilanzkreis", DataType::Utf8, true))
.scan_chunk_rows(1234)
.build()
.unwrap();
let spec = c.scan_spec();
assert_eq!(spec.merge_key(), c.merge_key().as_slice());
assert_eq!(spec.extra(), ["tenant", "bilanzkreis"]);
assert_eq!(spec.chunk_rows(), Some(1234));
assert_eq!(
spec.cursor_columns(),
["malo_id", "from", "obis_code", "tenant", "version"]
);
}
#[test]
fn a_zero_row_chunk_is_rejected() {
assert!(base().scan_chunk_rows(0).build().is_err());
}
#[test]
fn a_column_name_that_is_not_an_identifier_is_refused() {
for bad in [
r#"tenant" ; DROP TABLE readings --"#,
"has space",
"1leading_digit",
"",
"dotted.name",
"kebab-case",
] {
let err = base()
.attribute_column(Field::new(bad, DataType::Utf8, true))
.build()
.unwrap_err();
assert!(
err.to_string().contains("plain identifier"),
"{bad:?} was accepted: {err}"
);
}
for good in ["tenant", "_private", "bilanzkreis_2", "NetzGebiet"] {
assert!(
base()
.attribute_column(Field::new(good, DataType::Utf8, true))
.build()
.is_ok(),
"{good:?} was refused"
);
}
}
#[test]
fn extra_columns_must_not_collide_with_core_columns() {
let err = base()
.attribute_column(Field::new("malo_id", DataType::Utf8, true))
.build()
.unwrap_err();
assert!(err.to_string().contains("collides"));
}
#[test]
fn extra_columns_must_be_unique_across_both_kinds() {
let err = base()
.attribute_column(Field::new("bilanzkreis", DataType::Utf8, true))
.attribute_column(Field::new("bilanzkreis", DataType::Utf8, true))
.build()
.unwrap_err();
assert!(err.to_string().contains("duplicate"));
}
#[test]
fn identity_columns_join_the_merge_key() {
let c = base()
.identity_column(Field::new("tenant", DataType::Utf8, false))
.build()
.unwrap();
assert_eq!(
c.merge_key(),
vec!["malo_id", "obis_code", "from", "tenant"]
);
}
#[test]
fn a_point_table_identifies_a_reading_by_its_messlokation_by_default() {
let point = base().time_model(TimeModel::Point).build().unwrap();
assert!(point.melo_in_merge_key());
assert_eq!(
point.merge_key(),
vec!["malo_id", "melo_id", "obis_code", "from"]
);
assert_eq!(point.discriminator_columns(), vec!["melo_id"]);
}
#[test]
fn an_interval_table_does_not() {
let lastgang = base().build().unwrap();
assert!(!lastgang.melo_in_merge_key());
assert_eq!(lastgang.merge_key(), vec!["malo_id", "obis_code", "from"]);
assert!(lastgang.discriminator_columns().is_empty());
}
#[test]
fn identify_by_melo_pins_the_choice_either_way() {
assert!(
!base()
.time_model(TimeModel::Point)
.identify_by_melo(false)
.build()
.unwrap()
.melo_in_merge_key()
);
assert!(
base()
.identify_by_melo(true)
.build()
.unwrap()
.melo_in_merge_key()
);
}
#[test]
fn the_messlokation_leads_the_key_and_identity_columns_follow() {
let c = base()
.identify_by_melo(true)
.identity_column(Field::new("tenant", DataType::Utf8, false))
.build()
.unwrap();
assert_eq!(
c.merge_key(),
vec!["malo_id", "melo_id", "obis_code", "from", "tenant"]
);
assert_eq!(c.discriminator_columns(), vec!["melo_id", "tenant"]);
assert_eq!(
c.scan_spec().cursor_columns(),
[
"malo_id",
"from",
"melo_id",
"obis_code",
"tenant",
"version"
]
);
}
#[test]
fn a_table_name_too_long_for_postgres_identifiers_is_refused() {
let longest = "a".repeat(34);
assert!(TableConfig::new(&longest).build().is_ok());
let err = TableConfig::new("a".repeat(35)).build().unwrap_err();
let msg = err.to_string();
assert!(msg.contains("63"), "{msg}");
assert!(msg.contains("34"), "message must name the limit: {msg}");
}
#[test]
fn attribute_columns_stay_out_of_the_merge_key() {
let c = base()
.attribute_column(Field::new("bilanzkreis", DataType::Utf8, true))
.build()
.unwrap();
assert_eq!(c.merge_key(), vec!["malo_id", "obis_code", "from"]);
}
#[test]
fn a_nullable_identity_column_is_rejected() {
let err = base()
.identity_column(Field::new("tenant", DataType::Utf8, true))
.build()
.unwrap_err();
assert!(err.to_string().contains("non-nullable"));
}
#[test]
fn identity_columns_come_before_attributes_in_the_schema() {
let c = base()
.attribute_column(Field::new("bilanzkreis", DataType::Utf8, true))
.identity_column(Field::new("tenant", DataType::Utf8, false))
.build()
.unwrap();
let names: Vec<_> = c.extra_columns().iter().map(|f| f.name().clone()).collect();
assert_eq!(names, ["tenant", "bilanzkreis"]);
}
#[test]
fn valid_extra_columns_are_kept_in_order() {
let c = base()
.attribute_column(Field::new("bilanzkreis", DataType::Utf8, true))
.attribute_column(Field::new("netzgebiet", DataType::Utf8, true))
.build()
.unwrap();
let extra = c.extra_columns();
let names: Vec<_> = extra.iter().map(|f| f.name().as_str()).collect();
assert_eq!(names, ["bilanzkreis", "netzgebiet"]);
}
#[test]
fn expected_hot_partitions_covers_lag_plus_headroom() {
assert_eq!(base().build().unwrap().expected_hot_partitions(), 21);
let weekly = base()
.archival_step(Duration::weeks(1))
.settlement_lag(Duration::weeks(2))
.partition_headroom(Duration::weeks(2))
.build()
.unwrap();
assert_eq!(weekly.expected_hot_partitions(), 4);
}
#[test]
fn duration_formatting_is_human_readable() {
assert_eq!(fmt_duration(Duration::DAY), "1d");
assert_eq!(fmt_duration(Duration::weeks(1)), "7d");
assert_eq!(fmt_duration(Duration::hours(6)), "6h");
assert_eq!(fmt_duration(Duration::minutes(15)), "15m");
assert_eq!(fmt_duration(Duration::seconds(90)), "90s");
}
}