relmath-rs 0.5.0

use std::{
    collections::{BTreeMap, BTreeSet},
    error::Error,
    fmt,
};

use crate::traits::FiniteRelation;

/// Errors returned by schema-aware n-ary relation operations.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum NaryRelationError {
    /// The provided schema has no columns.
    EmptySchema,
    /// A column name is blank or only whitespace.
    BlankColumnName {
        /// The zero-based schema position with the blank name.
        index: usize,
    },
    /// A column name is repeated where uniqueness is required.
    DuplicateColumn {
        /// The repeated column name.
        column: String,
    },
    /// A row does not match the relation arity.
    RowArityMismatch {
        /// The schema arity expected by the relation.
        expected: usize,
        /// The number of values that actually appeared in the row.
        actual: usize,
    },
    /// An operation referenced a column that is not present in the schema.
    UnknownColumn {
        /// The missing column name.
        column: String,
    },
    /// A named-row input omitted a column required by the schema.
    MissingColumn {
        /// The missing column name.
        column: String,
    },
    /// A named-row input provided a column that is not in the schema.
    UnexpectedColumn {
        /// The unexpected column name.
        column: String,
    },
    /// An operation requires identical schemas but received different ones.
    SchemaMismatch {
        /// The left-hand schema.
        left: Vec<String>,
        /// The right-hand schema.
        right: Vec<String>,
    },
}

impl fmt::Display for NaryRelationError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::EmptySchema => f.write_str("n-ary relation schema cannot be empty"),
            Self::BlankColumnName { index } => {
                write!(f, "column at index {index} must have a non-blank name")
            }
            Self::DuplicateColumn { column } => {
                write!(f, "column `{column}` appears more than once")
            }
            Self::RowArityMismatch { expected, actual } => write!(
                f,
                "row arity mismatch: expected {expected} values but found {actual}"
            ),
            Self::UnknownColumn { column } => {
                write!(f, "column `{column}` is not present in the schema")
            }
            Self::MissingColumn { column } => {
                write!(f, "named row is missing required column `{column}`")
            }
            Self::UnexpectedColumn { column } => {
                write!(f, "named row contains unexpected column `{column}`")
            }
            Self::SchemaMismatch { left, right } => {
                write!(f, "schema mismatch: left = {:?}, right = {:?}", left, right)
            }
        }
    }
}

impl Error for NaryRelationError {}

/// A deterministic exact grouping of an n-ary relation by named key columns.
///
/// `GroupedRelation<T>` is the first G2 grouping surface for exact n-ary
/// relations. Groups are keyed by projected key rows and stored in a
/// `BTreeMap`, so group iteration order is deterministic.
///
/// In this first exact grouping slice, each grouped member relation keeps the
/// original relation schema rather than removing key columns. This avoids
/// introducing zero-ary relation semantics while keeping the grouped output
/// relation-first and easy to inspect.
///
/// The first exact aggregate on grouped output is [`Self::counts`], which
/// returns the number of stored rows in each group.
///
/// # Examples
///
/// ```rust
/// use relmath::NaryRelation;
///
/// let completed = NaryRelation::from_rows(
///     ["student", "course", "term"],
///     [
///         ["Alice", "Math", "Fall"],
///         ["Alice", "Physics", "Fall"],
///         ["Bob", "Math", "Spring"],
///     ],
/// )?;
///
/// let grouped = completed.group_by(["term", "student"])?;
///
/// assert_eq!(
///     grouped.key_schema(),
///     &["term".to_string(), "student".to_string()]
/// );
/// assert_eq!(
///     grouped.counts(),
///     vec![(vec!["Fall", "Alice"], 2), (vec!["Spring", "Bob"], 1)]
/// );
/// # Ok::<(), relmath::NaryRelationError>(())
/// ```
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct GroupedRelation<T: Ord> {
    key_schema: Vec<String>,
    member_schema: Vec<String>,
    groups: BTreeMap<Vec<T>, NaryRelation<T>>,
}

impl<T: Ord> GroupedRelation<T> {
    /// Returns the grouping key schema in order.
    #[must_use]
    pub fn key_schema(&self) -> &[String] {
        &self.key_schema
    }

