skade-katalog 0.1.8

// Apache-2.0 licensed.

//! Key encoding for redb tables.
//!
//! redb stores raw bytes; we use `\x1f` (ASCII Unit Separator) between
//! `(catalog_name, namespace_path, leaf)` components. Iceberg namespace
//! parts cannot themselves contain `\x1f`, and we additionally forbid `.`
//! inside any component (Iceberg's own conventions enforce this).

use iceberg::{NamespaceIdent, TableIdent};

pub(crate) const SEP: char = '\x1f';

/// Encode a namespace path. Multi-part namespaces are joined with `.`,
/// matching `iceberg-catalog-sql` behaviour.
pub(crate) fn ns_path(ns: &NamespaceIdent) -> String {
    ns.clone().inner().join(".")
}

/// Key for `NAMESPACES` table: `"{catalog}\x1f{ns_path}"`.
pub(crate) fn namespace_key(catalog: &str, ns: &NamespaceIdent) -> String {
    format!("{catalog}{SEP}{}", ns_path(ns))
}

/// Key for `NAMESPACE_PROPS` table: `"{catalog}\x1f{ns_path}\x1f{prop}"`.
pub(crate) fn namespace_prop_key(catalog: &str, ns: &NamespaceIdent, prop: &str) -> String {
    format!("{catalog}{SEP}{}{SEP}{prop}", ns_path(ns))
}

/// Inclusive lower bound for scanning all property entries of a namespace.
pub(crate) fn namespace_prop_prefix(catalog: &str, ns: &NamespaceIdent) -> String {
    format!("{catalog}{SEP}{}{SEP}", ns_path(ns))
}

/// Key for `TABLES` table: `"{catalog}\x1f{ns_path}\x1f{table}"`.
pub(crate) fn table_key(catalog: &str, table: &TableIdent) -> String {
    format!(
        "{catalog}{SEP}{}{SEP}{}",
        ns_path(table.namespace()),
        table.name()
    )
}

/// Inclusive lower bound for scanning all tables in a namespace.
pub(crate) fn table_prefix(catalog: &str, ns: &NamespaceIdent) -> String {
    format!("{catalog}{SEP}{}{SEP}", ns_path(ns))
}

/// Inclusive lower bound for scanning all keys in a catalog.
pub(crate) fn catalog_prefix(catalog: &str) -> String {
    format!("{catalog}{SEP}")
}

/// Compute an exclusive upper bound for a string prefix by incrementing the
/// last byte. Used to bound redb range scans.
pub(crate) fn prefix_upper(prefix: &str) -> String {
    let mut bytes = prefix.as_bytes().to_vec();
    // Walk back, incrementing the last non-0xFF byte.
    while let Some(last) = bytes.last_mut() {
        if *last < 0xFF {
            *last += 1;
            // After incrementing, the byte sequence may not be valid UTF-8 in
            // general, but since our prefixes always end with ASCII `\x1f` or
            // an ASCII catalog name + `\x1f`, incrementing yields ASCII again.
            return String::from_utf8(bytes).expect("prefix increment stays ASCII");
        } else {
            bytes.pop();
        }
    }
    // All bytes were 0xFF — caller should treat this as "scan to end".
    String::new()
}