glowdust 0.0.1

use crate::compiler::parser::ast::{
    Atom, AtomKind, Expr, ExprKind, MapTarget, NamedParameter, Stmt,
};
use crate::compiler::parser::ast_walker::{
    validate_variable_initialization, validate_variable_initialization_in_stmt, WalkContext,
};
use crate::compiler::parser::pointer::P;
use crate::compiler::parser::{ast_walker, CompileError};
use crate::compiler::query_solver::CursorFrame::{Call, Pull};
use crate::compiler::value::values::Value;
use crate::compiler::value::values::Value::FunctionNameValue;
use crate::runtime::bytecode::OpCode::{
    OpCall, OpConst, OpEquals, OpGetLocal, OpJump, OpJumpIfFalse, OpLoop, OpOpenCursor, OpPop,
    OpPull, OpSetLocal,
};
use crate::runtime::compiled_script::Bytecode;
use crate::store::Store;
use std::collections::HashSet;

#[derive(Debug)]
enum CompiledCursor {
    Produce {
        name: String,
        name_const: u8,
        produce_params: Vec<NamedParameter>,
        map_params: MapTarget,
    },
    Call {
        name: String,
        name_const: u8,
        call_params: Vec<NamedParameter>,
        map_params: MapTarget,
    },
    Block {
        code: Vec<Stmt>,
    },
}

#[derive(Debug)]
struct ValueGenerator {
    cursors: Vec<CompiledCursor>,
}

impl ValueGenerator {
    fn new() -> Self {
        ValueGenerator { cursors: vec![] }
    }
}

pub(crate) fn solve_query(
    query: &Vec<P<Atom>>,
    walk_context: &mut WalkContext,
) -> Result<Bytecode, CompileError> {
    let mut unbound_vars = HashSet::new();
    let solved = cut_into_pipes(query, walk_context, &mut unbound_vars)?;
    compile(
        solved,
        walk_context,
        &mut walk_context.initialized_variables.clone(),
    )
}

fn cut_into_pipes(
    parsed_query: &Vec<P<Atom>>,
    walk_context: &mut WalkContext,
    unbound: &mut HashSet<u8>,
) -> Result<Vec<ValueGenerator>, CompileError> {
    let mut result = vec![];
    let mut current_generator = ValueGenerator::new();

    for atom in parsed_query {
        match &atom.kind {
            AtomKind::FunCall(fun_name, params) => {
                let function_name = check_domain_arity(
                    walk_context.constants,
                    *fun_name,
                    params,
                    walk_context.store,
                )?;

                // First check codomain arity and synthesize the LHS discard, since it's common to both Produce and Map
                let pops = if let Some(codomain_arity) =
                    walk_context.store.codomain_arity(&function_name)
                {
                    vec![
                        P::P(Expr {
                            kind: ExprKind::ValueDiscard,
                            span: atom.span
                        });
                        codomain_arity as usize
                    ]
                } else {
                    return Err(CompileError::new(
                        format!(
                            "Store could not find codomain arity for function {}",
                            function_name
                        ),
                        atom.span,
                    ));
                };
                if function_call_has_unbound_arguments(params, unbound)? {
                    if !walk_context.store.has_values(&function_name) {
                        return Err(CompileError::new(format!(
                        "Function {} has unbound variables but there are no values to generate from it",
                        function_name
                    ), atom.span));
                    }
                    let produce = CompiledCursor::Produce {
                        name: function_name.clone(),
                        name_const: *fun_name,
                        produce_params: params.clone(),
                        map_params: MapTarget::Tuple(pops),
                    };
                    // This will _not_ trigger only on the first Produce ever - instead it will
                    // ignore it and create a new one. Every subsequent pass will always find
                    // at least the previous Produce so it will always add the previous generator
                    // to the result and then create a new one and add itself as the first element.
                    if !current_generator.cursors.is_empty() {
                        result.push(current_generator);
                    }
                    current_generator = ValueGenerator::new();
                    current_generator.cursors.push(produce);
                } else {
                    // For now this is safe. Once global variables are accessible in match clauses,
                    // it will definitely be possible to have a Call element as the first in a generator.
                    // Then this will cause breakage
                    let call = CompiledCursor::Call {
                        name: function_name.clone(),
                        name_const: *fun_name,
                        call_params: params.clone(),
                        map_params: MapTarget::Tuple(pops),
                    };
                    current_generator.cursors.push(call);
                }
            }
            AtomKind::Map(fun_name, params, bind) => {
                let function_name = check_domain_arity(
                    walk_context.constants,
                    *fun_name,
                    params,
                    walk_context.store,
                )?;
                let cursor;
                if function_call_has_unbound_arguments(params, unbound)? {
                    cursor = CompiledCursor::Produce {
                        name: function_name.clone(),
                        name_const: *fun_name,
                        produce_params: params.clone(),
                        map_params: bind.clone(),
                    };
                    // This will _not_ trigger only on the first Produce ever - instead it will
                    // ignore it and create a new one. Every subsequent pass will always find
                    // at least the previous Produce so it will always add the previous generator
                    // to the result and then create a new one and add itself as the first element.
                    if !current_generator.cursors.is_empty() {
                        result.push(current_generator);
                    }
                    current_generator = ValueGenerator::new();
                } else {
                    // For now this is safe. Once global variables are accessible in match clauses,
                    // it will definitely be possible to have a Call element as the first in a generator.
                    // Then this will cause breakage
                    cursor = CompiledCursor::Call {
                        name: function_name.clone(),
                        name_const: *fun_name,
                        call_params: params.clone(),
                        map_params: bind.clone(),
                    };
                }
                if let MapTarget::Tuple(exprs) = bind {
                    if let Some(codomain_arity) = walk_context.store.codomain_arity(&function_name)
                    {
                        if exprs.len() != codomain_arity as usize {
                            return Err(CompileError::new(format!(
                                "Function {} has codomain arity {} which doesn't match the query argument count {}",
                                function_name, codomain_arity, exprs.len()), atom.span));
                        }
                        bind_has_unbound_vars(exprs, unbound)?;
                    } else {
                        return Err(CompileError::new(
                            format!(
                                "2. Store could not find codomain arity for function/map {}",
                                function_name
                            ),
                            atom.span,
                        ));
                    }
                    current_generator.cursors.push(cursor);
                } else {
                    panic!()
                }
            }
            AtomKind::Block(block) => {
                // treat the block like normal code, except the bound variables are those that
                // have been bound so far
                let mut context = WalkContext {
                    is_local: true,
                    constants: &[],
                    initialized_variables: unbound.clone(),
                    store: walk_context.store,
                };
                for expr in &block.stmts {
                    validate_variable_initialization_in_stmt(&mut context, expr)?;
                }
                current_generator.cursors.push(CompiledCursor::Block {
                    code: block.stmts.clone(),
                })
            }
            AtomKind::Return(exprs) => {
                let mut context = WalkContext {
                    is_local: true,
                    constants: &[],
                    initialized_variables: unbound.clone(),
                    store: walk_context.store,
                };
                for expr in exprs {
                    validate_variable_initialization(&mut context, expr)?;
                }
                current_generator.cursors.push(CompiledCursor::Block {
                    code: vec![Stmt::Return(exprs.clone())],
                })
            }
        }
    }

    result.push(current_generator);
    walk_context.initialized_variables.extend(unbound.drain());
    Ok(result)
}