    /// Returns the schema of each member relation.
    ///
    /// In the current exact grouping slice this is the full original relation
    /// schema.
    #[must_use]
    pub fn member_schema(&self) -> &[String] {
        &self.member_schema
    }

    /// Returns the number of groups.
    #[must_use]
    pub fn len(&self) -> usize {
        self.groups.len()
    }

    /// Returns `true` when the grouping contains no groups.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.groups.is_empty()
    }

    /// Returns the member relation for the given grouping key.
    #[must_use]
    pub fn group(&self, key: &[T]) -> Option<&NaryRelation<T>> {
        self.groups.get(key)
    }

    /// Returns an iterator over groups in deterministic key order.
    pub fn iter(&self) -> impl Iterator<Item = (&[T], &NaryRelation<T>)> {
        self.groups
            .iter()
            .map(|(key, group)| (key.as_slice(), group))
    }

    /// Returns the first exact aggregate over the grouped relation: row counts.
    ///
    /// Each count is the number of stored rows in the corresponding group after
    /// the base relation's set semantics have already deduplicated equal rows.
    /// Results follow the deterministic key order of the grouping.
    #[must_use]
    pub fn counts(&self) -> Vec<(Vec<T>, usize)>
    where
        T: Clone,
    {
        self.groups
            .iter()
            .map(|(key, group)| (key.clone(), group.len()))
            .collect()
    }
}

/// A deterministic exact n-ary relation with a named schema.
///
/// `NaryRelation<T>` is the first G2 building block for schema-aware
/// relations. It stores column names explicitly and keeps rows in a `BTreeSet`
/// so iteration order is deterministic.
///
/// In this initial G2 step, all cells in a row share the same value type `T`.
///
/// # Examples
///
/// ```rust
/// use relmath::NaryRelation;
///
/// let mut passed = NaryRelation::new(["student", "course", "status"])?;
///
/// assert_eq!(passed.column_index("course")?, 1);
/// assert!(passed.insert_row(["Alice", "Math", "passed"])?);
/// assert!(passed.insert_row(["Alice", "Physics", "passed"])?);
/// assert!(passed.insert_row(["Bob", "Physics", "in_progress"])?);
///
/// let completed = passed.select(|row| row[2] == "passed");
/// let students = completed.project(["student"])?;
/// let renamed = students.rename("student", "learner")?;
///
/// assert_eq!(passed.arity(), 3);
/// assert!(completed.contains_row(&["Alice", "Math", "passed"]));
/// assert_eq!(renamed.schema(), &["learner".to_string()]);
/// # Ok::<(), relmath::NaryRelationError>(())
/// ```
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct NaryRelation<T: Ord> {
    schema: Vec<String>,
    rows: BTreeSet<Vec<T>>,
}

