use alloc::string::String;
use core::cmp::Ordering;
pub(crate) fn compare(collation: &str, a: &str, b: &str) -> Option<Ordering> {
let name = collation.trim();
let base = name.split(['.', '@']).next().unwrap_or(name);
if base.eq_ignore_ascii_case("C") || base.eq_ignore_ascii_case("POSIX") {
return Some(a.as_bytes().cmp(b.as_bytes()));
}
if base.eq_ignore_ascii_case("binary")
|| base
.rsplit_once('_')
.is_some_and(|(_, tail)| tail.eq_ignore_ascii_case("bin"))
{
return Some(a.as_bytes().cmp(b.as_bytes()));
}
let locale = icu_locale_core::Locale::try_from_str(&normalise(name)).ok()?;
let prefs = icu_collator::CollatorPreferences::from(&locale);
let collator = icu_collator::Collator::try_new(prefs, pg_options()).ok()?;
Some(
collator
.compare(a, b)
.then_with(|| a.as_bytes().cmp(b.as_bytes())),
)
}
pub(crate) fn sort_key(collation: &str, s: &str) -> Option<alloc::vec::Vec<u8>> {
let name = collation.trim();
let base = name.split(['.', '@']).next().unwrap_or(name);
if base.eq_ignore_ascii_case("C")
|| base.eq_ignore_ascii_case("POSIX")
|| base.eq_ignore_ascii_case("binary")
|| base
.rsplit_once('_')
.is_some_and(|(_, tail)| tail.eq_ignore_ascii_case("bin"))
{
return None;
}
Collated::resolve(name)?.sort_key_of(s)
}
pub(crate) struct Collated {
collator: Option<icu_collator::CollatorBorrowed<'static>>,
ascii_byte_order: bool,
}
impl Collated {
pub(crate) fn resolve(name: &str) -> Option<Self> {
if is_byte_wise(name) {
return Some(Self {
collator: None,
ascii_byte_order: true,
});
}
let normalised = normalise(name);
let locale = icu_locale_core::Locale::try_from_str(&normalised).ok()?;
let prefs = icu_collator::CollatorPreferences::from(&locale);
let collator = icu_collator::Collator::try_new(prefs, pg_options()).ok()?;
Some(Self {
collator: Some(collator),
ascii_byte_order: ascii_alnum_is_byte_order(&normalised),
})
}
pub(crate) const fn ascii_byte_order(&self) -> bool {
self.ascii_byte_order
}
pub(crate) fn sort_key_of(&self, s: &str) -> Option<alloc::vec::Vec<u8>> {
let c = self.collator.as_ref()?;
let mut key: alloc::vec::Vec<u8> = alloc::vec::Vec::new();
c.write_sort_key_to(s, &mut key).ok()?;
key.push(0);
key.extend_from_slice(s.as_bytes());
Some(key)
}
pub(crate) fn compare(&self, a: &str, b: &str) -> Ordering {
match &self.collator {
Some(c) => {
if self.ascii_byte_order && is_ascii_alnum_lower(a) && is_ascii_alnum_lower(b) {
return a.as_bytes().cmp(b.as_bytes());
}
c.compare(a, b).then_with(|| a.as_bytes().cmp(b.as_bytes()))
}
None => a.as_bytes().cmp(b.as_bytes()),
}
}
}
fn pg_options() -> icu_collator::options::CollatorOptions {
let mut o = icu_collator::options::CollatorOptions::default();
o.alternate_handling = Some(icu_collator::options::AlternateHandling::Shifted);
o.max_variable = Some(icu_collator::options::MaxVariable::Punctuation);
o
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct TextCompare {
pub fold_case: bool,
pub pads: bool,
pub order: Option<alloc::string::String>,
}
impl TextCompare {
pub fn is_plain_bytes(&self) -> bool {
!self.fold_case && !self.pads && self.order.is_none()
}
pub fn compare(&self, a: &str, b: &str) -> Option<core::cmp::Ordering> {
compare(self.order.as_deref()?, a, b)
}
}
pub(crate) use spg_sql::charset::MYSQL_CHARSET_DEFAULT_COLLATION;
pub(crate) fn mysql_charset_of_collation(name: &str) -> Option<&'static str> {
let lower = name.trim().to_ascii_lowercase();
if lower == "binary" {
return Some("binary");
}
MYSQL_CHARSET_DEFAULT_COLLATION
.iter()
.map(|(c, _)| *c)
.find(|c| lower.starts_with(*c) && lower.as_bytes().get(c.len()) == Some(&b'_'))
}
pub(crate) fn charset_default_collation(charset: &str) -> Option<&'static str> {
spg_sql::charset::charset_default_collation(charset)
}
pub(crate) const MYSQL_DEFAULT_CONNECTION_COLLATION: &str = "utf8mb4_0900_ai_ci";
pub(crate) fn pads_space(collation: Option<&str>) -> bool {
let Some(name) = collation else {
return false;
};
let n = name.trim();
if n.eq_ignore_ascii_case("binary") {
return false;
}
let lower = n.to_ascii_lowercase();
!(lower.contains("_0900_") || lower.contains("nopad"))
}
pub(crate) fn folds_case(collation: &str) -> bool {
if is_byte_wise(collation) {
return false;
}
!collation
.trim()
.rsplit_once('_')
.is_some_and(|(_, tail)| tail.eq_ignore_ascii_case("cs"))
}
pub(crate) fn is_byte_wise(collation: &str) -> bool {
spg_storage::collation_is_byte_wise(collation)
}
pub(crate) fn is_known(name: &str) -> bool {
let n = name.trim();
fn base(s: &str) -> &str {
let t = s.trim();
t.split(['.', '@']).next().unwrap_or(t)
}
let want = base(n);
if crate::collation_catalog::PG_COLLATIONS
.iter()
.any(|(_, c, ..)| c.eq_ignore_ascii_case(n) || base(c).eq_ignore_ascii_case(want))
{
return true;
}
let tag = normalise(n);
let lang: alloc::string::String = tag.split('-').next().unwrap_or(&tag).into();
if crate::collation_catalog::PG_COLLATIONS
.iter()
.filter(|(_, _, provider, ..)| *provider == "i")
.any(|(_, c, ..)| {
c.strip_suffix("-x-icu").is_some_and(|icu| {
icu.eq_ignore_ascii_case(&tag) || icu.eq_ignore_ascii_case(&lang)
})
})
{
return true;
}
let lower = n.to_ascii_lowercase();
if lower == "case_insensitive" || lower == "nocase" {
return true;
}
spg_sql::charset::is_mysql_collation(&lower)
}
pub(crate) fn is_spg_only_spelling(name: &str) -> bool {
let lower = name.trim().to_ascii_lowercase();
lower == "case_insensitive"
|| lower == "nocase"
|| lower == "binary"
|| lower.ends_with("_ci")
|| lower.ends_with("_cs")
|| lower.ends_with("_bin")
}
pub(crate) fn unknown_collation_error(name: &str, mysql: bool) -> crate::EngineError {
crate::EngineError::Unsupported(unknown_collation_text(name, mysql))
}
pub(crate) fn unknown_collation_text(name: &str, mysql: bool) -> alloc::string::String {
if mysql {
alloc::format!("Unknown collation: '{name}'")
} else {
alloc::format!("collation \"{name}\" for encoding \"UTF8\" does not exist")
}
}
pub(crate) fn is_supported(collation: &str) -> bool {
compare(collation, "a", "b").is_some()
}
pub(crate) fn is_ascii_alnum_lower(s: &str) -> bool {
let in_alphabet = |x: u8| (x.wrapping_sub(b'0') < 10) | (x.wrapping_sub(b'a') < 26);
let (chunks, rest) = s.as_bytes().as_chunks::<16>();
for c in chunks {
let mut ok = true;
for &x in c {
ok &= in_alphabet(x);
}
if !ok {
return false;
}
}
rest.iter().all(|&x| in_alphabet(x))
}
#[must_use]
pub(crate) const fn is_collatable(t: &spg_storage::DataType) -> bool {
use spg_storage::DataType as D;
matches!(
t,
D::Text
| D::Varchar(_)
| D::Char(_)
| D::Name
| D::TextArray
| D::VarcharArray
| D::CharArray
)
}
#[must_use]
pub(crate) fn not_collatable_error(type_name: &str) -> crate::EngineError {
crate::EngineError::Unsupported(alloc::format!(
"collations are not supported by type {type_name}"
))
}
fn ascii_alnum_is_byte_order(normalised: &str) -> bool {
let lang = normalised.split('-').next().unwrap_or(normalised);
matches!(
lang,
"und" | "en" | "de" | "fr" | "es" | "it" | "pt" | "nl" | "sv" | "tr" | "id" | "ms"
)
}
fn normalise(name: &str) -> String {
if name.eq_ignore_ascii_case("default")
|| name.eq_ignore_ascii_case("unicode")
|| name.eq_ignore_ascii_case("pg_unicode_fast")
|| name.eq_ignore_ascii_case("ucs_basic")
{
return String::from("und");
}
let head = name.split(['.', '@']).next().unwrap_or(name);
head.replace('_', "-")
}
pub(crate) fn column_key_is_bytewise(col: &spg_storage::ColumnSchema, mysql: bool) -> bool {
if !matches!(
col.ty,
spg_storage::DataType::Text
| spg_storage::DataType::Varchar(_)
| spg_storage::DataType::Char(_)
) {
return true;
}
if matches!(col.collation, spg_storage::Collation::CaseInsensitive) {
return false;
}
!mysql || matches!(col.collation, spg_storage::Collation::Binary)
}
#[cfg(test)]
#[path = "collate_survey.rs"]
mod survey;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn sort_key_bytes_order_the_way_compare_does() {
let words = [
"apple",
"Apple",
"APPLE",
"Bob",
"bob",
"client",
"DateStyle",
"Zebra",
"zebra",
"_under",
"cherry",
"de-luca",
"deluca",
"O'Brien",
"Obrien",
"résumé",
"resume",
"Résumé",
"1abc",
"",
" ",
"a",
"A",
"ä",
"Ä",
"z",
"Z",
"élan",
"elan",
];
for coll in ["en_US.utf8", "de_DE.utf8", "fr_FR.utf8"] {
for a in words {
for b in words {
let by_compare = compare(coll, a, b).expect("supported");
let ka = sort_key(coll, a).expect("supported");
let kb = sort_key(coll, b).expect("supported");
assert_eq!(
ka.cmp(&kb),
by_compare,
"{coll}: {a:?} vs {b:?} — sort keys say {:?}, compare says {by_compare:?}",
ka.cmp(&kb)
);
}
}
}
}
use alloc::vec;
use alloc::vec::Vec;
fn sorted<'a>(coll: &str, mut xs: Vec<&'a str>) -> Vec<&'a str> {
xs.sort_by(|a, b| compare(coll, a, b).expect("supported"));
xs
}
#[test]
fn en_us_reproduces_the_rules_measured_from_pg() {
assert_eq!(
sorted("en_US.utf8", vec!["b", "A", "a", "B"]),
["a", "A", "b", "B"]
);
assert_eq!(compare("en_US.utf8", "a", "A"), Some(Ordering::Less));
assert_eq!(
sorted("en_US.utf8", vec!["f", "ê", "E", "é", "e"]),
["e", "E", "é", "ê", "f"]
);
assert_eq!(
compare("en_US.utf8", "résumé", "resumes"),
Some(Ordering::Less)
);
assert_eq!(sorted("en_US.utf8", vec!["a", "1", "A"]), ["1", "a", "A"]);
let g = sorted("en_US.utf8", vec!["aB", "ab", "a-b", "a b"]);
assert_eq!(g[3], "aB", "case outranks punctuation: {g:?}");
assert_eq!(
sorted("en_US.utf8", vec!["a2", "a10", "a1"]),
["a1", "a10", "a2"]
);
assert_eq!(
sorted("en_US.utf8", vec!["中", "あ", "z", "Z"]),
["z", "Z", "あ", "中"]
);
}
#[test]
fn c_and_posix_are_byte_order() {
for c in ["C", "POSIX", "c", "posix"] {
assert_eq!(compare(c, "B", "a"), Some(Ordering::Less), "{c}");
assert_eq!(sorted(c, vec!["a", "B", "_"]), ["B", "_", "a"], "{c}");
}
}
#[test]
fn posix_locale_spellings_are_understood() {
for name in ["en_US", "en_US.utf8", "en_US.UTF-8", "en-US"] {
assert!(is_supported(name), "{name}");
assert_eq!(compare(name, "a", "A"), Some(Ordering::Less), "{name}");
}
}
#[test]
fn an_unknown_name_declines_instead_of_guessing() {
assert_eq!(compare("no_such_locale_at_all", "a", "b"), None);
assert!(!is_supported("no_such_locale_at_all"));
}
#[test]
fn punctuation_is_variable_weighted_like_pg() {
assert_eq!(
sorted("en_US.utf8", vec!["_under", "apple", "Zebra", "cherry"]),
["apple", "cherry", "_under", "Zebra"]
);
assert_eq!(
sorted("en_US.utf8", vec!["_id", "name", "zip", "_ts", "email"]),
["email", "_id", "name", "_ts", "zip"]
);
assert_eq!(
sorted("en_US.utf8", vec!["O'Brien", "Oakes", "Obrien", "O-Brien"]),
["Oakes", "Obrien", "O'Brien", "O-Brien"]
);
let g = sorted(
"en_US.utf8",
vec!["de luca", "deluca", "de-luca", "demarco"],
);
assert_eq!(
g[3], "demarco",
"the de-luca variants group before demarco: {g:?}"
);
}
#[test]
fn all_punctuation_values_now_match_pg() {
assert_eq!(
sorted("en_US.utf8", vec!["_", " ", "-", "."]),
[" ", "-", ".", "_"]
);
}
#[test]
fn a_locale_tailoring_actually_changes_the_answer() {
assert_eq!(compare("sv_SE.utf8", "z", "å"), Some(Ordering::Less));
assert_eq!(compare("en_US.utf8", "z", "å"), Some(Ordering::Greater));
}
#[test]
fn survey_pg18_collation_coverage() {
let all = super::survey::PG18_COLLATIONS;
let unsupported: Vec<&str> = all.iter().copied().filter(|n| !is_supported(n)).collect();
let supported = all.len() - unsupported.len();
assert_eq!(
unsupported,
Vec::<&str>::new(),
"{supported}/{} performable",
all.len()
);
}
}
#[cfg(test)]
mod ascii_shortcut_tests {
use super::*;
use alloc::string::String;
use alloc::vec::Vec;
const ALPHA: &[u8] = b"0123456789abcdefghijklmnopqrstuvwxyz";
fn corpus() -> Vec<String> {
let mut out: Vec<String> = Vec::new();
for &a in ALPHA {
out.push(String::from_utf8(alloc::vec![a]).unwrap());
for &b in ALPHA {
out.push(String::from_utf8(alloc::vec![a, b]).unwrap());
for &c in ALPHA {
out.push(String::from_utf8(alloc::vec![a, b, c]).unwrap());
for &d in ALPHA {
if (usize::from(a) + usize::from(b) + usize::from(c) + usize::from(d))
% 1301
== 0
{
out.push(String::from_utf8(alloc::vec![a, b, c, d]).unwrap());
}
}
}
}
}
out
}
fn first_disagreement(name: &str) -> Option<(String, String)> {
let c = Collated::resolve(name).expect("resolvable");
let icu = c.collator.as_ref().expect("a collator");
let mut by_bytes = corpus();
by_bytes.sort_unstable();
let mut by_coll = by_bytes.clone();
by_coll.sort_by(|x, y| {
icu.compare(x.as_str(), y.as_str())
.then_with(|| x.as_bytes().cmp(y.as_bytes()))
});
by_bytes
.iter()
.zip(by_coll.iter())
.find(|(a, b)| a != b)
.map(|(a, b)| (a.clone(), b.clone()))
}
#[test]
fn every_allowed_collation_orders_the_alphabet_by_bytes() {
for name in [
"und",
"en_US.utf8",
"en_GB",
"de_DE.utf8",
"fr_FR",
"es_ES",
"it_IT",
"pt_BR",
"nl_NL",
"sv_SE",
"tr_TR",
"id_ID",
"ms_MY",
"de_AT",
"fr_CA",
] {
let c = Collated::resolve(name).expect("resolvable");
assert!(
c.ascii_byte_order(),
"{name} must be on the allowlist for this test to mean anything"
);
assert_eq!(first_disagreement(name), None, "{name}");
}
}
#[test]
fn the_languages_that_retailor_the_alphabet_are_not_allowed() {
for (name, why) in [
("cs_CZ", "`ch` is a letter, sorting after `h`"),
("et_EE", "`z` sorts between `s` and `t`"),
("lt_LT", "`y` sorts after `i`"),
("da_DK", "`aa` is `å`, sorting after `z`"),
("hu_HU", "`cs`, `gy`, `sz` … are letters"),
] {
let c = Collated::resolve(name).expect("resolvable");
assert!(!c.ascii_byte_order(), "{name}: {why}");
assert!(
first_disagreement(name).is_some(),
"{name} was excluded for a reason that does not reproduce: {why}"
);
}
}
#[test]
fn the_chunked_walk_agrees_with_the_definition_at_every_length() {
let plain = |s: &str| {
s.as_bytes()
.iter()
.all(|&c| c.is_ascii_digit() || c.is_ascii_lowercase())
};
for len in [0usize, 1, 15, 16, 17, 31, 32, 33, 47, 48, 49, 192] {
let all_good: alloc::string::String = (0..len).map(|i| ALPHABET[i % 36]).collect();
assert_eq!(
is_ascii_alnum_lower(&all_good),
plain(&all_good),
"len {len}, every byte in the alphabet"
);
assert!(is_ascii_alnum_lower(&all_good), "len {len} is all alphabet");
for bad_at in 0..len {
for bad in ['A', ' ', '_', '{', '/', '\u{e9}', '\u{4e2d}'] {
let mut v: alloc::vec::Vec<char> = all_good.chars().collect();
v[bad_at] = bad;
let t: alloc::string::String = v.into_iter().collect();
assert_eq!(
is_ascii_alnum_lower(&t),
plain(&t),
"len {len}, `{bad}` at {bad_at}"
);
assert!(
!is_ascii_alnum_lower(&t),
"len {len}: `{bad}` at {bad_at} is not in the alphabet"
);
}
}
}
}
const ALPHABET: [char; 36] = [
'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h',
'i', 'j', 'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z',
];
#[test]
fn a_value_outside_the_alphabet_still_takes_the_collator() {
let c = Collated::resolve("en_US.utf8").expect("resolvable");
assert!(c.ascii_byte_order());
assert_eq!(c.compare("Bob", "apple"), Ordering::Greater);
assert_eq!("Bob".as_bytes().cmp("apple".as_bytes()), Ordering::Less);
assert_eq!(c.compare("bob", "apple"), Ordering::Greater);
assert!(!is_ascii_alnum_lower("a b"));
assert!(is_ascii_alnum_lower("a0z"));
assert!(!is_ascii_alnum_lower("A"));
assert!(!is_ascii_alnum_lower("\u{e9}"));
assert!(is_ascii_alnum_lower(""));
}
}