wasm-dbms 0.10.1

// Rust guideline compliant 2026-03-01
// X-WHERE-CLAUSE, M-CANONICAL-DOCS

//! Join execution engine for cross-table queries.

use std::collections::HashSet;

use wasm_dbms_api::prelude::{
    ColumnDef, DbmsResult, JoinColumnDef, JoinType, OrderDirection, Query, Value,
};
use wasm_dbms_memory::prelude::MemoryProvider;

use crate::database::WasmDbmsDatabase;
use crate::schema::DatabaseSchema;

/// A row in the joined result, organized by source table.
type JoinedRow = Vec<(String, Vec<(ColumnDef, Value)>)>;

/// Schema columns of every table in a joined row, in row order.
type JoinedTableColumns = Vec<(String, &'static [ColumnDef])>;

/// Engine that executes join queries using nested-loop join.
pub struct JoinEngine<'a, Schema: ?Sized, M>
where
    Schema: DatabaseSchema<M>,
    M: MemoryProvider,
{
    schema: &'a Schema,
    _marker: std::marker::PhantomData<M>,
}

impl<'a, Schema: ?Sized, M> JoinEngine<'a, Schema, M>
where
    Schema: DatabaseSchema<M>,
    M: MemoryProvider,
{
    pub fn new(schema: &'a Schema) -> Self {
        Self {
            schema,
            _marker: std::marker::PhantomData,
        }
    }
}

impl<Schema: ?Sized, M> JoinEngine<'_, Schema, M>
where
    Schema: DatabaseSchema<M>,
    M: MemoryProvider,
{
    /// Executes a join query using nested-loop join.
    pub fn join(
        &self,
        dbms: &WasmDbmsDatabase<'_, M>,
        from_table: &str,
        query: Query,
    ) -> DbmsResult<Vec<Vec<(JoinColumnDef, Value)>>> {
        let mut table_columns: JoinedTableColumns = vec![(
            from_table.to_string(),
            self.schema.table_columns(from_table)?,
        )];
        for join in &query.joins {
            table_columns.push((join.table.clone(), self.schema.table_columns(&join.table)?));
        }

        if let Some(filter) = &query.filter {
            let groups: Vec<(&str, &[ColumnDef])> = table_columns
                .iter()
                .map(|(table, columns)| (table.as_str(), *columns))
                .collect();
            filter.validate_joined(&groups)?;
        }

        let from_rows = self
            .schema
            .select(dbms, from_table, Query::builder().all().build())?;

        let mut joined_rows: Vec<JoinedRow> = from_rows
            .into_iter()
            .map(|row| vec![(from_table.to_string(), row)])
            .collect();
        // Schema columns of the tables already in `joined_rows`, used to
        // NULL-pad outer joins even when one side has no rows.
        let mut left_tables: JoinedTableColumns = vec![table_columns[0].clone()];

        for (join_index, join) in query.joins.iter().enumerate() {
            let (left_table, left_col) = self.resolve_column_ref(&join.left_column, from_table);
            let (_right_table_ref, right_col) =
                self.resolve_column_ref(&join.right_column, &join.table);

            let (keep_unmatched_left, keep_unmatched_right) = match join.join_type {
                JoinType::Inner => (false, false),
                JoinType::Left => (true, false),
                JoinType::Right => (false, true),
                JoinType::Full => (true, true),
            };

            let right_rows = self.load_join_right_rows(
                dbms,
                &joined_rows,
                &join.table,
                &left_table,
                left_col,
                right_col,
                keep_unmatched_right,
            )?;

            let right_columns = table_columns[join_index + 1].1;

            joined_rows = self.nested_loop_join(
                joined_rows,
                &right_rows,
                &left_tables,
                &join.table,
                right_columns,
                &left_table,
                left_col,
                right_col,
                keep_unmatched_left,
                keep_unmatched_right,
            );
            left_tables.push((join.table.clone(), right_columns));
        }

        if let Some(filter) = &query.filter {
            let mut filtered_rows = Vec::with_capacity(joined_rows.len());
            for row in joined_rows {
                let groups: Vec<(&str, Vec<(ColumnDef, Value)>)> = row
                    .iter()
                    .map(|(t, cols)| (t.as_str(), cols.clone()))
                    .collect();
                // Evaluation errors (ambiguous or out-of-scope columns, invalid
                // operands) are query errors, not non-matching rows.
                if filter.matches_joined_row(&groups)? {
                    filtered_rows.push(row);
                }
            }
            joined_rows = filtered_rows;
        }

        for (column, direction) in query.order_by.iter().rev() {
            self.sort_joined_rows(&mut joined_rows, column, *direction);
        }

        let offset = query.offset.unwrap_or_default();
        if offset > 0 {
            if offset >= joined_rows.len() {
                joined_rows.clear();
            } else {
                joined_rows = joined_rows.into_iter().skip(offset).collect();
            }
        }

        if let Some(limit) = query.limit {
            joined_rows.truncate(limit);
        }

        let results = joined_rows
            .into_iter()
            .map(|row| self.flatten_joined_row(row, &query))
            .collect::<DbmsResult<Vec<_>>>()?;

        Ok(results)
    }

    #[expect(
        clippy::too_many_arguments,
        reason = "arguments are necessary for loading right table rows based on join conditions"
    )]
    fn load_join_right_rows(
        &self,
        dbms: &WasmDbmsDatabase<'_, M>,
        left_rows: &[JoinedRow],
        right_table: &str,
        left_table: &str,
        left_col: &str,
        right_col: &str,
        keep_unmatched_right: bool,
    ) -> DbmsResult<Vec<Vec<(ColumnDef, Value)>>> {
        let unique_join_values: Vec<Value> = {
            let mut seen = HashSet::new();
            left_rows
                .iter()
                .filter_map(|row| self.get_column_value(row, left_table, left_col).cloned())
                .filter(|value| !value.is_null())
                .filter(|value| seen.insert(value.clone()))
                .collect()
        };

        if keep_unmatched_right {
            return self
                .schema
                .select(dbms, right_table, Query::builder().all().build());
        }

        if unique_join_values.is_empty() {
            return Ok(Vec::new());
        }

        self.schema.select(
            dbms,
            right_table,
            Query::builder()
                .all()
                .filter(Some(wasm_dbms_api::prelude::Filter::in_list(
                    right_col,
                    unique_join_values,
                )))
                .build(),
        )
    }

    /// Unified nested-loop join.
    ///
    /// Unmatched rows kept by an outer join are padded with `NULL` values for
    /// every schema column of the missing side, taken from `left_tables` and
    /// `right_columns` rather than from sample rows, so the row shape stays
    /// complete when that side is empty.
    #[allow(clippy::too_many_arguments)]
    fn nested_loop_join(
        &self,
        left_rows: Vec<JoinedRow>,
        right_rows: &[Vec<(ColumnDef, Value)>],
        left_tables: &[(String, &'static [ColumnDef])],
        right_table: &str,
        right_columns: &[ColumnDef],
        left_table: &str,
        left_col: &str,
        right_col: &str,
        keep_unmatched_left: bool,
        keep_unmatched_right: bool,
    ) -> Vec<JoinedRow> {
        let mut results = Vec::new();
        let mut right_matched = vec![false; right_rows.len()];

        for left_row in &left_rows {
            let left_value = self.get_column_value(left_row, left_table, left_col);
            let mut matched = false;

            for (i, right_row) in right_rows.iter().enumerate() {
                let right_value = right_row
                    .iter()
                    .find(|(c, _)| c.name == right_col)
                    .map(|(_, v)| v);

                if let (Some(left), Some(right)) = (left_value, right_value)
                    && !left.is_null()
                    && !right.is_null()
                    && left == right
                {
                    let mut new_row = left_row.clone();
                    new_row.push((right_table.to_string(), right_row.clone()));
                    results.push(new_row);
                    right_matched[i] = true;
                    matched = true;
                }
            }

            if keep_unmatched_left && !matched {
                let mut new_row = left_row.clone();
                new_row.push((
                    right_table.to_string(),
                    self.null_pad_columns(right_columns),
                ));
                results.push(new_row);
            }
        }

        if keep_unmatched_right {
            for (i, right_row) in right_rows.iter().enumerate() {
                if !right_matched[i] {
                    let mut new_row: JoinedRow = left_tables
                        .iter()
                        .map(|(table_name, columns)| {
                            (table_name.clone(), self.null_pad_columns(columns))
                        })
                        .collect();
                    new_row.push((right_table.to_string(), right_row.clone()));
                    results.push(new_row);
                }
            }
        }

        results
    }

    /// Resolves a column reference to (table_name, column_name).
    fn resolve_column_ref<'a>(&self, field: &'a str, default_table: &'a str) -> (String, &'a str) {
        if let Some((table, column)) = field.split_once('.') {
            (table.to_string(), column)
        } else {
            (default_table.to_string(), field)
        }
    }

    /// Finds a column value in a joined row.
    fn get_column_value<'a>(
        &self,
        row: &'a JoinedRow,
        table: &str,
        column: &str,
    ) -> Option<&'a Value> {
        row.iter()
            .find(|(t, _)| t == table)
            .and_then(|(_, cols)| cols.iter().find(|(c, _)| c.name == column).map(|(_, v)| v))
    }

    /// Creates a NULL-padded row from a table's schema columns.
    fn null_pad_columns(&self, columns: &[ColumnDef]) -> Vec<(ColumnDef, Value)> {
        columns.iter().map(|col| (*col, Value::Null)).collect()
    }

    /// Sorts joined rows by a column.
    fn sort_joined_rows(&self, rows: &mut [JoinedRow], column: &str, direction: OrderDirection) {
        let (table, col) = if let Some((t, c)) = column.split_once('.') {
            (Some(t), c)
        } else {
            (None, column)
        };

        rows.sort_by(|a, b| {
            let a_val = self.find_value_in_joined_row(a, table, col);
            let b_val = self.find_value_in_joined_row(b, table, col);

            crate::database::sort_values_with_direction(a_val, b_val, direction)
        });
    }

    /// Finds a column value in a joined row, optionally scoped to a table.
    fn find_value_in_joined_row<'a>(
        &self,
        row: &'a JoinedRow,
        table: Option<&str>,
        column: &str,
    ) -> Option<&'a Value> {
        if let Some(table) = table {
            return self.get_column_value(row, table, column);
        }
        row.iter()
            .flat_map(|(_, cols)| cols)
            .find_map(|(col, value)| {
                if col.name == column {
                    Some(value)
                } else {
                    None
                }
            })
    }

    /// Flattens a joined row into the output format.
    fn flatten_joined_row(
        &self,
        row: JoinedRow,
        query: &Query,
    ) -> DbmsResult<Vec<(JoinColumnDef, Value)>> {
        let mut result = Vec::new();

        for (table_name, cols) in row {
            for (col, val) in cols {
                let mut candid_col = JoinColumnDef::from(col);
                candid_col.table = Some(table_name.clone());

                if !query.all_selected() {
                    let selected = query.raw_columns();
                    let qualified_name = format!("{table_name}.{col}", col = candid_col.name);
                    if !selected.contains(&candid_col.name) && !selected.contains(&qualified_name) {
                        continue;
                    }
                }

                result.push((candid_col, val));
            }
        }

        Ok(result)
    }
}

