hamelin_translation 0.4.2

//! Fuse projection commands (SELECT, LET, DROP) into minimal SELECT commands.
//!
//! This pass runs after normalize_commands.
//! Pipeline-level pass contract: `Arc<TypedPipeline> -> Result<Arc<TypedPipeline>, ...>`

use std::collections::HashSet;
use std::sync::Arc;

use ordermap::OrderMap;

use hamelin_lib::{
    err::TranslationError,
    tree::{
        ast::{
            command::Command,
            expression::Expression,
            identifier::{CompoundIdentifier, Identifier, SimpleIdentifier},
            node::Span,
        },
        builder::{self, column_ref, field, select_command, ExpressionBuilder},
        typed_ast::{
            clause::Projections,
            command::{
                TypedCommand, TypedCommandKind, TypedDropCommand, TypedLetCommand,
                TypedSelectCommand,
            },
            context::StatementTranslationContext,
            environment::TypeEnvironment,
            expression::{TypedExpression, TypedExpressionKind},
            pipeline::TypedPipeline,
        },
    },
    types::{struct_type::Struct, Type},
};

/// Fuse projection commands (SELECT, LET, DROP) into minimal SELECT commands.
///
/// Pipeline-level pass contract: `Arc<TypedPipeline> -> Result<Arc<TypedPipeline>, ...>`
///
/// This creates a minimal number of SELECT commands by:
/// 1. Accumulating LET/DROP operations into a pending SELECT
/// 2. Detecting dependency barriers when LET references a field we assigned
/// 3. Emitting the pending SELECT when we hit a barrier or non-projection command
///
/// Returns the original `Rc` unchanged if no fusion is needed.
pub fn fuse_projections(
    pipeline: Arc<TypedPipeline>,
    ctx: &mut StatementTranslationContext,
) -> Result<Arc<TypedPipeline>, Arc<TranslationError>> {
    // Check if fusion is needed
    let needs_fusion = pipeline
        .valid_ref()?
        .commands
        .iter()
        .any(command_needs_fusion);

    if !needs_fusion {
        return Ok(pipeline);
    }

    let valid = pipeline.valid_ref()?;

    // Fuse commands
    let fused_commands = fuse_commands(&valid.commands)?;

    let mut pipe_builder = builder::pipeline().at(pipeline.ast.span.clone());
    for cmd in fused_commands {
        pipe_builder = pipe_builder.command(cmd);
    }
    let fused_ast = pipe_builder.build();

    // Re-typecheck
    Ok(Arc::new(TypedPipeline::from_ast_with_context(
        Arc::new(fused_ast),
        ctx,
    )))
}

/// Check if a command needs fusion (LET or DROP).
fn command_needs_fusion(cmd: &Arc<TypedCommand>) -> bool {
    matches!(
        &cmd.kind,
        TypedCommandKind::Let(_) | TypedCommandKind::Drop(_)
    )
}

/// Fuse commands into minimal SELECT commands.
fn fuse_commands(commands: &[Arc<TypedCommand>]) -> Result<Vec<Command>, Arc<TranslationError>> {
    let mut result: Vec<Command> = Vec::new();
    let mut pending: Option<PendingSelect> = None;

    for command in commands {
        match &command.kind {
            TypedCommandKind::Select(select_cmd) => {
                // SELECT is a barrier - emit pending and start new
                if let Some(p) = pending.take() {
                    result.push(p.emit());
                }
                pending = Some(PendingSelect::from_select(command, select_cmd)?);
            }

            TypedCommandKind::Let(let_cmd) => {
                let refs = extract_column_references_from_projections(&let_cmd.projections);

                if let Some(ref mut p) = pending {
                    // Check for dependency barrier
                    if refs
                        .iter()
                        .any(|r| p.assigned.iter().any(|a| identifiers_overlap(r, a)))
                    {
                        // BARRIER: LET references something we assigned
                        result.push(pending.take().unwrap().emit());
                        pending = Some(PendingSelect::from_let(command, let_cmd)?);
                    } else {
                        // FUSE: merge LET into pending SELECT
                        p.merge_let(command, let_cmd)?;
                    }
                } else {
                    pending = Some(PendingSelect::from_let(command, let_cmd)?);
                }
            }

            TypedCommandKind::Drop(drop_cmd) => {
                if let Some(ref mut p) = pending {
                    // DROP can always fuse - it just removes fields
                    p.merge_drop(command, drop_cmd);
                } else {
                    pending = Some(PendingSelect::from_drop(command, drop_cmd));
                }
            }

            _ => {
                // Non-projection command (WHERE, SORT, AGG, etc.) is a barrier
                if let Some(p) = pending.take() {
                    result.push(p.emit());
                }
                // Pass through the original AST command
                result.push(command.ast.as_ref().clone());
            }
        }
    }

    // Flush remaining
    if let Some(p) = pending {
        result.push(p.emit());
    }

    Ok(result)
}

