hamelin_legacy 0.4.4

use std::sync::Arc;

use hamelin_lib::{
    antlr::hamelinparser::{JoinCommandContext, JoinCommandContextAttrs},
    err::{TranslationError, TranslationErrors, UnexpectedType},
    sql::{
        expression::{
            identifier::{Identifier, SimpleIdentifier},
            literal::{BooleanLiteral, ColumnReference},
            Leaf, SQLExpression,
        },
        query::{
            projection::{Binding, ColumnProjection},
            Join, JoinClause, JoinType, SQLQuery, SubQuery,
        },
    },
    types::BOOLEAN,
};

use crate::ast::{
    expression::HamelinExpression, from_clause::HamelinFromClause, pipeline::HamelinPipeline,
};
use crate::translation::{
    projection_builder::{ProjectionBuilder, ProjectionBuilderExt},
    PendingQuery,
};

pub fn translate(
    ctx: &JoinCommandContext<'static>,
    pipeline: &HamelinPipeline,
    translation: &PendingQuery,
) -> Result<PendingQuery, TranslationErrors> {
    let join_type: JoinType = if ctx.LOOKUP_COMMAND().is_some() {
        JoinType::LEFT
    } else {
        JoinType::INNER
    };

    let hfc = HamelinFromClause::new(
        TranslationErrors::expect(ctx, ctx.fromClause())?,
        Arc::new(pipeline.cte.clone()),
        pipeline.context.clone(),
    );
    let (right_env, right_table_ref, right_alias) = hfc.reflect()?;
    hfc.append_completions();
    let right_alias_or_default = right_alias.unwrap_or(hfc.table_identifier()?.last().clone());

    let left_alias = SimpleIdentifier::new("_left");

    let new_env = translation.env.clone().with_binding(
        right_alias_or_default.clone().into(),
        right_env.fields.clone().into(),
    );

    let join_expression: SQLExpression = if let Some(on) = ctx.on.as_ref() {
        // JOIN right=some_table ON field == right.field
        // JOIN some_table ON field == some_table.field
        //
        // Interesting. When we translate the ON clause, the left references
        // will never be aliased. When we see a field reference come out
        // of the left side, we'll need to translate it to an aliased name.
        // But, when we see a reference to a right side, we know it will
        // already be aliased.
        //
        // Solve this by checking the expression against the correct env, but
        // then prepending the left alias to every column reference that does
        // not start with the right alias.
        let join_clause = HamelinExpression::new(
            on.clone(),
            pipeline
                .context
                .default_expression_translation_context(&new_env),
        );
        let translation = join_clause.translate()?;

        if translation.typ != BOOLEAN {
            return Err(TranslationError::wrap(
                on.as_ref(),
                UnexpectedType::new(translation.typ, vec![BOOLEAN]),
            )
            .single());
        }

        translation.sql.map_leaves(|leaf| match leaf {
            Leaf::ColumnReference(column_reference) => {
                let ident: Identifier =
                    if *column_reference.identifier.first() != right_alias_or_default {
                        column_reference
                            .identifier
                            .prefix_from_simples(&[left_alias.clone()])
                            .into()
                    } else {
                        column_reference.identifier
                    };

                ColumnReference::new(ident).into()
            }
            leaf @ _ => leaf.into(),
        })
    } else {
        BooleanLiteral::new(true).into()
    };

    let left_query =
        SubQuery::new(translation.query.clone().into()).alias(left_alias.clone().into());
    let join = Join::new(left_query.into()).with_clause(JoinClause {
        table: right_table_ref
            .alias(right_alias_or_default.clone().into())
            .into(),
        join_type,
        condition: Some(join_expression),
    });

    let right_cast: SQLExpression = ProjectionBuilder::deep_initialize_from_environment(&right_env)
        .build_cast()
        .map_err(|e| TranslationError::wrap_box(ctx, e.into()).single())?
        .into();

    let right_cast_mapped = right_cast.map_leaves(|leaf| {
        if let Leaf::ColumnReference(column_reference) = leaf {
            let ident = column_reference
                .identifier
                .prefix_from_simples(&[right_alias_or_default.clone()]);
            ColumnReference::new(ident.into()).into()
        } else {
            leaf.into()
        }
    });

    let query = SQLQuery::default().from(join.into()).select(
        translation
            .env
            .get_column_projections()
            .into_iter()
            .map(|cp| {
                ColumnProjection::new(
                    cp.identifier
                        .prefix_from_simples(&[left_alias.clone()])
                        .into(),
                )
                .into()
            })
            .chain(
                [Binding::new(
                    right_alias_or_default.clone().into(),
                    right_cast_mapped.into(),
                )
                .into()]
                .into_iter(),
            )
            .collect(),
    );

    Ok(PendingQuery::new(query, new_env.clone()))
}