// TODO semantics of this function (and its sibling that follows) are not 100% clear. In theory
// this determines the unbound arguments and allows naked variables to be marked as unbound
// to let function calls be marked as producers.
// In practice it also detects and fails expressions if they contain unbound variables. This is
// how it should be, but i don't think this is the place for that to happen.
fn function_call_has_unbound_arguments(
    params: &[NamedParameter],
    unbound: &mut HashSet<u8>,
) -> Result<bool, CompileError> {
    let mut has_unbound = false;
    for param in params {
        match param.value.kind {
            ExprKind::Variable(id) => {
                if unbound.insert(id) {
                    has_unbound = true
                }
            }
            ExprKind::ValueDiscard => has_unbound = true,
            _ => {
                // I am not sure this is actually correct, but all tests pass so far
                let unbound_found = expression_has_unbound_vars(&param.value, unbound)?;
                if unbound_found {
                    return Err(CompileError::new(
                        "Expressions cannot contain unbound arguments".to_string(),
                        param.value.span,
                    ));
                }
            }
        }
    }
    Ok(has_unbound)
}

fn bind_has_unbound_vars(
    params: &[P<Expr>],
    unbound: &mut HashSet<u8>,
) -> Result<bool, CompileError> {
    let mut has_unbound = false;
    for param in params {
        match param.kind {
            ExprKind::Variable(id) => {
                if unbound.insert(id) {
                    has_unbound = true
                }
            }
            ExprKind::ValueDiscard => has_unbound = true,
            _ => {
                // I am not sure this is actually correct, but all tests pass so far
                let unbound_found = expression_has_unbound_vars(&param, unbound)?;
                if unbound_found {
                    return Err(CompileError::new(
                        "Expressions cannot contain unbound arguments: {}".to_string(),
                        param.span,
                    ));
                }
            }
        }
    }
    Ok(has_unbound)
}