/// Pending SELECT being accumulated during fusion.
///
/// Tracks assignments by identifier, preserving insertion order via OrderMap.
struct PendingSelect {
    /// The SELECT assignments being built (identifier -> AST expression)
    /// Only contains explicit assignments from LET/SELECT, NOT passthroughs.
    /// Passthroughs are computed at emit() time from output_schema.
    assignments: OrderMap<Identifier, Arc<Expression>>,
    /// Set of identifiers that have been explicitly assigned (not passthrough)
    /// Used for barrier detection (LET referencing assigned field).
    assigned: HashSet<Identifier>,
    /// Set of identifiers that have been dropped.
    /// Used to determine if struct passthroughs need to recurse.
    dropped: HashSet<Identifier>,
    /// Output schema after all fused commands have been applied.
    /// Used to generate passthroughs at emit() time.
    output_schema: Arc<TypeEnvironment>,
    /// AST span for the synthesized command
    span: Span,
}

impl PendingSelect {
    /// Initialize from a SELECT command
    ///
    /// SELECT replaces the schema entirely, so we store all its assignments
    /// and use its output_schema directly.
    fn from_select(
        command: &TypedCommand,
        select_cmd: &TypedSelectCommand,
    ) -> Result<Self, Arc<TranslationError>> {
        let mut assignments = ordermap::OrderMap::new();
        let mut assigned = HashSet::new();

        for assignment in &select_cmd.projections.assignments {
            let id = assignment.identifier.clone().valid()?;
            // Use the original AST expression
            assignments.insert(id.clone(), assignment.expression.ast.clone());
            // Track as assigned - LET referencing these fields should trigger barrier
            assigned.insert(id);
        }

        Ok(Self {
            assignments,
            assigned,
            dropped: HashSet::new(),
            output_schema: command.output_schema.clone(),
            span: command.ast.span,
        })
    }

    /// Initialize from a LET command (LET assignments first, then passthrough)
    ///
    /// LET prepends fields to the schema, so we put LET fields first.
    /// Passthroughs are NOT computed here - they're generated at emit() time
    /// based on output_schema minus assignments.
    fn from_let(
        command: &TypedCommand,
        let_cmd: &TypedLetCommand,
    ) -> Result<Self, Arc<TranslationError>> {
        let mut assignments = ordermap::OrderMap::new();
        let mut assigned = HashSet::new();

        // Store only the LET's explicit assignments
        for assignment in &let_cmd.projections.assignments {
            let id = assignment.identifier.clone().valid()?;
            assignments.insert(id.clone(), assignment.expression.ast.clone());
            assigned.insert(id);
        }

        Ok(Self {
            assignments,
            assigned,
            dropped: HashSet::new(),
            output_schema: command.output_schema.clone(),
            span: command.ast.span,
        })
    }

    /// Initialize from a DROP command (passthrough minus dropped fields)
    ///
    /// DROP doesn't have any explicit assignments - all fields are passthroughs
    /// except for the dropped ones. Passthroughs are generated at emit() time.
    fn from_drop(command: &TypedCommand, drop_cmd: &TypedDropCommand) -> Self {
        // DROP has no explicit assignments - everything comes from output_schema
        // Track dropped fields so struct passthroughs know to recurse
        let dropped = drop_cmd.dropped_fields.iter().cloned().collect();
        Self {
            assignments: ordermap::OrderMap::new(),
            assigned: HashSet::new(),
            dropped,
            output_schema: command.output_schema.clone(),
            span: command.ast.span,
        }
    }

    /// Merge a LET command into this pending SELECT
    ///
    /// LET prepends fields to the schema, so we need to insert at the front.
    /// We only track explicit assignments here - passthroughs are computed at emit().
    fn merge_let(
        &mut self,
        command: &TypedCommand,
        let_cmd: &TypedLetCommand,
    ) -> Result<(), Arc<TranslationError>> {
        // Build new assignments with LET fields first, then existing fields
        let mut new_assignments = OrderMap::new();

        // First, add LET's assignments (they go to the front)
        for assignment in &let_cmd.projections.assignments {
            let id = assignment.identifier.clone().valid()?;
            new_assignments.insert(id.clone(), assignment.expression.ast.clone());
            self.assigned.insert(id);
        }

        // Then add existing explicit assignments (skipping any that the new LET shadows)
        // If LET assigns x.y and we had x = expr, the new LET shadows our x.
        // If LET assigns x and we had x.y = expr, the new LET also shadows our x.y.
        for (id, expr) in self.assignments.drain(..) {
            if new_assignments
                .keys()
                .any(|new_id| identifiers_overlap(&id, new_id))
            {
                continue;
            }
            new_assignments.insert(id, expr);
        }

        self.assignments = new_assignments;
        self.output_schema = command.output_schema.clone();
        self.expand_span(&command.ast.span);
        Ok(())
    }

