use super::CompiledPreAggregation;
use super::PreAggregationSource;
use crate::cube_bridge::join_hints::JoinHintItem;
use crate::cube_bridge::member_sql::MemberSql;
use crate::cube_bridge::pre_aggregation_description::PreAggregationDescription;
use crate::logical_plan::PreAggregationJoin;
use crate::logical_plan::PreAggregationJoinItem;
use crate::logical_plan::PreAggregationTable;
use crate::logical_plan::PreAggregationUnion;
use crate::logical_plan::PreAggregationUnionItem;
use crate::planner::join_hints::JoinHints;
use crate::planner::multi_fact_join_groups::{MeasuresJoinHints, MultiFactJoinGroups};
use crate::planner::planners::JoinPlanner;
use crate::planner::planners::ResolvedJoinItem;
use crate::planner::state::State;
use crate::planner::Compiler;
use crate::planner::GranularityHelper;
use crate::planner::MemberSymbol;
use crate::planner::SqlCall;
use crate::planner::SqlCallReference;
use crate::planner::SymbolPath;
use crate::planner::SymbolPathType;
use crate::planner::TimeDimensionSymbol;
use crate::utils::debug::DebugSql;
use cubenativeutils::CubeError;
use cubenativeutils::CubeErrorCauseType;
use itertools::Itertools;
use std::collections::HashMap;
use std::fmt::Debug;
use std::rc::Rc;

#[derive(Debug, Clone, Hash, PartialEq, Eq)]
pub struct PreAggregationFullName {
    pub cube_name: String,
    pub name: String,
}

impl PreAggregationFullName {
    pub fn from_string(name: &str) -> Result<Self, CubeError> {
        let parts = name.split('.').collect_vec();
        if parts.len() != 2 {
            Err(CubeError::user(format!(
                "Invalid pre-aggregation name: {}",
                name
            )))
        } else {
            Ok(Self {
                cube_name: parts[0].to_string(),
                name: parts[1].to_string(),
            })
        }
    }

    pub fn new(cube_name: String, name: String) -> Self {
        Self { cube_name, name }
    }
}

pub struct PreAggregationsCompiler {
    query_tools: Rc<State>,
    descriptions: Rc<Vec<(PreAggregationFullName, Rc<dyn PreAggregationDescription>)>>,
    compiled_cache: HashMap<PreAggregationFullName, Rc<CompiledPreAggregation>>,
}

impl PreAggregationsCompiler {
    pub fn try_new(query_tools: Rc<State>, cube_names: &Vec<String>) -> Result<Self, CubeError> {
        let mut descriptions = Vec::new();
        for cube_name in cube_names.iter() {
            let pre_aggregations = query_tools
                .cube_evaluator()
                .pre_aggregations_for_cube_as_array(cube_name.clone())?;
            for pre_aggregation in pre_aggregations.iter() {
                let full_name = PreAggregationFullName::new(
                    cube_name.clone(),
                    pre_aggregation.static_data().name.clone(),
                );
                descriptions.push((full_name, pre_aggregation.clone()));
            }
        }
        Ok(Self {
            query_tools,
            descriptions: Rc::new(descriptions),
            compiled_cache: HashMap::new(),
        })
    }