impl<T: Ord> NaryRelation<T> {
    /// Creates an empty n-ary relation with the given schema.
    ///
    /// Column names must be unique and non-blank.
    pub fn new<I, S>(schema: I) -> Result<Self, NaryRelationError>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        Ok(Self {
            schema: Self::validate_schema(schema)?,
            rows: BTreeSet::new(),
        })
    }

    /// Creates an n-ary relation from a schema and rows.
    ///
    /// Column names must be unique and non-blank, and every row must have the
    /// same arity as the schema.
    pub fn from_rows<I, S, R, Row>(schema: I, rows: R) -> Result<Self, NaryRelationError>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
        R: IntoIterator<Item = Row>,
        Row: IntoIterator<Item = T>,
    {
        let schema = Self::validate_schema(schema)?;
        let arity = schema.len();
        let mut dedup_rows = BTreeSet::new();

        for row in rows {
            dedup_rows.insert(Self::collect_row(arity, row)?);
        }

        Ok(Self {
            schema,
            rows: dedup_rows,
        })
    }

    /// Creates an n-ary relation from a schema and named rows.
    ///
    /// This is the current std-only G2 onboarding boundary. Each input row
    /// must provide exactly the schema columns by name. The schema order stays
    /// explicit in the relation rather than being inferred from map iteration.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use std::collections::BTreeMap;
    ///
    /// use relmath::NaryRelation;
    ///
    /// let progress = NaryRelation::from_named_rows(
    ///     ["student", "course", "status"],
    ///     [
    ///         BTreeMap::from([
    ///             ("course", "Math"),
    ///             ("status", "passed"),
    ///             ("student", "Alice"),
    ///         ]),
    ///         BTreeMap::from([
    ///             ("course", "Physics"),
    ///             ("status", "in_progress"),
    ///             ("student", "Bob"),
    ///         ]),
    ///     ],
    /// )?;
    ///
    /// assert_eq!(
    ///     progress.to_rows(),
    ///     vec![
    ///         vec!["Alice", "Math", "passed"],
    ///         vec!["Bob", "Physics", "in_progress"],
    ///     ]
    /// );
    /// # Ok::<(), relmath::NaryRelationError>(())
    /// ```
    pub fn from_named_rows<I, S, R, K>(schema: I, rows: R) -> Result<Self, NaryRelationError>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
        R: IntoIterator<Item = BTreeMap<K, T>>,
        K: Into<String> + Ord,
    {
        let schema = Self::validate_schema(schema)?;
        let mut dedup_rows = BTreeSet::new();

        for named_row in rows {
            dedup_rows.insert(Self::collect_named_row(&schema, named_row)?);
        }

        Ok(Self {
            schema,
            rows: dedup_rows,
        })
    }

    /// Returns the schema column names in order.
    #[must_use]
    pub fn schema(&self) -> &[String] {
        &self.schema
    }

    /// Returns the arity of the relation.
    #[must_use]
    pub fn arity(&self) -> usize {
        self.schema.len()
    }

    /// Returns the zero-based position of a named column.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use relmath::NaryRelation;
    ///
    /// let students = NaryRelation::<&str>::new(["student", "course"])?;
    ///
    /// assert_eq!(students.column_index("student")?, 0);
    /// assert_eq!(students.column_index("course")?, 1);
    /// # Ok::<(), relmath::NaryRelationError>(())
    /// ```
    pub fn column_index(&self, column: &str) -> Result<usize, NaryRelationError> {
        self.schema
            .iter()
            .position(|candidate| candidate == column)
            .ok_or_else(|| NaryRelationError::UnknownColumn {
                column: column.to_string(),
            })
    }

    /// Inserts a row into the relation.
    ///
    /// Returns `true` when the row was not already present.
    pub fn insert_row<I>(&mut self, row: I) -> Result<bool, NaryRelationError>
    where
        I: IntoIterator<Item = T>,
    {
        Ok(self.rows.insert(Self::collect_row(self.arity(), row)?))
    }

    /// Returns `true` when the relation contains the given row.
    #[must_use]
    pub fn contains_row(&self, row: &[T]) -> bool {
        self.rows.contains(row)
    }

    /// Returns an iterator over the stored rows in deterministic order.
    pub fn iter(&self) -> impl Iterator<Item = &[T]> {
        self.rows.iter().map(Vec::as_slice)
    }

    /// Returns the rows as a deterministically ordered vector.
    #[must_use]
    pub fn to_rows(&self) -> Vec<Vec<T>>
    where
        T: Clone,
    {
        self.rows.iter().cloned().collect()
    }

    /// Exports the relation as named rows.
    ///
    /// The returned vector follows the relation's deterministic row order.
    /// Each row is materialized as a `BTreeMap`, so key lookup is by column
    /// name rather than schema position. Preserve [`Self::schema`] separately
    /// when schema order matters to a downstream consumer.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use std::collections::BTreeMap;
    ///
    /// use relmath::NaryRelation;
    ///
    /// let progress = NaryRelation::from_rows(
    ///     ["student", "course"],
    ///     [["Alice", "Math"], ["Bob", "Physics"]],
    /// )?;
    ///
    /// assert_eq!(
    ///     progress.to_named_rows(),
    ///     vec![
    ///         BTreeMap::from([
    ///             ("course".to_string(), "Math"),
    ///             ("student".to_string(), "Alice"),
    ///         ]),
    ///         BTreeMap::from([
    ///             ("course".to_string(), "Physics"),
    ///             ("student".to_string(), "Bob"),
    ///         ]),
    ///     ]
    /// );
    /// # Ok::<(), relmath::NaryRelationError>(())
    /// ```
    #[must_use]
    pub fn to_named_rows(&self) -> Vec<BTreeMap<String, T>>
    where
        T: Clone,
    {
        self.rows
            .iter()
            .map(|row| {
                self.schema
                    .iter()
                    .cloned()
                    .zip(row.iter().cloned())
                    .collect()
            })
            .collect()
    }

    /// Groups the relation by the named key columns.
    ///
    /// Group keys follow the exact order of `columns`. Empty key selections,
    /// duplicate key columns, and unknown columns are rejected by the current
    /// exact core.
    ///
    /// Each member relation keeps the original relation schema and contains the
    /// subset of rows whose projected key matches the group key. Groups are
    /// stored in a `BTreeMap`, so group iteration order is deterministic.
    ///
    /// The first exact aggregate over the grouped output is
    /// [`GroupedRelation::counts`], which reports the number of stored rows in
    /// each group.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use relmath::NaryRelation;
    ///
    /// let completed = NaryRelation::from_rows(
    ///     ["student", "course", "term"],
    ///     [
    ///         ["Alice", "Math", "Fall"],
    ///         ["Alice", "Physics", "Fall"],
    ///         ["Bob", "Math", "Spring"],
    ///     ],
    /// )?;
    ///
    /// let grouped = completed.group_by(["term", "student"])?;
    ///
    /// assert_eq!(
    ///     grouped.key_schema(),
    ///     &["term".to_string(), "student".to_string()]
    /// );
    /// assert_eq!(
    ///     grouped.counts(),
    ///     vec![(vec!["Fall", "Alice"], 2), (vec!["Spring", "Bob"], 1)]
    /// );
    /// # Ok::<(), relmath::NaryRelationError>(())
    /// ```
    pub fn group_by<I, S>(&self, columns: I) -> Result<GroupedRelation<T>, NaryRelationError>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
        T: Clone,
    {
        let columns: Vec<String> = columns
            .into_iter()
            .map(|column| column.as_ref().to_string())
            .collect();
        let key_schema = Self::validate_schema(columns)?;
        let mut indices = Vec::with_capacity(key_schema.len());

        for column in &key_schema {
            indices.push(self.column_index(column)?);
        }

        let mut raw_groups: BTreeMap<Vec<T>, BTreeSet<Vec<T>>> = BTreeMap::new();
        for row in &self.rows {
            let key = indices.iter().map(|index| row[*index].clone()).collect();
            raw_groups.entry(key).or_default().insert(row.clone());
        }

        let member_schema = self.schema.clone();
        let groups = raw_groups
            .into_iter()
            .map(|(key, rows)| {
                (
                    key,
                    Self {
                        schema: member_schema.clone(),
                        rows,
                    },
                )
            })
            .collect();

        Ok(GroupedRelation {
            key_schema,
            member_schema,
            groups,
        })
    }

    /// Returns the subset of rows that satisfy `predicate`.
    ///
    /// The output keeps the same schema as `self`. Surviving rows remain in the
    /// deterministic order induced by the underlying `BTreeSet`.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use relmath::NaryRelation;
    ///
    /// let progress = NaryRelation::from_rows(
    ///     ["student", "course", "status"],
    ///     [
    ///         ["Alice", "Math", "passed"],
    ///         ["Alice", "Physics", "passed"],
    ///         ["Bob", "Physics", "in_progress"],
    ///     ],
    /// )?;
    ///
    /// let completed = progress.select(|row| row[2] == "passed");
    ///
    /// assert_eq!(
    ///     completed.to_rows(),
    ///     vec![
    ///         vec!["Alice", "Math", "passed"],
    ///         vec!["Alice", "Physics", "passed"],
    ///     ]
    /// );
    /// # Ok::<(), relmath::NaryRelationError>(())
    /// ```
    #[must_use]
    pub fn select<F>(&self, mut predicate: F) -> Self
    where
        F: FnMut(&[T]) -> bool,
        T: Clone,
    {
        let rows = self
            .rows
            .iter()
            .filter(|row| predicate(row))
            .cloned()
            .collect();

        Self {
            schema: self.schema.clone(),
            rows,
        }
    }

    /// Projects the relation onto the named columns in the requested order.
    ///
    /// The output schema follows the exact order of `columns`. Duplicate
    /// column names, unknown columns, and empty projections are rejected by the
    /// current exact core.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use relmath::NaryRelation;
    ///
    /// let passed = NaryRelation::from_rows(
    ///     ["student", "course", "status"],
    ///     [
    ///         ["Alice", "Math", "passed"],
    ///         ["Cara", "Math", "passed"],
    ///         ["Alice", "Physics", "passed"],
    ///     ],
    /// )?;
    ///
    /// let projected = passed.project(["course", "student"])?;
    ///
    /// assert_eq!(
    ///     projected.schema(),
    ///     &["course".to_string(), "student".to_string()]
    /// );
    /// assert_eq!(
    ///     projected.to_rows(),
    ///     vec![
    ///         vec!["Math", "Alice"],
    ///         vec!["Math", "Cara"],
    ///         vec!["Physics", "Alice"],
    ///     ]
    /// );
    /// # Ok::<(), relmath::NaryRelationError>(())
    /// ```
    pub fn project<I, S>(&self, columns: I) -> Result<Self, NaryRelationError>
    where
        I: IntoIterator<Item = S>,
        S: AsRef<str>,
        T: Clone,
    {
        let columns: Vec<String> = columns
            .into_iter()
            .map(|column| column.as_ref().to_string())
            .collect();
        let projected_schema = Self::validate_schema(columns)?;
        let mut indices = Vec::with_capacity(projected_schema.len());

        for column in &projected_schema {
            indices.push(self.column_index(column)?);
        }

        let rows = self
            .rows
            .iter()
            .map(|row| indices.iter().map(|index| row[*index].clone()).collect())
            .collect();

        Ok(Self {
            schema: projected_schema,
            rows,
        })
    }

    /// Returns a relation with one column renamed.
    ///
    /// Renaming a column to its current name is a no-op. The new column name
    /// must be non-blank and must not duplicate another column in the schema.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use relmath::NaryRelation;
    ///
    /// let completed = NaryRelation::from_rows(
    ///     ["student", "course"],
    ///     [["Alice", "Math"], ["Bob", "Physics"]],
    /// )?;
    ///
    /// let renamed = completed.rename("student", "learner")?;
    ///
    /// assert_eq!(
    ///     renamed.schema(),
    ///     &["learner".to_string(), "course".to_string()]
    /// );
    /// assert_eq!(
    ///     renamed.to_rows(),
    ///     vec![vec!["Alice", "Math"], vec!["Bob", "Physics"]]
    /// );
    /// # Ok::<(), relmath::NaryRelationError>(())
    /// ```
    pub fn rename<S>(&self, from: &str, to: S) -> Result<Self, NaryRelationError>
    where
        S: Into<String>,
        T: Clone,
    {
        let to = to.into();
        let index = self.column_index(from)?;

        Self::validate_column_name(&to, index)?;

        if from != to && self.schema.iter().any(|column| column == &to) {
            return Err(NaryRelationError::DuplicateColumn { column: to });
        }

        let mut schema = self.schema.clone();
        schema[index] = to;

        Ok(Self {
            schema,
            rows: self.rows.clone(),
        })
    }

    /// Returns the union of `self` and `other` when schemas match exactly.
    ///
    /// Schema compatibility for the current exact core means exact schema
    /// equality, including column order.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use relmath::NaryRelation;
    ///
    /// let left = NaryRelation::from_rows(["student"], [["Alice"], ["Bob"]])?;
    /// let right = NaryRelation::from_rows(["student"], [["Bob"], ["Cara"]])?;
    ///
    /// let union = left.union(&right)?;
    ///
    /// assert_eq!(
    ///     union.to_rows(),
    ///     vec![vec!["Alice"], vec!["Bob"], vec!["Cara"]]
    /// );
    /// # Ok::<(), relmath::NaryRelationError>(())
    /// ```
    pub fn union(&self, other: &Self) -> Result<Self, NaryRelationError>
    where
        T: Clone,
    {
        self.ensure_same_schema(other)?;

        Ok(Self {
            schema: self.schema.clone(),
            rows: self.rows.union(&other.rows).cloned().collect(),
        })
    }

    /// Returns the intersection of `self` and `other` when schemas match
    /// exactly.
    ///
    /// Schema compatibility for the current exact core means exact schema
    /// equality, including column order.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use relmath::NaryRelation;
    ///
    /// let left = NaryRelation::from_rows(["student"], [["Alice"], ["Bob"]])?;
    /// let right = NaryRelation::from_rows(["student"], [["Bob"], ["Cara"]])?;
    ///
    /// let intersection = left.intersection(&right)?;
    ///
    /// assert_eq!(intersection.to_rows(), vec![vec!["Bob"]]);
    /// # Ok::<(), relmath::NaryRelationError>(())
    /// ```
    pub fn intersection(&self, other: &Self) -> Result<Self, NaryRelationError>
    where
        T: Clone,
    {
        self.ensure_same_schema(other)?;

        Ok(Self {
            schema: self.schema.clone(),
            rows: self.rows.intersection(&other.rows).cloned().collect(),
        })
    }

    /// Returns the set difference `self \ other` when schemas match exactly.
    ///
    /// Schema compatibility for the current exact core means exact schema
    /// equality, including column order.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use relmath::NaryRelation;
    ///
    /// let left = NaryRelation::from_rows(["student"], [["Alice"], ["Bob"]])?;
    /// let right = NaryRelation::from_rows(["student"], [["Bob"], ["Cara"]])?;
    ///
    /// let only_left = left.difference(&right)?;
    ///
    /// assert_eq!(only_left.to_rows(), vec![vec!["Alice"]]);
    /// # Ok::<(), relmath::NaryRelationError>(())
    /// ```
    pub fn difference(&self, other: &Self) -> Result<Self, NaryRelationError>
    where
        T: Clone,
    {
        self.ensure_same_schema(other)?;

        Ok(Self {
            schema: self.schema.clone(),
            rows: self.rows.difference(&other.rows).cloned().collect(),
        })
    }

    /// Returns the natural join of `self` and `other`.
    ///
    /// Shared column names constrain row matching: a pair of rows joins only
    /// when every shared column has equal values on both sides. When the
    /// schemas are disjoint, natural join behaves as a cartesian product.
    ///
    /// The output schema keeps the entire left-hand schema in order, followed
    /// by the right-hand columns that are not shared, in their original order.
    /// Output rows are stored in a `BTreeSet`, so iteration remains
    /// deterministic. When no rows match, the result is empty but still keeps
    /// the joined schema.
    ///
    /// # Examples
    ///
    /// ```rust
    /// use relmath::NaryRelation;
    ///
    /// let completed = NaryRelation::from_rows(
    ///     ["student", "course"],
    ///     [
    ///         ["Alice", "Math"],
    ///         ["Alice", "Physics"],
    ///         ["Bob", "Math"],
    ///     ],
    /// )?;
    /// let rooms = NaryRelation::from_rows(
    ///     ["course", "room"],
    ///     [["Math", "R101"], ["Physics", "R201"]],
    /// )?;
    ///
    /// let scheduled = completed.natural_join(&rooms);
    ///
    /// assert_eq!(
    ///     scheduled.schema(),
    ///     &["student".to_string(), "course".to_string(), "room".to_string()]
    /// );
    /// assert_eq!(
    ///     scheduled.to_rows(),
    ///     vec![
    ///         vec!["Alice", "Math", "R101"],
    ///         vec!["Alice", "Physics", "R201"],
    ///         vec!["Bob", "Math", "R101"],
    ///     ]
    /// );
    /// # Ok::<(), relmath::NaryRelationError>(())
    /// ```
    pub fn natural_join(&self, other: &Self) -> Self
    where
        T: Clone,
    {
        let right_index_by_name: BTreeMap<&str, usize> = other
            .schema
            .iter()
            .enumerate()
            .map(|(index, column)| (column.as_str(), index))
            .collect();

        let mut shared_indices = Vec::new();
        let mut shared_right_indices = BTreeSet::new();
        for (left_index, column) in self.schema.iter().enumerate() {
            if let Some(&right_index) = right_index_by_name.get(column.as_str()) {
                shared_indices.push((left_index, right_index));
                shared_right_indices.insert(right_index);
            }
        }

        let mut schema = self.schema.clone();
        let mut right_only_indices = Vec::new();
        for (right_index, column) in other.schema.iter().enumerate() {
            if !shared_right_indices.contains(&right_index) {
                schema.push(column.clone());
                right_only_indices.push(right_index);
            }
        }

        let mut rows = BTreeSet::new();
        for left_row in &self.rows {
            for right_row in &other.rows {
                if shared_indices.iter().all(|(left_index, right_index)| {
                    left_row[*left_index] == right_row[*right_index]
                }) {
                    let mut joined_row =
                        Vec::with_capacity(self.arity() + right_only_indices.len());
                    joined_row.extend(left_row.iter().cloned());
                    for right_index in &right_only_indices {
                        joined_row.push(right_row[*right_index].clone());
                    }
                    rows.insert(joined_row);
                }
            }
        }

        Self { schema, rows }
    }

    fn validate_schema<I, S>(schema: I) -> Result<Vec<String>, NaryRelationError>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        let schema: Vec<String> = schema.into_iter().map(Into::into).collect();

        if schema.is_empty() {
            return Err(NaryRelationError::EmptySchema);
        }

        let mut seen = BTreeSet::new();
        for (index, column) in schema.iter().enumerate() {
            Self::validate_column_name(column, index)?;

            if !seen.insert(column.clone()) {
                return Err(NaryRelationError::DuplicateColumn {
                    column: column.clone(),
                });
            }
        }

        Ok(schema)
    }

    fn collect_row<I>(arity: usize, row: I) -> Result<Vec<T>, NaryRelationError>
    where
        I: IntoIterator<Item = T>,
    {
        let row: Vec<T> = row.into_iter().collect();

        if row.len() != arity {
            return Err(NaryRelationError::RowArityMismatch {
                expected: arity,
                actual: row.len(),
            });
        }

        Ok(row)
    }

    fn collect_named_row<K>(
        schema: &[String],
        named_row: BTreeMap<K, T>,
    ) -> Result<Vec<T>, NaryRelationError>
    where
        K: Into<String> + Ord,
    {
        let mut normalized_row = BTreeMap::new();
        for (column, value) in named_row {
            let column = column.into();
            if normalized_row.insert(column.clone(), value).is_some() {
                return Err(NaryRelationError::DuplicateColumn { column });
            }
        }

        let mut row = Vec::with_capacity(schema.len());
        for column in schema {
            row.push(normalized_row.remove(column.as_str()).ok_or_else(|| {
                NaryRelationError::MissingColumn {
                    column: column.clone(),
                }
            })?);
        }

        if let Some((column, _)) = normalized_row.into_iter().next() {
            return Err(NaryRelationError::UnexpectedColumn { column });
        }

        Ok(row)
    }

    fn validate_column_name(column: &str, index: usize) -> Result<(), NaryRelationError> {
        if column.trim().is_empty() {
            Err(NaryRelationError::BlankColumnName { index })
        } else {
            Ok(())
        }
    }

    fn ensure_same_schema(&self, other: &Self) -> Result<(), NaryRelationError> {
        if self.schema == other.schema {
            Ok(())
        } else {
            Err(NaryRelationError::SchemaMismatch {
                left: self.schema.clone(),
                right: other.schema.clone(),
            })
        }
    }
}

impl<T: Ord> FiniteRelation for NaryRelation<T> {
    fn len(&self) -> usize {
        self.rows.len()
    }
}