    /// Merge a DROP command into this pending SELECT
    ///
    /// DROP removes fields. We remove any explicit assignments for dropped fields,
    /// and update output_schema. Passthroughs will be computed at emit() time.
    fn merge_drop(&mut self, command: &TypedCommand, drop_cmd: &TypedDropCommand) {
        for dropped in &drop_cmd.dropped_fields {
            self.assignments.remove(dropped);
            self.assigned.remove(dropped);
            self.dropped.insert(dropped.clone());
        }
        self.output_schema = command.output_schema.clone();
        self.expand_span(&command.ast.span);
    }

    /// Expand the span to include another span (union of the two ranges)
    fn expand_span(&mut self, other: &Span) {
        if other.is_none() {
            return;
        }
        if self.span.is_none() {
            self.span = other.clone();
            return;
        }

        let self_start = self.span.start().unwrap();
        let self_end = self.span.end().unwrap();
        let other_start = other.start().unwrap();
        let other_end = other.end().unwrap();

        let new_start = self_start.min(other_start);
        let new_end = self_end.max(other_end);
        self.span = Span::new(new_start, new_end);
    }

    /// Emit as an AST SELECT command.
    ///
    /// Emits explicit assignments first (in order), then generates passthroughs
    /// for any fields in output_schema that aren't covered by assignments.
    /// Compound identifiers are preserved as-is; the IR conversion layer will
    /// pack them into struct literals.
    fn emit(self) -> Command {
        let mut builder = select_command();

        // First, emit all explicit assignments (LET fields first)
        for (identifier, expr) in &self.assignments {
            builder = builder.named_field(identifier.clone(), expr.clone());
        }

        // Then, generate passthroughs from output_schema for uncovered fields
        let assignment_set: HashSet<_> = self.assignments.keys().cloned().collect();
        let output_struct = self.output_schema.flatten();

        let mut passthroughs = Vec::new();
        emit_passthroughs_recursive(
            &output_struct,
            &assignment_set,
            &self.dropped,
            &[],
            &mut passthroughs,
        );

        for (identifier, expr) in passthroughs {
            builder = builder.named_field(identifier, expr);
        }

        builder.at(self.span).build()
    }
}

/// Check if any assignment or drop overlaps with a struct field.
///
/// Returns true if we need to recurse into the struct to handle partial coverage.
/// Returns false if we can emit the struct directly as a passthrough.
fn any_modification_overlaps_struct(
    assignments: &HashSet<Identifier>,
    dropped: &HashSet<Identifier>,
    struct_id: &Identifier,
) -> bool {
    let overlaps_assignments = assignments
        .iter()
        .any(|a| a == struct_id || a.has_prefix(struct_id) || struct_id.has_prefix(a));
    let overlaps_dropped = dropped
        .iter()
        .any(|d| d == struct_id || d.has_prefix(struct_id) || struct_id.has_prefix(d));
    overlaps_assignments || overlaps_dropped
}

/// Recursively emit passthroughs for fields in schema not covered by assignments.
///
/// This handles nested structs correctly:
/// - If a field is exactly in assignments → skip (fully covered)
/// - If a field is a struct with NO overlapping modifications → emit struct directly
/// - If a field is a struct with overlapping assignments/drops → recurse for leaf passthroughs
/// - Otherwise → emit passthrough for the leaf field
fn emit_passthroughs_recursive(
    schema: &Struct,
    assignments: &HashSet<Identifier>,
    dropped: &HashSet<Identifier>,
    prefix: &[SimpleIdentifier],
    output: &mut Vec<(Identifier, Expression)>,
) {
    for (field_name, field_type) in schema.fields.iter() {
        // Build the full identifier path
        let mut path = prefix.to_vec();
        path.push(field_name.clone().into());
        let full_id = identifier_from_path(&path);

        // Check if this field is exactly covered by an assignment
        if assignments.contains(&full_id) {
            continue;
        }

        // For struct types, check if any modification overlaps with this struct
        if let Type::Struct(inner_struct) = field_type {
            if any_modification_overlaps_struct(assignments, dropped, &full_id) {
                // Partial coverage - recurse to emit leaf-level passthroughs
                emit_passthroughs_recursive(inner_struct, assignments, dropped, &path, output);
            } else {
                // No overlap - emit the struct directly as a passthrough
                let passthrough_expr = synthesize_passthrough_ast(&full_id);
                output.push((full_id, passthrough_expr));
            }
            continue;
        }

        // Leaf field not covered - emit passthrough
        let passthrough_expr = synthesize_passthrough_ast(&full_id);
        output.push((full_id, passthrough_expr));
    }
}