fn expression_has_unbound_vars(arg: &P<Expr>, unbound: &HashSet<u8>) -> Result<bool, CompileError> {
    match &arg.kind {
        ExprKind::Tuple(contents) => {
            for content in contents {
                if expression_has_unbound_vars(content, unbound)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        ExprKind::Unary(_, expr) => expression_has_unbound_vars(expr, unbound),
        ExprKind::Binary(_, lhs, rhs) => Ok(expression_has_unbound_vars(lhs, unbound)?
            || expression_has_unbound_vars(rhs, unbound)?),
        ExprKind::Variable(id) => Ok(!unbound.contains(id)),
        ExprKind::Literal(_) => Ok(false),
        ExprKind::FunCall(_, params) => {
            for param in params {
                if expression_has_unbound_vars(&param.value, unbound)? {
                    return Ok(true);
                }
            }
            Ok(false)
        }
        ExprKind::Assignment(lhs, rhs) => Ok(expression_has_unbound_vars(lhs, unbound)?
            || expression_has_unbound_vars(rhs, unbound)?),
        ExprKind::FieldAccess(lhs, _) => Ok(expression_has_unbound_vars(lhs, unbound)?),
        ExprKind::ValueDiscard => Ok(false),
    }
}

fn check_domain_arity(
    constants: &[Value],
    fun_name: u8,
    params: &[NamedParameter],
    store: &dyn Store,
) -> Result<String, String> {
    let FunctionNameValue(function_name) = &(constants[fun_name as usize].clone()) else {
        return Err("Expected function name constant to be a FunctionNameValue".to_string());
    };
    if let Some(domain_arity) = store.domain_arity(function_name) {
        if params.len() != domain_arity as usize {
            return Err(format!(
                "Function {} has domain arity {} which doesn't match the query argument count {}",
                function_name,
                domain_arity,
                params.len()
            ));
        }
    } else {
        return Err(format!(
            "Store could not find domain arity for function {}",
            function_name
        ));
    }
    Ok(function_name.clone())
}

#[derive(Default)]
struct PullFrame {
    // the cursor id for this cursor. incremented externally, used for the OpCloseCursor in the epilogue
    cursor_id: u8,
    // this is the location where the pull opcode is. That's the target for the loop in the epilogue
    pull_position: usize,
    // this is the location of the exit jump. this will be patched when the epilogue is generated
    exit_jump_patch_position: usize,
}

#[derive(Default)]
struct CallFrame {
    // the cursor id for this cursor. incremented externally, used for the OpCloseCursor in the epilogue
    cursor_id: u8,
    // this is the location of the exit jump. this will be patched when the epilogue is generated
    exit_jump_patch_position: usize,
}

enum CursorFrame {
    Pull(PullFrame),
    Call(CallFrame),
}

// TODO The direct bytecode OpPops that follow are a symptom of improper frame stack setup for query atoms. If there
// were proper stack frames, then on false we could just OpReturn and all the variables would disappear without
// having to count. This will have to do for now, it's probably the next thing to fix.

// The following 3 methods basically implement the "jump to next generated value if false" functionality. They
// should be called at the end of query atoms
// These will either loop back to the top of the loop if a generator cursor is running, or to the
// end of the query if a call cursor is running.
// I'm pretty sure that this distinction should go away at some point, but the common ground between them is
// not clear as of this writing.

fn jump_to_pull_on_false(
    bytecode: &mut Bytecode,
    pull_position: usize,
    pops: usize,
    from_call: bool,
) {
    // if the top of the stack is false jump over two bytecode bytes (over the pop/jump and to the loop)
    bytecode.byte(OpJumpIfFalse as u8);
    if from_call {
        // if this is a call, we need to leave the value on the stack if it is true-ish
        bytecode.byte(2_u8);
    } else {
        // otherwise we need to pop it
        bytecode.byte(3_u8);
        bytecode.byte(OpPop as u8);
    }
    // and jump over the pops loop back to the rest of the cursor atom
    bytecode.byte(OpJump as u8);
    bytecode.byte(3_u8 + pops as u8);
    // Otherwise, pop whatever has been pushed
    for _ in 0..pops + 1 {
        // + 1 because we need to pop the false value that caused the OpJumpIfFalse to jump here
        bytecode.byte(OpPop as u8);
    }
    bytecode.byte(OpLoop as u8);
    bytecode.byte((bytecode.bytes.len() - pull_position + 1) as u8);
}

fn optionally_jump_to_pull_on_false(
    bytecode: &mut Bytecode,
    cursor_stack: &[CursorFrame],
    pops: usize,
) {
    if let Some(Pull(pull_frame)) = cursor_stack.last() {
        jump_to_pull_on_false(bytecode, pull_frame.pull_position, pops, false);
    }
}

fn jump_to_pull_or_return_on_false(
    result: &mut Bytecode,
    cursor_stack: &mut Vec<CursorFrame>,
    from_call: bool,
) {
    // If we are in a Pull cursor chain jump to next pull if false...
    if let Some(Pull(pull_frame)) = cursor_stack.last() {
        jump_to_pull_on_false(result, pull_frame.pull_position, 0, from_call);
    } else {
        // ...otherwise jump to return if false (even if there is another CallFrame on the stack)
        let mut current_frame = CallFrame::default();
        result.byte(OpJumpIfFalse as u8);
        current_frame.exit_jump_patch_position = result.byte(0);
        cursor_stack.push(Call(current_frame));
    }
}

fn compile(
    solved_query: Vec<ValueGenerator>,
    walk_context: &mut WalkContext,
    unbound: &mut HashSet<u8>,
) -> Result<Bytecode, CompileError> {
    let mut result = Bytecode::new();
    let mut cursor_stack: Vec<CursorFrame> = vec![];

    for generator in &solved_query {
        for element in &generator.cursors {
            match element {
                CompiledCursor::Produce {
                    name,
                    name_const,
                    produce_params,
                    map_params,
                } => {
                    let mut current_frame = PullFrame {
                        cursor_id: cursor_stack.len() as u8,
                        ..PullFrame::default()
                    };
                    assert!(
                        walk_context.store.has_values(name),
                        "Store claims no values for function {}",
                        name
                    );

                    // function name
                    result.byte(OpConst as u8);
                    result.byte(*name_const);

                    // create cursor
                    result.byte(OpOpenCursor as u8);
                    result.byte(current_frame.cursor_id);

                    // loop start - pull value
                    current_frame.pull_position = result.byte(OpPull as u8);
                    result.byte(current_frame.cursor_id);

                    result.byte(OpJumpIfFalse as u8);
                    current_frame.exit_jump_patch_position = result.byte(0);
                    /*
                     * Domain and codomain values are on the stack now. We need to bind/join them
                     *
                     * The pop count is kind of awkward, but necessary. Bound, undiscarded parameters
                     * will leave a bool on the stack, since they are making a comparison/join. For each
                     * one of those, we'll need to OpPop, so we need to keep count. If, however, a variable
                     * is unbound or discarded, no boolean remains, so we need one less OpPop. This
                     * is checked by bind_and_join_var() and we just update the pops count variable.
                     */
                    let mut pops = produce_params.len();
                    // TODO There is a lot of complexity in this (and in the identical piece in Map below),
                    // TODO complexity that doesn't seem required. This can more than likely be streamlined.
                    for named_param in produce_params {
                        let param = &named_param.value;
                        if bind_and_join_var(&param, unbound, &mut result, walk_context)? {
                            pops -= 1;
                            continue;
                        }
                        jump_to_pull_on_false(
                            &mut result,
                            current_frame.pull_position,
                            pops,
                            false,
                        );
                    }
                    if let MapTarget::Tuple(map) = map_params {
                        pops = map.len();
                        for param in map {
                            if bind_and_join_var(&param, unbound, &mut result, walk_context)? {
                                pops -= 1;
                                continue;
                            }
                            pops -= 1; // TODO check this if it is really needed
                            jump_to_pull_on_false(
                                &mut result,
                                current_frame.pull_position,
                                pops,
                                false,
                            );
                        }
                    } else {
                        panic!()
                    }
                    cursor_stack.push(Pull(current_frame));
                }
                CompiledCursor::Call {
                    name: _,
                    name_const,
                    call_params,
                    map_params,
                } => {
                    // push arguments
                    for param in call_params {
                        ast_walker::generate_bytecode(walk_context, &mut result, &param.value)?;
                    }

                    // push function name
                    result.byte(OpConst as u8);
                    result.byte(*name_const);

                    // do the call
                    result.byte(OpCall as u8);
                    result.byte(call_params.len() as u8);

                    jump_to_pull_or_return_on_false(&mut result, &mut cursor_stack, true);

                    if let MapTarget::Tuple(map) = map_params {
                        let mut pops = map.len();
                        for param in map {
                            if bind_and_join_var(&param, unbound, &mut result, walk_context)? {
                                pops -= 1;
                                continue;
                            }
                            pops -= 1;
                            optionally_jump_to_pull_on_false(&mut result, &cursor_stack, pops);
                        }
                    } else {
                        panic!()
                    }
                }
                CompiledCursor::Block { code } => {
                    let mut last_expr_was_generate = false;
                    for stmt in code {
                        ast_walker::generate_bytecode_from_statement(
                            walk_context,
                            &mut result,
                            stmt,
                        )?;
                        match stmt {
                            Stmt::Return(_) => last_expr_was_generate = true,
                            _ => last_expr_was_generate = false,
                        }
                    }
                    // This is wrong, this will skip jump checks only on `return` expressions
                    // INCLUDING blocks where the last expression is return.
                    // Each function call and expression in a block can leave a value on
                    // the stack and this must be dealt with.
                    // The check must be much broader than that, but at least return is valid
                    if !last_expr_was_generate {
                        jump_to_pull_or_return_on_false(&mut result, &mut cursor_stack, false);
                    }
                }
            }
        }
    }

    while let Some(frame) = cursor_stack.pop() {
        match frame {
            Pull(pull_frame) => {
                result.byte(OpLoop as u8);
                result.byte((result.bytes.len() - pull_frame.pull_position + 1) as u8);
                // result.bytes.len() is where the OpReturn _will_ be added right after this if block.
                result.bytes[pull_frame.exit_jump_patch_position] =
                    (result.bytes.len() - pull_frame.exit_jump_patch_position - 1) as u8;
                // This OpPop is needed to pop off the empty tuple produced by the last produce that
                // returned empty and stops the outermost iteration
                result.byte(OpPop as u8);
            }
            Call(call_frame) => {
                // result.bytes.len() is where the OpReturn _will_ be added right after this if block.
                result.bytes[call_frame.exit_jump_patch_position] =
                    (result.bytes.len() - call_frame.exit_jump_patch_position - 1) as u8;
            }
        }
    }
    Ok(result)
}

/*
 * This method generates the bytecode for a variable position in an atom. It either sets a variable
 * value or it computes a value and joins the variable on that value. In more detail:
 *
 * 1. If this is a variable, it checks if it is unbound (i.e. seen for the first time
 *  1a. If it is unbound, then it needs to be set as bound (since it just had a value assigned),
 *      and have the value set.
 *  1b. If it is bound, then it must join to the value that it already has. To do that, it
 *      gets the variable value, performs an OpEquals.
 * 2. If this is a _ pattern, it needs to just discard the value, no OpEquals
 * 3. If this is anything else, then it needs to compute the expression and do an OpEquals
 *
 * Return value is Ok(true) if no OpEquals was emitted, Ok(false) otherwise.
 */
fn bind_and_join_var(
    param: &P<Expr>,
    unbound: &mut HashSet<u8>,
    mut result: &mut Bytecode,
    walk_context: &mut WalkContext,
) -> Result<bool, CompileError> {
    if let ExprKind::Variable(id) = param.kind {
        if unbound.remove(&id) {
            result.byte_and_line(OpSetLocal as u8, param.span.line);
            result.byte_and_line(id, param.span.line);
            result.byte(OpPop as u8);
            return Ok(true);
        } else {
            result.byte_and_line(OpGetLocal as u8, param.span.line);
            result.byte_and_line(id, param.span.line);
        }
    } else if let ExprKind::ValueDiscard = param.kind {
        result.byte(OpPop as u8);
        return Ok(true);
    } else {
        ast_walker::generate_bytecode(walk_context, &mut result, param)?;
    }
    result.byte(OpEquals as u8);
    Ok(false)
}

#[cfg(test)]
mod tests {
    use crate::compiler::parser::ast::ExprKind::Variable;
    use crate::compiler::parser::ast::{
        Atom, AtomKind, Expr, MapTarget, NamedParameter, DUMMY_SPAN,
    };
    use crate::compiler::parser::ast_walker::WalkContext;
    use crate::compiler::parser::pointer::P;
    use crate::compiler::query_solver::CompiledCursor::Call;
    use crate::compiler::query_solver::{cut_into_pipes, CompiledCursor};
    use crate::compiler::value::values::Value;
    use crate::compiler::value::values::Value::FunctionNameValue;
    use crate::store::test_mock::StoreHasAllValuesMock;
    use crate::string_to_value;
    use std::collections::{HashMap, HashSet};
    use std::io::Error;

    fn assert_produce(pipes: &[CompiledCursor], index: usize) {
        if let Some(CompiledCursor::Produce {
            name: _,
            name_const: _,
            produce_params: _,
            map_params: _,
        }) = pipes.get(index)
        {
            return;
        } else {
            panic!("")
        }
    }

    fn assert_call(pipes: &[CompiledCursor], index: usize) {
        if let Some(Call {
            name: _,
            name_const: _,
            call_params: _,
            map_params: _,
        }) = pipes.get(index)
        {
            return;
        } else {
            panic!("Expected Call, got {:?} instead", pipes.get(index));
        }
    }

    fn raw_to_string_values(raw: &[&str]) -> Vec<Value> {
        raw.iter()
            .map(|string| string_to_value!(string.to_string()))
            .collect()
    }

    fn raw_to_function_name_values(raw: &[&str]) -> Vec<Value> {
        raw.iter()
            .map(|string| FunctionNameValue(string.to_string()))
            .collect()
    }

    fn strings_to_constant_table<'a>(strings: &[&'a str], table: &mut HashMap<&'a str, u8>) {
        let base_size = table.len() as u8;
        for (index, string) in strings.iter().enumerate() {
            table.insert(*string, base_size + index as u8);
        }
    }

    fn funcall_base<'a>(
        fn_name: &str,
        params: &[&str],
        constant_table: &HashMap<&'a str, u8>,
    ) -> (u8, Vec<NamedParameter>) {
        let name_const = *constant_table.get(fn_name).unwrap();

        let mut args = vec![];
        for param in params {
            args.push(NamedParameter {
                name: None,
                value: P::P(Expr {
                    kind: Variable(*constant_table.get(param).unwrap()),
                    span: DUMMY_SPAN,
                }),
            })
        }
        (name_const, args)
    }

    fn funcall_atom<'a>(
        fn_name: &str,
        params: &[&str],
        constant_table: &HashMap<&'a str, u8>,
    ) -> AtomKind {
        let (name, params) = funcall_base(fn_name, params, constant_table);
        AtomKind::FunCall(name, params)
    }

    fn map_atom<'a>(
        fn_name: &str,
        params: &[&str],
        bind_var_names: &[&str],
        constant_table: &HashMap<&'a str, u8>,
    ) -> AtomKind {
        let (name, params) = funcall_base(fn_name, params, constant_table);

        let mut binds = vec![];

        for bind_var_name in bind_var_names {
            binds.push(P::P(Expr {
                kind: Variable(*constant_table.get(bind_var_name).unwrap()),
                span: DUMMY_SPAN,
            }))
        }

        AtomKind::Map(name, params, MapTarget::Tuple(binds))
    }

    #[test]
    fn check_joins_on_star_join() {
        /*
        match f(x1, x2) -> (y1, y2), g1(x1) -> (w1), g2(x2) -> (w2), g3(y1), g4(y2)
        */
        let variables_raw = vec!["x1", "x2", "y1", "y2", "w1", "w2"];
        let function_names_raw = vec!["f", "g1", "g2", "g3", "g4"];
        let mut constant_table = HashMap::new();

        strings_to_constant_table(&variables_raw, &mut constant_table);
        strings_to_constant_table(&function_names_raw, &mut constant_table);

        let mut variable_names = raw_to_string_values(&variables_raw);
        let mut function_names = raw_to_function_name_values(&function_names_raw);

        variable_names.append(&mut function_names);

        let mut match_stmts = vec![];

        let g3_tuple = P::P(Atom {
            kind: funcall_atom("g3", &vec!["y1"], &constant_table),
            span: DUMMY_SPAN,
        });

        let g4_tuple = P::P(Atom {
            kind: funcall_atom("g4", &vec!["y2"], &constant_table),
            span: DUMMY_SPAN,
        });

        let f_map = P::P(Atom {
            kind: map_atom("f", &vec!["x1", "x2"], &vec!["y1", "y2"], &constant_table),
            span: DUMMY_SPAN,
        });

        let g1_map = P::P(Atom {
            kind: map_atom("g1", &vec!["x1"], &vec!["w1"], &constant_table),
            // kind: AtomKind::Map(g1_tuple, g1_bind),
            span: DUMMY_SPAN,
        });
        let g2_map = P::P(Atom {
            kind: map_atom("g2", &vec!["x2"], &vec!["w2"], &constant_table),
            span: DUMMY_SPAN,
        });

        let mut arities = HashMap::new();
        arities.insert("f".to_string(), (2, 2));
        arities.insert("g1".to_string(), (1, 1));
        arities.insert("g2".to_string(), (1, 1));
        arities.insert("g3".to_string(), (1, 1));
        arities.insert("g4".to_string(), (1, 1));
        let mut store = StoreHasAllValuesMock { arities };

        match_stmts.push(f_map);
        match_stmts.push(g1_map);
        match_stmts.push(g2_map);
        match_stmts.push(g3_tuple);
        match_stmts.push(g4_tuple);

        let pipes = cut_into_pipes(
            &mut match_stmts,
            &mut WalkContext {
                is_local: true,
                constants: &variable_names,
                store: &mut store,
                initialized_variables: HashSet::new(),
            },
            &mut HashSet::new(),
        )
        .unwrap();

        /*
         solution is
        0. produce(f)(x1, x2) -> (y1, y2)
        1. call(g1)(x1) -> (w1)
        2. call(g2)(x2) -> (w2)
        3. call(g3)(y1)
        4. call(g4)(y2)
          */

        assert_eq!(1, pipes.len());
        let pipe = &pipes.get(0).unwrap().cursors;
        println!("{:?}", pipe);

        assert_eq!(5, pipe.len());
        assert_produce(&pipe, 0);
        assert_call(&pipe, 1);
        assert_call(&pipe, 2);
        assert_call(&pipe, 3);
        assert_call(&pipe, 4);
    }

    #[test]
    fn simple_join() -> Result<(), Error> {
        /*
        match f(x) -> (y), g(y) -> (w)
        */

        let variables_raw = vec!["x", "y", "w"];
        let functions_raw = vec!["f", "g"];
        let mut constant_table = HashMap::new();

        strings_to_constant_table(&variables_raw, &mut constant_table);
        strings_to_constant_table(&functions_raw, &mut constant_table);

        let mut variable_names = raw_to_string_values(&variables_raw);
        let mut function_names = raw_to_function_name_values(&functions_raw);

        variable_names.append(&mut function_names);
        let mut match_stmts = vec![];

        let f_map = P::P(Atom {
            kind: map_atom("f", &vec!["x"], &vec!["y"], &constant_table),
            // kind: AtomKind::Map(f_tuple, f_bind),
            span: DUMMY_SPAN,
        });

        let g_map = P::P(Atom {
            kind: map_atom("g", &vec!["y"], &vec!["w"], &constant_table),
            // kind: AtomKind::Map(g_tuple, g_bind),
            span: DUMMY_SPAN,
        });

        match_stmts.push(f_map);
        match_stmts.push(g_map);

        let mut arities = HashMap::new();
        arities.insert("f".to_string(), (1, 1));
        arities.insert("g".to_string(), (1, 1));
        let mut store = StoreHasAllValuesMock { arities };
        let pipes = cut_into_pipes(
            &mut match_stmts,
            &mut WalkContext {
                is_local: true,
                constants: &variable_names,
                store: &mut store,
                initialized_variables: HashSet::new(),
            },
            &mut HashSet::new(),
        )
        .unwrap();

        /*
        solution is
        0. produce(f)(x)->(y)
        1. call(g)->(w)
         */

        assert_eq!(1, pipes.len());
        let pipe = &pipes.get(0).unwrap().cursors;
        assert_eq!(2, pipe.len());

        assert_produce(&pipe, 0);
        assert_call(&pipe, 1);
        Ok(())
    }

    #[test]
    fn simple_cartesian() {
        /*
        match f(x) -> (y), g(v) -> (w)
         */

        let variables_raw = vec!["x", "y", "w", "v"];
        let functions_raw = vec!["f", "g"];
        let mut constant_table = HashMap::new();

        strings_to_constant_table(&variables_raw, &mut constant_table);
        strings_to_constant_table(&functions_raw, &mut constant_table);

        let mut variable_names = raw_to_string_values(&variables_raw);
        let mut function_names = raw_to_function_name_values(&functions_raw);

        variable_names.append(&mut function_names);

        let mut match_stmts = vec![];

        let f_map = P::P(Atom {
            kind: map_atom("f", &vec!["x"], &vec!["y"], &constant_table),
            span: DUMMY_SPAN,
        });

        let g_map = P::P(Atom {
            kind: map_atom("g", &vec!["v"], &vec!["w"], &constant_table),
            span: DUMMY_SPAN,
        });

        match_stmts.push(f_map);
        match_stmts.push(g_map);

        let mut arities = HashMap::new();
        arities.insert("f".to_string(), (1, 1));
        arities.insert("g".to_string(), (1, 1));
        let mut store = StoreHasAllValuesMock { arities };

        let pipes = cut_into_pipes(
            &mut match_stmts,
            &mut WalkContext {
                is_local: true,
                constants: &variable_names,
                store: &mut store,
                initialized_variables: HashSet::new(),
            },
            &mut HashSet::new(),
        )
        .unwrap();

        /*
        0. produce(f)->(y)

        0. produce(g)->(w)
         */

        assert_eq!(2, pipes.len());
        let pipe1 = &pipes.get(0).unwrap().cursors;
        let pipe2 = &pipes.get(1).unwrap().cursors;
        assert_produce(&pipe1, 0);
        assert_produce(&pipe2, 0);
    }

    #[test]
    fn kinda_complicated() {
        /*
        match   f(x, y, z) -> (w),
                g(x, y) -> (x),
                print(w),
                h(k, x),
                print(k),
                greater_than(x, k),
                f(p, p, p)->(p)
        */

        let variables_raw = vec!["x", "y", "z", "w", "k", "p"];
        let functions_raw = vec!["f", "g", "print", "h", "greater_than"];
        let mut constant_table = HashMap::new();

        strings_to_constant_table(&variables_raw, &mut constant_table);
        strings_to_constant_table(&functions_raw, &mut constant_table);

        let mut variable_names = raw_to_string_values(&variables_raw);
        let mut function_names = raw_to_function_name_values(&functions_raw);

        variable_names.append(&mut function_names);

        let f1_map = P::P(Atom {
            kind: map_atom("f", &vec!["x", "y", "z"], &vec!["w"], &constant_table),
            span: DUMMY_SPAN,
        });

        let g_map = P::P(Atom {
            kind: map_atom("g", &vec!["x", "y"], &vec!["x"], &constant_table),
            span: DUMMY_SPAN,
        });

        let print1_call = P::P(Atom {
            kind: funcall_atom("print", &vec!["x"], &constant_table),
            span: DUMMY_SPAN,
        });

        let h_call = P::P(Atom {
            kind: funcall_atom("h", &vec!["k", "x"], &constant_table),
            span: DUMMY_SPAN,
        });

        let print2_call = P::P(Atom {
            kind: funcall_atom("print", &vec!["k"], &constant_table),
            span: DUMMY_SPAN,
        });

        let greater_than_call = P::P(Atom {
            kind: funcall_atom("greater_than", &vec!["x", "k"], &constant_table),
            span: DUMMY_SPAN,
        });

        let f2_map = P::P(Atom {
            kind: map_atom("f", &vec!["p", "p", "p"], &vec!["p"], &constant_table),
            span: DUMMY_SPAN,
        });

        let mut arities = HashMap::new();
        arities.insert("f".to_string(), (3, 1));
        arities.insert("g".to_string(), (2, 1));
        arities.insert("print".to_string(), (1, 1));
        arities.insert("h".to_string(), (2, 1));
        arities.insert("greater_than".to_string(), (2, 1));
        let mut store = StoreHasAllValuesMock { arities };

        let mut match_stmts = vec![];
        match_stmts.push(f1_map);
        match_stmts.push(g_map);
        match_stmts.push(print1_call);
        match_stmts.push(h_call);
        match_stmts.push(print2_call);
        match_stmts.push(greater_than_call);
        match_stmts.push(f2_map);

        let pipes = cut_into_pipes(
            &mut match_stmts,
            &mut WalkContext {
                is_local: true,
                constants: &variable_names,
                store: &mut store,
                initialized_variables: HashSet::new(),
            },
            &mut HashSet::new(),
        )
        .unwrap();

        /*
         solution is
        0. produce(f)(x,y,z) -> (w)
        1. call(g)(x, y) -> (x)
        2. call(print)

        0. produce(h)(k, x)
        1. call(print)(k)
        2. call(greater_than)(x, k)

        0. produce(f) -> (p)
        */

        assert_eq!(3, pipes.len());
        let pipe1 = &pipes.get(0).unwrap().cursors;
        let pipe2 = &pipes.get(1).unwrap().cursors;
        let pipe3 = &pipes.get(2).unwrap().cursors;

        assert_eq!(3, pipe1.len());
        assert_eq!(3, pipe2.len());
        assert_eq!(1, pipe3.len());

        assert_produce(&pipe1, 0);
        assert_call(&pipe1, 1);
        assert_call(&pipe1, 2);

        assert_produce(&pipe2, 0);
        assert_call(&pipe2, 1);
        assert_call(&pipe2, 2);

        assert_produce(&pipe3, 0);
    }

    #[test]
    fn single_element() {
        /*
        match f(x, y)
         */
        let variables_raw = vec!["x", "y"];
        let functions_raw = vec!["f"];
        let mut constant_table = HashMap::new();

        strings_to_constant_table(&variables_raw, &mut constant_table);
        strings_to_constant_table(&functions_raw, &mut constant_table);

        let mut variable_names = raw_to_string_values(&variables_raw);
        let mut function_names = raw_to_function_name_values(&functions_raw);

        variable_names.append(&mut function_names);

        let mut arities = HashMap::new();
        arities.insert("f".to_string(), (2, 1));
        let mut store = StoreHasAllValuesMock { arities };

        let f_call = Atom {
            kind: funcall_atom("f", &vec!["x", "y"], &constant_table),
            span: DUMMY_SPAN,
        };
        let mut match_stmts = vec![];

        match_stmts.push(P::P(f_call));

        let pipes = cut_into_pipes(
            &mut match_stmts,
            &mut WalkContext {
                is_local: true,
                constants: &variable_names,
                store: &mut store,
                initialized_variables: HashSet::new(),
            },
            &mut HashSet::new(),
        )
        .unwrap();
        assert_eq!(1, pipes.len());
        assert_produce(&pipes.get(0).unwrap().cursors, 0);
    }

    #[test]
    fn skip_join() {
        /*
            match f(x) -> (y), g(v) -> (w), h(x) -> (w)
        */
        let variables_raw = vec!["x", "y", "v", "w"];
        let functions_raw = vec!["f", "g", "h"];
        let mut constant_table = HashMap::new();

        strings_to_constant_table(&variables_raw, &mut constant_table);
        strings_to_constant_table(&functions_raw, &mut constant_table);

        let mut variable_names = raw_to_string_values(&variables_raw);
        let mut function_names = raw_to_function_name_values(&functions_raw);

        variable_names.append(&mut function_names);

        let mut arities = HashMap::new();
        arities.insert("f".to_string(), (1, 1));
        arities.insert("g".to_string(), (1, 1));
        arities.insert("h".to_string(), (1, 1));
        let mut store = StoreHasAllValuesMock { arities };

        let mut match_stmts = vec![];

        let f_map = P::P(Atom {
            kind: map_atom("f", &vec!["x"], &vec!["y"], &constant_table),
            span: DUMMY_SPAN,
        });

        let g_map = P::P(Atom {
            kind: map_atom("g", &vec!["v"], &vec!["w"], &constant_table),
            span: DUMMY_SPAN,
        });

        let h_map = P::P(Atom {
            kind: map_atom("h", &vec!["x"], &vec!["w"], &constant_table),
            span: DUMMY_SPAN,
        });

        match_stmts.push(f_map);
        match_stmts.push(g_map);
        match_stmts.push(h_map);

        let pipes = cut_into_pipes(
            &mut match_stmts,
            &mut WalkContext {
                is_local: true,
                constants: &variable_names,
                store: &mut store,
                initialized_variables: HashSet::new(),
            },
            &mut HashSet::new(),
        )
        .unwrap();

        /*
        solution is
        0. produce(f) -> (y)

        0. produce(g) -> (w)
        1. call(h) -> (w)
         */

        assert_eq!(2, pipes.len());
        let pipe1 = &pipes.get(0).unwrap().cursors;
        let pipe2 = &pipes.get(1).unwrap().cursors;
        assert_produce(&pipe1, 0);
        assert_produce(&pipe2, 0);
        assert_call(&pipe2, 1);
    }

    #[test]
    fn check_unbound_var_on_call_is_failure() {
        /*
        // g doesn't have values, so this should fail since x is unbound
        match g(x)
         */

        let variables_raw = vec!["x"];
        let functions_raw = vec!["g"];
        let mut constant_table = HashMap::new();

        strings_to_constant_table(&variables_raw, &mut constant_table);
        strings_to_constant_table(&functions_raw, &mut constant_table);

        let mut variable_names = raw_to_string_values(&variables_raw);
        let mut function_names = raw_to_function_name_values(&functions_raw);

        variable_names.append(&mut function_names);

        // no arities for g means no values
        let mut store = StoreHasAllValuesMock {
            arities: HashMap::new(),
        };

        let g_call = Atom {
            kind: funcall_atom("g", &vec!["x"], &constant_table),
            span: DUMMY_SPAN,
        };

        let mut match_stmts = vec![];

        match_stmts.push(P::P(g_call));

        let pipes = cut_into_pipes(
            &mut match_stmts,
            &mut WalkContext {
                is_local: true,
                constants: &variable_names,
                store: &mut store,
                initialized_variables: HashSet::new(),
            },
            &mut HashSet::new(),
        );
        assert!(pipes.is_err());
    }
}