hamelin_legacy 0.3.10

use std::rc::Rc;
use std::sync::Arc;

use crate::ast::from_clause::HamelinFromClause;
use crate::ast::pipeline::HamelinPipeline;
use crate::env::Environment;
use crate::translation::projection_builder::{ProjectionBuilder, ProjectionBuilderExt};
use crate::translation::sql_query_helpers::{add_filter_condition, prepend_projections};
use crate::translation::PendingQuery;
use hamelin_lib::antlr::hamelinparser::FromClauseContextAll;
use hamelin_lib::err::{TranslationError, TranslationErrors};
use hamelin_lib::sql::query::set::SetOperation;
use hamelin_lib::sql::query::{SQLQuery, SQLQueryExpression, SubQuery};

#[derive(Clone, Copy, PartialEq, Eq)]
pub enum FromAliasing<'a> {
    AliasingAllowed,
    RaiseErrorOnAlias(&'a str),
}

#[derive(Clone, Copy, PartialEq, Eq)]
pub enum FromAnonymous {
    DefaultAlias,
    DoNotNest,
}

/// The shared code that implements the "from"ing for FROM, UNION, and MATCH.
pub fn uber_from(
    ctx: Vec<Rc<FromClauseContextAll<'static>>>,
    pipeline: &HamelinPipeline,
    previous_command: &PendingQuery,
    aliasing: FromAliasing,
    anonymous: FromAnonymous,
) -> Result<PendingQuery, TranslationErrors> {
    let mut merged_env = Environment::default();

    let hfc_vec: Vec<HamelinFromClause> = ctx
        .iter()
        .map(|from| {
            HamelinFromClause::new(
                from.clone(),
                Arc::new(pipeline.cte.clone()),
                pipeline.context.clone(),
            )
        })
        .collect();

    for from in hfc_vec.iter() {
        from.append_completions();
    }

    // Merge the environment (finding fields that are the same, and overlapping them correctly)
    // - Push a table alias nested field into the environment (for the ones that are assigned)
    // - Remember the environment for each of these aliases
    let mut table_refs = vec![];
    for (hfc, from) in hfc_vec.iter().zip(ctx.iter()) {
        let (env, tr, maybe_si) = hfc.reflect()?;

        let si = match (maybe_si, &aliasing, &anonymous) {
            (Some(_), FromAliasing::RaiseErrorOnAlias(error), _) => {
                let err = TranslationError::msg(from.as_ref(), error).single();

                return Err(err);
            }
            (Some(si), FromAliasing::AliasingAllowed, _) => Some(si),
            (None, _, FromAnonymous::DefaultAlias) => Some(hfc.table_identifier()?.last().clone()),
            (None, _, FromAnonymous::DoNotNest) => None,
        };

        table_refs.push((tr, si.clone(), env.clone(), from.clone()));

        let env_to_merge = match si {
            Some(si) => env
                .clone()
                .with_binding(si.into(), env.fields.clone().into()),
            None => env.clone(),
        };

        merged_env = merged_env.merge(&env_to_merge).map_err(|e| match &e.at {
            Some(at) => TranslationError::msg(
                from.as_ref(),
                format!("Field {} is of type {}.", at.to_hamelin(), e.other).as_str(),
            )
            .with_source(e),

            None => TranslationError::wrap(from.as_ref(), e),
        })?;
    }

    if table_refs.is_empty() {
        // The code below assumes that there is at least one clause in the from.
        // That code is tedious to refactor, so I am just going to bail and do
        // nothing in the event that the required minimum is not present.
        return Ok(previous_command.clone());
    }

    // Build each leg of the union. Remember each leg's environment in case we need it.
    // We take a little shortcut for the simplest FROM whatever. We don't initialize a ProjectionBuilder
    // so that we don't blow up the query with CAST ROW AS ROW's that don't add any value.
    let mut queries = vec![];
    let (first_table_ref, first_table_alias, _, _) = table_refs.get(0).unwrap();
    if table_refs.len() == 1 && first_table_alias.is_none() {
        let query = previous_command
            .query
            .clone()
            .from(first_table_ref.clone().into())
            .select(
                merged_env
                    .get_column_projections()
                    .into_iter()
                    .map(|cp| cp.into())
                    .collect(),
            );
        queries.push(query);
    } else {
        for (table_ref, table_alias, env, from_clause) in table_refs {
            let mut projection_builder = ProjectionBuilder::deep_initialize_from_environment(&env);

            if let Some(ta) = table_alias {
                // Add the table alias "nested" column.
                // Note that we already added this column to the env during merging.
                // We have to merge the env together the first time through the loop

                projection_builder.bind(
                    ta.into(),
                    ProjectionBuilder::shallow_initialize_from_environment(&env)
                        .build_cast()
                        .map_err(|e| TranslationError::wrap_box(from_clause.as_ref(), e.into()))?
                        .into(),
                    env.fields.clone().into(),
                );
            }

            let query = prepend_projections(
                &previous_command
                    .query
                    .clone()
                    .from(table_ref.clone().into()),
                projection_builder
                    .expand(merged_env.fields.clone())
                    .map_err(|e| TranslationError::wrap_box(from_clause.as_ref(), e.into()))?
                    .build_projections()
                    .map_err(|e| TranslationError::wrap_box(from_clause.as_ref(), e.into()))?,
                &merged_env,
            );
            queries.push(query);
        }
    }

    let query = if queries.len() == 1 {
        queries.pop().unwrap()
    } else {
        SQLQuery::default()
            .from(
                SubQuery::new(
                    queries
                        .into_iter()
                        .map(|q| SQLQueryExpression::from(q))
                        .reduce(|a, b| SetOperation::new(a.into(), b.into()).into())
                        .unwrap_or_else(|| SQLQuery::default().into())
                        .into(),
                )
                .into(),
            )
            .select(
                merged_env
                    .get_column_projections()
                    .into_iter()
                    .map(|cp| cp.into())
                    .collect(),
            )
    };

    let mut ret_query = query;
    if let Some(wf) = pipeline.within.clone() {
        if merged_env.lookup(&"timestamp".parse()?).is_ok() {
            ret_query = add_filter_condition(&ret_query, wf, &merged_env);
        }
    }
    Ok(PendingQuery::new(ret_query, merged_env))
}