    pub fn compile_pre_aggregation(
        &mut self,
        name: &PreAggregationFullName,
    ) -> Result<Rc<CompiledPreAggregation>, CubeError> {
        if let Some(compiled) = self.compiled_cache.get(&name) {
            return Ok(compiled.clone());
        }

        let description = if let Some((_, description)) =
            self.descriptions.clone().iter().find(|(n, _)| n == name)
        {
            description.clone()
        } else {
            if let Some(descr) = self
                .query_tools
                .cube_evaluator()
                .pre_aggregation_description_by_name(name.cube_name.clone(), name.name.clone())?
            {
                descr
            } else {
                return Err(CubeError::internal(format!(
                    "Undefined pre-aggregation {}.{}",
                    name.cube_name, name.name
                )));
            }
        };

        let static_data = description.static_data();

        if static_data.pre_aggregation_type == "rollupLambda" {
            return self.build_lambda(name, &description);
        }

        let measures = if let Some(refs) = description.measure_references()? {
            Self::symbols_from_ref(self.query_tools.clone(), name, refs, Self::check_is_measure)?
        } else {
            Vec::new()
        };
        let dimensions = if let Some(refs) = description.dimension_references()? {
            Self::symbols_from_ref(
                self.query_tools.clone(),
                name,
                refs,
                Self::check_is_dimension,
            )?
        } else {
            Vec::new()
        };
        let time_dimensions = if let Some(td_refs) = description.time_dimension_references()? {
            let mut resolved = Vec::new();
            for td_ref in td_refs.iter() {
                let base_symbol = Self::time_dimension_symbol_from_ref(
                    self.query_tools.clone(),
                    name,
                    td_ref.dimension()?,
                )?;
                resolved.push((base_symbol, td_ref.static_data().granularity.clone()));
            }

            let evaluator_compiler_cell = self.query_tools.compiler().clone();
            let mut evaluator_compiler = evaluator_compiler_cell.borrow_mut();
            let mut result = Vec::new();
            for (base_symbol, granularity) in resolved {
                let granularity_obj = GranularityHelper::make_granularity_obj(
                    self.query_tools.cube_evaluator().clone(),
                    &mut evaluator_compiler,
                    &base_symbol.cube_name(),
                    &base_symbol.name(),
                    Some(granularity.clone()),
                )?;
                let symbol = MemberSymbol::new_time_dimension(TimeDimensionSymbol::new(
                    base_symbol,
                    Some(granularity),
                    granularity_obj,
                    None,
                ));
                result.push(symbol);
            }
            result
        } else if let Some(refs) = description.time_dimension_reference()? {
            let base_symbol =
                Self::time_dimension_symbol_from_ref(self.query_tools.clone(), name, refs)?;

            if static_data.granularity.is_some() {
                let evaluator_compiler_cell = self.query_tools.compiler().clone();
                let mut evaluator_compiler = evaluator_compiler_cell.borrow_mut();

                let granularity_obj = GranularityHelper::make_granularity_obj(
                    self.query_tools.cube_evaluator().clone(),
                    &mut evaluator_compiler,
                    &base_symbol.cube_name(),
                    &base_symbol.name(),
                    static_data.granularity.clone(),
                )?;
                let symbol = MemberSymbol::new_time_dimension(TimeDimensionSymbol::new(
                    base_symbol,
                    static_data.granularity.clone(),
                    granularity_obj,
                    None,
                ));

                vec![symbol]
            } else {
                vec![base_symbol]
            }
        } else {
            Vec::new()
        };
        let segments = if let Some(refs) = description.segment_references()? {
            Self::symbols_from_ref(self.query_tools.clone(), name, refs, Self::check_is_segment)?
        } else {
            Vec::new()
        };
        let allow_non_strict_date_range_match = description
            .static_data()
            .allow_non_strict_date_range_match
            .unwrap_or(false);

        let measures_join_hints = MeasuresJoinHints::builder(&JoinHints::new())
            .add_dimensions(&dimensions)
            .add_dimensions(&time_dimensions)
            .build(&measures)?;
        let multi_fact_join_groups =
            MultiFactJoinGroups::try_new(self.query_tools.clone(), measures_join_hints)?;

        let rollups = if let Some(refs) = description.rollup_references()? {
            let r = self
                .query_tools
                .cube_evaluator()
                .evaluate_rollup_references(name.cube_name.clone(), refs)?;
            r
        } else {
            Vec::new()
        };

        let source = if static_data.pre_aggregation_type == "rollupJoin" {
            let all_dimensions = dimensions
                .iter()
                .cloned()
                .chain(time_dimensions.iter().cloned())
                .chain(segments.iter().cloned())
                .collect_vec();
            PreAggregationSource::Join(self.build_join_source(
                &measures,
                &all_dimensions,
                &rollups,
            )?)
        } else {
            let cube = self
                .query_tools
                .cube_evaluator()
                .cube_from_path(name.cube_name.clone())?;
            let cube_alias = if let Some(alias) = &cube.static_data().sql_alias {
                alias.clone()
            } else {
                name.cube_name.clone()
            };
            PreAggregationSource::Single(PreAggregationTable {
                cube_name: name.cube_name.clone(),
                cube_alias,
                name: name.name.clone(),
                alias: static_data.sql_alias.clone(),
                usage_index: None,
            })
        };

        let res = Rc::new(CompiledPreAggregation {
            name: static_data.name.clone(),
            cube_name: name.cube_name.clone(),
            source: Rc::new(source),
            granularity: static_data.granularity.clone(),
            external: static_data.external,
            measures,
            dimensions,
            time_dimensions,
            segments,
            allow_non_strict_date_range_match,
            multi_fact_join_groups,
        });
        self.compiled_cache.insert(name.clone(), res.clone());
        Ok(res)
    }

    fn build_lambda(
        &mut self,
        name: &PreAggregationFullName,
        description: &Rc<dyn PreAggregationDescription>,
    ) -> Result<Rc<CompiledPreAggregation>, CubeError> {
        let rollups = if let Some(refs) = description.rollup_references()? {
            let r = self
                .query_tools
                .cube_evaluator()
                .evaluate_rollup_references(name.cube_name.clone(), refs)?;
            r
        } else {
            Vec::new()
        };
        if rollups.is_empty() {
            return Err(CubeError::user(format!(
                "rollupLambda '{}.{}' should reference at least one rollup",
                name.cube_name, name.name
            )));
        }

        let pre_aggrs_for_lambda = rollups
            .iter()
            .map(|item| -> Result<_, CubeError> {
                self.compile_pre_aggregation(&PreAggregationFullName::from_string(item)?)
            })
            .collect::<Result<Vec<_>, _>>()?;

        let mut sources = vec![];
        for (i, rollup) in pre_aggrs_for_lambda.clone().iter().enumerate() {
            match rollup.source.as_ref() {
                PreAggregationSource::Single(table) => {
                    // Carry this branch's own symbols: each member rollup stores its
                    // columns under its own cube aliases, while the lambda exposes the
                    // first rollup's symbols. The physical builder maps lambda members
                    // to each branch's stored column through these.
                    sources.push(PreAggregationUnionItem {
                        table: Rc::new(table.clone()),
                        measures: rollup.measures.clone(),
                        dimensions: rollup.dimensions.clone(),
                        time_dimensions: rollup.time_dimensions.clone(),
                    });
                }
                _ => {
                    return Err(CubeError::user(format!("Rollup lambda can't be nested")));
                }
            }
            if i >= 1 {
                Self::match_symbols(&rollup.measures, &pre_aggrs_for_lambda[0].measures)?;
                Self::match_symbols(&rollup.dimensions, &pre_aggrs_for_lambda[0].dimensions)?;
                Self::match_time_dimensions(
                    &rollup.time_dimensions,
                    &pre_aggrs_for_lambda[0].time_dimensions,
                )?;
            }
        }

        let measures = pre_aggrs_for_lambda[0].measures.clone();
        let dimensions = pre_aggrs_for_lambda[0].dimensions.clone();
        let time_dimensions = pre_aggrs_for_lambda[0].time_dimensions.clone();
        let segments = pre_aggrs_for_lambda[0].segments.clone();
        let allow_non_strict_date_range_match = description
            .static_data()
            .allow_non_strict_date_range_match
            .unwrap_or(false);
        let granularity = pre_aggrs_for_lambda[0].granularity.clone();
        let multi_fact_join_groups = pre_aggrs_for_lambda[0].multi_fact_join_groups.clone();
        let source = PreAggregationSource::Union(PreAggregationUnion { items: sources });

        let static_data = description.static_data();
        // A rollupLambda is left without an `external` default in the data model;
        // its storage is whatever its member rollups use. The union is rendered as a
        // single query, so it is external only when every member rollup is — then the
        // read renders with the right dialect (e.g. CubeStore `?` placeholders).
        let external = if pre_aggrs_for_lambda
            .iter()
            .all(|p| p.external == Some(true))
        {
            Some(true)
        } else {
            static_data.external
        };
        let res = Rc::new(CompiledPreAggregation {
            name: static_data.name.clone(),
            cube_name: name.cube_name.clone(),
            source: Rc::new(source),
            granularity,
            external,
            measures,
            dimensions,
            time_dimensions,
            segments,
            allow_non_strict_date_range_match,
            multi_fact_join_groups,
        });
        self.compiled_cache.insert(name.clone(), res.clone());
        Ok(res)
    }