#[cfg(test)]
mod tests {

    use wasm_dbms_api::prelude::{
        Database as _, DbmsError, Filter, InsertRecord as _, JoinColumnDef, Nullable, Query,
        QueryError, TableSchema as _, Text, Uint32, Value,
    };
    use wasm_dbms_macros::{DatabaseSchema, Table};
    use wasm_dbms_memory::prelude::HeapMemoryProvider;

    use crate::prelude::{DbmsContext, WasmDbmsDatabase};

    // Use tables WITHOUT foreign key constraints so we can test all join
    // types including unmatched rows without FK validation failures.

    #[derive(Debug, Table, Clone, PartialEq, Eq)]
    #[table = "departments"]
    pub struct Department {
        #[primary_key]
        pub id: Uint32,
        pub name: Text,
        pub code: Nullable<Uint32>,
    }

    #[derive(Debug, Table, Clone, PartialEq, Eq)]
    #[table = "employees"]
    pub struct Employee {
        #[primary_key]
        pub id: Uint32,
        pub name: Text,
        pub dept_id: Uint32,
        pub dept_code: Nullable<Uint32>,
    }

    #[derive(DatabaseSchema)]
    #[tables(Department = "departments", Employee = "employees")]
    pub struct TestSchema;

    #[derive(Debug, Table, Clone, PartialEq, Eq)]
    #[table = "indexed_departments"]
    pub struct IndexedDepartment {
        #[primary_key]
        pub id: Uint32,
        pub name: Text,
    }

