mod cursor;
pub mod decimal;
mod fidelity;
mod mapping;
mod override_type;
pub mod policy;
mod rivet_type;
mod source_column;
pub mod target;
pub use cursor::CursorState;
pub use fidelity::TypeFidelity;
#[allow(unused_imports)]
pub use mapping::{TypeMapping, build_arrow_field, derive_fidelity, rivet_type_to_arrow};
pub use override_type::parse_type_str;
pub use rivet_type::{RivetType, TimeUnit};
pub use source_column::SourceColumn;
#[allow(unused_imports)]
pub use source_column::ColumnOverride;
pub type ColumnOverrides = std::collections::HashMap<String, RivetType>;
pub fn overrides_for_table(all: &ColumnOverrides, table: &str) -> ColumnOverrides {
let mut out: ColumnOverrides = ColumnOverrides::new();
for (k, v) in all {
if !k.contains('.') {
out.insert(k.clone(), v.clone());
}
}
for (k, v) in all {
if let Some((t, col)) = k.split_once('.')
&& t == table
{
out.insert(col.to_string(), v.clone());
}
}
out
}
pub fn resolve_or(
overrides: &ColumnOverrides,
column: &str,
autodetect: impl FnOnce() -> RivetType,
) -> RivetType {
overrides.get(column).cloned().unwrap_or_else(autodetect)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn overrides_for_table_bare_applies_qualified_wins_foreign_excluded() {
let mut all = ColumnOverrides::new();
all.insert("v".into(), RivetType::Text);
all.insert("orders.v".into(), RivetType::Int64);
all.insert("ghost.v".into(), RivetType::Bool);
let orders = overrides_for_table(&all, "orders");
assert_eq!(orders.get("v"), Some(&RivetType::Int64), "qualified wins");
let users = overrides_for_table(&all, "users");
assert_eq!(users.get("v"), Some(&RivetType::Text), "bare fallback");
assert_eq!(users.len(), 1, "foreign-qualified keys never leak");
}
#[test]
fn resolve_or_prefers_override_then_autodetect() {
let mut ov = ColumnOverrides::new();
ov.insert(
"amount".into(),
RivetType::Decimal {
precision: 18,
scale: 2,
},
);
assert_eq!(
resolve_or(&ov, "amount", || panic!(
"autodetect must not run when overridden"
)),
RivetType::Decimal {
precision: 18,
scale: 2
}
);
assert_eq!(
resolve_or(&ov, "other", || RivetType::Int64),
RivetType::Int64
);
}
use crate::types::mapping::{META_FIDELITY, META_LOGICAL_TYPE, META_NATIVE_TYPE};
use arrow::datatypes::DataType;
#[test]
fn end_to_end_payments_schema_matches_definition_of_done() {
let cols: Vec<(SourceColumn, RivetType)> = vec![
(
SourceColumn::simple("id", "bigint", false),
RivetType::Int64,
),
(
SourceColumn::decimal("amount", "numeric", false, 18, 2),
RivetType::Decimal {
precision: 18,
scale: 2,
},
),
(
SourceColumn::simple("created_at", "timestamptz", false),
RivetType::Timestamp {
unit: TimeUnit::Microsecond,
timezone: Some("UTC".into()),
},
),
(
SourceColumn::simple("payload", "jsonb", true),
RivetType::Json,
),
];
let mappings: Vec<TypeMapping> = cols
.into_iter()
.map(|(s, t)| TypeMapping::from_source(&s, t))
.collect();
assert_eq!(mappings[0].fidelity, TypeFidelity::Exact);
assert_eq!(mappings[1].fidelity, TypeFidelity::Exact);
assert_eq!(mappings[2].fidelity, TypeFidelity::Exact);
assert_eq!(mappings[3].fidelity, TypeFidelity::LogicalString);
assert_eq!(mappings[0].arrow_type, Some(DataType::Int64));
assert_eq!(mappings[1].arrow_type, Some(DataType::Decimal128(18, 2)));
assert!(matches!(
mappings[2].arrow_type,
Some(DataType::Timestamp(_, Some(_)))
));
assert_eq!(mappings[3].arrow_type, Some(DataType::Utf8));
let amount_field = build_arrow_field(&mappings[1]).expect("amount");
assert_eq!(
amount_field
.metadata()
.get(META_NATIVE_TYPE)
.map(String::as_str),
Some("numeric")
);
assert_eq!(
amount_field
.metadata()
.get(META_FIDELITY)
.map(String::as_str),
Some("exact")
);
let payload_field = build_arrow_field(&mappings[3]).expect("payload");
assert_eq!(
payload_field
.metadata()
.get(META_LOGICAL_TYPE)
.map(String::as_str),
Some("json")
);
}
#[test]
fn mapping_helpers_reexported_for_contract_tests() {
let dec = RivetType::Decimal {
precision: 18,
scale: 2,
};
assert!(matches!(
rivet_type_to_arrow(&dec),
Some(DataType::Decimal128(18, 2))
));
assert_eq!(derive_fidelity(&dec), TypeFidelity::Exact);
}
}