/// Build an Identifier from a path of SimpleIdentifiers.
fn identifier_from_path(path: &[SimpleIdentifier]) -> Identifier {
    match path.len() {
        0 => panic!("identifier_from_path called with empty path"),
        1 => path[0].clone().into(),
        2 => CompoundIdentifier::new(path[0].clone(), path[1].clone(), vec![]).into(),
        _ => CompoundIdentifier::new(path[0].clone(), path[1].clone(), path[2..].to_vec()).into(),
    }
}

/// Synthesize a passthrough AST expression for an identifier.
///
/// For simple identifiers (x), creates a column reference.
/// For compound identifiers (x.y.z), creates a field lookup chain.
fn synthesize_passthrough_ast(identifier: &Identifier) -> Expression {
    match identifier {
        Identifier::Simple(simple) => column_ref(simple.as_str()).build(),
        Identifier::Compound(compound) => {
            // Build chain: column_ref(first).field(second).field(third)...
            let parts = &compound.parts;
            assert!(!parts.is_empty());

            let mut current: Box<dyn ExpressionBuilder> = Box::new(column_ref(parts[0].as_str()));

            for part in &parts[1..] {
                current = Box::new(field(current, part.as_str()));
            }

            current.build()
        }
    }
}

/// Check if two identifiers overlap (one is prefix of the other, or they're equal).
fn identifiers_overlap(a: &Identifier, b: &Identifier) -> bool {
    a == b || a.has_prefix(b) || b.has_prefix(a)
}

/// Extract all column references from a Projections structure.
fn extract_column_references_from_projections(projections: &Projections) -> HashSet<Identifier> {
    let mut refs = HashSet::new();
    for assignment in &projections.assignments {
        extract_column_references_from_expression(&assignment.expression, &mut refs);
    }
    refs
}

/// Extract all column references from a TypedExpression using find() traversal.
fn extract_column_references_from_expression(
    expr: &TypedExpression,
    refs: &mut HashSet<Identifier>,
) {
    // Use find() to traverse all nodes, collecting column references
    // We return false to keep searching (visit all nodes)
    expr.find(&mut |e| {
        if let TypedExpressionKind::ColumnReference(col_ref) = &e.kind {
            if let Ok(simple) = col_ref.column_name.clone().valid() {
                refs.insert(simple.into());
            }
        }
        false // Continue searching to visit all nodes
    });
}

#[cfg(test)]
mod tests {
    use super::*;
    use hamelin_lib::{
        tree::ast::identifier::CompoundIdentifier,
        tree::{
            ast::{pipeline::Pipeline, IntoTyped, TypeCheckExecutor},
            builder::{
                add, array, column_ref, drop_command, let_command, pipeline, select_command,
            },
        },
        types::{struct_type::Struct, INT},
    };
    use pretty_assertions::assert_eq;
    use rstest::rstest;
    use std::sync::Arc;