    fn match_symbols(
        a: &Vec<Rc<MemberSymbol>>,
        b: &Vec<Rc<MemberSymbol>>,
    ) -> Result<(), CubeError> {
        if !a.iter().zip(b.iter()).all(|(a, b)| a.name() == b.name()) {
            return Err(CubeError::user(format!(
                "Names for pre-aggregation symbols in lambda pre-aggragation don't match"
            )));
        }
        Ok(())
    }

    fn match_time_dimensions(
        a: &Vec<Rc<MemberSymbol>>,
        b: &Vec<Rc<MemberSymbol>>,
    ) -> Result<(), CubeError> {
        if !a.iter().zip(b.iter()).all(|(a, b)| {
            if let (Ok(td_a), Ok(td_b)) = (a.as_time_dimension(), b.as_time_dimension()) {
                td_a.name() == td_a.name() && td_a.granularity() == td_b.granularity()
            } else {
                false
            }
        }) {
            return Err(CubeError::user(format!(
                "Names for pre-aggregation symbols in lambda pre-aggragation don't match"
            )));
        }
        Ok(())
    }

    fn join_hints_from_pre_aggregation(&self, symbols: &Vec<Rc<MemberSymbol>>) -> JoinHints {
        let mut result = JoinHints::new();
        for symbol in symbols {
            let path = symbol.path();
            if path.len() == 1 {
                result.push(JoinHintItem::Single(path[0].clone()));
            } else {
                result.push(JoinHintItem::Vector(path.clone()));
            }
        }
        result
    }
    fn build_join_source(
        &mut self,
        measures: &Vec<Rc<MemberSymbol>>,
        all_dimensions: &Vec<Rc<MemberSymbol>>,
        rollups: &Vec<String>,
    ) -> Result<PreAggregationJoin, CubeError> {
        let all_symbols = measures
            .iter()
            .cloned()
            .chain(all_dimensions.iter().cloned())
            .collect_vec();
        let pre_aggr_join_hints = self.join_hints_from_pre_aggregation(&all_symbols);

        let join_planner = JoinPlanner::new(self.query_tools.clone());
        let pre_aggrs_for_join = rollups
            .iter()
            .map(|item| -> Result<_, CubeError> {
                self.compile_pre_aggregation(&PreAggregationFullName::from_string(item)?)
            })
            .collect::<Result<Vec<_>, _>>()?;

        let target_joins = join_planner.resolve_join_members_by_hints(&pre_aggr_join_hints)?;
        let mut existing_joins = vec![];
        for join_pre_aggr in pre_aggrs_for_join.iter() {
            let all_symbols = join_pre_aggr
                .measures
                .iter()
                .cloned()
                .chain(join_pre_aggr.dimensions.iter().cloned())
                .chain(join_pre_aggr.time_dimensions.iter().cloned())
                .chain(join_pre_aggr.segments.iter().cloned())
                .collect_vec();
            let join_pre_aggr_join_hints = self.join_hints_from_pre_aggregation(&all_symbols);
            let mut existing =
                join_planner.resolve_join_members_by_hints(&join_pre_aggr_join_hints)?;
            existing_joins.append(&mut existing);
        }

        let not_existing_joins = target_joins
            .into_iter()
            .filter(|join| {
                !existing_joins
                    .iter()
                    .any(|existing| existing.is_same_as(join))
            })
            .collect_vec();

        if not_existing_joins.is_empty() {
            return Err(CubeError::user(format!("Nothing to join in rollup join. Target joins are included in existing rollup joins")));
        }

        let items = not_existing_joins
            .iter()
            .map(|item| self.make_pre_aggregation_join_item(&pre_aggrs_for_join, item))
            .collect::<Result<Vec<_>, _>>()?;

        let res = PreAggregationJoin {
            root: items[0].from.clone(),
            items,
        };
        Ok(res)
    }

    fn make_pre_aggregation_join_item(
        &self,
        pre_aggrs_for_join: &Vec<Rc<CompiledPreAggregation>>,
        join_item: &ResolvedJoinItem,
    ) -> Result<PreAggregationJoinItem, CubeError> {
        let from_pre_aggr =
            self.find_pre_aggregation_for_join(pre_aggrs_for_join, &join_item.from_members)?;
        let to_pre_aggr =
            self.find_pre_aggregation_for_join(pre_aggrs_for_join, &join_item.to_members)?;

        let res = PreAggregationJoinItem {
            from: from_pre_aggr.source.clone(),
            to: to_pre_aggr.source.clone(),
            from_members: join_item.from_members.clone(),
            to_members: join_item.to_members.clone(),
            on_sql: join_item.on_sql.clone(),
        };
        Ok(res)
    }