    #[derive(Debug, Table, Clone, PartialEq, Eq)]
    #[table = "indexed_employees"]
    pub struct IndexedEmployee {
        #[primary_key]
        pub id: Uint32,
        pub name: Text,
        #[index]
        pub dept_id: Uint32,
    }

    #[derive(DatabaseSchema)]
    #[tables(
        IndexedDepartment = "indexed_departments",
        IndexedEmployee = "indexed_employees"
    )]
    pub struct IndexedJoinSchema;

    fn setup() -> DbmsContext<HeapMemoryProvider> {
        let ctx = DbmsContext::new(HeapMemoryProvider::default());
        TestSchema::register_tables(&ctx).unwrap();
        ctx
    }

    fn setup_indexed() -> DbmsContext<HeapMemoryProvider> {
        let ctx = DbmsContext::new(HeapMemoryProvider::default());
        IndexedJoinSchema::register_tables(&ctx).unwrap();
        ctx
    }

    fn insert_dept(db: &WasmDbmsDatabase<'_, HeapMemoryProvider>, id: u32, name: &str) {
        let insert = DepartmentInsertRequest::from_values(&[
            (Department::columns()[0], Value::Uint32(Uint32(id))),
            (
                Department::columns()[1],
                Value::Text(Text(name.to_string())),
            ),
        ])
        .unwrap();
        db.insert::<Department>(insert).unwrap();
    }

    fn insert_emp(
        db: &WasmDbmsDatabase<'_, HeapMemoryProvider>,
        id: u32,
        name: &str,
        dept_id: u32,
    ) {
        let insert = EmployeeInsertRequest::from_values(&[
            (Employee::columns()[0], Value::Uint32(Uint32(id))),
            (Employee::columns()[1], Value::Text(Text(name.to_string()))),
            (Employee::columns()[2], Value::Uint32(Uint32(dept_id))),
        ])
        .unwrap();
        db.insert::<Employee>(insert).unwrap();
    }

    fn insert_indexed_dept(db: &WasmDbmsDatabase<'_, HeapMemoryProvider>, id: u32, name: &str) {
        let insert = IndexedDepartmentInsertRequest::from_values(&[
            (IndexedDepartment::columns()[0], Value::Uint32(Uint32(id))),
            (
                IndexedDepartment::columns()[1],
                Value::Text(Text(name.to_string())),
            ),
        ])
        .unwrap();
        db.insert::<IndexedDepartment>(insert).unwrap();
    }

    fn insert_indexed_emp(
        db: &WasmDbmsDatabase<'_, HeapMemoryProvider>,
        id: u32,
        name: &str,
        dept_id: u32,
    ) {
        let insert = IndexedEmployeeInsertRequest::from_values(&[
            (IndexedEmployee::columns()[0], Value::Uint32(Uint32(id))),
            (
                IndexedEmployee::columns()[1],
                Value::Text(Text(name.to_string())),
            ),
            (
                IndexedEmployee::columns()[2],
                Value::Uint32(Uint32(dept_id)),
            ),
        ])
        .unwrap();
        db.insert::<IndexedEmployee>(insert).unwrap();
    }

    #[test]
    fn test_inner_join_does_not_match_null_keys() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_dept(&db, 2, "hr");
        insert_emp(&db, 10, "alice", 1);
        insert_emp(&db, 11, "bob", 1);

        let query = Query::builder()
            .all()
            .inner_join("employees", "code", "dept_code")
            .build();
        let results = db.select_join("departments", query).unwrap();
        assert!(results.is_empty());
    }

    #[test]
    fn test_outer_join_keeps_rows_with_null_keys_unmatched() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_emp(&db, 10, "alice", 1);

        let query = Query::builder()
            .all()
            .full_join("employees", "code", "dept_code")
            .build();
        let results = db.select_join("departments", query).unwrap();
        assert_eq!(results.len(), 2);
        let non_null_ids: Vec<Vec<(&str, Value)>> = results
            .iter()
            .map(|row| {
                row.iter()
                    .filter(|(column, value)| column.name == "id" && !value.is_null())
                    .map(|(column, value)| {
                        (
                            column
                                .table
                                .as_deref()
                                .expect("join column must name its table"),
                            value.clone(),
                        )
                    })
                    .collect()
            })
            .collect();
        assert_eq!(
            non_null_ids,
            vec![
                vec![("departments", Value::from(1u32))],
                vec![("employees", Value::from(10u32))],
            ]
        );
    }

    #[test]
    fn test_inner_join() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_dept(&db, 2, "hr");
        insert_emp(&db, 10, "alice", 1);
        insert_emp(&db, 11, "bob", 1);

        let query = Query::builder()
            .all()
            .inner_join("employees", "id", "dept_id")
            .build();
        let results = db.select_join("departments", query).unwrap();
        // eng has 2 employees, hr has 0 → 2 rows
        assert_eq!(results.len(), 2);
    }

    #[test]
    fn test_left_join() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_dept(&db, 2, "hr");
        insert_emp(&db, 10, "alice", 1);

        let query = Query::builder()
            .all()
            .left_join("employees", "id", "dept_id")
            .build();
        let results = db.select_join("departments", query).unwrap();
        // eng has 1 employee, hr has 0 but LEFT keeps unmatched left → 2 rows
        assert_eq!(results.len(), 2);

        // Find hr's row: employee columns should be Null
        let hr_row = results
            .iter()
            .find(|row| {
                row.iter().any(|(col, val)| {
                    col.name == "name"
                        && col.table.as_deref() == Some("departments")
                        && *val == Value::Text(Text("hr".to_string()))
                })
            })
            .expect("hr should be in results");

        // hr's employee name should be Null
        let emp_name = hr_row
            .iter()
            .find(|(col, _)| col.name == "name" && col.table.as_deref() == Some("employees"))
            .expect("employee name column should exist for hr");
        assert_eq!(emp_name.1, Value::Null);
    }

    #[test]
    fn test_right_join() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_emp(&db, 10, "alice", 1);
        // charlie references dept 999 which doesn't exist (no FK constraint)
        insert_emp(&db, 11, "charlie", 999);

        let query = Query::builder()
            .all()
            .right_join("employees", "id", "dept_id")
            .build();
        let results = db.select_join("departments", query).unwrap();
        // alice matches eng, charlie (dept_id=999) is unmatched right → 2 rows
        assert_eq!(results.len(), 2);

        // charlie should have null department columns
        let charlie_row = results
            .iter()
            .find(|row| {
                row.iter().any(|(col, val)| {
                    col.name == "name"
                        && col.table.as_deref() == Some("employees")
                        && *val == Value::Text(Text("charlie".to_string()))
                })
            })
            .expect("charlie should be in results");

        let dept_name = charlie_row
            .iter()
            .find(|(col, _)| col.name == "name" && col.table.as_deref() == Some("departments"))
            .expect("department name column should exist for charlie");
        assert_eq!(dept_name.1, Value::Null);
    }

    #[test]
    fn test_full_join() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_dept(&db, 2, "hr");
        insert_emp(&db, 10, "alice", 1);
        // charlie references dept 999 which doesn't exist
        insert_emp(&db, 11, "charlie", 999);

        let query = Query::builder()
            .all()
            .full_join("employees", "id", "dept_id")
            .build();
        let results = db.select_join("departments", query).unwrap();
        // eng-alice matched (1), hr unmatched left (1), charlie unmatched right (1) = 3
        assert_eq!(results.len(), 3);
    }

    #[test]
    fn test_join_with_filter() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_dept(&db, 2, "hr");
        insert_emp(&db, 10, "alice", 1);
        insert_emp(&db, 11, "bob", 2);

        let query = Query::builder()
            .all()
            .inner_join("employees", "id", "dept_id")
            .and_where(Filter::eq(
                "departments.name",
                Value::Text(Text("eng".to_string())),
            ))
            .build();
        let results = db.select_join("departments", query).unwrap();
        assert_eq!(results.len(), 1);
    }

    #[test]
    fn test_join_with_order_by() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_dept(&db, 2, "hr");
        insert_emp(&db, 10, "zzz", 1);
        insert_emp(&db, 11, "aaa", 2);

        let query = Query::builder()
            .all()
            .inner_join("employees", "id", "dept_id")
            .order_by_asc("employees.name")
            .build();
        let results = db.select_join("departments", query).unwrap();
        assert_eq!(results.len(), 2);
        let first_name = results[0]
            .iter()
            .find(|(col, _)| col.name == "name" && col.table.as_deref() == Some("employees"))
            .unwrap();
        assert_eq!(first_name.1, Value::Text(Text("aaa".to_string())));
    }

    #[test]
    fn test_join_with_limit() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_dept(&db, 2, "hr");
        insert_emp(&db, 10, "alice", 1);
        insert_emp(&db, 11, "bob", 2);

        let query = Query::builder()
            .all()
            .inner_join("employees", "id", "dept_id")
            .limit(1)
            .build();
        let results = db.select_join("departments", query).unwrap();
        assert_eq!(results.len(), 1);
    }

    #[test]
    fn test_join_with_offset() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_dept(&db, 2, "hr");
        insert_emp(&db, 10, "alice", 1);
        insert_emp(&db, 11, "bob", 2);

        let query = Query::builder()
            .all()
            .inner_join("employees", "id", "dept_id")
            .offset(1)
            .build();
        let results = db.select_join("departments", query).unwrap();
        assert_eq!(results.len(), 1);
    }

    #[test]
    fn test_join_with_column_selection() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_emp(&db, 10, "alice", 1);

        let query = Query::builder()
            .field("departments.name")
            .field("employees.name")
            .inner_join("employees", "id", "dept_id")
            .build();
        let results = db.select_join("departments", query).unwrap();
        assert_eq!(results.len(), 1);
        assert_eq!(results[0].len(), 2);
    }

    #[test]
    fn test_inner_join_empty_result() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        // No employees

        let query = Query::builder()
            .all()
            .inner_join("employees", "id", "dept_id")
            .build();
        let results = db.select_join("departments", query).unwrap();
        assert!(results.is_empty());
    }

    #[test]
    fn test_join_offset_exceeding_results_returns_empty() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_emp(&db, 10, "alice", 1);

        let query = Query::builder()
            .all()
            .inner_join("employees", "id", "dept_id")
            .offset(100)
            .build();
        let results = db.select_join("departments", query).unwrap();
        assert!(results.is_empty());
    }

    #[test]
    fn test_join_on_indexed_column() {
        let ctx = setup_indexed();
        let db = WasmDbmsDatabase::oneshot(&ctx, IndexedJoinSchema);
        insert_indexed_dept(&db, 1, "eng");
        insert_indexed_dept(&db, 2, "hr");
        insert_indexed_emp(&db, 10, "alice", 1);
        insert_indexed_emp(&db, 11, "bob", 2);

        let query = Query::builder()
            .all()
            .inner_join(
                "indexed_employees",
                "indexed_departments.id",
                "indexed_employees.dept_id",
            )
            .build();
        let results = db.select_join("indexed_departments", query).unwrap();

        assert_eq!(results.len(), 2);
        assert!(results.iter().any(|row| {
            row.iter().any(|(column, value)| {
                column.name == "name"
                    && column.table.as_deref() == Some("indexed_employees")
                    && *value == Value::Text(Text("alice".to_string()))
            })
        }));
    }

    /// Returns the value of `table.column` in a joined row, if the column is present.
    fn joined_value<'a>(
        row: &'a [(JoinColumnDef, Value)],
        table: &str,
        column: &str,
    ) -> Option<&'a Value> {
        row.iter()
            .find(|(col, _)| col.table.as_deref() == Some(table) && col.name == column)
            .map(|(_, value)| value)
    }

    /// Asserts that `row` has every department and employee column, in schema
    /// order, and that the columns of `null_table` are all `NULL`.
    fn assert_complete_row_with_null_side(row: &[(JoinColumnDef, Value)], null_table: &str) {
        let shape: Vec<(Option<&str>, &str)> = row
            .iter()
            .map(|(col, _)| (col.table.as_deref(), col.name.as_str()))
            .collect();
        let expected_shape: Vec<(Option<&str>, &str)> = Department::columns()
            .iter()
            .map(|col| (Some("departments"), col.name))
            .chain(
                Employee::columns()
                    .iter()
                    .map(|col| (Some("employees"), col.name)),
            )
            .collect();
        assert_eq!(shape, expected_shape);

        for (col, value) in row {
            if col.table.as_deref() == Some(null_table) {
                assert_eq!(
                    *value,
                    Value::Null,
                    "{null_table}.{} must be NULL",
                    col.name
                );
            }
        }
    }

    #[test]
    fn test_left_join_with_empty_right_table_keeps_null_right_columns() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");

        let query = Query::builder()
            .all()
            .left_join("employees", "departments.id", "employees.dept_id")
            .build();
        let results = db.select_join("departments", query).unwrap();

        assert_eq!(results.len(), 1);
        assert_complete_row_with_null_side(&results[0], "employees");
        assert_eq!(
            joined_value(&results[0], "departments", "name"),
            Some(&Value::Text(Text("eng".to_string())))
        );
    }

    #[test]
    fn test_right_join_with_empty_left_table_keeps_null_left_columns() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_emp(&db, 10, "alice", 1);

        let query = Query::builder()
            .all()
            .right_join("employees", "departments.id", "employees.dept_id")
            .build();
        let results = db.select_join("departments", query).unwrap();

        assert_eq!(results.len(), 1);
        assert_complete_row_with_null_side(&results[0], "departments");
        assert_eq!(
            joined_value(&results[0], "employees", "name"),
            Some(&Value::Text(Text("alice".to_string())))
        );
    }

    #[test]
    fn test_full_join_with_empty_right_table_keeps_null_right_columns() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");

        let query = Query::builder()
            .all()
            .full_join("employees", "departments.id", "employees.dept_id")
            .build();
        let results = db.select_join("departments", query).unwrap();

        assert_eq!(results.len(), 1);
        assert_complete_row_with_null_side(&results[0], "employees");
    }

    #[test]
    fn test_full_join_with_empty_left_table_keeps_null_left_columns() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_emp(&db, 10, "alice", 1);

        let query = Query::builder()
            .all()
            .full_join("employees", "departments.id", "employees.dept_id")
            .build();
        let results = db.select_join("departments", query).unwrap();

        assert_eq!(results.len(), 1);
        assert_complete_row_with_null_side(&results[0], "departments");
    }

    #[test]
    fn test_full_join_with_unmatched_rows_on_both_sides_has_stable_shape() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_emp(&db, 10, "alice", 999);

        let query = Query::builder()
            .all()
            .full_join("employees", "departments.id", "employees.dept_id")
            .build();
        let results = db.select_join("departments", query).unwrap();

        assert_eq!(results.len(), 2);
        let eng_row = results
            .iter()
            .find(|row| joined_value(row, "departments", "id") == Some(&Value::Uint32(Uint32(1))))
            .expect("eng should be in results");
        assert_complete_row_with_null_side(eng_row, "employees");
        let alice_row = results
            .iter()
            .find(|row| joined_value(row, "employees", "id") == Some(&Value::Uint32(Uint32(10))))
            .expect("alice should be in results");
        assert_complete_row_with_null_side(alice_row, "departments");
    }

    #[test]
    fn test_join_filter_with_ambiguous_column_returns_error() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_emp(&db, 10, "alice", 1);

        let query = Query::builder()
            .all()
            .inner_join("employees", "departments.id", "employees.dept_id")
            .and_where(Filter::eq("id", Value::Uint32(Uint32(1))))
            .build();
        let result = db.select_join("departments", query);

        assert!(
            matches!(
                &result,
                Err(DbmsError::Query(QueryError::InvalidQuery(message)))
                    if message.contains("Ambiguous column 'id'")
            ),
            "expected ambiguous column error, got {result:?}"
        );
    }

    #[test]
    fn test_join_filter_with_out_of_scope_table_returns_error() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_emp(&db, 10, "alice", 1);

        let query = Query::builder()
            .all()
            .inner_join("employees", "departments.id", "employees.dept_id")
            .and_where(Filter::eq("projects.id", Value::Uint32(Uint32(1))))
            .build();
        let result = db.select_join("departments", query);

        assert!(
            matches!(
                &result,
                Err(DbmsError::Query(QueryError::InvalidQuery(message)))
                    if message.contains("Table 'projects' not in query scope")
            ),
            "expected out-of-scope table error, got {result:?}"
        );
    }

    #[test]
    fn test_join_filter_with_invalid_like_operand_returns_error() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_emp(&db, 10, "alice", 1);

        let query = Query::builder()
            .all()
            .inner_join("employees", "departments.id", "employees.dept_id")
            .and_where(Filter::like("employees.dept_id", "1%"))
            .build();
        let result = db.select_join("departments", query);

        assert!(
            matches!(
                &result,
                Err(DbmsError::Query(QueryError::InvalidQuery(message)))
                    if message.contains("LIKE operator can only be applied to Text values")
            ),
            "expected LIKE type error, got {result:?}"
        );
    }

    #[test]
    fn test_join_filter_with_ambiguous_column_returns_error_for_empty_join() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        let query = Query::builder()
            .all()
            .inner_join("employees", "departments.id", "employees.dept_id")
            .and_where(Filter::eq("id", Value::Uint32(Uint32(1))))
            .build();

        let result = db.select_join("departments", query);

        assert!(
            matches!(
                &result,
                Err(DbmsError::Query(QueryError::InvalidQuery(message)))
                    if message.contains("Ambiguous column 'id'")
            ),
            "expected ambiguous column error, got {result:?}"
        );
    }

    #[test]
    fn test_join_filter_with_out_of_scope_table_returns_error_for_empty_join() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        let query = Query::builder()
            .all()
            .inner_join("employees", "departments.id", "employees.dept_id")
            .and_where(Filter::eq("projects.id", Value::Uint32(Uint32(1))))
            .build();

        let result = db.select_join("departments", query);

        assert!(
            matches!(
                &result,
                Err(DbmsError::Query(QueryError::InvalidQuery(message)))
                    if message.contains("Table 'projects' not in query scope")
            ),
            "expected out-of-scope table error, got {result:?}"
        );
    }

    #[test]
    fn test_join_filter_with_invalid_like_operand_returns_error_for_empty_join() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        let query = Query::builder()
            .all()
            .inner_join("employees", "departments.id", "employees.dept_id")
            .and_where(Filter::like("employees.dept_id", "1%"))
            .build();

        let result = db.select_join("departments", query);

        assert!(
            matches!(
                &result,
                Err(DbmsError::Query(QueryError::InvalidQuery(message)))
                    if message.contains("LIKE operator can only be applied to Text values")
            ),
            "expected LIKE type error, got {result:?}"
        );
    }

    #[test]
    fn test_left_join_like_filter_skips_null_padded_rows() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_dept(&db, 2, "hr");
        insert_emp(&db, 10, "alice", 1);

        let query = Query::builder()
            .all()
            .left_join("employees", "departments.id", "employees.dept_id")
            .and_where(Filter::like("employees.name", "a%"))
            .build();
        let results = db.select_join("departments", query).unwrap();

        assert_eq!(results.len(), 1);
        assert_eq!(
            joined_value(&results[0], "employees", "name"),
            Some(&Value::Text(Text("alice".to_string())))
        );

        let query = Query::builder()
            .all()
            .left_join("employees", "departments.id", "employees.dept_id")
            .and_where(Filter::like("employees.name", "a%").not())
            .build();
        let results = db.select_join("departments", query).unwrap();

        assert_eq!(results.len(), 1);
        assert_eq!(
            joined_value(&results[0], "departments", "name"),
            Some(&Value::Text(Text("hr".to_string())))
        );
        assert_complete_row_with_null_side(&results[0], "employees");
    }

    #[test]
    fn test_join_filter_with_dangling_like_escape_returns_error() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_dept(&db, 1, "eng");
        insert_emp(&db, 10, "abc", 1);

        let query = Query::builder()
            .all()
            .inner_join("employees", "departments.id", "employees.dept_id")
            .and_where(Filter::like("employees.name", "abc\\"))
            .build();
        let result = db.select_join("departments", query);

        assert!(
            matches!(
                &result,
                Err(DbmsError::Query(QueryError::InvalidQuery(message)))
                    if message.contains("Invalid LIKE pattern")
            ),
            "expected invalid LIKE pattern error, got {result:?}"
        );
    }
}