hamelin_legacy 0.3.10

use hamelin_lib::{
    antlr::hamelinparser::{ExplodeCommandContext, ExplodeCommandContextAttrs},
    err::{TranslationError, TranslationErrors},
    sql::{
        expression::{
            identifier::{Identifier, SimpleIdentifier},
            literal::{BooleanLiteral, ColumnReference},
            Leaf, SQLExpression,
        },
        query::{
            Join, JoinClause, JoinType, SQLQuery, SubQuery, TableAlias, TableExpression,
            TableFunctionApply,
        },
    },
    types::Type,
};

use crate::ast::{assignment_clause::HamelinAssignmentClause, pipeline::HamelinPipeline};
use crate::env::Environment;
use crate::translation::{projection_builder::ProjectionBuilder, PendingQuery};

pub fn translate(
    ctx: &ExplodeCommandContext<'static>,
    pipeline: &HamelinPipeline,
    previous_command: &PendingQuery,
) -> Result<PendingQuery, TranslationErrors> {
    let assignment_ctx = TranslationErrors::expect(ctx, ctx.assignmentClause())?;
    let (identifier, translation) = HamelinAssignmentClause::new(
        assignment_ctx.clone(),
        pipeline
            .context
            .default_expression_translation_context(&previous_command.env),
    )
    .to_sql()?;

    let left_alias = SimpleIdentifier::new("_left");
    let right_alias = SimpleIdentifier::new("_right");
    let right_nested_alias = SimpleIdentifier::new("_nested");

    let left_sql = previous_command.query.clone();
    let left_query = SubQuery::new(left_sql.clone().into()).alias(left_alias.clone().into());

    match translation.typ {
        Type::Array(array) => {
            /*
            This is the alias for the right hand side of the join. We have to be careful in how
            we construct this. If the element type is struct, the database will automatically
            "lift" each subfield into a named field in the result set. If the element type is
            not a struct, we have to give that inner field a name, so that we can then refer to
            it later. We just pick "_nested" so we can find it later, when building the projections.
            */
            let table_alias = match array.element_type.as_ref() {
                Type::Struct(_) => TableAlias::new(right_alias.clone().into()),
                _ => TableAlias::new(right_alias.clone().into())
                    .with_field_alias(right_nested_alias.clone()),
            };

            let unnest = TableFunctionApply::new("unnest".to_string())
                .with_argument(translation.sql.clone())
                .with_alias(table_alias.into());

            let from: TableExpression = if left_sql == SQLQuery::default() {
                unnest.into()
            } else {
                Join::new(left_query.into())
                    .with_clause(JoinClause {
                        table: unnest.into(),
                        join_type: JoinType::LEFT,
                        condition: Some(BooleanLiteral::new(true).into()),
                    })
                    .into()
            };

            let mut pb = ProjectionBuilder::default();

            // First, load up the projection builder with "_left".* for all the old fields.
            for (k, v) in previous_command.env.fields.fields.iter() {
                let ident: Identifier = k.clone().into();
                let cr =
                    ColumnReference::new(ident.prefix_from_simples(&[left_alias.clone()]).into());
                pb.bind(ident, cr.into(), v.clone());
            }

            match array.element_type.as_ref() {
                Type::Struct(srct) => {
                    // Next, generate the cast expression that represents the inner struct type.
                    let cast: SQLExpression =
                        ProjectionBuilder::deep_initialize_from_struct(None, srct.clone())
                            .build_cast()
                            .map_err(|e| TranslationError::wrap_box(ctx, e.into()))?
                            .into();

                    // Prefix any inner references with the "_right" alias.
                    let cast_with_prefix = cast.map_leaves(|leaf| match leaf {
                        Leaf::ColumnReference(column_reference) => ColumnReference::new(
                            column_reference
                                .identifier
                                .prefix_from_simples(&[right_alias.clone()])
                                .into(),
                        )
                        .into(),
                        l => l.into(),
                    });

                    // Lastly, bind the struct expression on the right hand side.
                    pb.bind(identifier, cast_with_prefix, srct.clone().into());
                }
                t => {
                    // No fancy business needed for the non-struct case. Just column ref _right.
                    let right_nested_identifier: Identifier = right_nested_alias.clone().into();
                    let cr = ColumnReference::new(
                        right_nested_identifier
                            .prefix_from_simples(&[right_alias])
                            .into(),
                    );

                    pb.bind(identifier, cr.into(), t.clone());
                }
            }

            let projections = pb
                .clone()
                .build_projections()
                .map_err(|e| TranslationError::wrap_box(ctx, e.into()))?;

            let query = SQLQuery::default().from(from).select(projections);
            let merged_env = Environment::new(pb.build_hamelin_type());

            Ok(PendingQuery::new(query, merged_env))
        }
        t => {
            return Err(TranslationError::msg(
                assignment_ctx.as_ref(),
                &format!("Unsupported type {} for explode command. Must be array.", t),
            )
            .single())
        }
    }
}