    fn find_pre_aggregation_for_join(
        &self,
        pre_aggrs_for_join: &Vec<Rc<CompiledPreAggregation>>,
        members: &Vec<Rc<MemberSymbol>>,
    ) -> Result<Rc<CompiledPreAggregation>, CubeError> {
        let found_pre_aggr = pre_aggrs_for_join
            .iter()
            .filter(|pa| {
                members
                    .iter()
                    .all(|m| pa.dimensions.iter().any(|pa_m| m == pa_m))
            })
            .collect_vec();
        if found_pre_aggr.is_empty() {
            return Err(CubeError::user(format!(
                "No rollups found that can be used for rollup join"
            )));
        }
        if found_pre_aggr.len() > 1 {
            return Err(CubeError::user(format!(
                "Multiple rollups found that can be used for rollup join"
            )));
        }

        Ok(found_pre_aggr[0].clone())
    }

    pub fn compile_all_pre_aggregations(
        &mut self,
        disable_external_pre_aggregations: bool,
    ) -> Result<Vec<Rc<CompiledPreAggregation>>, CubeError> {
        let mut result = Vec::new();
        for (name, _) in self.descriptions.clone().iter() {
            let pre_aggregation = self.compile_pre_aggregation(name)?;
            if !(disable_external_pre_aggregations && pre_aggregation.external == Some(true)) {
                result.push(pre_aggregation);
            }
        }
        Ok(result)
    }

    pub fn compile_origin_sql_pre_aggregation(
        &mut self,
        cube_name: &String,
    ) -> Result<Option<Rc<CompiledPreAggregation>>, CubeError> {
        let res = if let Some((name, _)) = self.descriptions.clone().iter().find(|(name, descr)| {
            &name.cube_name == cube_name
                && &descr.static_data().pre_aggregation_type == "originalSql"
        }) {
            Some(self.compile_pre_aggregation(name)?)
        } else {
            None
        };
        Ok(res)
    }

    fn symbols_from_ref<F: Fn(&MemberSymbol) -> Result<(), CubeError>>(
        query_tools: Rc<State>,
        name: &PreAggregationFullName,
        ref_func: Rc<dyn MemberSql>,
        check_type_fn: F,
    ) -> Result<Vec<Rc<MemberSymbol>>, CubeError> {
        let evaluator_compiler_cell = query_tools.compiler().clone();
        let mut evaluator_compiler = evaluator_compiler_cell.borrow_mut();
        let sql_call = evaluator_compiler.compile_sql_call(&name.cube_name, ref_func)?;
        let symbols =
            Self::reference_symbols(&query_tools, &mut evaluator_compiler, name, &sql_call)?;
        for symbol in symbols.iter() {
            check_type_fn(symbol)?;
        }
        Ok(symbols)
    }

    fn time_dimension_symbol_from_ref(
        query_tools: Rc<State>,
        name: &PreAggregationFullName,
        ref_func: Rc<dyn MemberSql>,
    ) -> Result<Rc<MemberSymbol>, CubeError> {
        let evaluator_compiler_cell = query_tools.compiler().clone();
        let mut evaluator_compiler = evaluator_compiler_cell.borrow_mut();
        let sql_call = evaluator_compiler.compile_sql_call(&name.cube_name, ref_func)?;

        let symbols =
            Self::reference_symbols(&query_tools, &mut evaluator_compiler, name, &sql_call)?;
        for symbol in symbols.iter() {
            Self::check_is_time_dimension(symbol)?;
        }

        symbols.into_iter().next().ok_or_else(|| {
            let path = sql_call.debug_sql(true);
            Self::reference_not_found_error(&path, name)
        })
    }

    /// Members a pre-aggregation reference declaration names, in declaration
    /// order. An element that interpolated the cube instead of the member is
    /// resolved here; an element naming no member at all is an error, so a
    /// reference is never silently dropped.
    fn reference_symbols(
        query_tools: &Rc<State>,
        evaluator_compiler: &mut Compiler,
        name: &PreAggregationFullName,
        sql_call: &SqlCall,
    ) -> Result<Vec<Rc<MemberSymbol>>, CubeError> {
        let mut result = Vec::new();
        for item in sql_call.reference_items() {
            let symbol = match item {
                SqlCallReference::Symbol(symbol) => symbol,
                SqlCallReference::Path(path) => {
                    let full_name = path.join(".");
                    let symbol_path =
                        SymbolPath::parse(query_tools.cube_evaluator().clone(), &full_name)
                            // A path the data model doesn't know is reported with the
                            // pre-aggregation it came from; anything else (a failure
                            // reaching the model at all) is passed through as it is.
                            .map_err(|e| match e.cause {
                                CubeErrorCauseType::User => {
                                    Self::reference_not_found_error(&full_name, name)
                                }
                                _ => e,
                            })?;
                    match symbol_path.path_type() {
                        SymbolPathType::Dimension => {
                            evaluator_compiler.add_dimension_evaluator_by_path(symbol_path)?
                        }
                        SymbolPathType::Measure => {
                            evaluator_compiler.add_measure_evaluator_by_path(symbol_path)?
                        }
                        SymbolPathType::Segment => {
                            evaluator_compiler.add_segment_evaluator_by_path(symbol_path)?
                        }
                        _ => return Err(Self::reference_not_found_error(&full_name, name)),
                    }
                }
                SqlCallReference::Unresolved(rendered) => {
                    return Err(Self::reference_not_found_error(&rendered, name))
                }
            };
            result.push(symbol);
        }
        Ok(result)
    }

    fn reference_not_found_error(path: &str, name: &PreAggregationFullName) -> CubeError {
        let member_name = path.rsplit('.').next().unwrap_or(path);
        CubeError::user(format!(
            "'{}' not found for path '{}' in pre-aggregation '{}.{}'",
            member_name, path, name.cube_name, name.name
        ))
    }

    fn check_is_measure(symbol: &MemberSymbol) -> Result<(), CubeError> {
        symbol
            .as_measure()
            .map_err(|_| CubeError::user(format!("Pre-aggregation measure must be a measure")))?;
        Ok(())
    }

