//! Outer join differentiation.
//!
//! LEFT JOIN = INNER JOIN + anti-join for non-matching left rows.
//!
//! Differentiate the inner join part normally, then handle the anti-join:
//! - Left rows that lose their last match → INSERT with NULL right columns
//! - Left rows that gain their first match → DELETE the NULL-padded row
//!
//! ## EC-01 fix: R₀ for DELETE deltas in Part 1 and Part 3
//!
//! Part 1 (delta_left ⋈ right) and Part 3 (delta_left anti-join right)
//! together partition all delta_left rows into "matched" and "unmatched".
//! When the right partner is simultaneously deleted, the post-change
//! right table (R₁) no longer contains that partner, causing:
//! - Part 1 to miss the DELETE (no match in R₁)
//! - Part 3 to emit a DELETE with NULL right cols (anti-join succeeds)
//!
//! Fix: split both parts by action:
//! - Part 1a/3a: INSERTs check R₁ (inserts need current right state)
//! - Part 1b/3b: DELETEs check R₀ (deletes need pre-change right state)
//!
//! R₀ = R_current EXCEPT ALL ΔR_inserts UNION ALL ΔR_deletes

use crate::dvm::diff::{DiffContext, DiffResult, quote_ident};
use crate::dvm::operators::join_common::{
    build_snapshot_sql, rewrite_join_condition, use_pre_change_snapshot,
};
use crate::dvm::parser::OpTree;
use crate::error::PgTrickleError;

