fluidattacks-database-core-domain 0.1.0

use alloc::vec::Vec;

use super::error::SchemaError;
use crate::cell::{valid_decimal, CellType};
use crate::core::ColumnId;

/// One column of a table: its name, its cell type and whether it accepts nulls.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Column {
    name: ColumnId,
    cell_type: CellType,
    nullable: bool,
}

impl Column {
    #[must_use]
    pub const fn new(name: ColumnId, cell_type: CellType, nullable: bool) -> Self {
        Self {
            name,
            cell_type,
            nullable,
        }
    }

    #[must_use]
    pub const fn name(&self) -> &ColumnId {
        &self.name
    }

    #[must_use]
    pub const fn cell_type(&self) -> &CellType {
        &self.cell_type
    }

    #[must_use]
    pub const fn nullable(&self) -> bool {
        self.nullable
    }
}

/// The columns of a table, in the order of the row values, and its key. An
/// empty key is a table with no key.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Schema {
    columns: Vec<Column>,
    key: Vec<ColumnId>,
}

/// The first `(precision, scale)` of a decimal in `cell_type` that no target
/// can hold.
fn invalid_decimal(cell_type: &CellType) -> Option<(u8, u8)> {
    match cell_type {
        CellType::Decimal { precision, scale } => {
            (!valid_decimal(*precision, *scale)).then_some((*precision, *scale))
        }
        CellType::Array { element, .. } => invalid_decimal(element),
        _ => None,
    }
}

/// The first name in `items` that an earlier item repeats.
fn first_repeated<T>(items: &[T], name: impl Fn(&T) -> &ColumnId) -> Option<&ColumnId> {
    items
        .iter()
        .enumerate()
        .find(|(i, item)| {
            items
                .iter()
                .take(*i)
                .any(|earlier| name(earlier) == name(item))
        })
        .map(|(_, item)| name(item))
}

impl Schema {
    pub fn new(columns: Vec<Column>, key: Vec<ColumnId>) -> Result<Self, SchemaError> {
        if columns.is_empty() {
            return Err(SchemaError::NoColumns);
        }
        if let Some(name) = first_repeated(&columns, Column::name) {
            return Err(SchemaError::DuplicateColumn(name.clone()));
        }
        if let Some((column, (precision, scale))) = columns
            .iter()
            .find_map(|c| invalid_decimal(c.cell_type()).map(|d| (c.name(), d)))
        {
            return Err(SchemaError::InvalidDecimal {
                column: column.clone(),
                precision,
                scale,
            });
        }
        if let Some(name) = first_repeated(&key, |name| name) {
            return Err(SchemaError::DuplicateKeyColumn(name.clone()));
        }
        for name in &key {
            match columns.iter().find(|c| c.name() == name) {
                None => return Err(SchemaError::UnknownKeyColumn(name.clone())),
                Some(c) if c.nullable() => {
                    return Err(SchemaError::NullableKeyColumn(name.clone()))
                }
                Some(_) => {}
            }
        }
        Ok(Self { columns, key })
    }

    #[must_use]
    pub fn columns(&self) -> &[Column] {
        &self.columns
    }

    #[must_use]
    pub fn key(&self) -> &[ColumnId] {
        &self.key
    }

    #[must_use]
    pub fn column(&self, name: &ColumnId) -> Option<&Column> {
        self.columns.iter().find(|c| c.name() == name)
    }
}

#[cfg(test)]
mod tests {
    use alloc::boxed::Box;
    use alloc::vec;
    use alloc::vec::Vec;

    use super::{Column, Schema};
    use crate::cell::CellType;
    use crate::core::{ColumnId, Identifier};
    use crate::table::SchemaError;

    fn id(name: &str) -> ColumnId {
        ColumnId::new(Identifier::new(name))
    }

    fn column(name: &str, cell_type: CellType, nullable: bool) -> Column {
        Column::new(id(name), cell_type, nullable)
    }

    fn decimal(precision: u8, scale: u8) -> CellType {
        CellType::Decimal { precision, scale }
    }

    #[test]
    fn a_valid_schema_keeps_its_columns_and_key() {
        let columns = vec![
            column("id", CellType::Integer, false),
            column("amount", decimal(38, 37), true),
        ];
        let schema = Schema::new(columns.clone(), vec![id("id")]);
        assert_eq!(schema.as_ref().map(Schema::columns), Ok(columns.as_slice()));
        assert_eq!(schema.as_ref().map(Schema::key), Ok([id("id")].as_slice()));
    }

    #[test]
    fn a_table_may_have_no_key() {
        let schema = Schema::new(vec![column("id", CellType::Integer, true)], Vec::new());
        assert_eq!(schema.map(|s| s.key().is_empty()), Ok(true));
    }

    #[test]
    fn a_table_needs_columns() {
        assert_eq!(
            Schema::new(Vec::new(), Vec::new()),
            Err(SchemaError::NoColumns)
        );
    }

    #[test]
    fn repeated_columns_are_rejected_by_exact_name() {
        let repeated = vec![
            column("date", CellType::Date, false),
            column("date", CellType::Text, false),
        ];
        assert_eq!(
            Schema::new(repeated, Vec::new()),
            Err(SchemaError::DuplicateColumn(id("date")))
        );
        let cased = vec![
            column("subject", CellType::Text, false),
            column("SUBJECT", CellType::Text, false),
        ];
        assert!(Schema::new(cased, Vec::new()).is_ok());
    }

    #[test]
    fn invalid_decimals_are_rejected_inside_arrays_too() {
        let over_scale = vec![column("amount", decimal(38, 38), false)];
        assert_eq!(
            Schema::new(over_scale, Vec::new()),
            Err(SchemaError::InvalidDecimal {
                column: id("amount"),
                precision: 38,
                scale: 38
            })
        );
        let in_array = vec![column(
            "amounts",
            CellType::Array {
                element: Box::new(decimal(10, 12)),
                nullable_elements: false,
            },
            false,
        )];
        assert_eq!(
            Schema::new(in_array, Vec::new()),
            Err(SchemaError::InvalidDecimal {
                column: id("amounts"),
                precision: 10,
                scale: 12
            })
        );
        let zero_precision = vec![column("amount", decimal(0, 0), false)];
        assert!(Schema::new(zero_precision, Vec::new()).is_err());
    }

    #[test]
    fn key_columns_must_exist_once_and_not_be_nullable() {
        let columns = || {
            vec![
                column("id", CellType::Integer, false),
                column("subject", CellType::Text, true),
            ]
        };
        assert_eq!(
            Schema::new(columns(), vec![id("ID")]),
            Err(SchemaError::UnknownKeyColumn(id("ID")))
        );
        assert_eq!(
            Schema::new(columns(), vec![id("id"), id("id")]),
            Err(SchemaError::DuplicateKeyColumn(id("id")))
        );
        assert_eq!(
            Schema::new(columns(), vec![id("subject")]),
            Err(SchemaError::NullableKeyColumn(id("subject")))
        );
    }

    #[test]
    fn a_column_is_found_by_its_exact_name() {
        let schema = Schema::new(vec![column("subject", CellType::Text, true)], Vec::new());
        assert_eq!(
            schema.as_ref().map(|s| s.column(&id("subject")).is_some()),
            Ok(true)
        );
        assert_eq!(
            schema.as_ref().map(|s| s.column(&id("SUBJECT")).is_none()),
            Ok(true)
        );
    }
}