    fn check_is_dimension(symbol: &MemberSymbol) -> Result<(), CubeError> {
        symbol.as_dimension().map_err(|_| {
            CubeError::user(format!("Pre-aggregation dimension must be a dimension"))
        })?;
        Ok(())
    }

    fn check_is_time_dimension(symbol: &MemberSymbol) -> Result<(), CubeError> {
        let dimension = symbol.as_dimension().map_err(|_| {
            CubeError::user(format!(
                "Pre-aggregation time dimension must be a dimension"
            ))
        })?;
        if !dimension.is_time() {
            return Err(CubeError::user(format!(
                "Pre-aggregation time dimension must be a dimension"
            )));
        }
        Ok(())
    }

    fn check_is_segment(symbol: &MemberSymbol) -> Result<(), CubeError> {
        symbol.as_member_expression().map_err(|_| {
            CubeError::user(
                "Pre-aggregation segment reference must be a member expression".to_string(),
            )
        })?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_fixtures::cube_bridge::MockSchema;
    use crate::test_fixtures::test_utils::TestContext;
    use indoc::indoc;

    fn create_reference_context() -> TestContext {
        // Template syntax like `\"{CUBE}.created_at\"` models a reference built by
        // interpolating the cube itself — `(CUBE) => `${CUBE}.created_at``, where
        // the member name arrives as literal text next to a cube reference
        // instead of as a member symbol.
        let schema = MockSchema::from_yaml(indoc! {"
            cubes:
              - name: orders
                sql: SELECT * FROM orders
                dimensions:
                  - name: id
                    type: number
                    sql: id
                    primary_key: true
                  - name: status
                    type: string
                    sql: status
                  - name: created_at
                    type: time
                    sql: created_at
                  - name: city
                    type: string
                    sql: city
                measures:
                  - name: count
                    type: count
                  - name: total
                    type: sum
                    sql: amount
                segments:
                  - name: completed
                    sql: \"{CUBE}.status = 'completed'\"
                pre_aggregations:
                  - name: symbol_rollup
                    type: rollup
                    measures:
                      - count
                    time_dimension: created_at
                    granularity: day
                  - name: interpolated_rollup
                    type: rollup
                    measures:
                      - '{CUBE}.count'
                    dimensions:
                      - '{CUBE}.status'
                    segments:
                      - '{CUBE}.completed'
                    time_dimension: '{CUBE}.created_at'
                    granularity: day
                  - name: interpolated_rollup_no_granularity
                    type: rollup
                    measures:
                      - count
                    time_dimension: '{CUBE}.created_at'
                  - name: interpolated_rollup_mixed_list
                    type: rollup
                    measures:
                      - '{CUBE}.total'
                      - count
                    dimensions:
                      - status
                      - '{CUBE}.city'
                    time_dimension: '{CUBE}.created_at'
                    granularity: day
                  - name: broken_rollup_granularity_suffix
                    type: rollup
                    measures:
                      - count
                    time_dimension: '{CUBE}.created_at_day'
                    granularity: day
                  - name: broken_rollup_unknown_measure
                    type: rollup
                    measures:
                      - '{CUBE}.unknown_total'
                    time_dimension: created_at
                    granularity: day
                  - name: interpolated_rollup_granularity_segment
                    type: rollup
                    measures:
                      - count
                    time_dimension: '{CUBE}.created_at.day'
                    granularity: day
                  - name: symbol_rollup_granularity_segment
                    type: rollup
                    measures:
                      - count
                    time_dimension: created_at.day
                    granularity: day
                  - name: broken_rollup_expression_dimension
                    type: rollup
                    measures:
                      - count
                    dimensions:
                      - \"{CUBE}.status = 'completed'\"
                    time_dimension: created_at
                    granularity: day
        "})
        .unwrap();
        TestContext::new(schema).unwrap()
    }

    fn compile_pre_agg(
        ctx: &TestContext,
        pre_agg_name: &str,
    ) -> Result<Rc<CompiledPreAggregation>, CubeError> {
        let cube_names = vec!["orders".to_string()];
        let mut compiler =
            PreAggregationsCompiler::try_new(ctx.query_tools().clone(), &cube_names).unwrap();
        let name = PreAggregationFullName::new("orders".to_string(), pre_agg_name.to_string());
        compiler.compile_pre_aggregation(&name)
    }

    #[test]
    fn test_time_dimension_resolves_to_member_symbol() {
        let ctx = create_reference_context();
        let compiled = compile_pre_agg(&ctx, "symbol_rollup").unwrap();

        assert_eq!(compiled.time_dimensions.len(), 1);
        assert_eq!(
            compiled.time_dimensions[0].full_name(),
            "orders.created_at_day"
        );
        assert_eq!(compiled.granularity, Some("day".to_string()));
    }

    #[test]
    fn test_interpolated_references_resolve_to_member_symbols() {
        let ctx = create_reference_context();
        let compiled = compile_pre_agg(&ctx, "interpolated_rollup").unwrap();

        assert_eq!(
            compiled
                .measures
                .iter()
                .map(|m| m.full_name())
                .collect_vec(),
            vec!["orders.count".to_string()]
        );
        assert_eq!(
            compiled
                .dimensions
                .iter()
                .map(|d| d.full_name())
                .collect_vec(),
            vec!["orders.status".to_string()]
        );
        assert_eq!(
            compiled
                .segments
                .iter()
                .map(|sg| sg.full_name())
                .collect_vec(),
            vec!["expr:orders.completed".to_string()]
        );
        assert_eq!(compiled.time_dimensions.len(), 1);
        assert_eq!(
            compiled.time_dimensions[0].full_name(),
            "orders.created_at_day"
        );
        assert_eq!(compiled.granularity, Some("day".to_string()));
    }

    #[test]
    fn test_interpolated_time_dimension_without_granularity_resolves() {
        let ctx = create_reference_context();
        let compiled = compile_pre_agg(&ctx, "interpolated_rollup_no_granularity").unwrap();

        assert_eq!(compiled.time_dimensions.len(), 1);
        assert_eq!(compiled.time_dimensions[0].full_name(), "orders.created_at");
        assert_eq!(compiled.granularity, None);
    }

    // One list mixing both forms keeps every member it names, in declaration
    // order — join hints and lambda member matching read the list positionally.
    #[test]
    fn test_interpolated_and_symbol_references_in_one_list() {
        let ctx = create_reference_context();
        let compiled = compile_pre_agg(&ctx, "interpolated_rollup_mixed_list").unwrap();

        assert_eq!(
            compiled
                .measures
                .iter()
                .map(|m| m.full_name())
                .collect_vec(),
            vec!["orders.total".to_string(), "orders.count".to_string()]
        );
        assert_eq!(
            compiled
                .dimensions
                .iter()
                .map(|d| d.full_name())
                .collect_vec(),
            vec!["orders.status".to_string(), "orders.city".to_string()]
        );
        assert_eq!(
            compiled.time_dimensions[0].full_name(),
            "orders.created_at_day"
        );
    }

    // An interpolated reference naming the granularity-suffixed member instead
    // of the member itself, e.g. `(CUBE) => `${CUBE}.created_at_day``.
    #[test]
    fn test_interpolated_time_dimension_with_granularity_suffix_returns_error() {
        let ctx = create_reference_context();
        let err = compile_pre_agg(&ctx, "broken_rollup_granularity_suffix")
            .expect_err("Pre-aggregation with unresolvable time dimension should fail to compile");
        assert_eq!(
            err.message,
            "'created_at_day' not found for path 'orders.created_at_day' in pre-aggregation 'orders.broken_rollup_granularity_suffix'"
        );
    }

    // Naming the granularity inside the reference, `${CUBE}.created_at.day`,
    // instead of through `granularity:`. Rejected — and rejected the same way as
    // the equivalent symbol reference `CUBE.created_at.day`.
    #[test]
    fn test_granularity_inside_the_reference_is_rejected_like_the_symbol_form() {
        let ctx = create_reference_context();
        let interpolated = compile_pre_agg(&ctx, "interpolated_rollup_granularity_segment")
            .expect_err("Granularity inside a time dimension reference should fail to compile");
        let symbol = compile_pre_agg(&ctx, "symbol_rollup_granularity_segment")
            .expect_err("Granularity inside a time dimension reference should fail to compile");

        assert_eq!(interpolated.message, symbol.message);
        assert_eq!(
            interpolated.message,
            "Pre-aggregation time dimension must be a dimension"
        );
    }

    #[test]
    fn test_interpolated_measure_that_does_not_exist_returns_error() {
        let ctx = create_reference_context();
        let err = compile_pre_agg(&ctx, "broken_rollup_unknown_measure")
            .expect_err("Pre-aggregation with unresolvable measure should fail to compile");
        assert_eq!(
            err.message,
            "'unknown_total' not found for path 'orders.unknown_total' in pre-aggregation 'orders.broken_rollup_unknown_measure'"
        );
    }

    // An element built as an expression rather than a member reference names no
    // member, so it is reported instead of dropped from the reference list.
    #[test]
    fn test_reference_that_names_no_member_returns_error() {
        let ctx = create_reference_context();
        let err = compile_pre_agg(&ctx, "broken_rollup_expression_dimension")
            .expect_err("Pre-aggregation with an expression reference should fail to compile");
        assert_eq!(
            err.message,
            "'status = 'completed'' not found for path 'orders.status = 'completed'' in pre-aggregation 'orders.broken_rollup_expression_dimension'"
        );
    }

    #[test]
    fn test_compile_simple_rollup() {
        let schema = MockSchema::from_yaml_file("common/pre_aggregations_test.yaml");
        let test_context = TestContext::new(schema).unwrap();
        let query_tools = test_context.query_tools().clone();

        let cube_names = vec!["visitors".to_string()];
        let mut compiler = PreAggregationsCompiler::try_new(query_tools, &cube_names).unwrap();

        let pre_agg_name =
            PreAggregationFullName::new("visitors".to_string(), "daily_rollup".to_string());
        let compiled = compiler.compile_pre_aggregation(&pre_agg_name).unwrap();

        assert_eq!(compiled.name, "daily_rollup");
        assert_eq!(compiled.cube_name, "visitors");
        assert_eq!(compiled.granularity, Some("day".to_string()));

        // Check measures
        assert_eq!(compiled.measures.len(), 2);
        let measure_names: Vec<String> = compiled.measures.iter().map(|m| m.full_name()).collect();
        assert!(measure_names.contains(&"visitors.count".to_string()));
        assert!(measure_names.contains(&"visitors.unique_source_count".to_string()));

        // Check dimensions
        assert_eq!(compiled.dimensions.len(), 1);
        assert_eq!(compiled.dimensions[0].full_name(), "visitors.source");

        // Check time dimensions (with granularity suffix)
        assert_eq!(compiled.time_dimensions.len(), 1);
        assert_eq!(
            compiled.time_dimensions[0].full_name(),
            "visitors.created_at_day"
        );
    }

    #[test]
    fn test_compile_joined_rollup() {
        let schema = MockSchema::from_yaml_file("common/pre_aggregations_test.yaml");
        let test_context = TestContext::new(schema).unwrap();
        let query_tools = test_context.query_tools().clone();

        let cube_names = vec!["visitor_checkins".to_string()];
        let mut compiler = PreAggregationsCompiler::try_new(query_tools, &cube_names).unwrap();

        let pre_agg_name = PreAggregationFullName::new(
            "visitor_checkins".to_string(),
            "joined_rollup".to_string(),
        );
        let compiled = compiler.compile_pre_aggregation(&pre_agg_name).unwrap();

        assert_eq!(compiled.name, "joined_rollup");
        assert_eq!(compiled.cube_name, "visitor_checkins");
        assert_eq!(compiled.granularity, Some("day".to_string()));

        // Check measures
        assert_eq!(compiled.measures.len(), 1);
        assert_eq!(compiled.measures[0].full_name(), "visitor_checkins.count");

        // Check dimensions
        assert_eq!(compiled.dimensions.len(), 2);
        let dimension_names: Vec<String> =
            compiled.dimensions.iter().map(|d| d.full_name()).collect();
        assert!(dimension_names.contains(&"visitor_checkins.visitor_id".to_string()));
        assert!(dimension_names.contains(&"visitors.source".to_string()));

        // Check time dimensions (with granularity suffix)
        assert_eq!(compiled.time_dimensions.len(), 1);
        assert_eq!(
            compiled.time_dimensions[0].full_name(),
            "visitor_checkins.created_at_day"
        );
    }

    #[test]
    fn test_compile_multiplied_rollup() {
        let schema = MockSchema::from_yaml_file("common/pre_aggregations_test.yaml");
        let test_context = TestContext::new(schema).unwrap();
        let query_tools = test_context.query_tools().clone();

        let cube_names = vec!["visitors".to_string()];
        let mut compiler = PreAggregationsCompiler::try_new(query_tools, &cube_names).unwrap();

        let pre_agg_name =
            PreAggregationFullName::new("visitors".to_string(), "multiplied_rollup".to_string());
        let compiled = compiler.compile_pre_aggregation(&pre_agg_name).unwrap();

        assert_eq!(compiled.name, "multiplied_rollup");
        assert_eq!(compiled.cube_name, "visitors");
        assert_eq!(compiled.granularity, Some("day".to_string()));

        // Check measures
        assert_eq!(compiled.measures.len(), 1);
        assert_eq!(compiled.measures[0].full_name(), "visitors.count");

        // Check dimensions
        assert_eq!(compiled.dimensions.len(), 2);
        let dimension_names: Vec<String> =
            compiled.dimensions.iter().map(|d| d.full_name()).collect();
        assert!(dimension_names.contains(&"visitors.source".to_string()));
        assert!(dimension_names.contains(&"visitor_checkins.source".to_string()));

        // Check time dimensions (with granularity suffix)
        assert_eq!(compiled.time_dimensions.len(), 1);
        assert_eq!(
            compiled.time_dimensions[0].full_name(),
            "visitors.created_at_day"
        );
    }

    #[test]
    fn test_compile_all_pre_aggregations() {
        let schema = MockSchema::from_yaml_file("common/pre_aggregations_test.yaml");
        let test_context = TestContext::new(schema).unwrap();
        let query_tools = test_context.query_tools().clone();

        let cube_names = vec!["visitors".to_string(), "visitor_checkins".to_string()];
        let mut compiler = PreAggregationsCompiler::try_new(query_tools, &cube_names).unwrap();

        let compiled = compiler.compile_all_pre_aggregations(false).unwrap();

        // Should compile all 8 pre-aggregations from visitors and visitor_checkins cubes
        //        assert_eq!(compiled.len(), 8);

        let names: Vec<String> = compiled.iter().map(|pa| pa.name.clone()).collect();

        // visitors pre-aggregations
        assert!(names.contains(&"daily_rollup".to_string()));
        assert!(names.contains(&"multiplied_rollup".to_string()));
        assert!(names.contains(&"for_join".to_string()));

        // visitor_checkins pre-aggregations
        assert!(names.contains(&"joined_rollup".to_string()));
        assert!(names.contains(&"checkins_with_visitor_source".to_string()));
        assert!(names.contains(&"for_lambda".to_string()));
        assert!(names.contains(&"lambda_union".to_string()));
    }

    #[test]
    fn test_compile_nonexistent_pre_aggregation() {
        let schema = MockSchema::from_yaml_file("common/pre_aggregations_test.yaml");
        let test_context = TestContext::new(schema).unwrap();
        let query_tools = test_context.query_tools().clone();

        let cube_names = vec!["visitors".to_string()];
        let mut compiler = PreAggregationsCompiler::try_new(query_tools, &cube_names).unwrap();

        let pre_agg_name =
            PreAggregationFullName::new("visitors".to_string(), "nonexistent".to_string());
        let result = compiler.compile_pre_aggregation(&pre_agg_name);

        assert!(result.is_err());
    }

    #[test]
    fn test_compile_rollup_join() {
        let schema = MockSchema::from_yaml_file("common/pre_aggregations_test.yaml");
        let test_context = TestContext::new(schema).unwrap();
        let query_tools = test_context.query_tools().clone();

        // Need both cubes for rollupJoin: visitor_checkins and visitors
        let cube_names = vec!["visitor_checkins".to_string(), "visitors".to_string()];
        let mut compiler = PreAggregationsCompiler::try_new(query_tools, &cube_names).unwrap();

        let pre_agg_name = PreAggregationFullName::new(
            "visitor_checkins".to_string(),
            "checkins_with_visitor_source".to_string(),
        );
        let compiled = compiler.compile_pre_aggregation(&pre_agg_name).unwrap();

        assert_eq!(compiled.name, "checkins_with_visitor_source");
        assert_eq!(compiled.cube_name, "visitor_checkins");

        // Check measures
        assert_eq!(compiled.measures.len(), 1);
        assert_eq!(compiled.measures[0].full_name(), "visitor_checkins.count");

        // Check dimensions
        assert_eq!(compiled.dimensions.len(), 2);
        let dimension_names: Vec<String> =
            compiled.dimensions.iter().map(|d| d.full_name()).collect();
        assert!(dimension_names.contains(&"visitor_checkins.visitor_id".to_string()));
        assert!(dimension_names.contains(&"visitors.source".to_string()));

        // Check source is Join
        match compiled.source.as_ref() {
            PreAggregationSource::Join(_) => {} // Expected
            _ => panic!("Expected PreAggregationSource::Join"),
        }
    }

    #[test]
    fn test_compile_rollup_lambda() {
        let schema = MockSchema::from_yaml_file("common/pre_aggregations_test.yaml");
        let test_context = TestContext::new(schema).unwrap();
        let query_tools = test_context.query_tools().clone();

        let cube_names = vec!["visitor_checkins".to_string()];
        let mut compiler = PreAggregationsCompiler::try_new(query_tools, &cube_names).unwrap();

        let pre_agg_name =
            PreAggregationFullName::new("visitor_checkins".to_string(), "lambda_union".to_string());
        let compiled = compiler.compile_pre_aggregation(&pre_agg_name).unwrap();

        assert_eq!(compiled.name, "lambda_union");
        assert_eq!(compiled.cube_name, "visitor_checkins");
        assert_eq!(compiled.granularity, Some("day".to_string()));

        // Check measures
        assert_eq!(compiled.measures.len(), 1);
        assert_eq!(compiled.measures[0].full_name(), "visitor_checkins.count");

        // Check dimensions
        assert_eq!(compiled.dimensions.len(), 1);
        assert_eq!(
            compiled.dimensions[0].full_name(),
            "visitor_checkins.visitor_id"
        );

        // Check time dimensions
        assert_eq!(compiled.time_dimensions.len(), 1);
        assert_eq!(
            compiled.time_dimensions[0].full_name(),
            "visitor_checkins.created_at_day"
        );

        // Check source is Union
        match compiled.source.as_ref() {
            PreAggregationSource::Union(_) => {} // Expected
            _ => panic!("Expected PreAggregationSource::Union"),
        }
    }

    #[test]
    fn test_pre_aggregation_full_name_from_string() {
        let name = PreAggregationFullName::from_string("visitors.daily_rollup").unwrap();
        assert_eq!(name.cube_name, "visitors");
        assert_eq!(name.name, "daily_rollup");
    }

    #[test]
    fn test_pre_aggregation_full_name_invalid() {
        let result = PreAggregationFullName::from_string("invalid_name");
        assert!(result.is_err());

        let result2 = PreAggregationFullName::from_string("too.many.parts");
        assert!(result2.is_err());
    }

    #[test]
    fn test_compile_rollup_with_segments() {
        let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml");
        let test_context = TestContext::new(schema).unwrap();
        let query_tools = test_context.query_tools().clone();

        let cube_names = vec!["orders".to_string()];
        let mut compiler = PreAggregationsCompiler::try_new(query_tools, &cube_names).unwrap();

        let pre_agg_name =
            PreAggregationFullName::new("orders".to_string(), "segment_rollup".to_string());
        let compiled = compiler.compile_pre_aggregation(&pre_agg_name).unwrap();

        assert_eq!(compiled.name, "segment_rollup");
        assert_eq!(compiled.cube_name, "orders");
        assert_eq!(compiled.granularity, Some("day".to_string()));

        // Check segments
        assert_eq!(compiled.segments.len(), 1);
        assert_eq!(
            compiled.segments[0].full_name(),
            "expr:orders.high_priority"
        );

        // Check measures
        assert_eq!(compiled.measures.len(), 2);
        let measure_names: Vec<String> = compiled.measures.iter().map(|m| m.full_name()).collect();
        assert!(measure_names.contains(&"orders.count".to_string()));
        assert!(measure_names.contains(&"orders.total_amount".to_string()));

        // Check dimensions
        assert_eq!(compiled.dimensions.len(), 1);
        assert_eq!(compiled.dimensions[0].full_name(), "orders.status");
    }

    #[test]
    fn test_compile_rollup_join_calculated_measures() {
        let schema = MockSchema::from_yaml_file("common/rollup_join_calculated_measures.yaml");
        let test_context = TestContext::new(schema).unwrap();
        let query_tools = test_context.query_tools().clone();

        let cube_names = vec![
            "line_items".to_string(),
            "facts".to_string(),
            "campaigns".to_string(),
        ];
        let mut compiler = PreAggregationsCompiler::try_new(query_tools, &cube_names).unwrap();

        let pre_agg_name = PreAggregationFullName::new(
            "line_items".to_string(),
            "combined_rollup_join".to_string(),
        );
        let compiled = compiler.compile_pre_aggregation(&pre_agg_name).unwrap();

        // Check source is Join with correct structure
        match compiled.source.as_ref() {
            PreAggregationSource::Join(join) => {
                // Root should be li_rollup
                match join.root.as_ref() {
                    PreAggregationSource::Single(table) => {
                        assert_eq!(table.name, "li_rollup", "Root should be li_rollup");
                    }
                    _ => panic!("Expected Single source for root"),
                }
                // Should have 2 join items (line_items→facts, line_items→campaigns)
                assert_eq!(
                    join.items.len(),
                    2,
                    "Should have 2 join items, got {}",
                    join.items.len()
                );
                let to_names: Vec<String> = join
                    .items
                    .iter()
                    .map(|item| match item.to.as_ref() {
                        PreAggregationSource::Single(table) => table.name.clone(),
                        _ => panic!("Expected Single source"),
                    })
                    .collect();
                assert!(
                    to_names.contains(&"facts_rollup".to_string()),
                    "Should join to facts_rollup, got: {:?}",
                    to_names
                );
                assert!(
                    to_names.contains(&"campaigns_rollup".to_string()),
                    "Should join to campaigns_rollup, got: {:?}",
                    to_names
                );
            }
            _ => panic!("Expected PreAggregationSource::Join"),
        }
    }
}