    #[rstest]
    // Case 1: No LET or DROP commands - passes through unchanged
    #[case::no_fusion_needed(
        pipeline()
            .command(select_command().named_field("a", 1).named_field("b", 2).build())
            .build(),
        pipeline()
            .command(select_command().named_field("a", 1).named_field("b", 2).build())
            .build(),
        Struct::default().with_str("a", INT).with_str("b", INT)
    )]
    // Case 2: SELECT + LET → fused to single SELECT (both projections merged)
    #[case::select_let_fused(
        pipeline()
            .command(select_command().named_field("a", 1).build())
            .command(let_command().named_field("b", 2).build())
            .build(),
        pipeline()
            .command(select_command().named_field("b", 2).named_field("a", 1).build())
            .build(),
        Struct::default().with_str("b", INT).with_str("a", INT)
    )]
    // Case 3: SELECT + DROP → fused to single SELECT (field removed)
    #[case::select_drop_fused(
        pipeline()
            .command(select_command().named_field("a", 1).named_field("b", 2).build())
            .command(drop_command().field("b").build())
            .build(),
        pipeline()
            .command(select_command().named_field("a", 1).build())
            .build(),
        Struct::default().with_str("a", INT)
    )]
    // Case 4: SELECT + multiple LETs → single SELECT
    #[case::select_multiple_lets_fused(
        pipeline()
            .command(select_command().named_field("a", 1).build())
            .command(let_command().named_field("b", 2).build())
            .command(let_command().named_field("c", 3).build())
            .build(),
        pipeline()
            .command(select_command()
                .named_field("c", 3)
                .named_field("b", 2)
                .named_field("a", 1)
                .build())
            .build(),
        Struct::default().with_str("c", INT).with_str("b", INT).with_str("a", INT)
    )]
    // Case 5: LET references field assigned by SELECT → barrier
    // SELECT a = 1 | LET b = a + 1
    // The LET references 'a' which SELECT assigned, so they cannot fuse.
    #[case::barrier_let_refs_select_field(
        pipeline()
            .command(select_command().named_field("a", 1).build())
            .command(let_command().named_field("b", add(column_ref("a"), 1)).build())
            .build(),
        pipeline()
            .command(select_command().named_field("a", 1).build())
            .command(select_command()
                .named_field("b", add(column_ref("a"), 1))
                .named_field("a", column_ref("a"))
                .build())
            .build(),
        Struct::default().with_str("b", INT).with_str("a", INT)
    )]
    // Case 6: LET references field assigned by preceding LET → barrier
    // SELECT a = 1 | LET b = 2 | LET c = b + 1
    // First LET doesn't reference 'a', so it fuses with SELECT.
    // Second LET references 'b' which first LET assigned, so barrier.
    #[case::barrier_let_refs_let_field(
        pipeline()
            .command(select_command().named_field("a", 1).build())
            .command(let_command().named_field("b", 2).build())
            .command(let_command().named_field("c", add(column_ref("b"), 1)).build())
            .build(),
        pipeline()
            .command(select_command()
                .named_field("b", 2)
                .named_field("a", 1)
                .build())
            .command(select_command()
                .named_field("c", add(column_ref("b"), 1))
                .named_field("b", column_ref("b"))
                .named_field("a", column_ref("a"))
                .build())
            .build(),
        Struct::default().with_str("c", INT).with_str("b", INT).with_str("a", INT)
    )]
    // Case 7: Chained LET dependencies → multiple barriers
    // SELECT a = 1 | LET b = a + 1 | LET c = b + 1
    // Each LET references the field assigned by the previous command.
    #[case::barrier_chained_dependencies(
        pipeline()
            .command(select_command().named_field("a", 1).build())
            .command(let_command().named_field("b", add(column_ref("a"), 1)).build())
            .command(let_command().named_field("c", add(column_ref("b"), 1)).build())
            .build(),
        pipeline()
            .command(select_command().named_field("a", 1).build())
            .command(select_command()
                .named_field("b", add(column_ref("a"), 1))
                .named_field("a", column_ref("a"))
                .build())
            .command(select_command()
                .named_field("c", add(column_ref("b"), 1))
                .named_field("b", column_ref("b"))
                .named_field("a", column_ref("a"))
                .build())
            .build(),
        Struct::default().with_str("c", INT).with_str("b", INT).with_str("a", INT)
    )]
    // Case 8: LET overwrites same field without reference → fuses (last write wins)
    // LET a = 1 | LET a = 2
    // Second LET doesn't reference 'a', it just overwrites it. No barrier needed.
    #[case::no_barrier_overwrite_without_ref(
        pipeline()
            .command(let_command().named_field("a", 1).build())
            .command(let_command().named_field("a", 2).build())
            .build(),
        pipeline()
            // Second LET's assignment wins (last write)
            .command(select_command().named_field("a", 2).build())
            .build(),
        Struct::default().with_str("a", INT)
    )]
    // Case 9: LET self-reference → barrier
    // SELECT a = 1 | LET a = a + 1
    // The second LET references 'a' which SELECT assigned.
    #[case::barrier_self_reference(
        pipeline()
            .command(select_command().named_field("a", 1).build())
            .command(let_command().named_field("a", add(column_ref("a"), 1)).build())
            .build(),
        pipeline()
            .command(select_command().named_field("a", 1).build())
            .command(select_command().named_field("a", add(column_ref("a"), 1)).build())
            .build(),
        Struct::default().with_str("a", INT)
    )]
    // Case 10: Independent LETs fuse (no dependency)
    // SELECT a = 1 | LET b = 2 | LET c = 3
    // None of the LETs reference assigned fields, so all fuse.
    #[case::no_barrier_independent_lets(
        pipeline()
            .command(select_command().named_field("a", 1).build())
            .command(let_command().named_field("b", 2).build())
            .command(let_command().named_field("c", 3).build())
            .build(),
        pipeline()
            .command(select_command()
                .named_field("c", 3)
                .named_field("b", 2)
                .named_field("a", 1)
                .build())
            .build(),
        Struct::default().with_str("c", INT).with_str("b", INT).with_str("a", INT)
    )]
    // Case 11: Three LETs without SELECT - verifies from_let prepend order
    // LET a = 1 | LET b = 2 | LET c = 3
    // First LET uses from_let, subsequent LETs use merge_let.
    // Each successive LET prepends, so final order is c, b, a.
    #[case::three_lets_prepend_order(
        pipeline()
            .command(let_command().named_field("a", 1).build())
            .command(let_command().named_field("b", 2).build())
            .command(let_command().named_field("c", 3).build())
            .build(),
        pipeline()
            .command(select_command()
                .named_field("c", 3)
                .named_field("b", 2)
                .named_field("a", 1)
                .build())
            .build(),
        Struct::default().with_str("c", INT).with_str("b", INT).with_str("a", INT)
    )]
    // Case 12: Compound LET assignments are preserved (struct packing happens in IR)
    // LET x.a = 1 | LET x.b = 2 fuses to SELECT x.b = 2, x.a = 1
    #[case::compound_let_preserved(
        pipeline()
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .build())
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    2,
                )
                .build())
            .build(),
        pipeline()
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    2,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .build())
            .build(),
        Struct::default()
            .with_str("x", Struct::default().with_str("b", INT).with_str("a", INT).into())
    )]
    // Case 13: Compound LET after EXPLODE barrier with nested struct in input schema
    // This is the exact scenario from the EXPLODE normalization bug:
    // After EXPLODE, input schema has {temp: Int, data: Struct{arr: Array<Int>}}
    // Then LET data.arr = temp assigns a compound identifier.
    // Bug: from_let was adding passthrough for `data` (simple) from flatten() output,
    // which conflicts with the compound assignment `data.arr`.
    // Fix: from_let must skip passthroughs that overlap with compound assignments.
    #[case::compound_let_after_explode_barrier(
        pipeline()
            // Setup: create nested struct and temp field
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("data".into(), "arr".into(), vec![]),
                    array().element(1).element(2).element(3),
                )
                .named_field("temp", 42)
                .build())
            // EXPLODE is a barrier
            .command(hamelin_lib::tree::builder::explode_command()
                .named_field("temp", column_ref("temp"))
                .build())
            // After barrier: LET with compound identifier into existing struct
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("data".into(), "arr".into(), vec![]),
                    column_ref("temp"),
                )
                .build())
            .build(),
        pipeline()
            // First SELECT (before barrier)
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("data".into(), "arr".into(), vec![]),
                    array().element(1).element(2).element(3),
                )
                .named_field("temp", 42)
                .build())
            // EXPLODE barrier preserved
            .command(hamelin_lib::tree::builder::explode_command()
                .named_field("temp", column_ref("temp"))
                .build())
            // Second SELECT: compound assignment + temp passthrough, but NO `data` passthrough
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("data".into(), "arr".into(), vec![]),
                    column_ref("temp"),
                )
                .named_field("temp", column_ref("temp"))
                .build())
            .build(),
        Struct::default()
            .with_str("data", Struct::default().with_str("arr", INT).into())
            .with_str("temp", INT)
    )]
    // Case 14: Compound LET adding sibling field - must preserve existing siblings
    // LET x.a = 1, x.b = 2 | WHERE true | LET x.y = 3
    // After barrier, input schema has {x: Struct{a: Int, b: Int}}.
    // The LET assigns x.y, which should ADD to the struct, not replace it.
    // Expected output schema: {x: Struct{y: Int, a: Int, b: Int}}
    // BUG: Current fix skips passthrough for `x` entirely, losing x.a and x.b.
    #[case::compound_let_sibling_preserves_existing(
        pipeline()
            // Setup: create struct with two fields
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    2,
                )
                .build())
            // WHERE is a barrier
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // After barrier: LET adds a new sibling field to existing struct
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "y".into(), vec![]),
                    3,
                )
                .build())
            .build(),
        pipeline()
            // First SELECT (before barrier)
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    2,
                )
                .build())
            // WHERE barrier preserved
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // Second SELECT: new field + passthroughs for existing siblings
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "y".into(), vec![]),
                    3,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    hamelin_lib::tree::builder::field(column_ref("x"), "a"),
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    hamelin_lib::tree::builder::field(column_ref("x"), "b"),
                )
                .build())
            .build(),
        Struct::default()
            .with_str("x", Struct::default().with_str("y", INT).with_str("a", INT).with_str("b", INT).into())
    )]
    // Case 15: Deep sibling preservation
    // LET x.c.d = 4, x.c.e = 5 | WHERE true | LET x.c.d = 6
    // Expect x.c.e preserved, x.c.d overwritten.
    #[case::deep_sibling_preservation(
        pipeline()
            // Setup: create deeply nested struct with two fields
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["d".into()]),
                    4,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["e".into()]),
                    5,
                )
                .build())
            // WHERE is a barrier
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // After barrier: LET overwrites one deep field
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["d".into()]),
                    6,
                )
                .build())
            .build(),
        pipeline()
            // First SELECT (before barrier)
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["d".into()]),
                    4,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["e".into()]),
                    5,
                )
                .build())
            // WHERE barrier preserved
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // Second SELECT: overwritten field + passthrough for sibling
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["d".into()]),
                    6,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["e".into()]),
                    hamelin_lib::tree::builder::field(
                        hamelin_lib::tree::builder::field(column_ref("x"), "c"),
                        "e"
                    ),
                )
                .build())
            .build(),
        Struct::default()
            .with_str("x", Struct::default()
                .with_str("c", Struct::default()
                    .with_str("d", INT)
                    .with_str("e", INT)
                    .into())
                .into())
    )]
    // Case 16: Parent vs child conflict - parent first, then child assignment
    // LET x = {a: 1, b: 2} | WHERE true | LET x.a = 3
    // Expect x.a overwritten, x.b preserved via leaf passthrough.
    // Note: struct literal `{a:1, b:2}` is desugared to `x.a=1, x.b=2` before fuse_projections runs.
    #[case::parent_then_child_assignment(
        pipeline()
            // Setup: create struct via struct literal (will be desugared)
            .command(let_command()
                .named_field(
                    "x",
                    hamelin_lib::tree::builder::struct_literal()
                        .field("a", 1)
                        .field("b", 2),
                )
                .build())
            // WHERE is a barrier
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // After barrier: LET assigns a child field
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    3,
                )
                .build())
            .build(),
        pipeline()
            // First SELECT: struct literal desugared to compound assignments
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    2,
                )
                .build())
            // WHERE barrier preserved
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // Second SELECT: child assignment + leaf passthrough for sibling
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    3,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    hamelin_lib::tree::builder::field(column_ref("x"), "b"),
                )
                .build())
            .build(),
        Struct::default()
            .with_str("x", Struct::default().with_str("a", INT).with_str("b", INT).into())
    )]
    // Case 17: Child vs parent conflict - child first, then parent assignment
    // LET x.a = 1 | WHERE true | LET x = {y: 2}
    // Expected: parent struct literal should REPLACE x entirely, not merge.
    // BUG: struct literal desugaring doesn't implement replace semantics yet.
    // This test documents the CORRECT expected behavior. It will fail until
    // struct literal desugaring is fixed to properly replace parent bindings.
    #[case::child_then_parent_assignment(
        pipeline()
            // Setup: create child field
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .build())
            // WHERE is a barrier
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // After barrier: LET assigns struct literal - should REPLACE x entirely
            .command(let_command()
                .named_field(
                    "x",
                    hamelin_lib::tree::builder::struct_literal().field("y", 2),
                )
                .build())
            .build(),
        pipeline()
            // First SELECT (before barrier)
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .build())
            // WHERE barrier preserved
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // Second SELECT: struct literal replaces x, so only x.y exists
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "y".into(), vec![]),
                    2,
                )
                .build())
            .build(),
        Struct::default()
            .with_str("x", Struct::default().with_str("y", INT).into())
    )]
    // Case 18: Mixed deep + new field (recursive example)
    // Input schema x: {a, b, c: {d, e}}, LET assigns x.y = 3, x.c.d = 4
    // Expect passthroughs for x.a, x.b, x.c.e only.
    #[case::mixed_deep_and_new_field(
        pipeline()
            // Setup: create struct with nested structure
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    2,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["d".into()]),
                    10,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["e".into()]),
                    20,
                )
                .build())
            // WHERE is a barrier
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // After barrier: LET assigns new sibling + overwrites deep field
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "y".into(), vec![]),
                    3,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["d".into()]),
                    4,
                )
                .build())
            .build(),
        pipeline()
            // First SELECT (before barrier)
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    2,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["d".into()]),
                    10,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["e".into()]),
                    20,
                )
                .build())
            // WHERE barrier preserved
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // Second SELECT: assignments first, then passthroughs in schema order
            // Schema order after LET: x.y, x.c.d (from LET), then x.c.e, x.a, x.b (preserved)
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "y".into(), vec![]),
                    3,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["d".into()]),
                    4,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec!["e".into()]),
                    hamelin_lib::tree::builder::field(
                        hamelin_lib::tree::builder::field(column_ref("x"), "c"),
                        "e"
                    ),
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    hamelin_lib::tree::builder::field(column_ref("x"), "a"),
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    hamelin_lib::tree::builder::field(column_ref("x"), "b"),
                )
                .build())
            .build(),
        Struct::default()
            .with_str("x", Struct::default()
                .with_str("y", INT)
                .with_str("c", Struct::default().with_str("d", INT).with_str("e", INT).into())
                .with_str("a", INT)
                .with_str("b", INT)
                .into())
    )]
    // Case 19: DROP + LET within same fused block
    // LET x.a = 1, x.b = 2, x.c = 3 | WHERE true | DROP x.b | LET x.d = 4
    // Expect x.a, x.c, x.d present; x.b dropped.
    #[case::drop_and_let_fused(
        pipeline()
            // Setup: create struct with three fields
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    2,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec![]),
                    3,
                )
                .build())
            // WHERE is a barrier
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // After barrier: DROP one field, then LET adds another
            .command(drop_command()
                .field(CompoundIdentifier::new("x".into(), "b".into(), vec![]))
                .build())
            .command(let_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "d".into(), vec![]),
                    4,
                )
                .build())
            .build(),
        pipeline()
            // First SELECT (before barrier)
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    2,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec![]),
                    3,
                )
                .build())
            // WHERE barrier preserved
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // Second SELECT: new field + passthroughs (excluding dropped x.b)
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "d".into(), vec![]),
                    4,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    hamelin_lib::tree::builder::field(column_ref("x"), "a"),
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec![]),
                    hamelin_lib::tree::builder::field(column_ref("x"), "c"),
                )
                .build())
            .build(),
        Struct::default()
            .with_str("x", Struct::default()
                .with_str("d", INT)
                .with_str("a", INT)
                .with_str("c", INT)
                .into())
    )]
    // Case 20: DROP child from direct struct binding
    // LET x = {a: 1, b: 2, c: 3} | WHERE true | DROP x.a
    // Expect x.b, x.c preserved; x.a dropped.
    #[case::drop_child_from_struct_binding(
        pipeline()
            // Setup: create struct via struct literal (will be desugared)
            .command(let_command()
                .named_field(
                    "x",
                    hamelin_lib::tree::builder::struct_literal()
                        .field("a", 1)
                        .field("b", 2)
                        .field("c", 3),
                )
                .build())
            // WHERE is a barrier
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // After barrier: DROP a child field
            .command(drop_command()
                .field(CompoundIdentifier::new("x".into(), "a".into(), vec![]))
                .build())
            .build(),
        pipeline()
            // First SELECT: struct literal desugared to compound assignments
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "a".into(), vec![]),
                    1,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    2,
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec![]),
                    3,
                )
                .build())
            // WHERE barrier preserved
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // Second SELECT: passthroughs for remaining children (x.a dropped)
            .command(select_command()
                .named_field(
                    CompoundIdentifier::new("x".into(), "b".into(), vec![]),
                    hamelin_lib::tree::builder::field(column_ref("x"), "b"),
                )
                .named_field(
                    CompoundIdentifier::new("x".into(), "c".into(), vec![]),
                    hamelin_lib::tree::builder::field(column_ref("x"), "c"),
                )
                .build())
            .build(),
        Struct::default()
            .with_str("x", Struct::default().with_str("b", INT).with_str("c", INT).into())
    )]
    // Case 21: No overlap sanity - unrelated fields passthrough unchanged
    // LET a = 1 | WHERE true | LET b = 2
    // Expect both fields present, no interference.
    #[case::no_overlap_unrelated_fields(
        pipeline()
            // Setup: create one field
            .command(let_command()
                .named_field("a", 1)
                .build())
            // WHERE is a barrier
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // After barrier: LET adds unrelated field
            .command(let_command()
                .named_field("b", 2)
                .build())
            .build(),
        pipeline()
            // First SELECT (before barrier)
            .command(select_command()
                .named_field("a", 1)
                .build())
            // WHERE barrier preserved
            .command(hamelin_lib::tree::builder::where_command(true).build())
            // Second SELECT: new field + passthrough for existing
            .command(select_command()
                .named_field("b", 2)
                .named_field("a", column_ref("a"))
                .build())
            .build(),
        Struct::default().with_str("b", INT).with_str("a", INT)
    )]
    fn test_fuse_projections(
        #[case] input: Pipeline,
        #[case] expected: Pipeline,
        #[case] expected_output_schema: Struct,
    ) {
        let input_typed = input.typed_with().typed();
        let expected_typed = expected.typed_with().typed();

        let mut ctx = StatementTranslationContext::default();
        let result = fuse_projections(Arc::new(input_typed), &mut ctx).unwrap();

        // Compare ASTs
        assert_eq!(result.ast, expected_typed.ast);

        // Verify output schema
        let result_schema = result.environment().flatten();
        assert_eq!(result_schema, expected_output_schema);
    }
}