/// Differentiate a LeftJoin node.
pub fn diff_left_join(ctx: &mut DiffContext, op: &OpTree) -> Result<DiffResult, PgTrickleError> {
    let OpTree::LeftJoin {
        condition,
        left,
        right,
    } = op
    else {
        return Err(PgTrickleError::InternalError(
            "diff_left_join called on non-LeftJoin node".into(),
        ));
    };

    // For LEFT JOIN, we reuse inner join differentiation for the matching part
    // and add the anti-join handling for non-matching left rows.

    // Differentiate both children
    let left_result = ctx.diff_node(left)?;
    let right_result = ctx.diff_node(right)?;

    // Rewrite join condition aliases for each part of the delta query.
    // For nested join children, column names are disambiguated with the
    // original table alias prefix (e.g., o.cust_id → dl."o__cust_id").
    let join_cond_part1 = rewrite_join_condition(condition, left, "dl", right, "r");
    let join_cond_part2 = rewrite_join_condition(condition, left, "l", right, "dr");
    let join_cond_antijoin = rewrite_join_condition(condition, left, "dl", right, "r");

    let left_cols = &left_result.columns;
    let right_cols = &right_result.columns;

    // Disambiguate output columns with table-alias prefix, matching
    // inner join convention so diff_project can resolve qualified refs.
    let left_prefix = left.alias();
    let right_prefix = right.alias();

    let mut output_cols = Vec::new();
    for c in left_cols {
        output_cols.push(format!("{left_prefix}__{c}"));
    }
    for c in right_cols {
        output_cols.push(format!("{right_prefix}__{c}"));
    }

    let right_table = build_snapshot_sql(right);
    let left_table = build_snapshot_sql(left);

    // Build column references with AS aliases for disambiguation
    let dl_cols: Vec<String> = left_cols
        .iter()
        .map(|c| {
            format!(
                "dl.{} AS {}",
                quote_ident(c),
                quote_ident(&format!("{left_prefix}__{c}"))
            )
        })
        .collect();
    let r_cols: Vec<String> = right_cols
        .iter()
        .map(|c| {
            format!(
                "r.{} AS {}",
                quote_ident(c),
                quote_ident(&format!("{right_prefix}__{c}"))
            )
        })
        .collect();
    let l_cols: Vec<String> = left_cols
        .iter()
        .map(|c| {
            format!(
                "l.{} AS {}",
                quote_ident(c),
                quote_ident(&format!("{left_prefix}__{c}"))
            )
        })
        .collect();
    let dr_cols: Vec<String> = right_cols
        .iter()
        .map(|c| {
            format!(
                "dr.{} AS {}",
                quote_ident(c),
                quote_ident(&format!("{right_prefix}__{c}"))
            )
        })
        .collect();
    let null_right_cols: Vec<String> = right_cols
        .iter()
        .map(|c| format!("NULL AS {}", quote_ident(&format!("{right_prefix}__{c}"))))
        .collect();

    let part1_cols = [dl_cols.as_slice(), r_cols.as_slice()].concat().join(", ");
    let part2_cols = [l_cols.as_slice(), dr_cols.as_slice()].concat().join(", ");
    let antijoin_cols = [dl_cols.as_slice(), null_right_cols.as_slice()]
        .concat()
        .join(", ");

    // For Parts 4 & 5: current_left JOIN conditions with different aliases
    // Part 4/5 JOIN: uses l and dr (same as Part 2)
    // Part 5 NOT EXISTS: uses l (current_left) and r (current_right)
    let not_exists_cond = rewrite_join_condition(condition, left, "l", right, "r");

    // R_old condition: uses l (current_left) and __pgt_r_old (pre-change right)
    let r_old_cond = rewrite_join_condition(condition, left, "l", right, "__pgt_r_old");

    // Build R_old snapshot for Parts 4/5: pre-change right state.
    // R_old = R_current EXCEPT ALL Δ_inserts UNION ALL Δ_deletes
    // Used to check whether a left row had ANY matching right row BEFORE
    // the current cycle's changes, preventing spurious NULL-padded D/I.
    let right_user_cols: Vec<&String> = right_cols.iter().filter(|c| *c != "__pgt_count").collect();
    let right_col_list: String = right_user_cols
        .iter()
        .map(|c| quote_ident(c))
        .collect::<Vec<_>>()
        .join(", ");
    let right_alias = right.alias();

    let r_old_snapshot = format!(
        "(SELECT {right_col_list} FROM {right_table} {ra} \
         EXCEPT ALL \
         SELECT {right_col_list} FROM {delta_right} WHERE __pgt_action = 'I' \
         UNION ALL \
         SELECT {right_col_list} FROM {delta_right} WHERE __pgt_action = 'D')",
        ra = quote_ident(right_alias),
        delta_right = right_result.cte_name,
    );

    // Null-padded columns for Parts 4 & 5 (left from `l`, right all NULL)
    let l_null_padded_cols = [l_cols.as_slice(), null_right_cols.as_slice()]
        .concat()
        .join(", ");

    // ── EC-01: Pre-change right snapshot for Part 1b / Part 3b ──────
    //
    // Same fix as diff_inner_join: when a left DELETE's old right
    // partner is simultaneously deleted, R₁ misses the match. Split
    // Part 1 and Part 3 by action, using R₀ for DELETEs.
    //
    // R₀ = R_current EXCEPT ALL ΔR_inserts UNION ALL ΔR_deletes
    let use_r0 = use_pre_change_snapshot(right, ctx.inside_semijoin);

    // Build R₀ for Parts 1b/3b (includes all right_cols for JOIN/anti-join).
    // Separate from r_old_snapshot (used for Parts 4/5 NOT EXISTS only,
    // which filters out __pgt_count).
    let r0_snapshot = if use_r0 {
        let right_all_cols: String = right_cols
            .iter()
            .map(|c| quote_ident(c))
            .collect::<Vec<_>>()
            .join(", ");
        let right_alias = right.alias();
        let r0 = format!(
            "(SELECT {right_all_cols} FROM {right_table} {ra} \
             EXCEPT ALL \
             SELECT {right_all_cols} FROM {delta_right} WHERE __pgt_action = 'I' \
             UNION ALL \
             SELECT {right_all_cols} FROM {delta_right} WHERE __pgt_action = 'D')",
            ra = quote_ident(right_alias),
            delta_right = right_result.cte_name,
        );
        Some(r0)
    } else {
        None
    };

    // When use_r0 is true, mark the right delta CTE as NOT MATERIALIZED
    // to prevent PostgreSQL from spilling temp files for the multiple
    // EXCEPT ALL / UNION ALL references.
    if use_r0 {
        ctx.mark_cte_not_materialized(&right_result.cte_name);
    }

    // ── G-J2: Pre-compute right-delta action flags ──────────────────
    //
    // Parts 4 and 5 each scan all current left rows joined with the right
    // delta. When the right delta is INSERT-only, Part 5 (which handles
    // right DELETEs) scans left_table for nothing. And vice versa.
    //
    // A single bool_or CTE evaluated once tells us which parts to run,
    // allowing PostgreSQL to skip the full left_table scan for whichever
    // part returns no rows.
    let flags_cte = ctx.next_cte_name("lj_right_flags");
    ctx.add_cte(
        flags_cte.clone(),
        format!(
            "SELECT bool_or(__pgt_action = 'I') AS has_ins,\
                    bool_or(__pgt_action = 'D') AS has_del \
             FROM {delta_right}",
            delta_right = right_result.cte_name,
        ),
    );

    let cte_name = ctx.next_cte_name("left_join");

    let sql = if let Some(ref r0) = r0_snapshot {
        // ── EC-01: Split Part 1 and Part 3 by action ────────────────
        // Part 1a: INSERTs ⋈ R₁  (inserts need current right partners)
        // Part 1b: DELETEs ⋈ R₀  (deletes need pre-change right partners)
        // Part 3a: INSERTs anti-join R₁  (no match in current right → NULL pad)
        // Part 3b: DELETEs anti-join R₀  (no match in pre-change right → NULL pad)
        format!(
            "\
-- Part 1a: delta_left INSERTS JOIN current_right R₁ (matching insert rows)
SELECT pgtrickle.pg_trickle_hash_multi(ARRAY[dl.__pgt_row_id::TEXT, pgtrickle.pg_trickle_hash(row_to_json(r)::text)::TEXT]) AS __pgt_row_id,
       dl.__pgt_action,
       {part1_cols}
FROM {delta_left} dl
JOIN {right_table} r ON {join_cond_part1}
WHERE dl.__pgt_action = 'I'

UNION ALL

-- Part 1b: delta_left DELETES JOIN pre-change_right R₀ (EC-01 fix)
-- R₀ = R_current EXCEPT ALL ΔR_inserts UNION ALL ΔR_deletes
-- Ensures deleted left rows find their old right partner even when
-- the right partner was simultaneously deleted.
SELECT pgtrickle.pg_trickle_hash_multi(ARRAY[dl.__pgt_row_id::TEXT, pgtrickle.pg_trickle_hash(row_to_json(r)::text)::TEXT]) AS __pgt_row_id,
       dl.__pgt_action,
       {part1_cols}
FROM {delta_left} dl
JOIN {r0_snapshot} r ON {join_cond_part1}
WHERE dl.__pgt_action = 'D'

UNION ALL

-- Part 2: current_left JOIN delta_right
SELECT pgtrickle.pg_trickle_hash_multi(ARRAY[pgtrickle.pg_trickle_hash(row_to_json(l)::text)::TEXT, dr.__pgt_row_id::TEXT]) AS __pgt_row_id,
       dr.__pgt_action,
       {part2_cols}
FROM {left_table} l
JOIN {delta_right} dr ON {join_cond_part2}

UNION ALL

-- Part 3a: delta_left INSERTS anti-join R₁ (non-matching inserts → NULL right cols)
SELECT dl.__pgt_row_id,
       dl.__pgt_action,
       {antijoin_cols}
FROM {delta_left} dl
WHERE dl.__pgt_action = 'I'
  AND NOT EXISTS (
    SELECT 1 FROM {right_table} r WHERE {join_cond_antijoin}
)

UNION ALL

-- Part 3b: delta_left DELETES anti-join R₀ (non-matching deletes → NULL right cols)
-- Uses R₀ to match the Part 1b partition: a delete that matched R₀ goes to
-- Part 1b, a delete that didn't match R₀ goes here.
SELECT dl.__pgt_row_id,
       dl.__pgt_action,
       {antijoin_cols}
FROM {delta_left} dl
WHERE dl.__pgt_action = 'D'
  AND NOT EXISTS (
    SELECT 1 FROM {r0_snapshot} r WHERE {join_cond_antijoin}
)

UNION ALL

-- Part 4: Delete stale NULL-padded rows when a left row gains its FIRST right match.
-- When a right INSERT creates a new match for a left row that previously had NO
-- matching right rows (was NULL-padded), the NULL-padded ST row must be removed.
-- We check R_old (pre-change right) to verify the left row truly had no matches
-- before. Without this check, left rows that ALREADY had matches would get
-- spurious D(NULL-padded) rows that corrupt intermediate aggregate old-state
-- reconstruction via EXCEPT ALL/UNION ALL.
SELECT 0::BIGINT AS __pgt_row_id,
       'D'::TEXT AS __pgt_action,
       {l_null_padded_cols}
FROM {left_table} l
JOIN {delta_right} dr ON {join_cond_part2}
WHERE dr.__pgt_action = 'I'
  AND (SELECT has_ins FROM {flags_cte})
  AND NOT EXISTS (
    SELECT 1 FROM {r_old_snapshot} __pgt_r_old WHERE {r_old_cond}
  )

UNION ALL

-- Part 5: Insert NULL-padded rows when a left row loses ALL right matches.
-- When a right DELETE removes the last match for a left row, the left row
-- reverts to NULL-padded. Check current right (post-changes) to verify no
-- remaining matches exist, AND check R_old to confirm the left row previously
-- HAD matches (otherwise it was already NULL-padded — no change needed).
SELECT 0::BIGINT AS __pgt_row_id,
       'I'::TEXT AS __pgt_action,
       {l_null_padded_cols}
FROM {left_table} l
JOIN {delta_right} dr ON {join_cond_part2}
WHERE dr.__pgt_action = 'D'
  AND (SELECT has_del FROM {flags_cte})
  AND NOT EXISTS (
    SELECT 1 FROM {right_table} r WHERE {not_exists_cond}
  )
  AND EXISTS (
    SELECT 1 FROM {r_old_snapshot} __pgt_r_old WHERE {r_old_cond}
  )",
            delta_left = left_result.cte_name,
            delta_right = right_result.cte_name,
            r0_snapshot = r0,
            flags_cte = flags_cte,
        )
    } else {
        // Right child is complex — keep Part 1 and Part 3 unsplit.
        format!(
            "\
-- Part 1: delta_left JOIN current_right (matching rows)
SELECT pgtrickle.pg_trickle_hash_multi(ARRAY[dl.__pgt_row_id::TEXT, pgtrickle.pg_trickle_hash(row_to_json(r)::text)::TEXT]) AS __pgt_row_id,
       dl.__pgt_action,
       {part1_cols}
FROM {delta_left} dl
JOIN {right_table} r ON {join_cond_part1}

UNION ALL

-- Part 2: current_left JOIN delta_right
SELECT pgtrickle.pg_trickle_hash_multi(ARRAY[pgtrickle.pg_trickle_hash(row_to_json(l)::text)::TEXT, dr.__pgt_row_id::TEXT]) AS __pgt_row_id,
       dr.__pgt_action,
       {part2_cols}
FROM {left_table} l
JOIN {delta_right} dr ON {join_cond_part2}

UNION ALL

-- Part 3: delta_left anti-join right (non-matching left rows get NULL right cols)
SELECT dl.__pgt_row_id,
       dl.__pgt_action,
       {antijoin_cols}
FROM {delta_left} dl
WHERE NOT EXISTS (
    SELECT 1 FROM {right_table} r WHERE {join_cond_antijoin}
)

UNION ALL

-- Part 4: Delete stale NULL-padded rows when a left row gains its FIRST right match.
SELECT 0::BIGINT AS __pgt_row_id,
       'D'::TEXT AS __pgt_action,
       {l_null_padded_cols}
FROM {left_table} l
JOIN {delta_right} dr ON {join_cond_part2}
WHERE dr.__pgt_action = 'I'
  AND (SELECT has_ins FROM {flags_cte})
  AND NOT EXISTS (
    SELECT 1 FROM {r_old_snapshot} __pgt_r_old WHERE {r_old_cond}
  )

UNION ALL

-- Part 5: Insert NULL-padded rows when a left row loses ALL right matches.
SELECT 0::BIGINT AS __pgt_row_id,
       'I'::TEXT AS __pgt_action,
       {l_null_padded_cols}
FROM {left_table} l
JOIN {delta_right} dr ON {join_cond_part2}
WHERE dr.__pgt_action = 'D'
  AND (SELECT has_del FROM {flags_cte})
  AND NOT EXISTS (
    SELECT 1 FROM {right_table} r WHERE {not_exists_cond}
  )
  AND EXISTS (
    SELECT 1 FROM {r_old_snapshot} __pgt_r_old WHERE {r_old_cond}
  )",
            delta_left = left_result.cte_name,
            delta_right = right_result.cte_name,
            flags_cte = flags_cte,
        )
    };

    ctx.add_cte(cte_name.clone(), sql);

    Ok(DiffResult {
        cte_name,
        columns: output_cols,
        is_deduplicated: false,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::dvm::operators::test_helpers::*;

    #[test]
    fn test_diff_left_join_basic() {
        let mut ctx = test_ctx();
        let left = scan(1, "orders", "public", "o", &["id", "cust_id", "amount"]);
        let right = scan(2, "customers", "public", "c", &["id", "name"]);
        let cond = eq_cond("o", "cust_id", "c", "id");
        let tree = left_join(cond, left, right);
        let result = diff_left_join(&mut ctx, &tree).unwrap();

        // Output columns should be disambiguated
        assert!(result.columns.contains(&"o__id".to_string()));
        assert!(result.columns.contains(&"c__name".to_string()));
    }

    #[test]
    fn test_diff_left_join_has_five_parts() {
        let mut ctx = test_ctx();
        let left = scan(1, "orders", "public", "o", &["id", "cust_id"]);
        let right = scan(2, "customers", "public", "c", &["id", "name"]);
        let cond = eq_cond("o", "cust_id", "c", "id");
        let tree = left_join(cond, left, right);
        let result = diff_left_join(&mut ctx, &tree).unwrap();
        let sql = ctx.build_with_query(&result.cte_name);

        // EC-01: Parts 1 and 3 are split when right child is simple (Scan)
        assert_sql_contains(&sql, "Part 1a");
        assert_sql_contains(&sql, "Part 1b");
        assert_sql_contains(&sql, "Part 2");
        assert_sql_contains(&sql, "Part 3a");
        assert_sql_contains(&sql, "Part 3b");
        assert_sql_contains(&sql, "Part 4");
        assert_sql_contains(&sql, "Part 5");
    }

    #[test]
    fn test_ec01_left_join_r0_uses_except_all() {
        // For Scan right children, Part 1b and Part 3b should use R₀ via
        // EXCEPT ALL to find pre-change right partners.
        let mut ctx = test_ctx();
        let left = scan(1, "a", "public", "a", &["id", "bid"]);
        let right = scan(2, "b", "public", "b", &["id", "name"]);
        let cond = eq_cond("a", "bid", "b", "id");
        let tree = left_join(cond, left, right);
        let result = diff_left_join(&mut ctx, &tree).unwrap();
        let sql = ctx.build_with_query(&result.cte_name);

        // R₀ uses EXCEPT ALL pattern
        assert_sql_contains(&sql, "EXCEPT ALL");
        // Part 1b filters DELETEs only
        assert_sql_contains(&sql, "Part 1b");
        // Part 3b filters DELETEs only
        assert_sql_contains(&sql, "Part 3b");
    }

    #[test]
    fn test_ec01_left_join_insert_delete_partition() {
        // Verify that Part 1a/3a handle INSERTs and Part 1b/3b handle DELETEs
        let mut ctx = test_ctx();
        let left = scan(1, "a", "public", "a", &["id"]);
        let right = scan(2, "b", "public", "b", &["id"]);
        let cond = eq_cond("a", "id", "b", "id");
        let tree = left_join(cond, left, right);
        let result = diff_left_join(&mut ctx, &tree).unwrap();
        let sql = ctx.build_with_query(&result.cte_name);

        // Count the INSERT/DELETE action filters
        let insert_filters = sql.matches("__pgt_action = 'I'").count();
        let delete_filters = sql.matches("__pgt_action = 'D'").count();
        // Part 1a uses 'I', Part 1b uses 'D', Part 3a uses 'I', Part 3b uses 'D'
        // Plus Parts 4/5 have action filters
        assert!(
            insert_filters >= 2,
            "expected at least 2 INSERT filters for Part 1a/3a, got {insert_filters}"
        );
        assert!(
            delete_filters >= 2,
            "expected at least 2 DELETE filters for Part 1b/3b, got {delete_filters}"
        );
    }

    #[test]
    fn test_diff_left_join_null_padding() {
        let mut ctx = test_ctx();
        let left = scan(1, "orders", "public", "o", &["id", "cust_id"]);
        let right = scan(2, "customers", "public", "c", &["id", "name"]);
        let cond = eq_cond("o", "cust_id", "c", "id");
        let tree = left_join(cond, left, right);
        let result = diff_left_join(&mut ctx, &tree).unwrap();
        let sql = ctx.build_with_query(&result.cte_name);

        // Anti-join part should pad right columns with NULL
        assert_sql_contains(&sql, "NULL AS");
    }

    #[test]
    fn test_diff_left_join_right_delta_flags() {
        let mut ctx = test_ctx();
        let left = scan(1, "a", "public", "a", &["id"]);
        let right = scan(2, "b", "public", "b", &["id"]);
        let cond = eq_cond("a", "id", "b", "id");
        let tree = left_join(cond, left, right);
        let result = diff_left_join(&mut ctx, &tree).unwrap();
        let sql = ctx.build_with_query(&result.cte_name);

        // G-J2 optimization: pre-computed right-delta action flags
        assert_sql_contains(&sql, "has_ins");
        assert_sql_contains(&sql, "has_del");
    }

    #[test]
    fn test_diff_left_join_not_deduplicated() {
        let mut ctx = test_ctx();
        let left = scan(1, "a", "public", "a", &["id"]);
        let right = scan(2, "b", "public", "b", &["id"]);
        let cond = eq_cond("a", "id", "b", "id");
        let tree = left_join(cond, left, right);
        let result = diff_left_join(&mut ctx, &tree).unwrap();
        assert!(!result.is_deduplicated);
    }

    #[test]
    fn test_diff_left_join_error_on_non_left_join_node() {
        let mut ctx = test_ctx();
        let tree = scan(1, "t", "public", "t", &["id"]);
        let result = diff_left_join(&mut ctx, &tree);
        assert!(result.is_err());
    }

    // ── Nested join tests ───────────────────────────────────────────

    #[test]
    fn test_diff_left_join_nested_three_tables() {
        // (a ⋈ b) LEFT JOIN c — left child is a nested inner join
        let a = scan(1, "a", "public", "a", &["id", "bid"]);
        let b = scan(2, "b", "public", "b", &["id"]);
        let inner = inner_join(eq_cond("a", "bid", "b", "id"), a, b);
        let c = scan(3, "c", "public", "c", &["id"]);
        let tree = left_join(eq_cond("a", "id", "c", "id"), inner, c);

        let mut ctx = test_ctx();
        let result = diff_left_join(&mut ctx, &tree);
        assert!(
            result.is_ok(),
            "nested 3-table left join should diff: {result:?}"
        );
        let dr = result.unwrap();
        let sql = ctx.build_with_query(&dr.cte_name);
        assert_sql_contains(&sql, "UNION ALL");
    }

    #[test]
    fn test_diff_left_join_nested_right_child() {
        // a LEFT JOIN (b ⋈ c) — right child is a nested inner join
        let a = scan(1, "a", "public", "a", &["id"]);
        let b = scan(2, "b", "public", "b", &["id", "cid"]);
        let c = scan(3, "c", "public", "c", &["id"]);
        let inner = inner_join(eq_cond("b", "cid", "c", "id"), b, c);
        let tree = left_join(eq_cond("a", "id", "b", "id"), a, inner);

        let mut ctx = test_ctx();
        let result = diff_left_join(&mut ctx, &tree);
        assert!(
            result.is_ok(),
            "left join with nested right child should diff: {result:?}"
        );
    }

    // ── NATURAL LEFT JOIN diff tests ────────────────────────────────

    #[test]
    fn test_diff_left_join_with_natural_condition() {
        // Simulate NATURAL LEFT JOIN: tables share "id" column
        let left = scan(1, "orders", "public", "o", &["id", "customer_id"]);
        let right = scan(2, "items", "public", "i", &["id", "order_id"]);
        let cond = natural_join_cond(&left, &right);
        let tree = left_join(cond, left, right);

        let mut ctx = test_ctx();
        let result = diff_left_join(&mut ctx, &tree).unwrap();
        let sql = ctx.build_with_query(&result.cte_name);
        // Left join diff should have multiple parts with UNION ALL
        assert_sql_contains(&sql, "UNION ALL");
        // Disambiguated columns from both sides
        assert!(result.columns.contains(&"o__id".to_string()));
        assert!(result.columns.contains(&"i__id".to_string()));
    }

    #[test]
    fn test_diff_left_join_natural_multiple_common_cols() {
        // Two tables sharing "id" and "region"
        let left = scan(1, "a", "public", "a", &["id", "region", "val"]);
        let right = scan(2, "b", "public", "b", &["id", "region", "score"]);
        let cond = natural_join_cond(&left, &right);
        let tree = left_join(cond, left, right);

        let mut ctx = test_ctx();
        let result = diff_left_join(&mut ctx, &tree).unwrap();
        assert!(result.columns.contains(&"a__id".to_string()));
        assert!(result.columns.contains(&"a__region".to_string()));
        assert!(result.columns.contains(&"b__region".to_string()));
        assert!(result.columns.contains(&"b__score".to_string()));
    }
}
